[WIP] Add Adaptive Sampling Support for gRPC Remote Storage

.github/workflows/ci-e2e-grpc.yml

@@ -38,10 +38,10 @@ jobs:
       run: |
         case ${{ matrix.version }} in
           v1)
-            SPAN_STORAGE_TYPE=memory make grpc-storage-integration-test
+            SAMPLING_STORAGE_TYPE=memory SPAN_STORAGE_TYPE=memory make grpc-storage-integration-test
             ;;
           v2)
-            STORAGE=grpc SPAN_STORAGE_TYPE=memory make jaeger-v2-storage-integration-test
+            SAMPLING_STORAGE_TYPE=memory STORAGE=grpc SPAN_STORAGE_TYPE=memory make jaeger-v2-storage-integration-test
             ;;
         esac
 

cmd/remote-storage/app/server.go

@@ -22,6 +22,7 @@
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 	"github.com/jaegertracing/jaeger/storage"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
+	"github.com/jaegertracing/jaeger/storage/samplingstore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
@@ -36,8 +37,8 @@
 }
 
 // NewServer creates and initializes Server.
-func NewServer(options *Options, storageFactory storage.BaseFactory, tm *tenancy.Manager, telset telemetry.Settings) (*Server, error) {
-	handler, err := createGRPCHandler(storageFactory, telset.Logger)
+func NewServer(options *Options, storageFactory storage.BaseFactory, tm *tenancy.Manager, telset telemetry.Settings, samplingStoreFactory storage.SamplingStoreFactory) (*Server, error) {
+	handler, err := createGRPCHandler(storageFactory, samplingStoreFactory, telset.Logger)
 	if err != nil {
 		return nil, err
 	}
@@ -54,7 +55,7 @@
 	}, nil
 }
 
