chore: Add bool to metrics.go log line whether query used bloom filters (#14645)

Better observability of which queries use bloom filters for chunk filtering.

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

Author: chaudum

pkg/bloomgateway/querier.go

@@ -101,10 +101,10 @@ func convertToShortRef(ref *logproto.ChunkRef) *logproto.ShortRef {
 	return &logproto.ShortRef{From: ref.From, Through: ref.Through, Checksum: ref.Checksum}
 }
 
-func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from, through model.Time, chunkRefs []*logproto.ChunkRef, queryPlan plan.QueryPlan) ([]*logproto.ChunkRef, error) {
+func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from, through model.Time, chunkRefs []*logproto.ChunkRef, queryPlan plan.QueryPlan) ([]*logproto.ChunkRef, bool, error) {
 	// Shortcut that does not require any filtering
 	if !bq.limits.BloomGatewayEnabled(tenant) || len(chunkRefs) == 0 || len(v1.ExtractTestableLabelMatchers(queryPlan.AST)) == 0 {
-		return chunkRefs, nil
+		return chunkRefs, false, nil
 	}
 
 	logger, ctx := spanlogger.NewWithLogger(ctx, bq.logger, "bloomquerier.FilterChunkRefs")
@@ -140,7 +140,7 @@ func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from
 			bq.metrics.seriesTotal.Add(float64(preFilterSeries))
 			bq.metrics.seriesFiltered.Add(0)
 
-			return chunkRefs, nil
+			return chunkRefs, false, nil
 		}
 	}
 
@@ -152,12 +152,12 @@ func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from
 		day := bloomshipper.NewInterval(s.day.Time, s.day.Add(Day))
 		blocks, skipped, err := bq.blockResolver.Resolve(ctx, tenant, day, s.series)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 
 		refs, err := bq.c.FilterChunks(ctx, tenant, s.interval, blocks, queryPlan)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 
 		skippedGrps = append(skippedGrps, skipped)
@@ -167,7 +167,7 @@ func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from
 	// add chunk refs from series that were not mapped to any blocks
 	skippedDeduped, err := mergeSeries(skippedGrps, nil)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to dedupe skipped series")
+		return nil, false, errors.Wrap(err, "failed to dedupe skipped series")
 	}
 
 	var chunksSkipped int
@@ -177,7 +177,7 @@ func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from
 
 	deduped, err := mergeSeries(responses, nil)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to dedupe results")
+		return nil, false, errors.Wrap(err, "failed to dedupe results")
 	}
 
 	result := make([]*logproto.ChunkRef, 0, len(chunkRefs))
@@ -218,7 +218,7 @@ func (bq *BloomQuerier) FilterChunkRefs(ctx context.Context, tenant string, from
 	bq.metrics.seriesSkipped.Add(float64(len(skippedDeduped)))
 	bq.metrics.seriesFiltered.Add(float64(preFilterSeries - postFilterSeries))
 
-	return result, nil
+	return result, true, nil
 }
 
 // groupChunkRefs takes a slice of chunk refs sorted by their fingerprint and

pkg/bloomgateway/querier_test.go

@@ -79,7 +79,7 @@ func TestBloomQuerier(t *testing.T) {
 		}
 		expr, err := syntax.ParseExpr(`{foo="bar"}`)
 		require.NoError(t, err)
-		res, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
+		res, _, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
 		require.NoError(t, err)
 		require.Equal(t, chunkRefs, res)
 		require.Equal(t, 0, c.callCount)
@@ -95,7 +95,7 @@ func TestBloomQuerier(t *testing.T) {
 		chunkRefs := []*logproto.ChunkRef{}
 		expr, err := syntax.ParseExpr(`{foo="bar"} | trace_id="exists"`)
 		require.NoError(t, err)
-		res, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
+		res, _, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
 		require.NoError(t, err)
 		require.Equal(t, chunkRefs, res)
 		require.Equal(t, 0, c.callCount)
@@ -115,7 +115,7 @@ func TestBloomQuerier(t *testing.T) {
 		}
 		expr, err := syntax.ParseExpr(`{foo="bar"} | trace_id="exists"`)
 		require.NoError(t, err)
-		res, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
+		res, _, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
 		require.Error(t, err)
 		require.Nil(t, res)
 	})
@@ -134,7 +134,7 @@ func TestBloomQuerier(t *testing.T) {
 		}
 		expr, err := syntax.ParseExpr(`{foo="bar"} | trace_id="exists"`)
 		require.NoError(t, err)
-		res, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
+		res, _, err := bq.FilterChunkRefs(ctx, tenant, from, through, chunkRefs, plan.QueryPlan{AST: expr})
 		require.NoError(t, err)
 		require.Equal(t, chunkRefs, res)
 		require.Equal(t, 2, c.callCount)

pkg/indexgateway/gateway.go

@@ -58,7 +58,7 @@ type IndexClientWithRange struct {
 }
 
 type BloomQuerier interface {
-	FilterChunkRefs(ctx context.Context, tenant string, from, through model.Time, chunks []*logproto.ChunkRef, plan plan.QueryPlan) ([]*logproto.ChunkRef, error)
+	FilterChunkRefs(ctx context.Context, tenant string, from, through model.Time, chunks []*logproto.ChunkRef, plan plan.QueryPlan) ([]*logproto.ChunkRef, bool, error)
 }
 
 type Gateway struct {
@@ -257,14 +257,15 @@ func (g *Gateway) GetChunkRef(ctx context.Context, req *logproto.GetChunkRefRequ
 		return result, nil
 	}
 
-	chunkRefs, err := g.bloomQuerier.FilterChunkRefs(ctx, instanceID, req.From, req.Through, result.Refs, req.Plan)
+	chunkRefs, used, err := g.bloomQuerier.FilterChunkRefs(ctx, instanceID, req.From, req.Through, result.Refs, req.Plan)
 	if err != nil {
 		return nil, err
 	}
 
 	result.Refs = chunkRefs
-	level.Info(logger).Log("msg", "return filtered chunk refs", "unfiltered", initialChunkCount, "filtered", len(result.Refs))
+	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
 	return result, nil
 }
 

pkg/logql/metrics.go

@@ -208,6 +208,7 @@ func RecordRangeAndInstantQueryMetrics(
 		"index_total_chunks", stats.Index.TotalChunks,
 		"index_post_bloom_filter_chunks", stats.Index.PostFilterChunks,
 		"index_bloom_filter_ratio", fmt.Sprintf("%.2f", bloomRatio),
+		"index_used_bloom_filters", stats.Index.UsedBloomFilters,
 		"index_shard_resolver_duration", time.Duration(stats.Index.ShardsDuration),
 	}...)
 

pkg/logqlmodel/stats/context.go

@@ -240,6 +240,9 @@ func (i *Index) Merge(m Index) {
 	i.TotalChunks += m.TotalChunks
 	i.PostFilterChunks += m.PostFilterChunks
 	i.ShardsDuration += m.ShardsDuration
+	if m.UsedBloomFilters {
+		i.UsedBloomFilters = m.UsedBloomFilters
+	}
 }
 
 func (c *Caches) Merge(m Caches) {