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

cmd/query/app/fixture/otlp2jaeger-in.json

@@ -0,0 +1,52 @@
+{
+   "resourceSpans":[
+      {
+         "resource":{
+            "attributes":[
+               {
+                  "key":"service.name",
+                  "value":{
+                     "stringValue":"telemetrygen"
+                  }
+               }
+            ]
+         },
+         "scopeSpans":[
+            {
+               "scope":{
+                  "name":"telemetrygen"
+               },
+               "spans":[
+                  {
+                     "traceId":"83a9efd15c1c98a977e0711cc93ee28b",
+                     "spanId":"e127af99e3b3e074",
+                     "parentSpanId":"909541b92cf05311",
+                     "name":"okey-dokey-0",
+                     "kind":2,
+                     "startTimeUnixNano":"1706678909209712000",
+                     "endTimeUnixNano":"1706678909209835000",
+                     "attributes":[
+                        {
+                           "key":"net.peer.ip",
+                           "value":{
+                              "stringValue":"1.2.3.4"
+                           }
+                        },
+                        {
+                           "key":"peer.service",
+                           "value":{
+                              "stringValue":"telemetrygen-client"
+                           }
+                        }
+                     ],
+                     "status":{
+
+                     }
+                  }
+               ]
+            }
+         ],
+         "schemaUrl":"https://opentelemetry.io/schemas/1.4.0"
+      }
+   ]
+}

cmd/query/app/fixture/otlp2jaeger-out.json

@@ -0,0 +1,59 @@
+{
+	"data": [
+		{
+			"traceID": "83a9efd15c1c98a977e0711cc93ee28b",
+			"spans": [
+				{
+					"traceID": "83a9efd15c1c98a977e0711cc93ee28b",
+					"spanID": "e127af99e3b3e074",
+					"operationName": "okey-dokey-0",
+					"references": [
+						{
+							"refType": "CHILD_OF",
+							"traceID": "83a9efd15c1c98a977e0711cc93ee28b",
+							"spanID": "909541b92cf05311"
+						}
+					],
+					"startTime": 1706678909209712,
+					"duration": 123,
+					"tags": [
+						{
+							"key": "otel.library.name",
+							"type": "string",
+							"value": "telemetrygen"
+						},
+						{
+							"key": "net.peer.ip",
+							"type": "string",
+							"value": "1.2.3.4"
+						},
+						{
+							"key": "peer.service",
+							"type": "string",
+							"value": "telemetrygen-client"
+						},
+						{
+							"key": "span.kind",
+							"type": "string",
+							"value": "server"
+						}
+					],
+					"logs": [],
+					"processID": "p1",
+					"warnings": null
+				}
+			],
+			"processes": {
+				"p1": {
+					"serviceName": "telemetrygen",
+					"tags": []
+				}
+			},
+			"warnings": null
+		}
+	],
+	"total": 0,
+	"limit": 0,
+	"offset": 0,
+	"errors": null
+}

cmd/query/app/http_handler.go

@@ -20,6 +20,7 @@
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -128,6 +129,7 @@
 	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 +195,42 @@
 	aH.writeJSON(w, r, &structuredRes)
 }
 
+func (aH *APIHandler) transformOTLP(w http.ResponseWriter, r *http.Request) {
+	body, err := io.ReadAll(r.Body)
+	if aH.handleError(w, err, http.StatusBadRequest) {
+		return
+	}
+
+	var uiErrors []structuredError
+	batches, err := otlp2model(body)
+
+	if aH.handleError(w, err, http.StatusInternalServerError) {
+		return
+	}
+
+	traces, err := 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 == "" {

cmd/query/app/http_handler_test.go

@@ -25,6 +25,7 @@ import (
 	"math"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"testing"
 	"time"
 
@@ -270,6 +271,24 @@ func TestWriteJSON(t *testing.T) {
 	}
 }
 
+func readOTLPTraces(t *testing.T) (interface{}, error) {
+	dat, err := os.ReadFile("./fixture/otlp2jaeger-in.json")
+	require.NoError(t, err)
+	out := new(interface{})
+	err = json.Unmarshal(dat, out)
+	require.NoError(t, err)
+	return out, nil
+}
+
+func readJaegerTraces(t *testing.T) (interface{}, error) {
+	dat, err := os.ReadFile("./fixture/otlp2jaeger-out.json")
+	out := new(interface{})
+	require.NoError(t, err)
+	err = json.Unmarshal(dat, out)
+	require.NoError(t, err)
+	return out, nil
+}
+
 func TestGetTrace(t *testing.T) {
 	testCases := []struct {
 		suffix      string
@@ -623,6 +642,28 @@ func TestGetOperationsLegacyStorageFailure(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestTransformOTLPSuccess(t *testing.T) {
+	withTestServer(func(ts *testServer) {
+		response := new(interface{})
+		request, err := readOTLPTraces(t)
+		require.NoError(t, err)
+		err = postJSON(ts.server.URL+"/api/transform", request, response)
+		require.NoError(t, err)
+		corectResponse, err := readJaegerTraces(t)
+		require.NoError(t, err)
+		assert.Equal(t, response, corectResponse)
+	}, querysvc.QueryServiceOptions{})
+}
+
+func TestTransformOTLPEmptyFailure(t *testing.T) {
+	withTestServer(func(ts *testServer) {
+		response := new(interface{})
+		request := ""
+		err := postJSON(ts.server.URL+"/api/transform", request, response)
+		require.Error(t, err)
+	}, querysvc.QueryServiceOptions{})
+}
+
 func TestGetMetricsSuccess(t *testing.T) {
 	mr := &metricsmocks.Reader{}
 	apiHandlerOptions := []HandlerOption{

cmd/query/app/otlp_translator.go

@@ -0,0 +1,60 @@
+// Copyright (c) 2024 The Jaeger Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package app
+
+import (
+	"fmt"
+
+	model2otel "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/jaegertracing/jaeger/model"
+)
+
+func otlp2model(otlpSpans []byte) ([]*model.Batch, error) {
+	ptraceUnmarshaler := ptrace.JSONUnmarshaler{}
+	otlpTraces, err := ptraceUnmarshaler.UnmarshalTraces(otlpSpans)
+	if err != nil {
+		return nil, fmt.Errorf("cannot unmarshal OTLP : %w", err)
+	}
+	jaegerBatches, err := model2otel.ProtoFromTraces(otlpTraces)
+	if err != nil {
+		return nil, fmt.Errorf("cannot transform OTLP to Jaeger: %w", err)
+	}
+	return jaegerBatches, nil
+}
+
+func batchesToTraces(jaegerBatches []*model.Batch) ([]*model.Trace, error) {
+	var traces []*model.Trace
+	traceMap := make(map[model.TraceID]*model.Trace)
+	for _, batch := range jaegerBatches {
+		for _, span := range batch.Spans {
+			if span.Process == nil {
+				span.Process = batch.Process
+			}
+			trace, ok := traceMap[span.TraceID]
+			if !ok {
+				newtrace := model.Trace{
+					Spans: []*model.Span{span},
+				}
+				traceMap[span.TraceID] = &newtrace
+				traces = append(traces, &newtrace)
+			} else {
+				trace.Spans = append(trace.Spans, span)
+			}
+		}
+	}
+	return traces, nil
+}

cmd/query/app/otlp_translator_test.go

@@ -0,0 +1,68 @@
+// Copyright (c) 2024 The Jaeger Authors.
+// SPDX-License-Identifier: Apache-2.0
+
+package app
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/jaegertracing/jaeger/model"
+)
+
+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.NoError(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)
+}