[v2] Add v1 factory converter to v2 storage factory

cmd/jaeger/internal/extension/jaegerstorage/converter/README.md

@@ -0,0 +1,5 @@
+# Storage Factory Converter
+
+A temporary v1 storage factory wrapper to implement v2 storage APIs.
+This way, the existing v1 storage factories declared in `jaegerstorageextension`
+can act as v2 storage while we migrate to v2 storage APIs.

cmd/jaeger/internal/extension/jaegerstorage/converter/empty_test.go

@@ -0,0 +1,14 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package converter
+
+import (
+	"testing"
+
+	"github.com/jaegertracing/jaeger/pkg/testutils"
+)
+
+func TestMain(m *testing.M) {
+	testutils.VerifyGoLeaks(m)
+}

cmd/jaeger/internal/extension/jaegerstorage/converter/factory.go

@@ -0,0 +1,58 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package converter
+
+import (
+	"context"
+	"io"
+
+	"go.uber.org/zap"
+
+	"github.com/jaegertracing/jaeger/pkg/metrics"
+	storage_v1 "github.com/jaegertracing/jaeger/storage"
+	"github.com/jaegertracing/jaeger/storage_v2/spanstore"
+)
+
+type Factory struct {
+	logger *zap.Logger
+	ss     storage_v1.Factory
+}
+
+func NewFactory(logger *zap.Logger, ss storage_v1.Factory) spanstore.Factory {
+	return &Factory{
+		logger: logger,
+		ss:     ss,
+	}
+}
+
+// Initialize implements spanstore.Factory.
+func (f *Factory) Initialize(ctx context.Context) error {
+	return f.ss.Initialize(metrics.NullFactory, f.logger)
+}
+
+// Close implements spanstore.Factory.
+func (f *Factory) Close(ctx context.Context) error {
+	if closer, ok := f.ss.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
+
+// CreateTraceReader implements spanstore.Factory.
+func (f *Factory) CreateTraceReader() (spanstore.Reader, error) {
+	spanReader, err := f.ss.CreateSpanReader()
+	if err != nil {
+		return nil, err
+	}
+	return NewTraceReader(spanReader)
+}
+
+// CreateTraceWriter implements spanstore.Factory.
+func (f *Factory) CreateTraceWriter() (spanstore.Writer, error) {
+	spanWriter, err := f.ss.CreateSpanWriter()
+	if err != nil {
+		return nil, err
+	}
+	return NewTraceWriter(spanWriter)
+}

cmd/jaeger/internal/extension/jaegerstorage/converter/reader.go

@@ -0,0 +1,139 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package converter
+
+import (
+	"context"
+	"encoding/binary"
+	"fmt"
+
+	jaeger2otlp "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/jaegertracing/jaeger/model"
+	spanstore_v1 "github.com/jaegertracing/jaeger/storage/spanstore"
+	"github.com/jaegertracing/jaeger/storage_v2/spanstore"
+)
+
+type TraceReader struct {
+	spanReader spanstore_v1.Reader
+}
+
+func NewTraceReader(spanReader spanstore_v1.Reader) (spanstore.Reader, error) {
+	return &TraceReader{
+		spanReader: spanReader,
+	}, nil
+}
+
+// GetTrace implements spanstore.Reader.
+func (s *TraceReader) GetTrace(ctx context.Context, traceID pcommon.TraceID) (ptrace.Traces, error) {
+	// otelcol-contrib has the translator to jaeger proto but declared in private function
+	// similar to https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/internal/coreinternal/idutils/big_endian_converter.go#L21
+	traceIDHigh, traceIDLow := binary.BigEndian.Uint64(traceID[:8]), binary.BigEndian.Uint64(traceID[8:])
+	id := model.TraceID{
+		Low:  traceIDLow,
+		High: traceIDHigh,
+	}
+	trace, err := s.spanReader.GetTrace(ctx, id)
+	if err != nil {
+		return ptrace.NewTraces(), err
+	}
+
+	batches := []*model.Batch{{Spans: trace.Spans}}
+	td, err := jaeger2otlp.ProtoToTraces(batches)
+	if err != nil {
+		return ptrace.NewTraces(), fmt.Errorf("cannot transform Jaeger trace to OTLP format: %w", err)
+	}
+
+	return td, nil
+}
+
+// GetServices implements spanstore.Reader.
+func (s *TraceReader) GetServices(ctx context.Context) ([]string, error) {
+	services, err := s.spanReader.GetServices(ctx)
+	if err != nil {
+		return []string{}, err
+	}
+	return services, nil
+}
+
+// GetOperations implements spanstore.Reader.
+func (s *TraceReader) GetOperations(ctx context.Context, query spanstore.OperationQueryParameters) ([]spanstore.Operation, error) {
+	ops, err := s.spanReader.GetOperations(ctx, spanstore_v1.OperationQueryParameters{
+		ServiceName: query.ServiceName,
+		SpanKind:    query.SpanKind,
+	})
+	if err != nil {
+		return []spanstore.Operation{}, err
+	}
+
+	operations := []spanstore.Operation{}
+	for _, op := range ops {
+		operations = append(operations, spanstore.Operation{
+			Name:     op.Name,
+			SpanKind: op.SpanKind,
+		})
+	}
+	return operations, nil
+}
+
+// FindTraces implements spanstore.Reader.
+func (s *TraceReader) FindTraces(ctx context.Context, query spanstore.TraceQueryParameters) ([]ptrace.Traces, error) {
+	traces, err := s.spanReader.FindTraces(ctx, &spanstore_v1.TraceQueryParameters{
+		ServiceName:   query.ServiceName,
+		OperationName: query.OperationName,
+		Tags:          query.Tags,
+		StartTimeMin:  query.StartTimeMin,
+		StartTimeMax:  query.StartTimeMax,
+		DurationMin:   query.DurationMin,
+		DurationMax:   query.DurationMax,
+		NumTraces:     query.NumTraces,
+	})
+	if err != nil {
+		return []ptrace.Traces{}, err
+	}
+
+	tds := []ptrace.Traces{}
+	for _, trace := range traces {
+		batch := []*model.Batch{{Spans: trace.Spans}}
+		td, err := jaeger2otlp.ProtoToTraces(batch)
+		if err != nil {
+			return []ptrace.Traces{}, fmt.Errorf("cannot transform Jaeger trace to OTLP format: %w", err)
+		}
+
+		tds = append(tds, td)
+	}
+
+	return tds, nil
+}
+
+// FindTraceIDs implements spanstore.Reader.
+func (s *TraceReader) FindTraceIDs(ctx context.Context, query spanstore.TraceQueryParameters) ([]pcommon.TraceID, error) {
+	ids, err := s.spanReader.FindTraceIDs(ctx, &spanstore_v1.TraceQueryParameters{
+		ServiceName:   query.ServiceName,
+		OperationName: query.OperationName,
+		Tags:          query.Tags,
+		StartTimeMin:  query.StartTimeMin,
+		StartTimeMax:  query.StartTimeMax,
+		DurationMin:   query.DurationMin,
+		DurationMax:   query.DurationMax,
+		NumTraces:     query.NumTraces,
+	})
+	if err != nil {
+		return []pcommon.TraceID{}, err
+	}
+
+	traceIDs := []pcommon.TraceID{}
+	for _, id := range ids {
+		// otelcol-contrib has the translator to OTLP but declared in private function
+		// similar to https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/internal/coreinternal/idutils/big_endian_converter.go#L13
+		traceID := [16]byte{}
+		binary.BigEndian.PutUint64(traceID[:8], id.High)
+		binary.BigEndian.PutUint64(traceID[8:], id.Low)
+		traceIDs = append(traceIDs, traceID)
+	}
+
+	return traceIDs, nil
+}

cmd/jaeger/internal/extension/jaegerstorage/converter/writer.go

@@ -0,0 +1,44 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package converter
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	otlp2jaeger "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	spanstore_v1 "github.com/jaegertracing/jaeger/storage/spanstore"
+	"github.com/jaegertracing/jaeger/storage_v2/spanstore"
+)
+
+type TraceWriter struct {
+	spanWriter spanstore_v1.Writer
+}
+
+func NewTraceWriter(spanWriter spanstore_v1.Writer) (spanstore.Writer, error) {
+	return &TraceWriter{
+		spanWriter: spanWriter,
+	}, nil
+}
+
+// WriteTraces implements spanstore.Writer.
+func (t *TraceWriter) WriteTraces(ctx context.Context, td ptrace.Traces) error {
+	batches, err := otlp2jaeger.ProtoFromTraces(td)
+	if err != nil {
+		return fmt.Errorf("cannot transform OTLP traces to Jaeger format: %w", err)
+	}
+	var errs []error
+	for _, batch := range batches {
+		for _, span := range batch.Spans {
+			if span.Process == nil {
+				span.Process = batch.Process
+			}
+			errs = append(errs, t.spanWriter.WriteSpan(ctx, span))
+		}
+	}
+	return errors.Join(errs...)
+}