support compression in grpc endpoint

Author: JayiceZ

app/vtinsert/opentelemetry/grpc.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
 	"github.com/VictoriaMetrics/VictoriaTraces/lib/protoparser/opentelemetry/pb"
 	"io"
 	"net/http"
@@ -29,7 +30,7 @@ func getProtobufData(r *http.Request) ([]byte, error) {
 		return nil, &httpserver.ErrorWithStatusCode{StatusCode: http.StatusBadRequest, Err: fmt.Errorf("invalid grpc header length: %d", len(reqBody))}
 	}
 	grpcHeader := reqBody[:5]
-	if isCompress := grpcHeader[0]; isCompress != 0 {
+	if isCompress := grpcHeader[0]; isCompress != 0 && isCompress != 1 {
 		return nil, &httpserver.ErrorWithStatusCode{StatusCode: http.StatusBadRequest, Err: fmt.Errorf("grpc compression not supporte")}
 	}
 	messageLength := binary.BigEndian.Uint32(grpcHeader[1:5])
@@ -55,7 +56,16 @@ func writeExportTraceResponses(w http.ResponseWriter, rejectedSpans int64, error
 	w.Header().Set("Content-Type", "application/grpc+proto")
 	w.Header().Set("Trailer", "grpc-status, grpc-message")
 
-	w.Write(grpcRespData)
+	writtenLen, err := w.Write(grpcRespData)
+	if writtenLen != len(grpcRespData) {
+		logger.Errorf("unexpected write of %d bytes in replying OLTP export grpc request, expected:%d", writtenLen, len(grpcRespData))
+		return
+	}
+	if err != nil {
+		logger.Errorf("failed to reply OLTP export grpc request , error:%s", err)
+		return
+	}
+
 	w.Header().Set("Grpc-Status", "0")
 	w.Header().Set("Grpc-Message", "")
 

app/vtinsert/opentelemetry/opentelemetry.go

@@ -1,6 +1,7 @@
 package opentelemetry
 
 import (
+	"bytes"
 	"fmt"
 	"net/http"
 	"strconv"
@@ -165,7 +166,6 @@ func GrpcRequestHandler(r *http.Request, w http.ResponseWriter) {
 		httpserver.Errorf(w, r, "failed to process grpc request:%s", &httpserver.ErrorWithStatusCode{Err: fmt.Errorf("grpc method not found: %s", r.URL.Path), StatusCode: http.StatusNotFound})
 		return
 	}
-
 	cp, err := insertutil.GetCommonParams(r)
 	if err != nil {
 		httpserver.Errorf(w, r, "cannot parse common params from request: %s", err)
@@ -186,15 +186,26 @@ func GrpcRequestHandler(r *http.Request, w http.ResponseWriter) {
 		httpserver.Errorf(w, r, "failed to get protobuf data from request, error: %s", err)
 		return
 	}
+	encoding := r.Header.Get("grpc-encoding")
+
+	err = protoparserutil.ReadUncompressedData(bytes.NewReader(protobufData), encoding, maxRequestSize, func(data []byte) error {
+		var (
+			req         otelpb.ExportTraceServiceRequest
+			callbackErr error
+		)
+		lmp := cp.NewLogMessageProcessor("opentelemetry_traces", false)
+		if callbackErr = req.UnmarshalProtobuf(data); callbackErr != nil {
+			return fmt.Errorf("cannot unmarshal request from %d protobuf bytes: %w", len(data), callbackErr)
+		}
+		callbackErr = pushExportTraceServiceRequest(&req, lmp)
+		lmp.MustClose()
+		return callbackErr
+	})
 
-	var req otelpb.ExportTraceServiceRequest
-	lmp := cp.NewLogMessageProcessor("opentelemetry_traces", false)
-	if err = req.UnmarshalProtobuf(protobufData); err != nil {
-		httpserver.Errorf(w, r, "cannot unmarshal request from %d protobuf bytes: %w", len(protobufData), err)
+	if err != nil {
+		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
 		return
 	}
-	_ = pushExportTraceServiceRequest(&req, lmp)
-	lmp.MustClose()
 
 	writeExportTraceResponses(w, 0, "")
 	return
@@ -232,7 +243,6 @@ func pushFieldsFromScopeSpans(ss *otelpb.ScopeSpans, commonFields []logstorage.F
 
 func pushFieldsFromSpan(span *otelpb.Span, scopeCommonFields []logstorage.Field, lmp insertutil.LogMessageProcessor) []logstorage.Field {
 	fields := scopeCommonFields
-	println(span.TraceID)
 	fields = append(fields,
 		logstorage.Field{Name: otelpb.TraceIDField, Value: span.TraceID},
 		logstorage.Field{Name: otelpb.SpanIDField, Value: span.SpanID},