[api_v3][query] Change api_v3 http handler to use v2 query service

cmd/query/app/apiv3/gateway_test.go

@@ -24,8 +24,10 @@
 	"github.com/jaegertracing/jaeger/internal/proto/api_v3"
 	"github.com/jaegertracing/jaeger/model"
 	_ "github.com/jaegertracing/jaeger/pkg/gogocodec" // force gogo codec registration
+	"github.com/jaegertracing/jaeger/pkg/iter"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
-	spanstoremocks "github.com/jaegertracing/jaeger/storage/spanstore/mocks"
+	"github.com/jaegertracing/jaeger/storage_v2/tracestore"
+	tracestoremocks "github.com/jaegertracing/jaeger/storage_v2/tracestore/mocks"
 )
 
 // Utility functions used from http_gateway_test.go.
@@ -43,7 +45,7 @@
 )
 
 type testGateway struct {
-	reader *spanstoremocks.Reader
+	reader *tracestoremocks.Reader
 	url    string
 	router *mux.Router
 	// used to set a tenancy header when executing requests
@@ -85,7 +87,7 @@
 	require.NoError(t, gogojsonpb.Unmarshal(bytes.NewBuffer(body), obj))
 }
 
 func makeTestTrace() (*model.Trace, spanstore.GetTraceParameters) {
 	traceID := model.NewTraceID(150, 160)
 	query := spanstore.GetTraceParameters{TraceID: traceID}
 	return &model.Trace{
@@ -112,9 +114,9 @@
 	spanA := scopes.Spans().AppendEmpty()
 	spanA.SetName("foobar")
 	spanA.SetTraceID(traceID)
-	spanA.SetSpanID(pcommon.SpanID([8]byte{180}))
-	spanA.SetKind(ptrace.SpanKindServer)
-	spanA.Attributes().PutBool("error", true)
+	spanA.SetSpanID(pcommon.SpanID([8]byte{0, 0, 0, 0, 0, 0, 0, 2}))
+	// spanA.SetKind(ptrace.SpanKindServer)
+	// spanA.Attributes().PutBool("error", true)
 
 	return trace
 }
@@ -145,10 +147,10 @@
 }
 
 func (gw *testGateway) runGatewayGetOperations(t *testing.T) {
-	qp := spanstore.OperationQueryParameters{ServiceName: "foo", SpanKind: "server"}
+	qp := tracestore.OperationQueryParams{ServiceName: "foo", SpanKind: "server"}
 	gw.reader.
 		On("GetOperations", matchContext, qp).
-		Return([]spanstore.Operation{{Name: "get_users", SpanKind: "server"}}, nil).Once()
+		Return([]tracestore.Operation{{Name: "get_users", SpanKind: "server"}}, nil).Once()
 
 	body, statusCode := gw.execRequest(t, "/api/v3/operations?service=foo&span_kind=server")
 	require.Equal(t, http.StatusOK, statusCode)
@@ -162,21 +164,25 @@
 }
 
 func (gw *testGateway) runGatewayGetTrace(t *testing.T) {
-	trace, query := makeTestTrace()
-	gw.reader.On("GetTrace", matchContext, query).Return(trace, nil).Once()
-	gw.getTracesAndVerify(t, "/api/v3/traces/"+query.TraceID.String(), query.TraceID)
+	query := tracestore.GetTraceParams{TraceID: traceID}
+	gw.reader.
+		On("GetTraces", matchContext, query).
+		Return(iter.Seq2[[]ptrace.Traces, error](func(yield func([]ptrace.Traces, error) bool) {
+			yield([]ptrace.Traces{makeTestTraceV2()}, nil)
+		})).Once()
+	gw.getTracesAndVerify(t, "/api/v3/traces/1", traceID)
 }
 
 func (gw *testGateway) runGatewayFindTraces(t *testing.T) {
-	trace, query := makeTestTrace()
 	q, qp := mockFindQueries()
-	gw.reader.
-		On("FindTraces", matchContext, qp).
-		Return([]*model.Trace{trace}, nil).Once()
-	gw.getTracesAndVerify(t, "/api/v3/traces?"+q.Encode(), query.TraceID)
+	gw.reader.On("FindTraces", matchContext, qp).
+		Return(iter.Seq2[[]ptrace.Traces, error](func(yield func([]ptrace.Traces, error) bool) {
+			yield([]ptrace.Traces{makeTestTraceV2()}, nil)
+		})).Once()
+	gw.getTracesAndVerify(t, "/api/v3/traces?"+q.Encode(), traceID)
 }
 
