Create new grpc storage configuration to align with OTEL

cmd/jaeger/config-remote-storage.yaml

@@ -14,8 +14,10 @@ extensions:
   jaeger_storage:
     grpc:
       external-storage:
-        server: localhost:17271
-        connection-timeout: 5s
+        endpoint: localhost:17271
+        timeout: 5s
+        tls:
+          insecure: true
 
 receivers:
   otlp:

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

@@ -18,7 +18,7 @@ import (
 type Config struct {
 	Memory        map[string]memoryCfg.Configuration   `mapstructure:"memory"`
 	Badger        map[string]badgerCfg.NamespaceConfig `mapstructure:"badger"`
-	GRPC          map[string]grpcCfg.Configuration     `mapstructure:"grpc"`
+	GRPC          map[string]grpcCfg.ConfigV2          `mapstructure:"grpc"`
 	Opensearch    map[string]esCfg.Configuration       `mapstructure:"opensearch"`
 	Elasticsearch map[string]esCfg.Configuration       `mapstructure:"elasticsearch"`
 	Cassandra     map[string]cassandra.Options         `mapstructure:"cassandra"`

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

@@ -117,7 +117,7 @@ func (s *storageExt) Start(ctx context.Context, host component.Host) error {
 		cfg:         s.config.Badger,
 		builder:     badger.NewFactoryWithConfig,
 	}
-	grpcStarter := &starter[grpcCfg.Configuration, *grpc.Factory]{
+	grpcStarter := &starter[grpcCfg.ConfigV2, *grpc.Factory]{
 		ext:         s,
 		storageKind: "grpc",
 		cfg:         s.config.GRPC,

pkg/config/tlscfg/options.go

@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"go.opentelemetry.io/collector/config/configtls"
 	"go.uber.org/zap"
 )
 
@@ -135,6 +136,22 @@ func (p Options) loadCertPool() (*x509.CertPool, error) {
 	return certPool, nil
 }
 
+func (o *Options) ToOtelClientConfig(v2 *configtls.ClientConfig) configtls.ClientConfig {
+	v2.Insecure = !o.Enabled
+
+	v2.ServerName = o.ServerName
+	v2.InsecureSkipVerify = o.SkipHostVerify
+	v2.CAFile = o.CAPath
+	v2.CertFile = o.CertPath
+	v2.KeyFile = o.KeyPath
+	v2.CipherSuites = o.CipherSuites
+	v2.MinVersion = o.MinVersion
+	v2.MaxVersion = o.MaxVersion
+	v2.ReloadInterval = o.ReloadInterval
+
+	return *v2
+}
+
 func addCertToPool(caPath string, certPool *x509.CertPool) error {
 	caPEM, err := os.ReadFile(filepath.Clean(caPath))
 	if err != nil {

plugin/storage/grpc/config/config.go

@@ -15,16 +15,18 @@
 package config
 
 import (
-	"errors"
+	"context"
 	"fmt"
 	"time"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/credentials"
-	"google.golang.org/grpc/credentials/insecure"
 
 	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
@@ -37,72 +39,86 @@ type Configuration struct {
 	RemoteTLS            tlscfg.Options
 	RemoteConnectTimeout time.Duration `yaml:"connection-timeout" mapstructure:"connection-timeout"`
 	TenancyOpts          tenancy.Options
+}
+
+type ConfigV2 struct {
+	TenancyOpts                    tenancy.Options
+	configgrpc.ClientConfig        `mapstructure:",squash"`
+	exporterhelper.TimeoutSettings `mapstructure:",squash"`
+}
+
+func (c *Configuration) translateToConfigV2() *ConfigV2 {
+	v2 := &ConfigV2{}
+	v2.Endpoint = c.RemoteServerAddr
+	v2.Timeout = c.RemoteConnectTimeout
+	v2.TLSSetting = c.RemoteTLS.ToOtelClientConfig(&v2.TLSSetting)
 
-	remoteConn *grpc.ClientConn
+	return v2
 }
 
 // ClientPluginServices defines services plugin can expose and its capabilities
 type ClientPluginServices struct {
 	shared.PluginServices
 	Capabilities shared.PluginCapabilities
+	remoteConn   *grpc.ClientConn
 }
 
 // PluginBuilder is used to create storage plugins. Implemented by Configuration.
 // TODO this interface should be removed and the building capability moved to Factory.
 type PluginBuilder interface {
 	Build(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error)
-	Close() error
 }
 
 // Build instantiates a PluginServices
 func (c *Configuration) Build(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
-	return c.buildRemote(logger, tracerProvider, grpc.NewClient)
+	v2Cfg := c.translateToConfigV2()
+	return v2Cfg.Build(logger, tracerProvider)
 }
 
-type newClientFn func(target string, opts ...grpc.DialOption) (conn *grpc.ClientConn, err error)
+func (c *ConfigV2) Build(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
+	s, err := newRemoteStorage(c, tracerProvider)
+	if err != nil {
+		return nil, err
+	}
+	return s, nil
+}
 
-func (c *Configuration) buildRemote(logger *zap.Logger, tracerProvider trace.TracerProvider, newClient newClientFn) (*ClientPluginServices, error) {
+func newRemoteStorage(c *ConfigV2, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
 	opts := []grpc.DialOption{
 		grpc.WithStatsHandler(otelgrpc.NewClientHandler(otelgrpc.WithTracerProvider(tracerProvider))),
 	}
-	if c.RemoteTLS.Enabled {
-		tlsCfg, err := c.RemoteTLS.Config(logger)
-		if err != nil {
-			return nil, err
-		}
-		creds := credentials.NewTLS(tlsCfg)
-		opts = append(opts, grpc.WithTransportCredentials(creds))
-	} else {
-		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	if c.Auth != nil {
+		return nil, fmt.Errorf("authenticator is not supported")
 	}
 
 	tenancyMgr := tenancy.NewManager(&c.TenancyOpts)
 	if tenancyMgr.Enabled {
 		opts = append(opts, grpc.WithUnaryInterceptor(tenancy.NewClientUnaryInterceptor(tenancyMgr)))
 		opts = append(opts, grpc.WithStreamInterceptor(tenancy.NewClientStreamInterceptor(tenancyMgr)))
 	}
-	var err error
-	c.remoteConn, err = newClient(c.RemoteServerAddr, opts...)
+
+	remoteConn, err := c.ToClientConn(context.Background(), componenttest.NewNopHost(), component.TelemetrySettings{}, opts...)
 	if err != nil {
 		return nil, fmt.Errorf("error creating remote storage client: %w", err)
 	}
-
-	grpcClient := shared.NewGRPCClient(c.remoteConn)
+	grpcClient := shared.NewGRPCClient(remoteConn)
 	return &ClientPluginServices{
 		PluginServices: shared.PluginServices{
 			Store:               grpcClient,
 			ArchiveStore:        grpcClient,
 			StreamingSpanWriter: grpcClient,
 		},
 		Capabilities: grpcClient,
+		remoteConn:   remoteConn,
 	}, nil
 }
 
-func (c *Configuration) Close() error {
-	var errs []error
+func (c *ClientPluginServices) Close() error {
 	if c.remoteConn != nil {
-		errs = append(errs, c.remoteConn.Close())
+		err := c.remoteConn.Close()
+		if err != nil {
+			return err
+		}
 	}
-	errs = append(errs, c.RemoteTLS.Close())
-	return errors.Join(errs...)
+	return nil
 }

plugin/storage/grpc/config/config_test.go

@@ -4,20 +4,16 @@
 package config
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/stretchr/testify/require"
-	"go.uber.org/zap"
-	"google.golang.org/grpc"
+	"go.opentelemetry.io/otel/trace/noop"
 )
 
 func TestBuildRemoteNewClientError(t *testing.T) {
 	// this is a silly test to verify handling of error from grpc.NewClient, which cannot be induced via params.
-	c := &Configuration{}
-	_, err := c.buildRemote(zap.NewNop(), nil, func(target string, opts ...grpc.DialOption) (conn *grpc.ClientConn, err error) {
-		return nil, errors.New("test error")
-	})
+	c := &ConfigV2{}
+	_, err := newRemoteStorage(c, noop.NewTracerProvider())
 	require.Error(t, err)
 	require.Contains(t, err.Error(), "error creating remote storage client")
 }

plugin/storage/grpc/factory.go

@@ -15,6 +15,7 @@
 package grpc
 
 import (
+	"errors"
 	"flag"
 	"fmt"
 	"io"
@@ -53,6 +54,7 @@ type Factory struct {
 	archiveStore        shared.ArchiveStoragePlugin
 	streamingSpanWriter shared.StreamingSpanWriterPlugin
 	capabilities        shared.PluginCapabilities
+	services            *config.ClientPluginServices
 }
 
 // NewFactory creates a new Factory.
@@ -62,12 +64,12 @@ func NewFactory() *Factory {
 
 // NewFactoryWithConfig is used from jaeger(v2).
 func NewFactoryWithConfig(
-	cfg config.Configuration,
+	cfg config.ConfigV2,
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
 ) (*Factory, error) {
 	f := NewFactory()
-	f.configureFromOptions(Options{Configuration: cfg})
+	f.configureFromOptions(Options{ConfigV2: cfg})
 	err := f.Initialize(metricsFactory, logger)
 	if err != nil {
 		return nil, err
@@ -92,23 +94,25 @@ func (f *Factory) InitFromViper(v *viper.Viper, logger *zap.Logger) {
 func (f *Factory) configureFromOptions(opts Options) {
 	f.options = opts
 	f.builder = &f.options.Configuration
+	f.builder = &f.options.ConfigV2
 }
 
 // Initialize implements storage.Factory
 func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error {
 	f.metricsFactory, f.logger = metricsFactory, logger
 	f.tracerProvider = otel.GetTracerProvider()
 
-	services, err := f.builder.Build(logger, f.tracerProvider)
+	var err error
+	f.services, err = f.builder.Build(logger, f.tracerProvider)
 	if err != nil {
 		return fmt.Errorf("grpc storage builder failed to create a store: %w", err)
 	}
 
-	f.store = services.Store
-	f.archiveStore = services.ArchiveStore
-	f.capabilities = services.Capabilities
-	f.streamingSpanWriter = services.StreamingSpanWriter
-	logger.Info("Remote storage configuration", zap.Any("configuration", f.options.Configuration))
+	f.store = f.services.Store
+	f.archiveStore = f.services.ArchiveStore
+	f.capabilities = f.services.Capabilities
+	f.streamingSpanWriter = f.services.StreamingSpanWriter
+	logger.Info("Remote storage configuration", zap.Any("configuration", f.options.ConfigV2))
 	return nil
 }
 
@@ -165,5 +169,11 @@ func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 // Close closes the resources held by the factory
 func (f *Factory) Close() error {
 	// TODO Close should move into Services type, instead of being in the Config.
-	return f.builder.Close()
+	var errs []error
+	err := f.services.Close()
+	if err != nil {
+		errs = append(errs, err)
+	}
+	errs = append(errs, f.options.Configuration.RemoteTLS.Close())
+	return errors.Join(errs...)
 }

plugin/storage/grpc/factory_test.go

@@ -24,6 +24,8 @@ import (
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -148,6 +150,9 @@ func TestGRPCStorageFactory(t *testing.T) {
 }
 
 func TestGRPCStorageFactoryWithConfig(t *testing.T) {
+	cfg := grpcConfig.ConfigV2{}
+	_, 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")
 	require.NoError(t, err, "failed to listen")
 
@@ -159,9 +164,13 @@ func TestGRPCStorageFactoryWithConfig(t *testing.T) {
 	}()
 	defer s.Stop()
 
-	cfg := grpcConfig.Configuration{
-		RemoteServerAddr:     lis.Addr().String(),
-		RemoteConnectTimeout: 1 * time.Second,
+	cfg = grpcConfig.ConfigV2{
+		ClientConfig: configgrpc.ClientConfig{
+			Endpoint: lis.Addr().String(),
+		},
+		TimeoutSettings: exporterhelper.TimeoutSettings{
+			Timeout: 1 * time.Second,
+		},
 	}
 	f, err := NewFactoryWithConfig(cfg, metrics.NullFactory, zap.NewNop())
 	require.NoError(t, err)

plugin/storage/grpc/options.go

@@ -37,6 +37,7 @@ const (
 // to bind them to command line flags
 type Options struct {
 	Configuration config.Configuration `mapstructure:",squash"`
+	ConfigV2      config.ConfigV2      `mapstructure:",squash"`
 }
 
 func tlsFlagsConfig() tlscfg.ClientFlagsConfig {