Support uploading traces to UI in OpenTelemetry format (OTLP JSON)

cmd/query/app/apiv3/otlp_translator.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 	model2otel "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
+	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.opentelemetry.io/collector/pdata/ptrace/ptraceotlp"
 
 	"github.com/jaegertracing/jaeger/model"
@@ -45,3 +46,38 @@ func modelToOTLP(spans []*model.Span) ([]*tracev1.ResourceSpans, error) {
 	}
 	return chunk.ResourceSpans, nil
 }
+
+func OTLP2model(OTLPSpans []byte) ([]*model.Batch, error) {
+	ptraceUnmarshaler := ptrace.JSONUnmarshaler{}
+	otlpTraces, err := ptraceUnmarshaler.UnmarshalTraces(OTLPSpans)
+	if err != nil {
+		fmt.Println(err)
+		return nil, fmt.Errorf("cannot unmarshal OTLP : %w", err)
+	}
+	jaegerBatches, err := model2otel.ProtoFromTraces(otlpTraces)
+	if err != nil {
+		fmt.Println(err)
+		return nil, fmt.Errorf("cannot transform OTLP to Jaeger: %w", err)
+	}
+
+	return jaegerBatches, nil
+}
+
+func BatchesToTraces(jaegerBatches []*model.Batch) ([]model.Trace, error) {
+	var jaegerTraces []model.Trace
+	spanMap := make(map[model.TraceID][]*model.Span)
+	for _, v := range jaegerBatches {
+		jaegerTrace := model.Trace{
+			Spans: v.Spans,
+		}
+		jaegerTrace.DenormalizeProcess(v.Process)
+		jaegerTrace.FlattenToSpansMaps(spanMap)
+	}
+	for _, v := range spanMap {
+		jaegerTrace := model.Trace{
+			Spans: v,
+		}
+		jaegerTraces = append(jaegerTraces, jaegerTrace)
+	}
+	return jaegerTraces, nil
+}

cmd/query/app/apiv3/package_test.go

@@ -6,9 +6,67 @@ package apiv3
 import (
 	"testing"
 
+	"github.com/jaegertracing/jaeger/model"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/goleak"
 )
 
 func TestMain(m *testing.M) {
 	goleak.VerifyTestMain(m)
 }
+
+func TestBatchesToTraces(t *testing.T) {
+	b1 := &model.Batch{
+		Spans: []*model.Span{
+			{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(1), OperationName: "x"},
+			{TraceID: model.NewTraceID(1, 3), SpanID: model.NewSpanID(2), OperationName: "y"},
+		},
+		Process: model.NewProcess("process1", model.KeyValues{}),
+	}
+
+	b2 := &model.Batch{
+		Spans: []*model.Span{
+			{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(2), OperationName: "z"},
+		},
+		Process: model.NewProcess("process2", model.KeyValues{}),
+	}
+
+	mainBatch := []*model.Batch{b1, b2}
+
+	traces, err := BatchesToTraces(mainBatch)
+	require.Nil(t, err)
+
+	s1 := []*model.Span{
+		{
+			TraceID:       model.NewTraceID(1, 2),
+			SpanID:        model.NewSpanID(1),
+			OperationName: "x",
+			Process:       model.NewProcess("process1", model.KeyValues{}),
+		},
+		{
+         TraceID: model.NewTraceID(1, 2), 
+         SpanID: model.NewSpanID(2), 
+         OperationName: "z",
+			Process:       model.NewProcess("process2", model.KeyValues{}),
+      },
+   }
+
+	s2 := []*model.Span{
+      {
+         TraceID: model.NewTraceID(1, 3), 
+         SpanID: model.NewSpanID(2), 
+         OperationName: "y",
+			Process:       model.NewProcess("process1", model.KeyValues{}),
+      },
+   }
+
+	t1 := model.Trace{
+		Spans: s1,
+	}
+	t2 := model.Trace{
+		Spans: s2,
+	}
+	mainTrace := []model.Trace{t1, t2}
+	assert.Equal(t, mainTrace, traces)
+}

cmd/query/app/http_handler.go

@@ -20,6 +20,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -31,6 +32,7 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 	"go.uber.org/zap"
 
+	"github.com/jaegertracing/jaeger/cmd/query/app/apiv3"
 	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
 	"github.com/jaegertracing/jaeger/model"
 	uiconv "github.com/jaegertracing/jaeger/model/converter/json"
