Skip to content

Commit

Permalink
Merge branch 'main' into cassandra_init
Browse files Browse the repository at this point in the history
  • Loading branch information
@hellspawn679
hellspawn679 authored Oct 11, 2024
2 parents 7c05fdc + 3551cf7 commit 3aa0941
Show file tree
Hide file tree
Showing 8 changed files with 148 additions and 34 deletions.
8 changes: 4 additions & 4 deletions plugin/storage/cassandra/factory.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down
10 changes: 8 additions & 2 deletions plugin/storage/cassandra/savetracetest/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down
8 changes: 6 additions & 2 deletions plugin/storage/cassandra/spanstore/operation_names.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Expand All @@ -113,7 +117,7 @@ func NewOperationNamesStorage(
TTL: writeCacheTTL,
InitialCapacity: 10000,
}),
}
}, nil
}

// Write saves Operation and Service name tuples
Expand Down
58 changes: 49 additions & 9 deletions plugin/storage/cassandra/spanstore/operation_names_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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{}
Expand Down Expand Up @@ -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) {
Expand Down
9 changes: 6 additions & 3 deletions plugin/storage/cassandra/spanstore/reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -127,7 +130,7 @@ func NewSpanReader(
},
logger: logger,
tracer: tracer,
}
}, nil
}

// GetServices returns all services traced by Jaeger
Expand Down
32 changes: 31 additions & 1 deletion plugin/storage/cassandra/spanstore/reader_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand Down
9 changes: 6 additions & 3 deletions plugin/storage/cassandra/spanstore/writer.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand All @@ -117,7 +120,7 @@ func NewSpanWriter(
tagFilter: opts.tagFilter,
storageMode: opts.storageMode,
indexFilter: opts.indexFilter,
}
}, nil
}

// Close closes SpanWriter
Expand Down
48 changes: 38 additions & 10 deletions plugin/storage/cassandra/spanstore/writer_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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{}
Expand All @@ -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)
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand All @@ -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{
Expand Down Expand Up @@ -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 }
Expand All @@ -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 := ""
Expand Down Expand Up @@ -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
Expand Down

0 comments on commit 3aa0941

Please sign in to comment.