Replace es-spanstore tracing instrumentation with OpenTelemetry

cmd/all-in-one/main.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"go.opentelemetry.io/otel"
 	_ "go.uber.org/automaxprocs"
 	"go.uber.org/zap"
 
@@ -104,6 +105,7 @@ by default uses only in-memory database.`,
 			if err != nil {
 				logger.Fatal("Failed to initialize tracer", zap.Error(err))
 			}
+			otel.SetTracerProvider(tracer.OTEL)
 
 			storageFactory.InitFromViper(v, logger)
 			if err := storageFactory.Initialize(metricsFactory, logger); err != nil {

cmd/query/main.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
+	"go.opentelemetry.io/otel"
 	_ "go.uber.org/automaxprocs"
 	"go.uber.org/zap"
 
@@ -76,6 +77,7 @@ func main() {
 			if err != nil {
 				logger.Fatal("Failed to create tracer:", zap.Error(err))
 			}
+			otel.SetTracerProvider(jtracer.OTEL)
 			queryOpts, err := new(app.QueryOptions).InitFromViper(v, logger)
 			if err != nil {
 				logger.Fatal("Failed to configure query service", zap.Error(err))

plugin/storage/es/factory.go

@@ -21,6 +21,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/es"
@@ -50,6 +52,7 @@ type Factory struct {
 
 	metricsFactory metrics.Factory
 	logger         *zap.Logger
+	tracer         trace.TracerProvider
 
 	newClientFn func(c *config.Configuration, logger *zap.Logger, metricsFactory metrics.Factory) (es.Client, error)
 
@@ -64,6 +67,7 @@ func NewFactory() *Factory {
 	return &Factory{
 		Options:     NewOptions(primaryNamespace, archiveNamespace),
 		newClientFn: config.NewClient,
+		tracer:      otel.GetTracerProvider(),
 	}
 }
 
@@ -108,49 +112,48 @@ func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger)
 
 // CreateSpanReader implements storage.Factory
 func (f *Factory) CreateSpanReader() (spanstore.Reader, error) {
-	return createSpanReader(f.metricsFactory, f.logger, f.primaryClient, f.primaryConfig, false)
+	return createSpanReader(f.primaryClient, f.primaryConfig, false, f.metricsFactory, f.logger, f.tracer)
 }
 
 // CreateSpanWriter implements storage.Factory
 func (f *Factory) CreateSpanWriter() (spanstore.Writer, error) {
-	return createSpanWriter(f.metricsFactory, f.logger, f.primaryClient, f.primaryConfig, false)
+	return createSpanWriter(f.primaryClient, f.primaryConfig, false, f.metricsFactory, f.logger)
 }
 
 // CreateDependencyReader implements storage.Factory
 func (f *Factory) CreateDependencyReader() (dependencystore.Reader, error) {
-	return createDependencyReader(f.logger, f.primaryClient, f.primaryConfig)
+	return createDependencyReader(f.primaryClient, f.primaryConfig, f.logger)
 }
 
 // CreateArchiveSpanReader implements storage.ArchiveFactory
 func (f *Factory) CreateArchiveSpanReader() (spanstore.Reader, error) {
 	if !f.archiveConfig.Enabled {
 		return nil, nil
 	}
-	return createSpanReader(f.metricsFactory, f.logger, f.archiveClient, f.archiveConfig, true)
+	return createSpanReader(f.archiveClient, f.archiveConfig, true, f.metricsFactory, f.logger, f.tracer)
 }
 
 // CreateArchiveSpanWriter implements storage.ArchiveFactory
 func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 	if !f.archiveConfig.Enabled {
 		return nil, nil
 	}
-	return createSpanWriter(f.metricsFactory, f.logger, f.archiveClient, f.archiveConfig, true)
+	return createSpanWriter(f.archiveClient, f.archiveConfig, true, f.metricsFactory, f.logger)
 }
 
 func createSpanReader(
-	mFactory metrics.Factory,
-	logger *zap.Logger,
 	client es.Client,
 	cfg *config.Configuration,
 	archive bool,
+	mFactory metrics.Factory,
+	logger *zap.Logger,
+	tp trace.TracerProvider,
 ) (spanstore.Reader, error) {
 	if cfg.UseILM && !cfg.UseReadWriteAliases {
 		return nil, fmt.Errorf("--es.use-ilm must always be used in conjunction with --es.use-aliases to ensure ES writers and readers refer to the single index mapping")
 	}
 	return esSpanStore.NewSpanReader(esSpanStore.SpanReaderParams{
 		Client:                        client,
-		Logger:                        logger,
-		MetricsFactory:                mFactory,
 		MaxDocCount:                   cfg.MaxDocCount,
 		MaxSpanAge:                    cfg.MaxSpanAge,
 		IndexPrefix:                   cfg.IndexPrefix,
@@ -162,15 +165,18 @@ func createSpanReader(
 		UseReadWriteAliases:           cfg.UseReadWriteAliases,
 		Archive:                       archive,
 		RemoteReadClusters:            cfg.RemoteReadClusters,
+		Logger:                        logger,
+		MetricsFactory:                mFactory,
+		Tracer:                        tp.Tracer("esSpanStore.SpanReader"),
 	}), nil
 }
 
 func createSpanWriter(
-	mFactory metrics.Factory,
-	logger *zap.Logger,
 	client es.Client,
 	cfg *config.Configuration,
 	archive bool,
+	mFactory metrics.Factory,
+	logger *zap.Logger,
 ) (spanstore.Writer, error) {
 	var tags []string
 	var err error
@@ -197,8 +203,6 @@ func createSpanWriter(
 	}
 	writer := esSpanStore.NewSpanWriter(esSpanStore.SpanWriterParams{
 		Client:                 client,
-		Logger:                 logger,
-		MetricsFactory:         mFactory,
 		IndexPrefix:            cfg.IndexPrefix,
 		SpanIndexDateLayout:    cfg.IndexDateLayoutSpans,
 		ServiceIndexDateLayout: cfg.IndexDateLayoutServices,
@@ -207,6 +211,8 @@ func createSpanWriter(
 		TagDotReplacement:      cfg.Tags.DotReplacement,
 		Archive:                archive,
 		UseReadWriteAliases:    cfg.UseReadWriteAliases,
+		Logger:                 logger,
+		MetricsFactory:         mFactory,
 	})
 
 	// Creating a template here would conflict with the one created for ILM resulting to no index rollover
@@ -220,9 +226,9 @@ func createSpanWriter(
 }
 
 func createDependencyReader(
-	logger *zap.Logger,
 	client es.Client,
 	cfg *config.Configuration,
+	logger *zap.Logger,
 ) (dependencystore.Reader, error) {
 	reader := esDepStore.NewDependencyStore(esDepStore.DependencyStoreParams{
 		Client:              client,

plugin/storage/es/spanstore/reader.go

@@ -24,9 +24,9 @@ import (
 	"time"
 
 	"github.com/olivere/elastic"
-	"github.com/opentracing/opentracing-go"
-	ottag "github.com/opentracing/opentracing-go/ext"
-	otlog "github.com/opentracing/opentracing-go/log"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/model"
@@ -93,7 +93,6 @@ var (
 // SpanReader can query for and load traces from ElasticSearch
 type SpanReader struct {
 	client es.Client
-	logger *zap.Logger
 	// The age of the oldest service/operation we will look for. Because indices in ElasticSearch are by day,
 	// this will be rounded down to UTC 00:00 of that day.
 	maxSpanAge                    time.Duration
@@ -109,15 +108,15 @@ type SpanReader struct {
 	sourceFn                      sourceFn
 	maxDocCount                   int
 	useReadWriteAliases           bool
+	logger                        *zap.Logger
+	tracer                        trace.Tracer
 }
 
 // SpanReaderParams holds constructor params for NewSpanReader
 type SpanReaderParams struct {
 	Client                        es.Client
-	Logger                        *zap.Logger
 	MaxSpanAge                    time.Duration
 	MaxDocCount                   int
-	MetricsFactory                metrics.Factory
 	IndexPrefix                   string
 	SpanIndexDateLayout           string
 	ServiceIndexDateLayout        string
@@ -127,6 +126,9 @@ type SpanReaderParams struct {
 	Archive                       bool
 	UseReadWriteAliases           bool
 	RemoteReadClusters            []string
+	MetricsFactory                metrics.Factory
+	Logger                        *zap.Logger
+	Tracer                        trace.Tracer
 }
 
 // NewSpanReader returns a new SpanReader with a metrics.
@@ -139,7 +141,6 @@ func NewSpanReader(p SpanReaderParams) *SpanReader {
 	}
 	return &SpanReader{
 		client:                        p.Client,
-		logger:                        p.Logger,
 		maxSpanAge:                    maxSpanAge,
 		serviceOperationStorage:       NewServiceOperationStorage(p.Client, p.Logger, 0), // the decorator takes care of metrics
 		spanIndexPrefix:               indexNames(p.IndexPrefix, spanIndex),
@@ -153,6 +154,8 @@ func NewSpanReader(p SpanReaderParams) *SpanReader {
 		sourceFn:                      getSourceFn(p.Archive, p.MaxDocCount),
 		maxDocCount:                   p.MaxDocCount,
 		useReadWriteAliases:           p.UseReadWriteAliases,
+		logger:                        p.Logger,
+		tracer:                        p.Tracer,
 	}
 }
 
@@ -238,8 +241,8 @@ func indexNames(prefix, index string) string {
 
 // GetTrace takes a traceID and returns a Trace associated with that traceID
 func (s *SpanReader) GetTrace(ctx context.Context, traceID model.TraceID) (*model.Trace, error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "GetTrace")
-	defer span.Finish()
+	ctx, span := s.tracer.Start(ctx, "GetTrace")
+	defer span.End()
 	currentTime := time.Now()
 	traces, err := s.multiRead(ctx, []model.TraceID{traceID}, currentTime.Add(-s.maxSpanAge), currentTime)
 	if err != nil {
@@ -283,8 +286,8 @@ func (s *SpanReader) unmarshalJSONSpan(esSpanRaw *elastic.SearchHit) (*dbmodel.S
 
 // GetServices returns all services traced by Jaeger, ordered by frequency
 func (s *SpanReader) GetServices(ctx context.Context) ([]string, error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "GetServices")
-	defer span.Finish()
+	ctx, span := s.tracer.Start(ctx, "GetService")
+	defer span.End()
 	currentTime := time.Now()
 	jaegerIndices := s.timeRangeIndices(s.serviceIndexPrefix, s.serviceIndexDateLayout, currentTime.Add(-s.maxSpanAge), currentTime, s.serviceIndexRolloverFrequency)
 	return s.serviceOperationStorage.getServices(ctx, jaegerIndices, s.maxDocCount)
@@ -295,8 +298,8 @@ func (s *SpanReader) GetOperations(
 	ctx context.Context,
 	query spanstore.OperationQueryParameters,
 ) ([]spanstore.Operation, error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "GetOperations")
-	defer span.Finish()
+	ctx, span := s.tracer.Start(ctx, "GetOperations")
+	defer span.End()
 	currentTime := time.Now()
 	jaegerIndices := s.timeRangeIndices(s.serviceIndexPrefix, s.serviceIndexDateLayout, currentTime.Add(-s.maxSpanAge), currentTime, s.serviceIndexRolloverFrequency)
 	operations, err := s.serviceOperationStorage.getOperations(ctx, jaegerIndices, query.ServiceName, s.maxDocCount)
@@ -329,8 +332,8 @@ func bucketToStringArray(buckets []*elastic.AggregationBucketKeyItem) ([]string,
 
 // FindTraces retrieves traces that match the traceQuery
 func (s *SpanReader) FindTraces(ctx context.Context, traceQuery *spanstore.TraceQueryParameters) ([]*model.Trace, error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "FindTraces")
-	defer span.Finish()
+	ctx, span := s.tracer.Start(ctx, "FindTraces")
+	defer span.End()
 
 	uniqueTraceIDs, err := s.FindTraceIDs(ctx, traceQuery)
 	if err != nil {
@@ -341,8 +344,8 @@ func (s *SpanReader) FindTraces(ctx context.Context, traceQuery *spanstore.Trace
 
 // FindTraceIDs retrieves traces IDs that match the traceQuery
 func (s *SpanReader) FindTraceIDs(ctx context.Context, traceQuery *spanstore.TraceQueryParameters) ([]model.TraceID, error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "FindTraceIDs")
-	defer span.Finish()
+	ctx, span := s.tracer.Start(ctx, "FindTraceIDs")
+	defer span.End()
 
 	if err := validateQuery(traceQuery); err != nil {
 		return nil, err
@@ -360,9 +363,16 @@ func (s *SpanReader) FindTraceIDs(ctx context.Context, traceQuery *spanstore.Tra
 }
 
 func (s *SpanReader) multiRead(ctx context.Context, traceIDs []model.TraceID, startTime, endTime time.Time) ([]*model.Trace, error) {
-	childSpan, _ := opentracing.StartSpanFromContext(ctx, "multiRead")
-	childSpan.LogFields(otlog.Object("trace_ids", traceIDs))
-	defer childSpan.Finish()
+	ctx, childSpan := s.tracer.Start(ctx, "multiRead")
+	defer childSpan.End()
+
+	if childSpan.IsRecording() {
+		tracesIDs := make([]string, len(traceIDs))
+		for i, traceID := range traceIDs {
+			tracesIDs[i] = traceID.String()
+		}
+		childSpan.SetAttributes(attribute.Key("trace_ids").StringSlice(tracesIDs))
+	}
 
 	if len(traceIDs) == 0 {
 		return []*model.Trace{}, nil
@@ -503,8 +513,8 @@ func validateQuery(p *spanstore.TraceQueryParameters) error {
 }
 
 func (s *SpanReader) findTraceIDs(ctx context.Context, traceQuery *spanstore.TraceQueryParameters) ([]string, error) {
-	childSpan, _ := opentracing.StartSpanFromContext(ctx, "findTraceIDs")
-	defer childSpan.Finish()
+	ctx, childSpan := s.tracer.Start(ctx, "findTraceIDs")
+	defer childSpan.End()
 	//  Below is the JSON body to our HTTP GET request to ElasticSearch. This function creates this.
 	// {
 	//      "size": 0,
@@ -686,7 +696,7 @@ func (s *SpanReader) buildObjectQuery(field string, k string, v string) elastic.
 	return elastic.NewBoolQuery().Must(keyQuery)
 }
 
-func logErrorToSpan(span opentracing.Span, err error) {
-	ottag.Error.Set(span, true)
-	span.LogFields(otlog.Error(err))
+func logErrorToSpan(span trace.Span, err error) {
+	span.RecordError(err)
+	span.SetStatus(codes.Error, err.Error())
 }