From 3ea46524c5505b8b459640e08dc13c580387e205 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Mon, 29 Jun 2026 09:54:17 +0900 Subject: [PATCH 1/6] resolve parquet shard count from bucket index to reduce object storage calls Signed-off-by: SungJin1212 --- CHANGELOG.md | 1 + docs/blocks-storage/querier.md | 5 +- docs/blocks-storage/store-gateway.md | 5 +- docs/configuration/config-file-reference.md | 5 +- pkg/storage/tsdb/config.go | 4 +- pkg/storegateway/bucket_stores.go | 4 ++ pkg/storegateway/gateway.go | 3 + pkg/storegateway/parquet_bucket_store.go | 67 ++++++++++++++----- pkg/storegateway/parquet_bucket_stores.go | 47 ++++++++++--- .../parquet_bucket_stores_test.go | 66 ++++++++++++++++++ schemas/cortex-config-schema.json | 4 +- 11 files changed, 178 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a9c2e68ece..cd7f5659978 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,6 +39,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] Store Gateway: Resolve the parquet shard count from the bucket index instead of reading the converter mark for each block, reducing object storage calls when the bucket index is enabled. #7648 * [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] Upgrade prometheus version to v3.9.1. #7535 diff --git a/docs/blocks-storage/querier.md b/docs/blocks-storage/querier.md index 73c3e2e2729..fa063b36519 100644 --- a/docs/blocks-storage/querier.md +++ b/docs/blocks-storage/querier.md @@ -2006,13 +2006,14 @@ blocks_storage: [enabled: | default = true] # How frequently a bucket index, which previously failed to load, should - # be tried to load again. This option is used only by querier. + # be tried to load again. This option is used by querier and store-gateway + # parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.update-on-error-interval [update_on_error_interval: | default = 1m] # How long a unused bucket index should be cached. Once this timeout # expires, the unused bucket index is removed from the in-memory cache. - # This option is used only by querier. + # This option is used by querier and store-gateway parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.idle-timeout [idle_timeout: | default = 1h] diff --git a/docs/blocks-storage/store-gateway.md b/docs/blocks-storage/store-gateway.md index 4e7e640ed20..a558d102658 100644 --- a/docs/blocks-storage/store-gateway.md +++ b/docs/blocks-storage/store-gateway.md @@ -2064,13 +2064,14 @@ blocks_storage: [enabled: | default = true] # How frequently a bucket index, which previously failed to load, should - # be tried to load again. This option is used only by querier. + # be tried to load again. This option is used by querier and store-gateway + # parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.update-on-error-interval [update_on_error_interval: | default = 1m] # How long a unused bucket index should be cached. Once this timeout # expires, the unused bucket index is removed from the in-memory cache. - # This option is used only by querier. + # This option is used by querier and store-gateway parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.idle-timeout [idle_timeout: | default = 1h] diff --git a/docs/configuration/config-file-reference.md b/docs/configuration/config-file-reference.md index e4d74aedffa..0b6609f9865 100644 --- a/docs/configuration/config-file-reference.md +++ b/docs/configuration/config-file-reference.md @@ -2656,13 +2656,14 @@ bucket_store: [enabled: | default = true] # How frequently a bucket index, which previously failed to load, should be - # tried to load again. This option is used only by querier. + # tried to load again. This option is used by querier and store-gateway + # parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.update-on-error-interval [update_on_error_interval: | default = 1m] # How long a unused bucket index should be cached. Once this timeout # expires, the unused bucket index is removed from the in-memory cache. This - # option is used only by querier. + # option is used by querier and store-gateway parquet mode. # CLI flag: -blocks-storage.bucket-store.bucket-index.idle-timeout [idle_timeout: | default = 1h] diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index 646a988cade..bd921400803 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -457,8 +457,8 @@ type BucketIndexConfig struct { func (cfg *BucketIndexConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) { f.BoolVar(&cfg.Enabled, prefix+"enabled", true, "True to enable querier and store-gateway to discover blocks in the storage via bucket index instead of bucket scanning. Disabling the bucket index is not recommended for production.") - f.DurationVar(&cfg.UpdateOnErrorInterval, prefix+"update-on-error-interval", time.Minute, "How frequently a bucket index, which previously failed to load, should be tried to load again. This option is used only by querier.") - f.DurationVar(&cfg.IdleTimeout, prefix+"idle-timeout", time.Hour, "How long a unused bucket index should be cached. Once this timeout expires, the unused bucket index is removed from the in-memory cache. This option is used only by querier.") + f.DurationVar(&cfg.UpdateOnErrorInterval, prefix+"update-on-error-interval", time.Minute, "How frequently a bucket index, which previously failed to load, should be tried to load again. This option is used by querier and store-gateway parquet mode.") + f.DurationVar(&cfg.IdleTimeout, prefix+"idle-timeout", time.Hour, "How long a unused bucket index should be cached. Once this timeout expires, the unused bucket index is removed from the in-memory cache. This option is used by querier and store-gateway parquet mode.") f.DurationVar(&cfg.MaxStalePeriod, prefix+"max-stale-period", time.Hour, "The maximum allowed age of a bucket index (last updated) before queries start failing because the bucket index is too old. The bucket index is periodically updated by the compactor, while this check is enforced in the querier (at query time).") } diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go index c905ff92396..20ef4b020e1 100644 --- a/pkg/storegateway/bucket_stores.go +++ b/pkg/storegateway/bucket_stores.go @@ -49,6 +49,7 @@ type BucketStores interface { storepb.StoreServer SyncBlocks(ctx context.Context) error InitialSync(ctx context.Context) error + Stop() error } // ThanosBucketStores is a multi-tenant wrapper of Thanos BucketStore. @@ -230,6 +231,9 @@ func (u *ThanosBucketStores) SyncBlocks(ctx context.Context) error { }) } +// Stop implements BucketStores. ThanosBucketStores has no background services to stop. +func (u *ThanosBucketStores) Stop() error { return nil } + func (u *ThanosBucketStores) syncUsersBlocksWithRetries(ctx context.Context, f func(context.Context, *store.BucketStore) error) error { retries := backoff.New(ctx, backoff.Config{ MinBackoff: 1 * time.Second, diff --git a/pkg/storegateway/gateway.go b/pkg/storegateway/gateway.go index 14724d60b51..857f91d24b6 100644 --- a/pkg/storegateway/gateway.go +++ b/pkg/storegateway/gateway.go @@ -391,6 +391,9 @@ func (g *StoreGateway) running(ctx context.Context) error { } func (g *StoreGateway) stopping(_ error) error { + if g.stores != nil { + _ = g.stores.Stop() + } if g.subservices != nil { return services.StopManagerAndAwaitStopped(context.Background(), g.subservices) } diff --git a/pkg/storegateway/parquet_bucket_store.go b/pkg/storegateway/parquet_bucket_store.go index 5a5d740bc41..abace8c6be7 100644 --- a/pkg/storegateway/parquet_bucket_store.go +++ b/pkg/storegateway/parquet_bucket_store.go @@ -27,16 +27,20 @@ import ( "google.golang.org/grpc/status" cortex_parquet "github.com/cortexproject/cortex/pkg/storage/parquet" + "github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex" "github.com/cortexproject/cortex/pkg/util/parquetutil" "github.com/cortexproject/cortex/pkg/util/spanlogger" "github.com/cortexproject/cortex/pkg/util/validation" ) type parquetBucketStore struct { - logger log.Logger - bucket objstore.InstrumentedBucket - limits *validation.Overrides - concurrency int + logger log.Logger + bucket objstore.InstrumentedBucket + indexLoader *bucketindex.Loader // nil when bucket index disabled + limits *validation.Overrides + userID string + bucketIndexEnabled bool + concurrency int chunksDecoder *schema.PrometheusParquetChunksDecoder @@ -67,20 +71,15 @@ func (p *parquetBucketStore) findParquetBlocks(ctx context.Context, blockMatcher bucketOpener := parquet_storage.NewParquetBucketOpener(p.bucket) noopQuota := search.NewQuota(search.NoopQuotaLimitFunc(ctx)) - // Read converter marks and expand to per-shard (blockID, shardID) lists. - // TODO(Sungjin1212): Read the shard count from the bucket index instead of reading the converter mark for each block. + shardCounts, err := p.resolveShardCounts(ctx, blockIDs) + if err != nil { + return nil, err + } + var shardBlockIDs []string var shardIDs []int for _, blockID := range blockIDs { - uid, err := ulid.Parse(blockID) - if err != nil { - return nil, errors.Wrapf(err, "failed to parse block ID %s", blockID) - } - marker, err := cortex_parquet.ReadConverterMark(ctx, uid, p.bucket, p.logger) - if err != nil { - return nil, errors.Wrapf(err, "failed to read converter mark for block %s", blockID) - } - numShards := marker.Shards + numShards := shardCounts[blockID] if numShards <= 0 { // backward compatibility: blocks without a shard count have one shard numShards = 1 @@ -112,6 +111,44 @@ func (p *parquetBucketStore) findParquetBlocks(ctx context.Context, blockMatcher return parquetBlocks, nil } +// resolveShardCounts returns the number of parquet shards for each requested block ID. +// +// When the bucket index is enabled, the shard count is read from the bucket index. +// When the bucket index is disabled, it falls back to reading the converter mark +// for each block. +func (p *parquetBucketStore) resolveShardCounts(ctx context.Context, blockIDs []string) (map[string]int, error) { + shardCounts := make(map[string]int, len(blockIDs)) + + if p.bucketIndexEnabled && p.indexLoader != nil { + idx, _, err := p.indexLoader.GetIndex(ctx, p.userID) + if err != nil { + return nil, errors.Wrap(err, "failed to get bucket index") + } + for _, b := range idx.Blocks { + numShards := 1 + if b.Parquet != nil && b.Parquet.Shards > 0 { + numShards = b.Parquet.Shards + } + shardCounts[b.ID.String()] = numShards + } + return shardCounts, nil + } + + // Fallback: read the converter mark for each block. + for _, blockID := range blockIDs { + uid, err := ulid.Parse(blockID) + if err != nil { + return nil, errors.Wrapf(err, "failed to parse block ID %s", blockID) + } + marker, err := cortex_parquet.ReadConverterMark(ctx, uid, p.bucket, p.logger) + if err != nil { + return nil, errors.Wrapf(err, "failed to read converter mark for block %s", blockID) + } + shardCounts[blockID] = marker.Shards + } + return shardCounts, nil +} + // Series implements the store interface for a single parquet bucket store func (p *parquetBucketStore) Series(req *storepb.SeriesRequest, seriesSrv storepb.Store_SeriesServer) (err error) { spanLog, ctx := spanlogger.New(seriesSrv.Context(), "ParquetBucketStore.Series") diff --git a/pkg/storegateway/parquet_bucket_stores.go b/pkg/storegateway/parquet_bucket_stores.go index f0c488ba686..9c6ac89bec5 100644 --- a/pkg/storegateway/parquet_bucket_stores.go +++ b/pkg/storegateway/parquet_bucket_stores.go @@ -5,6 +5,7 @@ import ( "fmt" "sort" "sync" + "time" "github.com/go-kit/log" "github.com/go-kit/log/level" @@ -31,9 +32,11 @@ import ( "github.com/cortexproject/cortex/pkg/querysharding" "github.com/cortexproject/cortex/pkg/storage/bucket" "github.com/cortexproject/cortex/pkg/storage/tsdb" + "github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex" cortex_util "github.com/cortexproject/cortex/pkg/util" cortex_errors "github.com/cortexproject/cortex/pkg/util/errors" "github.com/cortexproject/cortex/pkg/util/parquetutil" + "github.com/cortexproject/cortex/pkg/util/services" "github.com/cortexproject/cortex/pkg/util/spanlogger" "github.com/cortexproject/cortex/pkg/util/users" "github.com/cortexproject/cortex/pkg/util/validation" @@ -60,6 +63,10 @@ type ParquetBucketStores struct { rowRangesCache search.RowRangesForConstraintsCache inflightRequests *cortex_util.InflightRequestTracker + + // indexLoader lazily loads and caches the per-tenant bucket index in memory + // It is non-nil only when BucketIndex.Enabled. + indexLoader *bucketindex.Loader } // newParquetBucketStores creates a new multi-tenant parquet bucket stores @@ -108,6 +115,15 @@ func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore. u.matcherCache = storecache.NoopMatchersCache } + if cfg.BucketStore.BucketIndex.Enabled { + u.indexLoader = bucketindex.NewLoader(bucketindex.LoaderConfig{ + CheckInterval: time.Minute, + UpdateOnStaleInterval: cfg.BucketStore.SyncInterval, + UpdateOnErrorInterval: cfg.BucketStore.BucketIndex.UpdateOnErrorInterval, + IdleTimeout: cfg.BucketStore.BucketIndex.IdleTimeout, + }, bucketClient, limits, logger, reg) + } + return u, nil } @@ -218,6 +234,18 @@ func (u *ParquetBucketStores) SyncBlocks(ctx context.Context) error { // InitialSync implements BucketStores func (u *ParquetBucketStores) InitialSync(ctx context.Context) error { + if u.indexLoader != nil { + // Start indexLoader + return services.StartAndAwaitRunning(ctx, u.indexLoader) + } + return nil +} + +// Stop implements BucketStores +func (u *ParquetBucketStores) Stop() error { + if u.indexLoader != nil { + return services.StopAndAwaitTerminated(context.Background(), u.indexLoader) + } return nil } @@ -270,14 +298,17 @@ func (u *ParquetBucketStores) createParquetBucketStore(userID string, userLogger userBucket := bucket.NewUserBucketClient(userID, u.bucket, u.limits) store := &parquetBucketStore{ - logger: userLogger, - bucket: userBucket, - limits: u.limits, - concurrency: u.cfg.BucketStore.ParquetQueryConcurrency, - chunksDecoder: u.chunksDecoder, - matcherCache: u.matcherCache, - parquetShardCache: u.parquetShardCache, - rowRangesCache: u.rowRangesCache, + logger: userLogger, + bucket: userBucket, + indexLoader: u.indexLoader, + limits: u.limits, + userID: userID, + bucketIndexEnabled: u.cfg.BucketStore.BucketIndex.Enabled, + concurrency: u.cfg.BucketStore.ParquetQueryConcurrency, + chunksDecoder: u.chunksDecoder, + matcherCache: u.matcherCache, + parquetShardCache: u.parquetShardCache, + rowRangesCache: u.rowRangesCache, } return store, nil diff --git a/pkg/storegateway/parquet_bucket_stores_test.go b/pkg/storegateway/parquet_bucket_stores_test.go index d577de06551..797bd4c87c4 100644 --- a/pkg/storegateway/parquet_bucket_stores_test.go +++ b/pkg/storegateway/parquet_bucket_stores_test.go @@ -1,6 +1,7 @@ package storegateway import ( + "bytes" "context" "errors" "fmt" @@ -30,6 +31,7 @@ import ( "github.com/cortexproject/cortex/pkg/storage/bucket/filesystem" cortex_parquet "github.com/cortexproject/cortex/pkg/storage/parquet" cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb" + "github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex" "github.com/cortexproject/cortex/pkg/util/users" "github.com/cortexproject/cortex/pkg/util/validation" ) @@ -485,3 +487,67 @@ func TestParquetBucketStores_Series_MultiShard(t *testing.T) { require.NoError(t, err) assert.Equal(t, numSeries, len(series), "all series from all shards must be returned") } + +// TestParquetBucketStores_Series_MultiShard_BucketIndex verifies that, when the bucket +// index is enabled, the Store Gateway resolves the parquet shard count from the bucket +// index instead of reading the per-block converter mark. +func TestParquetBucketStores_Series_MultiShard_BucketIndex(t *testing.T) { + const ( + userID = "user-1" + metricName = "test_metric" + numSeries = 6 // 6 unique series + // numRowGroups=1, maxRowsPerRowGroup=2 → ceil(6/1*2) = 3 shards + numRowGroups = 1 + maxRowsPerRowGroup = 2 + expectedShards = 3 + ) + + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + cfg.BucketStore.BucketIndex.Enabled = true + + storageDir := t.TempDir() + + // Create a block with 6 unique series. + generateStorageBlockWithMultipleSeries(t, storageDir, userID, metricName, numSeries, 0, 100, 15) + + bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) + require.NoError(t, err) + + overrides := validation.NewOverrides(validation.Limits{}, nil) + uBucket := bucket.NewUserBucketClient(userID, bkt, overrides) + + // Convert to parquet with 3 shards and write the (correct) converter mark. + userPath := filepath.Join(storageDir, userID) + blockIDs, err := convertToParquetBlocksWithShardsForTesting(userPath, uBucket, numRowGroups, maxRowsPerRowGroup) + require.NoError(t, err) + require.Len(t, blockIDs, 1) + + uidV2, err := ulidv2.Parse(blockIDs[0]) + require.NoError(t, err) + + // The bucket index discovers parquet blocks via the global markers location + // (parquet-markers/-parquet-converter-mark.json), so upload it there too. + require.NoError(t, uBucket.Upload(context.Background(), bucketindex.ConverterMarkFilePath(uidV2), bytes.NewReader([]byte("{}")))) + + // Build the bucket index (with parquet info) so the shard count is recorded there. + idx, _, _, err := bucketindex.NewUpdater(bkt, userID, nil, log.NewNopLogger()).EnableParquet().UpdateIndex(context.Background(), nil) + require.NoError(t, err) + require.NoError(t, bucketindex.WriteIndex(context.Background(), bkt, userID, nil, idx)) + + // The bucket index must record the actual number of shards (3). + require.Len(t, idx.Blocks, 1) + require.NotNil(t, idx.Blocks[0].Parquet) + require.Equal(t, expectedShards, idx.Blocks[0].Parquet.Shards, "bucket index should record 3 shards") + + // Overwrite the converter mark with a wrong shard count (1). If the Store Gateway + // used the converter mark instead of the bucket index, it would only read 1 shard. + require.NoError(t, cortex_parquet.WriteConverterMark(context.Background(), uidV2, uBucket, 1)) + + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), prometheus.NewRegistry()) + require.NoError(t, err) + + series, _, err := querySeries(stores, userID, metricName, 0, 100, blockIDs...) + require.NoError(t, err) + assert.Equal(t, numSeries, len(series), "all series must be returned using the bucket index shard count") +} diff --git a/schemas/cortex-config-schema.json b/schemas/cortex-config-schema.json index a1bbeb2bd90..dba0f587120 100644 --- a/schemas/cortex-config-schema.json +++ b/schemas/cortex-config-schema.json @@ -1094,7 +1094,7 @@ }, "idle_timeout": { "default": "1h0m0s", - "description": "How long a unused bucket index should be cached. Once this timeout expires, the unused bucket index is removed from the in-memory cache. This option is used only by querier.", + "description": "How long a unused bucket index should be cached. Once this timeout expires, the unused bucket index is removed from the in-memory cache. This option is used by querier and store-gateway parquet mode.", "type": "string", "x-cli-flag": "blocks-storage.bucket-store.bucket-index.idle-timeout", "x-format": "duration" @@ -1108,7 +1108,7 @@ }, "update_on_error_interval": { "default": "1m0s", - "description": "How frequently a bucket index, which previously failed to load, should be tried to load again. This option is used only by querier.", + "description": "How frequently a bucket index, which previously failed to load, should be tried to load again. This option is used by querier and store-gateway parquet mode.", "type": "string", "x-cli-flag": "blocks-storage.bucket-store.bucket-index.update-on-error-interval", "x-format": "duration" From 2c651473289b2964889d9fb39126d7ca4d4eaa90 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Mon, 29 Jun 2026 11:20:58 +0900 Subject: [PATCH 2/6] add component labels for store-gateway and store-queryable Signed-off-by: SungJin1212 --- CHANGELOG.md | 2 +- integration/parquet_querier_test.go | 11 +++++++++++ pkg/querier/blocks_store_queryable.go | 2 +- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cd7f5659978..d09d0e865a9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -39,7 +39,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] Store Gateway: Resolve the parquet shard count from the bucket index instead of reading the converter mark for each block, reducing object storage calls when the bucket index is enabled. #7648 +* [ENHANCEMENT] Store Gateway: Resolve the parquet shard count from the bucket index instead of reading the converter mark for each block, reducing object storage calls when the bucket index is enabled. A `component` label is added to the bucket index loader metrics to distinguish store-queryable and store-gateway. #7648 * [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] Upgrade prometheus version to v3.9.1. #7535 diff --git a/integration/parquet_querier_test.go b/integration/parquet_querier_test.go index d696d181bf9..330e6d908a7 100644 --- a/integration/parquet_querier_test.go +++ b/integration/parquet_querier_test.go @@ -619,9 +619,20 @@ func TestParquetMultiShardQuery(t *testing.T) { if tc.viaStoreGateway { require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_querier_storegateway_instances_hit_per_query"}, e2e.WithMetricCount, e2e.SkipMissingMetrics)) require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_querier_blocks_consistency_checks_total"}, e2e.SkipMissingMetrics)) + // The store-gateway parquet bucket store resolves the shard count via the + // bucket index loader, which is tagged with component="store-gateway". + if tc.extraStoreGateways == 0 { + // Only assert on the single binary when there are no extra replicas, since + // with replication the query may be served by another store-gateway. + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_bucket_index_loads_total"}, e2e.WithLabelMatchers( + labels.MustNewMatcher(labels.MatchEqual, "component", "store-gateway")))) + } } else { require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_parquet_queryable_blocks_queried_total"}, e2e.WithLabelMatchers( labels.MustNewMatcher(labels.MatchEqual, "type", "parquet")))) + // The querier's bucket index loader is tagged with component="store-queryable". + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_bucket_index_loads_total"}, e2e.WithLabelMatchers( + labels.MustNewMatcher(labels.MatchEqual, "component", "store-queryable")))) } }) } diff --git a/pkg/querier/blocks_store_queryable.go b/pkg/querier/blocks_store_queryable.go index e7063d99bb8..d7ce7be4e81 100644 --- a/pkg/querier/blocks_store_queryable.go +++ b/pkg/querier/blocks_store_queryable.go @@ -210,7 +210,7 @@ func NewBlocksStoreQueryableFromConfig(querierCfg Config, gatewayCfg storegatewa MaxStalePeriod: storageCfg.BucketStore.BucketIndex.MaxStalePeriod, IgnoreDeletionMarksDelay: storageCfg.BucketStore.IgnoreDeletionMarksDelay, IgnoreBlocksWithin: storageCfg.BucketStore.IgnoreBlocksWithin, - }, bucketClient, limits, logger, reg) + }, bucketClient, limits, logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "store-queryable"}, reg)) } else { usersScanner, err := users.NewScanner(storageCfg.UsersScanner, bucketClient, logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "querier"}, reg)) if err != nil { From 9bddc9497db38864a354a2d1f1a71298f4d49b95 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 7 Jul 2026 11:12:23 +0900 Subject: [PATCH 3/6] add warning log Signed-off-by: SungJin1212 --- pkg/storegateway/gateway.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/pkg/storegateway/gateway.go b/pkg/storegateway/gateway.go index 857f91d24b6..3bcbdbbf64c 100644 --- a/pkg/storegateway/gateway.go +++ b/pkg/storegateway/gateway.go @@ -392,7 +392,9 @@ func (g *StoreGateway) running(ctx context.Context) error { func (g *StoreGateway) stopping(_ error) error { if g.stores != nil { - _ = g.stores.Stop() + if err := g.stores.Stop(); err != nil { + level.Warn(g.logger).Log("msg", "failed to stop bucket stores", "err", err) + } } if g.subservices != nil { return services.StopManagerAndAwaitStopped(context.Background(), g.subservices) From f6dbed134a74cdb5a1102e2bd8bbb330369f3bad Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Wed, 8 Jul 2026 16:34:39 +0900 Subject: [PATCH 4/6] optimize shard count retrieval by memoizing results from bucket index Signed-off-by: SungJin1212 --- pkg/storegateway/parquet_bucket_store.go | 47 +++++++++++++++---- .../parquet_bucket_stores_test.go | 34 ++++++++++++++ 2 files changed, 73 insertions(+), 8 deletions(-) diff --git a/pkg/storegateway/parquet_bucket_store.go b/pkg/storegateway/parquet_bucket_store.go index abace8c6be7..ae03666ce1f 100644 --- a/pkg/storegateway/parquet_bucket_store.go +++ b/pkg/storegateway/parquet_bucket_store.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "strings" + "sync" "github.com/go-kit/log" "github.com/gogo/protobuf/types" @@ -47,6 +48,12 @@ type parquetBucketStore struct { matcherCache storecache.MatchersCache parquetShardCache parquetutil.CacheInterface[parquet_storage.ParquetShard] rowRangesCache search.RowRangesForConstraintsCache + + shardCountsMu sync.Mutex + // cachedShardCounts maps a block ID to its parquet shard count. The map is rebuilt + // only when the bucket index is refreshed (detected via cachedIndexUpdatedAt). + cachedShardCounts map[string]int + cachedIndexUpdatedAt int64 } func (p *parquetBucketStore) Close() error { @@ -124,14 +131,7 @@ func (p *parquetBucketStore) resolveShardCounts(ctx context.Context, blockIDs [] if err != nil { return nil, errors.Wrap(err, "failed to get bucket index") } - for _, b := range idx.Blocks { - numShards := 1 - if b.Parquet != nil && b.Parquet.Shards > 0 { - numShards = b.Parquet.Shards - } - shardCounts[b.ID.String()] = numShards - } - return shardCounts, nil + return p.shardCountsFromIndex(idx), nil } // Fallback: read the converter mark for each block. @@ -149,6 +149,37 @@ func (p *parquetBucketStore) resolveShardCounts(ctx context.Context, blockIDs [] return shardCounts, nil } +// shardCountsFromIndex returns a block ID to shard count map built from the bucket index. +// +// The parquet shard count of a block is immutable once the block is converted, so the map +// is rebuilt only when the bucket index has been refreshed (detected via UpdatedAt). In +// between refreshes the previously built map is reused, avoiding a full iteration over all +// blocks in the index on every request. +func (p *parquetBucketStore) shardCountsFromIndex(idx *bucketindex.Index) map[string]int { + p.shardCountsMu.Lock() + defer p.shardCountsMu.Unlock() + + if p.cachedShardCounts != nil && p.cachedIndexUpdatedAt == idx.UpdatedAt { + return p.cachedShardCounts + } + + shardCounts := make(map[string]int, len(idx.Blocks)) + for _, b := range idx.Blocks { + if b.Parquet == nil { + continue + } + numShards := b.Parquet.Shards + if numShards <= 0 { + numShards = 1 + } + shardCounts[b.ID.String()] = numShards + } + + p.cachedShardCounts = shardCounts + p.cachedIndexUpdatedAt = idx.UpdatedAt + return shardCounts +} + // Series implements the store interface for a single parquet bucket store func (p *parquetBucketStore) Series(req *storepb.SeriesRequest, seriesSrv storepb.Store_SeriesServer) (err error) { spanLog, ctx := spanlogger.New(seriesSrv.Context(), "ParquetBucketStore.Series") diff --git a/pkg/storegateway/parquet_bucket_stores_test.go b/pkg/storegateway/parquet_bucket_stores_test.go index 797bd4c87c4..64e1616c35f 100644 --- a/pkg/storegateway/parquet_bucket_stores_test.go +++ b/pkg/storegateway/parquet_bucket_stores_test.go @@ -164,6 +164,40 @@ func TestParquetBucketStore_FindParquetBlocks_InvalidMatchers(t *testing.T) { assert.Equal(t, codes.InvalidArgument, s.Code()) } +func TestParquetBucketStore_ShardCountsFromIndex_Memoized(t *testing.T) { + uid1 := ulidv2.MustNew(1, nil) + uid2 := ulidv2.MustNew(2, nil) + uid3 := ulidv2.MustNew(3, nil) + idx := &bucketindex.Index{ + UpdatedAt: 100, + Blocks: bucketindex.Blocks{ + {ID: uid1, Parquet: &cortex_parquet.ConverterMarkMeta{Shards: 3}}, + {ID: uid2}, // Non-parquet block -> excluded from the map. + {ID: uid3, Parquet: &cortex_parquet.ConverterMarkMeta{Shards: 0}}, // Old parquet block without recorded shards -> 1. + }, + } + + store := &parquetBucketStore{logger: log.NewNopLogger()} + + got := store.shardCountsFromIndex(idx) + assert.Equal(t, 3, got[uid1.String()]) + _, ok := got[uid2.String()] + assert.False(t, ok, "non-parquet blocks must not be tracked in the shard count map") + assert.Equal(t, 1, got[uid3.String()], "parquet block without recorded shards defaults to 1") + assert.Equal(t, int64(100), store.cachedIndexUpdatedAt) + + // Same UpdatedAt -> the map is reused rather than rebuilt. + store.cachedShardCounts[uid1.String()] = 999 + reused := store.shardCountsFromIndex(idx) + assert.Equal(t, 999, reused[uid1.String()], "map should be reused when UpdatedAt is unchanged") + + // A new UpdatedAt -> the map is rebuilt from the index. + idx.UpdatedAt = 200 + rebuilt := store.shardCountsFromIndex(idx) + assert.Equal(t, 3, rebuilt[uid1.String()], "map should be rebuilt when UpdatedAt changes") + assert.Equal(t, int64(200), store.cachedIndexUpdatedAt) +} + func TestChunkToStoreEncoding(t *testing.T) { tests := []struct { name string From 0609a30dcc4937c39ffdc3cbddfff876b9636d85 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Thu, 9 Jul 2026 11:50:33 +0900 Subject: [PATCH 5/6] fallback to converter mark for shard count retrieval when bucket index is stale Signed-off-by: SungJin1212 --- pkg/storegateway/parquet_bucket_store.go | 36 ++++++++++--- .../parquet_bucket_stores_test.go | 50 +++++++++++++++++++ 2 files changed, 78 insertions(+), 8 deletions(-) diff --git a/pkg/storegateway/parquet_bucket_store.go b/pkg/storegateway/parquet_bucket_store.go index ae03666ce1f..2aa84a37836 100644 --- a/pkg/storegateway/parquet_bucket_store.go +++ b/pkg/storegateway/parquet_bucket_store.go @@ -86,7 +86,17 @@ func (p *parquetBucketStore) findParquetBlocks(ctx context.Context, blockMatcher var shardBlockIDs []string var shardIDs []int for _, blockID := range blockIDs { - numShards := shardCounts[blockID] + numShards, ok := shardCounts[blockID] + if !ok { + // The bucket index may not carry this block's shard count yet, for example + // right after the block has been converted but before the index has been + // refreshed. In that case, read the converter mark directly to get the + // actual shard count instead of assuming a single shard. + numShards, err = p.shardCountFromConverterMark(ctx, blockID) + if err != nil { + return nil, err + } + } if numShards <= 0 { // backward compatibility: blocks without a shard count have one shard numShards = 1 @@ -136,19 +146,29 @@ func (p *parquetBucketStore) resolveShardCounts(ctx context.Context, blockIDs [] // Fallback: read the converter mark for each block. for _, blockID := range blockIDs { - uid, err := ulid.Parse(blockID) + numShards, err := p.shardCountFromConverterMark(ctx, blockID) if err != nil { - return nil, errors.Wrapf(err, "failed to parse block ID %s", blockID) + return nil, err } - marker, err := cortex_parquet.ReadConverterMark(ctx, uid, p.bucket, p.logger) - if err != nil { - return nil, errors.Wrapf(err, "failed to read converter mark for block %s", blockID) - } - shardCounts[blockID] = marker.Shards + shardCounts[blockID] = numShards } return shardCounts, nil } +// shardCountFromConverterMark reads the parquet converter mark for the given block and +// returns the number of shards recorded in it. +func (p *parquetBucketStore) shardCountFromConverterMark(ctx context.Context, blockID string) (int, error) { + uid, err := ulid.Parse(blockID) + if err != nil { + return 0, errors.Wrapf(err, "failed to parse block ID %s", blockID) + } + marker, err := cortex_parquet.ReadConverterMark(ctx, uid, p.bucket, p.logger) + if err != nil { + return 0, errors.Wrapf(err, "failed to read converter mark for block %s", blockID) + } + return marker.Shards, nil +} + // shardCountsFromIndex returns a block ID to shard count map built from the bucket index. // // The parquet shard count of a block is immutable once the block is converted, so the map diff --git a/pkg/storegateway/parquet_bucket_stores_test.go b/pkg/storegateway/parquet_bucket_stores_test.go index 64e1616c35f..e7f3b8780c2 100644 --- a/pkg/storegateway/parquet_bucket_stores_test.go +++ b/pkg/storegateway/parquet_bucket_stores_test.go @@ -585,3 +585,53 @@ func TestParquetBucketStores_Series_MultiShard_BucketIndex(t *testing.T) { require.NoError(t, err) assert.Equal(t, numSeries, len(series), "all series must be returned using the bucket index shard count") } + +// TestParquetBucketStores_Series_MultiShard_BucketIndexStale_FallbackToConverterMark +// verifies that, when the bucket index is enabled but doesn't yet carry the parquet shard +// info for a block, the Store Gateway falls back to reading the converter mark instead of +// defaulting to a single shard. +func TestParquetBucketStores_Series_MultiShard_BucketIndexStale_FallbackToConverterMark(t *testing.T) { + const ( + userID = "user-1" + metricName = "test_metric" + numSeries = 6 // 6 unique series + // numRowGroups=1, maxRowsPerRowGroup=2 → ceil(6/1*2) = 3 shards + numRowGroups = 1 + maxRowsPerRowGroup = 2 + ) + + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + cfg.BucketStore.BucketIndex.Enabled = true + + storageDir := t.TempDir() + + generateStorageBlockWithMultipleSeries(t, storageDir, userID, metricName, numSeries, 0, 100, 15) + + bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) + require.NoError(t, err) + + overrides := validation.NewOverrides(validation.Limits{}, nil) + uBucket := bucket.NewUserBucketClient(userID, bkt, overrides) + + // Build the bucket index before converting to parquet, without parquet info. This models + // a stale index that knows about the block but doesn't record its shard count yet. + idx, _, _, err := bucketindex.NewUpdater(bkt, userID, nil, log.NewNopLogger()).UpdateIndex(context.Background(), nil) + require.NoError(t, err) + require.NoError(t, bucketindex.WriteIndex(context.Background(), bkt, userID, nil, idx)) + require.Len(t, idx.Blocks, 1) + require.Nil(t, idx.Blocks[0].Parquet, "bucket index must not carry parquet shard info yet") + + // Convert to parquet with 3 shards and write the correct converter mark. + userPath := filepath.Join(storageDir, userID) + blockIDs, err := convertToParquetBlocksWithShardsForTesting(userPath, uBucket, numRowGroups, maxRowsPerRowGroup) + require.NoError(t, err) + require.Len(t, blockIDs, 1) + + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), prometheus.NewRegistry()) + require.NoError(t, err) + + series, _, err := querySeries(stores, userID, metricName, 0, 100, blockIDs...) + require.NoError(t, err) + assert.Equal(t, numSeries, len(series), "all series from all shards must be returned via the converter mark fallback") +} From 59a512dd3b8d6f997fe2d21d1c96ae17beb71125 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Sat, 11 Jul 2026 09:29:51 +0900 Subject: [PATCH 6/6] add fallback when getting index fail Signed-off-by: SungJin1212 --- pkg/storegateway/parquet_bucket_store.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pkg/storegateway/parquet_bucket_store.go b/pkg/storegateway/parquet_bucket_store.go index 2aa84a37836..246e68e9abc 100644 --- a/pkg/storegateway/parquet_bucket_store.go +++ b/pkg/storegateway/parquet_bucket_store.go @@ -7,6 +7,7 @@ import ( "sync" "github.com/go-kit/log" + "github.com/go-kit/log/level" "github.com/gogo/protobuf/types" "github.com/oklog/ulid/v2" "github.com/pkg/errors" @@ -138,10 +139,10 @@ func (p *parquetBucketStore) resolveShardCounts(ctx context.Context, blockIDs [] if p.bucketIndexEnabled && p.indexLoader != nil { idx, _, err := p.indexLoader.GetIndex(ctx, p.userID) - if err != nil { - return nil, errors.Wrap(err, "failed to get bucket index") + if err == nil { + return p.shardCountsFromIndex(idx), nil } - return p.shardCountsFromIndex(idx), nil + level.Warn(p.logger).Log("msg", "failed to get bucket index, falling back to converter marks for parquet shard count", "err", err) } // Fallback: read the converter mark for each block.