[cassandra] Prevent fallback to old schema for operation names table in case of db issues

plugin/storage/cassandra/factory.go

@@ -157,7 +157,7 @@ func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger)
 
 // CreateSpanReader implements storage.Factory
 func (f *Factory) CreateSpanReader() (spanstore.Reader, error) {
-	return cSpanStore.NewSpanReader(f.primarySession, f.primaryMetricsFactory, f.logger, f.tracer.Tracer("cSpanStore.SpanReader")), nil
+	return cSpanStore.NewSpanReader(f.primarySession, f.primaryMetricsFactory, f.logger, f.tracer.Tracer("cSpanStore.SpanReader"))
 }
 
 // CreateSpanWriter implements storage.Factory
@@ -166,7 +166,7 @@ func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) {
 	if err != nil {
 		return nil, err
 	}
-	return cSpanStore.NewSpanWriter(f.primarySession, f.Options.SpanStoreWriteCacheTTL, f.primaryMetricsFactory, f.logger, options...), nil
+	return cSpanStore.NewSpanWriter(f.primarySession, f.Options.SpanStoreWriteCacheTTL, f.primaryMetricsFactory, f.logger, options...)
 }
 
 // CreateDependencyReader implements storage.Factory
@@ -180,7 +180,7 @@ func (f *Factory) CreateArchiveSpanReader() (spanstore.Reader, error) {
 	if f.archiveSession == nil {
 		return nil, storage.ErrArchiveStorageNotConfigured
 	}
-	return cSpanStore.NewSpanReader(f.archiveSession, f.archiveMetricsFactory, f.logger, f.tracer.Tracer("cSpanStore.SpanReader")), nil
+	return cSpanStore.NewSpanReader(f.archiveSession, f.archiveMetricsFactory, f.logger, f.tracer.Tracer("cSpanStore.SpanReader"))
 }
 
 // CreateArchiveSpanWriter implements storage.ArchiveFactory
@@ -192,7 +192,7 @@ func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 	if err != nil {
 		return nil, err
 	}
-	return cSpanStore.NewSpanWriter(f.archiveSession, f.Options.SpanStoreWriteCacheTTL, f.archiveMetricsFactory, f.logger, options...), nil
+	return cSpanStore.NewSpanWriter(f.archiveSession, f.Options.SpanStoreWriteCacheTTL, f.archiveMetricsFactory, f.logger, options...)
 }
 
 // CreateLock implements storage.SamplingStoreFactory

plugin/storage/cassandra/savetracetest/main.go

@@ -44,8 +44,14 @@ func main() {
 	if err != nil {
 		logger.Fatal("Failed to initialize tracer", zap.Error(err))
 	}
-	spanStore := cSpanStore.NewSpanWriter(cqlSession, time.Hour*12, noScope, logger)
-	spanReader := cSpanStore.NewSpanReader(cqlSession, noScope, logger, tracer.OTEL.Tracer("cSpanStore.SpanReader"))
+	spanStore, err := cSpanStore.NewSpanWriter(cqlSession, time.Hour*12, noScope, logger)
+	if err != nil {
+		logger.Fatal("Failed to create span writer", zap.Error(err))
+	}
+	spanReader, err := cSpanStore.NewSpanReader(cqlSession, noScope, logger, tracer.OTEL.Tracer("cSpanStore.SpanReader"))
+	if err != nil {
+		logger.Fatal("Failed to create span reader", zap.Error(err))
+	}
 	ctx := context.Background()
 	if err = spanStore.WriteSpan(ctx, getSomeSpan()); err != nil {
 		logger.Fatal("Failed to save", zap.Error(err))

plugin/storage/cassandra/spanstore/operation_names.go

@@ -92,9 +92,13 @@ func NewOperationNamesStorage(
 	writeCacheTTL time.Duration,
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
-) *OperationNamesStorage {
+) (*OperationNamesStorage, error) {
 	schemaVersion := latestVersion
 	if !tableExist(session, schemas[schemaVersion].tableName) {
+		if !tableExist(session, schemas[previousVersion].tableName) {
+			return nil, fmt.Errorf("neither table %s nor %s exist",
+				schemas[schemaVersion].tableName, schemas[previousVersion].tableName)
+		}
 		schemaVersion = previousVersion
 	}
 	table := schemas[schemaVersion]
@@ -113,7 +117,7 @@ func NewOperationNamesStorage(
 				TTL:             writeCacheTTL,
 				InitialCapacity: 10000,
 			}),
-	}
+	}, nil
 }
 
 // Write saves Operation and Service name tuples

