Merge branch 'main' into hotrod

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, adaptive_sampling]
       exporters: [jaeger_storage_exporter]
 
 extensions:
@@ -25,6 +25,16 @@ extensions:
       memstore_archive:
         max_traces: 100000
 
+  remote_sampling:
+    # You can either use file or adaptive sampling strategy in remote_sampling
+    # file:
+    #   path: ./cmd/jaeger/sampling-strategies.json
+    adaptive:
+      sampling_store: memstore
+      initial_sampling_probability: 0.1
+    http:
+    grpc:
+
 receivers:
   otlp:
     protocols:
@@ -42,6 +52,9 @@ receivers:
 
 processors:
   batch:
+  # Adaptive Sampling Processor is required to support adaptive sampling.
+  # It expects remote_sampling extension with `adaptive:` config to be enabled.
+  adaptive_sampling:
 
 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 {
@@ -62,7 +64,7 @@ func (b builders) build() (otelcol.Factories, error) {
 		jaegerquery.NewFactory(),
 		jaegerstorage.NewFactory(),
 		storagecleaner.NewFactory(),
-		// TODO add adaptive sampling
+		remotesampling.NewFactory(),
 	)
 	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/jaegerstorage/extension.go

@@ -54,7 +54,7 @@ func GetStorageFactory(name string, host component.Host) (storage.Factory, error
 	f, ok := comp.(Extension).Factory(name)
 	if !ok {
 		return nil, fmt.Errorf(
-			"cannot find storage '%s' declared by '%s' extension",
+			"cannot find definition of storage '%s' in the configuration for extension '%s'",
 			name, componentType,
 		)
 	}

cmd/jaeger/internal/extension/jaegerstorage/extension_test.go

@@ -83,7 +83,7 @@ func TestStorageFactoryBadHostError(t *testing.T) {
 func TestStorageFactoryBadNameError(t *testing.T) {
 	host := storageHost{t: t, ext: startStorageExtension(t, "foo")}
 	_, err := GetStorageFactory("bar", host)
-	require.ErrorContains(t, err, "cannot find storage 'bar'")
+	require.ErrorContains(t, err, "cannot find definition of storage 'bar'")
 }
 
 func TestStorageFactoryBadShutdownError(t *testing.T) {

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

@@ -0,0 +1,91 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package remotesampling
+
+import (
+	"errors"
+
+	"github.com/asaskevich/govalidator"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/confmap"
+
+	"github.com/jaegertracing/jaeger/plugin/sampling/strategyprovider/adaptive"
+)
+
+var (
+	errNoProvider        = errors.New("no sampling strategy provider specified, expecting 'adaptive' or 'file'")
+	errMultipleProviders = errors.New("only one sampling strategy provider can be specified, 'adaptive' or 'file'")
+)
+
+var (
+	_ component.Config          = (*Config)(nil)
+	_ component.ConfigValidator = (*Config)(nil)
+	_ confmap.Unmarshaler       = (*Config)(nil)
+)
+
+type Config struct {
+	File     *FileConfig              `mapstructure:"file"`
+	Adaptive *AdaptiveConfig          `mapstructure:"adaptive"`
+	HTTP     *confighttp.ServerConfig `mapstructure:"http"`
+	GRPC     *configgrpc.ServerConfig `mapstructure:"grpc"`
+}
+
+type FileConfig struct {
+	// File specifies a local file as the source of sampling strategies.
+	Path string `valid:"required" mapstructure:"path"`
+}
+
+type AdaptiveConfig struct {
+	// SamplingStore is the name of the storage defined in the jaegerstorage extension.
+	SamplingStore string `valid:"required" mapstructure:"sampling_store"`
+
+	adaptive.Options `mapstructure:",squash"`
+}
+
+// Unmarshal is a custom unmarshaler that allows the factory to provide default values
+// for nested configs (like GRPC endpoint) yes still reset the pointers to nil if the
+// config did not contain the corresponding sections.
+// This is a workaround for the lack of opional fields support in OTEL confmap.
+// Issue: https://github.com/open-telemetry/opentelemetry-collector/issues/10266
+func (cfg *Config) Unmarshal(conf *confmap.Conf) error {
+	// first load the config normally
+	err := conf.Unmarshal(cfg)
+	if err != nil {
+		return err
+	}
+
+	// use string names of fields to see if they are set in the confmap
+	if !conf.IsSet("file") {
+		cfg.File = nil
+	}
+
+	if !conf.IsSet("adaptive") {
+		cfg.Adaptive = nil
+	}
+
+	if !conf.IsSet("grpc") {
+		cfg.GRPC = nil
+	}
+
+	if !conf.IsSet("http") {
+		cfg.HTTP = nil
+	}
+
+	return nil
+}
+
+func (cfg *Config) Validate() error {
+	if cfg.File == nil && cfg.Adaptive == nil {
+		return errNoProvider
+	}
+
+	if cfg.File != nil && cfg.Adaptive != nil {
+		return errMultipleProviders
+	}
+
+	_, err := govalidator.ValidateStruct(cfg)
+	return err
+}

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

@@ -0,0 +1,134 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package remotesampling
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func Test_Validate(t *testing.T) {
+	tests := []struct {
+		name        string
+		config      *Config
+		expectedErr string
+	}{
+		{
+			name:        "No provider specified",
+			config:      &Config{},
+			expectedErr: "no sampling strategy provider specified, expecting 'adaptive' or 'file'",
+		},
+		{
+			name: "Both providers specified",
+			config: &Config{
+				File:     &FileConfig{Path: "test-path"},
+				Adaptive: &AdaptiveConfig{SamplingStore: "test-store"},
+			},
+			expectedErr: "only one sampling strategy provider can be specified, 'adaptive' or 'file'",
+		},
+		{
+			name: "Only File provider specified",
+			config: &Config{
+				File: &FileConfig{Path: "test-path"},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "Only Adaptive provider specified",
+			config: &Config{
+				Adaptive: &AdaptiveConfig{SamplingStore: "test-store"},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "Invalid File provider",
+			config: &Config{
+				File: &FileConfig{Path: ""},
+			},
+			expectedErr: "File.Path: non zero value required",
+		},
+		{
+			name: "Invalid Adaptive provider",
+			config: &Config{
+				Adaptive: &AdaptiveConfig{SamplingStore: ""},
+			},
+			expectedErr: "Adaptive.SamplingStore: non zero value required",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.config.Validate()
+			if tt.expectedErr == "" {
+				require.NoError(t, err)
+			} else {
+				assert.Equal(t, tt.expectedErr, err.Error())
+			}
+		})
+	}
+}
+
+func Test_Unmarshal(t *testing.T) {
+	tests := []struct {
+		name        string
+		input       map[string]any
+		expectedCfg *Config
+		expectedErr string
+	}{
+		{
+			name: "Valid config with File",
+			input: map[string]any{
+				"file": map[string]any{
+					"path": "test-path",
+				},
+			},
+			expectedCfg: &Config{
+				File: &FileConfig{Path: "test-path"},
+			},
+			expectedErr: "",
+		},
+		{
+			name: "Valid config with Adaptive",
+			input: map[string]any{
+				"adaptive": map[string]any{
+					"sampling_store": "test-store",
+				},
+			},
+			expectedCfg: &Config{
+				Adaptive: &AdaptiveConfig{SamplingStore: "test-store"},
+			},
+			expectedErr: "",
+		},
+		{
+			name:        "Empty config",
+			input:       map[string]any{},
+			expectedCfg: &Config{},
+			expectedErr: "",
+		},
+		{
+			name: "Invalid config",
+			input: map[string]any{
+				"foo": "bar",
+			},
+			expectedErr: "invalid keys: foo", // sensitive to lib implementation
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			conf := confmap.NewFromStringMap(tt.input)
+			var cfg Config
+			err := cfg.Unmarshal(conf)
+			if tt.expectedErr == "" {
+				require.NoError(t, err)
+				assert.Equal(t, tt.expectedCfg, &cfg)
+			} else {
+				assert.ErrorContains(t, err, tt.expectedErr)
+			}
+		})
+	}
+}