Rework SLO metrics

CHANGELOG.md

@@ -13,6 +13,14 @@
 * [BUGFIX] Fix panic in metrics summary api [#2738](https://github.com/grafana/tempo/pull/2738) (@mdisibio)
 * [BUGFIX] Fix node role auth IDMSv1 [#2760](https://github.com/grafana/tempo/pull/2760) (@coufalja)
 * [BUGFIX] Only search ingester blocks that fall within the request time range. [#2783](https://github.com/grafana/tempo/pull/2783) (@joe-elliott)
+* [BUGFIX] Align tempo_query_frontend_queries_total and tempo_query_frontend_queries_within_slo_total. [#2840](https://github.com/grafana/tempo/pull/2840) (@joe-elliott)
+  This query will now correctly tell you %age of requests that are within SLO:
+  ```
+  sum(rate(tempo_query_frontend_queries_within_slo_total{}[1m])) by (op) 
+  /
+  sum(rate(tempo_query_frontend_queries_total{}[1m])) by (op)
+  ```
+  **BREAKING CHANGE** Removed: tempo_query_frontend_queries_total{op="searchtags|metrics"}. 
 * [CHANGE] Overrides module refactor [#2688](https://github.com/grafana/tempo/pull/2688) (@mapno)
     Added new `defaults` block to the overrides' module. Overrides change to indented syntax.
     Old config:
@@ -38,7 +46,7 @@ defaults:
   forwarders: ['foo']
   metrics_generator:
     processors: [service-graphs, span-metrics]
-```
+```  
 
 ## v2.2.1 / 2023-08-??
 

example/docker-compose/shared/tempo.yaml

@@ -1,6 +1,13 @@
 server:
   http_listen_port: 3200
 
+query_frontend:
+  search:
+    duration_slo: 5s
+    throughput_bytes_slo: 1.073741824e+09
+  trace_by_id:
+    duration_slo: 5s
+
 distributor:
   receivers:                           # this configuration will listen on all ports and protocols that tempo is capable of.
     jaeger:                            # the receives all come from the OpenTelemetry collector.  more configuration information can

integration/e2e/e2e_test.go

@@ -117,7 +117,7 @@ func TestAllInOne(t *testing.T) {
 			// test metrics
 			require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(1), "tempo_ingester_blocks_flushed_total"))
 			require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"tempodb_blocklist_length"}, e2e.WaitMissingMetrics))
-			require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(4), "tempo_query_frontend_queries_total"))
+			require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(3), "tempo_query_frontend_queries_total"))
 
 			// query trace - should fetch from backend
 			queryAndAssertTrace(t, apiClient, info)
@@ -271,7 +271,7 @@ func TestMicroservicesWithKVStores(t *testing.T) {
 				require.NoError(t, i.WaitSumMetrics(e2e.Equals(1), "tempo_ingester_blocks_flushed_total"))
 			}
 			require.NoError(t, tempoQuerier.WaitSumMetrics(e2e.Equals(3), "tempodb_blocklist_length"))
-			require.NoError(t, tempoQueryFrontend.WaitSumMetrics(e2e.Equals(4), "tempo_query_frontend_queries_total"))
+			require.NoError(t, tempoQueryFrontend.WaitSumMetrics(e2e.Equals(3), "tempo_query_frontend_queries_total"))
 
 			// query trace - should fetch from backend
 			queryAndAssertTrace(t, apiClient, info)

integration/e2e/encodings_test.go

@@ -99,7 +99,7 @@ func TestEncodings(t *testing.T) {
 			require.NoError(t, tempo.WaitSumMetrics(e2e.Equals(1), "tempo_ingester_blocks_flushed_total"))
 			require.NoError(t, tempo.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"tempodb_blocklist_length"}, e2e.WaitMissingMetrics))
 			if enc.Version() != v2.VersionString {
-				require.NoError(t, tempo.WaitSumMetrics(e2e.Greater(20), "tempo_query_frontend_queries_total"))
+				require.NoError(t, tempo.WaitSumMetrics(e2e.Greater(15), "tempo_query_frontend_queries_total"))
 			}
 
 			// query trace - should fetch from backend

modules/frontend/frontend.go

@@ -17,21 +17,13 @@ import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promauto"
 
 	"github.com/grafana/tempo/modules/overrides"
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/tempodb"
 )
 
-const (
-	traceByIDOp  = "traces"
-	searchOp     = "search"
-	searchTagsOp = "searchtags"
-	metricsOp    = "metrics"
-)
-
 type streamingSearchHandler func(req *tempopb.SearchRequest, srv tempopb.StreamingQuerier_SearchServer) error
 
 type QueryFrontend struct {
@@ -60,12 +52,6 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 		return nil, fmt.Errorf("query backend after should be less than or equal to query ingester until")
 	}
 