plugin/storage/cassandra/spanstore/operation_names_test.go

@@ -31,33 +31,73 @@ type operationNameStorageTest struct {
 	storage        *OperationNamesStorage
 }
 
-func withOperationNamesStorage(writeCacheTTL time.Duration,
+func withOperationNamesStorage(t *testing.T,
+	writeCacheTTL time.Duration,
 	schemaVersion schemaVersion,
 	fn func(s *operationNameStorageTest),
 ) {
 	session := &mocks.Session{}
 	logger, logBuffer := testutils.NewLogger()
 	metricsFactory := metricstest.NewFactory(0)
-	query := &mocks.Query{}
+	latestTableCheckquery := &mocks.Query{}
 	session.On("Query",
-		fmt.Sprintf(tableCheckStmt, schemas[latestVersion].tableName), mock.Anything).Return(query)
-	if schemaVersion != latestVersion {
-		query.On("Exec").Return(errors.New("new table does not exist"))
+		fmt.Sprintf(tableCheckStmt, schemas[latestVersion].tableName), mock.Anything).Return(latestTableCheckquery)
+	if schemaVersion == latestVersion {
+		latestTableCheckquery.On("Exec").Return(nil)
 	} else {
-		query.On("Exec").Return(nil)
+		previousTableCheckquery := &mocks.Query{}
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[previousVersion].tableName), mock.Anything).Return(previousTableCheckquery)
+		latestTableCheckquery.On("Exec").Return(errors.New("table not found"))
+		previousTableCheckquery.On("Exec").Return(nil)
 	}
 
+	storage, err := NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger)
+	require.NoError(t, err)
+
 	s := &operationNameStorageTest{
 		session:        session,
 		writeCacheTTL:  writeCacheTTL,
 		metricsFactory: metricsFactory,
 		logger:         logger,
 		logBuffer:      logBuffer,
-		storage:        NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger),
+		storage:        storage,
 	}
 	fn(s)
 }
 
