Create Span for the codec MergeResponse method (#5075)

* Create Span for the codec MergeResponse method

Signed-off-by: Alan Protasio <[email protected]>

* Changelog

Signed-off-by: Alan Protasio <[email protected]>

Signed-off-by: Alan Protasio <[email protected]>

Author: alanprot

CHANGELOG.md

@@ -17,6 +17,7 @@
 * [FEATURE] Ingester: Enable snapshotting of In-memory TSDB on disk during shutdown via `-blocks-storage.tsdb.memory-snapshot-on-shutdown`. #5011
 * [FEATURE] Query Frontend/Scheduler: Add a new counter metric `cortex_request_queue_requests_total` for total requests going to queue. #5030
 * [FEATURE] Build ARM docker images. #5041
+* [FEATURE] Query-frontend/Querier: Create spans to measure time to merge promql responses. #5041
 * [BUGFIX] Updated `golang.org/x/net` dependency to fix CVE-2022-27664. #5008
 * [BUGFIX] Fix panic when otel and xray tracing is enabled. #5044
 * [BUGFIX] Fixed no compact block got grouped in shuffle sharding grouper. #5055

pkg/querier/tripperware/instantquery/instant_query.go

@@ -243,7 +243,11 @@ func (instantQueryCodec) EncodeResponse(ctx context.Context, res tripperware.Res
 	return &resp, nil
 }
 
-func (instantQueryCodec) MergeResponse(responses ...tripperware.Response) (tripperware.Response, error) {
+func (instantQueryCodec) MergeResponse(ctx context.Context, responses ...tripperware.Response) (tripperware.Response, error) {
+	sp, _ := opentracing.StartSpanFromContext(ctx, "PrometheusInstantQueryResponse.MergeResponse")
+	sp.SetTag("response_count", len(responses))
+	defer sp.Finish()
+
 	if len(responses) == 0 {
 		return NewEmptyPrometheusInstantQueryResponse(), nil
 	} else if len(responses) == 1 {

pkg/querier/tripperware/instantquery/instant_query_test.go

@@ -308,7 +308,7 @@ func TestMergeResponse(t *testing.T) {
 				require.NoError(t, err)
 				resps = append(resps, dr)
 			}
-			resp, err := InstantQueryCodec.MergeResponse(resps...)
+			resp, err := InstantQueryCodec.MergeResponse(context.Background(), resps...)
 			assert.Equal(t, err, tc.expectedErr)
 			if err != nil {
 				return

pkg/querier/tripperware/query.go

@@ -40,7 +40,7 @@ type Codec interface {
 // Merger is used by middlewares making multiple requests to merge back all responses into a single one.
 type Merger interface {
 	// MergeResponse merges responses from multiple requests into a single Response
-	MergeResponse(...Response) (Response, error)
+	MergeResponse(context.Context, ...Response) (Response, error)
 }
 
 // Response represents a query range response.

pkg/querier/tripperware/queryrange/query_range.go

@@ -127,7 +127,10 @@ func NewEmptyPrometheusResponse() *PrometheusResponse {
 	}
 }
 
-func (c prometheusCodec) MergeResponse(responses ...tripperware.Response) (tripperware.Response, error) {
+func (c prometheusCodec) MergeResponse(ctx context.Context, responses ...tripperware.Response) (tripperware.Response, error) {
+	sp, _ := opentracing.StartSpanFromContext(ctx, "QueryRangeResponse.MergeResponse")
+	sp.SetTag("response_count", len(responses))
+	defer sp.Finish()
 	if len(responses) == 0 {
 		return NewEmptyPrometheusResponse(), nil
 	}

pkg/querier/tripperware/queryrange/query_range_test.go

@@ -652,7 +652,7 @@ func TestMergeAPIResponses(t *testing.T) {
 			},
 		}} {
 		t.Run(tc.name, func(t *testing.T) {
-			output, err := PrometheusCodec.MergeResponse(tc.input...)
+			output, err := PrometheusCodec.MergeResponse(context.Background(), tc.input...)
 			require.NoError(t, err)
 			require.Equal(t, tc.expected, output)
 		})

pkg/querier/tripperware/queryrange/results_cache.go

@@ -407,7 +407,7 @@ func (s resultsCache) handleHit(ctx context.Context, r tripperware.Request, exte
 		return nil, nil, err
 	}
 	if len(requests) == 0 {
-		response, err := s.merger.MergeResponse(responses...)
+		response, err := s.merger.MergeResponse(context.Background(), responses...)
 		// No downstream requests so no need to write back to the cache.
 		return response, nil, err
 	}
@@ -469,7 +469,7 @@ func (s resultsCache) handleHit(ctx context.Context, r tripperware.Request, exte
 		if err != nil {
 			return nil, nil, err
 		}
-		merged, err := s.merger.MergeResponse(accumulator.Response, currentRes)
+		merged, err := s.merger.MergeResponse(ctx, accumulator.Response, currentRes)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -481,7 +481,7 @@ func (s resultsCache) handleHit(ctx context.Context, r tripperware.Request, exte
 		return nil, nil, err
 	}
 
-	response, err := s.merger.MergeResponse(responses...)
+	response, err := s.merger.MergeResponse(ctx, responses...)
 	return response, mergedExtents, err
 }
 

pkg/querier/tripperware/queryrange/split_by_interval.go

@@ -61,7 +61,7 @@ func (s splitByInterval) Do(ctx context.Context, r tripperware.Request) (tripper
 		resps = append(resps, reqResp.Response)
 	}
 
-	response, err := s.merger.MergeResponse(resps...)
+	response, err := s.merger.MergeResponse(ctx, resps...)
 	if err != nil {
 		return nil, err
 	}

pkg/querier/tripperware/queryrange/split_by_interval_test.go

@@ -266,7 +266,7 @@ func TestSplitQuery(t *testing.T) {
 }
 
 func TestSplitByDay(t *testing.T) {
-	mergedResponse, err := PrometheusCodec.MergeResponse(parsedResponse, parsedResponse)
+	mergedResponse, err := PrometheusCodec.MergeResponse(context.Background(), parsedResponse, parsedResponse)
 	require.NoError(t, err)
 
 	mergedHTTPResponse, err := PrometheusCodec.EncodeResponse(context.Background(), mergedResponse)

pkg/querier/tripperware/shard_by.go

@@ -78,7 +78,7 @@ func (s shardBy) Do(ctx context.Context, r Request) (Response, error) {
 		resps = append(resps, reqResp.Response)
 	}
 
-	return s.merger.MergeResponse(resps...)
+	return s.merger.MergeResponse(ctx, resps...)
 }
 
 func (s shardBy) shardQuery(l log.Logger, numShards int, r Request, analysis querysharding.QueryAnalysis) []Request {