chore: improve metrics.go stats logging to include bloom filter time as well as fix missing size info in query-frontend (#20093)

(cherry picked from commit 0ecb4fb)

Author: slim-bean

pkg/indexgateway/gateway.go

@@ -226,7 +226,9 @@ func (g *Gateway) GetChunkRef(ctx context.Context, req *logproto.GetChunkRefRequ
 	}
 
 	predicate := chunk.NewPredicate(matchers, &req.Plan)
+	chunkRefsLookupStart := time.Now()
 	chunks, _, err := g.indexQuerier.GetChunks(ctx, instanceID, req.From, req.Through, predicate, nil)
+	chunkRefsLookupDuration := time.Since(chunkRefsLookupStart)
 	if err != nil {
 		return nil, err
 	}
@@ -243,6 +245,7 @@ func (g *Gateway) GetChunkRef(ctx context.Context, req *logproto.GetChunkRefRequ
 	initialChunkCount := len(result.Refs)
 	result.Stats.TotalChunks = int64(initialChunkCount)
 	result.Stats.PostFilterChunks = int64(initialChunkCount) // populate early for error reponses
+	result.Stats.ChunkRefsLookupTime = chunkRefsLookupDuration.Seconds()
 
 	defer func() {
 		if err == nil {
@@ -278,17 +281,19 @@ func (g *Gateway) GetChunkRef(ctx context.Context, req *logproto.GetChunkRefRequ
 
 	start = time.Now()
 	chunkRefs, used, err := g.bloomQuerier.FilterChunkRefs(ctx, instanceID, req.From, req.Through, seriesMap, result.Refs, req.Plan)
+	bloomFilterDuration := time.Since(start)
 	if err != nil {
 		return nil, err
 	}
 	sp.AddEvent("bloomQuerier.FilterChunkRefs", trace.WithAttributes(
-		attribute.String("duration", time.Since(start).String()),
+		attribute.String("duration", bloomFilterDuration.String()),
 	))
 
 	result.Refs = chunkRefs
 	level.Info(logger).Log("msg", "return filtered chunk refs", "unfiltered", initialChunkCount, "filtered", len(result.Refs), "used_blooms", used)
 	result.Stats.PostFilterChunks = int64(len(result.Refs))
 	result.Stats.UsedBloomFilters = used
+	result.Stats.BloomFilterTime = bloomFilterDuration.Seconds()
 	return result, nil
 }
 

pkg/logql/metrics.go

@@ -115,13 +115,12 @@ func RecordRangeAndInstantQueryMetrics(
 	result promql_parser.Value,
 ) {
 	var (
-		logger              = fixLogger(ctx, log)
-		rangeType           = GetRangeType(p)
-		rt                  = string(rangeType)
-		latencyType         = latencyTypeFast
-		returnedLines       = 0
-		cardinalityEstimate = uint64(0)
-		queryTags, _        = ctx.Value(httpreq.QueryTagsHTTPHeader).(string) // it's ok to be empty.
+		logger        = fixLogger(ctx, log)
+		rangeType     = GetRangeType(p)
+		rt            = string(rangeType)
+		latencyType   = latencyTypeFast
+		returnedLines = 0
+		queryTags, _  = ctx.Value(httpreq.QueryTagsHTTPHeader).(string) // it's ok to be empty.
 	)
 
 	queryType, err := QueryType(p.GetExpression())
@@ -162,10 +161,6 @@ func RecordRangeAndInstantQueryMetrics(
 		bloomRatio = float64(stats.Index.TotalChunks-stats.Index.PostFilterChunks) / float64(stats.Index.TotalChunks)
 	}
 
-	if r, ok := result.(CountMinSketchVector); ok {
-		cardinalityEstimate = r.F.HyperLogLog.Estimate()
-	}
-
 	logValues = append(logValues, []interface{}{
 		"latency", latencyType, // this can be used to filter log lines.
 		"query", query,
@@ -213,8 +208,6 @@ func RecordRangeAndInstantQueryMetrics(
 		"cache_result_hit", resultCache.EntriesFound,
 		"cache_result_download_time", resultCache.CacheDownloadTime(),
 		"cache_result_query_length_served", resultCache.CacheQueryLengthServed(),
-		// Cardinality estimate for some approximate query types
-		"cardinality_estimate", cardinalityEstimate,
 		// The total of chunk reference fetched from index.
 		"ingester_chunk_refs", stats.Ingester.Store.GetTotalChunksRef(),
 		// Total number of chunks fetched.
@@ -240,8 +233,15 @@ func RecordRangeAndInstantQueryMetrics(
 		"index_bloom_filter_ratio", fmt.Sprintf("%.2f", bloomRatio),
 		"index_used_bloom_filters", stats.Index.UsedBloomFilters,
 		"index_shard_resolver_duration", time.Duration(stats.Index.ShardsDuration),
+		"index_bloom_filter_time", logql_stats.ConvertSecondsToNanoseconds(stats.Index.BloomFilterTime),
+		"index_chunk_refs_lookup_time", logql_stats.ConvertSecondsToNanoseconds(stats.Index.ChunkRefsLookupTime),
 	}...)
 
+	if r, ok := result.(CountMinSketchVector); ok {
+		cardinalityEstimate := r.F.HyperLogLog.Estimate()
+		logValues = append(logValues, "cardinality_estimate", cardinalityEstimate)
+	}
+
 	logValues = append(logValues, httpreq.TagsToKeyValues(queryTags)...)
 
 	if httpreq.ExtractHeader(ctx, httpreq.LokiDisablePipelineWrappersHeader) == "true" {

pkg/logqlmodel/stats/context.go

@@ -295,6 +295,8 @@ func (i *Index) Merge(m Index) {
 	i.PostFilterChunks += m.PostFilterChunks
 	i.ShardsDuration += m.ShardsDuration
 	i.TotalStreams += m.TotalStreams
+	i.ChunkRefsLookupTime += m.ChunkRefsLookupTime
+	i.BloomFilterTime += m.BloomFilterTime
 	if m.UsedBloomFilters {
 		i.UsedBloomFilters = m.UsedBloomFilters
 	}

pkg/logqlmodel/stats/stats.pb.go

@@ -114,6 +114,14 @@ message Index {
   bool usedBloomFilters = 4 [(gogoproto.jsontag) = "usedBloomFilters"];
   // Total unique streams matched
   int64 totalStreams = 5 [(gogoproto.jsontag) = "totalStreams"];
+  // Time spent fetching chunk refs from index in seconds.
+  // In addition to internal calculations this is also returned by the HTTP API.
+  // Grafana expects time values to be returned in seconds as float.
+  double chunkRefsLookupTime = 6 [(gogoproto.jsontag) = "chunkRefsLookupTime"];
+  // Time spent filtering chunks with bloom gateway in seconds.
+  // In addition to internal calculations this is also returned by the HTTP API.
+  // Grafana expects time values to be returned in seconds as float.
+  double bloomFilterTime = 7 [(gogoproto.jsontag) = "bloomFilterTime"];
 }
 
 message Querier {

pkg/logqlmodel/stats/stats.proto

@@ -1994,6 +1994,8 @@ var (
 			}
 		},
 		"index": {
+			"bloomFilterTime": 0,
+			"chunkRefsLookupTime": 0,
 			"postFilterChunks": 0,
 			"totalChunks": 0,
 			"totalStreams": 0,

pkg/querier/queryrange/codec_test.go

@@ -14,6 +14,8 @@ import (
 
 var emptyStats = `"stats": {
 	"index": {
+		"bloomFilterTime": 0,
+		"chunkRefsLookupTime": 0,
 		"postFilterChunks": 0,
 		"totalChunks": 0,
 		"totalStreams": 0,

pkg/querier/queryrange/prometheus_test.go

@@ -12,8 +12,10 @@ import (
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
 	"github.com/grafana/dskit/middleware"
+	"github.com/prometheus/prometheus/promql"
 	promql_parser "github.com/prometheus/prometheus/promql/parser"
 
+	"github.com/grafana/loki/v3/pkg/loghttp"
 	"github.com/grafana/loki/v3/pkg/logproto"
 
 	"github.com/grafana/loki/v3/pkg/logql"
@@ -146,12 +148,30 @@ func StatsCollectorMiddleware() queryrangebase.Middleware {
 				switch r := resp.(type) {
 				case *LokiResponse:
 					responseStats = &r.Statistics
-					totalEntries = int(logqlmodel.Streams(r.Data.Result).Lines())
+					res = logqlmodel.Streams(r.Data.Result)
+					totalEntries = int(res.(logqlmodel.Streams).Lines())
 					queryType = queryTypeLog
 				case *LokiPromResponse:
 					responseStats = &r.Statistics
+
 					if r.Response != nil {
 						totalEntries = len(r.Response.Data.Result)
+						// Convert the response to promql_parser.Value for stats calculation
+						switch r.Response.Data.ResultType {
+						case loghttp.ResultTypeVector:
+							res = sampleStreamToVector(r.Response.Data.Result)
+						case loghttp.ResultTypeMatrix:
+							res = sampleStreamToMatrix(r.Response.Data.Result)
+						case loghttp.ResultTypeScalar:
+							// Scalar is represented as a single SampleStream with one sample
+							if len(r.Response.Data.Result) > 0 && len(r.Response.Data.Result[0].Samples) > 0 {
+								sample := r.Response.Data.Result[0].Samples[0]
+								res = promql.Scalar{
+									T: sample.TimestampMs,
+									V: sample.Value,
+								}
+							}
+						}
 					}
 
 					queryType = queryTypeMetric

pkg/querier/queryrange/stats.go

@@ -57,6 +57,8 @@ var queryTests = []struct {
 			],
 			"stats" : {
 				"index": {
+					"bloomFilterTime": 0,
+					"chunkRefsLookupTime": 0,
 					"postFilterChunks": 0,
 					"totalChunks": 0,
 					"totalStreams": 0,

pkg/util/marshal/legacy/marshal_test.go

@@ -28,6 +28,8 @@ import (
 
 const emptyStats = `{
 	"index": {
+		"bloomFilterTime": 0,
+		"chunkRefsLookupTime": 0,
 		"postFilterChunks": 0,
 		"totalChunks": 0,
 		"totalStreams": 0,