+func TestNewOperationNamesStorage(t *testing.T) {
+	t.Run("test operation names storage creation with old schema", func(t *testing.T) {
+		withOperationNamesStorage(t, 0, previousVersion, func(s *operationNameStorageTest) {
+			assert.NotNil(t, s.storage)
+		})
+	})
+
+	t.Run("test operation names storage creation with new schema", func(t *testing.T) {
+		withOperationNamesStorage(t, 0, latestVersion, func(s *operationNameStorageTest) {
+			assert.NotNil(t, s.storage)
+		})
+	})
+
+	t.Run("test operation names storage creation error", func(t *testing.T) {
+		session := &mocks.Session{}
+		logger, _ := testutils.NewLogger()
+		metricsFactory := metricstest.NewFactory(0)
+		query := &mocks.Query{}
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[latestVersion].tableName),
+			mock.Anything).Return(query)
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[previousVersion].tableName),
+			mock.Anything).Return(query)
+		query.On("Exec").Return(errors.New("table does not exist"))
+
+		_, err := NewOperationNamesStorage(session, 0, metricsFactory, logger)
+
+		require.EqualError(t, err, "neither table operation_names_v2 nor operation_names exist")
+	})
+}
+
 func TestOperationNamesStorageWrite(t *testing.T) {
 	for _, test := range []struct {
 		name          string
@@ -70,7 +110,7 @@ func TestOperationNamesStorageWrite(t *testing.T) {
 		{name: "test new schema with 1min ttl", ttl: time.Minute, schemaVersion: latestVersion},
 	} {
 		t.Run(test.name, func(t *testing.T) {
-			withOperationNamesStorage(test.ttl, test.schemaVersion, func(s *operationNameStorageTest) {
+			withOperationNamesStorage(t, test.ttl, test.schemaVersion, func(s *operationNameStorageTest) {
 				execError := errors.New("exec error")
 				query := &mocks.Query{}
 				query1 := &mocks.Query{}
@@ -160,7 +200,7 @@ func TestOperationNamesStorageGetServices(t *testing.T) {
 		{name: "test new schema with scan error", schemaVersion: latestVersion, expErr: scanError},
 	} {
 		t.Run(test.name, func(t *testing.T) {
-			withOperationNamesStorage(0, test.schemaVersion, func(s *operationNameStorageTest) {
+			withOperationNamesStorage(t, 0, test.schemaVersion, func(s *operationNameStorageTest) {
 				assignPtr := func(vals ...string) any {
 					return mock.MatchedBy(func(args []any) bool {
 						if len(args) != len(vals) {

plugin/storage/cassandra/spanstore/reader.go

@@ -109,10 +109,13 @@ func NewSpanReader(
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
 	tracer trace.Tracer,
-) *SpanReader {
+) (*SpanReader, error) {
 	readFactory := metricsFactory.Namespace(metrics.NSOptions{Name: "read", Tags: nil})
 	serviceNamesStorage := NewServiceNamesStorage(session, 0, metricsFactory, logger)
-	operationNamesStorage := NewOperationNamesStorage(session, 0, metricsFactory, logger)
+	operationNamesStorage, err := NewOperationNamesStorage(session, 0, metricsFactory, logger)
+	if err != nil {
+		return nil, err
+	}
 	return &SpanReader{
 		session:              session,
 		serviceNamesReader:   serviceNamesStorage.GetServices,
@@ -127,7 +130,7 @@ func NewSpanReader(
 		},
 		logger: logger,
 		tracer: tracer,
-	}
+	}, nil
 }
 
 // GetServices returns all services traced by Jaeger

plugin/storage/cassandra/spanstore/reader_test.go

@@ -59,18 +59,48 @@ func withSpanReader(t *testing.T, fn func(r *spanReaderTest)) {
 	metricsFactory := metricstest.NewFactory(0)
 	tracer, exp, closer := tracerProvider(t)
 	defer closer()
+	reader, err := NewSpanReader(session, metricsFactory, logger, tracer.Tracer("test"))
+	require.NoError(t, err)
 	r := &spanReaderTest{
 		session:     session,
 		logger:      logger,
 		logBuffer:   logBuffer,
 		traceBuffer: exp,
-		reader:      NewSpanReader(session, metricsFactory, logger, tracer.Tracer("test")),
+		reader:      reader,
 	}
 	fn(r)
 }
 
 var _ spanstore.Reader = &SpanReader{} // check API conformance
 
+func TestNewSpanReader(t *testing.T) {
+	t.Run("test span reader creation", func(t *testing.T) {
+		withSpanReader(t, func(r *spanReaderTest) {
+			assert.NotNil(t, r.reader)
+		})
+	})
+
+	t.Run("test span reader creation error", func(t *testing.T) {
+		session := &mocks.Session{}
+		query := &mocks.Query{}
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[latestVersion].tableName),
+			mock.Anything).Return(query)
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[previousVersion].tableName),
+			mock.Anything).Return(query)
+		query.On("Exec").Return(errors.New("table does not exist"))
+		logger, _ := testutils.NewLogger()
+		metricsFactory := metricstest.NewFactory(0)
+		tracer, _, closer := tracerProvider(t)
+		defer closer()
+
+		_, err := NewSpanReader(session, metricsFactory, logger, tracer.Tracer("test"))
+
+		require.EqualError(t, err, "neither table operation_names_v2 nor operation_names exist")
+	})
+}
+
 func TestSpanReaderGetServices(t *testing.T) {
 	withSpanReader(t, func(r *spanReaderTest) {
 		r.reader.serviceNamesReader = func() ([]string, error) { return []string{"service-a"}, nil }

plugin/storage/cassandra/spanstore/writer.go

@@ -96,9 +96,12 @@ func NewSpanWriter(
 	metricsFactory metrics.Factory,
 	logger *zap.Logger,
 	options ...Option,
-) *SpanWriter {
+) (*SpanWriter, error) {
 	serviceNamesStorage := NewServiceNamesStorage(session, writeCacheTTL, metricsFactory, logger)
-	operationNamesStorage := NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger)
+	operationNamesStorage, err := NewOperationNamesStorage(session, writeCacheTTL, metricsFactory, logger)
+	if err != nil {
+		return nil, err
+	}
 	tagIndexSkipped := metricsFactory.Counter(metrics.Options{Name: "tag_index_skipped", Tags: nil})
 	opts := applyOptions(options...)
 	return &SpanWriter{
@@ -117,7 +120,7 @@ func NewSpanWriter(
 		tagFilter:       opts.tagFilter,
 		storageMode:     opts.storageMode,
 		indexFilter:     opts.indexFilter,
-	}
+	}, nil
 }
 
 // Close closes SpanWriter

plugin/storage/cassandra/spanstore/writer_test.go

@@ -33,7 +33,7 @@ type spanWriterTest struct {
 	writer    *SpanWriter
 }
 
-func withSpanWriter(writeCacheTTL time.Duration, fn func(w *spanWriterTest), options ...Option,
+func withSpanWriter(t *testing.T, writeCacheTTL time.Duration, fn func(w *spanWriterTest), options ...Option,
 ) {
 	session := &mocks.Session{}
 	query := &mocks.Query{}
@@ -43,19 +43,47 @@ func withSpanWriter(writeCacheTTL time.Duration, fn func(w *spanWriterTest), opt
 	query.On("Exec").Return(nil)
 	logger, logBuffer := testutils.NewLogger()
 	metricsFactory := metricstest.NewFactory(0)
+	writer, err := NewSpanWriter(session, writeCacheTTL, metricsFactory, logger, options...)
+	require.NoError(t, err)
 	w := &spanWriterTest{
 		session:   session,
 		logger:    logger,
 		logBuffer: logBuffer,
-		writer:    NewSpanWriter(session, writeCacheTTL, metricsFactory, logger, options...),
+		writer:    writer,
 	}
 	fn(w)
 }
 
 var _ spanstore.Writer = &SpanWriter{} // check API conformance
 
+func TestNewSpanWriter(t *testing.T) {
+	t.Run("test span writer creation", func(t *testing.T) {
+		withSpanWriter(t, 0, func(w *spanWriterTest) {
+			assert.NotNil(t, w.writer)
+		})
+	})
+
+	t.Run("test span writer creation error", func(t *testing.T) {
+		session := &mocks.Session{}
+		query := &mocks.Query{}
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[latestVersion].tableName),
+			mock.Anything).Return(query)
+		session.On("Query",
+			fmt.Sprintf(tableCheckStmt, schemas[previousVersion].tableName),
+			mock.Anything).Return(query)
+		query.On("Exec").Return(errors.New("table does not exist"))
+		logger, _ := testutils.NewLogger()
+		metricsFactory := metricstest.NewFactory(0)
+
+		_, err := NewSpanWriter(session, 0, metricsFactory, logger)
+
+		assert.EqualError(t, err, "neither table operation_names_v2 nor operation_names exist")
+	})
+}
+
 func TestClientClose(t *testing.T) {
-	withSpanWriter(0, func(w *spanWriterTest) {
+	withSpanWriter(t, 0, func(w *spanWriterTest) {
 		w.session.On("Close").Return(nil)
 		w.writer.Close()
 		w.session.AssertNumberOfCalls(t, "Close", 1)
@@ -150,7 +178,7 @@ func TestSpanWriter(t *testing.T) {
 	for _, tc := range testCases {
 		testCase := tc // capture loop var
 		t.Run(testCase.caption, func(t *testing.T) {
-			withSpanWriter(0, func(w *spanWriterTest) {
+			withSpanWriter(t, 0, func(w *spanWriterTest) {
 				span := &model.Span{
 					TraceID:       model.NewTraceID(0, 1),
 					OperationName: "operation-a",
@@ -242,7 +270,7 @@ func TestSpanWriterSaveServiceNameAndOperationName(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		testCase := tc // capture loop var
-		withSpanWriter(0, func(w *spanWriterTest) {
+		withSpanWriter(t, 0, func(w *spanWriterTest) {
 			w.writer.serviceNamesWriter = testCase.serviceNamesWriter
 			w.writer.operationNamesWriter = testCase.operationNamesWriter
 			err := w.writer.saveServiceNameAndOperationName(
@@ -274,7 +302,7 @@ func TestSpanWriterSkippingTags(t *testing.T) {
 	}
 	for _, tc := range testCases {
 		testCase := tc // capture loop var
-		withSpanWriter(0, func(w *spanWriterTest) {
+		withSpanWriter(t, 0, func(w *spanWriterTest) {
 			db := dbmodel.TagInsertion{
 				ServiceName: "service-a",
 				TagKey:      testCase.key,
@@ -287,7 +315,7 @@ func TestSpanWriterSkippingTags(t *testing.T) {
 }
 
 func TestStorageMode_IndexOnly(t *testing.T) {
-	withSpanWriter(0, func(w *spanWriterTest) {
+	withSpanWriter(t, 0, func(w *spanWriterTest) {
 		w.writer.serviceNamesWriter = func(_ /* serviceName */ string) error { return nil }
 		w.writer.operationNamesWriter = func(_ dbmodel.Operation) error { return nil }
 		span := &model.Span{
@@ -329,7 +357,7 @@ var filterEverything = func(*dbmodel.Span, int) bool {
 }
 
 func TestStorageMode_IndexOnly_WithFilter(t *testing.T) {
-	withSpanWriter(0, func(w *spanWriterTest) {
+	withSpanWriter(t, 0, func(w *spanWriterTest) {
 		w.writer.indexFilter = filterEverything
 		w.writer.serviceNamesWriter = func(_ /* serviceName */ string) error { return nil }
 		w.writer.operationNamesWriter = func(_ dbmodel.Operation) error { return nil }
@@ -349,7 +377,7 @@ func TestStorageMode_IndexOnly_WithFilter(t *testing.T) {
 }
 
 func TestStorageMode_IndexOnly_FirehoseSpan(t *testing.T) {
-	withSpanWriter(0, func(w *spanWriterTest) {
+	withSpanWriter(t, 0, func(w *spanWriterTest) {
 		var serviceWritten atomic.Pointer[string]
 		var operationWritten atomic.Pointer[dbmodel.Operation]
 		empty := ""
@@ -401,7 +429,7 @@ func TestStorageMode_IndexOnly_FirehoseSpan(t *testing.T) {
 }
 
 func TestStorageMode_StoreWithoutIndexing(t *testing.T) {
-	withSpanWriter(0, func(w *spanWriterTest) {
+	withSpanWriter(t, 0, func(w *spanWriterTest) {
 		w.writer.serviceNamesWriter = func(_ /* serviceName */ string) error {
 			assert.Fail(t, "Non indexing store shouldn't index")
 			return nil