Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[refactor] Remove archive reader and writer from remote storage grpc handler #6611

Merged
merged 6 commits into from
Jan 26, 2025
+88 −288
Merged

[refactor] Remove archive reader and writer from remote storage grpc handler #6611

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
498fe0b
Remove Archive Reader / Writer From GRPC Handler
mahadzaryab1 Jan 26, 2025
803f2fe
Deprecate Archive Fields In Capabilities Response
mahadzaryab1 Jan 26, 2025
5706e70
Remove GetArchiveTrace And WriteArchiveSpan
mahadzaryab1 Jan 26, 2025
eb6cf62
Regenerate Proto Files
mahadzaryab1 Jan 26, 2025
345a0ce
Merge branch 'main' into grpc-archive-cleanup
mahadzaryab1 Jan 26, 2025
e66496d
Merge branch 'main' into grpc-archive-cleanup
mahadzaryab1 Jan 26, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Remove GetArchiveTrace And WriteArchiveSpan 
Signed-off-by: Mahad Zaryab <mahadzaryab1@gmail.com>
@mahadzaryab1
mahadzaryab1 committed Jan 26, 2025

Verified

This commit was created on GitHub.com and signed with GitHub’s verified signature. The key has expired.
GPG key ID: 4AEE18F83AFDEB23
Expired
Verified
Learn about vigilant mode
commit 5706e7082e919099a74b23cb5119ffff3824a449
10 changes: 1 addition & 9 deletions cmd/remote-storage/app/server_test.go
View file
Original file line number Diff line number Diff line change
@@ -160,7 +160,7 @@ func TestCreateGRPCHandler(t *testing.T) {
depReader: depReader,
}

h, err := createGRPCHandler(f, zap.NewNop())
h, err := createGRPCHandler(f)
require.NoError(t, err)

