[grpc][v2] Implement gRPC v2 factory (#6968)

<!--
!! Please DELETE this comment before posting.
We appreciate your contribution to the Jaeger project! 👋🎉
-->

## Which problem is this PR solving?
- Towards #6965

## Description of the changes
- This PR implements the following factories for gRPC storage
  - `io.Closer`
  - `tracestore.Factory`
  - `depstore.Factory`
## How was this change tested?
- Added unit tests

## Checklist
- [x] I have read
https://github.com/jaegertracing/jaeger/blob/master/CONTRIBUTING_GUIDELINES.md
- [x] I have signed all commits
- [x] I have added unit tests for the new functionality
- [x] I have run lint and test steps successfully
  - for `jaeger`: `make lint test`
  - for `jaeger-ui`: `npm run lint` and `npm run test`

---------

Signed-off-by: Mahad Zaryab <[email protected]>

Author: mahadzaryab1

internal/storage/v2/grpc/config.go

@@ -0,0 +1,27 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package grpc
+
+import (
+	"time"
+
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+
+	"github.com/jaegertracing/jaeger/internal/tenancy"
+)
+
+type Config struct {
+	Tenancy                      tenancy.Options `mapstructure:"multi_tenancy"`
+	configgrpc.ClientConfig      `mapstructure:",squash"`
+	exporterhelper.TimeoutConfig `mapstructure:",squash"`
+}
+
+func DefaultConfig() Config {
+	return Config{
+		TimeoutConfig: exporterhelper.TimeoutConfig{
+			Timeout: time.Duration(5 * time.Second),
+		},
+	}
+}

internal/storage/v2/grpc/config_test.go

@@ -0,0 +1,15 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package grpc
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestDefaultConfig(t *testing.T) {
+	cfg := DefaultConfig()
+	require.NotEmpty(t, cfg.Timeout)
+}

internal/storage/v2/grpc/factory.go

@@ -0,0 +1,147 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package grpc
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	"go.opentelemetry.io/otel/trace"
+	"go.opentelemetry.io/otel/trace/noop"
+	"google.golang.org/grpc"
+
+	"github.com/jaegertracing/jaeger/internal/bearertoken"
+	"github.com/jaegertracing/jaeger/internal/storage/v2/api/depstore"
+	"github.com/jaegertracing/jaeger/internal/storage/v2/api/tracestore"
+	"github.com/jaegertracing/jaeger/internal/telemetry"
+	"github.com/jaegertracing/jaeger/internal/tenancy"
+)
+
+var (
+	_ io.Closer          = (*Factory)(nil)
+	_ tracestore.Factory = (*Factory)(nil)
+	_ depstore.Factory   = (*Factory)(nil)
+)
+
+type Factory struct {
+	telset telemetry.Settings
+	config Config
+	// readerConn is the gRPC connection used for reading data from the remote storage backend.
+	// It is safe for this connection to have instrumentation enabled without
+	// the risk of recursively generating traces.
+	readerConn *grpc.ClientConn
+	// writerConn is the gRPC connection used for writing data to the remote storage backend.
+	// This connection should not have instrumentation enabled to avoid recursively generating traces.
+	writerConn *grpc.ClientConn
+}
+
+// NewFactory initializes a new gRPC (remote) storage backend.
+func NewFactory(
+	cfg Config,
+	telset telemetry.Settings,
+) (*Factory, error) {
+	f := &Factory{
+		telset: telset,
+		config: cfg,
+	}
+
+	readerTelset := getTelset(f.telset, f.telset.TracerProvider)
+	writerTelset := getTelset(f.telset, noop.NewTracerProvider())
+	newClientFn := func(telset component.TelemetrySettings, opts ...grpc.DialOption) (conn *grpc.ClientConn, err error) {
+		clientOpts := make([]configgrpc.ToClientConnOption, 0)
+		for _, opt := range opts {
+			clientOpts = append(clientOpts, configgrpc.WithGrpcDialOption(opt))
+		}
+		return f.config.ToClientConn(context.Background(), f.telset.Host, telset, clientOpts...)
+	}
+
+	if err := f.initializeConnections(readerTelset, writerTelset, newClientFn); err != nil {
+		return nil, err
+	}
+
+	return f, nil
+}
+
+func (f *Factory) CreateTraceReader() (tracestore.Reader, error) {
+	return NewTraceReader(f.readerConn), nil
+}
+
+func (f *Factory) CreateTraceWriter() (tracestore.Writer, error) {
+	return NewTraceWriter(f.writerConn), nil
+}
+
+func (f *Factory) CreateDependencyReader() (depstore.Reader, error) {
+	return NewDependencyReader(f.readerConn), nil
+}
+
+func (f *Factory) Close() error {
+	var errs []error
+	if f.readerConn != nil {
+		errs = append(errs, f.readerConn.Close())
+	}
+	if f.writerConn != nil {
+		errs = append(errs, f.writerConn.Close())
+	}
+	return errors.Join(errs...)
+}
+
+func getTelset(
+	telset telemetry.Settings,
+	tracerProvider trace.TracerProvider,
+) component.TelemetrySettings {
+	return component.TelemetrySettings{
+		Logger:         telset.Logger,
+		TracerProvider: tracerProvider,
+		MeterProvider:  telset.MeterProvider,
+	}
+}
+
+type newClientFn func(telset component.TelemetrySettings, opts ...grpc.DialOption) (*grpc.ClientConn, error)
+
+func (f *Factory) initializeConnections(
+	readerTelset, writerTelset component.TelemetrySettings,
+	newClient newClientFn,
+) error {
+	if f.config.Auth != nil {
+		return errors.New("authenticator is not supported")
+	}
+
+	unaryInterceptors := []grpc.UnaryClientInterceptor{bearertoken.NewUnaryClientInterceptor()}
+	streamInterceptors := []grpc.StreamClientInterceptor{bearertoken.NewStreamClientInterceptor()}
+
+	if tenancyMgr := tenancy.NewManager(&f.config.Tenancy); tenancyMgr.Enabled {
+		unaryInterceptors = append(unaryInterceptors, tenancy.NewClientUnaryInterceptor(tenancyMgr))
+		streamInterceptors = append(streamInterceptors, tenancy.NewClientStreamInterceptor(tenancyMgr))
+	}
+
+	baseOpts := []grpc.DialOption{
+		grpc.WithChainUnaryInterceptor(unaryInterceptors...),
+		grpc.WithChainStreamInterceptor(streamInterceptors...),
+	}
+
+	createConn := func(telset component.TelemetrySettings) (*grpc.ClientConn, error) {
+		opts := append(baseOpts, grpc.WithStatsHandler(
+			otelgrpc.NewClientHandler(otelgrpc.WithTracerProvider(telset.TracerProvider)),
+		))
+		return newClient(telset, opts...)
+	}
+
+	readerConn, err := createConn(readerTelset)
+	if err != nil {
+		return fmt.Errorf("error creating reader client connection: %w", err)
+	}
+	writerConn, err := createConn(writerTelset)
+	if err != nil {
+		return fmt.Errorf("error creating writer client connection: %w", err)
+	}
+
+	f.readerConn, f.writerConn = readerConn, writerConn
+
+	return nil
+}

