Streaming labels values PR 1

[tcp13equals2]

What this PR does

This is the first PR for the implementation of a new streaming labels/values search API. See prometheus/proposals#74.

This PR leverages Prometheus vendored storage interface changes and search/filter utilities.

For reference - noting not all of these are merged as yet.

• util/strutil: add Jaro-Winkler similarity implementation prometheus/prometheus#18405
• util/strutil: add subsequence matching implementation prometheus/prometheus#18402
• strutil/subsequence: rank exact matches above non-exact matches prometheus/prometheus#18576
• storage: introduce search interface with scoring and filtering prometheus/prometheus#18497
• web/api: add search API endpoint prometheus/prometheus#18573

This PR focuses on the Searcher implementation which will run on the Ingester and Store-Gateways. The changes are at the gRPC boundary into these components.

Once this PR is merged, the next PR will focus on the Querier and fan-out of the search requests to the changes made in this PR.

TBD - should a change log be recorded for non-user facing change. None of the changes in this PR will be inline to any user operation at this time.

Which issue(s) this PR fixes or relates to

Fixes #

Checklist

• Tests updated.
• Documentation added.
• CHANGELOG.md updated - the order of entries should be [CHANGE], [FEATURE], [ENHANCEMENT], [BUGFIX]. If changelog entry is not needed, please add the changelog-not-needed label to the PR.
• about-versioning.md updated with experimental features.

---

Note

Medium Risk
Introduces new gRPC streaming APIs and non-trivial merge/ordering logic on the read path; while largely additive, it touches core ingester/store-gateway query surfaces and could impact correctness or performance under load.

Overview
Adds a new streaming labels/values search API across ingester and store-gateway, including new gRPC methods SearchLabelNames / SearchLabelValues, request/response protos (SearchFilter, SearchOrdering, SearchResultBatch), and regenerated client/server stubs.

Implements the ingester-side storage.Searcher integration with batched streaming, input validation mapped to InvalidArgument, warning propagation, and context-cancellation checks; wires the new RPCs through activity tracking and profiling wrappers.

Implements the store-gateway search path by running per-block searches concurrently, applying per-block filter/order/limit, and streaming a deduplicated k-way merge via a new PairwiseMergeSearchSets utility (ported from Prometheus), plus wrappers for tracing, activity tracking, and client error wrapping; adds comprehensive tests for batching, ordering, limits, warnings, errors, and cancellation.

