mirror of
https://github.com/prometheus/prometheus.git
synced 2026-10-28 10:44:59 +08:00
storage: Fix search label values limit order (#19694)
* storage: Fix search label values limit order --------- Signed-off-by: Andrew Hall <andrew.hall@grafana.com>
This commit is contained in:
1 parent
ba860bfd0e
commit
f6ccc4acc4
10 files changed
+595
-13
No files matched your search
@@ -366,6 +366,66 @@ func linearResultCap(numValues, limit int) int {
|
||||
return allocCap
|
||||
}
|
||||
|
||||
// smallestNValues returns the n smallest of values in ascending order. It does
|
||||
// not modify values, so it is safe to use on a shared slice. A bounded heap
|
||||
// avoids sorting or copying the whole input, which for a high cardinality label
|
||||
// is much larger than n.
|
||||
//
|
||||
// n must be greater than zero and less than len(values).
|
||||
func smallestNValues(values []string, n int) []string {
|
||||
h := make(smallestNHeap, n)
|
||||
copy(h, values[:n])
|
||||
h.init()
|
||||
for _, v := range values[n:] {
|
||||
// The heap maximum is the worst value currently kept, so a smaller
|
||||
// candidate replaces it and anything else cannot improve the result.
|
||||
if v < h[0] {
|
||||
h[0] = v
|
||||
h.siftDown(0)
|
||||
}
|
||||
}
|
||||
// The heap is only partially ordered, so the survivors need a sort. It runs
|
||||
// over n entries, not over the whole input.
|
||||
slices.Sort(h)
|
||||
return h
|
||||
}
|
||||
|
||||
// smallestNHeap is a typed binary max-heap over label values, so heap[0] is the
|
||||
// largest value currently kept. Replacing the maximum on smaller candidates
|
||||
// keeps the N smallest values without sorting the full input and without the
|
||||
// per-operation interface boxing that container/heap would introduce.
|
||||
type smallestNHeap []string
|
||||
|
||||
// init establishes the heap property over the whole slice. It works bottom-up
|
||||
// from the last internal node, which is cheaper than inserting the values one
|
||||
// at a time; the indices from len(h)/2 up are leaves and so are heaps already.
|
||||
func (h smallestNHeap) init() {
|
||||
for i := len(h)/2 - 1; i >= 0; i-- {
|
||||
h.siftDown(i)
|
||||
}
|
||||
}
|
||||
|
||||
// siftDown restores the heap property at index i, assuming both subtrees of i
|
||||
// are already heaps.
|
||||
func (h smallestNHeap) siftDown(i int) {
|
||||
n := len(h)
|
||||
for {
|
||||
left := 2*i + 1
|
||||
if left >= n {
|
||||
return
|
||||
}
|
||||
largest := left
|
||||
if right := left + 1; right < n && h[right] > h[left] {
|
||||
largest = right
|
||||
}
|
||||
if h[largest] <= h[i] {
|
||||
return
|
||||
}
|
||||
h[i], h[largest] = h[largest], h[i]
|
||||
i = largest
|
||||
}
|
||||
}
|
||||
|
||||
// topKByScore returns the top-K matches under the (Score desc, Value asc) total
|
||||
// order, using a min-heap of size limit. This avoids sorting the full matched
|
||||
// set when only the best Limit results are needed.
|
||||
|
||||
@@ -331,6 +331,34 @@ func ErrSearchResultSet(err error, warnings ...annotations.Annotations) SearchRe
|
||||
type LabelHints struct {
|
||||
// Maximum number of results returned. Use a value of 0 to disable.
|
||||
Limit int
|
||||
|
||||
// LimitSmallest requires Limit and will select the lexically smallest N
|
||||
// values of the full set.
|
||||
LimitSmallest bool
|
||||
}
|
||||
|
||||
// AllowsEarlyStop reports whether a read may stop as soon as it holds Limit
|
||||
// values. It will return false when no Limit is set or LimitSmallest is true.
|
||||
func (h *LabelHints) AllowsEarlyStop() bool {
|
||||
return h != nil && h.Limit > 0 && !h.LimitSmallest
|
||||
}
|
||||
|
||||
// ApplyLimit reduces values to at most Limit entries. It keeps the lexically
|
||||
// smallest values when LimitSmallest is set, and otherwise the leading ones. A
|
||||
// nil receiver, and a Limit of zero or less, both leave values unchanged.
|
||||
//
|
||||
// The limited result set is returned, along with a bool which indicates if the result
|
||||
// has re-used the same input slice or if a newly allocated slice has been created.
|
||||
//
|
||||
// Do not mutate the returned slice if the returned allocated=false.
|
||||
func (h *LabelHints) ApplyLimit(values []string) (limited []string, allocated bool) {
|
||||
if h == nil || h.Limit <= 0 || len(values) <= h.Limit {
|
||||
return values, false
|
||||
}
|
||||
if h.LimitSmallest {
|
||||
return smallestNValues(values, h.Limit), true
|
||||
}
|
||||
return values[:h.Limit], false
|
||||
}
|
||||
|
||||
// SearchHints configures search operations with filtering and scoring.
|
||||
|
||||
@@ -14,6 +14,10 @@
|
||||
package storage_test
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand/v2"
|
||||
"slices"
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
@@ -52,3 +56,108 @@ func TestMockSeriesWithST(t *testing.T) {
|
||||
require.Equal(t, []float64{1, 2, 3}, vs)
|
||||
require.Equal(t, []int64{0, 1, 2}, st)
|
||||
}
|
||||
|
||||
func TestLabelHintsApplyLimit(t *testing.T) {
|
||||
// Unordered, so a leading limit and a smallest limit differ.
|
||||
input := []string{"d", "b", "e", "a", "c"}
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
hints *storage.LabelHints
|
||||
want []string
|
||||
wantAllocated bool
|
||||
}{
|
||||
{"nil hints", nil, input, false},
|
||||
{"no limit", &storage.LabelHints{LimitSmallest: true}, input, false},
|
||||
{"limit at count", &storage.LabelHints{Limit: 5, LimitSmallest: true}, input, false},
|
||||
{"limit above count", &storage.LabelHints{Limit: 6, LimitSmallest: true}, input, false},
|
||||
{"negative limit", &storage.LabelHints{Limit: -1, LimitSmallest: true}, input, false},
|
||||
{"negative limit without flag", &storage.LabelHints{Limit: -1}, input, false},
|
||||
{"leading limit", &storage.LabelHints{Limit: 3}, []string{"d", "b", "e"}, false},
|
||||
{"smallest limit", &storage.LabelHints{Limit: 3, LimitSmallest: true}, []string{"a", "b", "c"}, true},
|
||||
{"smallest limit of one", &storage.LabelHints{Limit: 1, LimitSmallest: true}, []string{"a"}, true},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
values := slices.Clone(input)
|
||||
got, allocated := tc.hints.ApplyLimit(values)
|
||||
require.Equal(t, tc.want, got)
|
||||
require.Equal(t, tc.wantAllocated, allocated)
|
||||
require.Equal(t, input, values, "ApplyLimit must not modify its input")
|
||||
|
||||
if allocated {
|
||||
// The caller may keep and modify an allocated result.
|
||||
got[0] = "zzz"
|
||||
require.Equal(t, input, values, "an allocated result must not alias the input")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestLabelHintsApplyLimitSmallest compares the bounded selection against a
|
||||
// full sort over randomly ordered values, at limits either side of a power of
|
||||
// two so the heap is exercised at depth.
|
||||
func TestLabelHintsApplyLimitSmallest(t *testing.T) {
|
||||
const count = 500
|
||||
values := make([]string, count)
|
||||
for i := range values {
|
||||
values[i] = fmt.Sprintf("value_%04d", i)
|
||||
}
|
||||
|
||||
shuffled := slices.Clone(values)
|
||||
rand.New(rand.NewPCG(1, 2)).Shuffle(len(shuffled), func(i, j int) {
|
||||
shuffled[i], shuffled[j] = shuffled[j], shuffled[i]
|
||||
})
|
||||
|
||||
for _, limit := range []int{1, 2, 3, 7, 63, 64, 65, count - 1} {
|
||||
t.Run(strconv.Itoa(limit), func(t *testing.T) {
|
||||
hints := &storage.LabelHints{Limit: limit, LimitSmallest: true}
|
||||
got, allocated := hints.ApplyLimit(shuffled)
|
||||
require.True(t, allocated)
|
||||
require.Equal(t, values[:limit], got)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestLabelHintsApplyLimitEdgeInputs(t *testing.T) {
|
||||
t.Run("nil input", func(t *testing.T) {
|
||||
hints := &storage.LabelHints{Limit: 3, LimitSmallest: true}
|
||||
got, allocated := hints.ApplyLimit(nil)
|
||||
require.Nil(t, got)
|
||||
require.False(t, allocated)
|
||||
})
|
||||
|
||||
t.Run("empty input", func(t *testing.T) {
|
||||
hints := &storage.LabelHints{Limit: 3, LimitSmallest: true}
|
||||
got, allocated := hints.ApplyLimit([]string{})
|
||||
require.Empty(t, got)
|
||||
require.False(t, allocated)
|
||||
})
|
||||
|
||||
t.Run("duplicate values", func(t *testing.T) {
|
||||
// Duplicates must not be collapsed, and must not displace a smaller
|
||||
// value that appears later.
|
||||
hints := &storage.LabelHints{Limit: 3, LimitSmallest: true}
|
||||
got, allocated := hints.ApplyLimit([]string{"b", "b", "c", "a", "b"})
|
||||
require.Equal(t, []string{"a", "b", "b"}, got)
|
||||
require.True(t, allocated)
|
||||
})
|
||||
}
|
||||
|
||||
func TestLabelHintsAllowsEarlyStop(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
hints *storage.LabelHints
|
||||
want bool
|
||||
}{
|
||||
{"nil hints", nil, false},
|
||||
{"no limit", &storage.LabelHints{}, false},
|
||||
{"negative limit", &storage.LabelHints{Limit: -1}, false},
|
||||
{"limit only", &storage.LabelHints{Limit: 1}, true},
|
||||
{"limit with smallest", &storage.LabelHints{Limit: 1, LimitSmallest: true}, false},
|
||||
{"smallest without limit", &storage.LabelHints{LimitSmallest: true}, false},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
require.Equal(t, tc.want, tc.hints.AllowsEarlyStop())
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -451,6 +451,66 @@ func TestReadIndexFormatV1(t *testing.T) {
|
||||
}, query(t, q, labels.MustNewMatcher(labels.MatchNotRegexp, "foo", "^.?$")))
|
||||
}
|
||||
|
||||
// TestSearchLabelValuesIndexFormatV1 covers the limit selection on a FormatV1
|
||||
// index, whose label values are held in a map and so are read in an arbitrary
|
||||
// order. The current writer only emits FormatV2, so this fixture is the only
|
||||
// way to reach that branch of index.Reader.LabelValues.
|
||||
func TestSearchLabelValuesIndexFormatV1(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
block, err := OpenBlock(nil, filepath.Join("testdata", "index_format_v1"), nil, nil)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, block.Close()) })
|
||||
require.Equal(t, index.FormatV1, block.meta.Version, "fixture must exercise the FormatV1 branch")
|
||||
|
||||
// The fixture holds "bar" values "0" to "99", so the lexically smallest are
|
||||
// not the numerically smallest and an arbitrary subset is easy to spot.
|
||||
wantSmallest := []string{"0", "1", "10", "11", "12"}
|
||||
|
||||
t.Run("index reader keeps the smallest values", func(t *testing.T) {
|
||||
ir, err := block.Index()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, ir.Close()) })
|
||||
|
||||
all, err := ir.LabelValues(ctx, "bar", nil)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, all, 100)
|
||||
require.False(t, slices.IsSorted(all), "precondition: the map read is unordered")
|
||||
|
||||
got, err := ir.LabelValues(ctx, "bar", &storage.LabelHints{Limit: 5, LimitSmallest: true})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, wantSmallest, got)
|
||||
})
|
||||
|
||||
t.Run("legacy limit still takes any subset", func(t *testing.T) {
|
||||
ir, err := block.Index()
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, ir.Close()) })
|
||||
|
||||
// Without LimitSmallest the read may stop early, so only the count is
|
||||
// guaranteed. Asserting the values would pin Go's map order.
|
||||
got, err := ir.LabelValues(ctx, "bar", &storage.LabelHints{Limit: 5})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, got, 5)
|
||||
})
|
||||
|
||||
t.Run("search returns the smallest values", func(t *testing.T) {
|
||||
q, err := NewBlockQuerier(block, 0, 1000)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, q.Close()) })
|
||||
|
||||
rs := q.(storage.Searcher).SearchLabelValues(ctx, "bar", &storage.SearchHints{Limit: 5})
|
||||
t.Cleanup(func() { require.NoError(t, rs.Close()) })
|
||||
|
||||
var got []string
|
||||
for rs.Next() {
|
||||
got = append(got, rs.At().Value)
|
||||
}
|
||||
require.NoError(t, rs.Err())
|
||||
require.Equal(t, wantSmallest, got)
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkLabelValuesWithMatchers(b *testing.B) {
|
||||
tmpdir := b.TempDir()
|
||||
ctx := context.Background()
|
||||
|
||||
+8
-1
@@ -1371,13 +1371,20 @@ func (r *Reader) LabelValues(ctx context.Context, name string, hints *storage.La
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
// The offsets are held in a map, so they are read in an arbitrary
|
||||
// order. Stopping at the limit would return an arbitrary subset, which
|
||||
// a caller asking for the smallest values cannot use.
|
||||
earlyLimit := hints.AllowsEarlyStop()
|
||||
values := make([]string, 0, len(e))
|
||||
for k := range e {
|
||||
if hints != nil && hints.Limit > 0 && len(values) >= hints.Limit {
|
||||
if earlyLimit && len(values) >= hints.Limit {
|
||||
break
|
||||
}
|
||||
values = append(values, k)
|
||||
}
|
||||
if !earlyLimit {
|
||||
values, _ = hints.ApplyLimit(values)
|
||||
}
|
||||
return values, nil
|
||||
}
|
||||
e, ok := r.postings[name]
|
||||
|
||||
@@ -173,8 +173,14 @@ func (p *MemPostings) LabelValues(_ context.Context, name string, hints *storage
|
||||
values := p.lvs[name]
|
||||
p.mtx.RUnlock()
|
||||
|
||||
if hints != nil && hints.Limit > 0 && len(values) > hints.Limit {
|
||||
values = values[:hints.Limit]
|
||||
// The values are not held in order, so keeping the leading Limit of them
|
||||
// returns an arbitrary subset. ApplyLimit selects the smallest values when
|
||||
// the caller needs them.
|
||||
values, allocated := hints.ApplyLimit(values)
|
||||
if allocated {
|
||||
// The result no longer aliases p.lvs[name], so the copy below would be
|
||||
// a second copy of the same values.
|
||||
return values
|
||||
}
|
||||
|
||||
// The slice from p.lvs[name] is shared between all readers, and it is append-only.
|
||||
|
||||
@@ -1519,3 +1519,61 @@ func TestMemPostings_PostingsForLabelMatchingHonorsContextCancel(t *testing.T) {
|
||||
require.Error(t, p.Err())
|
||||
require.Equal(t, failAfter+1, ctx.Count()) // Plus one for the Err() call that puts the error in the result.
|
||||
}
|
||||
|
||||
func TestMemPostings_LabelValuesLimitSmallest(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// Added in descending order
|
||||
p := NewMemPostings()
|
||||
all := []string{"a", "b", "c", "d", "e"}
|
||||
for i, v := range slices.Backward(all) {
|
||||
p.Add(storage.SeriesRef(i+1), labels.FromStrings("lbl", v))
|
||||
}
|
||||
|
||||
shared := slices.Clone(p.lvs["lbl"])
|
||||
require.Equal(t, []string{"e", "d", "c", "b", "a"}, shared,
|
||||
"precondition: the shared value list is held in insertion order")
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
label string
|
||||
hints *storage.LabelHints
|
||||
want []string
|
||||
}{
|
||||
{"nil hints", "lbl", nil, nil},
|
||||
{"no limit", "lbl", &storage.LabelHints{LimitSmallest: true}, nil},
|
||||
{"limit below count", "lbl", &storage.LabelHints{Limit: 2, LimitSmallest: true}, []string{"a", "b"}},
|
||||
{"limit of one", "lbl", &storage.LabelHints{Limit: 1, LimitSmallest: true}, []string{"a"}},
|
||||
{"limit at count", "lbl", &storage.LabelHints{Limit: 5, LimitSmallest: true}, nil},
|
||||
{"limit above count", "lbl", &storage.LabelHints{Limit: 10, LimitSmallest: true}, nil},
|
||||
{"limit without flag", "lbl", &storage.LabelHints{Limit: 2}, []string{"e", "d"}},
|
||||
{"unknown label name", "missing", &storage.LabelHints{Limit: 2, LimitSmallest: true}, nil},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
got := p.LabelValues(ctx, tc.label, tc.hints)
|
||||
switch {
|
||||
case tc.label == "missing":
|
||||
require.Nil(t, got)
|
||||
case tc.want == nil:
|
||||
// The whole set, in whatever order it is held.
|
||||
require.ElementsMatch(t, all, got)
|
||||
default:
|
||||
require.Equal(t, tc.want, got)
|
||||
}
|
||||
|
||||
require.Equal(t, shared, p.lvs["lbl"], "the shared value list must not be mutated")
|
||||
if len(got) > 0 {
|
||||
// A write through the result must not reach the shared array.
|
||||
got[0] = "zzz"
|
||||
require.Equal(t, shared, p.lvs["lbl"], "the result must not alias the shared value list")
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Delete rebuilds the value list from a map, so the order afterwards is
|
||||
// arbitrary.
|
||||
p.Delete(map[storage.SeriesRef]struct{}{3: {}}, map[labels.Label]struct{}{
|
||||
{Name: "lbl", Value: "c"}: {},
|
||||
})
|
||||
require.Equal(t, []string{"a", "b", "d"}, p.LabelValues(ctx, "lbl", &storage.LabelHints{Limit: 3, LimitSmallest: true}))
|
||||
}
|
||||
+15
-10
@@ -114,22 +114,23 @@ func (q *blockBaseQuerier) SearchLabelValues(ctx context.Context, name string, h
|
||||
|
||||
// Limit pushdown is only correct when natural (ascending) index order
|
||||
// is preserved all the way to the output and no filtering discards
|
||||
// values ahead of the limit.
|
||||
// values ahead of the limit. LimitSmallest is also set to true to ensure
|
||||
// we obtain the smallest N results not just N results.
|
||||
labelHints := &storage.LabelHints{}
|
||||
if hints.OrderBy == storage.OrderByValueAsc && hints.Filter == nil {
|
||||
labelHints.Limit = hints.Limit
|
||||
labelHints.LimitSmallest = true
|
||||
}
|
||||
|
||||
var (
|
||||
values []string
|
||||
err error
|
||||
)
|
||||
switch hints.OrderBy {
|
||||
case storage.OrderByScoreDesc:
|
||||
// Score-based sorting happens in ApplySearchHints; avoid the
|
||||
// index-level sort.
|
||||
// ApplySearchHints needs the values ascending by value.
|
||||
// OrderByScore relies on the score calculated by the filter.
|
||||
if hints.OrderBy == storage.OrderByScoreDesc && hints.Filter != nil {
|
||||
values, err = q.index.LabelValues(ctx, name, labelHints, matchers...)
|
||||
default:
|
||||
} else {
|
||||
values, err = q.index.SortedLabelValues(ctx, name, labelHints, matchers...)
|
||||
}
|
||||
if err != nil {
|
||||
@@ -503,9 +504,7 @@ func labelValuesWithMatchers(ctx context.Context, r IndexReader, name string, hi
|
||||
|
||||
// If we don't have any matchers for other labels, then we're done.
|
||||
if !hasMatchersForOtherLabels {
|
||||
if hints != nil && hints.Limit > 0 && len(allValues) > hints.Limit {
|
||||
allValues = allValues[:hints.Limit]
|
||||
}
|
||||
allValues, _ = hints.ApplyLimit(allValues)
|
||||
return allValues, nil
|
||||
}
|
||||
|
||||
@@ -526,14 +525,20 @@ func labelValuesWithMatchers(ctx context.Context, r IndexReader, name string, hi
|
||||
return nil, fmt.Errorf("intersecting postings: %w", err)
|
||||
}
|
||||
|
||||
// FindIntersectingPostings returns the indexes ordered by series reference,
|
||||
// not by value, so stopping at the limit returns an arbitrary subset. A
|
||||
// caller asking for the smallest values has to see all of them first.
|
||||
earlyLimit := hints.AllowsEarlyStop()
|
||||
values := make([]string, 0, len(indexes))
|
||||
for _, idx := range indexes {
|
||||
values = append(values, allValues[idx])
|
||||
if hints != nil && hints.Limit > 0 && len(values) >= hints.Limit {
|
||||
if earlyLimit && len(values) >= hints.Limit {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// A no-op unless a limit needs the smallest of what we collected.
|
||||
values, _ = hints.ApplyLimit(values)
|
||||
return values, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -16,6 +16,7 @@ package tsdb
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math/rand/v2"
|
||||
"strconv"
|
||||
"testing"
|
||||
|
||||
@@ -24,6 +25,7 @@ import (
|
||||
"github.com/prometheus/prometheus/model/labels"
|
||||
"github.com/prometheus/prometheus/storage"
|
||||
"github.com/prometheus/prometheus/tsdb/index"
|
||||
"github.com/prometheus/prometheus/util/compression"
|
||||
)
|
||||
|
||||
// Make entries ~50B in size, to emulate real-world high cardinality.
|
||||
@@ -364,3 +366,69 @@ func BenchmarkQuerierSelectWithOutOfOrder(b *testing.B) {
|
||||
benchmarkSelect(b, db, numSeries, false)
|
||||
})
|
||||
}
|
||||
|
||||
// BenchmarkSearchLabelValues measures a full result consumption over a warm
|
||||
// index. The values are appended in a permuted order, so the head's in-memory
|
||||
// value list is not ascending and a limit cannot be applied while reading.
|
||||
func BenchmarkSearchLabelValues(b *testing.B) {
|
||||
for _, cardinality := range []int{1000, 100000} {
|
||||
b.Run(fmt.Sprintf("values=%d", cardinality), func(b *testing.B) {
|
||||
head, _ := newTestHead(b, 2000, compression.None, false)
|
||||
ctx := b.Context()
|
||||
app := head.Appender(ctx)
|
||||
for _, i := range rand.New(rand.NewPCG(1, 2)).Perm(cardinality) {
|
||||
_, err := app.Append(0, labels.FromStrings(
|
||||
"__name__", fmt.Sprintf("metric_%06d%s", i, postingsBenchSuffix),
|
||||
"job", "api",
|
||||
), 100, 1)
|
||||
require.NoError(b, err)
|
||||
}
|
||||
require.NoError(b, app.Commit())
|
||||
|
||||
block, err := OpenBlock(nil, createBlockFromHead(b, b.TempDir(), head), nil, nil)
|
||||
require.NoError(b, err)
|
||||
b.Cleanup(func() { require.NoError(b, block.Close()) })
|
||||
|
||||
for _, backend := range []struct {
|
||||
name string
|
||||
reader BlockReader
|
||||
}{
|
||||
{"Head", head},
|
||||
{"Block", block},
|
||||
} {
|
||||
b.Run(backend.name, func(b *testing.B) {
|
||||
q, err := NewBlockQuerier(backend.reader, 0, 200)
|
||||
require.NoError(b, err)
|
||||
b.Cleanup(func() { require.NoError(b, q.Close()) })
|
||||
searcher := q.(storage.Searcher)
|
||||
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
hints *storage.SearchHints
|
||||
matchers []*labels.Matcher
|
||||
}{
|
||||
{"limit=100", &storage.SearchHints{Limit: 100}, nil},
|
||||
{"limit=100_filter", &storage.SearchHints{Limit: 100, Filter: prefixFilter{"metric_00"}}, nil},
|
||||
{"limit=100_matcher", &storage.SearchHints{Limit: 100}, []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "job", "api")}},
|
||||
{"unlimited", &storage.SearchHints{}, nil},
|
||||
} {
|
||||
b.Run(tc.name, func(b *testing.B) {
|
||||
b.ReportAllocs()
|
||||
for b.Loop() {
|
||||
rs := searcher.SearchLabelValues(ctx, "__name__", tc.hints, tc.matchers...)
|
||||
count := 0
|
||||
for rs.Next() {
|
||||
_ = rs.At()
|
||||
count++
|
||||
}
|
||||
require.NoError(b, rs.Err())
|
||||
require.NoError(b, rs.Close())
|
||||
require.Positive(b, count)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -40,6 +40,7 @@ import (
|
||||
"github.com/prometheus/prometheus/tsdb/tombstones"
|
||||
"github.com/prometheus/prometheus/tsdb/tsdbutil"
|
||||
"github.com/prometheus/prometheus/util/annotations"
|
||||
"github.com/prometheus/prometheus/util/compression"
|
||||
"github.com/prometheus/prometheus/util/testutil"
|
||||
)
|
||||
|
||||
@@ -4310,3 +4311,183 @@ func TestBlockBaseQuerierSearchLabelValues(t *testing.T) {
|
||||
require.Equal(t, []string{"staging", "prod", "dev"}, gotValues)
|
||||
})
|
||||
}
|
||||
|
||||
// forEachSearchBackend appends the given label sets to a head, in the order
|
||||
// given, persists an equivalent block, and runs fn against a searcher for each.
|
||||
// Head insertion order is preserved in the in-memory label value lists, so the
|
||||
// caller controls whether that order differs from ascending value order.
|
||||
func forEachSearchBackend(t *testing.T, labelSets []labels.Labels, fn func(t *testing.T, q storage.Searcher)) {
|
||||
t.Helper()
|
||||
|
||||
h, _ := newTestHead(t, 1000, compression.None, false)
|
||||
app := h.Appender(t.Context())
|
||||
for _, ls := range labelSets {
|
||||
_, err := app.Append(0, ls, 2100, 1)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
require.NoError(t, app.Commit())
|
||||
|
||||
block, err := OpenBlock(nil, createBlockFromHead(t, t.TempDir(), h), nil, nil)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, block.Close()) })
|
||||
|
||||
for _, backend := range []struct {
|
||||
name string
|
||||
reader BlockReader
|
||||
}{
|
||||
{"Head", h},
|
||||
{"Block", block},
|
||||
} {
|
||||
t.Run(backend.name, func(t *testing.T) {
|
||||
q, err := NewBlockQuerier(backend.reader, 1500, 2500)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, q.Close()) })
|
||||
fn(t, q.(storage.Searcher))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// requireSearchValues collects rs and asserts that it yields exactly want, in order.
|
||||
func requireSearchValues(t *testing.T, rs storage.SearchResultSet, want []string, msgAndArgs ...any) {
|
||||
t.Helper()
|
||||
got := collectSearchResultSet(t, rs)
|
||||
gotValues := make([]string, len(got))
|
||||
for i, r := range got {
|
||||
gotValues[i] = r.Value
|
||||
}
|
||||
require.Equal(t, want, gotValues, msgAndArgs...)
|
||||
}
|
||||
|
||||
// TestSearchLabelValuesLimitAppliedAfterOrdering checks that a limited ascending
|
||||
// search returns the lexically smallest values. The limit must be applied after
|
||||
// ordering: the head holds label values unsorted, and labelValuesWithMatchers
|
||||
// collects them in postings-intersection order, so truncating first selects an
|
||||
// arbitrary subset.
|
||||
func TestSearchLabelValuesLimitAppliedAfterOrdering(t *testing.T) {
|
||||
// Appended in descending order, so the head's in-memory value list for
|
||||
// __name__ is the reverse of the expected result.
|
||||
unorderedInsertion := []labels.Labels{
|
||||
labels.FromStrings("__name__", "z_metric", "job", "api"),
|
||||
labels.FromStrings("__name__", "m_metric", "job", "api"),
|
||||
labels.FromStrings("__name__", "a_metric", "job", "api"),
|
||||
}
|
||||
// Blocks store series sorted by label set, so the series carrying the
|
||||
// smallest "pod" value gets the highest series reference.
|
||||
// FindIntersectingPostings yields values in series-reference order, which
|
||||
// here is the reverse of ascending "pod" order.
|
||||
unorderedSeriesRefs := []labels.Labels{
|
||||
labels.FromStrings("__name__", "a_metric", "job", "api", "pod", "z_pod"),
|
||||
labels.FromStrings("__name__", "b_metric", "job", "api", "pod", "m_pod"),
|
||||
labels.FromStrings("__name__", "c_metric", "job", "api", "pod", "a_pod"),
|
||||
}
|
||||
jobAPI := []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "job", "api")}
|
||||
|
||||
t.Run("no matchers", func(t *testing.T) {
|
||||
forEachSearchBackend(t, unorderedInsertion, func(t *testing.T, q storage.Searcher) {
|
||||
want := []string{"a_metric", "m_metric", "z_metric"}
|
||||
for limit := 1; limit <= len(want); limit++ {
|
||||
t.Run(strconv.Itoa(limit), func(t *testing.T) {
|
||||
// A nil filter and an accept-all filter must agree.
|
||||
for _, filter := range []storage.Filter{nil, prefixFilter{""}} {
|
||||
rs := q.SearchLabelValues(t.Context(), "__name__", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByValueAsc,
|
||||
Limit: limit,
|
||||
Filter: filter,
|
||||
})
|
||||
requireSearchValues(t, rs, want[:limit], "filter: %T", filter)
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("matcher on another label", func(t *testing.T) {
|
||||
forEachSearchBackend(t, unorderedSeriesRefs, func(t *testing.T, q storage.Searcher) {
|
||||
want := []string{"a_pod", "m_pod", "z_pod"}
|
||||
for limit := 1; limit <= len(want); limit++ {
|
||||
t.Run(strconv.Itoa(limit), func(t *testing.T) {
|
||||
rs := q.SearchLabelValues(t.Context(), "pod", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByValueAsc,
|
||||
Limit: limit,
|
||||
}, jobAPI...)
|
||||
requireSearchValues(t, rs, want[:limit])
|
||||
})
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("matcher on the searched label", func(t *testing.T) {
|
||||
forEachSearchBackend(t, unorderedInsertion, func(t *testing.T, q storage.Searcher) {
|
||||
rs := q.SearchLabelValues(t.Context(), "__name__", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByValueAsc,
|
||||
Limit: 2,
|
||||
}, labels.MustNewMatcher(labels.MatchRegexp, "__name__", ".*_metric"))
|
||||
requireSearchValues(t, rs, []string{"a_metric", "m_metric"})
|
||||
})
|
||||
})
|
||||
|
||||
// ApplySearchHints documents that its input must be ascending by value. With
|
||||
// a nil filter the score ordering degenerates to value ascending, so the
|
||||
// index-level sort cannot be skipped.
|
||||
t.Run("OrderByScoreDesc without a filter", func(t *testing.T) {
|
||||
forEachSearchBackend(t, unorderedInsertion, func(t *testing.T, q storage.Searcher) {
|
||||
rs := q.SearchLabelValues(t.Context(), "__name__", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByScoreDesc,
|
||||
Limit: 2,
|
||||
})
|
||||
requireSearchValues(t, rs, []string{"a_metric", "m_metric"})
|
||||
})
|
||||
})
|
||||
|
||||
// OrderByValueDesc walks the ascending input in reverse, so it also needs
|
||||
// ordered input.
|
||||
t.Run("OrderByValueDesc", func(t *testing.T) {
|
||||
forEachSearchBackend(t, unorderedInsertion, func(t *testing.T, q storage.Searcher) {
|
||||
rs := q.SearchLabelValues(t.Context(), "__name__", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByValueDesc,
|
||||
Limit: 2,
|
||||
})
|
||||
requireSearchValues(t, rs, []string{"z_metric", "m_metric"})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
// TestSearchLabelValuesLimitAcrossMergedQueriers checks the merged case. The
|
||||
// merge truncates the k-way merge of its children, which is only correct if
|
||||
// every child returned its own smallest N.
|
||||
func TestSearchLabelValuesLimitAcrossMergedQueriers(t *testing.T) {
|
||||
ctx := t.Context()
|
||||
|
||||
persisted, _ := newTestHead(t, 1000, compression.None, false)
|
||||
app := persisted.Appender(ctx)
|
||||
for _, name := range []string{"n_metric", "p_metric", "r_metric"} {
|
||||
_, err := app.Append(0, labels.FromStrings("__name__", name, "job", "api"), 2100, 1)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
require.NoError(t, app.Commit())
|
||||
block, err := OpenBlock(nil, createBlockFromHead(t, t.TempDir(), persisted), nil, nil)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, block.Close()) })
|
||||
blockQ, err := NewBlockQuerier(block, 1500, 2500)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Appended in descending order so the head's value list is unordered.
|
||||
head, _ := newTestHead(t, 1000, compression.None, false)
|
||||
app = head.Appender(ctx)
|
||||
for _, name := range []string{"z_metric", "m_metric", "a_metric"} {
|
||||
_, err := app.Append(0, labels.FromStrings("__name__", name, "job", "api"), 2100, 1)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
require.NoError(t, app.Commit())
|
||||
headQ, err := NewBlockQuerier(head, 1500, 2500)
|
||||
require.NoError(t, err)
|
||||
|
||||
q := storage.NewMergeQuerier([]storage.Querier{headQ, blockQ}, nil, storage.ChainedSeriesMerge)
|
||||
t.Cleanup(func() { require.NoError(t, q.Close()) })
|
||||
|
||||
rs := q.(storage.Searcher).SearchLabelValues(ctx, "__name__", &storage.SearchHints{
|
||||
OrderBy: storage.OrderByValueAsc,
|
||||
Limit: 2,
|
||||
})
|
||||
requireSearchValues(t, rs, []string{"a_metric", "m_metric"})
|
||||
}
|
||||
Reference in new issue
Block a user