refactor and move to translator

Signed-off-by: Yuri Shkuro <github@ysh.us>

plugin/storage/integration/badgerstore_test.go

@@ -39,7 +39,7 @@ func (s *BadgerIntegrationStorage) initialize(t *testing.T) {
 	spanReader, err := s.factory.CreateSpanReader()
 	require.NoError(t, err)
 	s.TraceReader = v1adapter.NewTraceReader(spanReader)
-	
+
 	s.SamplingStore, err = s.factory.CreateSamplingStore(0)
 	require.NoError(t, err)
 }

plugin/storage/integration/integration.go

@@ -159,7 +159,7 @@ func (s *StorageIntegration) testGetServices(t *testing.T) {
 				iterTraces := s.TraceReader.FindTraces(context.Background(), tracestore.TraceQueryParams{
 					ServiceName: service,
 				})
-				traces, err := v1adapter.PTracesSeq2ToModel(iterTraces)
+				traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
 				if err != nil {
 					t.Log(err)
 					continue
@@ -220,7 +220,7 @@ func (s *StorageIntegration) testGetLargeSpan(t *testing.T) {
 	var actual *model.Trace
 	found := s.waitForCondition(t, func(_ *testing.T) bool {
 		iterTraces := s.TraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: expectedTraceID.ToOTELTraceID()})
-		traces, err := v1adapter.PTracesSeq2ToModel(iterTraces)
+		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
 		require.NotEmpty(t, traces)
 		actual = traces[0]
 		return err == nil && len(actual.Spans) >= len(expected.Spans)
@@ -295,7 +295,7 @@ func (s *StorageIntegration) testGetTrace(t *testing.T) {
 	var actual *model.Trace
 	found := s.waitForCondition(t, func(t *testing.T) bool {
 		iterTraces := s.TraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: expectedTraceID.ToOTELTraceID()})
-		traces, err := v1adapter.PTracesSeq2ToModel(iterTraces)
+		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
 		if err != nil {
 			t.Log(err)
 		}
@@ -310,7 +310,7 @@ func (s *StorageIntegration) testGetTrace(t *testing.T) {
 	t.Run("NotFound error", func(t *testing.T) {
 		fakeTraceID := model.TraceID{High: 0, Low: 1}
 		iterTraces := s.TraceReader.GetTraces(context.Background(), tracestore.GetTraceParams{TraceID: fakeTraceID.ToOTELTraceID()})
-		traces, err := v1adapter.PTracesSeq2ToModel(iterTraces)
+		traces, err := v1adapter.V1TracesFromSeq2(iterTraces)
 		assert.Equal(t, spanstore.ErrTraceNotFound, err)
 		assert.Nil(t, traces)
 	})
@@ -355,7 +355,7 @@ func (s *StorageIntegration) findTracesByQuery(t *testing.T, query *tracestore.T
 	found := s.waitForCondition(t, func(t *testing.T) bool {
 		var err error
 		iterTraces := s.TraceReader.FindTraces(context.Background(), *query)
-		traces, err = v1adapter.PTracesSeq2ToModel(iterTraces)
+		traces, err = v1adapter.V1TracesFromSeq2(iterTraces)
 		if err != nil {
 			t.Log(err)
 			return false

storage_v2/v1adapter/translator.go

@@ -10,10 +10,20 @@ import (
 
 	"github.com/jaegertracing/jaeger/internal/jptrace"
 	"github.com/jaegertracing/jaeger/model"
+	"github.com/jaegertracing/jaeger/pkg/iter"
+	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
+// V1BatchesFromTraces converts OpenTelemetry traces (ptrace.Traces)
+// to Jaeger model batches ([]*model.Batch).
+func V1BatchesFromTraces(traces ptrace.Traces) []*model.Batch {
+	return ProtoFromTraces(traces)
+}
+
 // ProtoFromTraces converts OpenTelemetry traces (ptrace.Traces)
 // to Jaeger model batches ([]*model.Batch).
+//
+// TODO remove this function in favor of V1BatchesFromTraces
 func ProtoFromTraces(traces ptrace.Traces) []*model.Batch {
 	batches := jaegerTranslator.ProtoFromTraces(traces)
 	spanMap := createSpanMapFromBatches(batches)
@@ -32,6 +42,46 @@ func V1BatchesToTraces(batches []*model.Batch) ptrace.Traces {
 	return traces
 }
 
+// V1TracesFromSeq2 converts an interator of ptrace.Traces chunks into v1 traces.
+// Returns spanstore.ErrTraceNotFound for empty iterators.
+func V1TracesFromSeq2(otelSeq iter.Seq2[[]ptrace.Traces, error]) ([]*model.Trace, error) {
+	var (
+		jaegerTraces []*model.Trace
+		iterErr      error
+	)
+	jptrace.AggregateTraces(otelSeq)(func(otelTrace ptrace.Traces, err error) bool {
+		if err != nil {
+			iterErr = err
+			return false
+		}
+		jaegerTraces = append(jaegerTraces, modelTraceFromOtelTrace(otelTrace))
+		return true
+	})
+	if iterErr != nil {
+		return nil, iterErr
+	}
+	if len(jaegerTraces) == 0 {
+		return nil, spanstore.ErrTraceNotFound
+	}
+	return jaegerTraces, nil
+}
+
+// modelTraceFromOtelTrace extracts spans from otel traces
+func modelTraceFromOtelTrace(otelTrace ptrace.Traces) *model.Trace {
+	var spans []*model.Span
+	batches := V1BatchesFromTraces(otelTrace)
+	for _, batch := range batches {
+		for _, span := range batch.Spans {
+			if span.Process == nil {
+				proc := *batch.Process // shallow clone
+				span.Process = &proc
+			}
+			spans = append(spans, span)
+		}
+	}
+	return &model.Trace{Spans: spans}
+}
+
 func createSpanMapFromBatches(batches []*model.Batch) map[model.SpanID]*model.Span {
 	spanMap := make(map[model.SpanID]*model.Span)
 	for _, batch := range batches {