internal/storage/v2/grpc/factory_test.go

@@ -0,0 +1,101 @@
+// Copyright (c) 2025 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package grpc
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configauth"
+	"go.opentelemetry.io/collector/config/configgrpc"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+	"google.golang.org/grpc"
+
+	"github.com/jaegertracing/jaeger/internal/telemetry"
+	"github.com/jaegertracing/jaeger/internal/tenancy"
+)
+
+func TestNewFactory_NonEmptyAuthenticator(t *testing.T) {
+	cfg := &Config{
+		ClientConfig: configgrpc.ClientConfig{
+			Auth: &configauth.Authentication{},
+		},
+	}
+	_, err := NewFactory(*cfg, telemetry.NoopSettings())
+	require.ErrorContains(t, err, "authenticator is not supported")
+}
+
+func TestNewFactory(t *testing.T) {
+	lis, err := net.Listen("tcp", ":0")
+	require.NoError(t, err, "failed to listen")
+
+	s := grpc.NewServer()
+
+	startServer(t, s, lis)
+
+	cfg := Config{
+		ClientConfig: configgrpc.ClientConfig{
+			Endpoint: lis.Addr().String(),
+		},
+		TimeoutConfig: exporterhelper.TimeoutConfig{
+			Timeout: 1 * time.Second,
+		},
+		Tenancy: tenancy.Options{
+			Enabled: true,
+		},
+	}
+	telset := telemetry.NoopSettings()
+	f, err := NewFactory(cfg, telset)
+	require.NoError(t, err)
+	require.NoError(t, f.Close())
+}
+
+func TestFactory(t *testing.T) {
+	lis, err := net.Listen("tcp", ":0")
+	require.NoError(t, err, "failed to listen")
+
+	s := grpc.NewServer()
+
+	conn := startServer(t, s, lis)
+	f := &Factory{
+		readerConn: conn,
+	}
+
+	t.Run("CreateTraceReader", func(t *testing.T) {
+		tr, err := f.CreateTraceReader()
+		require.NoError(t, err)
+		require.NotNil(t, tr)
+	})
+
+	t.Run("CreateTraceWriter", func(t *testing.T) {
+		tr, err := f.CreateTraceWriter()
+		require.NoError(t, err)
+		require.NotNil(t, tr)
+	})
+
+	t.Run("CreateDependencyReader", func(t *testing.T) {
+		tr, err := f.CreateDependencyReader()
+		require.NoError(t, err)
+		require.NotNil(t, tr)
+	})
+}
+
+func TestInitializeConnections_ClientError(t *testing.T) {
+	f, err := NewFactory(Config{
+		ClientConfig: configgrpc.ClientConfig{
+			Endpoint: ":0",
+		},
+	}, telemetry.NoopSettings())
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, f.Close()) })
+	newClientFn := func(_ component.TelemetrySettings, _ ...grpc.DialOption) (conn *grpc.ClientConn, err error) {
+		return nil, assert.AnError
+	}
+	err = f.initializeConnections(component.TelemetrySettings{}, component.TelemetrySettings{}, newClientFn)
+	assert.ErrorContains(t, err, "error creating reader client connection")
+}