Implemented chunked encoding

cmd/query/app/apiv3/http_gateway.go

@@ -137,8 +137,7 @@ func (h *HTTPGateway) returnTraces(traces []ptrace.Traces, err error, w http.Res
 		http.Error(w, string(resp), http.StatusNotFound)
 		return
 	}
-	// TODO: the response should be streamed back to the client
-	// https://github.com/jaegertracing/jaeger/issues/6467
+
 	combinedTrace := ptrace.NewTraces()
 	for _, t := range traces {
 		resources := t.ResourceSpans()
@@ -150,6 +149,79 @@ func (h *HTTPGateway) returnTraces(traces []ptrace.Traces, err error, w http.Res
 	h.returnTrace(combinedTrace, w)
 }
 
+func (h *HTTPGateway) streamTraces(tracesIter func(yield func([]ptrace.Traces, error) bool), w http.ResponseWriter) {
+	flusher, ok := w.(http.Flusher)
+	if !ok {
+		traces, err := jiter.FlattenWithErrors(tracesIter)
+		h.returnTraces(traces, err, w)
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	w.Header().Set("Content-Encoding", "identity")
+
+	tracesFound := false
+	firstChunk := true
+	hasError := false
+
+	tracesIter(func(traces []ptrace.Traces, err error) bool {
+		if err != nil {
+			if firstChunk {
+				h.tryHandleError(w, err, http.StatusInternalServerError)
+			} else {
+				h.Logger.Error("Error while streaming traces", zap.Error(err))
+			}
+			hasError = true
+			return false
+		}
+
+		if len(traces) == 0 {
+			return true
+		}
+
+		tracesFound = true
+
+		for _, td := range traces {
+			tracesData := jptrace.TracesData(td)
+			response := &api_v3.GRPCGatewayWrapper{
+				Result: &tracesData,
+			}
+
+			if firstChunk {
+				w.WriteHeader(http.StatusOK)
+				firstChunk = false
+			}
+
+			marshaler := jsonpb.Marshaler{}
+			if err := marshaler.Marshal(w, response); err != nil {
+				h.Logger.Error("Failed to marshal trace chunk", zap.Error(err))
+				return false
+			}
+
+			if _, err := w.Write([]byte("\n")); err != nil {
+				h.Logger.Error("Failed to write chunk separator", zap.Error(err))
+				return false
+			}
+
+			flusher.Flush()
+		}
+
+		return true
+	})
+
+	if !tracesFound && !hasError {
+		errorResponse := api_v3.GRPCGatewayError{
+			Error: &api_v3.GRPCGatewayError_GRPCGatewayErrorDetails{
+				HttpCode: http.StatusNotFound,
+				Message:  "No traces found",
+			},
+		}
+		resp, _ := json.Marshal(&errorResponse)
+		http.Error(w, string(resp), http.StatusNotFound)
+	}
+}
+
 func (*HTTPGateway) marshalResponse(response proto.Message, w http.ResponseWriter) {
 	_ = new(jsonpb.Marshaler).Marshal(w, response)
 }
@@ -193,8 +265,7 @@ func (h *HTTPGateway) getTrace(w http.ResponseWriter, r *http.Request) {
 		request.RawTraces = rawTraces
 	}
 	getTracesIter := h.QueryService.GetTraces(r.Context(), request)
-	trc, err := jiter.FlattenWithErrors(getTracesIter)
-	h.returnTraces(trc, err, w)
+	h.streamTraces(getTracesIter, w)
 }
 
 func (h *HTTPGateway) findTraces(w http.ResponseWriter, r *http.Request) {
@@ -204,8 +275,7 @@ func (h *HTTPGateway) findTraces(w http.ResponseWriter, r *http.Request) {
 	}
 
 	findTracesIter := h.QueryService.FindTraces(r.Context(), *queryParams)
-	traces, err := jiter.FlattenWithErrors(findTracesIter)
-	h.returnTraces(traces, err, w)
+	h.streamTraces(findTracesIter, w)
 }
 
 func (h *HTTPGateway) parseFindTracesQuery(q url.Values, w http.ResponseWriter) (*querysvc.TraceQueryParams, bool) {