From fe2da16f3c54cf975a1352b27194009b57292143 Mon Sep 17 00:00:00 2001 From: Kyle Stang Date: Tue, 7 Jul 2026 22:37:10 +0000 Subject: [PATCH] perf(ingester): invalidate head postings cache per-label instead of per-metric The head expanded postings cache previously keyed a per-metric "seed" into the cache key, so creating or deleting any series for a metric invalidated every cached entry for that metric name, even entries whose result could not have changed. Replace the seed with a per-(tenant, label) write counter. On series create/delete each of the series' label counts is incremented. On creation, a cache entry snapshots the counts of its matched labels and stays valid as long as at least one tracked count is unchanged. Creating or deleting a series which matches all of the tracked labels expires the cache entries. Matchers that can match an absent label (e.g. foo!="x", foo=~".*") are excluded from the snapshot since a matching series need not carry the label, so they cannot vouch for an entry. Name-only queries fall back to the old whole-metric behaviour via the __name__ count. The counter array is shared across tenants and namespaced by userId; it is uint32 (8MB) and read under the existing stripe locks. Adds a reason="stale" value to the cache miss metric to distinguish count invalidation from TTL expiry. Signed-off-by: Kyle Stang --- CHANGELOG.md | 1 + pkg/ingester/ingester_test.go | 6 +- pkg/storage/tsdb/expanded_postings_cache.go | 232 ++++++++++----- .../tsdb/expanded_postings_cache_test.go | 271 ++++++++++++++++-- 4 files changed, 405 insertions(+), 105 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1ab3c71ffb..8f7d3a5acd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -36,6 +36,7 @@ * [ENHANCEMENT] Distributor: Added `cortex_distributor_received_histogram_buckets` metric to track number of buckets in received native histogram samples before validation, per user. #7569 * [ENHANCEMENT] Distributor: Add `WrappedHistogram` with configurable size limit (`-validation.max-native-histogram-size-bytes`) to cap native histogram protobuf size before unmarshalling. #7570 * [ENHANCEMENT] Ingester: Add lazy regex evaluation on head postings cache miss. Defers expensive regex matchers on high-cardinality labels to per-series filtering when a selective equality matcher already narrows the result set. Configured via `-blocks-storage.expanded_postings_cache.head.lazy-matcher-max-cardinality` (disabled by default). #7553 +* [ENHANCEMENT] Ingester: Improve head expanded postings cache hit rate by invalidating only the cache entries whose matched labels changed when a series is created or deleted, instead of invalidating all entries for the metric name. #7678 * [ENHANCEMENT] Query Frontend: Improve the slow query log with `source`, `user_agent`, `engine_type`, `block_store_type`, and query stats fields to aid slow query diagnosis. #7601 * [ENHANCEMENT] Ring: Add ring metric to count number of duplicate tokens. #7626 * [ENHANCEMENT] Ring: Cache `ShuffleShardWithLookback` subrings. The cached entry is invalidated on topology change or once `now` reaches the earliest `RegisteredTimestamp + lookbackPeriod` of any included instance. #7628 diff --git a/pkg/ingester/ingester_test.go b/pkg/ingester/ingester_test.go index fe396e6568..5e16aada6d 100644 --- a/pkg/ingester/ingester_test.go +++ b/pkg/ingester/ingester_test.go @@ -6424,7 +6424,7 @@ func TestExpendedPostingsCacheIsolation(t *testing.T) { cfg.BlocksStorageConfig.TSDB.BlockRanges = []time.Duration{2 * time.Hour} cfg.LifecyclerConfig.JoinAfter = 0 cfg.BlocksStorageConfig.TSDB.PostingsCache = cortex_tsdb.TSDBPostingsCacheConfig{ - SeedSize: 3, // lets make sure all metric names collide + LabelCounterSize: 3, // lets make sure all metric names collide Head: cortex_tsdb.PostingsCacheConfig{ Enabled: true, Ttl: time.Hour, @@ -6925,9 +6925,9 @@ func TestExpendedPostingsCache(t *testing.T) { // Query block and head require.Equal(t, postingsForMatchersCalls.Load(), int64(c.expectedBlockPostingCall+c.expectedHeadPostingCall)) - // Adding a metric for the first metric name so we expire all caches for that metric name + // Adding a series with the queried label "a" to expire the caches for that metric name and label _, err = i.Push(ctx, cortexpb.ToWriteRequest( - []labels.Labels{labels.FromStrings(labels.MetricName, metricNames[0], "extra", "1")}, []cortexpb.Sample{{Value: 2, TimestampMs: 4 * 60 * 60 * 1000}}, nil, nil, cortexpb.API)) + []labels.Labels{labels.FromStrings(labels.MetricName, metricNames[0], "a", "new", "extra", "1")}, []cortexpb.Sample{{Value: 2, TimestampMs: 4 * 60 * 60 * 1000}}, nil, nil, cortexpb.API)) require.NoError(t, err) for in, name := range metricNames { diff --git a/pkg/storage/tsdb/expanded_postings_cache.go b/pkg/storage/tsdb/expanded_postings_cache.go index 78558ab314..f10bef43d1 100644 --- a/pkg/storage/tsdb/expanded_postings_cache.go +++ b/pkg/storage/tsdb/expanded_postings_cache.go @@ -6,7 +6,6 @@ import ( "errors" "flag" "slices" - "strconv" "strings" "sync" "time" @@ -15,6 +14,7 @@ import ( "github.com/oklog/ulid/v2" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/common/model" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/tsdb" @@ -31,11 +31,10 @@ var ( ) const ( - // size of the seed array. Each seed is a 64bits int (8 bytes) - // totaling 16mb - seedArraySize = 2 * 1024 * 1024 + // size of the label-count array. Each count is a uint32 (4 bytes) totaling 8mb. + labelCounterArraySize = 2 * 1024 * 1024 - numOfSeedsStripes = 512 + numOfCounterStripes = 512 ) type ExpandedPostingsCacheMetrics struct { @@ -101,7 +100,7 @@ type TSDBPostingsCacheConfig struct { // The configurations below are used only for testing purpose PostingsForMatchers func(ctx context.Context, ix tsdb.IndexReader, ms ...*labels.Matcher) (index.Postings, error) `yaml:"-"` - SeedSize int `yaml:"-"` + LabelCounterSize int `yaml:"-"` timeNow func() time.Time `yaml:"-"` } @@ -129,19 +128,19 @@ func (cfg *PostingsCacheConfig) RegisterFlagsWithPrefix(prefix, block string, f } type ExpandedPostingsCacheFactory struct { - seedByHash *seedByHash - cfg TSDBPostingsCacheConfig + labelCounter *labelCounter + cfg TSDBPostingsCacheConfig } func NewExpandedPostingsCacheFactory(cfg TSDBPostingsCacheConfig) *ExpandedPostingsCacheFactory { if cfg.Head.Enabled || cfg.Blocks.Enabled { - if cfg.SeedSize == 0 { - cfg.SeedSize = seedArraySize + if cfg.LabelCounterSize == 0 { + cfg.LabelCounterSize = labelCounterArraySize } logutil.WarnExperimentalUse("expanded postings cache") return &ExpandedPostingsCacheFactory{ - cfg: cfg, - seedByHash: newSeedByHash(cfg.SeedSize), + cfg: cfg, + labelCounter: newLabelCounter(cfg.LabelCounterSize), } } @@ -149,7 +148,7 @@ func NewExpandedPostingsCacheFactory(cfg TSDBPostingsCacheConfig) *ExpandedPosti } func (f *ExpandedPostingsCacheFactory) NewExpandedPostingsCache(userId string, metrics *ExpandedPostingsCacheMetrics) ExpandedPostingsCache { - return newBlocksPostingsForMatchersCache(userId, f.cfg, metrics, f.seedByHash) + return newBlocksPostingsForMatchersCache(userId, f.cfg, metrics, f.labelCounter) } type ExpandedPostingsCache interface { @@ -171,8 +170,8 @@ type blocksPostingsForMatchersCache struct { lazyMatcherCfg lazyMatcherConfig labelCardinalityCache *expirable.LRU[string, int] - metrics *ExpandedPostingsCacheMetrics - seedByHash *seedByHash + metrics *ExpandedPostingsCacheMetrics + labelCounter *labelCounter } func (c *blocksPostingsForMatchersCache) Clear() { @@ -180,7 +179,7 @@ func (c *blocksPostingsForMatchersCache) Clear() { c.blocksCache.clear() } -func newBlocksPostingsForMatchersCache(userId string, cfg TSDBPostingsCacheConfig, metrics *ExpandedPostingsCacheMetrics, seedByHash *seedByHash) ExpandedPostingsCache { +func newBlocksPostingsForMatchersCache(userId string, cfg TSDBPostingsCacheConfig, metrics *ExpandedPostingsCacheMetrics, labelCounter *labelCounter) ExpandedPostingsCache { if cfg.PostingsForMatchers == nil { cfg.PostingsForMatchers = tsdb.PostingsForMatchers } @@ -201,7 +200,7 @@ func newBlocksPostingsForMatchersCache(userId string, cfg TSDBPostingsCacheConfi }, labelCardinalityCache: newLabelCardinalityCache(), metrics: metrics, - seedByHash: seedByHash, + labelCounter: labelCounter, userId: userId, } } @@ -211,7 +210,14 @@ func (c *blocksPostingsForMatchersCache) ExpireSeries(metric labels.Labels) { if err != nil { return } - c.seedByHash.incrementSeed(c.userId, metricName) + + metric.Range(func(label labels.Label) { + if label.Name == model.MetricNameLabel { + c.labelCounter.increment(c.userId, label.Value) + } else { + c.labelCounter.increment(c.userId, metricName, label.Name) + } + }) } func (c *blocksPostingsForMatchersCache) PurgeExpiredItems() { @@ -228,15 +234,17 @@ func (c *blocksPostingsForMatchersCache) PostingsForMatchers(ctx context.Context } func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsdb.IndexReader, ms ...*labels.Matcher) func(context.Context) (index.Postings, error) { - var seed string cache := c.blocksCache + isHead := isHeadBlock(blockID) + var metricName string - // If is a head block, lets add the seed on the cache key so we can - // invalidate the cache when new series are created for this metric name - if isHeadBlock(blockID) { + // If is a head block, we snapshot the per-label write counts so we can invalidate + // the cache when new series are created for this metric name. + if isHead { cache = c.headCache if cache.cfg.Enabled { - metricName, ok := metricNameFromMatcher(ms) + var ok bool + metricName, ok = metricNameFromMatcher(ms) // Lets not cache head if we don;t find an equal matcher for the label __name__ if !ok { c.metrics.NonCacheableQueries.WithLabelValues(cache.name).Inc() @@ -244,8 +252,6 @@ func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsd return tsdb.PostingsForMatchers(ctx, ix, ms...) } } - - seed = c.getSeedForMetricName(metricName) } } @@ -258,7 +264,7 @@ func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsd c.metrics.CacheRequests.WithLabelValues(cache.name).Inc() - fetch := func() ([]storage.SeriesRef, int64, error) { + fetch := func() ([]storage.SeriesRef, int64, func() bool, error) { // Use a context with timeout instead of context.Background() to prevent runaway queries. // This promise is maybe shared across calls, so we can't use any single caller's context. // However, we need a timeout to prevent the fetch from running indefinitely when all @@ -270,13 +276,23 @@ func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsd defer cancel() } + // Count-based invalidation only applies to the head cache: series are created + // there and bump the per-label counts. Block entries are immutable, so they + // get a nil validator and rely solely on TTL expiry. + var stillValid func() bool + var snapshotSize int64 + if isHead { + stillValid, snapshotSize = c.snapshotCounts(metricName, ms...) + } + // For head blocks, try to avoid expensive regex scans by splitting matchers: // resolve postings with selective matchers only, then filter by regex lazily. - if isHeadBlock(blockID) && c.lazyMatcherCfg.MaxCardinality > 0 { + if isHead && c.lazyMatcherCfg.MaxCardinality > 0 { selectMs, lazyMs := splitMatchersForHeadWithConfig(fetchCtx, ix, ms, c.lazyMatcherCfg, c.labelCardinalityCache) if len(lazyMs) > 0 { c.metrics.LazyMatcherQueries.Inc() - return c.fetchWithLazyMatchers(fetchCtx, ix, selectMs, lazyMs) + series, size, err := c.fetchWithLazyMatchers(fetchCtx, ix, selectMs, lazyMs) + return series, size + snapshotSize, stillValid, err } } @@ -284,13 +300,13 @@ func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsd if err == nil { ids, err := index.ExpandPostings(postings) - return ids, int64(len(ids) * 8), err + return ids, int64(len(ids)*8) + snapshotSize, stillValid, err } - return nil, 0, err + return nil, 0, stillValid, err } - key := cacheKey(seed, blockID, ms...) + key := cacheKey(blockID, ms...) promise, loaded := cache.getPromiseForKey(key, fetch) if loaded { c.metrics.CacheHits.WithLabelValues(cache.name).Inc() @@ -299,6 +315,54 @@ func (c *blocksPostingsForMatchersCache) fetchPostings(blockID ulid.ULID, ix tsd return c.result(promise) } +// snapshotCounts records the current per-label write counts for the matchers so the +// cached entry can later be validated. The slot for each matcher MUST be keyed +// identically to ExpireSeries' increment: __name__ by its value, every other label by +// (metricName, labelName). A mismatch would read a slot that never moves and keep stale +// entries alive forever. +// +// The returned stillValid closure reports whether the entry is still complete: it holds +// as long as ANY tracked count is unchanged, because a series matching all matchers bumps +// every tracked slot, so all-counts-changed is the only proof of a possibly-missed series. +// +// Empty-matching matchers (e.g. foo!="x", foo=~".*") match series that lack the label, and +// such a series bumps __name__ but not (metric, foo). Their count could stay frozen while a +// matching series was created, so they cannot vouch for the entry and are skipped. The +// __name__ equal matcher never matches "", so at least one snapshot always remains. +func (c *blocksPostingsForMatchersCache) snapshotCounts(metricName string, ms ...*labels.Matcher) (func() bool, int64) { + snapshots := make([]countSnapshot, 0, len(ms)) + for _, matcher := range ms { + // Matchers that can match an absent label cannot vouch for the entry. + if matcher.Matches("") { + continue + } + + var slot uint32 + if matcher.Name == model.MetricNameLabel { + slot = c.labelCounter.slotIndex(c.userId, matcher.Value) + } else { + slot = c.labelCounter.slotIndex(c.userId, metricName, matcher.Name) + } + snapshots = append(snapshots, countSnapshot{ + index: slot, + count: c.labelCounter.readSlot(slot), + }) + } + stillValid := func() bool { + for _, snapshot := range snapshots { + newCount := c.labelCounter.readSlot(snapshot.index) + if newCount == snapshot.count { + return true + } + } + + return false + } + + // countSnapshot is two uint32s (8 bytes total) + return stillValid, int64(len(snapshots) * 8) +} + func (c *blocksPostingsForMatchersCache) result(ce *cacheEntryPromise[[]storage.SeriesRef]) func(ctx context.Context) (index.Postings, error) { return func(ctx context.Context) (index.Postings, error) { select { @@ -373,11 +437,7 @@ func (c *blocksPostingsForMatchersCache) fetchWithLazyMatchers(ctx context.Conte return filtered, int64(len(filtered) * 8), nil } -func (c *blocksPostingsForMatchersCache) getSeedForMetricName(metricName string) string { - return c.seedByHash.getSeed(c.userId, metricName) -} - -func cacheKey(seed string, blockID ulid.ULID, ms ...*labels.Matcher) string { +func cacheKey(blockID ulid.ULID, ms ...*labels.Matcher) string { slices.SortFunc(ms, func(i, j *labels.Matcher) int { if i.Type != j.Type { return int(i.Type - j.Type) @@ -396,14 +456,12 @@ func cacheKey(seed string, blockID ulid.ULID, ms ...*labels.Matcher) string { sepLen = 1 ) - size := len(seed) + len(blockID.String()) + 2*sepLen + size := len(blockID.String()) + sepLen for _, m := range ms { size += len(m.Name) + len(m.Value) + typeLen + sepLen } sb := strings.Builder{} sb.Grow(size) - sb.WriteString(seed) - sb.WriteByte('|') sb.WriteString(blockID.String()) sb.WriteByte('|') for _, m := range ms { @@ -422,7 +480,7 @@ func isHeadBlock(blockID ulid.ULID) bool { func metricNameFromMatcher(ms []*labels.Matcher) (string, bool) { for _, m := range ms { - if m.Name == labels.MetricName && m.Type == labels.MatchEqual { + if m.Name == model.MetricNameLabel && m.Type == labels.MatchEqual { return m.Value, true } } @@ -430,34 +488,41 @@ func metricNameFromMatcher(ms []*labels.Matcher) (string, bool) { return "", false } -type seedByHash struct { - strippedLock []sync.RWMutex - seedByHash []int +// labelCounter tracks, per (tenant, label) hash slot, how many times a series carrying that +// label has been created or deleted. Cache entries snapshot these counts to detect when a +// matching series may have been added since the entry was stored. Collisions are safe: two +// distinct labels sharing a slot only cause extra (never missed) invalidations. +type labelCounter struct { + stripedLock []sync.RWMutex + counts []uint32 } -func newSeedByHash(size int) *seedByHash { - return &seedByHash{ - seedByHash: make([]int, size), - strippedLock: make([]sync.RWMutex, numOfSeedsStripes), +func newLabelCounter(size int) *labelCounter { + return &labelCounter{ + counts: make([]uint32, size), + stripedLock: make([]sync.RWMutex, numOfCounterStripes), } } -func (s *seedByHash) getSeed(userId string, v string) string { - h := memHashString(userId, v) - i := h % uint64(len(s.seedByHash)) - l := i % uint64(len(s.strippedLock)) - s.strippedLock[l].RLock() - defer s.strippedLock[l].RUnlock() - return strconv.Itoa(s.seedByHash[i]) +func (s *labelCounter) slotIndex(userId string, dimensions ...string) uint32 { + hash := memHashStrings(userId, dimensions...) + return hash % uint32(len(s.counts)) +} + +func (s *labelCounter) readSlot(index uint32) uint32 { + lock := index % uint32(len(s.stripedLock)) + s.stripedLock[lock].RLock() + defer s.stripedLock[lock].RUnlock() + return s.counts[index] } -func (s *seedByHash) incrementSeed(userId string, v string) { - h := memHashString(userId, v) - i := h % uint64(len(s.seedByHash)) - l := i % uint64(len(s.strippedLock)) - s.strippedLock[l].Lock() - defer s.strippedLock[l].Unlock() - s.seedByHash[i]++ +func (s *labelCounter) increment(userId string, dimensions ...string) { + h := memHashStrings(userId, dimensions...) + i := h % uint32(len(s.counts)) + l := i % uint32(len(s.stripedLock)) + s.stripedLock[l].Lock() + defer s.stripedLock[l].Unlock() + s.counts[i]++ } type lruCache[V any] struct { @@ -516,14 +581,14 @@ func (c *lruCache[V]) size() int { return c.cached.Len() } -func (c *lruCache[V]) getPromiseForKey(k string, fetch func() (V, int64, error)) (*cacheEntryPromise[V], bool) { +func (c *lruCache[V]) getPromiseForKey(k string, fetch func() (V, int64, func() bool, error)) (*cacheEntryPromise[V], bool) { r := &cacheEntryPromise[V]{ done: make(chan struct{}), } defer close(r.done) if !c.cfg.Enabled { - r.v, _, r.err = fetch() + r.v, _, _, r.err = fetch() return r, false } @@ -531,7 +596,7 @@ func (c *lruCache[V]) getPromiseForKey(k string, fetch func() (V, int64, error)) if !ok { c.metrics.CacheMiss.WithLabelValues(c.name, "miss").Inc() - r.v, r.sizeBytes, r.err = fetch() + r.v, r.sizeBytes, r.stillValid, r.err = fetch() r.sizeBytes += int64(len(k)) r.ts = c.timeNow() c.created(k, r.sizeBytes) @@ -539,20 +604,29 @@ func (c *lruCache[V]) getPromiseForKey(k string, fetch func() (V, int64, error)) } if ok { + entry := loaded.(*cacheEntryPromise[V]) // If the promise is already in the cache, lets wait it to fetch the data. - <-loaded.(*cacheEntryPromise[V]).done + <-entry.done // LRU: move to back on access c.cachedMtx.Lock() - if elem := loaded.(*cacheEntryPromise[V]).elem; elem != nil { - c.cached.MoveToBack(elem) + if entry.elem != nil { + c.cached.MoveToBack(entry.elem) } c.cachedMtx.Unlock() - // If is cached but is expired, lets try to replace the cache value. - if loaded.(*cacheEntryPromise[V]).isExpired(c.cfg.Ttl, c.timeNow()) && c.cachedValues.CompareAndSwap(k, loaded, r) { - c.metrics.CacheMiss.WithLabelValues(c.name, "expired").Inc() - r.v, r.sizeBytes, r.err = fetch() + // A nil stillValid means the entry has no count-based invalidation (blocks + // cache, or callers that don't snapshot counts); only time expiry applies. + stale := entry.stillValid != nil && !entry.stillValid() + + // If is cached but is expired or invalidated, lets try to replace the cache value. + if (stale || entry.isExpired(c.cfg.Ttl, c.timeNow())) && c.cachedValues.CompareAndSwap(k, loaded, r) { + if stale { + c.metrics.CacheMiss.WithLabelValues(c.name, "stale").Inc() + } else { + c.metrics.CacheMiss.WithLabelValues(c.name, "expired").Inc() + } + r.v, r.sizeBytes, r.stillValid, r.err = fetch() r.sizeBytes += int64(len(k)) c.updateSize(loaded.(*cacheEntryPromise[V]).sizeBytes, r.sizeBytes) r.ts = c.timeNow() @@ -636,6 +710,8 @@ type cacheEntryPromise[V any] struct { sizeBytes int64 elem *list.Element + stillValid func() bool + done chan struct{} v V err error @@ -647,7 +723,15 @@ func (ce *cacheEntryPromise[V]) isExpired(ttl time.Duration, now time.Time) bool return r >= ttl } -func memHashString(userId, v string) uint64 { - h := fnv1a.HashString64(userId) - return fnv1a.AddString64(h, v) +func memHashStrings(userId string, dimensions ...string) uint32 { + h := fnv1a.HashString32(userId) + for _, d := range dimensions { + h = fnv1a.AddString32(h, d) + } + return h +} + +type countSnapshot struct { + index uint32 // slot in labelCounter + count uint32 // value at store time } diff --git a/pkg/storage/tsdb/expanded_postings_cache_test.go b/pkg/storage/tsdb/expanded_postings_cache_test.go index 4f4f0741a0..73d9ec1acc 100644 --- a/pkg/storage/tsdb/expanded_postings_cache_test.go +++ b/pkg/storage/tsdb/expanded_postings_cache_test.go @@ -13,6 +13,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/tsdb" "github.com/prometheus/prometheus/tsdb/index" "github.com/stretchr/testify/require" @@ -21,7 +22,6 @@ import ( func TestCacheKey(t *testing.T) { blockID := ulid.MustNew(1, nil) - seed := "seed123" matchers := []*labels.Matcher{ { Type: labels.MatchEqual, @@ -44,8 +44,8 @@ func TestCacheKey(t *testing.T) { Value: "value_4", }, } - r := cacheKey(seed, blockID, matchers...) - require.Equal(t, "seed123|00000000010000000000000000|name_1=value_1|name_2!=value_2|name_3=~value_4|name_5!~value_4|", r) + r := cacheKey(blockID, matchers...) + require.Equal(t, "00000000010000000000000000|name_1=value_1|name_2!=value_2|name_3=~value_4|name_5!~value_4|", r) } func Test_ShouldFetchPromiseOnlyOnce(t *testing.T) { @@ -61,10 +61,10 @@ func Test_ShouldFetchPromiseOnlyOnce(t *testing.T) { wg := sync.WaitGroup{} wg.Add(concurrency) - fetchFunc := func() (int, int64, error) { + fetchFunc := func() (int, int64, func() bool, error) { calls.Inc() time.Sleep(100 * time.Millisecond) - return 0, 0, nil + return 0, 0, nil, nil } for range 100 { @@ -85,8 +85,8 @@ func TestFifoCacheDisabled(t *testing.T) { m := NewPostingCacheMetrics(prometheus.NewPedanticRegistry()) timeNow := time.Now cache := newLruCache[int](cfg, "test", m, timeNow) - old, loaded := cache.getPromiseForKey("key1", func() (int, int64, error) { - return 1, 0, nil + old, loaded := cache.getPromiseForKey("key1", func() (int, int64, func() bool, error) { + return 1, 0, nil, nil }) require.False(t, loaded) require.Equal(t, 1, old.v) @@ -131,14 +131,14 @@ func TestFifoCacheExpire(t *testing.T) { for i := range numberOfKeys { key := RepeatStringIfNeeded(fmt.Sprintf("key%d", i), keySize) - p, loaded := cache.getPromiseForKey(key, func() (int, int64, error) { - return 1, 8, nil + p, loaded := cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { + return 1, 8, nil, nil }) require.False(t, loaded) require.Equal(t, 1, p.v) require.True(t, cache.contains(key)) - p, loaded = cache.getPromiseForKey(key, func() (int, int64, error) { - return 1, 0, nil + p, loaded = cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { + return 1, 0, nil, nil }) require.True(t, loaded) require.Equal(t, 1, p.v) @@ -173,8 +173,8 @@ func TestFifoCacheExpire(t *testing.T) { for i := range numberOfKeys { key := RepeatStringIfNeeded(fmt.Sprintf("key%d", i), keySize) originalSize := cache.cachedBytes - p, loaded := cache.getPromiseForKey(key, func() (int, int64, error) { - return 2, 18, nil + p, loaded := cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { + return 2, 18, nil, nil }) require.False(t, loaded) // New value @@ -195,8 +195,8 @@ func TestFifoCacheExpire(t *testing.T) { return timeNow().Add(5 * c.cfg.Ttl) } - cache.getPromiseForKey("newKwy", func() (int, int64, error) { - return 2, 18, nil + cache.getPromiseForKey("newKwy", func() (int, int64, func() bool, error) { + return 2, 18, nil, nil }) // Should expire all keys expired keys @@ -214,14 +214,14 @@ func TestFifoCacheExpire(t *testing.T) { func Test_memHashString(test *testing.T) { numberOfTenants := 200 numberOfMetrics := 100 - occurrences := map[uint64]int{} + occurrences := map[uint32]int{} for range 10 { for j := range numberOfMetrics { metricName := fmt.Sprintf("metricName%v", j) for i := range numberOfTenants { userId := fmt.Sprintf("user%v", i) - occurrences[memHashString(userId, metricName)]++ + occurrences[memHashStrings(userId, metricName)]++ } } @@ -267,7 +267,7 @@ func TestPostingsCacheFetchTimeout(t *testing.T) { } m := NewPostingCacheMetrics(prometheus.NewPedanticRegistry()) - cache := newBlocksPostingsForMatchersCache("user1", cfg, m, newSeedByHash(seedArraySize)) + cache := newBlocksPostingsForMatchersCache("user1", cfg, m, newLabelCounter(labelCounterArraySize)) // Start a query that will trigger the fetch blockID := headULID @@ -306,19 +306,19 @@ func TestLruCacheEvictsLeastRecentlyUsed(t *testing.T) { cache := newLruCache[int](cfg, "test", m, time.Now) // Insert 3 entries: A, B, C - cache.getPromiseForKey("aaaa", func() (int, int64, error) { return 1, 8, nil }) - cache.getPromiseForKey("bbbb", func() (int, int64, error) { return 2, 8, nil }) - cache.getPromiseForKey("cccc", func() (int, int64, error) { return 3, 8, nil }) + cache.getPromiseForKey("aaaa", func() (int, int64, func() bool, error) { return 1, 8, nil, nil }) + cache.getPromiseForKey("bbbb", func() (int, int64, func() bool, error) { return 2, 8, nil, nil }) + cache.getPromiseForKey("cccc", func() (int, int64, func() bool, error) { return 3, 8, nil, nil }) require.True(t, cache.contains("aaaa")) require.True(t, cache.contains("bbbb")) require.True(t, cache.contains("cccc")) // Access A to make it recently used (B is now least recently used) - cache.getPromiseForKey("aaaa", func() (int, int64, error) { return 1, 8, nil }) + cache.getPromiseForKey("aaaa", func() (int, int64, func() bool, error) { return 1, 8, nil, nil }) // Insert D — should evict B (least recently used), not A - cache.getPromiseForKey("dddd", func() (int, int64, error) { return 4, 8, nil }) + cache.getPromiseForKey("dddd", func() (int, int64, func() bool, error) { return 4, 8, nil, nil }) require.True(t, cache.contains("aaaa"), "A should still be cached (recently accessed)") require.False(t, cache.contains("bbbb"), "B should be evicted (least recently used)") @@ -346,7 +346,7 @@ func BenchmarkLruCacheHitUnderPressure(b *testing.B) { hotKeys := make([]string, 50) for i := range hotKeys { hotKeys[i] = RepeatStringIfNeeded(fmt.Sprintf("hot-%d", i), keySize) - cache.getPromiseForKey(hotKeys[i], func() (int, int64, error) { return i, 8, nil }) + cache.getPromiseForKey(hotKeys[i], func() (int, int64, func() bool, error) { return i, 8, nil, nil }) } coldIdx := 0 @@ -356,12 +356,12 @@ func BenchmarkLruCacheHitUnderPressure(b *testing.B) { if i%3 == 0 { // 1/3 of accesses are hot queries (simulating ruler every 30s) key := hotKeys[i%len(hotKeys)] - cache.getPromiseForKey(key, func() (int, int64, error) { return 1, 8, nil }) + cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { return 1, 8, nil, nil }) } else { // 2/3 are cold unique queries (ad-hoc from Grafana) key := RepeatStringIfNeeded(fmt.Sprintf("cold-%d", coldIdx), keySize) coldIdx++ - cache.getPromiseForKey(key, func() (int, int64, error) { return 1, 8, nil }) + cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { return 1, 8, nil, nil }) } } @@ -384,15 +384,230 @@ func BenchmarkCacheGetPromise(b *testing.B) { // Pre-populate 1000 keys for i := range 1000 { key := fmt.Sprintf("key-%04d", i) - cache.getPromiseForKey(key, func() (int, int64, error) { return i, 100, nil }) + cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { return i, 100, nil, nil }) } b.RunParallel(func(pb *testing.PB) { i := 0 for pb.Next() { key := fmt.Sprintf("key-%04d", i%1000) - cache.getPromiseForKey(key, func() (int, int64, error) { return 1, 100, nil }) + cache.getPromiseForKey(key, func() (int, int64, func() bool, error) { return 1, 100, nil, nil }) i++ } }) } + +// countingPostingsCache builds a head-enabled cache whose PostingsForMatchers records how +// many times it was actually invoked, so tests can distinguish a cache hit (fetch NOT +// re-run) from an invalidation (fetch re-run). The returned fetches pointer is bumped once +// per underlying postings lookup. +func countingPostingsCache(t *testing.T, labelCounter *labelCounter) (ExpandedPostingsCache, *atomic.Int64) { + t.Helper() + fetches := atomic.NewInt64(0) + cfg := TSDBPostingsCacheConfig{ + Head: PostingsCacheConfig{ + Enabled: true, + Ttl: time.Hour, // long TTL so only count invalidation can force a refetch + MaxBytes: 10 << 20, + }, + PostingsForMatchers: func(ctx context.Context, ix tsdb.IndexReader, ms ...*labels.Matcher) (index.Postings, error) { + fetches.Inc() + return index.NewListPostings([]storage.SeriesRef{1, 2, 3}), nil + }, + } + m := NewPostingCacheMetrics(prometheus.NewPedanticRegistry()) + cache := newBlocksPostingsForMatchersCache("user1", cfg, m, labelCounter) + return cache, fetches +} + +// query runs a head-block PostingsForMatchers and drains the result so any lazy error surfaces. +func query(t *testing.T, cache ExpandedPostingsCache, ms ...*labels.Matcher) { + t.Helper() + p, err := cache.PostingsForMatchers(context.Background(), headULID, nil, ms...) + require.NoError(t, err) + // Drain so the ListPostings is fully consumed, matching real caller behaviour. + for p.Next() { + } + require.NoError(t, p.Err()) +} + +func TestExpandedPostingsCache_InvalidatesOnlyWhenAllTrackedCountsChange(t *testing.T) { + labelCounter := newLabelCounter(labelCounterArraySize) + cache, fetches := countingPostingsCache(t, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + jobMatcher := labels.MustNewMatcher(labels.MatchEqual, "job", "api") + + // First query populates the cache. + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load()) + + // Second identical query is a pure cache hit: no new series created. + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load(), "identical query should hit the cache") + + // A new series is created that shares the metric name but a DIFFERENT job value. + // This bumps the __name__ count and the (metric, job) count. Because every tracked + // count changed, the entry is invalidated and the next query refetches. + cache.ExpireSeries(labels.FromStrings("__name__", "http_requests", "job", "web")) + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(2), fetches.Load(), "series touching all tracked labels should invalidate") +} + +func TestExpandedPostingsCache_StableLabelKeepsEntryValid(t *testing.T) { + labelCounter := newLabelCounter(labelCounterArraySize) + cache, fetches := countingPostingsCache(t, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + jobMatcher := labels.MustNewMatcher(labels.MatchEqual, "job", "api") + + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load()) + + // A new series is created for the SAME metric but with a label the query does not + // track (region). This bumps __name__ and (metric, region), but NOT (metric, job). + // Since the tracked (metric, job) count is unchanged, the entry stays valid: any + // series matching {__name__="http_requests", job="api"} would have bumped (metric, job). + cache.ExpireSeries(labels.FromStrings("__name__", "http_requests", "region", "us-east")) + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load(), "unchanged tracked label should keep the entry valid") +} + +func TestExpandedPostingsCache_EmptyMatchingMatcherDoesNotVouch(t *testing.T) { + // A negative matcher like job!="batch" matches series that LACK a job label. Such a + // series bumps __name__ but not (metric, job), so the (metric, job) count must NOT be + // allowed to vouch for the entry. With only __name__ eligible to invalidate, any series + // creation for the metric must force a refetch. + labelCounter := newLabelCounter(labelCounterArraySize) + cache, fetches := countingPostingsCache(t, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + jobMatcher := labels.MustNewMatcher(labels.MatchNotEqual, "job", "batch") + + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load()) + + // New series for the metric WITHOUT a job label — matches job!="batch". It bumps only + // __name__ (and its other labels), not (metric, job). If the empty-matching job matcher + // were wrongly tracked, its frozen count would keep this stale entry alive. + cache.ExpireSeries(labels.FromStrings("__name__", "http_requests", "region", "us-east")) + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(2), fetches.Load(), "empty-matching matcher must not keep a stale entry alive") +} + +func TestExpandedPostingsCache_NameOnlyQueryInvalidatesOnAnySeriesChurn(t *testing.T) { + // A {__name__="foo"}-only query has no selective label to track, so it degrades to the + // old whole-metric behaviour: any series create/delete for the metric invalidates it. + labelCounter := newLabelCounter(labelCounterArraySize) + cache, fetches := countingPostingsCache(t, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + + query(t, cache, nameMatcher) + require.Equal(t, int64(1), fetches.Load()) + + query(t, cache, nameMatcher) + require.Equal(t, int64(1), fetches.Load(), "identical name-only query should hit the cache") + + cache.ExpireSeries(labels.FromStrings("__name__", "http_requests", "job", "api")) + query(t, cache, nameMatcher) + require.Equal(t, int64(2), fetches.Load(), "name-only query must invalidate on any series churn") +} + +func TestExpandedPostingsCache_DifferentMetricDoesNotInvalidate(t *testing.T) { + // Series churn on an unrelated metric must not invalidate this metric's entry. + labelCounter := newLabelCounter(labelCounterArraySize) + cache, fetches := countingPostingsCache(t, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + jobMatcher := labels.MustNewMatcher(labels.MatchEqual, "job", "api") + + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load()) + + cache.ExpireSeries(labels.FromStrings("__name__", "other_metric", "job", "api")) + query(t, cache, nameMatcher, jobMatcher) + require.Equal(t, int64(1), fetches.Load(), "churn on a different metric must not invalidate") +} + +// TestExpandedPostingsCache_SnapshotTakenBeforePostingsRead pins the ordering guarantee: +// the count snapshot MUST be captured before postings are read. If a series is created +// after the snapshot but before/around the postings read, the stored postings may miss it, +// so the stored count must be the pre-creation value and a later read must invalidate. +// +// The mock simulates that concurrent creation by bumping the metric's count from inside +// PostingsForMatchers (which runs after snapshotCounts in the correct implementation). With +// correct ordering the snapshot predates the bump, so the follow-up query refetches. If the +// snapshot were taken after the postings read, it would capture the post-bump count and the +// follow-up query would wrongly hit — failing this test. +func TestExpandedPostingsCache_SnapshotTakenBeforePostingsRead(t *testing.T) { + labelCounter := newLabelCounter(labelCounterArraySize) + fetches := atomic.NewInt64(0) + bumped := atomic.NewBool(false) + + cfg := TSDBPostingsCacheConfig{ + Head: PostingsCacheConfig{ + Enabled: true, + Ttl: time.Hour, + MaxBytes: 10 << 20, + }, + PostingsForMatchers: func(ctx context.Context, ix tsdb.IndexReader, ms ...*labels.Matcher) (index.Postings, error) { + fetches.Inc() + // Simulate a series for this metric being created during the fetch, once. The + // keying matches ExpireSeries' __name__ increment: (userId, metricValue). + if bumped.CompareAndSwap(false, true) { + labelCounter.increment("user1", "http_requests") + } + return index.NewListPostings([]storage.SeriesRef{1, 2, 3}), nil + }, + } + m := NewPostingCacheMetrics(prometheus.NewPedanticRegistry()) + cache := newBlocksPostingsForMatchersCache("user1", cfg, m, labelCounter) + + nameMatcher := labels.MustNewMatcher(labels.MatchEqual, "__name__", "http_requests") + + // Q1 populates the cache; the mock bumps the count after the snapshot was taken. + query(t, cache, nameMatcher) + require.Equal(t, int64(1), fetches.Load()) + + // Q2 sees the bumped count vs the pre-bump snapshot and must refetch. + query(t, cache, nameMatcher) + require.Equal(t, int64(2), fetches.Load(), "snapshot must predate the postings read; a series created during the fetch must invalidate") + + // Q3: the one-shot bump is done, so the Q2 snapshot now matches the live count → hit. + query(t, cache, nameMatcher) + require.Equal(t, int64(2), fetches.Load(), "with no further churn the entry should be stable again") +} + +func TestSnapshotCounts_Size(t *testing.T) { + // snapshotCounts reports the byte cost that gets added to the cache entry's sizeBytes. + // Each tracked matcher is one countSnapshot (two uint32s = 8 bytes); empty-matching + // matchers are skipped and must contribute nothing. + cfg := TSDBPostingsCacheConfig{ + Head: PostingsCacheConfig{Enabled: true, Ttl: time.Hour, MaxBytes: 1 << 20}, + } + m := NewPostingCacheMetrics(prometheus.NewPedanticRegistry()) + c := newBlocksPostingsForMatchersCache("user1", cfg, m, newLabelCounter(labelCounterArraySize)).(*blocksPostingsForMatchersCache) + + name := labels.MustNewMatcher(labels.MatchEqual, "__name__", "m") + job := labels.MustNewMatcher(labels.MatchEqual, "job", "api") + // Empty-matching (matches a series lacking "job"), so it is skipped and adds no bytes. + neg := labels.MustNewMatcher(labels.MatchNotEqual, "job", "batch") + require.True(t, neg.Matches(""), "sanity: neg must be empty-matching for this test to mean anything") + + tc := map[string]struct { + ms []*labels.Matcher + expectedSize int64 + }{ + "name only": {ms: []*labels.Matcher{name}, expectedSize: 8}, + "name + selective label": {ms: []*labels.Matcher{name, job}, expectedSize: 16}, + "empty-matcher contributes nothing": {ms: []*labels.Matcher{name, job, neg}, expectedSize: 16}, + } + + for title, tt := range tc { + t.Run(title, func(t *testing.T) { + _, size := c.snapshotCounts("m", tt.ms...) + require.Equal(t, tt.expectedSize, size) + }) + } +}