[TraceQL] quantile_over_time 2 of 2 - engine

CHANGELOG.md

@@ -1,6 +1,7 @@
 ## main / unreleased
 
 * [FEATURE] Add TLS support for Memcached Client [#3585](https://github.com/grafana/tempo/pull/3585) (@sonisr)
+* [FEATURE] TraceQL metrics queries: add quantile_over_time [#3605](https://github.com/grafana/tempo/pull/3605) [#3633](https://github.com/grafana/tempo/pull/3633) (@mdisibio) 
 * [ENHANCEMENT] Add querier metrics for requests executed [#3524](https://github.com/grafana/tempo/pull/3524) (@electron0zero)
 * [FEATURE] Added gRPC streaming endpoints for Tempo APIs.
   * Added gRPC streaming endpoints for all tag queries. [#3460](https://github.com/grafana/tempo/pull/3460) (@joe-elliott)

modules/frontend/combiner/metrics_query_range.go

@@ -11,8 +11,11 @@ import (
 var _ GRPCCombiner[*tempopb.QueryRangeResponse] = (*genericCombiner[*tempopb.QueryRangeResponse])(nil)
 
 // NewQueryRange returns a query range combiner.
-func NewQueryRange() Combiner {
-	combiner := traceql.QueryRangeCombiner{}
+func NewQueryRange(req *tempopb.QueryRangeRequest) (Combiner, error) {
+	combiner, err := traceql.QueryRangeCombinerFor(req, traceql.AggregateModeFinal)
+	if err != nil {
+		return nil, err
+	}
 
 	return &genericCombiner[*tempopb.QueryRangeResponse]{
 		httpStatusCode: 200,
@@ -65,11 +68,16 @@ func NewQueryRange() Combiner {
 			sortResponse(resp)
 			return resp, nil
 		},
-	}
+	}, nil
 }
 
-func NewTypedQueryRange() GRPCCombiner[*tempopb.QueryRangeResponse] {
-	return NewQueryRange().(GRPCCombiner[*tempopb.QueryRangeResponse])
+func NewTypedQueryRange(req *tempopb.QueryRangeRequest) (GRPCCombiner[*tempopb.QueryRangeResponse], error) {
+	c, err := NewQueryRange(req)
+	if err != nil {
+		return nil, err
+	}
+
+	return c.(GRPCCombiner[*tempopb.QueryRangeResponse]), nil
 }
 
 func sortResponse(res *tempopb.QueryRangeResponse) {

modules/frontend/metrics_query_range_handler.go

@@ -38,14 +38,18 @@ func newQueryRangeStreamingGRPCHandler(cfg Config, next pipeline.AsyncRoundTripp
 		start := time.Now()
 
 		var finalResponse *tempopb.QueryRangeResponse
-		c := combiner.NewTypedQueryRange()
+		c, err := combiner.NewTypedQueryRange(req)
+		if err != nil {
+			return err
+		}
+
 		collector := pipeline.NewGRPCCollector(next, c, func(qrr *tempopb.QueryRangeResponse) error {
 			finalResponse = qrr // sadly we can't pass srv.Send directly into the collector. we need bytesProcessed for the SLO calculations
 			return srv.Send(qrr)
 		})
 
 		logQueryRangeRequest(logger, tenant, req)
-		err := collector.RoundTrip(httpReq)
+		err = collector.RoundTrip(httpReq)
 
 		duration := time.Since(start)
 		bytesProcessed := uint64(0)
@@ -80,7 +84,15 @@ func newMetricsQueryRangeHTTPHandler(cfg Config, next pipeline.AsyncRoundTripper
 		logQueryRangeRequest(logger, tenant, queryRangeReq)
 
 		// build and use roundtripper
-		combiner := combiner.NewTypedQueryRange()
+		combiner, err := combiner.NewTypedQueryRange(queryRangeReq)
+		if err != nil {
+			level.Error(logger).Log("msg", "query range: query range combiner failed", "err", err)
+			return &http.Response{
+				StatusCode: http.StatusInternalServerError,
+				Status:     http.StatusText(http.StatusInternalServerError),
+				Body:       io.NopCloser(strings.NewReader(err.Error())),
+			}, nil
+		}
 		rt := pipeline.NewHTTPCollector(next, combiner)
 
 		resp, err := rt.RoundTrip(req)

modules/frontend/metrics_query_range_handler_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/grafana/dskit/user"
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/tempopb"
+	v1 "github.com/grafana/tempo/pkg/tempopb/common/v1"
 	"github.com/stretchr/testify/require"
 )
 
@@ -22,13 +23,16 @@ func TestQueryRangeHandlerSucceeds(t *testing.T) {
 		Series: []*tempopb.TimeSeries{
 			{
 				PromLabels: "foo",
+				Labels: []v1.KeyValue{
+					{Key: "foo", Value: &v1.AnyValue{Value: &v1.AnyValue_StringValue{StringValue: "bar"}}},
+				},
 				Samples: []tempopb.Sample{
 					{
-						TimestampMs: 2,
+						TimestampMs: 1200_000,
 						Value:       2,
 					},
 					{
-						TimestampMs: 1,
+						TimestampMs: 1100_000,
 						Value:       1,
 					},
 				},
@@ -40,15 +44,17 @@ func TestQueryRangeHandlerSucceeds(t *testing.T) {
 		responseFn: func() proto.Message {
 			return resp
 		},
-	}, nil, nil, nil)
+	}, nil, nil, nil, func(c *Config) {
+		c.Metrics.Sharder.Interval = time.Hour
+	})
 	tenant := "foo"
 
 	httpReq := httptest.NewRequest("GET", api.PathMetricsQueryRange, nil)
 	httpReq = api.BuildQueryRangeRequest(httpReq, &tempopb.QueryRangeRequest{
 		Query: "{} | rate()",
-		Start: 1,
-		End:   uint64(10000 * time.Second),
-		Step:  uint64(1 * time.Second),
+		Start: uint64(1100 * time.Second),
+		End:   uint64(1200 * time.Second),
+		Step:  uint64(100 * time.Second),
 	})
 
 	ctx := user.InjectOrgID(httpReq.Context(), tenant)
@@ -63,24 +69,27 @@ func TestQueryRangeHandlerSucceeds(t *testing.T) {
 	// for reasons I don't understand, this query turns into 408 jobs.
 	expectedResp := &tempopb.QueryRangeResponse{
 		Metrics: &tempopb.SearchMetrics{
-			CompletedJobs:   408,
-			InspectedTraces: 408,
-			InspectedBytes:  408,
-			TotalJobs:       408,
+			CompletedJobs:   4,
+			InspectedTraces: 4,
+			InspectedBytes:  4,
+			TotalJobs:       4,
 			TotalBlocks:     2,
 			TotalBlockBytes: 419430400,
 		},
 		Series: []*tempopb.TimeSeries{
 			{
 				PromLabels: "foo",
+				Labels: []v1.KeyValue{
+					{Key: "foo", Value: &v1.AnyValue{Value: &v1.AnyValue_StringValue{StringValue: "bar"}}},
+				},
 				Samples: []tempopb.Sample{
 					{
-						TimestampMs: 1,
-						Value:       408,
+						TimestampMs: 1100_000,
+						Value:       4,
 					},
 					{
-						TimestampMs: 2,
-						Value:       816,
+						TimestampMs: 1200_000,
+						Value:       8,
 					},
 				},
 			},
@@ -102,13 +111,16 @@ func TestQueryRangeHandlerRespectsSamplingRate(t *testing.T) {
 		Series: []*tempopb.TimeSeries{
 			{
 				PromLabels: "foo",
+				Labels: []v1.KeyValue{
+					{Key: "foo", Value: &v1.AnyValue{Value: &v1.AnyValue_StringValue{StringValue: "bar"}}},
+				},
 				Samples: []tempopb.Sample{
 					{
-						TimestampMs: 2,
+						TimestampMs: 1200_000,
 						Value:       2,
 					},
 					{
-						TimestampMs: 1,
+						TimestampMs: 1100_000,
 						Value:       1,
 					},
 				},
@@ -120,15 +132,17 @@ func TestQueryRangeHandlerRespectsSamplingRate(t *testing.T) {
 		responseFn: func() proto.Message {
 			return resp
 		},
-	}, nil, nil, nil)
+	}, nil, nil, nil, func(c *Config) {
+		c.Metrics.Sharder.Interval = time.Hour
+	})
 	tenant := "foo"
 
 	httpReq := httptest.NewRequest("GET", api.PathMetricsQueryRange, nil)
 	httpReq = api.BuildQueryRangeRequest(httpReq, &tempopb.QueryRangeRequest{
 		Query: "{} | rate() with (sample=.2)",
-		Start: 1,
-		End:   uint64(10000 * time.Second),
-		Step:  uint64(1 * time.Second),
+		Start: uint64(1100 * time.Second),
+		End:   uint64(1200 * time.Second),
+		Step:  uint64(100 * time.Second),
 	})
 
 	ctx := user.InjectOrgID(httpReq.Context(), tenant)
@@ -140,27 +154,29 @@ func TestQueryRangeHandlerRespectsSamplingRate(t *testing.T) {
 
 	require.Equal(t, 200, httpResp.Code)
 
-	// for reasons I don't understand, this query turns into 408 jobs.
 	expectedResp := &tempopb.QueryRangeResponse{
 		Metrics: &tempopb.SearchMetrics{
-			CompletedJobs:   102,
-			InspectedTraces: 102,
-			InspectedBytes:  102,
-			TotalJobs:       102,
+			CompletedJobs:   1,
+			InspectedTraces: 1,
+			InspectedBytes:  1,
+			TotalJobs:       1,
 			TotalBlocks:     2,
 			TotalBlockBytes: 419430400,
 		},
 		Series: []*tempopb.TimeSeries{
 			{
 				PromLabels: "foo",
+				Labels: []v1.KeyValue{
+					{Key: "foo", Value: &v1.AnyValue{Value: &v1.AnyValue_StringValue{StringValue: "bar"}}},
+				},
 				Samples: []tempopb.Sample{
 					{
-						TimestampMs: 1,
-						Value:       510,
+						TimestampMs: 1100_000,
+						Value:       5,
 					},
 					{
-						TimestampMs: 2,
-						Value:       1020,
+						TimestampMs: 1200_000,
+						Value:       10,
 					},
 				},
 			},

modules/frontend/metrics_query_range_sharder.go

@@ -68,7 +68,7 @@ func (s queryRangeSharder) RoundTrip(r *http.Request) (pipeline.Responses[combin
 		return pipeline.NewBadRequest(err), nil
 	}
 
-	expr, err := traceql.Parse(req.Query)
+	expr, _, _, _, err := traceql.NewEngine().Compile(req.Query)
 	if err != nil {
 		return pipeline.NewBadRequest(err), nil
 	}

modules/frontend/search_handlers_test.go

@@ -659,7 +659,9 @@ func cacheResponsesEqual(t *testing.T, cacheResponse *tempopb.SearchResponse, pi
 
 // frontendWithSettings returns a new frontend with the given settings. any nil options
 // are given "happy path" defaults
-func frontendWithSettings(t *testing.T, next http.RoundTripper, rdr tempodb.Reader, cfg *Config, cacheProvider cache.Provider) *QueryFrontend {
+func frontendWithSettings(t *testing.T, next http.RoundTripper, rdr tempodb.Reader, cfg *Config, cacheProvider cache.Provider,
+	opts ...func(*Config),
+) *QueryFrontend {
 	if next == nil {
 		next = &mockRoundTripper{
 			responseFn: func() proto.Message {
@@ -721,6 +723,10 @@ func frontendWithSettings(t *testing.T, next http.RoundTripper, rdr tempodb.Read
 		}
 	}
 
+	for _, o := range opts {
+		o(cfg)
+	}
+
 	o, err := overrides.NewOverrides(overrides.Config{}, nil, prometheus.DefaultRegisterer)
 	require.NoError(t, err)
 

modules/generator/processor/localblocks/processor.go

@@ -466,6 +466,7 @@ func (p *Processor) QueryRange(ctx context.Context, req *tempopb.QueryRangeReque
 		concurrency = uint(v)
 	}
 
+	// Compile the sharded version of the query
 	eval, err := traceql.NewEngine().CompileMetricsQueryRange(req, false, timeOverlapCutoff, unsafe)
 	if err != nil {
 		return nil, err
@@ -519,7 +520,7 @@ func (p *Processor) QueryRange(ctx context.Context, req *tempopb.QueryRangeReque
 		return nil, err
 	}
 
-	return eval.Results()
+	return eval.Results(), nil
 }
 
 func (p *Processor) metricsCacheGet(key string) *traceqlmetrics.MetricsResults {

modules/querier/querier_query_range.go

@@ -47,7 +47,11 @@ func (q *Querier) queryRangeRecent(ctx context.Context, req *tempopb.QueryRangeR
 		return nil, fmt.Errorf("error querying generators in Querier.MetricsQueryRange: %w", err)
 	}
 
-	c := traceql.QueryRangeCombiner{}
+	c, err := traceql.QueryRangeCombinerFor(req, traceql.AggregateModeSum)
+	if err != nil {
+		return nil, err
+	}
+
 	for _, result := range lookupResults {
 		c.Combine(result.response.(*tempopb.QueryRangeResponse))
 	}
@@ -98,6 +102,7 @@ func (q *Querier) queryBackend(ctx context.Context, req *tempopb.QueryRangeReque
 		concurrency = v
 	}
 
+	// Compile the sharded version of the query
 	eval, err := traceql.NewEngine().CompileMetricsQueryRange(req, dedupe, timeOverlapCutoff, unsafe)
 	if err != nil {
 		return nil, err
@@ -139,10 +144,7 @@ func (q *Querier) queryBackend(ctx context.Context, req *tempopb.QueryRangeReque
 		return nil, err
 	}
 
-	res, err := eval.Results()
-	if err != nil {
-		return nil, err
-	}
+	res := eval.Results()
 
 	inspectedBytes, spansTotal, _ := eval.Metrics()