-	queriesPerTenant := promauto.With(registerer).NewCounterVec(prometheus.CounterOpts{
-		Namespace: "tempo",
-		Name:      "query_frontend_queries_total",
-		Help:      "Total queries received per tenant.",
-	}, []string{"tenant", "op", "status"})
-
 	retryWare := newRetryWare(cfg.MaxRetries, registerer)
 
 	// tracebyid middleware
@@ -75,21 +61,16 @@ func New(cfg Config, next http.RoundTripper, o overrides.Interface, reader tempo
 
 	spanMetricsMiddleware := MergeMiddlewares(newSpanMetricsMiddleware(), retryWare)
 
-	traceByIDCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": traceByIDOp})
-	searchCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": searchOp})
-	searchTagsCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": searchTagsOp})
-	spanMetricsCounter := queriesPerTenant.MustCurryWith(prometheus.Labels{"op": metricsOp})
-
 	traces := traceByIDMiddleware.Wrap(next)
 	search := searchMiddleware.Wrap(next)
 	searchTags := searchTagsMiddleware.Wrap(next)
 	metrics := spanMetricsMiddleware.Wrap(next)
 
 	return &QueryFrontend{
-		TraceByIDHandler:          newHandler(traces, traceByIDCounter, logger),
-		SearchHandler:             newHandler(search, searchCounter, logger),
-		SearchTagsHandler:         newHandler(searchTags, searchTagsCounter, logger),
-		SpanMetricsSummaryHandler: newHandler(metrics, spanMetricsCounter, logger),
+		TraceByIDHandler:          newHandler(traces, traceByIDSLOPostHook(cfg.TraceByID.SLO), nil, logger),
+		SearchHandler:             newHandler(search, searchSLOPostHook(cfg.Search.SLO), searchSLOPreHook, logger),
+		SearchTagsHandler:         newHandler(searchTags, nil, nil, logger),
+		SpanMetricsSummaryHandler: newHandler(metrics, nil, nil, logger),
 		streamingSearch:           newSearchStreamingHandler(cfg, o, retryWare.Wrap(next), reader, apiPrefix, logger),
 		logger:                    logger,
 	}, nil
@@ -191,7 +172,7 @@ func newTraceByIDMiddleware(cfg Config, logger log.Logger) Middleware {
 // newSearchMiddleware creates a new frontend middleware to handle search and search tags requests.
 func newSearchMiddleware(cfg Config, o overrides.Interface, reader tempodb.Reader, logger log.Logger) Middleware {
 	return MiddlewareFunc(func(next http.RoundTripper) http.RoundTripper {
-		searchRT := NewRoundTripper(next, newSearchSharder(reader, o, cfg.Search.Sharder, cfg.Search.SLO, newSearchProgress, logger))
+		searchRT := NewRoundTripper(next, newSearchSharder(reader, o, cfg.Search.Sharder, newSearchProgress, logger))
 
 		return RoundTripperFunc(func(r *http.Request) (*http.Response, error) {
 			// backend search queries require sharding, so we pass through a special roundtripper

modules/frontend/frontend_test.go

@@ -13,6 +13,11 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+var testSLOcfg = SLOConfig{
+	ThroughputBytesSLO: 0,
+	DurationSLO:        0,
+}
+
 type mockNextTripperware struct{}
 
 func (s *mockNextTripperware) RoundTrip(_ *http.Request) (*http.Response, error) {

modules/frontend/handler.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"io"
 	"net/http"
-	"strconv"
 	"strings"
 	"time"
 
@@ -16,7 +15,6 @@ import (
 	"github.com/grafana/dskit/tracing"
 	"github.com/grafana/dskit/user"
 	"github.com/opentracing/opentracing-go"
-	"github.com/prometheus/client_golang/prometheus"
 )
 
 const (
@@ -32,20 +30,27 @@ var (
 	errRequestEntityTooLarge = httpgrpc.Errorf(http.StatusRequestEntityTooLarge, "http: request body too large")
 )
 
+type (
+	handlerPostHook func(ctx context.Context, resp *http.Response, tenant string, latency time.Duration, err error)
+	handlerPreHook  func(ctx context.Context) context.Context
+)
+
 // handler exists to wrap a roundtripper with an HTTP handler. It wraps all
 // frontend endpoints and should only contain functionality that is common to all.
 type handler struct {
-	roundTripper     http.RoundTripper
-	logger           log.Logger
-	queriesPerTenant *prometheus.CounterVec
+	roundTripper http.RoundTripper
+	logger       log.Logger
+	post         handlerPostHook
+	pre          handlerPreHook
 }
 
 // newHandler creates a handler
-func newHandler(rt http.RoundTripper, queriesPerTenant *prometheus.CounterVec, logger log.Logger) http.Handler {
+func newHandler(rt http.RoundTripper, post handlerPostHook, pre handlerPreHook, logger log.Logger) http.Handler {
 	return &handler{
-		roundTripper:     rt,
-		logger:           logger,
-		queriesPerTenant: queriesPerTenant,
+		roundTripper: rt,
+		logger:       logger,
+		post:         post,
+		pre:          pre,
 	}
 }
 
@@ -60,27 +65,31 @@ func (f *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	orgID, _ := user.ExtractOrgID(ctx)
 	traceID, _ := tracing.ExtractTraceID(ctx)
 
-	var statusCode int
-	defer func(status int) {
-		f.queriesPerTenant.WithLabelValues(orgID, strconv.Itoa(status)).Inc()
-	}(statusCode)
-
 	// add orgid to existing spans
 	span := opentracing.SpanFromContext(r.Context())
 	if span != nil {
 		span.SetTag("orgID", orgID)
 	}
 
+	if f.pre != nil {
+		ctx = f.pre(ctx)
+		r = r.WithContext(ctx)
+	}
 	resp, err := f.roundTripper.RoundTrip(r)
+	elapsed := time.Since(start)
+	if f.post != nil {
+		f.post(ctx, resp, orgID, elapsed, err)
+	}
+
 	if err != nil {
-		statusCode = http.StatusInternalServerError
+		statusCode := http.StatusInternalServerError
 		err = writeError(w, err)
 		level.Info(f.logger).Log(
 			"tenant", orgID,
 			"method", r.Method,
 			"traceID", traceID,
 			"url", r.URL.RequestURI(),
-			"duration", time.Since(start).String(),
+			"duration", elapsed.String(),
 			"response_size", 0,
 			"status", statusCode,
 			"err", err.Error(),
@@ -89,14 +98,14 @@ func (f *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 
 	if resp == nil {
-		statusCode = http.StatusInternalServerError
+		statusCode := http.StatusInternalServerError
 		err = writeError(w, errors.New(NilResponseError))
 		level.Info(f.logger).Log(
 			"tenant", orgID,
 			"method", r.Method,
 			"traceID", traceID,
 			"url", r.URL.RequestURI(),
-			"duration", time.Since(start).String(),
+			"duration", elapsed.String(),
 			"response_size", 0,
 			"status", statusCode,
 			"err", err.Error(),
@@ -113,6 +122,7 @@ func (f *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	// request/response logging
 	var contentLength int64
+	var statusCode int
 	if httpResp, ok := httpgrpc.HTTPResponseFromError(err); ok {
 		statusCode = int(httpResp.Code)
 		contentLength = int64(len(httpResp.Body))
@@ -126,7 +136,7 @@ func (f *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		"method", r.Method,
 		"traceID", traceID,
 		"url", r.URL.RequestURI(),
-		"duration", time.Since(start).String(),
+		"duration", elapsed.String(),
 		"response_size", contentLength,
 		"status", statusCode,
 	)

modules/frontend/search_streaming.go

@@ -16,6 +16,7 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/atomic"
 
+	"github.com/grafana/dskit/user"
 	"github.com/grafana/tempo/modules/overrides"
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/tempopb"
@@ -94,6 +95,8 @@ func (p *diffSearchProgress) finalResult() *shardedSearchResults {
 
 // newSearchStreamingHandler returns a handler that streams results from the HTTP handler
 func newSearchStreamingHandler(cfg Config, o overrides.Interface, downstream http.RoundTripper, reader tempodb.Reader, apiPrefix string, logger log.Logger) streamingSearchHandler {
+	postSLOHook := searchSLOPostHook(cfg.Search.SLO)
+
 	downstreamPath := path.Join(apiPrefix, api.PathSearch)
 	return func(req *tempopb.SearchRequest, srv tempopb.StreamingQuerier_SearchServer) error {
 		// build search request and propagate context
@@ -110,6 +113,12 @@ func newSearchStreamingHandler(cfg Config, o overrides.Interface, downstream htt
 			return fmt.Errorf("build search request failed: %w", err)
 		}
 		ctx := srv.Context()
+
+		// SLOS - start timer and prep context
+		start := time.Now()
+		tenant, _ := user.ExtractOrgID(ctx)
+		ctx = searchSLOPreHook(ctx)
+
 		httpReq = httpReq.WithContext(ctx)
 
 		// streaming search only accepts requests with backend components
@@ -125,7 +134,7 @@ func newSearchStreamingHandler(cfg Config, o overrides.Interface, downstream htt
 			return p
 		}
 		// build roundtripper
-		rt := NewRoundTripper(downstream, newSearchSharder(reader, o, cfg.Search.Sharder, cfg.Search.SLO, fn, logger))
+		rt := NewRoundTripper(downstream, newSearchSharder(reader, o, cfg.Search.Sharder, fn, logger))
 
 		type roundTripResult struct {
 			resp *http.Response
@@ -138,6 +147,9 @@ func newSearchStreamingHandler(cfg Config, o overrides.Interface, downstream htt
 			resp, err := rt.RoundTrip(httpReq)
 			resultChan <- roundTripResult{resp, err}
 			close(resultChan)
+
+			// SLOs record results
+			postSLOHook(ctx, resp, tenant, time.Since(start), err)
 		}()
 
 		// collect and return results