frontend: refactor response metrics collection (#4997)

* frontend: refactor metrics collection

The main goal is to provide accurate inspected bytes metrics (and othes) in responses for a potential future fair-use policy. Until now, we were counting inspected bytes regardless of the cache. With this PR, the responses that come from cache do not count towards inspected bytes.

- ensure that we collect response metrics in the same way for all endpoints.
- add cache header to querier responses, to indicate whether the response came from the cache
- set 0 to metrics from cached querier responses

* Update changelog

* Ignore gosec integer conversion rule in a few lines

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## main / unreleased
 
+* [CHANGE] Do not count cached querier responses for SLO metrics such as inspected bytes [#5185](https://github.com/grafana/tempo/pull/5185) (@carles-grafana)
 * [CHANGE] Assert max live traces limits in local-blocks processor [#5170](https://github.com/grafana/tempo/pull/5170) (@mapno)
 * [ENHANCEMENT] Include backendwork dashboard and include additional alert [#5159](https://github.com/grafana/tempo/pull/5159) (@zalegrala)
 * [ENHANCEMENT] Add alert for high error rate reported by vulture [#5206](https://github.com/grafana/tempo/pull/5206) (@ruslan-mikhailov)
@@ -7,6 +8,7 @@
 * [BUGFIX] Add nil check to partitionAssignmentVar [#5198](https://github.com/grafana/tempo/pull/5198) (@mapno)
 * [BUGFIX] Fix structural metrics rate by aggregation [#5204](https://github.com/grafana/tempo/pull/5204) (@zalegrala)
 
+
 # v2.8.0-rc.0
 
 * [CHANGE] **BREAKING CHANGE** Change default http-listen-port from 80 to 3200 [#4960](https://github.com/grafana/tempo/pull/4960) (@martialblog)

modules/frontend/combiner/header.go

@@ -0,0 +1,18 @@
+package combiner
+
+import "net/http"
+
+const (
+	TempoCacheHeader = "X-Tempo-Cache"
+
+	TempoCacheHit = "HIT"
+
+	TempoCacheMiss = "MISS"
+)
+
+func IsCacheHit(resp *http.Response) bool {
+	if resp == nil {
+		return false
+	}
+	return resp.Header.Get(TempoCacheHeader) == TempoCacheHit
+}

modules/frontend/combiner/metrics_query_range.go

@@ -30,22 +30,14 @@ func NewQueryRange(req *tempopb.QueryRangeRequest, maxSeriesLimit int) (Combiner
 	var prevResp *tempopb.QueryRangeResponse
 	maxSeriesReachedErrorMsg := fmt.Sprintf("Response exceeds maximum series limit of %d, a partial response is returned. Warning: the accuracy of each individual value is not guaranteed.", maxSeries)
 
+	metricsCombiner := NewQueryRangeMetricsCombiner()
 	c := &genericCombiner[*tempopb.QueryRangeResponse]{
 		httpStatusCode: 200,
 		new:            func() *tempopb.QueryRangeResponse { return &tempopb.QueryRangeResponse{} },
 		current:        &tempopb.QueryRangeResponse{Metrics: &tempopb.SearchMetrics{}},
-		combine: func(partial *tempopb.QueryRangeResponse, _ *tempopb.QueryRangeResponse, _ PipelineResponse) error {
-			if partial.Metrics != nil {
-				// this is a coordination between the sharder and combiner. the sharder returns one response with summary metrics
-				// only. the combiner correctly takes and accumulates that job. however, if the response has no jobs this is
-				// an indicator this is a "real" response so we set CompletedJobs to 1 to increment in the combiner.
-				if partial.Metrics.TotalJobs == 0 {
-					partial.Metrics.CompletedJobs = 1
-				}
-			}
-
+		combine: func(partial *tempopb.QueryRangeResponse, _ *tempopb.QueryRangeResponse, resp PipelineResponse) error {
 			combiner.Combine(partial)
-
+			metricsCombiner.Combine(partial.Metrics, resp)
 			return nil
 		},
 		finalize: func(_ *tempopb.QueryRangeResponse) (*tempopb.QueryRangeResponse, error) {
@@ -63,7 +55,7 @@ func NewQueryRange(req *tempopb.QueryRangeRequest, maxSeriesLimit int) (Combiner
 				resp.Message = maxSeriesReachedErrorMsg
 			}
 			attachExemplars(req, resp)
-
+			resp.Metrics = metricsCombiner.Metrics
 			return resp, nil
 		},
 		diff: func(_ *tempopb.QueryRangeResponse) (*tempopb.QueryRangeResponse, error) {
@@ -89,7 +81,7 @@ func NewQueryRange(req *tempopb.QueryRangeRequest, maxSeriesLimit int) (Combiner
 				diff.Status = tempopb.PartialStatus_PARTIAL
 				diff.Message = maxSeriesReachedErrorMsg
 			}
-
+			diff.Metrics = metricsCombiner.Metrics
 			return diff, nil
 		},
 		quit: func(_ *tempopb.QueryRangeResponse) bool {

modules/frontend/combiner/response_metrics.go

@@ -0,0 +1,100 @@
+package combiner
+
+import (
+	"github.com/grafana/tempo/pkg/tempopb"
+)
+
+// These structs combine response metrics in a single place
+
+type SearchMetricsCombiner struct {
+	Metrics *tempopb.SearchMetrics
+}
+
+func NewSearchMetricsCombiner() *SearchMetricsCombiner {
+	return &SearchMetricsCombiner{
+		Metrics: &tempopb.SearchMetrics{},
+	}
+}
+
+func (mc *SearchMetricsCombiner) Combine(newMetrics *tempopb.SearchMetrics, resp PipelineResponse) {
+	if newMetrics != nil {
+		mc.Metrics.CompletedJobs++
+		if !IsCacheHit(resp.HTTPResponse()) {
+			mc.Metrics.InspectedTraces += newMetrics.InspectedTraces
+			mc.Metrics.InspectedBytes += newMetrics.InspectedBytes
+		}
+	}
+}
+
+func (mc *SearchMetricsCombiner) CombineMetadata(newMetrics *tempopb.SearchMetrics, _ PipelineResponse) {
+	// These "Total" metrics are calculated by the frontend in the sharder.
+	// TotalBlockBytes is the total bytes of all blocks considered for the search, irrelevant of the cache.
+	// InspectedBytes is the total bytes actually read in the Parquet files, calculated by the queriers and conditional to the response being cached.
+	if newMetrics != nil {
+		mc.Metrics.TotalBlocks += newMetrics.TotalBlocks
+		mc.Metrics.TotalJobs += newMetrics.TotalJobs
+		mc.Metrics.TotalBlockBytes += newMetrics.TotalBlockBytes
+	}
+}
+
+type TraceByIDMetricsCombiner struct {
+	Metrics *tempopb.TraceByIDMetrics
+}
+
+func NewTraceByIDMetricsCombiner() *TraceByIDMetricsCombiner {
+	return &TraceByIDMetricsCombiner{
+		Metrics: &tempopb.TraceByIDMetrics{},
+	}
+}
+
+func (mc *TraceByIDMetricsCombiner) Combine(newMetrics *tempopb.TraceByIDMetrics, resp PipelineResponse) {
+	if newMetrics != nil && !IsCacheHit(resp.HTTPResponse()) {
+		mc.Metrics.InspectedBytes += newMetrics.InspectedBytes
+	}
+}
+
+type MetadataMetricsCombiner struct {
+	Metrics *tempopb.MetadataMetrics
+}
+
+func NewMetadataMetricsCombiner() *MetadataMetricsCombiner {
+	return &MetadataMetricsCombiner{
+		Metrics: &tempopb.MetadataMetrics{},
+	}
+}
+
+func (mc *MetadataMetricsCombiner) Combine(newMetrics *tempopb.MetadataMetrics, resp PipelineResponse) {
+	if newMetrics != nil && !IsCacheHit(resp.HTTPResponse()) {
+		mc.Metrics.InspectedBytes += newMetrics.InspectedBytes
+	}
+}
+
+type QueryRangeMetricsCombiner struct {
+	Metrics *tempopb.SearchMetrics
+}
+
+func NewQueryRangeMetricsCombiner() *QueryRangeMetricsCombiner {
+	return &QueryRangeMetricsCombiner{
+		Metrics: &tempopb.SearchMetrics{},
+	}
+}
+
+func (mc *QueryRangeMetricsCombiner) Combine(newMetrics *tempopb.SearchMetrics, resp PipelineResponse) {
+	if newMetrics != nil {
+		// this is a coordination between the sharder and combiner. the sharder returns one response with summary metrics
+		// only. the combiner correctly takes and accumulates that job. however, if the response has no jobs this is
+		// an indicator this is a "real" response so we set CompletedJobs to 1 to increment in the combiner.
+		if newMetrics.TotalJobs == 0 {
+			newMetrics.CompletedJobs = 1
+			mc.Metrics.CompletedJobs += newMetrics.CompletedJobs
+		}
+		if !IsCacheHit(resp.HTTPResponse()) {
+			mc.Metrics.TotalJobs += newMetrics.TotalJobs
+			mc.Metrics.TotalBlocks += newMetrics.TotalBlocks
+			mc.Metrics.TotalBlockBytes += newMetrics.TotalBlockBytes
+			mc.Metrics.InspectedBytes += newMetrics.InspectedBytes
+			mc.Metrics.InspectedTraces += newMetrics.InspectedTraces
+			mc.Metrics.InspectedSpans += newMetrics.InspectedSpans
+		}
+	}
+}

modules/frontend/combiner/search.go

@@ -42,6 +42,7 @@ func NewSearch(limit int, keepMostRecent bool) Combiner {
 	metadataCombiner := traceql.NewMetadataCombiner(limit, keepMostRecent)
 	diffTraces := map[string]struct{}{}
 	completedThroughTracker := &ShardCompletionTracker{}
+	metricsCombiner := NewSearchMetricsCombiner()
 
 	c := &genericCombiner[*tempopb.SearchResponse]{
 		httpStatusCode: 200,
@@ -60,19 +61,18 @@ func NewSearch(limit int, keepMostRecent bool) Combiner {
 				}
 			}
 
-			if partial.Metrics != nil {
-				final.Metrics.CompletedJobs++
-				final.Metrics.InspectedBytes += partial.Metrics.InspectedBytes
-				final.Metrics.InspectedTraces += partial.Metrics.InspectedTraces
-			}
+			metricsCombiner.Combine(partial.Metrics, resp)
 
 			return nil
 		},
 		metadata: func(resp PipelineResponse, final *tempopb.SearchResponse) error {
 			if sj, ok := resp.(*SearchJobResponse); ok && sj != nil {
-				final.Metrics.TotalBlocks += uint32(sj.TotalBlocks)
-				final.Metrics.TotalJobs += uint32(sj.TotalJobs)
-				final.Metrics.TotalBlockBytes += sj.TotalBytes
+				sjMetrics := &tempopb.SearchMetrics{
+					TotalBlocks:     uint32(sj.TotalBlocks), //nolint:gosec
+					TotalJobs:       uint32(sj.TotalJobs),   //nolint:gosec
+					TotalBlockBytes: sj.TotalBytes,
+				}
+				metricsCombiner.CombineMetadata(sjMetrics, resp)
 
 				if keepMostRecent {
 					completedThroughTracker.addShards(sj.Shards)
@@ -84,7 +84,7 @@ func NewSearch(limit int, keepMostRecent bool) Combiner {
 		finalize: func(final *tempopb.SearchResponse) (*tempopb.SearchResponse, error) {
 			// metrics are already combined on the passed in final
 			final.Traces = metadataCombiner.Metadata()
-
+			final.Metrics = metricsCombiner.Metrics
 			addRootSpanNotReceivedText(final.Traces)
 			return final, nil
 		},
@@ -92,15 +92,14 @@ func NewSearch(limit int, keepMostRecent bool) Combiner {
 			// wipe out any existing traces and recreate from the map
 			diff := &tempopb.SearchResponse{
 				Traces:  make([]*tempopb.TraceSearchMetadata, 0, len(diffTraces)),
-				Metrics: current.Metrics,
+				Metrics: metricsCombiner.Metrics,
 			}
-
 			metadataFn := metadataCombiner.Metadata
 			if keepMostRecent {
 				metadataFn = func() []*tempopb.TraceSearchMetadata {
 					completedThroughSeconds := completedThroughTracker.completedThroughSeconds
 					// if all jobs are completed then let's just return everything the combiner has
-					if current.Metrics.CompletedJobs == current.Metrics.TotalJobs && current.Metrics.TotalJobs > 0 {
+					if diff.Metrics.CompletedJobs == diff.Metrics.TotalJobs && diff.Metrics.TotalJobs > 0 {
 						completedThroughSeconds = 1
 					}
 
@@ -127,8 +126,6 @@ func NewSearch(limit int, keepMostRecent bool) Combiner {
 
 			return diff, nil
 		},
-		// search combiner doesn't use current in the way i would have expected. it only tracks metrics through current and uses the results map for the actual traces.
-		//  should we change this?
 		quit: func(_ *tempopb.SearchResponse) bool {
 			completedThroughSeconds := completedThroughTracker.completedThroughSeconds
 			// have we completed any shards?

modules/frontend/combiner/search_tag_values.go

@@ -4,7 +4,6 @@ import (
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/collector"
 	"github.com/grafana/tempo/pkg/tempopb"
-	"go.uber.org/atomic"
 )
 
 var (
@@ -15,26 +14,21 @@ var (
 func NewSearchTagValues(maxDataBytes int, maxTagsValues uint32, staleValueThreshold uint32) Combiner {
 	// Distinct collector with no limit
 	d := collector.NewDistinctStringWithDiff(maxDataBytes, maxTagsValues, staleValueThreshold)
-	inspectedBytes := atomic.NewUint64(0)
-
+	metricsCombiner := NewMetadataMetricsCombiner()
 	c := &genericCombiner[*tempopb.SearchTagValuesResponse]{
 		httpStatusCode: 200,
 		new:            func() *tempopb.SearchTagValuesResponse { return &tempopb.SearchTagValuesResponse{} },
 		current:        &tempopb.SearchTagValuesResponse{TagValues: make([]string, 0)},
-		combine: func(partial, _ *tempopb.SearchTagValuesResponse, _ PipelineResponse) error {
+		combine: func(partial, _ *tempopb.SearchTagValuesResponse, pipelineResp PipelineResponse) error {
 			for _, v := range partial.TagValues {
 				d.Collect(v)
 			}
-			if partial.Metrics != nil {
-				inspectedBytes.Add(partial.Metrics.InspectedBytes)
-			}
+			metricsCombiner.Combine(partial.Metrics, pipelineResp)
 			return nil
 		},
 		finalize: func(final *tempopb.SearchTagValuesResponse) (*tempopb.SearchTagValuesResponse, error) {
 			final.TagValues = d.Strings()
-			// return metrics in final response
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			final.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+			final.Metrics = metricsCombiner.Metrics
 			return final, nil
 		},
 		quit: func(_ *tempopb.SearchTagValuesResponse) bool {
@@ -46,9 +40,8 @@ func NewSearchTagValues(maxDataBytes int, maxTagsValues uint32, staleValueThresh
 				return nil, err
 			}
 			response.TagValues = resp
-			// also return latest metrics along with diff
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			response.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+			response.Metrics = metricsCombiner.Metrics
+
 			return response, nil
 		},
 	}
@@ -63,19 +56,17 @@ func NewTypedSearchTagValues(maxDataBytes int, maxTagsValues uint32, staleValueT
 func NewSearchTagValuesV2(maxDataBytes int, maxTagsValues uint32, staleValueThreshold uint32) Combiner {
 	// Distinct collector with no limit and diff enabled
 	d := collector.NewDistinctValueWithDiff(maxDataBytes, maxTagsValues, staleValueThreshold, func(tv tempopb.TagValue) int { return len(tv.Type) + len(tv.Value) })
-	inspectedBytes := atomic.NewUint64(0)
+	metricsCombiner := NewMetadataMetricsCombiner()
 
 	c := &genericCombiner[*tempopb.SearchTagValuesV2Response]{
 		httpStatusCode: 200,
 		current:        &tempopb.SearchTagValuesV2Response{TagValues: []*tempopb.TagValue{}},
 		new:            func() *tempopb.SearchTagValuesV2Response { return &tempopb.SearchTagValuesV2Response{} },
-		combine: func(partial, _ *tempopb.SearchTagValuesV2Response, _ PipelineResponse) error {
+		combine: func(partial, _ *tempopb.SearchTagValuesV2Response, pipelineResp PipelineResponse) error {
 			for _, v := range partial.TagValues {
 				d.Collect(*v)
 			}
-			if partial.Metrics != nil {
-				inspectedBytes.Add(partial.Metrics.InspectedBytes)
-			}
+			metricsCombiner.Combine(partial.Metrics, pipelineResp)
 			return nil
 		},
 		finalize: func(final *tempopb.SearchTagValuesV2Response) (*tempopb.SearchTagValuesV2Response, error) {
@@ -85,9 +76,9 @@ func NewSearchTagValuesV2(maxDataBytes int, maxTagsValues uint32, staleValueThre
 				v2 := v
 				final.TagValues = append(final.TagValues, &v2)
 			}
-			// load Inspected Bytes here and return along with final response
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			final.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+
+			final.Metrics = metricsCombiner.Metrics
+
 			return final, nil
 		},
 		quit: func(_ *tempopb.SearchTagValuesV2Response) bool {
@@ -103,9 +94,9 @@ func NewSearchTagValuesV2(maxDataBytes int, maxTagsValues uint32, staleValueThre
 				v2 := v
 				response.TagValues = append(response.TagValues, &v2)
 			}
-			// also return metrics along with diffs
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			response.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+
+			response.Metrics = metricsCombiner.Metrics
+
 			return response, nil
 		},
 	}

modules/frontend/combiner/search_tags.go

@@ -4,7 +4,6 @@ import (
 	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/collector"
 	"github.com/grafana/tempo/pkg/tempopb"
-	"go.uber.org/atomic"
 )
 
 var (
@@ -14,27 +13,25 @@ var (
 
 func NewSearchTags(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold uint32) Combiner {
 	d := collector.NewDistinctStringWithDiff(maxDataBytes, maxTagsPerScope, staleValueThreshold)
-	inspectedBytes := atomic.NewUint64(0)
+	metricsCombiner := NewMetadataMetricsCombiner()
 
 	c := &genericCombiner[*tempopb.SearchTagsResponse]{
 		httpStatusCode: 200,
 		new:            func() *tempopb.SearchTagsResponse { return &tempopb.SearchTagsResponse{} },
 		current:        &tempopb.SearchTagsResponse{TagNames: make([]string, 0)},
-		combine: func(partial, _ *tempopb.SearchTagsResponse, _ PipelineResponse) error {
+		combine: func(partial, _ *tempopb.SearchTagsResponse, pipelineResp PipelineResponse) error {
+			metricsCombiner.Combine(partial.Metrics, pipelineResp)
+
 			for _, v := range partial.TagNames {
 				d.Collect(v)
 			}
-			if partial.Metrics != nil {
-				inspectedBytes.Add(partial.Metrics.InspectedBytes)
-			}
 			return nil
 		},
-		finalize: func(response *tempopb.SearchTagsResponse) (*tempopb.SearchTagsResponse, error) {
-			response.TagNames = d.Strings()
-			// return metrics with final results
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			response.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
-			return response, nil
+		finalize: func(final *tempopb.SearchTagsResponse) (*tempopb.SearchTagsResponse, error) {
+			final.TagNames = d.Strings()
+			final.Metrics = metricsCombiner.Metrics
+
+			return final, nil
 		},
 		quit: func(_ *tempopb.SearchTagsResponse) bool {
 			return d.Exceeded()
@@ -46,9 +43,7 @@ func NewSearchTags(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold
 			}
 
 			response.TagNames = resp
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			// return metrics with diff results
-			response.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+			response.Metrics = metricsCombiner.Metrics
 			return response, nil
 		},
 	}
@@ -56,28 +51,23 @@ func NewSearchTags(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold
 	return c
 }
 
-func NewTypedSearchTags(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold uint32) GRPCCombiner[*tempopb.SearchTagsResponse] {
-	return NewSearchTags(maxDataBytes, maxTagsPerScope, staleValueThreshold).(GRPCCombiner[*tempopb.SearchTagsResponse])
-}
-
 func NewSearchTagsV2(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold uint32) Combiner {
 	// Distinct collector map to collect scopes and scope values
 	distinctValues := collector.NewScopedDistinctStringWithDiff(maxDataBytes, maxTagsPerScope, staleValueThreshold)
-	inspectedBytes := atomic.NewUint64(0)
+	metricsCombiner := NewMetadataMetricsCombiner()
 
 	c := &genericCombiner[*tempopb.SearchTagsV2Response]{
 		httpStatusCode: 200,
 		new:            func() *tempopb.SearchTagsV2Response { return &tempopb.SearchTagsV2Response{} },
 		current:        &tempopb.SearchTagsV2Response{Scopes: make([]*tempopb.SearchTagsV2Scope, 0)},
-		combine: func(partial, _ *tempopb.SearchTagsV2Response, _ PipelineResponse) error {
+		combine: func(partial, _ *tempopb.SearchTagsV2Response, pipelineResp PipelineResponse) error {
+			metricsCombiner.Combine(partial.Metrics, pipelineResp)
+
 			for _, res := range partial.GetScopes() {
 				for _, tag := range res.Tags {
 					distinctValues.Collect(res.Name, tag)
 				}
 			}
-			if partial.Metrics != nil {
-				inspectedBytes.Add(partial.Metrics.InspectedBytes)
-			}
 			return nil
 		},
 		finalize: func(final *tempopb.SearchTagsV2Response) (*tempopb.SearchTagsV2Response, error) {
@@ -90,9 +80,8 @@ func NewSearchTagsV2(maxDataBytes int, maxTagsPerScope uint32, staleValueThresho
 					Tags: vals,
 				})
 			}
-			// return metrics with final results
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			final.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+
+			final.Metrics = metricsCombiner.Metrics
 			return final, nil
 		},
 		quit: func(_ *tempopb.SearchTagsV2Response) bool {
@@ -111,9 +100,8 @@ func NewSearchTagsV2(maxDataBytes int, maxTagsPerScope uint32, staleValueThresho
 					Tags: vals,
 				})
 			}
-			// TODO: merge with other metrics as well, when we have them, return only InspectedBytes for now
-			// also return metrics with diff results
-			response.Metrics = &tempopb.MetadataMetrics{InspectedBytes: inspectedBytes.Load()}
+
+			response.Metrics = metricsCombiner.Metrics
 			return response, nil
 		},
 	}
@@ -121,6 +109,10 @@ func NewSearchTagsV2(maxDataBytes int, maxTagsPerScope uint32, staleValueThresho
 	return c
 }
 
+func NewTypedSearchTags(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold uint32) GRPCCombiner[*tempopb.SearchTagsResponse] {
+	return NewSearchTags(maxDataBytes, maxTagsPerScope, staleValueThreshold).(GRPCCombiner[*tempopb.SearchTagsResponse])
+}
+
 func NewTypedSearchTagsV2(maxDataBytes int, maxTagsPerScope uint32, staleValueThreshold uint32) GRPCCombiner[*tempopb.SearchTagsV2Response] {
 	return NewSearchTagsV2(maxDataBytes, maxTagsPerScope, staleValueThreshold).(GRPCCombiner[*tempopb.SearchTagsV2Response])
 }

modules/frontend/combiner/trace_by_id.go

@@ -9,8 +9,6 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/grafana/tempo/pkg/collector"
-
 	"github.com/gogo/protobuf/proto"
 	"github.com/grafana/tempo/pkg/api"
 	tempo_io "github.com/grafana/tempo/pkg/io"
@@ -31,7 +29,7 @@ type TraceByIDCombiner struct {
 	code          int
 	statusMessage string
 
-	mc *collector.MetricsCollector
+	MetricsCombiner *TraceByIDMetricsCombiner
 }
 
 // NewTraceByID returns a trace id combiner. The trace by id combiner has a few different behaviors then the others
@@ -41,10 +39,10 @@ type TraceByIDCombiner struct {
 // - encode the returned trace as either json or proto depending on the request
 func NewTraceByID(maxBytes int, contentType string) Combiner {
 	return &TraceByIDCombiner{
-		c:           trace.NewCombiner(maxBytes, false),
-		code:        http.StatusNotFound,
-		contentType: contentType,
-		mc:          collector.NewMetricsCollector(),
+		c:               trace.NewCombiner(maxBytes, false),
+		code:            http.StatusNotFound,
+		contentType:     contentType,
+		MetricsCombiner: NewTraceByIDMetricsCombiner(),
 	}
 }
 
@@ -52,10 +50,6 @@ func NewTypedTraceByID(maxBytes int, contentType string) *TraceByIDCombiner {
 	return NewTraceByID(maxBytes, contentType).(*TraceByIDCombiner)
 }
 
-func (c *TraceByIDCombiner) TotalBytesProcessed() uint64 {
-	return c.mc.TotalValue()
-}
-
 func (c *TraceByIDCombiner) AddResponse(r PipelineResponse) error {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -103,9 +97,7 @@ func (c *TraceByIDCombiner) AddResponse(r PipelineResponse) error {
 		c.statusMessage = fmt.Sprint(err)
 		return nil
 	}
-	if resp.Metrics != nil {
-		c.mc.Add(resp.Metrics.InspectedBytes)
-	}
+	c.MetricsCombiner.Combine(resp.Metrics, r)
 
 	return err
 }

modules/frontend/combiner/trace_by_id_v2.go

@@ -14,15 +14,15 @@ func NewTypedTraceByIDV2(maxBytes int, marshalingFormat string) GRPCCombiner[*te
 func NewTraceByIDV2(maxBytes int, marshalingFormat string) Combiner {
 	combiner := trace.NewCombiner(maxBytes, true)
 	var partialTrace bool
-	var inspectedBytes uint64
+	metricsCombiner := NewTraceByIDMetricsCombiner()
 	gc := &genericCombiner[*tempopb.TraceByIDResponse]{
-		combine: func(partial *tempopb.TraceByIDResponse, _ *tempopb.TraceByIDResponse, _ PipelineResponse) error {
+		combine: func(partial *tempopb.TraceByIDResponse, _ *tempopb.TraceByIDResponse, pipelineResp PipelineResponse) error {
 			if partial.Status == tempopb.PartialStatus_PARTIAL {
 				partialTrace = true
 			}
-			if partial.Metrics != nil {
-				inspectedBytes += partial.Metrics.InspectedBytes
-			}
+
+			metricsCombiner.Combine(partial.Metrics, pipelineResp)
+
 			_, err := combiner.Consume(partial.Trace)
 			return err
 		},
@@ -36,7 +36,7 @@ func NewTraceByIDV2(maxBytes int, marshalingFormat string) Combiner {
 			deduper := newDeduper()
 			traceResult = deduper.dedupe(traceResult)
 			resp.Trace = traceResult
-			resp.Metrics = &tempopb.TraceByIDMetrics{InspectedBytes: inspectedBytes}
+			resp.Metrics = metricsCombiner.Metrics
 
 			if partialTrace || combiner.IsPartialTrace() {
 				resp.Status = tempopb.PartialStatus_PARTIAL

modules/frontend/metrics_query_range_handler_test.go

@@ -280,3 +280,111 @@ func TestQueryRangeHandlerV2MaxSeries(t *testing.T) {
 	require.Equal(t, maxSeries, len(actualResp.Series))
 	require.Equal(t, tempopb.PartialStatus_PARTIAL, actualResp.Status)
 }
+
+func TestQueryRangeCachedMetrics(t *testing.T) {
+	// set up backend
+	tenant := "foo"
+	meta := &backend.BlockMeta{
+		StartTime:         time.Unix(15, 0),
+		EndTime:           time.Unix(16, 0),
+		Size_:             defaultTargetBytesPerRequest,
+		TotalRecords:      1,
+		BlockID:           backend.MustParse("00000000-0000-0000-0000-000000000123"),
+		ReplicationFactor: 1,
+	}
+	rdr := &mockReader{
+		metas: []*backend.BlockMeta{meta},
+	}
+
+	// set up cache
+	c := test.NewMockClient()
+	p := test.NewMockProvider()
+	err := p.AddCache(cache.RoleFrontendSearch, c)
+	require.NoError(t, err)
+	f := frontendWithSettings(t, &mockRoundTripper{
+		responseFn: func() proto.Message {
+			return &tempopb.QueryRangeResponse{
+				Metrics: &tempopb.SearchMetrics{
+					InspectedTraces: 2,
+					InspectedBytes:  33,
+				},
+				Series: []*tempopb.TimeSeries{
+					{
+						PromLabels: "foo",
+						Labels: []v1.KeyValue{
+							{Key: "foo", Value: &v1.AnyValue{Value: &v1.AnyValue_StringValue{StringValue: "bar"}}},
+						},
+						Samples: []tempopb.Sample{
+							{
+								TimestampMs: 1100_000,
+								Value:       1,
+							},
+						},
+					},
+				},
+			}
+		},
+	}, rdr, nil, p, func(c *Config, _ *overrides.Config) {
+		c.Metrics.Sharder.Interval = time.Hour
+	})
+
+	// set up query
+	query := "{} | rate()"
+	var step uint64 = 1000000000
+	hash := hashForQueryRangeRequest(&tempopb.QueryRangeRequest{Query: query, Step: step})
+	startNS := uint64(10 * time.Second)
+	endNS := uint64(20 * time.Second)
+	cacheKey := queryRangeCacheKey(tenant, hash, time.Unix(0, int64(startNS)), time.Unix(0, int64(endNS)), meta, 0, 1)
+
+	// confirm cache key doesn't exist
+	_, bufs, _ := c.Fetch(context.Background(), []string{cacheKey})
+	require.Equal(t, 0, len(bufs))
+
+	// execute query
+	path := fmt.Sprintf("/?start=%d&end=%d&q=%s", startNS, endNS, url.QueryEscape(query))
+	req := httptest.NewRequest("GET", path, nil)
+	ctx := req.Context()
+	ctx = user.InjectOrgID(ctx, tenant)
+	req = req.WithContext(ctx)
+	respWriter := httptest.NewRecorder()
+	f.MetricsQueryRangeHandler.ServeHTTP(respWriter, req)
+	resp := respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse response
+	actualResp := &tempopb.QueryRangeResponse{}
+	bytesResp, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are collected
+	require.Equal(t, uint64(33), actualResp.Metrics.InspectedBytes)
+	require.Equal(t, uint32(2), actualResp.Metrics.InspectedTraces)
+	require.Equal(t, uint32(1), actualResp.Metrics.CompletedJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalBlocks)
+	require.Equal(t, uint64(defaultTargetBytesPerRequest), actualResp.Metrics.TotalBlockBytes)
+
+	// execute query again
+	respWriter = httptest.NewRecorder()
+	f.MetricsQueryRangeHandler.ServeHTTP(respWriter, req)
+	resp = respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse cached response
+	actualResp = &tempopb.QueryRangeResponse{}
+	bytesResp, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are 0 because the response was cached
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedBytes)
+	require.Equal(t, uint32(0), actualResp.Metrics.InspectedTraces)
+	require.Equal(t, uint32(1), actualResp.Metrics.CompletedJobs)
+	// these are metadata metrics and are not affected by caching
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalBlocks)
+	require.Equal(t, uint64(defaultTargetBytesPerRequest), actualResp.Metrics.TotalBlockBytes)
+}

modules/frontend/pipeline/sync_handler_cache.go

@@ -7,6 +7,8 @@ import (
 	"io"
 	"net/http"
 
+	"github.com/grafana/tempo/modules/frontend/combiner"
+
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/grafana/tempo/pkg/api"
@@ -39,30 +41,30 @@ func (c cachingWare) RoundTrip(req Request) (*http.Response, error) {
 	if len(key) > 0 {
 		body := c.cache.fetchBytes(req.Context(), key)
 		if len(body) > 0 {
+			contentType := determineContentType(body)
+
 			resp := &http.Response{
-				Header:     http.Header{},
-				StatusCode: http.StatusOK,
-				Status:     http.StatusText(http.StatusOK),
-				Body:       io.NopCloser(bytes.NewBuffer(body)),
+				Header:        http.Header{api.HeaderContentType: []string{contentType}, combiner.TempoCacheHeader: []string{combiner.TempoCacheHit}},
+				StatusCode:    http.StatusOK,
+				Status:        http.StatusText(http.StatusOK),
+				Body:          io.NopCloser(bytes.NewBuffer(body)),
+				ContentLength: int64(len(body)),
 			}
 
-			// We aren't capturing the original content type in the cache, just the raw bytes.
-			// Detect it and readd it, so the upstream code can parse the body.
-			// TODO - Cache should capture all of the relevant parts of the
-			// original response including both content-type and content-length headers, possibly more.
-			// But upgrading the cache format requires migration/detection of previous format either way.
-			// It's tempting to use https://pkg.go.dev/net/http#DetectContentType but it doesn't detect
-			// json or proto.
-			if body[0] == '{' {
-				resp.Header.Add(api.HeaderContentType, api.HeaderAcceptJSON)
-			} else {
-				resp.Header.Add(api.HeaderContentType, api.HeaderAcceptProtobuf)
-			}
 			return resp, nil
 		}
 	}
 
 	resp, err := c.next.RoundTrip(req)
+
+	// Add cache miss header for all responses that weren't from cache
+	if resp != nil {
+		if resp.Header == nil {
+			resp.Header = http.Header{}
+		}
+		resp.Header[combiner.TempoCacheHeader] = []string{combiner.TempoCacheMiss}
+	}
+
 	// do not cache if there was an error
 	if err != nil {
 		return resp, err
@@ -100,6 +102,18 @@ func (c cachingWare) RoundTrip(req Request) (*http.Response, error) {
 	return resp, nil
 }
 
+func determineContentType(body []byte) string {
+	// TODO - Cache should capture all of the relevant parts of the
+	// original response including both content-type and content-length headers, possibly more.
+	// But upgrading the cache format requires migration/detection of previous format either way.
+	// It's tempting to use https://pkg.go.dev/net/http#DetectContentType but it doesn't detect
+	// json or proto.
+	if body[0] == '{' {
+		return api.HeaderAcceptJSON
+	}
+	return api.HeaderAcceptProtobuf
+}
+
 func shouldCache(statusCode int) bool {
 	return statusCode/100 == 2
 }

modules/frontend/pipeline/sync_handler_cache_test.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/jsonpb"
+	"github.com/grafana/tempo/pkg/api"
 	"github.com/grafana/tempo/pkg/cache"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/util/test"
@@ -45,3 +46,42 @@ func TestCacheCaches(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, expected, actual)
 }
+
+func TestDetermineContentType(t *testing.T) {
+	// Create and marshal a real protobuf message
+	protoMsg := &tempopb.SearchTagsResponse{
+		TagNames: []string{"foo", "bar"},
+	}
+	protobufContent, err := protoMsg.Marshal()
+	require.NoError(t, err)
+
+	// Also create JSON content for comparison
+	jsonBuf := bytes.NewBuffer([]byte{})
+	err = (&jsonpb.Marshaler{}).Marshal(jsonBuf, protoMsg)
+	require.NoError(t, err)
+	jsonContent := jsonBuf.Bytes()
+
+	testCases := []struct {
+		name         string
+		body         []byte
+		expectedType string
+	}{
+		{
+			name:         "JSON content",
+			body:         jsonContent,
+			expectedType: api.HeaderAcceptJSON,
+		},
+		{
+			name:         "Protobuf content",
+			body:         protobufContent,
+			expectedType: api.HeaderAcceptProtobuf,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			contentType := determineContentType(tc.body)
+			require.Equal(t, tc.expectedType, contentType)
+		})
+	}
+}

modules/frontend/search_handlers_test.go

@@ -589,7 +589,7 @@ func TestSearchAccessesCache(t *testing.T) {
 	end := uint32(20)
 	cacheKey := searchJobCacheKey(tenant, hash, time.Unix(int64(start), 0), time.Unix(int64(end), 0), meta, 0, 1)
 
-	// confirm cache key coesn't exist
+	// confirm cache key doesn't exist
 	_, bufs, _ := c.Fetch(context.Background(), []string{cacheKey})
 	require.Equal(t, 0, len(bufs))
 
@@ -632,7 +632,7 @@ func TestSearchAccessesCache(t *testing.T) {
 			RootTraceName:   "bar2",
 		}},
 		Metrics: &tempopb.SearchMetrics{
-			InspectedBytes: 11,
+			InspectedBytes: 0, // the cached response will have 0 inspected bytes no matter what value we have here
 		},
 	}
 	overwriteString, err := (&jsonpb.Marshaler{}).MarshalToString(overwriteResp)
@@ -672,6 +672,89 @@ func cacheResponsesEqual(t *testing.T, cacheResponse *tempopb.SearchResponse, pi
 	require.Equal(t, pipelineResp, cacheResponse)
 }
 
+func TestSearchCachedMetrics(t *testing.T) {
+	// set up backend
+	tenant := "foo"
+	meta := &backend.BlockMeta{
+		StartTime:    time.Unix(15, 0),
+		EndTime:      time.Unix(16, 0),
+		Size_:        defaultTargetBytesPerRequest,
+		TotalRecords: 1,
+		BlockID:      backend.MustParse("00000000-0000-0000-0000-000000000123"),
+	}
+	rdr := &mockReader{
+		metas: []*backend.BlockMeta{meta},
+	}
+
+	// set up cache
+	c := test.NewMockClient()
+	p := test.NewMockProvider()
+	err := p.AddCache(cache.RoleFrontendSearch, c)
+	require.NoError(t, err)
+	f := frontendWithSettings(t, nil, rdr, nil, p)
+
+	// set up query
+	query := "{}"
+	hash := hashForSearchRequest(&tempopb.SearchRequest{Query: query, Limit: 3, SpansPerSpanSet: 2})
+	start := uint32(10)
+	end := uint32(20)
+	cacheKey := searchJobCacheKey(tenant, hash, time.Unix(int64(start), 0), time.Unix(int64(end), 0), meta, 0, 1)
+
+	// confirm cache key doesn't exist
+	_, bufs, _ := c.Fetch(context.Background(), []string{cacheKey})
+	require.Equal(t, 0, len(bufs))
+
+	// execute query
+	path := fmt.Sprintf("/?start=%d&end=%d&q=%s&limit=3&spss=2", start, end, query)
+	req := httptest.NewRequest("GET", path, nil)
+	ctx := req.Context()
+	ctx = user.InjectOrgID(ctx, tenant)
+	req = req.WithContext(ctx)
+	respWriter := httptest.NewRecorder()
+	f.SearchHandler.ServeHTTP(respWriter, req)
+	resp := respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse response
+	actualResp := &tempopb.SearchResponse{}
+	bytesResp, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are collected
+	require.Equal(t, uint64(1), actualResp.Metrics.InspectedBytes)
+	require.Equal(t, uint32(1), actualResp.Metrics.InspectedTraces)
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedSpans)
+	require.Equal(t, uint32(1), actualResp.Metrics.CompletedJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalBlocks)
+	require.Equal(t, uint64(defaultTargetBytesPerRequest), actualResp.Metrics.TotalBlockBytes)
+
+	// execute query again
+	respWriter = httptest.NewRecorder()
+	f.SearchHandler.ServeHTTP(respWriter, req)
+	resp = respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse cached response
+	actualResp = &tempopb.SearchResponse{}
+	bytesResp, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are 0 because the the response was cached
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedBytes)
+	require.Equal(t, uint32(0), actualResp.Metrics.InspectedTraces)
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedSpans)
+	// these are metadata metrics and are not affected by caching
+	require.Equal(t, uint32(1), actualResp.Metrics.CompletedJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalJobs)
+	require.Equal(t, uint32(1), actualResp.Metrics.TotalBlocks)
+	require.Equal(t, uint64(defaultTargetBytesPerRequest), actualResp.Metrics.TotalBlockBytes)
+}
+
 func BenchmarkSearchPipeline(b *testing.B) {
 	tenant := "foo"
 

modules/frontend/tag_handlers_test.go

@@ -645,6 +645,197 @@ func TestSearchTagsV2AccessesCache(t *testing.T) {
 	require.Equal(t, overwriteResp, actualResp)
 }
 
+func TestTagValuesCachedMetrics(t *testing.T) {
+	// set up backend
+	meta := &backend.BlockMeta{
+		StartTime:    time.Unix(15, 0),
+		EndTime:      time.Unix(16, 0),
+		Size_:        defaultTargetBytesPerRequest,
+		TotalRecords: 1,
+		BlockID:      backend.MustParse("00000000-0000-0000-0000-000000000123"),
+	}
+	rdr := &mockReader{
+		metas: []*backend.BlockMeta{meta},
+	}
+
+	// set up cache
+	c := test.NewMockClient()
+	p := test.NewMockProvider()
+	err := p.AddCache(cache.RoleFrontendSearch, c)
+	require.NoError(t, err)
+
+	// Set up a mock response with specific metrics that should be cleared when cached
+	f := frontendWithSettings(t, &mockRoundTripper{
+		responseFn: func() proto.Message {
+			return &tempopb.SearchTagValuesV2Response{
+				TagValues: []*tempopb.TagValue{
+					{
+						Type:  "keyword",
+						Value: "frontend",
+					},
+					{
+						Type:  "keyword",
+						Value: "backend",
+					},
+					{
+						Type:  "keyword",
+						Value: "database",
+					},
+				},
+				Metrics: &tempopb.MetadataMetrics{
+					InspectedBytes: 1024,
+				},
+			}
+		},
+	}, rdr, nil, p)
+
+	// setup query
+	tenant := "foo"
+	tagName := "service.name"
+	hash := fnv1a.HashString64(tagName)
+	start := uint32(10)
+	end := uint32(20)
+	startTime := time.Unix(int64(start), 0)
+	endTime := time.Unix(int64(end), 0)
+	cacheKey := cacheKey(cacheKeyPrefixSearchTagValues, tenant, hash, startTime, endTime, meta, 0, 1)
+
+	// confirm cache key doesn't exist
+	_, bufs, _ := c.Fetch(context.Background(), []string{cacheKey})
+	require.Equal(t, 0, len(bufs))
+
+	// execute query
+	path := fmt.Sprintf("/api/v2/search/tag/%s/values?start=%d&end=%d", tagName, start, end)
+	req := httptest.NewRequest("GET", path, nil)
+	req = mux.SetURLVars(req, map[string]string{"tagName": tagName})
+	ctx := req.Context()
+	ctx = user.InjectOrgID(ctx, tenant)
+	req = req.WithContext(ctx)
+
+	respWriter := httptest.NewRecorder()
+	f.SearchTagsValuesV2Handler.ServeHTTP(respWriter, req)
+	resp := respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse response
+	actualResp := &tempopb.SearchTagValuesV2Response{}
+	bytesResp, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are collected
+	require.Equal(t, uint64(1024), actualResp.Metrics.InspectedBytes)
+
+	// execute query again
+	respWriter = httptest.NewRecorder()
+	f.SearchTagsValuesV2Handler.ServeHTTP(respWriter, req)
+	resp = respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse cached response
+	actualResp = &tempopb.SearchTagValuesV2Response{}
+	bytesResp, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are 0 because the response was cached
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedBytes)
+}
+
+func TestTagsCachedMetrics(t *testing.T) {
+	// set up backend
+	meta := &backend.BlockMeta{
+		StartTime:    time.Unix(15, 0),
+		EndTime:      time.Unix(16, 0),
+		Size_:        defaultTargetBytesPerRequest,
+		TotalRecords: 1,
+		BlockID:      backend.MustParse("00000000-0000-0000-0000-000000000123"),
+	}
+	rdr := &mockReader{
+		metas: []*backend.BlockMeta{meta},
+	}
+
+	// set up cache
+	c := test.NewMockClient()
+	p := test.NewMockProvider()
+	err := p.AddCache(cache.RoleFrontendSearch, c)
+	require.NoError(t, err)
+
+	// Set up a mock response with specific metrics that should be cleared when cached
+	f := frontendWithSettings(t, &mockRoundTripper{
+		responseFn: func() proto.Message {
+			return &tempopb.SearchTagsV2Response{
+				Scopes: []*tempopb.SearchTagsV2Scope{
+					{
+						Name: "resource",
+						Tags: []string{"service.name", "http.method", "http.status_code"},
+					},
+					{
+						Name: "span",
+						Tags: []string{"span.kind", "span.name", "span.status"},
+					},
+				},
+				Metrics: &tempopb.MetadataMetrics{
+					InspectedBytes: 2048,
+				},
+			}
+		},
+	}, rdr, nil, p)
+
+	// setup query
+	tenant := "foo"
+	scope := "resource"
+	hash := fnv1a.HashString64(scope)
+	start := uint32(10)
+	end := uint32(20)
+	startTime := time.Unix(int64(start), 0)
+	endTime := time.Unix(int64(end), 0)
+	cacheKey := cacheKey(cacheKeyPrefixSearchTag, tenant, hash, startTime, endTime, meta, 0, 1)
+
+	// confirm cache key doesn't exist
+	_, bufs, _ := c.Fetch(context.Background(), []string{cacheKey})
+	require.Equal(t, 0, len(bufs))
+
+	// execute query
+	path := fmt.Sprintf("/api/v2/search/tags?start=%d&end=%d&scope=%s", start, end, scope)
+	req := httptest.NewRequest("GET", path, nil)
+	ctx := req.Context()
+	ctx = user.InjectOrgID(ctx, tenant)
+	req = req.WithContext(ctx)
+
+	respWriter := httptest.NewRecorder()
+	f.SearchTagsV2Handler.ServeHTTP(respWriter, req)
+	resp := respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse response
+	actualResp := &tempopb.SearchTagsV2Response{}
+	bytesResp, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are collected
+	require.Equal(t, uint64(2048), actualResp.Metrics.InspectedBytes)
+
+	// execute query again
+	respWriter = httptest.NewRecorder()
+	f.SearchTagsV2Handler.ServeHTTP(respWriter, req)
+	resp = respWriter.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// parse cached response
+	actualResp = &tempopb.SearchTagsV2Response{}
+	bytesResp, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = jsonpb.Unmarshal(bytes.NewReader(bytesResp), actualResp)
+	require.NoError(t, err)
+
+	// verify metrics are 0 because the response was cached
+	require.Equal(t, uint64(0), actualResp.Metrics.InspectedBytes)
+}
+
 func TestParseParams(t *testing.T) {
 	tests := []struct {
 		name                   string

modules/frontend/traceid_handlers.go

@@ -73,16 +73,19 @@ func newTraceIDHandler(cfg Config, next pipeline.AsyncRoundTripper[combiner.Pipe
 		resp, err := rt.RoundTrip(req)
 		elapsed := time.Since(start)
 
-		bytesProcessed := comb.TotalBytesProcessed()
-		postSLOHook(resp, tenant, bytesProcessed, elapsed, err)
+		var inspectBytes uint64
+		if comb.MetricsCombiner != nil && comb.MetricsCombiner.Metrics != nil {
+			inspectBytes = comb.MetricsCombiner.Metrics.InspectedBytes
+		}
+		postSLOHook(resp, tenant, inspectBytes, elapsed, err)
 
 		level.Info(logger).Log(
 			"msg", "trace id response",
 			"tenant", tenant,
 			"path", req.URL.Path,
 			"duration_seconds", elapsed.Seconds(),
-			"inspected_bytes", bytesProcessed,
-			"request_throughput", float64(bytesProcessed)/elapsed.Seconds(),
+			"inspected_bytes", inspectBytes,
+			"request_throughput", float64(inspectBytes)/elapsed.Seconds(),
 			"err", err)
 
 		return resp, err

modules/frontend/traceid_handlers_test.go

@@ -7,6 +7,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -486,3 +487,110 @@ func TestTraceIDHandlerV2WithJSONResponse(t *testing.T) {
 	err := new(jsonpb.Unmarshaler).Unmarshal(resp.Body, actualResp)
 	require.NoError(t, err)
 }
+
+func TestTraceByIDHandlerV2CachedMetrics(t *testing.T) {
+	// Setup test trace
+	testTrace := test.MakeTrace(2, []byte{0x01, 0x02})
+	traceID := "1234abcd"
+
+	callCount := 0
+	var mu sync.Mutex
+	next := pipeline.RoundTripperFunc(func(_ pipeline.Request) (*http.Response, error) {
+		mu.Lock()
+		callCount++
+		mu.Unlock()
+
+		metrics := &tempopb.TraceByIDMetrics{
+			InspectedBytes: 1024,
+		}
+
+		resBytes, err := proto.Marshal(&tempopb.TraceByIDResponse{
+			Trace:   testTrace,
+			Metrics: metrics,
+		})
+		require.NoError(t, err)
+
+		return &http.Response{
+			Body:       io.NopCloser(bytes.NewReader(resBytes)),
+			StatusCode: 200,
+			Header: map[string][]string{
+				"Content-Type": {"application/protobuf"},
+			},
+		}, nil
+	})
+
+	// Create frontend
+	f := frontendWithSettings(t, next, nil, &Config{
+		MultiTenantQueriesEnabled: true,
+		TraceByID: TraceByIDConfig{
+			QueryShards: 2, // Minimum value required
+			SLO:         testSLOcfg,
+		},
+		Search: SearchConfig{
+			Sharder: SearchSharderConfig{
+				ConcurrentRequests:    defaultConcurrentRequests,
+				TargetBytesPerRequest: defaultTargetBytesPerRequest,
+				MostRecentShards:      defaultMostRecentShards,
+			},
+			SLO: testSLOcfg,
+		},
+		Metrics: MetricsConfig{
+			Sharder: QueryRangeSharderConfig{
+				ConcurrentRequests:    defaultConcurrentRequests,
+				TargetBytesPerRequest: defaultTargetBytesPerRequest,
+				Interval:              time.Second,
+			},
+			SLO: testSLOcfg,
+		},
+	}, nil)
+
+	tenant := "test-org"
+
+	// First request - should hit the backend
+	req := httptest.NewRequest("GET", "/api/v2/traces/"+traceID, nil)
+	ctx := user.InjectOrgID(req.Context(), tenant)
+	req = req.WithContext(ctx)
+	req = mux.SetURLVars(req, map[string]string{"traceID": traceID})
+	req.Header.Set("Accept", "application/protobuf")
+
+	httpResp := httptest.NewRecorder()
+	f.TraceByIDHandlerV2.ServeHTTP(httpResp, req)
+	resp := httpResp.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// Parse first response
+	actualResp := &tempopb.TraceByIDResponse{}
+	bytesTrace, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = proto.Unmarshal(bytesTrace, actualResp)
+	require.NoError(t, err)
+
+	// Verify first response metrics
+	require.Equal(t, uint64(2048), actualResp.Metrics.InspectedBytes)
+	require.Equal(t, 2, callCount)
+
+	// Second request - TraceIDHandlerV2 does not cache
+	req = httptest.NewRequest("GET", "/api/v2/traces/"+traceID, nil)
+	ctx = user.InjectOrgID(req.Context(), tenant)
+	req = req.WithContext(ctx)
+	req = mux.SetURLVars(req, map[string]string{"traceID": traceID})
+	req.Header.Set("Accept", "application/protobuf")
+
+	httpResp = httptest.NewRecorder()
+	f.TraceByIDHandlerV2.ServeHTTP(httpResp, req)
+	resp = httpResp.Result()
+	require.Equal(t, 200, resp.StatusCode)
+
+	// Parse second response
+	actualResp = &tempopb.TraceByIDResponse{}
+	bytesTrace, err = io.ReadAll(resp.Body)
+	require.NoError(t, err)
+	err = proto.Unmarshal(bytesTrace, actualResp)
+	require.NoError(t, err)
+
+	// Verify second response metrics. They should be the same as the first response as we're not caching
+	require.Equal(t, uint64(2048), actualResp.Metrics.InspectedBytes)
+
+	// Verify the backend was called again (callCount should be 4)
+	require.Equal(t, 4, callCount)
+}

pkg/collector/metrics_collector.go

@@ -4,8 +4,9 @@ import (
 	"go.uber.org/atomic"
 )
 
-// MetricsCollector is a simple collector that can be used to accumulate a metric
-// we primarily use it to collect the total bytes read from a reader across a request
+// MetricsCollector is a thread-safe collector that uses atomic operations
+// to accumulate metrics. We primarily use it to collect the total bytes read from
+// a reader across a request
 type MetricsCollector struct {
 	totalValue *atomic.Uint64
 }

pkg/traceql/combine.go

@@ -358,6 +358,8 @@ func (q *QueryRangeCombiner) Combine(resp *tempopb.QueryRangeResponse) {
 		q.maxSeriesReached = true
 	}
 
+	// TODO: this is here only for the querier
+	// we want to use the metrics combiner from the frontend
 	if resp.Metrics != nil {
 		q.metrics.TotalJobs += resp.Metrics.TotalJobs
 		q.metrics.TotalBlocks += resp.Metrics.TotalBlocks