[storage] Remove distinction between primary and archive storage interfaces

cmd/all-in-one/main.go

@@ -171,8 +171,8 @@ by default uses only in-memory database.`,
 			queryTelset := baseTelset // copy
 			queryTelset.Metrics = queryMetricsFactory
 			querySrv := startQuery(
-				svc, qOpts, qOpts.BuildQueryServiceOptions(storageFactory, logger),
-				qOpts.BuildQueryServiceOptionsV2(storageFactory, logger),
+				svc, qOpts, qOpts.BuildQueryServiceOptions(storageFactory.InitArchiveStorage, logger),
+				qOpts.BuildQueryServiceOptionsV2(storageFactory.InitArchiveStorage, logger),
 				traceReader, dependencyReader, metricsQueryService,
 				tm, queryTelset,
 			)

cmd/jaeger/config-remote-storage.yaml

@@ -24,6 +24,7 @@ extensions:
   jaeger_query:
     storage:
       traces: some-storage
+      traces_archive: another-storage
     ui:
       config_file: ./cmd/jaeger/config-ui.json
 
@@ -34,6 +35,11 @@ extensions:
           endpoint: localhost:17271
           tls:
             insecure: true
+      another-storage:
+        grpc:
+          endpoint: localhost:17272
+          tls:
+            insecure: true  
 
 receivers:
   otlp:

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

@@ -74,7 +74,7 @@ func (cfg *TraceBackend) Unmarshal(conf *confmap.Conf) error {
 	}
 	if conf.IsSet("cassandra") {
 		cfg.Cassandra = &cassandra.Options{
-			Primary: cassandra.NamespaceConfig{
+			NamespaceConfig: cassandra.NamespaceConfig{
 				Configuration: casCfg.DefaultConfiguration(),
 				Enabled:       true,
 			},

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

@@ -83,7 +83,7 @@ backends:
 `)
 	cfg := createDefaultConfig().(*Config)
 	require.NoError(t, conf.Unmarshal(cfg))
-	assert.NotEmpty(t, cfg.TraceBackends["some_storage"].Cassandra.Primary.Connection.Servers)
+	assert.NotEmpty(t, cfg.TraceBackends["some_storage"].Cassandra.Configuration.Connection.Servers)
 }
 
 func TestConfigDefaultElasticsearch(t *testing.T) {

cmd/jaeger/internal/integration/grpc_test.go

@@ -7,20 +7,24 @@ import (
 	"testing"
 
 	"github.com/jaegertracing/jaeger/plugin/storage/integration"
+	"github.com/jaegertracing/jaeger/ports"
 )
 
 type GRPCStorageIntegration struct {
 	E2EStorageIntegration
 
-	remoteStorage *integration.RemoteMemoryStorage
+	remoteStorage        *integration.RemoteMemoryStorage
+	archiveRemoteStorage *integration.RemoteMemoryStorage
 }
 
 func (s *GRPCStorageIntegration) initialize(t *testing.T) {
-	s.remoteStorage = integration.StartNewRemoteMemoryStorage(t)
+	s.remoteStorage = integration.StartNewRemoteMemoryStorage(t, ports.RemoteStorageGRPC)
+	s.archiveRemoteStorage = integration.StartNewRemoteMemoryStorage(t, ports.RemoteStorageGRPC+1)
 }
 
 func (s *GRPCStorageIntegration) cleanUp(t *testing.T) {
 	s.remoteStorage.Close(t)
+	s.archiveRemoteStorage.Close(t)
 	s.initialize(t)
 }
 
@@ -37,6 +41,7 @@ func TestGRPCStorage(t *testing.T) {
 	s.e2eInitialize(t, "grpc")
 	t.Cleanup(func() {
 		s.remoteStorage.Close(t)
+		s.archiveRemoteStorage.Close(t)
 	})
 	s.RunSpanStoreTests(t)
 }

cmd/query/app/flags.go

@@ -30,7 +30,7 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
 	"github.com/jaegertracing/jaeger/ports"
-	"github.com/jaegertracing/jaeger/storage"
+	"github.com/jaegertracing/jaeger/storage/spanstore"
 	"github.com/jaegertracing/jaeger/storage_v2/v1adapter"
 )
 
@@ -138,10 +138,19 @@ func (qOpts *QueryOptions) InitFromViper(v *viper.Viper, logger *zap.Logger) (*Q
 	return qOpts, nil
 }
 
+type InitArchiveStorageFn func(logger *zap.Logger) (spanstore.Reader, spanstore.Writer)
+
 // BuildQueryServiceOptions creates a QueryServiceOptions struct with appropriate adjusters and archive config