^{Reviewed by Cursor Bugbot for commit 3e5862c. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Test assertion is vacuously true on empty slice
  • Updated the test to generate searchBatchSize results so a mid-iteration batch is sent and warnings suppression is meaningfully asserted instead of vacuously passing.

Or push these changes by commenting:

@cursor push d8289ba067

Preview (d8289ba067)

diff --git a/pkg/ingester/ingester_search_test.go b/pkg/ingester/ingester_search_test.go
--- a/pkg/ingester/ingester_search_test.go
+++ b/pkg/ingester/ingester_search_test.go
@@ -231,18 +231,19 @@
 
 func TestStreamSearchResultsPropagatesErrInsteadOfWarnings(t *testing.T) {
 	want := errors.New("boom")
-	rs := &fakeSearchResultSet{
-		results: []storage.SearchResult{{Value: "a", Score: 1.0}},
-		err:     want,
-		warns:   addAnnotation(nil, "should not appear"),
+	// Generate enough results to trigger at least one mid-iteration batch send.
+	results := make([]storage.SearchResult, searchBatchSize)
+	for i := 0; i < searchBatchSize; i++ {
+		results[i] = storage.SearchResult{Value: "a", Score: 1.0}
 	}
+	rs := &fakeSearchResultSet{results: results, err: want, warns: addAnnotation(nil, "should not appear")}
 	var sent []*client.SearchResultBatch
 	send := func(b *client.SearchResultBatch) error {
 		sent = append(sent, b)
 		return nil
 	}
 	require.ErrorIs(t, streamSearchResults(rs, send), want)
-	// The single result was sent before iteration ended; the trailer batch
+	// A batch was sent before iteration ended; the trailer batch
 	// (which would carry warnings) is suppressed because rs.Err is non-nil.
 	for _, b := range sent {
 		assert.Empty(t, b.Warnings, "warnings must not be sent when iteration errored")

_{You can send follow-ups to the cloud agent here.}

[tcp13equals2]

@cursor push d8289ba

[cursor]

Could not push Autofix changes. The PR branch has conflicting changes.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Duplicate proto-to-domain conversion logic across packages
  • Centralized shared wire-to-domain conversion helpers in streaminglabelvalues and refactored both packages to use them.

Or push these changes by commenting:

@cursor push e749d6cdba

Preview (e749d6cdba)

diff --git a/pkg/ingester/ingester_search.go b/pkg/ingester/ingester_search.go
--- a/pkg/ingester/ingester_search.go
+++ b/pkg/ingester/ingester_search.go
@@ -128,24 +128,21 @@
 	if wf == nil {
 		return nil, nil
 	}
-	alg := streaminglabelvalues.FuzzAlgSubsequence
-	if wf.FuzzAlg == client.FUZZ_ALG_JARO_WINKLER {
-		alg = streaminglabelvalues.FuzzAlgJaroWinkler
-	}
-	return streaminglabelvalues.NewParams(wf.Terms, !wf.CaseInsensitive, alg, int(wf.FuzzThreshold))
+	return streaminglabelvalues.ParamsFromWire(
+		wf.Terms,
+		wf.CaseInsensitive,
+		wf.FuzzAlg == client.FUZZ_ALG_JARO_WINKLER,
+		int(wf.FuzzThreshold),
+	)
 }
 
 // protoToOrdering maps the wire SearchOrdering enum onto the Prometheus
 // storage.Ordering enum.
 func protoToOrdering(o client.SearchOrdering) storage.Ordering {
-	switch o {
-	case client.ORDER_BY_VALUE_DESC:
-		return storage.OrderByValueDesc
-	case client.ORDER_BY_SCORE_DESC:
-		return storage.OrderByScoreDesc
-	default:
-		return storage.OrderByValueAsc
-	}
+	return streaminglabelvalues.OrderingFromWire(
+		o == client.ORDER_BY_VALUE_DESC,
+		o == client.ORDER_BY_SCORE_DESC,
+	)
 }
 
 // searchResultSender is the minimal interface satisfied by both

diff --git a/pkg/storegateway/bucket_search.go b/pkg/storegateway/bucket_search.go
--- a/pkg/storegateway/bucket_search.go
+++ b/pkg/storegateway/bucket_search.go
@@ -214,23 +214,20 @@
 	if wf == nil {
 		return nil, nil
 	}
-	alg := streaminglabelvalues.FuzzAlgSubsequence
-	if wf.FuzzAlg == storepb.FUZZ_ALG_JARO_WINKLER {
-		alg = streaminglabelvalues.FuzzAlgJaroWinkler
-	}
-	return streaminglabelvalues.NewParams(wf.Terms, !wf.CaseInsensitive, alg, int(wf.FuzzThreshold))
+	return streaminglabelvalues.ParamsFromWire(
+		wf.Terms,
+		wf.CaseInsensitive,
+		wf.FuzzAlg == storepb.FUZZ_ALG_JARO_WINKLER,
+		int(wf.FuzzThreshold),
+	)
 }
 
 // storepbToOrdering maps the wire SearchOrdering enum onto storage.Ordering.
 func storepbToOrdering(o storepb.SearchOrdering) storage.Ordering {
-	switch o {
-	case storepb.ORDER_BY_VALUE_DESC:
-		return storage.OrderByValueDesc
-	case storepb.ORDER_BY_SCORE_DESC:
-		return storage.OrderByScoreDesc
-	default:
-		return storage.OrderByValueAsc
-	}
+	return streaminglabelvalues.OrderingFromWire(
+		o == storepb.ORDER_BY_VALUE_DESC,
+		o == storepb.ORDER_BY_SCORE_DESC,
+	)
 }
 
 // streamBucketSearchResults sends results in batches of searchBatchSize via

diff --git a/pkg/streaminglabelvalues/wire.go b/pkg/streaminglabelvalues/wire.go
new file mode 100644
--- /dev/null
+++ b/pkg/streaminglabelvalues/wire.go
@@ -1,0 +1,33 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+//
+// Helper conversions for translating wire-level search options into
+// streaminglabelvalues and Prometheus domain types. Kept here to avoid
+// duplicating the same logic across multiple gRPC servers.
+package streaminglabelvalues
+
+import (
+	"github.com/prometheus/prometheus/storage"
+)
+
+// ParamsFromWire constructs Params from wire-level fields.
+// - caseInsensitive matches the wire polarity and is inverted internally.
+// - useJaroWinkler selects the Jaro-Winkler fuzzy algorithm; subsequence is default.
+func ParamsFromWire(terms []string, caseInsensitive bool, useJaroWinkler bool, fuzzThreshold int) (*Params, error) {
+	alg := FuzzAlgSubsequence
+	if useJaroWinkler {
+		alg = FuzzAlgJaroWinkler
+	}
+	return NewParams(terms, !caseInsensitive, alg, fuzzThreshold)
+}
+
+// OrderingFromWire maps booleans derived from a wire enum onto storage.Ordering.
+// Prefer score desc when both flags are set (defensive default).
+func OrderingFromWire(isValueDesc bool, isScoreDesc bool) storage.Ordering {
+	if isScoreDesc {
+		return storage.OrderByScoreDesc
+	}
+	if isValueDesc {
+		return storage.OrderByValueDesc
+	}
+	return storage.OrderByValueAsc
+}

_{You can send follow-ups to the cloud agent here.}

[cursor]

Duplicate proto-to-domain conversion logic across packages

Low Severity

storepbToParams/storepbToOrdering in bucket_search.go and protoToParams/protoToOrdering in ingester_search.go contain identical domain-mapping logic (fuzz algorithm translation, ordering translation) duplicated across two packages. The shared streaminglabelvalues package already exists and could host the common ordering-to-storage.Ordering mapping and the fuzz-algorithm conversion, avoiding divergence risk if new enum values are added to only one copy.

Additional Locations (1)

• pkg/ingester/ingester_search.go#L126-L149

 

^{Reviewed by Cursor Bugbot for commit 11c63dd. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Duplicated warningsToStrings function across two packages
  • Extracted the helper to pkg/storage.WarningsToStrings and updated ingester and storegateway to use it, removing both duplicates.

Or push these changes by commenting:

@cursor push 9d8497c3f6

Preview (9d8497c3f6)

diff --git a/pkg/ingester/ingester_search.go b/pkg/ingester/ingester_search.go
--- a/pkg/ingester/ingester_search.go
+++ b/pkg/ingester/ingester_search.go
@@ -9,11 +9,11 @@
 	"github.com/grafana/dskit/tenant"
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/storage"
-	"github.com/prometheus/prometheus/util/annotations"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
 	"github.com/grafana/mimir/pkg/ingester/client"
+	mimirstorage "github.com/grafana/mimir/pkg/storage"
 	"github.com/grafana/mimir/pkg/streaminglabelvalues"
 )
 
@@ -179,7 +179,7 @@
 	if err := rs.Err(); err != nil {
 		return err
 	}
-	batch.Warnings = warningsToStrings(rs.Warnings())
+	batch.Warnings = mimirstorage.WarningsToStrings(rs.Warnings())
 	if len(batch.Results) > 0 || len(batch.Warnings) > 0 {
 		if err := send(batch); err != nil {
 			return err
@@ -188,15 +188,4 @@
 	return nil
 }
 
-// warningsToStrings flattens annotations into a string slice for wire transport.
-// Returns nil for empty input so the proto field is omitted on the wire.
-func warningsToStrings(a annotations.Annotations) []string {
-	if len(a) == 0 {
-		return nil
-	}
-	out := make([]string, 0, len(a))
-	for _, w := range a {
-		out = append(out, w.Error())
-	}
-	return out
-}
+ 

diff --git a/pkg/storage/warnings.go b/pkg/storage/warnings.go
new file mode 100644
--- /dev/null
+++ b/pkg/storage/warnings.go
@@ -1,0 +1,19 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package storage
+
+import "github.com/prometheus/prometheus/util/annotations"
+
+// WarningsToStrings flattens annotations into a string slice for wire transport.
+// Returns nil for empty input so the proto field is omitted on the wire.
+func WarningsToStrings(a annotations.Annotations) []string {
+	if len(a) == 0 {
+		return nil
+	}
+	out := make([]string, 0, len(a))
+	for _, w := range a {
+		out = append(out, w.Error())
+	}
+	return out
+}
+

diff --git a/pkg/storegateway/bucket_search.go b/pkg/storegateway/bucket_search.go
--- a/pkg/storegateway/bucket_search.go
+++ b/pkg/storegateway/bucket_search.go
@@ -10,7 +10,6 @@
 	"github.com/grafana/dskit/runutil"
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/storage"
-	"github.com/prometheus/prometheus/util/annotations"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -251,22 +250,11 @@
 	if err := rs.Err(); err != nil {
 		return err
 	}
-	batch.Warnings = warningsToStrings(rs.Warnings())
+	batch.Warnings = mimirstorage.WarningsToStrings(rs.Warnings())
 	if len(batch.Results) > 0 || len(batch.Warnings) > 0 {
 		return send(batch)
 	}
 	return nil
 }
 
-// warningsToStrings flattens annotations into a string slice for wire transport.
-// Returns nil for empty input so the proto field is omitted on the wire.
-func warningsToStrings(a annotations.Annotations) []string {
-	if len(a) == 0 {
-		return nil
-	}
-	out := make([]string, 0, len(a))
-	for _, w := range a {
-		out = append(out, w.Error())
-	}
-	return out
-}
+

_{You can send follow-ups to the cloud agent here.}

[cursor]

Duplicated warningsToStrings function across two packages

Low Severity

The warningsToStrings function is identically defined in both pkg/ingester/ingester_search.go and pkg/storegateway/bucket_search.go. Both convert annotations.Annotations to []string with the exact same logic. This could live in a shared utility package (e.g., pkg/streaminglabelvalues or pkg/storage) to avoid maintaining two copies.

Additional Locations (1)

• pkg/ingester/ingester_search.go#L192-L202

 

^{Reviewed by Cursor Bugbot for commit f69c271. Configure here.}

[tcp13equals2]

@ldufr - I have addressed your comments so far - do you have any other review comments to add? Thanks!

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

There are 3 total unresolved issues (including 2 from previous reviews).

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Unused index reader in SearchLabelValues block fan-out
  • Removed the unnecessary indexReader creation and close in SearchLabelValues since blockLabelValues manages its own reader.

Or push these changes by commenting:

@cursor push 7fb125ab8b

Preview (7fb125ab8b)

diff --git a/pkg/storegateway/bucket_search.go b/pkg/storegateway/bucket_search.go
--- a/pkg/storegateway/bucket_search.go
+++ b/pkg/storegateway/bucket_search.go
@@ -135,11 +135,7 @@
 	)
 
 	s.blockSet.filter(req.Start, req.End, nil, func(b *bucketBlock) {
-		// indexReader is created here (outside the goroutine) to hold the block open.
-		indexr := b.indexReader(nil)
 		g.Go(func() error {
-			defer runutil.CloseWithLogOnErr(s.logger, indexr, "search label values")
-
 			b.ensureIndexHeaderLoaded(gctx, stats)
 
 			result, err := blockLabelValues(gctx, b, s.postingsStrategy, s.maxSeriesPerBatch, req.Label, matchers, s.logger, stats)

_{You can send follow-ups to the cloud agent here.}

^{Reviewed by Cursor Bugbot for commit 144e481. Configure here.}

[cursor]

Unused index reader in SearchLabelValues block fan-out

Low Severity

In SearchLabelValues, indexr is created via b.indexReader(nil) and closed in the goroutine, but it is never passed to blockLabelValues — unlike SearchLabelNames, where indexr is created with s.postingsStrategy and actually used by blockLabelNames. The nil strategy argument and the unused reader suggest this was copied from SearchLabelNames without adapting it to match blockLabelValues's different signature, which takes the block directly.

 

^{Reviewed by Cursor Bugbot for commit 144e481. Configure here.}

[tcp13equals2]

Updated with better comment

[tcp13equals2]

Note to reviewers - I am doing these links since the PR where this is introduced in Prometheus has not yet merged. I will update these once Prometheus PRs are merged.

[tcp13equals2]

@ldufr I fixed those couple of comments - can I get you to give the review stamp again please