feature: [vt] support json encoding in opentelemery insert (#51)

related to #41

Opentelemetry support both protobuf and json encoding (https://opentelemetry.io/docs/specs/otlp/#otlphttp), so it will be nice if VictoriaTraces can accept json format traces.

Addition:
 - Regarding the mapping between json and protobuf, I followed the standards described in the [otel documentation](https://opentelemetry.io/docs/specs/otlp/#json-protobuf-encoding).  
 - How i test this feature？
     - I ran [otel demo](https://github.com/open-telemetry/opentelemetry-demo) to push json-format traces to VictoriaTraces.
     - I added a new case of json-format request in the integration test

---------

Signed-off-by: JayiceZ <[email protected]>
Co-authored-by: Jiekun <[email protected]>

Author: JayiceZ

app/victoria-traces/main.go

@@ -90,6 +90,7 @@ func requestHandler(w http.ResponseWriter, r *http.Request) bool {
 		})
 		return true
 	}
+
 	if vtinsert.RequestHandler(w, r) {
 		return true
 	}

app/vtinsert/opentelemetry/opentelemetry.go

@@ -22,9 +22,12 @@ var maxRequestSize = flagutil.NewBytes("opentelemetry.traces.maxRequestSize", 64
 
 var (
 	requestsProtobufTotal = metrics.NewCounter(`vt_http_requests_total{path="/insert/opentelemetry/v1/traces",format="protobuf"}`)
-	errorsTotal           = metrics.NewCounter(`vt_http_errors_total{path="/insert/opentelemetry/v1/traces",format="protobuf"}`)
+	errorsProtobufTotal   = metrics.NewCounter(`vt_http_errors_total{path="/insert/opentelemetry/v1/traces",format="protobuf"}`)
+	requestsJSONTotal     = metrics.NewCounter(`vt_http_requests_total{path="/insert/opentelemetry/v1/traces",format="json"}`)
+	errorsJSONTotal       = metrics.NewCounter(`vt_http_errors_total{path="/insert/opentelemetry/v1/traces",format="json"}`)
 
 	requestProtobufDuration = metrics.NewSummary(`vt_http_request_duration_seconds{path="/insert/opentelemetry/v1/traces",format="protobuf"}`)
+	requestJSONDuration     = metrics.NewSummary(`vt_http_request_duration_seconds{path="/insert/opentelemetry/v1/traces",format="json"}`)
 )
 
 var (
@@ -37,25 +40,37 @@ var (
 	traceIDCache = fastcache.New(32 * 1024 * 1024)
 )
 
+const (
+	contentTypeProtobuf = "application/x-protobuf"
+	contentTypeJSON     = "application/json"
+)
+
 // RequestHandler processes Opentelemetry insert requests
 func RequestHandler(path string, w http.ResponseWriter, r *http.Request) bool {
 	switch path {
 	// use the same path as opentelemetry collector
 	// https://opentelemetry.io/docs/specs/otlp/#otlphttp-request
 	case "/insert/opentelemetry/v1/traces":
-		if r.Header.Get("Content-Type") == "application/json" {
-			httpserver.Errorf(w, r, "json encoding isn't supported for opentelemetry format. Use protobuf encoding")
-			return true
-		}
-		handleProtobuf(r, w)
-		return true
+		return handleTracesRequest(r, w)
+	default:
+		return false
+	}
+}
+
+func handleTracesRequest(r *http.Request, w http.ResponseWriter) bool {
+	switch contentType := r.Header.Get("Content-Type"); contentType {
+	case contentTypeProtobuf:
+		handleProtobufRequest(r, w)
+	case contentTypeJSON:
+		handleJSONRequest(r, w)
 	default:
+		httpserver.Errorf(w, r, "Content-Type %s isn't supported for opentelemetry format. Use protobuf or JSON encoding", contentType)
 		return false
 	}
+	return true
 }
 
-// handleProtobuf handles the trace ingestion request.
-func handleProtobuf(r *http.Request, w http.ResponseWriter) {
+func handleProtobufRequest(r *http.Request, w http.ResponseWriter) {
 	startTime := time.Now()
 	requestsProtobufTotal.Inc()
 
@@ -69,36 +84,81 @@ func handleProtobuf(r *http.Request, w http.ResponseWriter) {
 	// for potentially better efficiency.
 	cp.StreamFields = append(mandatoryStreamFields, cp.StreamFields...)
 
-	if err := insertutil.CanWriteData(); err != nil {
+	if err = insertutil.CanWriteData(); err != nil {
 		httpserver.Errorf(w, r, "%s", err)
 		return
 	}
 
 	encoding := r.Header.Get("Content-Encoding")
 	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
-		lmp := cp.NewLogMessageProcessor("opentelemetry_traces_protobuf", false)
-		err := pushProtobufRequest(data, lmp)
+		var (
+			req         otelpb.ExportTraceServiceRequest
+			callbackErr error
+		)
+		lmp := cp.NewLogMessageProcessor("opentelemetry_traces", false)
+		if callbackErr = req.UnmarshalProtobuf(data); callbackErr != nil {
+			errorsProtobufTotal.Inc()
+			return fmt.Errorf("cannot unmarshal request from %d protobuf bytes: %w", len(data), callbackErr)
+		}
+		callbackErr = pushExportTraceServiceRequest(&req, lmp)
 		lmp.MustClose()
-		return err
+		return callbackErr
 	})
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
 		return
 	}
-
 	// update requestProtobufDuration only for successfully parsed requests
 	// There is no need in updating requestProtobufDuration for request errors,
 	// since their timings are usually much smaller than the timing for successful request parsing.
 	requestProtobufDuration.UpdateDuration(startTime)
 }
 
-func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor) error {
-	var req otelpb.ExportTraceServiceRequest
-	if err := req.UnmarshalProtobuf(data); err != nil {
-		errorsTotal.Inc()
-		return fmt.Errorf("cannot unmarshal request from %d bytes: %w", len(data), err)
+func handleJSONRequest(r *http.Request, w http.ResponseWriter) {
+	startTime := time.Now()
+	requestsJSONTotal.Inc()
+
+	cp, err := insertutil.GetCommonParams(r)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
+		return
+	}
+	// stream fields must contain the service name and span name.
+	// by using arguments and headers, users can also add other fields as stream fields
+	// for potentially better efficiency.
+	cp.StreamFields = append(mandatoryStreamFields, cp.StreamFields...)
+
+	if err = insertutil.CanWriteData(); err != nil {
+		httpserver.Errorf(w, r, "%s", err)
+		return
 	}
 
+	encoding := r.Header.Get("Content-Encoding")
+	err = protoparserutil.ReadUncompressedData(r.Body, encoding, maxRequestSize, func(data []byte) error {
+		var (
+			req         otelpb.ExportTraceServiceRequest
+			callbackErr error
+		)
+		lmp := cp.NewLogMessageProcessor("opentelemetry_traces", false)
+		if callbackErr = req.UnmarshalJSONCustom(data); callbackErr != nil {
+			errorsJSONTotal.Inc()
+			return fmt.Errorf("cannot unmarshal request from %d protobuf bytes: %w", len(data), callbackErr)
+		}
+		callbackErr = pushExportTraceServiceRequest(&req, lmp)
+		lmp.MustClose()
+		return callbackErr
+	})
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
+		return
+	}
+	// update requestJSONDuration only for successfully parsed requests
+	// There is no need in updating requestJSONDuration for request errors,
+	// since their timings are usually much smaller than the timing for successful request parsing.
+	requestJSONDuration.UpdateDuration(startTime)
+}
+
+func pushExportTraceServiceRequest(req *otelpb.ExportTraceServiceRequest, lmp insertutil.LogMessageProcessor) error {
 	var commonFields []logstorage.Field
 	for _, rs := range req.ResourceSpans {
 		commonFields = commonFields[:0]

docs/victoriatraces/changelog/CHANGELOG.md

@@ -17,7 +17,8 @@ The following `tip` changes can be tested by building VictoriaTraces components
 * FEATURE: [logstorage](https://docs.victoriametrics.com/victorialogs/): Upgrade VictoriaLogs dependency from [v1.27.0 to v1.33.1](https://github.com/VictoriaMetrics/VictoriaLogs/compare/v1.27.0...v1.33.1).
 * FEATURE: [docker compose](https://github.com/VictoriaMetrics/VictoriaTraces/tree/master/deployment/docker): add cluster docker compose environment.
 * FEATURE: [dashboards](https://github.com/VictoriaMetrics/VictoriaTraces/blob/master/dashboards): update dashboard for VictoriaTraces single-node and cluster to provide more charts.
-
+* FEATURE: [Single-node VictoriaTraces](https://docs.victoriametrics.com/victoriatraces/) and vtinsert in [VictoriaTraces cluster](https://docs.victoriametrics.com/victoriatraces/cluster/): support [JSON protobuf encoding](https://opentelemetry.io/docs/specs/otlp/#json-protobuf-encoding) in the OpenTelemetry protocol (OTLP) for data ingestion. See [this issue](https://github.com/VictoriaMetrics/VictoriaTraces/issues/41) for details. Thanks to @JayiceZ for the [pull request](https://github.com/VictoriaMetrics/VictoriaTraces/pull/51).
+* 
 * BUGFIX: [Single-node VictoriaTraces](https://docs.victoriametrics.com/victoriatraces/) and vtinsert in [VictoriaTraces cluster](https://docs.victoriametrics.com/victoriatraces/cluster/): Rename various [HTTP headers](https://docs.victoriametrics.com/victoriatraces/data-ingestion/#http-headers) prefix from `VL-` to `VT-`. These headers help with debugging and customizing stream fields. Thank @JayiceZ for [the pull request](https://github.com/VictoriaMetrics/VictoriaTraces/pull/56). 
 * BUGFIX: all components: properly expose metadata for summaries and histograms in VictoriaMetrics components with enabled `-metrics.exposeMetadata` cmd-line flag. See [metrics#98](https://github.com/VictoriaMetrics/metrics/issues/98) for details.
 

lib/protoparser/opentelemetry/pb/common.go

@@ -10,7 +10,7 @@ import (
 
 // Resource represents the corresponding OTEL protobuf message
 type Resource struct {
-	Attributes []*KeyValue
+	Attributes []*KeyValue `json:"attributes"`
 }
 
 // marshalProtobuf marshals
@@ -49,8 +49,8 @@ func (r *Resource) unmarshalProtobuf(src []byte) (err error) {
 
 // KeyValue represents the corresponding OTEL protobuf message
 type KeyValue struct {
-	Key   string
-	Value *AnyValue
+	Key   string    `json:"key"`
+	Value *AnyValue `json:"value"`
 }
 
 func (kv *KeyValue) marshalProtobuf(mm *easyproto.MessageMarshaler) {
@@ -95,13 +95,13 @@ func (kv *KeyValue) unmarshalProtobuf(src []byte) (err error) {
 
 // AnyValue represents the corresponding OTEL protobuf message
 type AnyValue struct {
-	StringValue  *string
-	BoolValue    *bool
-	IntValue     *int64
-	DoubleValue  *float64
-	ArrayValue   *ArrayValue
-	KeyValueList *KeyValueList
-	BytesValue   *[]byte
+	StringValue  *string       `json:"stringValue"`
+	BoolValue    *bool         `json:"boolValue"`
+	IntValue     *int64        `json:"intValue,string"`
+	DoubleValue  *float64      `json:"doubleValue"`
+	ArrayValue   *ArrayValue   `json:"arrayValue"`
+	KeyValueList *KeyValueList `json:"keyValueList"`
+	BytesValue   *[]byte       `json:"BytesValue"`
 }
 
 func (av *AnyValue) marshalProtobuf(mm *easyproto.MessageMarshaler) {
@@ -200,7 +200,7 @@ func (av *AnyValue) unmarshalProtobuf(src []byte) (err error) {
 
 // ArrayValue represents the corresponding OTEL protobuf message
 type ArrayValue struct {
-	Values []*AnyValue
+	Values []*AnyValue `json:"values"`
 }
 
 func (av *ArrayValue) marshalProtobuf(mm *easyproto.MessageMarshaler) {
@@ -238,7 +238,7 @@ func (av *ArrayValue) unmarshalProtobuf(src []byte) (err error) {
 
 // KeyValueList represents the corresponding OTEL protobuf message
 type KeyValueList struct {
-	Values []*KeyValue
+	Values []*KeyValue `json:"values"`
 }
 
 func (kvl *KeyValueList) marshalProtobuf(mm *easyproto.MessageMarshaler) {

lib/protoparser/opentelemetry/pb/testdata/json/1.bin

@@ -0,0 +1 @@
+{"resourceSpans":[{"resource":{"attributes":[{"key":"os.type","value":{"stringValue":"linux"}},{"key":"os.description","value":{"stringValue":"Debian GNU/Linux 12 (bookworm)"}},{"key":"os.build_id","value":{"stringValue":"6.10.14-linuxkit"}},{"key":"os.name","value":{"stringValue":"Debian GNU/Linux"}},{"key":"os.version","value":{"stringValue":"12"}},{"key":"host.name","value":{"stringValue":"8e296dd17f8c"}},{"key":"process.owner","value":{"stringValue":"app"}},{"key":"process.pid","value":{"intValue":"1"}},{"key":"process.runtime.description","value":{"stringValue":".NET 8.0.17"}},{"key":"process.runtime.name","value":{"stringValue":".NET"}},{"key":"process.runtime.version","value":{"stringValue":"8.0.17"}},{"key":"container.id","value":{"stringValue":"8e296dd17f8cbb2949a2483b6cc214d4e2a7faf7b9a71c71b0ea9927d6b4a4e1"}},{"key":"telemetry.distro.name","value":{"stringValue":"opentelemetry-dotnet-instrumentation"}},{"key":"telemetry.distro.version","value":{"stringValue":"1.11.0"}},{"key":"telemetry.sdk.name","value":{"stringValue":"opentelemetry"}},{"key":"telemetry.sdk.language","value":{"stringValue":"dotnet"}},{"key":"telemetry.sdk.version","value":{"stringValue":"1.11.2"}},{"key":"service.name","value":{"stringValue":"accounting"}},{"key":"service.namespace","value":{"stringValue":"opentelemetry-demo"}},{"key":"service.version","value":{"stringValue":"2.0.2"}}]},"scopeSpans":[{"scope":{"name":"OpenTelemetry.AutoInstrumentation.Kafka"},"spans":[{"traceId":"eaa99d2871f6e5d0bd9ac14bdad6629c","spanId":"5e9a917c821deb60","parentSpanId":"","flags":257,"name":"orders receive","kind":5,"startTimeUnixNano":"1757320933956646600","endTimeUnixNano":"1757320934273447500","attributes":[{"key":"messaging.client_id","value":{"stringValue":"rdkafka#consumer-1"}},{"key":"messaging.kafka.consumer.group","value":{"stringValue":"accounting"}},{"key":"messaging.operation","value":{"stringValue":"receive"}},{"key":"messaging.system","value":{"stringValue":"kafka"}},{"key":"messaging.destination.name","value":{"stringValue":"orders"}},{"key":"messaging.kafka.destination.partition","value":{"intValue":"0"}},{"key":"messaging.kafka.message.offset","value":{"intValue":"-1001"}}],"events":[{"timeUnixNano":"1757320934268187900","name":"exception","attributes":[{"key":"exception.message","value":{"stringValue":"Subscribed topic not available: orders: Broker: Unknown topic or partition"}},{"key":"exception.stacktrace","value":{"stringValue":"Confluent.Kafka.ConsumeException: Subscribed topic not available: orders: Broker: Unknown topic or partition\n   at Confluent.Kafka.Consumer`2.Consume(Int32 millisecondsTimeout)\n   at Confluent.Kafka.Consumer`2.Consume(CancellationToken cancellationToken)\n   at Accounting.Consumer.StartListening() in /src/Accounting/Consumer.cs:line 41"}},{"key":"exception.type","value":{"stringValue":"Confluent.Kafka.ConsumeException"}}]}],"status":{"message":"Subscribed topic not available: orders: Broker: Unknown topic or partition","code":2}}]}]},{"resource":{"attributes":[{"key":"service.name","value":{"stringValue":"frontend-proxy"}},{"key":"service.namespace","value":{"stringValue":"opentelemetry-demo"}},{"key":"service.version","value":{"stringValue":"2.0.2"}}]},"scopeSpans":[{"scope":{},"spans":[{"traceId":"ccfc43e08430655c257ebe80d3dd2bd4","spanId":"28d244b8e8788f7e","parentSpanId":"cbb37701f41cb9ee","name":"router frontend egress","kind":3,"startTimeUnixNano":"1757320936665313000","endTimeUnixNano":"1757320936733610000","attributes":[{"key":"http.protocol","value":{"stringValue":"HTTP/1.1"}},{"key":"upstream_address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"peer.address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"component","value":{"stringValue":"proxy"}},{"key":"upstream_cluster","value":{"stringValue":"frontend"}},{"key":"upstream_cluster.name","value":{"stringValue":"frontend"}},{"key":"http.status_code","value":{"stringValue":"200"}},{"key":"response_flags","value":{"stringValue":"-"}}],"status":{}},{"traceId":"ccfc43e08430655c257ebe80d3dd2bd4","spanId":"cbb37701f41cb9ee","parentSpanId":"","name":"ingress","kind":2,"startTimeUnixNano":"1757320936664715000","endTimeUnixNano":"1757320936733661000","attributes":[{"key":"node_id","value":{"stringValue":""}},{"key":"zone","value":{"stringValue":""}},{"key":"guid:x-request-id","value":{"stringValue":"56d080ef-fc25-9c26-8e42-979ee008027b"}},{"key":"http.url","value":{"stringValue":"http://frontend-proxy:8080/_next/static/chunks/framework-8c66923947fb0c35.js"}},{"key":"http.method","value":{"stringValue":"GET"}},{"key":"downstream_cluster","value":{"stringValue":"-"}},{"key":"user_agent","value":{"stringValue":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/136.0.0.0 Safari/537.36"}},{"key":"http.protocol","value":{"stringValue":"HTTP/1.1"}},{"key":"peer.address","value":{"stringValue":"172.18.0.25"}},{"key":"request_size","value":{"stringValue":"0"}},{"key":"response_size","value":{"stringValue":"57621"}},{"key":"component","value":{"stringValue":"proxy"}},{"key":"upstream_cluster","value":{"stringValue":"frontend"}},{"key":"upstream_cluster.name","value":{"stringValue":"frontend"}},{"key":"http.status_code","value":{"stringValue":"200"}},{"key":"response_flags","value":{"stringValue":"-"}}],"status":{}},{"traceId":"0f781dd1c80ba5328d134314c6003b48","spanId":"aa749a44860f70e7","parentSpanId":"55b80a9567dd47b6","name":"router frontend egress","kind":3,"startTimeUnixNano":"1757320936813439000","endTimeUnixNano":"1757320936827038000","attributes":[{"key":"http.protocol","value":{"stringValue":"HTTP/1.1"}},{"key":"upstream_address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"peer.address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"component","value":{"stringValue":"proxy"}},{"key":"upstream_cluster","value":{"stringValue":"frontend"}},{"key":"upstream_cluster.name","value":{"stringValue":"frontend"}},{"key":"http.status_code","value":{"stringValue":"200"}},{"key":"response_flags","value":{"stringValue":"-"}}],"status":{}},{"traceId":"0f781dd1c80ba5328d134314c6003b48","spanId":"55b80a9567dd47b6","parentSpanId":"","name":"ingress","kind":2,"startTimeUnixNano":"1757320936813279000","endTimeUnixNano":"1757320936827054000","attributes":[{"key":"node_id","value":{"stringValue":""}},{"key":"zone","value":{"stringValue":""}},{"key":"guid:x-request-id","value":{"stringValue":"ebc78a99-d4f3-9ac6-9e8c-a415e8a81df7"}},{"key":"http.url","value":{"stringValue":"http://frontend-proxy:8080/_next/static/_g5lJ7kKnYpQWGjSpoVo5/_ssgManifest.js"}},{"key":"http.method","value":{"stringValue":"GET"}},{"key":"downstream_cluster","value":{"stringValue":"-"}},{"key":"user_agent","value":{"stringValue":"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) HeadlessChrome/136.0.0.0 Safari/537.36"}},{"key":"http.protocol","value":{"stringValue":"HTTP/1.1"}},{"key":"peer.address","value":{"stringValue":"172.18.0.25"}},{"key":"request_size","value":{"stringValue":"0"}},{"key":"response_size","value":{"stringValue":"77"}},{"key":"component","value":{"stringValue":"proxy"}},{"key":"upstream_cluster","value":{"stringValue":"frontend"}},{"key":"upstream_cluster.name","value":{"stringValue":"frontend"}},{"key":"http.status_code","value":{"stringValue":"200"}},{"key":"response_flags","value":{"stringValue":"-"}}],"status":{}},{"traceId":"3c0a3fa93f41d1bc5f93acdc7cecb577","spanId":"d40390d9c3b5d1e5","parentSpanId":"e4ed749bdebb51a9","name":"router frontend egress","kind":3,"startTimeUnixNano":"1757320937207177000","endTimeUnixNano":"1757320937214023000","attributes":[{"key":"http.protocol","value":{"stringValue":"HTTP/1.1"}},{"key":"upstream_address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"peer.address","value":{"stringValue":"172.18.0.24:8080"}},{"key":"component","value":{"stringValue":"proxy"}},{"key":"upstream_cluster","value":{"stringValue":"frontend"}},{"key":"upstream_cluster.name","value":{"stringValue":"frontend"}},{"key":"http.status_code","value":{"stringValue":"200"}},{"key":"response_flags","value":{"stringValue":"-"}}],"status":{}}]}]}]}