diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cc35782015..41e5a56b36d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,7 @@ * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446 * [CHANGE] HA Tracker: Move `-distributor.ha-tracker.failover-timeout` from a global config to a per-tenant runtime config. The flag name and default value (30s) remain the same. #7481 +* [CHANGE] Querier: When the bucket index is disabled, the bucket-scan blocks finder now keeps its on-disk block-meta cache under the reserved `/__querier__//` directory instead of sharing `//meta-syncer/` with a co-located store-gateway, so the two components no longer delete each other's caches in single-binary deployments. On upgrade the querier re-populates its meta cache once; old `//meta-syncer` directories are left in place (in single-binary mode they are the store-gateway's live cache). `__querier__` is no longer usable as a tenant ID. #7748 * [FEATURE] Parquet: Support sharded parquet file conversion and querying. #7610 * [FEATURE] Parquet Converter: Add experimental `-parquet-converter.max-num-columns` flag to automatically shard parquet files when the number of columns exceeds the configured limit. This prevents failures when a TSDB block has more unique label names than the parquet library's column limit (32767). #7624 * [FEATURE] Distributor: Add experimental `-distributor.num-query-workers` flag to use a goroutine worker pool for query fan-out calls to ingesters. Reuses pre-grown goroutine stacks to eliminate the `runtime.copystack` overhead (~8% CPU) observed on rulers with wide ingester fan-out. Falls back to spawning a new goroutine when no worker is available. #7623 diff --git a/pkg/querier/blocks_finder_bucket_scan.go b/pkg/querier/blocks_finder_bucket_scan.go index aae57708108..fec78eb65b5 100644 --- a/pkg/querier/blocks_finder_bucket_scan.go +++ b/pkg/querier/blocks_finder_bucket_scan.go @@ -41,9 +41,12 @@ var ( ) type BucketScanBlocksFinderConfig struct { - ScanInterval time.Duration - TenantsConcurrency int - MetasConcurrency int + ScanInterval time.Duration + TenantsConcurrency int + MetasConcurrency int + // CacheDir is the bucket store sync directory, which the finder may share with a co-located + // store-gateway. The finder doesn't cache anything directly in it: it keeps its per-tenant + // metadata caches in its own sub directory (see metaCacheDirForUser). CacheDir string ConsistencyDelay time.Duration IgnoreDeletionMarksDelay time.Duration @@ -420,7 +423,7 @@ func (d *BucketScanBlocksFinder) createMetaFetcher(userID string) (block.Metadat userBucket, blockLister, // The fetcher stores cached metas in the "meta-syncer/" sub directory. - filepath.Join(d.cfg.CacheDir, userID), + d.metaCacheDirForUser(userID), userReg, filters, ) @@ -432,15 +435,23 @@ func (d *BucketScanBlocksFinder) createMetaFetcher(userID string) (block.Metadat return f, userBucket, deletionMarkFilter, nil } -// metaSyncerCacheDirName is the sub-directory, under the per-tenant cache directory, where -// block.NewMetaFetcher stores its cached meta.json files (see createMetaFetcher). -const metaSyncerCacheDirName = "meta-syncer" +// metaCacheDirForUser returns the directory where the metadata fetcher of the given tenant caches +// its meta.json files (the fetcher appends its own "meta-syncer" sub directory to it). +// +// The finder keeps its caches under its own root inside the bucket store sync directory, instead of +// in /, because a co-located store-gateway (single-binary mode with the bucket +// index disabled) uses / for its own fetcher cache and block data. Both +// components reconcile their on-disk state against different sets of tenants, so sharing a +// directory means each of them deletes cache entries the other still considers live. +func (d *BucketScanBlocksFinder) metaCacheDirForUser(userID string) string { + return filepath.Join(d.cfg.CacheDir, users.QuerierMetaCacheDirName, userID) +} // evictInactiveUserFetchers reconciles the per-tenant metadata fetchers against the set of // currently active tenants. For every tenant that is no longer active it removes the cached -// fetcher, unregisters its per-tenant Prometheus registry, and deletes the fetcher's on-disk meta -// cache. Without this, d.fetchers, d.fetchersMetrics and the on-disk cache would grow unbounded -// for the lifetime of the process as tenants are deleted from storage. +// fetcher, unregisters its per-tenant Prometheus registry and deletes its on-disk metadata cache. +// Without this, d.fetchers, d.fetchersMetrics and the on-disk caches would grow unbounded for the +// lifetime of the process as tenants are deleted from storage. func (d *BucketScanBlocksFinder) evictInactiveUserFetchers(activeUserIDs []string) { active := make(map[string]struct{}, len(activeUserIDs)) for _, userID := range activeUserIDs { @@ -461,29 +472,48 @@ func (d *BucketScanBlocksFinder) evictInactiveUserFetchers(activeUserIDs []strin } d.fetchersMx.Unlock() - if len(evicted) == 0 { + if len(evicted) > 0 { + level.Info(d.logger).Log("msg", "evicted metadata fetchers for inactive tenants", "count", len(evicted)) + } + + // Reconcile the on-disk caches too, outside the lock to keep disk I/O off it. + d.deleteInactiveUserMetaCacheDirs(active) +} + +// deleteInactiveUserMetaCacheDirs deletes the on-disk metadata caches of the tenants that are no +// longer active. It sweeps the finder's own cache root rather than keying off the fetchers evicted +// by the current process, so that directories left behind by a previous one (e.g. tenants deleted +// while this process was down) are reclaimed as well; it therefore runs on every scan, not only +// when something was evicted — the cost is a single ReadDir of the root, negligible next to the +// bucket scan that precedes it. That is safe because the root is exclusive to the finder: a +// store-gateway co-located in the same process keeps its own metadata caches and its block data in +// //, which we never touch. +func (d *BucketScanBlocksFinder) deleteInactiveUserMetaCacheDirs(activeUserIDs map[string]struct{}) { + root := filepath.Join(d.cfg.CacheDir, users.QuerierMetaCacheDirName) + + entries, err := os.ReadDir(root) + if err != nil { + // The root is created lazily, together with the first tenant's metadata fetcher. + if !os.IsNotExist(err) { + level.Warn(d.logger).Log("msg", "failed to list the cached metadata fetcher directory", "dir", root, "err", err) + } return } - level.Info(d.logger).Log("msg", "evicted metadata fetchers for inactive tenants", "count", len(evicted)) + for _, entry := range entries { + if !entry.IsDir() { + continue + } - // Delete each evicted fetcher's on-disk meta cache, outside the lock to keep disk I/O off it. - // We remove only the fetcher's own "meta-syncer" sub-directory, not the whole CacheDir/ - // tree: in single-binary mode CacheDir is the store-gateway's SyncDir, whose block data also - // lives under CacheDir// and must not be deleted here. We key this off the fetchers - // this process evicted rather than sweeping CacheDir, so we never reach into a co-located - // store-gateway's cache; stale directories left by a previous process are reaped by the - // store-gateway's own cleanup (single-binary) and are otherwise a negligible disk residual. - for _, userID := range evicted { - metaCacheDir := filepath.Join(d.cfg.CacheDir, userID, metaSyncerCacheDirName) - if err := os.RemoveAll(metaCacheDir); err != nil { - level.Warn(d.logger).Log("msg", "failed to delete cached metadata fetcher directory for inactive user", "user", userID, "dir", metaCacheDir, "err", err) + userID := entry.Name() + if _, ok := activeUserIDs[userID]; ok { continue } - // Best-effort removal of the now-empty per-tenant directory. os.Remove only succeeds on an - // empty directory, so a co-located store-gateway's data under the same path is preserved. - _ = os.Remove(filepath.Join(d.cfg.CacheDir, userID)) + metaCacheDir := d.metaCacheDirForUser(userID) + if err := os.RemoveAll(metaCacheDir); err != nil { + level.Warn(d.logger).Log("msg", "failed to delete cached metadata fetcher directory for inactive user", "user", userID, "dir", metaCacheDir, "err", err) + } } } diff --git a/pkg/querier/blocks_finder_bucket_scan_test.go b/pkg/querier/blocks_finder_bucket_scan_test.go index 515e090bf0e..79fd9e5e8c1 100644 --- a/pkg/querier/blocks_finder_bucket_scan_test.go +++ b/pkg/querier/blocks_finder_bucket_scan_test.go @@ -28,6 +28,10 @@ import ( "github.com/cortexproject/cortex/pkg/util/users" ) +// metaSyncerCacheDirName is the sub directory that block.NewMetaFetcher appends to the directory it +// is given to cache its meta.json files. +const metaSyncerCacheDirName = "meta-syncer" + func TestBucketScanBlocksFinder_InitialScan(t *testing.T) { t.Parallel() ctx := context.Background() @@ -433,7 +437,7 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsDeletedUserFetcher(t *testing. require.Equal(t, 1, len(s.fetchers)) s.fetchersMx.Unlock() - userCacheDir := filepath.Join(s.cfg.CacheDir, "user-1") + userCacheDir := filepath.Join(s.cfg.CacheDir, users.QuerierMetaCacheDirName, "user-1") metaSyncerDir := filepath.Join(userCacheDir, metaSyncerCacheDirName) require.DirExists(t, metaSyncerDir) @@ -454,8 +458,7 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsDeletedUserFetcher(t *testing. assert.Equal(t, 0, len(s.fetchers)) s.fetchersMx.Unlock() - // The fetcher's own meta-syncer cache must be removed; the now-empty parent dir is then - // removed on a best-effort basis. + // The whole per-tenant directory the fetcher cached its metas in must be removed. assert.NoDirExists(t, metaSyncerDir) assert.NoDirExists(t, userCacheDir) @@ -586,35 +589,67 @@ func TestBucketScanBlocksFinder_PeriodicScanEvictsInactiveUserDespiteOtherTenant assert.True(t, has2) } -// TestBucketScanBlocksFinder_PeriodicScanPreservesNonMetaSyncerDataOnEviction guards the -// single-binary safety property: CacheDir is the store-gateway's SyncDir, whose block data lives -// under the same per-tenant directory, so evicting a tenant must remove only the fetcher's own -// meta-syncer cache and never the sibling block data. -func TestBucketScanBlocksFinder_PeriodicScanPreservesNonMetaSyncerDataOnEviction(t *testing.T) { +// TestBucketScanBlocksFinder_PeriodicScanPreservesColocatedStoreGatewayCache guards the +// single-binary safety property: the finder shares -blocks-storage.bucket-store.sync-dir with a +// co-located store-gateway, which keeps syncing and serving tenants marked for deletion for as long +// as their blocks are still in the bucket. The finder tracks active tenants only, so evicting a +// tenant must not remove any on-disk state the store-gateway still considers live. +func TestBucketScanBlocksFinder_PeriodicScanPreservesColocatedStoreGatewayCache(t *testing.T) { t.Parallel() ctx := context.Background() - s, bucket, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig()) + s, bkt, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig()) - cortex_testutil.MockStorageBlock(t, bucket, "user-1", 10, 20) + cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20) require.NoError(t, services.StartAndAwaitRunning(ctx, s)) - metaSyncerDir := filepath.Join(s.cfg.CacheDir, "user-1", metaSyncerCacheDirName) - require.DirExists(t, metaSyncerDir) + // The finder caches its metas under its own root, not in the store-gateway's per-tenant dir. + require.DirExists(t, filepath.Join(s.cfg.CacheDir, users.QuerierMetaCacheDirName, "user-1", metaSyncerCacheDirName)) - // Simulate a co-located store-gateway's block data under the same per-tenant directory. + // Simulate the on-disk state of a co-located store-gateway for the same tenant: its own + // metadata fetcher cache and a downloaded block. + sgMetaFile := filepath.Join(s.cfg.CacheDir, "user-1", metaSyncerCacheDirName, "01DTVP434PA9VFXSW2JKB3392D", "meta.json") + require.NoError(t, os.MkdirAll(filepath.Dir(sgMetaFile), 0o755)) + require.NoError(t, os.WriteFile(sgMetaFile, []byte("{}"), 0o644)) sgBlockFile := filepath.Join(s.cfg.CacheDir, "user-1", "01DTVP434PA9VFXSW2JKB3392D", "index") require.NoError(t, os.MkdirAll(filepath.Dir(sgBlockFile), 0o755)) require.NoError(t, os.WriteFile(sgBlockFile, []byte("block-data"), 0o644)) - require.NoError(t, bucket.Delete(ctx, "user-1")) + // Mark the tenant for deletion: it leaves the finder's active set, while the store-gateway + // keeps it in its own sync set (active + deleting) until its blocks are gone. + require.NoError(t, users.WriteTenantDeletionMark(ctx, objstore.WithNoopInstr(bkt), "user-1", users.NewTenantDeletionMark(time.Now()))) require.NoError(t, s.scan(ctx)) - // The fetcher's own meta-syncer cache is deleted... - assert.NoDirExists(t, metaSyncerDir) - // ...but the co-located store-gateway block data under the same per-tenant dir must survive. + // The finder released its own cache for the tenant it no longer tracks... + s.fetchersMx.Lock() + require.Empty(t, s.fetchers) + s.fetchersMx.Unlock() + assert.NoDirExists(t, filepath.Join(s.cfg.CacheDir, users.QuerierMetaCacheDirName, "user-1")) + + // ...but the co-located store-gateway's cache for a tenant it still serves is untouched. + assert.FileExists(t, sgMetaFile) assert.FileExists(t, sgBlockFile) } +// TestBucketScanBlocksFinder_ScanDeletesMetaCacheDirsLeftByAPreviousProcess covers that the finder's +// cache root is exclusive to it, so it can also reclaim the per-tenant directories left behind by a +// previous process, e.g. for tenants deleted while this one was down. +func TestBucketScanBlocksFinder_ScanDeletesMetaCacheDirsLeftByAPreviousProcess(t *testing.T) { + t.Parallel() + ctx := context.Background() + s, bkt, _, _ := prepareBucketScanBlocksFinder(t, prepareBucketScanBlocksFinderConfig()) + + cortex_testutil.MockStorageBlock(t, bkt, "user-1", 10, 20) + + // The cache of a tenant this process never created a metadata fetcher for. + staleUserCacheDir := filepath.Join(s.cfg.CacheDir, users.QuerierMetaCacheDirName, "user-2") + require.NoError(t, os.MkdirAll(filepath.Join(staleUserCacheDir, metaSyncerCacheDirName, "01DTVP434PA9VFXSW2JKB3392D"), 0o755)) + + require.NoError(t, services.StartAndAwaitRunning(ctx, s)) + + assert.NoDirExists(t, staleUserCacheDir) + assert.DirExists(t, filepath.Join(s.cfg.CacheDir, users.QuerierMetaCacheDirName, "user-1", metaSyncerCacheDirName)) +} + func TestBucketScanBlocksFinder_GetBlocks(t *testing.T) { //parallel testing causes data race ctx := context.Background() diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go index a5ef68a969e..b3b5dbb1f24 100644 --- a/pkg/storegateway/bucket_stores.go +++ b/pkg/storegateway/bucket_stores.go @@ -759,6 +759,14 @@ func (u *ThanosBucketStores) deleteLocalFilesForExcludedTenants(includeUserIDs m continue } + // Not a tenant directory: it's the cache root of a querier co-located in the same process + // and sharing the sync dir (single-binary mode with the bucket index disabled). That querier + // is not sharded, so it caches block metas for tenants outside this shard too, and reclaims + // them on its own once a tenant is gone from the bucket. + if userID == users.QuerierMetaCacheDirName { + continue + } + err := u.closeEmptyBucketStore(userID) switch { case errors.Is(err, errBucketStoreNotEmpty): diff --git a/pkg/storegateway/bucket_stores_test.go b/pkg/storegateway/bucket_stores_test.go index 4bf3e7af065..68f74863399 100644 --- a/pkg/storegateway/bucket_stores_test.go +++ b/pkg/storegateway/bucket_stores_test.go @@ -918,6 +918,35 @@ func TestBucketStores_deleteLocalFilesForExcludedTenants(t *testing.T) { `), metricNames...)) } +// TestBucketStores_deleteLocalFilesForExcludedTenantsPreservesQuerierMetaCache pins that the +// per-shard cleanup of the sync dir doesn't reach into the cache of a co-located querier. In +// single-binary mode with the bucket index disabled, the querier's bucket-scan blocks finder shares +// the sync dir but is not sharded, so it caches block metas for every tenant, including the ones +// this store-gateway instance doesn't own. +func TestBucketStores_deleteLocalFilesForExcludedTenantsPreservesQuerierMetaCache(t *testing.T) { + t.Parallel() + + cfg := prepareStorageConfig(t) + stores := &ThanosBucketStores{cfg: cfg, logger: log.NewNopLogger()} + + ownedDir := filepath.Join(cfg.BucketStore.SyncDir, "user-1") + excludedDir := filepath.Join(cfg.BucketStore.SyncDir, "user-2") + require.NoError(t, os.MkdirAll(ownedDir, 0o755)) + require.NoError(t, os.MkdirAll(excludedDir, 0o755)) + + querierMetaFile := filepath.Join(cfg.BucketStore.SyncDir, users.QuerierMetaCacheDirName, "user-2", "meta-syncer", "01DTVP434PA9VFXSW2JKB3392D", "meta.json") + require.NoError(t, os.MkdirAll(filepath.Dir(querierMetaFile), 0o755)) + require.NoError(t, os.WriteFile(querierMetaFile, []byte("{}"), 0o644)) + + stores.deleteLocalFilesForExcludedTenants(map[string]struct{}{"user-1": {}}) + + // The sync dir of the tenant outside the shard is reclaimed, the owned one is kept... + assert.DirExists(t, ownedDir) + assert.NoDirExists(t, excludedDir) + // ...and the co-located querier's own cache root is never a tenant directory to reclaim. + assert.FileExists(t, querierMetaFile) +} + func TestBucketStores_tokenBuckets(t *testing.T) { const ( user1 = "user-1" diff --git a/pkg/util/users/scanner.go b/pkg/util/users/scanner.go index 3b5bf21ea03..c37c2e8646a 100644 --- a/pkg/util/users/scanner.go +++ b/pkg/util/users/scanner.go @@ -16,7 +16,7 @@ import ( ) var ( - userIDsToSkip = []string{GlobalMarkersDir, UserIndexCompressedFilename} + userIDsToSkip = []string{GlobalMarkersDir, UserIndexCompressedFilename, QuerierMetaCacheDirName} ) type Scanner interface { diff --git a/pkg/util/users/tenant.go b/pkg/util/users/tenant.go index 8443c548812..077338d8942 100644 --- a/pkg/util/users/tenant.go +++ b/pkg/util/users/tenant.go @@ -10,13 +10,26 @@ import ( "github.com/weaveworks/common/user" ) -const GlobalMarkersDir = "__markers__" +const ( + GlobalMarkersDir = "__markers__" + + // QuerierMetaCacheDirName is the reserved name of the directory, inside the bucket store + // sync directory (-blocks-storage.bucket-store.sync-dir), where the querier's bucket-scan + // blocks finder keeps its own on-disk block-meta caches + // (/__querier__//). The querier and the store-gateway can share the sync + // directory when running in the same process, and each reconciles its own on-disk state + // against a different view of which tenants are live, so their per-tenant caches must not + // overlap. Like GlobalMarkersDir, the name is rejected as a tenant ID and skipped by the + // users scanner, so nothing can treat it as a tenant. + QuerierMetaCacheDirName = "__querier__" +) var ( - errTenantIDTooLong = errors.New("tenant ID is too long: max 150 characters") - errTenantIDUnsafe = errors.New("tenant ID is '.' or '..'") - errTenantIDMarkers = errors.New("tenant ID '__markers__' is not allowed") - errTenantIDUserIndex = errors.New("tenant ID 'user-index.json.gz' is not allowed") + errTenantIDTooLong = errors.New("tenant ID is too long: max 150 characters") + errTenantIDUnsafe = errors.New("tenant ID is '.' or '..'") + errTenantIDMarkers = errors.New("tenant ID '__markers__' is not allowed") + errTenantIDUserIndex = errors.New("tenant ID 'user-index.json.gz' is not allowed") + errTenantIDQuerierMetaDir = errors.New("tenant ID '__querier__' is not allowed") ) type errTenantIDUnsupportedCharacter struct { @@ -95,6 +108,11 @@ func CheckTenantIDIsSupported(s string) error { return errTenantIDUserIndex } + // check tenantID is the querier's reserved meta cache directory + if s == QuerierMetaCacheDirName { + return errTenantIDQuerierMetaDir + } + // check tenantID is "." or ".." if containsUnsafePathSegments(s) { return errTenantIDUnsafe diff --git a/pkg/util/users/tenant_test.go b/pkg/util/users/tenant_test.go index 386c35ab94e..6bc866e6ec0 100644 --- a/pkg/util/users/tenant_test.go +++ b/pkg/util/users/tenant_test.go @@ -45,6 +45,10 @@ func TestValidTenantIDs(t *testing.T) { name: "user-index.json.gz", err: new("tenant ID 'user-index.json.gz' is not allowed"), }, + { + name: "__querier__", + err: new("tenant ID '__querier__' is not allowed"), + }, } { t.Run(tc.name, func(t *testing.T) { err := ValidTenantID(tc.name)