include resolved tenant fingerprint in result cache keys to prevent stale data after tenant changes (#7562)

Signed-off-by: SungJin1212 <tjdwls1201@gmail.com>

Author: SungJin1212

CHANGELOG.md

@@ -43,6 +43,7 @@
 * [BUGFIX] Security: Fix stored XSS vulnerability in Alertmanager and Store Gateway status pages by replacing `text/template` with `html/template`. #7512
 * [BUGFIX] Security: Limit decompressed gzip output in `ParseProtoReader` and OTLP ingestion path. The decompressed body is now capped by `-distributor.otlp-max-recv-msg-size`. #7515
 * [BUGFIX] Ingester: Close TSDB when compaction fails during `createTSDB`, preventing resource leaks (file descriptors, mmap handles) that could lead to ingester instability. #7560
+* [BUGFIX] Tenant Federation: Fix result cache returning stale data after a new tenant is added when `-tenant-federation.regex-matcher-enabled=true`. The resolved tenant set is now hashed and included in the cache key so that any change to the matched tenant list automatically invalidates cached entries. Non-regex users are unaffected. #7562
 * [BUGFIX] Tenant Federation: Fix regex resolver clearing known users list when user scan fails. #7534
 * [BUGFIX] Ingester: Release the TSDB appender on every early-return path in `Push` (e.g. out-of-order label set) by deferring `Rollback`. Previously such requests leaked TSDB head series references, mmap'd chunks and pending state per request, causing the `cortex_ingester_tsdb_head_active_appenders` gauge to grow unbounded. #7528
 * [BUGFIX] Ring: Fix ring token conflict resolution only applied to updated instance and make constantly token conflict check during instance observe period.

integration/querier_tenant_federation_test.go

@@ -3,7 +3,10 @@
 package integration
 
 import (
+	"context"
 	"fmt"
+	"math/rand"
+	"path/filepath"
 	"strings"
 	"testing"
 	"time"
@@ -13,13 +16,139 @@ import (
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/block"
+	"github.com/thanos-io/thanos/pkg/block/metadata"
 
 	"github.com/cortexproject/cortex/integration/e2e"
 	e2ecache "github.com/cortexproject/cortex/integration/e2e/cache"
 	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
 	"github.com/cortexproject/cortex/integration/e2ecortex"
+	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/util/log"
 )
 
+// extractTenantIDsFromMatrix returns the unique __tenant_id__ label values
+// present across all streams in a query range result matrix.
+func extractTenantIDsFromMatrix(m model.Matrix) []string {
+	seen := map[string]struct{}{}
+	for _, stream := range m {
+		if tid, ok := stream.Metric[model.LabelName("__tenant_id__")]; ok {
+			seen[string(tid)] = struct{}{}
+		}
+	}
+	result := make([]string, 0, len(seen))
+	for tid := range seen {
+		result = append(result, tid)
+	}
+	return result
+}
+
+// TestRegexResolver_ResultCacheStale verifies the behavior of
+// query result caching with dynamic tenant discovery.
+func TestRegexResolver_ResultCacheStale(t *testing.T) {
+	ctx := context.Background()
+	rnd := rand.New(rand.NewSource(time.Now().Unix()))
+
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	memcached := e2ecache.NewMemcached()
+	consul := e2edb.NewConsulWithName("consul")
+	require.NoError(t, s.StartAndWaitReady(consul, memcached))
+
+	flags := mergeFlags(BlocksStorageFlags(), map[string]string{
+		// Enable result cache with memcached.
+		"-querier.cache-results":             "true",
+		"-querier.split-queries-by-interval": "24h",
+		"-frontend.memcached.addresses":      "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
+
+		// Regex tenant federation with fast discovery.
+		"-tenant-federation.enabled":               "true",
+		"-tenant-federation.regex-matcher-enabled": "true",
+		"-tenant-federation.user-sync-interval":    "5s",
+
+		// Disable bucket index so the store-gateway picks up new blocks via
+		// direct bucket scan
+		"-blocks-storage.bucket-store.bucket-index.enabled": "false",
+		"-blocks-storage.bucket-store.sync-interval":        "2s",
+	})
+
+	minio := e2edb.NewMinio(9000, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, s.StartAndWaitReady(minio))
+
+	storage, err := e2ecortex.NewS3ClientForMinio(minio, flags["-blocks-storage.s3.bucket-name"])
+	require.NoError(t, err)
+
+	// entire range is cached on the first query
+	blockStart := time.Now().Add(-20 * time.Minute)
+	blockEnd := time.Now().Add(-10 * time.Minute)
+	seriesLabels := []labels.Labels{labels.FromStrings(labels.MetricName, "series_1")}
+	dir := t.TempDir()
+
+	// Upload blocks for user-0 and user-1.
+	for _, tenantID := range []string{"user-0", "user-1"} {
+		bkt := bucket.NewUserBucketClient(tenantID, storage.GetBucket(), nil)
+		id, err := e2e.CreateBlock(ctx, rnd, dir, seriesLabels, 10,
+			blockStart.UnixMilli(), blockEnd.UnixMilli(), 30_000, 1)
+		require.NoError(t, err)
+		require.NoError(t, block.Upload(ctx, log.Logger, bkt,
+			filepath.Join(dir, id.String()), metadata.NoneFunc))
+	}
+
+	// Start all services.
+	ingester := e2ecortex.NewIngester("ingester", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	distributor := e2ecortex.NewDistributor("distributor", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	storeGateway := e2ecortex.NewStoreGateway("store-gateway", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), flags, "")
+	queryFrontend := e2ecortex.NewQueryFrontend("query-frontend", flags, "")
+	require.NoError(t, s.StartAndWaitReady(ingester, distributor, storeGateway))
+	require.NoError(t, s.Start(queryFrontend))
+
+	querier := e2ecortex.NewQuerier("querier", e2ecortex.RingStoreConsul, consul.NetworkHTTPEndpoint(), mergeFlags(flags, map[string]string{
+		"-querier.frontend-address": queryFrontend.NetworkGRPCEndpoint(),
+	}), "")
+	require.NoError(t, s.StartAndWaitReady(querier))
+	require.NoError(t, s.WaitReady(queryFrontend))
+
+	// Wait for the store-gateway to load the 2 initial blocks.
+	require.NoError(t, storeGateway.WaitSumMetricsWithOptions(
+		e2e.Equals(2), []string{"cortex_blocks_meta_synced"}, e2e.WaitMissingMetrics))
+
+	// Wait for the regex resolver to discover user-0 and user-1.
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(2), "cortex_regex_resolver_discovered_users"))
+
+	c, err := e2ecortex.NewClient(distributor.HTTPEndpoint(), queryFrontend.HTTPEndpoint(), "", "", "user-.+")
+	require.NoError(t, err)
+
+	result1, err := c.QueryRange("series_1", blockStart, blockEnd, 30*time.Second)
+	require.NoError(t, err)
+	require.Equal(t, model.ValMatrix, result1.Type())
+	require.ElementsMatch(t, []string{"user-0", "user-1"},
+		extractTenantIDsFromMatrix(result1.(model.Matrix)),
+		"first query must return data for exactly the initial two tenants")
+
+	bkt2 := bucket.NewUserBucketClient("user-2", storage.GetBucket(), nil)
+	id2, err := e2e.CreateBlock(ctx, rnd, dir, seriesLabels, 10,
+		blockStart.UnixMilli(), blockEnd.UnixMilli(), 30_000, 1)
+	require.NoError(t, err)
+	require.NoError(t, block.Upload(ctx, log.Logger, bkt2,
+		filepath.Join(dir, id2.String()), metadata.NoneFunc))
+
+	// Wait for user-2's block to be picked up by the store-gateway.
+	require.NoError(t, storeGateway.WaitSumMetricsWithOptions(
+		e2e.Equals(3), []string{"cortex_blocks_meta_synced"}, e2e.WaitMissingMetrics))
+
+	// Wait for the regex resolver to discover user-2.
+	require.NoError(t, querier.WaitSumMetrics(e2e.Equals(3), "cortex_regex_resolver_discovered_users"))
+
+	result2, err := c.QueryRange("series_1", blockStart, blockEnd, 30*time.Second)
+	require.NoError(t, err)
+	require.Equal(t, model.ValMatrix, result2.Type())
+	require.ElementsMatch(t, []string{"user-0", "user-1", "user-2"},
+		extractTenantIDsFromMatrix(result2.(model.Matrix)),
+		"second query must include newly discovered user-2")
+}
+
 type querierTenantFederationConfig struct {
 	querySchedulerEnabled  bool
 	shuffleShardingEnabled bool

pkg/cortex/cortex.go

@@ -356,6 +356,9 @@ type Cortex struct {
 	StoreGateway     *storegateway.StoreGateway
 	MemberlistKV     *memberlist.KVInitService
 
+	// RegexResolver is set when tenant-federation.regex-matcher-enabled=true.
+	RegexResolver *tenantfederation.RegexResolver
+
 	// Queryables that the querier should use to query the long
 	// term storage. It depends on the storage engine used.
 	StoreQueryables []querier.QueryableWithFilter

pkg/cortex/modules.go

@@ -98,6 +98,7 @@ const (
 	Purger                   string = "purger"
 	QueryScheduler           string = "query-scheduler"
 	TenantFederation         string = "tenant-federation"
+	RegexResolverService     string = "regex-resolver"
 	ResourceMonitor          string = "resource-monitor"
 	All                      string = "all"
 )
@@ -300,6 +301,28 @@ func (t *Cortex) initQueryable() (serv services.Service, err error) {
 	return nil, nil
 }
 
+func (t *Cortex) initRegexResolverService() (serv services.Service, err error) {
+	if !t.Cfg.TenantFederation.Enabled || !t.Cfg.TenantFederation.RegexMatcherEnabled {
+		return nil, nil
+	}
+
+	util_log.WarnExperimentalUse("tenant-federation.regex-matcher-enabled")
+
+	reg := prometheus.DefaultRegisterer
+	bucketClientFactory := func(ctx context.Context) (objstore.InstrumentedBucket, error) {
+		return bucket.NewClient(ctx, t.Cfg.BlocksStorage.Bucket, nil, "regex-resolver", util_log.Logger, reg)
+	}
+
+	regexResolver, err := tenantfederation.NewRegexResolver(t.Cfg.BlocksStorage.UsersScanner, t.Cfg.TenantFederation, reg, bucketClientFactory, util_log.Logger)
+	if err != nil {
+		return nil, fmt.Errorf("failed to initialize regex resolver: %v", err)
+	}
+	users.WithDefaultResolver(regexResolver)
+	t.RegexResolver = regexResolver
+
+	return regexResolver, nil
+}
+
 // Enable merge querier if multi tenant query federation is enabled
 func (t *Cortex) initTenantFederation() (serv services.Service, err error) {
 	if t.Cfg.TenantFederation.Enabled {
@@ -312,24 +335,6 @@ func (t *Cortex) initTenantFederation() (serv services.Service, err error) {
 		t.QuerierQueryable = querier.NewSampleAndChunkQueryable(tenantfederation.NewQueryable(t.QuerierQueryable, t.Cfg.TenantFederation, byPassForSingleQuerier, reg))
 		t.MetadataQuerier = tenantfederation.NewMetadataQuerier(t.MetadataQuerier, t.Cfg.TenantFederation, reg)
 		t.ExemplarQueryable = tenantfederation.NewExemplarQueryable(t.ExemplarQueryable, t.Cfg.TenantFederation, byPassForSingleQuerier, reg)
-
-		if t.Cfg.TenantFederation.RegexMatcherEnabled {
-			util_log.WarnExperimentalUse("tenant-federation.regex-matcher-enabled")
-
-			bucketClientFactory := func(ctx context.Context) (objstore.InstrumentedBucket, error) {
-				return bucket.NewClient(ctx, t.Cfg.BlocksStorage.Bucket, nil, "regex-resolver", util_log.Logger, reg)
-			}
-
-			regexResolver, err := tenantfederation.NewRegexResolver(t.Cfg.BlocksStorage.UsersScanner, t.Cfg.TenantFederation, reg, bucketClientFactory, util_log.Logger)
-			if err != nil {
-				return nil, fmt.Errorf("failed to initialize regex resolver: %v", err)
-			}
-			users.WithDefaultResolver(regexResolver)
-
-			return regexResolver, nil
-		}
-
-		return nil, nil
 	}
 
 	return nil, nil
@@ -555,6 +560,14 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
 		users.WithDefaultResolver(tenantfederation.NewRegexValidator())
 	}
 
+	// Build a lazy resolver function for the result cache.
+	var tenantResolverFn func() users.Resolver
+	if t.Cfg.TenantFederation.Enabled && t.Cfg.TenantFederation.RegexMatcherEnabled {
+		tenantResolverFn = func() users.Resolver {
+			return t.RegexResolver
+		}
+	}
+
 	queryRangeMiddlewares, cache, err := queryrange.Middlewares(
 		t.Cfg.QueryRange,
 		util_log.Logger,
@@ -568,6 +581,7 @@ func (t *Cortex) initQueryFrontendTripperware() (serv services.Service, err erro
 		t.Cfg.Querier.DefaultEvaluationInterval,
 		t.Cfg.Querier.DistributedExecEnabled,
 		t.Cfg.Querier.ThanosEngine.LogicalOptimizers,
+		tenantResolverFn,
 	)
 	if err != nil {
 		return nil, err
@@ -945,6 +959,7 @@ func (t *Cortex) setupModuleManager() error {
 	mm.RegisterModule(Purger, nil)
 	mm.RegisterModule(QueryScheduler, t.initQueryScheduler)
 	mm.RegisterModule(TenantFederation, t.initTenantFederation, modules.UserInvisibleModule)
+	mm.RegisterModule(RegexResolverService, t.initRegexResolverService, modules.UserInvisibleModule)
 	mm.RegisterModule(All, nil)
 
 	// Add dependencies
@@ -964,7 +979,7 @@ func (t *Cortex) setupModuleManager() error {
 		Queryable:                {OverridesConfig, DistributorService, OverridesConfig, Ring, API, StoreQueryable, MemberlistKV, ResourceMonitor},
 		Querier:                  {TenantFederation},
 		StoreQueryable:           {OverridesConfig, OverridesConfig, MemberlistKV, GrpcClientService},
-		QueryFrontendTripperware: {API, OverridesConfig},
+		QueryFrontendTripperware: {API, OverridesConfig, RegexResolverService},
 		QueryFrontend:            {QueryFrontendTripperware},
 		QueryScheduler:           {API, OverridesConfig},
 		Ruler:                    {DistributorService, OverridesConfig, StoreQueryable, RulerStorage},
@@ -976,7 +991,8 @@ func (t *Cortex) setupModuleManager() error {
 		StoreGateway:             {API, OverridesConfig, MemberlistKV, ResourceMonitor},
 		TenantDeletion:           {API, OverridesConfig},
 		Purger:                   {TenantDeletion},
-		TenantFederation:         {Queryable},
+		TenantFederation:         {Queryable, RegexResolverService},
+		RegexResolverService:     {},
 		All:                      {QueryFrontend, Querier, Ingester, Distributor, Purger, StoreGateway, Ruler, Compactor, AlertManager},
 	}
 	if t.Cfg.ExternalPusher != nil && t.Cfg.ExternalQueryable != nil {

pkg/querier/tripperware/queryrange/query_range_middlewares.go

@@ -30,6 +30,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/querier"
 	"github.com/cortexproject/cortex/pkg/querier/tripperware"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
+	"github.com/cortexproject/cortex/pkg/util/users"
 )
 
 const day = 24 * time.Hour
@@ -107,6 +108,7 @@ func Middlewares(
 	defaultEvaluationInterval time.Duration,
 	distributedExecEnabled bool,
 	localOptimizers []logicalplan.Optimizer,
+	tenantResolverFn func() users.Resolver,
 ) ([]tripperware.Middleware, cache.Cache, error) {
 	// Metric used to keep track of each middleware execution duration.
 	metrics := tripperware.NewInstrumentMiddlewareMetrics(registerer)
@@ -132,7 +134,7 @@ func Middlewares(
 			return false
 		}
 
-		queryCacheMiddleware, cache, err := NewResultsCacheMiddleware(log, cfg.ResultsCacheConfig, splitter(cfg.SplitQueriesByInterval), limits, prometheusCodec, cacheExtractor, shouldCache, registerer)
+		queryCacheMiddleware, cache, err := NewResultsCacheMiddleware(log, cfg.ResultsCacheConfig, splitter(cfg.SplitQueriesByInterval), limits, prometheusCodec, cacheExtractor, shouldCache, registerer, tenantResolverFn)
 		if err != nil {
 			return nil, nil, err
 		}

pkg/querier/tripperware/queryrange/query_range_middlewares_test.go

@@ -70,6 +70,7 @@ func TestRoundTrip(t *testing.T) {
 		time.Minute,
 		false,
 		logicalplan.DefaultOptimizers,
+		nil,
 	)
 	require.NoError(t, err)
 
@@ -195,6 +196,7 @@ func TestRoundTripWithAndWithoutDistributedExec(t *testing.T) {
 				time.Minute,
 				tc.distributedEnabled,
 				logicalplan.DefaultOptimizers,
+				nil,
 			)
 			require.NoError(t, err)
 

pkg/querier/tripperware/queryrange/results_cache.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"hash/fnv"
 	"net/http"
 	"slices"
 	"sort"
@@ -179,6 +180,7 @@ type resultsCache struct {
 	merger                     tripperware.Merger
 	shouldCache                ShouldCacheFn
 	cacheQueryableSamplesStats bool
+	tenantResolverFn           func() users.Resolver
 }
 
 // NewResultsCacheMiddleware creates results cache middleware from config.
@@ -196,6 +198,7 @@ func NewResultsCacheMiddleware(
 	extractor Extractor,
 	shouldCache ShouldCacheFn,
 	reg prometheus.Registerer,
+	tenantResolverFn func() users.Resolver,
 ) (tripperware.Middleware, cache.Cache, error) {
 	c, err := cache.New(cfg.CacheConfig, reg, logger)
 	if err != nil {
@@ -219,6 +222,7 @@ func NewResultsCacheMiddleware(
 			now:                        time.Now,
 			shouldCache:                shouldCache,
 			cacheQueryableSamplesStats: cfg.CacheQueryableSamplesStats,
+			tenantResolverFn:           tenantResolverFn,
 		}
 	}), c, nil
 }
@@ -242,7 +246,21 @@ func (s resultsCache) Do(ctx context.Context, r tripperware.Request) (tripperwar
 		return s.next.Do(ctx, r)
 	}
 
-	key := s.splitter.GenerateCacheKey(ctx, users.JoinTenantIDs(tenantIDs), r)
+	// Build the user ID portion of the cache key.
+	// For regex-federation users the resolved tenant set is hashed and appended
+	// so that adding or removing a tenant invalidates cached entries automatically.
+	cacheUserID := users.JoinTenantIDs(tenantIDs)
+	if s.tenantResolverFn != nil {
+		if resolver := s.tenantResolverFn(); resolver != nil {
+			if resolvedIDs, resolveErr := resolver.TenantIDs(ctx); resolveErr == nil && len(resolvedIDs) > 0 {
+				h := fnv.New64a()
+				_, _ = h.Write([]byte(strings.Join(resolvedIDs, "|")))
+				cacheUserID = fmt.Sprintf("%s:h%x", cacheUserID, h.Sum64())
+			}
+		}
+	}
+
+	key := s.splitter.GenerateCacheKey(ctx, cacheUserID, r)
 
 	var (
 		extents  []tripperware.Extent