-func createGRPCHandler(f storage.BaseFactory, logger *zap.Logger) (*shared.GRPCHandler, error) {
+func createGRPCHandler(f storage.BaseFactory, samplingStoreFactory storage.SamplingStoreFactory, logger *zap.Logger) (*shared.GRPCHandler, error) {
 	reader, err := f.CreateSpanReader()
 	if err != nil {
 		return nil, err
@@ -67,12 +68,17 @@
 	if err != nil {
 		return nil, err
 	}
+	samplingStore, err := samplingStoreFactory.CreateSamplingStore(1)
+	if err != nil {
+		return nil, err
+	}
 
 	impl := &shared.GRPCHandlerStorageImpl{
 		SpanReader:          func() spanstore.Reader { return reader },
 		SpanWriter:          func() spanstore.Writer { return writer },
 		DependencyReader:    func() dependencystore.Reader { return depReader },
 		StreamingSpanWriter: func() spanstore.Writer { return nil },
+		SamplingStore:       func() samplingstore.Store { return samplingStore },
 	}
 
 	// borrow code from Query service for archive storage

cmd/remote-storage/app/server_test.go

@@ -47,6 +47,7 @@ func TestNewServer_CreateStorageErrors(t *testing.T) {
 			factory,
 			tenancy.NewManager(&tenancy.Options{}),
 			telemetry.NoopSettings(),
+			nil,
 		)
 	}
 	_, err := f()
@@ -118,13 +119,14 @@ func TestNewServer_TLSConfigError(t *testing.T) {
 		storageMocks.factory,
 		tenancy.NewManager(&tenancy.Options{}),
 		telset,
+		nil,
 	)
 	assert.ErrorContains(t, err, "invalid TLS config")
 }
 
 func TestCreateGRPCHandler(t *testing.T) {
 	storageMocks := newStorageMocks()
-	h, err := createGRPCHandler(storageMocks.factory, zap.NewNop())
+	h, err := createGRPCHandler(storageMocks.factory, nil, zap.NewNop())
 	require.NoError(t, err)
 
 	storageMocks.writer.On("WriteSpan", mock.Anything, mock.Anything).Return(errors.New("writer error"))
@@ -325,6 +327,7 @@ func TestServerGRPCTLS(t *testing.T) {
 				storageMocks.factory,
 				tm,
 				telset,
+				nil,
 			)
 			require.NoError(t, err)
 			require.NoError(t, server.Start())
@@ -374,6 +377,7 @@ func TestServerHandlesPortZero(t *testing.T) {
 		storageMocks.factory,
 		tenancy.NewManager(&tenancy.Options{}),
 		telset,
+		nil,
 	)
 	require.NoError(t, err)
 

cmd/remote-storage/main.go

@@ -77,7 +77,7 @@ func main() {
 			tm := tenancy.NewManager(&opts.Tenancy)
 			telset := baseTelset // copy
 			telset.Metrics = metricsFactory
-			server, err := app.NewServer(opts, storageFactory, tm, telset)
+			server, err := app.NewServer(opts, storageFactory, tm, telset, nil)
 			if err != nil {
 				logger.Fatal("Failed to create server", zap.Error(err))
 			}

plugin/storage/grpc/factory.go

@@ -30,6 +30,7 @@ import (
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 	"github.com/jaegertracing/jaeger/storage"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
+	"github.com/jaegertracing/jaeger/storage/samplingstore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	"github.com/jaegertracing/jaeger/storage/spanstore/spanstoremetrics"
 )
@@ -161,6 +162,7 @@ func (f *Factory) newRemoteStorage(
 			Store:               grpcClient,
 			ArchiveStore:        grpcClient,
 			StreamingSpanWriter: grpcClient,
+			SamplingStore:       grpcClient,
 		},
 		Capabilities: grpcClient,
 	}, nil
@@ -230,6 +232,10 @@ func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 	return f.services.ArchiveStore.ArchiveSpanWriter(), nil
 }
 
+func (f *Factory) CreateSamplingStore() (samplingstore.Store, error) {
+	return f.services.SamplingStore.SamplingStore(), nil
+}
+
 // Close closes the resources held by the factory
 func (f *Factory) Close() error {
 	var errs []error

plugin/storage/grpc/proto/storage.proto

@@ -148,6 +148,63 @@ message FindTraceIDsResponse {
     ];
 }
 
+message Throughput {
+  string Service = 1;
+  string Operation = 2;
+  int64 Count = 3;
+  repeated double Probabilities = 4;
+}
+
+message InsertThroughputRequest {
+  repeated Throughput throughput = 1;
+}
+
+message InsertThroughputResponse {
+}
+
+message StringFloatMap {
+  map<string, double> stringFloatMap = 1;
+}
+
+message ServiceOperationProbabilities {
+  map<string, StringFloatMap> serviceOperationProbabilities = 1;
+}
+
+message ServiceOperationQPS {
+  map<string, StringFloatMap> serviceOperationQPS = 1;
+}
+
+message InsertProbabilitiesAndQPSRequest {
+  string hostname = 1;
+  ServiceOperationProbabilities probabilities = 2;
+  ServiceOperationQPS qps = 3;
+}
+
+message InsertProbabilitiesAndQPSResponse {
+}
+
+message GetThroughputRequest {
+  google.protobuf.Timestamp start_time = 1[
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false
+  ];
+  google.protobuf.Timestamp end_time = 2 [
+    (gogoproto.stdtime) = true,
+    (gogoproto.nullable) = false
+  ];
+}
+
+message GetThroughputResponse {
+  repeated Throughput throughput = 1;
+}
+
+message GetLatestProbabilitiesRequest {
+}
+
+message GetLatestProbabilitiesResponse {
+  ServiceOperationProbabilities serviceOperationProbabilities = 1;
+}
+
 service SpanWriterPlugin {
     // spanstore/Writer
     rpc WriteSpan(WriteSpanRequest) returns (WriteSpanResponse);
@@ -182,6 +239,16 @@ service DependenciesReaderPlugin {
     rpc GetDependencies(GetDependenciesRequest) returns (GetDependenciesResponse);
 }
 
+service SamplingStorePlugin{
+    rpc InsertThroughput(InsertThroughputRequest) returns (InsertThroughputResponse);
+
+    rpc InsertProbabilitiesAndQPS(InsertProbabilitiesAndQPSRequest) returns (InsertProbabilitiesAndQPSResponse);
+
+    rpc GetThroughput(GetThroughputRequest) returns (GetThroughputResponse);
+
+    rpc GetLatestProbabilities(GetLatestProbabilitiesRequest) returns (GetLatestProbabilitiesResponse);
+}
+
 // empty; extensible in the future
 message CapabilitiesRequest {
 

plugin/storage/grpc/shared/grpc_client.go

@@ -14,10 +14,12 @@
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
+	samplingStoreModel "github.com/jaegertracing/jaeger/cmd/collector/app/sampling/model"
 	"github.com/jaegertracing/jaeger/model"
 	_ "github.com/jaegertracing/jaeger/pkg/gogocodec" // force gogo codec registration
 	"github.com/jaegertracing/jaeger/proto-gen/storage_v1"
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
+	"github.com/jaegertracing/jaeger/storage/samplingstore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
@@ -39,6 +41,7 @@
 	capabilitiesClient  storage_v1.PluginCapabilitiesClient
 	depsReaderClient    storage_v1.DependenciesReaderPluginClient
 	streamWriterClient  storage_v1.StreamingSpanWriterPluginClient
+	samplingStoreClient storage_v1.SamplingStorePluginClient
 }
 
 func NewGRPCClient(tracedConn *grpc.ClientConn, untracedConn *grpc.ClientConn) *GRPCClient {
@@ -50,6 +53,7 @@
 		capabilitiesClient:  storage_v1.NewPluginCapabilitiesClient(tracedConn),
 		depsReaderClient:    storage_v1.NewDependenciesReaderPluginClient(tracedConn),
 		streamWriterClient:  storage_v1.NewStreamingSpanWriterPluginClient(untracedConn),
+		samplingStoreClient: storage_v1.NewSamplingStorePluginClient(untracedConn),
 	}
 }
 
@@ -68,6 +72,10 @@
 	return c
 }
 