@@ -128,6 +130,7 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router) {
 	aH.handleFunc(router, aH.getOperations, "/operations").Methods(http.MethodGet)
 	// TODO - remove this when UI catches up
 	aH.handleFunc(router, aH.getOperationsLegacy, "/services/{%s}/operations", serviceParam).Methods(http.MethodGet)
+	aH.handleFunc(router, aH.transformOTLP, "/transform").Methods(http.MethodPost)
 	aH.handleFunc(router, aH.dependencies, "/dependencies").Methods(http.MethodGet)
 	aH.handleFunc(router, aH.latencies, "/metrics/latencies").Methods(http.MethodGet)
 	aH.handleFunc(router, aH.calls, "/metrics/calls").Methods(http.MethodGet)
@@ -193,6 +196,43 @@ func (aH *APIHandler) getOperationsLegacy(w http.ResponseWriter, r *http.Request
 	aH.writeJSON(w, r, &structuredRes)
 }
 
+func (aH *APIHandler) transformOTLP(w http.ResponseWriter, r *http.Request) {
+	body, err := ioutil.ReadAll(r.Body)
+	if aH.handleError(w, err, http.StatusBadRequest) {
+		return
+	}
+
+	var uiErrors []structuredError
+	batches, err := apiv3.OTLP2model(body)
+
+	if aH.handleError(w, err, http.StatusInternalServerError) {
+		return
+	}
+
+	traces, err := apiv3.BatchesToTraces(batches)
+
+	if aH.handleError(w, err, http.StatusInternalServerError) {
+		return
+	}
+
+	uiTraces := make([]*ui.Trace, len(traces))
+
+	for i, v := range traces {
+		uiTrace, uiErr := aH.convertModelToUI(&v, false)
+		if uiErr != nil {
+			uiErrors = append(uiErrors, *uiErr)
+		}
+		uiTraces[i] = uiTrace
+	}
+
+	structuredRes := structuredResponse{
+		Data:   uiTraces,
+		Errors: uiErrors,
+	}
+	aH.writeJSON(w, r, structuredRes)
+
+}
+
 func (aH *APIHandler) getOperations(w http.ResponseWriter, r *http.Request) {
 	service := r.FormValue(serviceParam)
 	if service == "" {

model/trace.go

@@ -32,3 +32,21 @@ func (t *Trace) NormalizeTimestamps() {
 		span.NormalizeTimestamps()
 	}
 }
+
+func (m *Trace) DenormalizeProcess(p *Process) {
+	for _, v := range m.Spans {
+		v.Process = p
+	}
+}
+
+func (m *Trace) FlattenToSpansMaps(spanMap map[TraceID][]*Span) {
+	for _, v := range m.Spans {
+		val, ok := spanMap[v.TraceID]
+		if !ok {
+			spanMap[v.TraceID] = []*Span{v}
+		} else {
+			spanMap[v.TraceID] = append(val, v)
+		}
+	}
+
+}

model/trace_test.go

@@ -67,3 +67,68 @@ func TestTraceNormalizeTimestamps(t *testing.T) {
 	assert.Equal(t, span.StartTime, tt1.UTC())
 	assert.Equal(t, span.Logs[0].Timestamp, tt2.UTC())
 }
+
+func TestFlattenToSpanMaps(t *testing.T) {
+	b1 := &model.Trace{
+		Spans: []*model.Span{
+			{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(1), OperationName: "x"},
+			{TraceID: model.NewTraceID(1, 3), SpanID: model.NewSpanID(2), OperationName: "y"},
+		},
+	}
+
+	b2 := &model.Trace{
+		Spans: []*model.Span{
+			{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(2), OperationName: "z"},
+		},
+	}
+
+	t1 := []*model.Span{
+		{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(1), OperationName: "x"},
+		{TraceID: model.NewTraceID(1, 2), SpanID: model.NewSpanID(2), OperationName: "z"}}
+
+	t2 := []*model.Span{{TraceID: model.NewTraceID(1, 3), SpanID: model.NewSpanID(2), OperationName: "y"}}
+	spanMap := make(map[model.TraceID][]*model.Span)
+	b1.FlattenToSpansMaps(spanMap)
+	b2.FlattenToSpansMaps(spanMap)
+	assert.Equal(t, t1, spanMap[model.NewTraceID(1, 2)])
+	assert.Equal(t, t2, spanMap[model.NewTraceID(1, 3)])
+}
+
+
+func TestDenormalizeProcess(t *testing.T){
+	t1 := &model.Trace{
+		Spans: []*model.Span{
+			{
+            TraceID: model.NewTraceID(1, 2), 
+            SpanID: model.NewSpanID(1), 
+            OperationName: "x",
+         },
+			{
+            TraceID: model.NewTraceID(1, 3), 
+            SpanID: model.NewSpanID(2), 
+            OperationName: "y",
+         },
+		},
+	}
+   p1 := model.NewProcess("process1", model.KeyValues{})
+
+	t2 := &model.Trace{
+		Spans: []*model.Span{
+			{
+            TraceID: model.NewTraceID(1, 2), 
+            SpanID: model.NewSpanID(1), 
+            OperationName: "x",
+            Process: model.NewProcess("process1", model.KeyValues{}),
+         },
+			{
+            TraceID: model.NewTraceID(1, 3), 
+            SpanID: model.NewSpanID(2), 
+            OperationName: "y",
+            Process: model.NewProcess("process1", model.KeyValues{}),
+         },
+		},
+	}
+   t1.DenormalizeProcess(p1)
+	assert.Equal(t, t1, t2)
+
+}