[jaeger-v2] Add support for GRPC storarge

cmd/jaeger/grpc_config.yaml

@@ -0,0 +1,35 @@
+service:
+  extensions: [jaeger_storage, jaeger_query]
+  pipelines:
+    traces:
+      receivers: [otlp]
+      processors: [batch]
+      exporters: [jaeger_storage_exporter]
+
+extensions:
+  jaeger_query:
+    trace_storage: external-storage
+    trace_storage_archive: external-storage-archive
+    ui_config: ./cmd/jaeger/config-ui.json
+
+  jaeger_storage:
+    grpc:
+      external-storage:
+        server: localhost:17271
+        connection-timeout: 5s
+      external-storage-archive:
+        server: localhost:17281
+        connection-timeout: 5s
+
+receivers:
+  otlp:
+    protocols:
+      grpc:
+      http:
+
+processors:
+  batch:
+
+exporters:
+  jaeger_storage_exporter:
+    trace_storage: external-storage

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

@@ -9,12 +9,14 @@ import (
 
 	memoryCfg "github.com/jaegertracing/jaeger/pkg/memory/config"
 	badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
+	grpcCfg "github.com/jaegertracing/jaeger/plugin/storage/grpc/config"
 )
 
 // Config has the configuration for jaeger-query,
 type Config struct {
 	Memory map[string]memoryCfg.Configuration   `mapstructure:"memory"`
 	Badger map[string]badgerCfg.NamespaceConfig `mapstructure:"badger"`
+	GRPC   map[string]grpcCfg.Configuration     `mapstructure:"grpc"`
 	// TODO add other storage types here
 	// TODO how will this work with 3rd party storage implementations?
 	//      Option: instead of looking for specific name, check interface.
@@ -27,6 +29,7 @@ type MemoryStorage struct {
 
 func (cfg *Config) Validate() error {
 	emptyCfg := createDefaultConfig().(*Config)
+	//nolint:govet // The remoteRPCClient field in GRPC.Configuration contains error type
 	if reflect.DeepEqual(*cfg, *emptyCfg) {
 		return fmt.Errorf("%s: no storage type present in config", ID)
 	} else {

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

@@ -17,6 +17,8 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/metrics"
 	"github.com/jaegertracing/jaeger/plugin/storage/badger"
 	badgerCfg "github.com/jaegertracing/jaeger/plugin/storage/badger"
+	"github.com/jaegertracing/jaeger/plugin/storage/grpc"
+	grpcCfg "github.com/jaegertracing/jaeger/plugin/storage/grpc/config"
 	"github.com/jaegertracing/jaeger/plugin/storage/memory"
 	"github.com/jaegertracing/jaeger/storage"
 )
@@ -107,10 +109,17 @@ func (s *storageExt) Start(ctx context.Context, host component.Host) error {
 		cfg:         s.config.Badger,
 		builder:     badger.NewFactoryWithConfig,
 	}
+	grpcStarter := &starter[grpcCfg.Configuration, *grpc.Factory]{
+		ext:         s,
+		storageKind: "grpc",
+		cfg:         s.config.GRPC,
+		builder:     grpc.NewFactoryWithConfig,
+	}
 
 	builders := []func(ctx context.Context, host component.Host) error{
 		memStarter.build,
 		badgerStarter.build,
+		grpcStarter.build,
 		// TODO add support for other backends
 	}
 	for _, builder := range builders {

plugin/storage/grpc/config/config.go

@@ -50,6 +50,7 @@
 	pluginHealthCheck     *time.Ticker
 	pluginHealthCheckDone chan bool
 	pluginRPCClient       plugin.ClientProtocol
+	remoteRPCClient       *grpc.ClientConn
 }
 
 // ClientPluginServices defines services plugin can expose and its capabilities
@@ -78,6 +79,9 @@
 		c.pluginHealthCheck.Stop()
 		c.pluginHealthCheckDone <- true
 	}
+	if c.remoteRPCClient != nil {
+		c.remoteRPCClient.Close()
+	}
 
 	return c.RemoteTLS.Close()
 }
@@ -106,12 +110,13 @@
 		opts = append(opts, grpc.WithUnaryInterceptor(tenancy.NewClientUnaryInterceptor(tenancyMgr)))
 		opts = append(opts, grpc.WithStreamInterceptor(tenancy.NewClientStreamInterceptor(tenancyMgr)))
 	}
-	conn, err := grpc.DialContext(ctx, c.RemoteServerAddr, opts...)
+	var err error
+	c.remoteRPCClient, err = grpc.DialContext(ctx, c.RemoteServerAddr, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("error connecting to remote storage: %w", err)
 	}
 
-	grpcClient := shared.NewGRPCClient(conn)
+	grpcClient := shared.NewGRPCClient(c.remoteRPCClient)
 	return &ClientPluginServices{
 		PluginServices: shared.PluginServices{
 			Store:               grpcClient,

plugin/storage/grpc/factory.go

@@ -60,6 +60,21 @@ func NewFactory() *Factory {
 	return &Factory{}
 }
 
+// NewFactoryWithConfig is used from jaeger(v2).
+func NewFactoryWithConfig(
+	cfg config.Configuration,
+	metricsFactory metrics.Factory,
+	logger *zap.Logger,
+) (*Factory, error) {
+	f := NewFactory()
+	f.InitFromOptions(Options{Configuration: cfg})
+	err := f.Initialize(metricsFactory, logger)
+	if err != nil {
+		return nil, err
+	}
+	return f, nil
+}
+
 // AddFlags implements plugin.Configurable
 func (f *Factory) AddFlags(flagSet *flag.FlagSet) {
 	f.options.AddFlags(flagSet)

plugin/storage/grpc/factory_test.go

@@ -16,13 +16,17 @@ package grpc
 
 import (
 	"errors"
+	"log"
+	"net"
 	"testing"
+	"time"
 
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
+	"google.golang.org/grpc"
 
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/metrics"
@@ -143,6 +147,36 @@ func TestGRPCStorageFactory(t *testing.T) {
 	assert.Equal(t, f.store.DependencyReader(), depReader)
 }
 
+func TestGRPCStorageFactoryWithConfig(t *testing.T) {
+	cfg := grpcConfig.Configuration{}
+	_, err := NewFactoryWithConfig(cfg, metrics.NullFactory, zap.NewNop())
+	require.ErrorContains(t, err, "grpc-plugin builder failed to create a store: error connecting to remote storage")
+
+	lis, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatalf("failed to listen: %v", err)
+	}
+
+	s := grpc.NewServer()
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("Server exited with error: %v", err)
+		}
+	}()
+	defer s.Stop()
+
+	cfg.RemoteServerAddr = lis.Addr().String()
+	cfg.RemoteConnectTimeout = 1 * time.Second
+	f, err := NewFactoryWithConfig(cfg, metrics.NullFactory, zap.NewNop())
+	defer func() {
+		err := f.Close()
+		if err != nil {
+			log.Fatalf("Client exited with error: %v", err)
+		}
+	}()
+	require.NoError(t, err)
+}
+
 func TestGRPCStorageFactory_Capabilities(t *testing.T) {
 	f := NewFactory()
 	v := viper.New()