respect grpc standard status

Author: JayiceZ

app/victoria-traces/main.go

@@ -123,7 +123,7 @@ func httpRequestHandler(w http.ResponseWriter, r *http.Request) bool {
 }
 
 func http2RequestHandler(w http.ResponseWriter, r *http.Request) {
-	vtinsert.GrpcInsertHandler(w, r)
+	vtinsert.GrpcExportHandler(w, r)
 }
 
 func usage() {

app/vtinsert/main.go

@@ -50,8 +50,8 @@ func RequestHandler(w http.ResponseWriter, r *http.Request) bool {
 	return false
 }
 
-func GrpcInsertHandler(w http.ResponseWriter, r *http.Request) {
-	opentelemetry.GrpcRequestHandler(r, w)
+func GrpcExportHandler(w http.ResponseWriter, r *http.Request) {
+	opentelemetry.GrpcExportHandler(r, w)
 }
 
 func insertHandler(w http.ResponseWriter, r *http.Request, path string) bool {

app/vtinsert/opentelemetry/grpc.go

@@ -10,10 +10,31 @@ import (
 	"net/http"
 )
 
+// https://grpc.github.io/grpc/core/md_doc_statuscodes.html
+const (
+	GrpcOk                 = "0"
+	GrpcCancelled          = "1"
+	GrpcUnknown            = "2"
+	GrpcInvalidArgument    = "3"
+	GrpcDeadlineExceeded   = "4"
+	GrpcNotFound           = "5"
+	GrpcAlreadyExist       = "6"
+	GrpcPermissionDenied   = "7"
+	GrpcResourceExhausted  = "8"
+	GrpcFailedPrecondition = "9"
+	GrpcAbort              = "10"
+	GrpcOutOfRange         = "11"
+	GrpcUnimplemented      = "12"
+	GrpcInternal           = "13"
+	GrpcUnavailable        = "14"
+	GrpcDataLoss           = "15"
+	GrpcUnauthenticated    = "16"
+)
+
 func getProtobufData(r *http.Request) ([]byte, error) {
 	reqBody, err := io.ReadAll(r.Body)
 	if err != nil {
-		return nil, &httpserver.ErrorWithStatusCode{StatusCode: http.StatusBadRequest, Err: fmt.Errorf("cannot read request body: %s", err)}
+		return nil, &httpserver.ErrorWithStatusCode{StatusCode: http.StatusInternalServerError, Err: fmt.Errorf("cannot read request body: %s", err)}
 	}
 	// +--------+-------------------------------------------------+
 	// | 1 byte |                    4 bytes                      |
@@ -26,6 +47,7 @@ func getProtobufData(r *http.Request) ([]byte, error) {
 	// |                 (variable length)                        |
 	// |                                                          |
 	// +----------------------------------------------------------+
+	// See https://grpc.github.io/grpc/core/md_doc__p_r_o_t_o_c_o_l-_h_t_t_p2.html
 	if len(reqBody) < 5 {
 		return nil, &httpserver.ErrorWithStatusCode{StatusCode: http.StatusBadRequest, Err: fmt.Errorf("invalid grpc header length: %d", len(reqBody))}
 	}
@@ -40,17 +62,27 @@ func getProtobufData(r *http.Request) ([]byte, error) {
 	return reqBody[5:], nil
 }
 
+func writeErrorGrpcResponse(w http.ResponseWriter, grpcErrorCode, grpcErrorMessage string) {
+	w.Header().Set("Content-Type", "application/grpc+proto")
+	w.Header().Set("Trailer", "grpc-status, grpc-message")
+	w.Header().Set("Grpc-Status", grpcErrorCode)
+	w.Header().Set("Grpc-Message", grpcErrorMessage)
+}
+
 // https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#message-encoding
-func writeExportTraceResponses(w http.ResponseWriter, rejectedSpans int64, errorMessage string) {
+func writeExportTracesGrpcResponse(w http.ResponseWriter, rejectedSpans int64, errorMessage string) {
 	resp := pb.ExportTraceServiceResponse{
 		ExportTracePartialSuccess: pb.ExportTracePartialSuccess{
 			RejectedSpans: rejectedSpans,
 			ErrorMessage:  errorMessage,
 		},
 	}
 	respData := resp.MarshalProtobuf(nil)
+
 	grpcRespData := make([]byte, 5+len(respData))
+	// Compressed flag
 	grpcRespData[0] = 0
+	// Message Length
 	binary.BigEndian.PutUint32(grpcRespData[1:5], uint32(len(respData)))
 	copy(grpcRespData[5:], respData)
 	w.Header().Set("Content-Type", "application/grpc+proto")
@@ -61,12 +93,14 @@ func writeExportTraceResponses(w http.ResponseWriter, rejectedSpans int64, error
 		logger.Errorf("unexpected write of %d bytes in replying OLTP export grpc request, expected:%d", writtenLen, len(grpcRespData))
 		return
 	}
+	grpcStatus := GrpcOk
+
 	if err != nil {
-		logger.Errorf("failed to reply OLTP export grpc request , error:%s", err)
-		return
+		grpcStatus = GrpcInternal
+		grpcErrorMessage := fmt.Sprintf("failed to reply OLTP export grpc request , error:%s", err)
+		logger.Errorf(grpcErrorMessage)
+		w.Header().Set("Grpc-Message", grpcErrorMessage)
 	}
 
-	w.Header().Set("Grpc-Status", "0")
-	w.Header().Set("Grpc-Message", "")
-
+	w.Header().Set("Grpc-Status", grpcStatus)
 }

app/vtinsert/opentelemetry/opentelemetry.go

@@ -161,14 +161,14 @@ func handleJSONRequest(r *http.Request, w http.ResponseWriter) {
 	requestJSONDuration.UpdateDuration(startTime)
 }
 
-func GrpcRequestHandler(r *http.Request, w http.ResponseWriter) {
+func GrpcExportHandler(r *http.Request, w http.ResponseWriter) {
 	if r.URL.Path != OLTPExportTracesGrpcPath {
 		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)
+		writeErrorGrpcResponse(w, GrpcInternal, fmt.Sprintf("cannot parse common params from request: %s", err))
 		return
 	}
 	// stream fields must contain the service name and span name.
@@ -177,13 +177,13 @@ func GrpcRequestHandler(r *http.Request, w http.ResponseWriter) {
 	cp.StreamFields = append(mandatoryStreamFields, cp.StreamFields...)
 
 	if err = insertutil.CanWriteData(); err != nil {
-		httpserver.Errorf(w, r, "%s", err)
+		writeErrorGrpcResponse(w, GrpcInternal, err.Error())
 		return
 	}
 
 	protobufData, err := getProtobufData(r)
 	if err != nil {
-		httpserver.Errorf(w, r, "failed to get protobuf data from request, error: %s", err)
+		writeErrorGrpcResponse(w, GrpcInvalidArgument, fmt.Sprintf("failed to get protobuf data from request, error: %s", err))
 		return
 	}
 	encoding := r.Header.Get("grpc-encoding")
@@ -203,11 +203,11 @@ func GrpcRequestHandler(r *http.Request, w http.ResponseWriter) {
 	})
 
 	if err != nil {
-		httpserver.Errorf(w, r, "cannot read OpenTelemetry protocol data: %s", err)
+		writeErrorGrpcResponse(w, GrpcInternal, fmt.Sprintf("cannot read OpenTelemetry protocol data: %s", err))
 		return
 	}
 
-	writeExportTraceResponses(w, 0, "")
+	writeExportTracesGrpcResponse(w, 0, "")
 	return
 }