-func (gw *testGateway) getTracesAndVerify(t *testing.T, url string, expectedTraceID model.TraceID) {
+func (gw *testGateway) getTracesAndVerify(t *testing.T, url string, expectedTraceID pcommon.TraceID) {
 	body, statusCode := gw.execRequest(t, url)
 	require.Equal(t, http.StatusOK, statusCode, "response=%s", string(body))
 	body = gw.verifySnapshot(t, body)

cmd/query/app/apiv3/http_gateway.go

@@ -20,10 +20,12 @@ import (
 	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 
-	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
+	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc/v2/querysvc"
 	"github.com/jaegertracing/jaeger/internal/proto/api_v3"
 	"github.com/jaegertracing/jaeger/model"
+	"github.com/jaegertracing/jaeger/pkg/iter"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
+	"github.com/jaegertracing/jaeger/storage_v2/tracestore"
 )
 
 const (
@@ -108,18 +110,8 @@ func (h *HTTPGateway) tryParamError(w http.ResponseWriter, err error, paramName
 	return h.tryHandleError(w, fmt.Errorf("malformed parameter %s: %w", paramName, err), http.StatusBadRequest)
 }
 
-func (h *HTTPGateway) returnSpans(spans []*model.Span, w http.ResponseWriter) {
-	// modelToOTLP does not easily return an error, so allow mocking it
-	h.returnSpansTestable(spans, w, modelToOTLP)
-}
-
-func (h *HTTPGateway) returnSpansTestable(
-	spans []*model.Span,
-	w http.ResponseWriter,
-	modelToOTLP func(_ []*model.Span) ptrace.Traces,
-) {
-	td := modelToOTLP(spans)
-	tracesData := api_v3.TracesData(td)
+func (h *HTTPGateway) returnTrace(t ptrace.Traces, w http.ResponseWriter) {
+	tracesData := api_v3.TracesData(t)
 	response := &api_v3.GRPCGatewayWrapper{
 		Result: &tracesData,
 	}
@@ -137,9 +129,11 @@ func (h *HTTPGateway) getTrace(w http.ResponseWriter, r *http.Request) {
 	if h.tryParamError(w, err, paramTraceID) {
 		return
 	}
-	request := querysvc.GetTraceParameters{
-		GetTraceParameters: spanstore.GetTraceParameters{
-			TraceID: traceID,
+	request := querysvc.GetTraceParams{
+		TraceIDs: []tracestore.GetTraceParams{
+			{
+				TraceID: traceID.ToOTELTraceID(),
+			},
 		},
 	}
 	http_query := r.URL.Query()
@@ -149,15 +143,15 @@ func (h *HTTPGateway) getTrace(w http.ResponseWriter, r *http.Request) {
 		if h.tryParamError(w, err, paramStartTime) {
 			return
 		}
-		request.StartTime = timeParsed.UTC()
+		request.TraceIDs[0].Start = timeParsed.UTC()
 	}
 	endTime := http_query.Get(paramEndTime)
 	if endTime != "" {
 		timeParsed, err := time.Parse(time.RFC3339Nano, endTime)
 		if h.tryParamError(w, err, paramEndTime) {
 			return
 		}
-		request.EndTime = timeParsed.UTC()
+		request.TraceIDs[0].End = timeParsed.UTC()
 	}
 	if r := http_query.Get(paramRawTraces); r != "" {
 		rawTraces, err := strconv.ParseBool(r)
@@ -166,11 +160,24 @@ func (h *HTTPGateway) getTrace(w http.ResponseWriter, r *http.Request) {
 		}
 		request.RawTraces = rawTraces
 	}
-	trc, err := h.QueryService.GetTrace(r.Context(), request)
+	getTracesIter := h.QueryService.GetTraces(r.Context(), request)
+	trc, err := iter.FlattenWithErrors(getTracesIter)
+
 	if h.tryHandleError(w, err, http.StatusInternalServerError) {
 		return
 	}
-	h.returnSpans(trc.Spans, w)
+	if len(trc) == 0 {
+		errorResponse := api_v3.GRPCGatewayError{
+			Error: &api_v3.GRPCGatewayError_GRPCGatewayErrorDetails{
+				HttpCode: http.StatusNotFound,
+				Message:  err.Error(),
+			},
+		}
+		resp, _ := json.Marshal(&errorResponse)
+		http.Error(w, string(resp), http.StatusNotFound)
+		return
+	}
+	h.returnTrace(trc[0], w)
 }
 
 func (h *HTTPGateway) findTraces(w http.ResponseWriter, r *http.Request) {
@@ -179,21 +186,26 @@ func (h *HTTPGateway) findTraces(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	traces, err := h.QueryService.FindTraces(r.Context(), queryParams)
+	findTracesIter := h.QueryService.FindTraces(r.Context(), *queryParams)
+	traces, err := iter.FlattenWithErrors(findTracesIter)
 	// TODO how do we distinguish internal error from bad parameters for FindTrace?
 	if h.tryHandleError(w, err, http.StatusInternalServerError) {
 		return
 	}
-	var spans []*model.Span
+	combinedTrace := ptrace.NewTraces()
 	for _, t := range traces {
-		spans = append(spans, t.Spans...)
+		resources := t.ResourceSpans()
+		for i := 0; i < resources.Len(); i++ {
+			resource := resources.At(i)
+			resource.CopyTo(combinedTrace.ResourceSpans().AppendEmpty())
+		}
 	}
-	h.returnSpans(spans, w)
+	h.returnTrace(combinedTrace, w)
 }
 
-func (h *HTTPGateway) parseFindTracesQuery(q url.Values, w http.ResponseWriter) (*querysvc.TraceQueryParameters, bool) {
-	queryParams := &querysvc.TraceQueryParameters{
-		TraceQueryParameters: spanstore.TraceQueryParameters{
+func (h *HTTPGateway) parseFindTracesQuery(q url.Values, w http.ResponseWriter) (*querysvc.TraceQueryParams, bool) {
+	queryParams := &querysvc.TraceQueryParams{
+		TraceQueryParams: tracestore.TraceQueryParams{
 			ServiceName:   q.Get(paramServiceName),
 			OperationName: q.Get(paramOperationName),
 			Tags:          nil, // most curiously not supported by grpc-gateway
@@ -262,7 +274,7 @@ func (h *HTTPGateway) getServices(w http.ResponseWriter, r *http.Request) {
 
 func (h *HTTPGateway) getOperations(w http.ResponseWriter, r *http.Request) {
 	query := r.URL.Query()
-	queryParams := spanstore.OperationQueryParameters{
+	queryParams := tracestore.OperationQueryParams{
 		ServiceName: query.Get("service"),
 		SpanKind:    query.Get("span_kind"),
 	}