-func (qOpts *QueryOptions) BuildQueryServiceOptions(storageFactory storage.BaseFactory, logger *zap.Logger) *querysvc.QueryServiceOptions {
+func (qOpts *QueryOptions) BuildQueryServiceOptions(
+	initArchiveStorageFn InitArchiveStorageFn,
+	logger *zap.Logger,
+) *querysvc.QueryServiceOptions {
 	opts := &querysvc.QueryServiceOptions{}
-	if !opts.InitArchiveStorage(storageFactory, logger) {
+	ar, aw := initArchiveStorageFn(logger)
+	if ar != nil && aw != nil {
+		opts.ArchiveSpanReader = ar
+		opts.ArchiveSpanWriter = aw
+	} else {
 		logger.Info("Archive storage not initialized")
 	}
 
@@ -150,13 +159,13 @@ func (qOpts *QueryOptions) BuildQueryServiceOptions(storageFactory storage.BaseF
 	return opts
 }
 
-func (qOpts *QueryOptions) BuildQueryServiceOptionsV2(storageFactory storage.BaseFactory, logger *zap.Logger) *v2querysvc.QueryServiceOptions {
+func (qOpts *QueryOptions) BuildQueryServiceOptionsV2(initArchiveStorageFn InitArchiveStorageFn, logger *zap.Logger) *v2querysvc.QueryServiceOptions {
 	opts := &v2querysvc.QueryServiceOptions{}
 
-	ar, aw := v1adapter.InitializeArchiveStorage(storageFactory, logger)
+	ar, aw := initArchiveStorageFn(logger)
 	if ar != nil && aw != nil {
-		opts.ArchiveTraceReader = ar
-		opts.ArchiveTraceWriter = aw
+		opts.ArchiveTraceReader = v1adapter.NewTraceReader(ar)
+		opts.ArchiveTraceWriter = v1adapter.NewTraceWriter(aw)
 	} else {
 		logger.Info("Archive storage not initialized")
 	}

cmd/query/app/flags_test.go

@@ -17,8 +17,8 @@ import (
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/testutils"
 	"github.com/jaegertracing/jaeger/ports"
-	"github.com/jaegertracing/jaeger/storage/mocks"
-	spanstore_mocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
+	"github.com/jaegertracing/jaeger/storage/spanstore"
+	spanstoremocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
 )
 
 func TestQueryBuilderFlags(t *testing.T) {
@@ -86,60 +86,51 @@ func TestStringSliceAsHeader(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func initializedFn(*zap.Logger) (spanstore.Reader, spanstore.Writer) {
+	return &spanstoremocks.Reader{}, &spanstoremocks.Writer{}
+}
+
+func uninitializedFn(*zap.Logger) (spanstore.Reader, spanstore.Writer) {
+	return nil, nil
+}
+
 func TestBuildQueryServiceOptions(t *testing.T) {
 	v, _ := config.Viperize(AddFlags)
 	qOpts, err := new(QueryOptions).InitFromViper(v, zap.NewNop())
 	require.NoError(t, err)
 	assert.NotNil(t, qOpts)
 
-	qSvcOpts := qOpts.BuildQueryServiceOptions(&mocks.Factory{}, zap.NewNop())
-	assert.NotNil(t, qSvcOpts)
-	assert.NotNil(t, qSvcOpts.Adjuster)
-	assert.Nil(t, qSvcOpts.ArchiveSpanReader)
-	assert.Nil(t, qSvcOpts.ArchiveSpanWriter)
-
-	comboFactory := struct {
-		*mocks.Factory
-		*mocks.ArchiveFactory
-	}{
-		&mocks.Factory{},
-		&mocks.ArchiveFactory{},
-	}
-
-	comboFactory.ArchiveFactory.On("CreateArchiveSpanReader").Return(&spanstore_mocks.Reader{}, nil)
-	comboFactory.ArchiveFactory.On("CreateArchiveSpanWriter").Return(&spanstore_mocks.Writer{}, nil)
-
-	qSvcOpts = qOpts.BuildQueryServiceOptions(comboFactory, zap.NewNop())
+	qSvcOpts := qOpts.BuildQueryServiceOptions(initializedFn, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
 	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.NotNil(t, qSvcOpts.ArchiveSpanReader)
 	assert.NotNil(t, qSvcOpts.ArchiveSpanWriter)
 }
 
-func TestBuildQueryServiceOptionsV2(t *testing.T) {
+func TestBuildQueryServiceOptions_NoArchiveStorage(t *testing.T) {
 	v, _ := config.Viperize(AddFlags)
 	qOpts, err := new(QueryOptions).InitFromViper(v, zap.NewNop())
 	require.NoError(t, err)
 	assert.NotNil(t, qOpts)
 
-	qSvcOpts := qOpts.BuildQueryServiceOptionsV2(&mocks.Factory{}, zap.NewNop())
+	logger, logBuf := testutils.NewLogger()
+	qSvcOpts := qOpts.BuildQueryServiceOptions(uninitializedFn, logger)
 	assert.NotNil(t, qSvcOpts)
 	assert.NotNil(t, qSvcOpts.Adjuster)
-	assert.Nil(t, qSvcOpts.ArchiveTraceReader)
-	assert.Nil(t, qSvcOpts.ArchiveTraceWriter)
+	assert.Nil(t, qSvcOpts.ArchiveSpanReader)
+	assert.Nil(t, qSvcOpts.ArchiveSpanWriter)
 
-	comboFactory := struct {
-		*mocks.Factory
-		*mocks.ArchiveFactory
-	}{
-		&mocks.Factory{},
-		&mocks.ArchiveFactory{},
-	}
+	require.Contains(t, logBuf.String(), "Archive storage not initialized")
+}
+
+func TestBuildQueryServiceOptionsV2(t *testing.T) {
+	v, _ := config.Viperize(AddFlags)
+	qOpts, err := new(QueryOptions).InitFromViper(v, zap.NewNop())
+	require.NoError(t, err)
+	assert.NotNil(t, qOpts)
 
-	comboFactory.ArchiveFactory.On("CreateArchiveSpanReader").Return(&spanstore_mocks.Reader{}, nil)
-	comboFactory.ArchiveFactory.On("CreateArchiveSpanWriter").Return(&spanstore_mocks.Writer{}, nil)
+	qSvcOpts := qOpts.BuildQueryServiceOptionsV2(initializedFn, zap.NewNop())
 
-	qSvcOpts = qOpts.BuildQueryServiceOptionsV2(comboFactory, zap.NewNop())
 	assert.NotNil(t, qSvcOpts)
 	assert.NotNil(t, qSvcOpts.Adjuster)
 	assert.NotNil(t, qSvcOpts.ArchiveTraceReader)
@@ -153,7 +144,7 @@ func TestBuildQueryServiceOptionsV2_NoArchiveStorage(t *testing.T) {
 	assert.NotNil(t, qOpts)
 
 	logger, logBuf := testutils.NewLogger()
-	qSvcOpts := qOpts.BuildQueryServiceOptionsV2(&mocks.Factory{}, logger)
+	qSvcOpts := qOpts.BuildQueryServiceOptionsV2(uninitializedFn, logger)
 	assert.Nil(t, qSvcOpts.ArchiveTraceReader)
 	assert.Nil(t, qSvcOpts.ArchiveTraceWriter)
 

cmd/query/app/querysvc/query_service.go

@@ -8,11 +8,8 @@ import (
 	"errors"
 	"time"
 
-	"go.uber.org/zap"
-
 	"github.com/jaegertracing/jaeger-idl/model/v1"
 	"github.com/jaegertracing/jaeger/model/adjuster"
-	"github.com/jaegertracing/jaeger/storage"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	"github.com/jaegertracing/jaeger/storage_v2/depstore"
 	"github.com/jaegertracing/jaeger/storage_v2/tracestore"
@@ -164,36 +161,6 @@ func (qs QueryService) GetCapabilities() StorageCapabilities {
 	}
 }
 
-// InitArchiveStorage tries to initialize archive storage reader/writer if storage factory supports them.
-func (opts *QueryServiceOptions) InitArchiveStorage(storageFactory storage.BaseFactory, logger *zap.Logger) bool {
-	archiveFactory, ok := storageFactory.(storage.ArchiveFactory)
-	if !ok {
-		logger.Info("Archive storage not supported by the factory")
-		return false
-	}
-	reader, err := archiveFactory.CreateArchiveSpanReader()
-	if errors.Is(err, storage.ErrArchiveStorageNotConfigured) || errors.Is(err, storage.ErrArchiveStorageNotSupported) {
-		logger.Info("Archive storage not created", zap.String("reason", err.Error()))
-		return false
-	}
-	if err != nil {
-		logger.Error("Cannot init archive storage reader", zap.Error(err))
-		return false
-	}
-	writer, err := archiveFactory.CreateArchiveSpanWriter()
-	if errors.Is(err, storage.ErrArchiveStorageNotConfigured) || errors.Is(err, storage.ErrArchiveStorageNotSupported) {
-		logger.Info("Archive storage not created", zap.String("reason", err.Error()))
-		return false
-	}
-	if err != nil {
-		logger.Error("Cannot init archive storage writer", zap.Error(err))
-		return false
-	}
-	opts.ArchiveSpanReader = reader
-	opts.ArchiveSpanWriter = writer
-	return true
-}
-
 // hasArchiveStorage returns true if archive storage reader/writer are initialized.
 func (opts *QueryServiceOptions) hasArchiveStorage() bool {
 	return opts.ArchiveSpanReader != nil && opts.ArchiveSpanWriter != nil

cmd/query/app/querysvc/query_service_test.go

@@ -451,64 +451,14 @@ func TestGetCapabilitiesWithSupportsArchive(t *testing.T) {
 
 type fakeStorageFactory1 struct{}
 
-type fakeStorageFactory2 struct {
-	fakeStorageFactory1
-	r    spanstore.Reader
-	w    spanstore.Writer
-	rErr error
-	wErr error
-}
-
 func (*fakeStorageFactory1) Initialize(metrics.Factory, *zap.Logger) error {
 	return nil
 }
 func (*fakeStorageFactory1) CreateSpanReader() (spanstore.Reader, error)             { return nil, nil }
 func (*fakeStorageFactory1) CreateSpanWriter() (spanstore.Writer, error)             { return nil, nil }
 func (*fakeStorageFactory1) CreateDependencyReader() (dependencystore.Reader, error) { return nil, nil }
 
-func (f *fakeStorageFactory2) CreateArchiveSpanReader() (spanstore.Reader, error) { return f.r, f.rErr }
-func (f *fakeStorageFactory2) CreateArchiveSpanWriter() (spanstore.Writer, error) { return f.w, f.wErr }
-
-var (
-	_ storage.Factory        = new(fakeStorageFactory1)
-	_ storage.ArchiveFactory = new(fakeStorageFactory2)
-)
-
-func TestInitArchiveStorageErrors(t *testing.T) {
-	opts := &QueryServiceOptions{}
-	logger := zap.NewNop()
-
-	assert.False(t, opts.InitArchiveStorage(new(fakeStorageFactory1), logger))
-	assert.False(t, opts.InitArchiveStorage(
-		&fakeStorageFactory2{rErr: storage.ErrArchiveStorageNotConfigured},
-		logger,
-	))
-	assert.False(t, opts.InitArchiveStorage(
-		&fakeStorageFactory2{rErr: errors.New("error")},
-		logger,
-	))
-	assert.False(t, opts.InitArchiveStorage(
-		&fakeStorageFactory2{wErr: storage.ErrArchiveStorageNotConfigured},
-		logger,
-	))
-	assert.False(t, opts.InitArchiveStorage(
-		&fakeStorageFactory2{wErr: errors.New("error")},
-		logger,
-	))
-}
-
-func TestInitArchiveStorage(t *testing.T) {
-	opts := &QueryServiceOptions{}
-	logger := zap.NewNop()
-	reader := &spanstoremocks.Reader{}
-	writer := &spanstoremocks.Writer{}
-	assert.True(t, opts.InitArchiveStorage(
-		&fakeStorageFactory2{r: reader, w: writer},
-		logger,
-	))
-	assert.Equal(t, reader, opts.ArchiveSpanReader)
-	assert.Equal(t, writer, opts.ArchiveSpanWriter)
-}
+var _ storage.Factory = new(fakeStorageFactory1)
 
 func TestMain(m *testing.M) {
 	testutils.VerifyGoLeaks(m)

cmd/query/app/token_propagation_test.go

@@ -81,7 +81,7 @@ func runQueryService(t *testing.T, esURL string) *Server {
 	}))
 	f.InitFromViper(v, flagsSvc.Logger)
 	// set AllowTokenFromContext manually because we don't register the respective CLI flag from query svc
-	f.Options.Primary.Authentication.BearerTokenAuthentication.AllowFromContext = true
+	f.Options.Config.Authentication.BearerTokenAuthentication.AllowFromContext = true
 	require.NoError(t, f.Initialize(telset.Metrics, telset.Logger))
 	defer f.Close()
 

cmd/query/main.go

@@ -107,13 +107,19 @@ func main() {
 			if err != nil {
 				logger.Fatal("Failed to create metrics query service", zap.Error(err))
 			}
-			queryServiceOptions := queryOpts.BuildQueryServiceOptions(storageFactory, logger)
+			queryServiceOptions := queryOpts.BuildQueryServiceOptions(
+				storageFactory.InitArchiveStorage,
+				logger,
+			)
 			queryService := querysvc.NewQueryService(
 				traceReader,
 				dependencyReader,
 				*queryServiceOptions)
 
-			queryServiceOptionsV2 := queryOpts.BuildQueryServiceOptionsV2(storageFactory, logger)
+			queryServiceOptionsV2 := queryOpts.BuildQueryServiceOptionsV2(
+				storageFactory.InitArchiveStorage,
+				logger,
+			)
 			queryServiceV2 := querysvcv2.NewQueryService(
 				traceReader,
 				dependencyReader,