writer.On("WriteSpan", mock.Anything, mock.Anything).Return(errors.New("writer error"))
@@ -176,12 +176,6 @@ func TestCreateGRPCHandler(t *testing.T) {
_, err = h.GetDependencies(context.Background(), &storage_v1.GetDependenciesRequest{})
require.ErrorContains(t, err, "deps error")

err = h.GetArchiveTrace(nil, nil)
require.ErrorContains(t, err, "not implemented")

_, err = h.WriteArchiveSpan(context.Background(), nil)
require.ErrorContains(t, err, "not implemented")

err = h.WriteSpanStream(nil)
assert.ErrorContains(t, err, "not implemented")
}
@@ -449,8 +443,6 @@ func validateGRPCServer(t *testing.T, hostPort string) {
"jaeger.storage.v1.SpanWriterPlugin",
"jaeger.storage.v1.DependenciesReaderPlugin",
"jaeger.storage.v1.PluginCapabilities",
"jaeger.storage.v1.ArchiveSpanReaderPlugin",
"jaeger.storage.v1.ArchiveSpanWriterPlugin",
"jaeger.storage.v1.StreamingSpanWriterPlugin",
"grpc.health.v1.Health",
},
12 changes: 0 additions & 12 deletions plugin/storage/grpc/shared/grpc_handler.go
View file
Original file line number Diff line number Diff line change
@@ -50,16 +50,12 @@ func NewGRPCHandler(impl *GRPCHandlerStorageImpl) *GRPCHandler {
func (s *GRPCHandler) Register(ss *grpc.Server, hs *health.Server) error {
storage_v1.RegisterSpanReaderPluginServer(ss, s)
storage_v1.RegisterSpanWriterPluginServer(ss, s)
storage_v1.RegisterArchiveSpanReaderPluginServer(ss, s)
storage_v1.RegisterArchiveSpanWriterPluginServer(ss, s)
storage_v1.RegisterPluginCapabilitiesServer(ss, s)
storage_v1.RegisterDependenciesReaderPluginServer(ss, s)
storage_v1.RegisterStreamingSpanWriterPluginServer(ss, s)

hs.SetServingStatus("jaeger.storage.v1.SpanReaderPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.SpanWriterPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.ArchiveSpanReaderPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.ArchiveSpanWriterPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.PluginCapabilities", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.DependenciesReaderPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
hs.SetServingStatus("jaeger.storage.v1.StreamingSpanWriterPlugin", grpc_health_v1.HealthCheckResponse_SERVING)
@@ -246,11 +242,3 @@ func (s *GRPCHandler) Capabilities(context.Context, *storage_v1.CapabilitiesRequ
StreamingSpanWriter: s.impl.StreamingSpanWriter() != nil,
}, nil
}

func (s *GRPCHandler) GetArchiveTrace(r *storage_v1.GetTraceRequest, stream storage_v1.ArchiveSpanReaderPlugin_GetArchiveTraceServer) error {
return s.GetTrace(r, stream)
}

func (s *GRPCHandler) WriteArchiveSpan(ctx context.Context, r *storage_v1.WriteSpanRequest) (*storage_v1.WriteSpanResponse, error) {
return s.WriteSpan(ctx, r)
}
152 changes: 2 additions & 150 deletions plugin/storage/grpc/shared/grpc_handler_test.go
View file
Original file line number Diff line number Diff line change
@@ -76,12 +76,6 @@ func withGRPCServer(fn func(r *grpcServerTest)) {
DependencyReader: func() dependencystore.Reader {
return mockPlugin.depsReader
},
ArchiveSpanReader: func() spanstore.Reader {
return mockPlugin.spanReader
},
ArchiveSpanWriter: func() spanstore.Writer {
return mockPlugin.spanWriter
},
StreamingSpanWriter: func() spanstore.Writer {
return mockPlugin.streamWriter
},
@@ -280,150 +274,11 @@ func TestGRPCServerGetDependencies(t *testing.T) {
})
}

func TestGRPCServerGetArchiveTrace(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
traceSteam := new(grpcMocks.SpanReaderPlugin_GetTraceServer)
traceSteam.On("Context").Return(context.Background())
traceSteam.On("Send", &storage_v1.SpansResponseChunk{Spans: mockTraceSpans}).
Return(nil)

var traceSpans []*model.Span
for i := range mockTraceSpans {
traceSpans = append(traceSpans, &mockTraceSpans[i])
}
r.impl.spanReader.On("GetTrace", mock.Anything, spanstore.GetTraceParameters{TraceID: mockTraceID}).
Return(&model.Trace{Spans: traceSpans}, nil)

err := r.server.GetArchiveTrace(&storage_v1.GetTraceRequest{
TraceID: mockTraceID,
}, traceSteam)
require.NoError(t, err)
})
}

func TestGRPCServerGetArchiveTrace_NotFound(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
traceSteam := new(grpcMocks.SpanReaderPlugin_GetTraceServer)
traceSteam.On("Context").Return(context.Background())

r.impl.spanReader.On("GetTrace", mock.Anything, spanstore.GetTraceParameters{TraceID: mockTraceID}).
Return(nil, spanstore.ErrTraceNotFound)

err := r.server.GetArchiveTrace(&storage_v1.GetTraceRequest{
TraceID: mockTraceID,
}, traceSteam)
assert.Equal(t, codes.NotFound, status.Code(err))
})
}

func TestGRPCServerGetArchiveTrace_Error(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
traceSteam := new(grpcMocks.SpanReaderPlugin_GetTraceServer)
traceSteam.On("Context").Return(context.Background())

r.impl.spanReader.On("GetTrace", mock.Anything, spanstore.GetTraceParameters{TraceID: mockTraceID}).
Return(nil, errors.New("some error"))

err := r.server.GetArchiveTrace(&storage_v1.GetTraceRequest{
TraceID: mockTraceID,
}, traceSteam)
require.Error(t, err)
})
}

func TestGRPCServerGetArchiveTrace_NoImpl(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
r.server.impl.ArchiveSpanReader = func() spanstore.Reader { return nil }
traceSteam := new(grpcMocks.SpanReaderPlugin_GetTraceServer)

r.impl.spanReader.On("GetTrace", mock.Anything, spanstore.GetTraceParameters{TraceID: mockTraceID}).
Return(nil, errors.New("some error"))

err := r.server.GetArchiveTrace(&storage_v1.GetTraceRequest{
TraceID: mockTraceID,
}, traceSteam)
assert.Equal(t, codes.Unimplemented, status.Code(err))
})
}

