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

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

@@ -0,0 +1,74 @@
+{
+  "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": {}
+            },
+            {
+              "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,98 @@
+{
+  "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
+        },
+        {
+          "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 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -128,6 +129,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 +195,22 @@ 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 := io.ReadAll(r.Body)
+	if aH.handleError(w, err, http.StatusBadRequest) {
+		return
+	}
+
+	traces, err := otlp2traces(body)
+	if aH.handleError(w, err, http.StatusInternalServerError) {
+		return
+	}
+
+	var uiErrors []structuredError
+	structuredRes := aH.tracesToResponse(traces, false, uiErrors)
+	aH.writeJSON(w, r, structuredRes)
+}
+
 func (aH *APIHandler) getOperations(w http.ResponseWriter, r *http.Request) {
 	service := r.FormValue(serviceParam)
 	if service == "" {
@@ -243,20 +261,24 @@ func (aH *APIHandler) search(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	uiTraces := make([]*ui.Trace, len(tracesFromStorage))
-	for i, v := range tracesFromStorage {
-		uiTrace, uiErr := aH.convertModelToUI(v, true)
+	structuredRes := aH.tracesToResponse(tracesFromStorage, true, uiErrors)
+	aH.writeJSON(w, r, structuredRes)
+}
+
+func (aH *APIHandler) tracesToResponse(traces []*model.Trace, adjust bool, uiErrors []structuredError) *structuredResponse {
+	uiTraces := make([]*ui.Trace, len(traces))
+	for i, v := range traces {
+		uiTrace, uiErr := aH.convertModelToUI(v, adjust)
 		if uiErr != nil {
 			uiErrors = append(uiErrors, *uiErr)
 		}
 		uiTraces[i] = uiTrace
 	}
 
-	structuredRes := structuredResponse{
+	return &structuredResponse{
 		Data:   uiTraces,
 		Errors: uiErrors,
 	}
-	aH.writeJSON(w, r, &structuredRes)
 }
 
 func (aH *APIHandler) tracesByIDs(ctx context.Context, traceIDs []model.TraceID) ([]*model.Trace, []structuredError, error) {
@@ -436,18 +458,8 @@ func (aH *APIHandler) getTrace(w http.ResponseWriter, r *http.Request) {
 	}
 
 	var uiErrors []structuredError
-	uiTrace, uiErr := aH.convertModelToUI(trace, shouldAdjust(r))
-	if uiErr != nil {
-		uiErrors = append(uiErrors, *uiErr)
-	}
-
-	structuredRes := structuredResponse{
-		Data: []*ui.Trace{
-			uiTrace,
-		},
-		Errors: uiErrors,
-	}
-	aH.writeJSON(w, r, &structuredRes)
+	structuredRes := aH.tracesToResponse([]*model.Trace{trace}, shouldAdjust(r), uiErrors)
+	aH.writeJSON(w, r, structuredRes)
 }
 
 func shouldAdjust(r *http.Request) bool {

cmd/query/app/http_handler_test.go

@@ -25,6 +25,7 @@ import (
 	"math"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"testing"
 	"time"
 
@@ -56,6 +57,15 @@ import (
 
 const millisToNanosMultiplier = int64(time.Millisecond / time.Nanosecond)
 
+type IoReaderMock struct {
+	mock.Mock
+}
+
+func (m *IoReaderMock) Read(b []byte) (int, error) {
+	args := m.Called(b)
+	return args.Int(0), args.Error(1)
+}
+
 var (
 	errStorageMsg = "storage error"
 	errStorage    = errors.New(errStorageMsg)
@@ -623,6 +633,52 @@ func TestGetOperationsLegacyStorageFailure(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestTransformOTLPSuccess(t *testing.T) {
+	reformat := func(in []byte) []byte {
+		obj := new(interface{})
+		require.NoError(t, json.Unmarshal(in, obj))
+		// format json similar to `jq .`
+		out, err := json.MarshalIndent(obj, "", "  ")
+		require.NoError(t, err)
+		return out
+	}
+	withTestServer(func(ts *testServer) {
+		inFile, err := os.Open("./fixture/otlp2jaeger-in.json")
+		require.NoError(t, err)
+
+		resp, err := ts.server.Client().Post(ts.server.URL+"/api/transform", "application/json", inFile)
+		require.NoError(t, err)
+
+		responseBytes, err := io.ReadAll(resp.Body)
+		require.NoError(t, err)
+		responseBytes = reformat(responseBytes)
+
+		expectedBytes, err := os.ReadFile("./fixture/otlp2jaeger-out.json")
+		require.NoError(t, err)
+		expectedBytes = reformat(expectedBytes)
+
+		assert.Equal(t, string(expectedBytes), string(responseBytes))
+	}, querysvc.QueryServiceOptions{})
+}
+
+func TestTransformOTLPReadError(t *testing.T) {
+	withTestServer(func(ts *testServer) {
+		bytesReader := &IoReaderMock{}
+		bytesReader.On("Read", mock.AnythingOfType("[]uint8")).Return(0, errors.New("Mocked error"))
+		_, err := ts.server.Client().Post(ts.server.URL+"/api/transform", "application/json", bytesReader)
+		require.Error(t, err)
+	}, querysvc.QueryServiceOptions{})
+}
+
+func TestTransformOTLPBadPayload(t *testing.T) {
+	withTestServer(func(ts *testServer) {
+		response := new(interface{})
+		request := "Bad Payload"
+		err := postJSON(ts.server.URL+"/api/transform", request, response)
+		require.ErrorContains(t, err, "cannot unmarshal OTLP")
+	}, querysvc.QueryServiceOptions{})
+}
+
 func TestGetMetricsSuccess(t *testing.T) {
 	mr := &metricsmocks.Reader{}
 	apiHandlerOptions := []HandlerOption{

cmd/query/app/otlp_translator.go

@@ -0,0 +1,55 @@
+// 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 otlp2traces(otlpSpans []byte) ([]*model.Trace, error) {
+	ptraceUnmarshaler := ptrace.JSONUnmarshaler{}
+	otlpTraces, err := ptraceUnmarshaler.UnmarshalTraces(otlpSpans)
+	if err != nil {
+		return nil, fmt.Errorf("cannot unmarshal OTLP : %w", err)
+	}
+	jaegerBatches, _ := model2otel.ProtoFromTraces(otlpTraces)
+	// ProtoFromTraces will not give an 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
+}