[jaeger-v2] Add remotesampling extension

cmd/jaeger/config-badger.yaml

@@ -1,12 +1,20 @@
 service:
-  extensions: [jaeger_storage, jaeger_query]
+  extensions: [jaeger_storage, jaeger_query, remote_sampling]
   pipelines:
     traces:
       receivers: [otlp]
-      processors: [batch]
+      processors: [batch, adaptivesampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    file:
+      path: ./cmd/jaeger/sampling-strategies.json
+    # adaptive:
+    #   initial_sampling_probability: 0.1
+    #   strategy_store: badger_main
+
   jaeger_query:
     trace_storage: badger_main
     trace_storage_archive: badger_archive
@@ -37,6 +45,8 @@ receivers:
 
 processors:
   batch:
+  adaptivesampling:
+    strategy_store: badger_main
 
 exporters:
   jaeger_storage_exporter:

cmd/jaeger/config-cassandra.yaml

@@ -1,12 +1,20 @@
 service:
-  extensions: [jaeger_storage, jaeger_query]
+  extensions: [jaeger_storage, jaeger_query, remote_sampling]
   pipelines:
     traces:
       receivers: [otlp]
-      processors: [batch]
+      processors: [batch, adaptivesampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    file:
+      path: ./cmd/jaeger/sampling-strategies.json
+    # adaptive:
+    #   initial_sampling_probability: 0.1
+    #   strategy_store: cassandra_main
+
   jaeger_query:
     trace_storage: cassandra_main
     trace_storage_archive: cassandra_archive
@@ -35,6 +43,9 @@ receivers:
 
 processors:
   batch:
+  adaptivesampling:
+    strategy_store: cassandra_main
+
 
 exporters:
   jaeger_storage_exporter:

cmd/jaeger/config-elasticsearch.yaml

@@ -1,12 +1,20 @@
 service:
-  extensions: [jaeger_storage, jaeger_query]
+  extensions: [jaeger_storage, jaeger_query, remote_sampling]
   pipelines:
     traces:
       receivers: [otlp]
-      processors: [batch]
+      processors: [batch, adaptivesampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    file:
+      path: ./cmd/jaeger/sampling-strategies.json
+    # adaptive:
+    #   initial_sampling_probability: 0.1
+    #   strategy_store: es_main
+
   jaeger_query:
     trace_storage: es_main
     trace_storage_archive: es_archive
@@ -35,6 +43,8 @@ receivers:
 
 processors:
   batch:
+  adaptivesampling:
+    strategy_store: es_main
 
 exporters:
   jaeger_storage_exporter:

cmd/jaeger/config-opensearch.yaml

@@ -1,12 +1,20 @@
 service:
-  extensions: [jaeger_storage, jaeger_query]
+  extensions: [jaeger_storage, jaeger_query, remote_sampling]
   pipelines:
     traces:
       receivers: [otlp]
-      processors: [batch]
+      processors: [batch, adaptivesampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    file:
+      path: ./cmd/jaeger/sampling-strategies.json
+    # adaptive:
+    #   initial_sampling_probability: 0.1
+    #   strategy_store: os_main
+
   jaeger_query:
     trace_storage: os_main
     trace_storage_archive: os_archive
@@ -36,6 +44,8 @@ receivers:
 
 processors:
   batch:
+  adaptivesampling:
+    strategy_store: os_main
 
 exporters:
   jaeger_storage_exporter:

cmd/jaeger/config.yaml

@@ -1,9 +1,9 @@
 service:
-  extensions: [jaeger_storage, jaeger_query]
+  extensions: [jaeger_storage, jaeger_query, remote_sampling]
   pipelines:
     traces:
       receivers: [otlp, jaeger, zipkin]
-      processors: [batch]
+      processors: [batch, adaptivesampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
@@ -13,6 +13,14 @@ extensions:
   # zpages:
   #   endpoint: 0.0.0.0:55679
 
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    file:
+      path: ./cmd/jaeger/sampling-strategies.json
+    # adaptive:
+    #   initial_sampling_probability: 0.1
+    #   strategy_store: memstore
+
   jaeger_query:
     trace_storage: memstore
     trace_storage_archive: memstore_archive
@@ -42,6 +50,8 @@ receivers:
 
 processors:
   batch:
+  adaptivesampling:
+    strategy_store: memstore
 
 exporters:
   jaeger_storage_exporter:

cmd/jaeger/internal/components.go

@@ -29,7 +29,9 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/exporters/storageexporter"
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerquery"
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerstorage"
+	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/remotesampling"
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/integration/storagecleaner"
+	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/processors/adaptivesampling"
 )
 
 type builders struct {
@@ -61,8 +63,8 @@ func (b builders) build() (otelcol.Factories, error) {
 		// add-ons
 		jaegerquery.NewFactory(),
 		jaegerstorage.NewFactory(),
+		remotesampling.NewFactory(),
 		storagecleaner.NewFactory(),
-		// TODO add adaptive sampling
 	)
 	if err != nil {
 		return otelcol.Factories{}, err
@@ -99,7 +101,7 @@ func (b builders) build() (otelcol.Factories, error) {
 		batchprocessor.NewFactory(),
 		memorylimiterprocessor.NewFactory(),
 		// add-ons
-		// TODO add adaptive sampling
+		adaptivesampling.NewFactory(),
 	)
 	if err != nil {
 		return otelcol.Factories{}, err

cmd/jaeger/internal/extension/remotesampling/config.go

@@ -0,0 +1,103 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package remotesampling
+
+import (
+	"errors"
+	"reflect"
+	"time"
+)
+
+var (
+	errNoSource       = errors.New("no sampling strategy specified, has to be either 'adaptive' or 'static'")
+	errMultipleSource = errors.New("only one sampling strategy can be specified, has to be either 'adaptive' or 'static'")
+)
+
+type FileConfig struct {
+	// File specifies a local file as the strategies source
+	Path string `mapstructure:"path"`
+}
+
+type AdaptiveConfig struct {
+	StrategyStore string `mapstructure:"strategy_store"`
+
+	// TargetSamplesPerSecond is the global target rate of samples per operation.
+	// TODO implement manual overrides per service/operation.
+	TargetSamplesPerSecond float64 `mapstructure:"target_samples_per_second"`
+
+	// DeltaTolerance is the acceptable amount of deviation between the observed and the desired (target)
+	// throughput for an operation, expressed as a ratio. For example, the value of 0.3 (30% deviation)
+	// means that if abs((actual-expected) / expected) < 0.3, then the actual sampling rate is "close enough"
+	// and the system does not need to send an updated sampling probability (the "control signal" u(t)
+	// in the PID Controller terminology) to the sampler in the application.
+	//
+	// Increase this to reduce the amount of fluctuation in the calculated probabilities.
+	DeltaTolerance float64 `mapstructure:"delta_tolerance"`
+
+	// CalculationInterval determines how often new probabilities are calculated. E.g. if it is 1 minute,
+	// new sampling probabilities are calculated once a minute and each bucket will contain 1 minute worth
+	// of aggregated throughput data.
+	CalculationInterval time.Duration `mapstructure:"calculation_interval"`
+
+	// AggregationBuckets is the total number of aggregated throughput buckets kept in memory, ie. if
+	// the CalculationInterval is 1 minute (each bucket contains 1 minute of thoughput data) and the
+	// AggregationBuckets is 3, the adaptive sampling processor will keep at most 3 buckets in memory for
+	// all operations.
+	// TODO(wjang): Expand on why this is needed when BucketsForCalculation seems to suffice.
+	AggregationBuckets int `mapstructure:"aggregation_buckets"`
+
+	// BucketsForCalculation determines how many previous buckets used in calculating the weighted QPS,
+	// ie. if BucketsForCalculation is 1, only the most recent bucket will be used in calculating the weighted QPS.
+	BucketsForCalculation int `mapstructure:"buckets_for_calculation"`
+
+	// Delay is the amount of time to delay probability generation by, ie. if the CalculationInterval
+	// is 1 minute, the number of buckets is 10, and the delay is 2 minutes, then at one time
+	// we'll have [now()-12m,now()-2m] range of throughput data in memory to base the calculations
+	// off of. This delay is necessary to counteract the rate at which the jaeger clients poll for
+	// the latest sampling probabilities. The default client poll rate is 1 minute, which means that
+	// during any 1 minute interval, the clients will be fetching new probabilities in a uniformly
+	// distributed manner throughout the 1 minute window. By setting the delay to 2 minutes, we can
+	// guarantee that all clients can use the latest calculated probabilities for at least 1 minute.
+	Delay time.Duration `mapstructure:"delay"`
+
+	// InitialSamplingProbability is the initial sampling probability for all new operations.
+	InitialSamplingProbability float64 `mapstructure:"initial_sampling_probability"`
+
+	// MinSamplingProbability is the minimum sampling probability for all operations. ie. the calculated sampling
+	// probability will be in the range [MinSamplingProbability, 1.0].
+	MinSamplingProbability float64 `mapstructure:"min_sampling_probability"`
+
+	// MinSamplesPerSecond determines the min number of traces that are sampled per second.
+	// For example, if the value is 0.01666666666 (one every minute), then the sampling processor will do
+	// its best to sample at least one trace a minute for an operation. This is useful for low QPS operations
+	// that may never be sampled by the probabilistic sampler.
+	MinSamplesPerSecond float64 `mapstructure:"min_samples_per_second"`
+
+	// LeaderLeaseRefreshInterval is the duration to sleep if this processor is elected leader before
+	// attempting to renew the lease on the leader lock. NB. This should be less than FollowerLeaseRefreshInterval
+	// to reduce lock thrashing.
+	LeaderLeaseRefreshInterval time.Duration `mapstructure:"leader_lease_refresh_interval"`
+
+	// FollowerLeaseRefreshInterval is the duration to sleep if this processor is a follower
+	// (ie. failed to gain the leader lock).
+	FollowerLeaseRefreshInterval time.Duration `mapstructure:"follower_lease_refresh_interval"`
+}
+
+type Config struct {
+	File     FileConfig     `mapstructure:"file"`
+	Adaptive AdaptiveConfig `mapstructure:"adaptive"`
+	Port     string         `mapstructure:"port"`
+}
+
+func (cfg *Config) Validate() error {
+	emptyCfg := createDefaultConfig().(*Config)
+	if reflect.DeepEqual(*cfg, *emptyCfg) {
+		return errNoSource
+	}
+
+	if cfg.File.Path != "" && cfg.Adaptive.StrategyStore != "" {
+		return errMultipleSource
+	}
+	return nil
+}