func TestGRPCServerGetArchiveTrace_StreamError(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
traceSteam := new(grpcMocks.SpanReaderPlugin_GetTraceServer)
traceSteam.On("Context").Return(context.Background())
traceSteam.On("Send", &storage_v1.SpansResponseChunk{Spans: mockTraceSpans}).
Return(errors.New("some error"))

var traceSpans []*model.Span
for i := range mockTraceSpans {
traceSpans = append(traceSpans, &mockTraceSpans[i])
}
r.impl.spanReader.On("GetTrace", mock.Anything, spanstore.GetTraceParameters{TraceID: mockTraceID}).
Return(&model.Trace{Spans: traceSpans}, nil)

err := r.server.GetArchiveTrace(&storage_v1.GetTraceRequest{
TraceID: mockTraceID,
}, traceSteam)
require.Error(t, err)
})
}

func TestGRPCServerWriteArchiveSpan_NoImpl(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
r.server.impl.ArchiveSpanWriter = func() spanstore.Writer { return nil }

_, err := r.server.WriteArchiveSpan(context.Background(), &storage_v1.WriteSpanRequest{
Span: &mockTraceSpans[0],
})
assert.Equal(t, codes.Unimplemented, status.Code(err))
})
}

func TestGRPCServerWriteArchiveSpan(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
r.impl.spanWriter.On("WriteSpan", mock.Anything, &mockTraceSpans[0]).
Return(nil)

s, err := r.server.WriteArchiveSpan(context.Background(), &storage_v1.WriteSpanRequest{
Span: &mockTraceSpans[0],
})
require.NoError(t, err)
assert.Equal(t, &storage_v1.WriteSpanResponse{}, s)
})
}

func TestGRPCServerWriteArchiveSpan_Error(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
r.impl.spanWriter.On("WriteSpan", mock.Anything, &mockTraceSpans[0]).
Return(errors.New("some error"))

_, err := r.server.WriteArchiveSpan(context.Background(), &storage_v1.WriteSpanRequest{
Span: &mockTraceSpans[0],
})
require.Error(t, err)
})
}

func TestGRPCServerCapabilities(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
capabilities, err := r.server.Capabilities(context.Background(), &storage_v1.CapabilitiesRequest{})
require.NoError(t, err)
assert.Equal(t, &storage_v1.CapabilitiesResponse{ArchiveSpanReader: true, ArchiveSpanWriter: true, StreamingSpanWriter: true}, capabilities)
})
}

func TestGRPCServerCapabilities_NoArchive(t *testing.T) {
withGRPCServer(func(r *grpcServerTest) {
r.server.impl.ArchiveSpanReader = func() spanstore.Reader { return nil }
r.server.impl.ArchiveSpanWriter = func() spanstore.Writer { return nil }

capabilities, err := r.server.Capabilities(context.Background(), &storage_v1.CapabilitiesRequest{})
require.NoError(t, err)
expected := &storage_v1.CapabilitiesResponse{
ArchiveSpanReader: false,
ArchiveSpanWriter: false,
StreamingSpanWriter: true,
}
assert.Equal(t, expected, capabilities)
assert.Equal(t, &storage_v1.CapabilitiesResponse{ArchiveSpanReader: false, ArchiveSpanWriter: false, StreamingSpanWriter: true}, capabilities)
})
}

@@ -433,10 +288,7 @@ func TestGRPCServerCapabilities_NoStreamWriter(t *testing.T) {

capabilities, err := r.server.Capabilities(context.Background(), &storage_v1.CapabilitiesRequest{})
require.NoError(t, err)
expected := &storage_v1.CapabilitiesResponse{
ArchiveSpanReader: true,
ArchiveSpanWriter: true,
}
expected := &storage_v1.CapabilitiesResponse{}
assert.Equal(t, expected, capabilities)
})
}