+func (c *GRPCClient) SamplingStore() samplingstore.Store {
+	return c
+}
+
 func (c *GRPCClient) StreamingSpanWriter() spanstore.Writer {
 	return newStreamingSpanWriter(c.streamWriterClient)
 }
@@ -266,3 +274,62 @@
 
 	return &trace, nil
 }
+
+func (c *GRPCClient) InsertThroughput(throughputs []*samplingStoreModel.Throughput) error {
+	ctx := context.Background()
+	storageV1Throughput, err := samplingStoreThroughputsToStorageV1Throughputs(throughputs)
+	if err != nil {
+		return err
+	}
+
+	_, err = c.samplingStoreClient.InsertThroughput(ctx, &storage_v1.InsertThroughputRequest{
+		Throughput: storageV1Throughput,
+	})
+	if err != nil {
+		return fmt.Errorf("plugin error: %w", err)
+	}
+
+	return nil
+}
+
+func (c *GRPCClient) InsertProbabilitiesAndQPS(hostname string, probabilities samplingStoreModel.ServiceOperationProbabilities, qps samplingStoreModel.ServiceOperationQPS) error {
+	ctx := context.Background()
+
+	_, err := c.samplingStoreClient.InsertProbabilitiesAndQPS(ctx, &storage_v1.InsertProbabilitiesAndQPSRequest{
+		Hostname: hostname,
+		Probabilities: &storage_v1.ServiceOperationProbabilities{
+			ServiceOperationProbabilities: sSFloatMapToStorageV1SSFloatMap(probabilities),
+		},
+		Qps: &storage_v1.ServiceOperationQPS{
+			ServiceOperationQPS: sSFloatMapToStorageV1SSFloatMap(qps),
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("plugin error: %w", err)
+	}
+
+	return nil
+}
+
+func (c *GRPCClient) GetThroughput(start, end time.Time) ([]*samplingStoreModel.Throughput, error) {
+	ctx := context.Background()
+	resp, err := c.samplingStoreClient.GetThroughput(ctx, &storage_v1.GetThroughputRequest{
+		StartTime: start,
+		EndTime:   end,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("plugin error: %w", err)
+	}
+
+	return storageV1ThroughputsToSamplingStoreThroughputs(resp.Throughput), nil
+}
+
+func (c *GRPCClient) GetLatestProbabilities() (samplingStoreModel.ServiceOperationProbabilities, error) {
+	ctx := context.Background()
+	resp, err := c.samplingStoreClient.GetLatestProbabilities(ctx, &storage_v1.GetLatestProbabilitiesRequest{})
+	if err != nil {
+		return nil, fmt.Errorf("plugin error: %w", err)
+	}
+
+	return storageV1SSFloatMapToSSFloatMap(resp.ServiceOperationProbabilities.ServiceOperationProbabilities), nil
+}

plugin/storage/grpc/shared/grpc_client_test.go

@@ -112,6 +112,7 @@ func TestNewGRPCClient(t *testing.T) {
 	assert.Implements(t, (*storage_v1.PluginCapabilitiesClient)(nil), client.capabilitiesClient)
 	assert.Implements(t, (*storage_v1.DependenciesReaderPluginClient)(nil), client.depsReaderClient)
 	assert.Implements(t, (*storage_v1.StreamingSpanWriterPluginClient)(nil), client.streamWriterClient)
+	assert.Implements(t, (*storage_v1.SamplingStorePluginClient)(nil), client.samplingStoreClient)
 }
 
 func TestGRPCClientGetServices(t *testing.T) {