Skip to content

Commit

Permalink
Replace OpenTracing instrumentation with OpenTelemetry in grpc storag…
Browse files Browse the repository at this point in the history
…e plugin (#4611)

## Which problem is this PR solving?
* Part of #3381 
* This PR adds `otelgrpc` plugin to storage

## Short description of the changes
- Replaces `otgrpc` plugin storage with otel grpc interceptors

---------

Signed-off-by: Afzal <[email protected]>
Co-authored-by: Afzal <[email protected]>
  • Loading branch information
afzal442 and afzalbin64 authored Aug 4, 2023
1 parent 0d06c56 commit ab6732b
Show file tree
Hide file tree
Showing 4 changed files with 25 additions and 16 deletions.
4 changes: 2 additions & 2 deletions plugin/storage/grpc/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,8 +200,8 @@ grpc.ServeWithGRPCServer(&shared.PluginServices{
ArchiveStore: memStorePlugin,
}, func(options []googleGRPC.ServerOption) *googleGRPC.Server {
return plugin.DefaultGRPCServer([]googleGRPC.ServerOption{
googleGRPC.UnaryInterceptor(otgrpc.OpenTracingServerInterceptor(tracer)),
googleGRPC.StreamInterceptor(otgrpc.OpenTracingStreamServerInterceptor(tracer)),
grpc.UnaryInterceptor(otelgrpc.UnaryServerInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
grpc.StreamInterceptor(otelgrpc.StreamServerInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
})
})
```
Expand Down
28 changes: 16 additions & 12 deletions plugin/storage/grpc/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,10 @@ import (
"runtime"
"time"

"github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/go-plugin"
"github.com/opentracing/opentracing-go"
"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
Expand Down Expand Up @@ -60,16 +60,16 @@ type ClientPluginServices struct {

// PluginBuilder is used to create storage plugins. Implemented by Configuration.
type PluginBuilder interface {
Build(logger *zap.Logger) (*ClientPluginServices, error)
Build(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error)
Close() error
}

// Build instantiates a PluginServices
func (c *Configuration) Build(logger *zap.Logger) (*ClientPluginServices, error) {
func (c *Configuration) Build(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
if c.PluginBinary != "" {
return c.buildPlugin(logger)
return c.buildPlugin(logger, tracerProvider)
} else {
return c.buildRemote(logger)
return c.buildRemote(logger, tracerProvider)
}
}

Expand All @@ -82,10 +82,12 @@ func (c *Configuration) Close() error {
return c.RemoteTLS.Close()
}

func (c *Configuration) buildRemote(logger *zap.Logger) (*ClientPluginServices, error) {
func (c *Configuration) buildRemote(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
opts := []grpc.DialOption{
grpc.WithUnaryInterceptor(otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer())),
grpc.WithStreamInterceptor(otgrpc.OpenTracingStreamClientInterceptor(opentracing.GlobalTracer())),
grpc.WithUnaryInterceptor(
otelgrpc.UnaryClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
grpc.WithStreamInterceptor(
otelgrpc.StreamClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
grpc.WithBlock(),
}
if c.RemoteTLS.Enabled {
Expand Down Expand Up @@ -123,10 +125,12 @@ func (c *Configuration) buildRemote(logger *zap.Logger) (*ClientPluginServices,
}, nil
}

func (c *Configuration) buildPlugin(logger *zap.Logger) (*ClientPluginServices, error) {
func (c *Configuration) buildPlugin(logger *zap.Logger, tracerProvider trace.TracerProvider) (*ClientPluginServices, error) {
opts := []grpc.DialOption{
grpc.WithUnaryInterceptor(otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer())),
grpc.WithStreamInterceptor(otgrpc.OpenTracingStreamClientInterceptor(opentracing.GlobalTracer())),
grpc.WithUnaryInterceptor(
otelgrpc.UnaryClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
grpc.WithStreamInterceptor(
otelgrpc.StreamClientInterceptor(otelgrpc.WithTracerProvider(tracerProvider))),
}

tenancyMgr := tenancy.NewManager(&c.TenancyOpts)
Expand Down
6 changes: 5 additions & 1 deletion plugin/storage/grpc/factory.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,8 @@ import (
"io"

"github.com/spf13/viper"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"

"github.com/jaegertracing/jaeger/pkg/metrics"
Expand All @@ -41,6 +43,7 @@ type Factory struct {
options Options
metricsFactory metrics.Factory
logger *zap.Logger
tracerProvider trace.TracerProvider

builder config.PluginBuilder

Expand Down Expand Up @@ -77,8 +80,9 @@ func (f *Factory) InitFromOptions(opts Options) {
// Initialize implements storage.Factory
func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error {
f.metricsFactory, f.logger = metricsFactory, logger
f.tracerProvider = otel.GetTracerProvider()

services, err := f.builder.Build(logger)
services, err := f.builder.Build(logger, f.tracerProvider)
if err != nil {
return fmt.Errorf("grpc-plugin builder failed to create a store: %w", err)
}
Expand Down
3 changes: 2 additions & 1 deletion plugin/storage/grpc/factory_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ import (
"github.com/spf13/viper"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/otel/trace"
"go.uber.org/zap"

"github.com/jaegertracing/jaeger/pkg/config"
Expand All @@ -43,7 +44,7 @@ type mockPluginBuilder struct {
err error
}

func (b *mockPluginBuilder) Build(logger *zap.Logger) (*grpcConfig.ClientPluginServices, error) {
func (b *mockPluginBuilder) Build(logger *zap.Logger, tracer trace.TracerProvider) (*grpcConfig.ClientPluginServices, error) {
if b.err != nil {
return nil, b.err
}
Expand Down

0 comments on commit ab6732b

Please sign in to comment.