Replace AlwaysSample with ProbabilitySampler in router (#1348)

charts/README.md

@@ -70,6 +70,7 @@ Parameter | Description | Default
 `router.svcAddressUpdateTimeout` | The length of update lock expiry time for router to get a service URL returns from executor | `30`
 `router.svcAnnotations` | Annotations for router service | None
 `router.useEncodedPath` | For router to match encoded path. If true, "/foo%2Fbar" will match the path "/{var}"; Otherwise, it will match the path "/foo/bar". | `false`
+`router.traceSamplingRate` | Uniformly sample traces with the given probabilistic sampling rate | `0.5`
 `router.roundTrip.disableKeepAlive` | Disable transport keep-alive for fast switching function version | `true`
 `router.roundTrip.keepAliveTime` | The keep-alive period for an active network connection to function pod | `30s`
 `router.roundTrip.timeout` | HTTP transport request timeout | `50ms`

charts/fission-all/templates/router.yaml

@@ -54,7 +54,7 @@ spec:
           - name: ROUTER_SVC_ADDRESS_UPDATE_TIMEOUT
             value: {{ .Values.router.svcAddressUpdateTimeout | default "30s" | quote }}
           - name: TRACING_SAMPLING_RATE
-            value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
+            value: {{ .Values.router.traceSamplingRate | default "0.5" | quote }}
           - name: USE_ENCODED_PATH
             value: {{ .Values.router.useEncodedPath | default false | quote }}
           - name: DEBUG_ENV

charts/fission-all/values.yaml

@@ -98,6 +98,9 @@ router:
     ## Max retries times of a failed request
     maxRetries: 10
 
+  ## Sample with a rate per time window (traces/second)
+  traceSamplingRate: 0.5
+
 ## Message queue trigger config
 ### NATS Streaming, enabled by default
 nats:

charts/fission-core/templates/router.yaml

@@ -54,7 +54,7 @@ spec:
           - name: ROUTER_SVC_ADDRESS_UPDATE_TIMEOUT
             value: {{ .Values.router.svcAddressUpdateTimeout | default "30s" | quote }}
           - name: TRACING_SAMPLING_RATE
-            value: {{ .Values.traceSamplingRate | default "0.5" | quote }}
+            value: {{ .Values.router.traceSamplingRate | default "0.5" | quote }}
           - name: USE_ENCODED_PATH
             value: {{ .Values.router.useEncodedPath | default false | quote }}
           - name: DEBUG_ENV

charts/fission-core/values.yaml

@@ -85,6 +85,9 @@ router:
     ## Max retries times of a failed request
     maxRetries: 10
 
+  ## Sample with a rate per time window (traces/second)
+  traceSamplingRate: 0.5
+
 ## Persist data to a persistent volume.
 persistence:
   ## If true, fission will create/use a Persistent Volume Claim

pkg/router/router.go

@@ -77,13 +77,14 @@ func router(ctx context.Context, logger *zap.Logger, httpTriggerSet *HTTPTrigger
 	return mr
 }
 
-func serve(ctx context.Context, logger *zap.Logger, port int, httpTriggerSet *HTTPTriggerSet, resolver *functionReferenceResolver) {
+func serve(ctx context.Context, logger *zap.Logger, port int, tracingSamplingRate float64, httpTriggerSet *HTTPTriggerSet, resolver *functionReferenceResolver) {
 	mr := router(ctx, logger, httpTriggerSet, resolver)
 	url := fmt.Sprintf(":%v", port)
+
 	http.ListenAndServe(url, &ochttp.Handler{
 		Handler: mr,
 		StartOptions: trace.StartOptions{
-			Sampler: trace.AlwaysSample(),
+			Sampler: trace.ProbabilitySampler(tracingSamplingRate),
 		},
 	})
 }
@@ -191,6 +192,16 @@ func Start(logger *zap.Logger, port int, executorUrl string) {
 			zap.Duration("default", svcAddrUpdateTimeout))
 	}
 
+	tracingSamplingRateStr := os.Getenv("TRACING_SAMPLING_RATE")
+	tracingSamplingRate, err := strconv.ParseFloat(tracingSamplingRateStr, 64)
+	if err != nil {
+		tracingSamplingRate = .5
+		logger.Error("failed to parse tracing sampling rate from 'TRACING_SAMPLING_RATE' - set to the default value",
+			zap.Error(err),
+			zap.String("value", tracingSamplingRateStr),
+			zap.Float64("default", tracingSamplingRate))
+	}
+
 	triggers, _, fnStore := makeHTTPTriggerSet(logger.Named("triggerset"), fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{
 		timeout:           timeout,
 		timeoutExponent:   timeoutExponent,
@@ -207,5 +218,5 @@ func Start(logger *zap.Logger, port int, executorUrl string) {
 	logger.Info("starting router", zap.Int("port", port))
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	serve(ctx, logger, port, triggers, resolver)
+	serve(ctx, logger, port, tracingSamplingRate, triggers, resolver)
 }

pkg/router/router_test.go

@@ -96,9 +96,10 @@ func TestRouter(t *testing.T) {
 
 	// run the router
 	port := 4242
+	tracingSamplingRate := .5
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	go serve(ctx, logger, port, triggers, frr)
+	go serve(ctx, logger, port, tracingSamplingRate, triggers, frr)
 	time.Sleep(100 * time.Millisecond)
 
 	// hit the router