From e4b13acb8f8d6b2413b5a8cd3f40215bd6a6849f Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 14 Jul 2026 16:43:11 +0900 Subject: [PATCH 1/5] Implement hybrid mode Store Gateway Signed-off-by: SungJin1212 --- CHANGELOG.md | 1 + integration/parquet_querier_test.go | 209 ++++++ pkg/storage/parquet/converter_marker.go | 10 +- pkg/storage/tsdb/bucketindex/updater.go | 5 +- pkg/storage/tsdb/config.go | 7 + .../bucket_index_metadata_fetcher.go | 3 +- .../bucket_index_metadata_fetcher_test.go | 8 + pkg/storegateway/bucket_stores.go | 79 ++- pkg/storegateway/bucket_stores_test.go | 61 ++ pkg/storegateway/hybrid_bucket_stores.go | 576 +++++++++++++++++ pkg/storegateway/hybrid_bucket_stores_test.go | 603 ++++++++++++++++++ pkg/storegateway/metadata_fetcher_filters.go | 50 ++ .../metadata_fetcher_filters_test.go | 68 ++ pkg/storegateway/parquet_bucket_stores.go | 53 +- .../parquet_bucket_stores_test.go | 115 ++-- 15 files changed, 1717 insertions(+), 131 deletions(-) create mode 100644 pkg/storegateway/hybrid_bucket_stores.go create mode 100644 pkg/storegateway/hybrid_bucket_stores_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index f9e2588a717..89377632af5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,7 @@ * [FEATURE] Querier: Add timeout classification to classify query timeouts as 4XX (user error) or 5XX (system error) based on phase timing. When enabled, queries that spend most of their time in PromQL evaluation return `422 Unprocessable Entity` instead of `503 Service Unavailable`. #7374 * [FEATURE] Querier: Implement Resource Based Throttling in Querier. #7442 * [FEATURE] Querier: Add resource-based query eviction that automatically cancels the heaviest running query when CPU or heap utilization exceeds configured thresholds. #7488 +* [FEATURE] Store Gateway: Implement a hybrid mode of the Store Gateway. #7689 * [ENHANCEMENT] Upgrade prometheus alertmanager version to v0.32.1. #7462 * [ENHANCEMENT] Tenant Federation: Avoid purging the regex resolver LRU cache on user-sync ticks when the set of known users has not changed. #7489 * [ENHANCEMENT] Memberlist: Add `-memberlist.packet-read-timeout`, `-memberlist.max-packet-size`, and `-memberlist.max-concurrent-connections` flags to bound inbound gossip TCP connections, preventing slow-read, OOM, and connection-flood attacks on the gossip port. #7518 diff --git a/integration/parquet_querier_test.go b/integration/parquet_querier_test.go index 330e6d908a7..b46e1cf1e2f 100644 --- a/integration/parquet_querier_test.go +++ b/integration/parquet_querier_test.go @@ -7,13 +7,19 @@ import ( "fmt" "path/filepath" "slices" + "sort" "strconv" + "strings" "testing" "time" "github.com/cortexproject/promqlsmith" + "github.com/prometheus-community/parquet-common/convert" "github.com/prometheus/common/model" + "github.com/prometheus/common/promslog" "github.com/prometheus/prometheus/model/labels" + prom_tsdb "github.com/prometheus/prometheus/tsdb" + "github.com/prometheus/prometheus/tsdb/chunkenc" "github.com/stretchr/testify/require" "github.com/thanos-io/objstore" "github.com/thanos-io/thanos/pkg/block" @@ -637,3 +643,206 @@ func TestParquetMultiShardQuery(t *testing.T) { }) } } + +func TestParquetStoreGateway_HybridMode(t *testing.T) { + s, err := e2e.NewScenario(networkName) + require.NoError(t, err) + defer s.Close() + + consul := e2edb.NewConsulWithName("consul") + require.NoError(t, s.StartAndWaitReady(consul)) + + baseFlags := mergeFlags(AlertmanagerLocalFlags(), BlocksStorageFlags()) + flags := mergeFlags(baseFlags, map[string]string{ + // No parquet-converter service: TSDB block will never be auto-converted. + "-target": "all", + "-blocks-storage.tsdb.block-ranges-period": "1m,24h", + "-blocks-storage.tsdb.ship-interval": "1s", + "-blocks-storage.bucket-store.sync-interval": "1s", + "-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl": "1s", + "-blocks-storage.bucket-store.bucket-index.idle-timeout": "1s", + "-blocks-storage.bucket-store.bucket-index.enabled": "true", + // Route reads through the store-gateway Parquet bucket store. + "-blocks-storage.bucket-store.bucket-store-type": "parquet", + "-compactor.cleanup-interval": "1s", + "-ring.store": "consul", + "-consul.hostname": consul.NetworkHTTPEndpoint(), + "-distributor.replication-factor": "1", + "-store-gateway.sharding-enabled": "true", + "-store-gateway.sharding-ring.store": "consul", + "-store-gateway.sharding-ring.consul.hostname": consul.NetworkHTTPEndpoint(), + "-store-gateway.sharding-ring.replication-factor": "1", + "-querier.enable-parquet-queryable": "false", + "-limits.query-ingesters-within": "2h", + "-alertmanager.web.external-url": "http://localhost/alertmanager", + "-parquet-converter.enabled": "true", // enables EnableParquet() in the compactor's bucket index updater + }) + + require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{})) + + const ( + userID = "user-1" + metricParquet = "series_parquet" + metricTSDB = "series_tsdb" + metricMerge = "series_merge" + numSamples = 60 + ) + + ctx := context.Background() + rnd := newFuzzRand(t) + dir := filepath.Join(s.SharedDir(), "data") + scrapeInterval := time.Minute + now := time.Now() + // Both time ranges must be older than -limits.query-ingesters-within (2h). + midPoint := now.Add(-time.Hour * 10) + start := now.Add(-time.Hour * 24) + end := now.Add(-time.Hour * 3) + + // Block A: series_parquet [start, midPoint) — will be converted to Parquet block. + // Also carries two series_merge series (labeled "pk", a Parquet-only label) to verify + // that Series/LabelNames/LabelValues correctly merge results across both blocks. + idA, err := e2e.CreateBlock(ctx, rnd, dir, + []labels.Labels{ + labels.FromStrings(labels.MetricName, metricParquet, "job", "test"), + labels.FromStrings(labels.MetricName, metricMerge, "job", "test", "series", "a", "pk", "1"), + labels.FromStrings(labels.MetricName, metricMerge, "job", "test", "series", "c", "pk", "1"), + }, + numSamples, start.UnixMilli(), midPoint.UnixMilli(), scrapeInterval.Milliseconds(), 10) + require.NoError(t, err) + + // Block B: series_tsdb [midPoint, end) — stays as TSDB block. + // Also carries three series_merge series (labeled "tk", a TSDB-only label). The "series" + // value "c" is shared with block A (under a different label set) to exercise + // de-duplication in the merged LabelValues response. + idB, err := e2e.CreateBlock(ctx, rnd, dir, + []labels.Labels{ + labels.FromStrings(labels.MetricName, metricTSDB, "job", "test"), + labels.FromStrings(labels.MetricName, metricMerge, "job", "test", "series", "b", "tk", "1"), + labels.FromStrings(labels.MetricName, metricMerge, "job", "test", "series", "c", "tk", "1"), + labels.FromStrings(labels.MetricName, metricMerge, "job", "test", "series", "d", "tk", "1"), + }, + numSamples, midPoint.UnixMilli(), end.UnixMilli(), scrapeInterval.Milliseconds(), 10) + require.NoError(t, err) + + 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) + userBkt := bucket.NewUserBucketClient(userID, storage.GetBucket(), nil) + + // Upload both TSDB blocks to object storage. + require.NoError(t, block.Upload(ctx, log.Logger, userBkt, filepath.Join(dir, idA.String()), metadata.NoneFunc)) + require.NoError(t, block.Upload(ctx, log.Logger, userBkt, filepath.Join(dir, idB.String()), metadata.NoneFunc)) + + // Manually convert block A to Parquet in object storage. + { + tsdbBlock, openErr := prom_tsdb.OpenBlock(nil, filepath.Join(dir, idA.String()), chunkenc.NewPool(), prom_tsdb.DefaultPostingsDecoderFactory) + require.NoError(t, openErr) + numShards, convertErr := convert.ConvertTSDBBlock(ctx, userBkt, + tsdbBlock.MinTime(), tsdbBlock.MaxTime(), + []convert.Convertible{tsdbBlock}, promslog.NewNopLogger(), + convert.WithName(idA.String())) + require.NoError(t, tsdbBlock.Close()) + require.NoError(t, convertErr) + require.NoError(t, cortex_parquet.WriteConverterMark(ctx, idA, userBkt, numShards)) + // Upload at the global marker path so the bucket index updater discovers it. + require.NoError(t, userBkt.Upload(ctx, bucketindex.ConverterMarkFilePath(idA), strings.NewReader("{}"))) + } + + cortex := e2ecortex.NewSingleBinary("cortex", flags, "") + require.NoError(t, s.StartAndWaitReady(cortex)) + + c, err := e2ecortex.NewClient("", cortex.HTTPEndpoint(), "", "", userID) + require.NoError(t, err) + + // Wait until the compactor has built the bucket index with the correct state: + // block A must be tagged as Parquet, block B must be TSDB. + cortex_testutil.Poll(t, 60*time.Second, true, func() any { + idx, idxErr := bucketindex.ReadIndex(ctx, storage.GetBucket(), userID, nil, log.Logger) + if idxErr != nil { + return false + } + foundParquetBlock, foundTSDBBlock := false, false + for _, b := range idx.Blocks { + switch b.ID { + case idA: + if b.Parquet != nil { + foundParquetBlock = true + } + case idB: + if b.Parquet == nil { + foundTSDBBlock = true + } + } + } + return foundParquetBlock && foundTSDBBlock + }) + + // Wait until both metrics are queryable via the store-gateway. + cortex_testutil.Poll(t, 120*time.Second, true, func() any { + labelSets, err := c.Series([]string{`{job="test"}`}, start, end) + if err != nil { + return false + } + foundParquet, foundTSDB := false, false + for _, ls := range labelSets { + switch string(ls[model.MetricNameLabel]) { + case metricParquet: + foundParquet = true + case metricTSDB: + foundTSDB = true + } + } + return foundParquet && foundTSDB + }) + + // series_parquet must be served by the Parquet store. + resParquet, err := c.QueryRange(metricParquet, start, midPoint, scrapeInterval) + require.NoError(t, err) + matrixParquet, ok := resParquet.(model.Matrix) + require.True(t, ok) + require.Len(t, matrixParquet, 1, "series_parquet must return one series (served from Parquet store)") + + // series_tsdb must be served by the TSDB store. + resTSDB, err := c.QueryRange(metricTSDB, midPoint, end, scrapeInterval) + require.NoError(t, err) + matrixTSDB, ok := resTSDB.(model.Matrix) + require.True(t, ok) + require.Len(t, matrixTSDB, 1, "series_tsdb must return one series (served from TSDB store)") + + // Series() must return the union of series_merge series from both the Parquet block + // (pk-labeled: a, c) and the TSDB block (tk-labeled: b, c, d). No extra poll is needed here: + // series_merge lives in the same blocks A/B already confirmed synced by the poll above. + mergeMatcher := fmt.Sprintf(`{__name__=%q}`, metricMerge) + mergedSeries, err := c.Series([]string{mergeMatcher}, start, end) + require.NoError(t, err) + require.Len(t, mergedSeries, 5, "series_merge must return the union of series from both stores") + var gotSeriesValues []string + for _, ls := range mergedSeries { + gotSeriesValues = append(gotSeriesValues, string(ls["series"])) + } + sort.Strings(gotSeriesValues) + require.Equal(t, []string{"a", "b", "c", "c", "d"}, gotSeriesValues, + "series_merge series values must include both blocks' series (c appears twice, once per block)") + + // LabelNames() must merge label names from both stores: "pk" only exists in the Parquet + // block, "tk" only in the TSDB block. + names, err := c.LabelNames(start, end, mergeMatcher) + require.NoError(t, err) + require.Equal(t, []string{labels.MetricName, "job", "pk", "series", "tk"}, names, + "LabelNames must merge Parquet-only and TSDB-only label names") + + // LabelValues() must merge and de-duplicate values from both stores: "c" is present in + // both blocks but must appear only once in the merged, sorted result. + values, err := c.LabelValues("series", start, end, []string{mergeMatcher}) + require.NoError(t, err) + require.Equal(t, model.LabelValues{"a", "b", "c", "d"}, values, + "LabelValues must merge and de-duplicate values across both stores") + + // TSDB sub-store must have loaded only block B (loaded=1) and excluded block A (parquet-converted=1). + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_blocks_meta_synced"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "state", "loaded")))) + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_blocks_meta_synced"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "state", "parquet-converted")))) +} diff --git a/pkg/storage/parquet/converter_marker.go b/pkg/storage/parquet/converter_marker.go index 55d85c45b3c..64cfc456ec4 100644 --- a/pkg/storage/parquet/converter_marker.go +++ b/pkg/storage/parquet/converter_marker.go @@ -6,6 +6,7 @@ import ( "encoding/json" "io" "path" + "time" "github.com/efficientgo/core/errors" "github.com/go-kit/log" @@ -32,6 +33,8 @@ type ConverterMark struct { // Shards is the number of parquet shards created for this block. // This field is optional for backward compatibility. Shards int `json:"shards,omitempty"` + // ConvertedAt is the unix timestamp (seconds) when the block was converted. + ConvertedAt int64 `json:"converted_at,omitempty"` } func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*ConverterMark, error) { @@ -58,8 +61,9 @@ func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Instr func WriteConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Bucket, shards int) error { marker := ConverterMark{ - Version: CurrentVersion, - Shards: shards, + Version: CurrentVersion, + Shards: shards, + ConvertedAt: time.Now().Unix(), } markerPath := path.Join(id.String(), ConverterMarkerFileName) b, err := json.Marshal(marker) @@ -75,6 +79,8 @@ type ConverterMarkMeta struct { // Shards is the number of parquet shards created for this block. // This field is optional for backward compatibility. Shards int `json:"shards,omitempty"` + // ConvertedAt is the unix timestamp (seconds) when the block was converted. + ConvertedAt int64 `json:"converted_at,omitempty"` } func ValidConverterMarkVersion(version int) bool { diff --git a/pkg/storage/tsdb/bucketindex/updater.go b/pkg/storage/tsdb/bucketindex/updater.go index 42f5fb1620f..7b9e3d67068 100644 --- a/pkg/storage/tsdb/bucketindex/updater.go +++ b/pkg/storage/tsdb/bucketindex/updater.go @@ -210,8 +210,9 @@ func (w *Updater) updateParquetBlockIndexEntry(ctx context.Context, id ulid.ULID } block.Parquet = &parquet.ConverterMarkMeta{ - Version: marker.Version, - Shards: marker.Shards, + Version: marker.Version, + Shards: marker.Shards, + ConvertedAt: marker.ConvertedAt, } return nil } diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index bd921400803..47d227c50e0 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -448,6 +448,13 @@ func (cfg *BucketStoreConfig) Validate() error { return nil } +// ParquetBlocksGracePeriod is how long the TSDB store keeps syncing a just-converted +// Parquet block before dropping it, avoiding a data gap right after conversion. +// 2 * SyncInterval comfortably covers that window. +func (cfg *BucketStoreConfig) ParquetBlocksGracePeriod() time.Duration { + return 2 * cfg.SyncInterval +} + type BucketIndexConfig struct { Enabled bool `yaml:"enabled"` UpdateOnErrorInterval time.Duration `yaml:"update_on_error_interval"` diff --git a/pkg/storegateway/bucket_index_metadata_fetcher.go b/pkg/storegateway/bucket_index_metadata_fetcher.go index a66e8072482..75a1a4d520b 100644 --- a/pkg/storegateway/bucket_index_metadata_fetcher.go +++ b/pkg/storegateway/bucket_index_metadata_fetcher.go @@ -21,6 +21,7 @@ const ( corruptedBucketIndex = "corrupted-bucket-index" keyAccessDenied = "key-access-denied" noBucketIndex = "no-bucket-index" + parquetConvertedMeta = "parquet-converted" ) // BucketIndexMetadataFetcher is a Thanos MetadataFetcher implementation leveraging on the Cortex bucket index. @@ -50,7 +51,7 @@ func NewBucketIndexMetadataFetcher( cfgProvider: cfgProvider, logger: logger, filters: filters, - metrics: block.NewFetcherMetrics(reg, [][]string{{corruptedBucketIndex}, {noBucketIndex}}, nil), + metrics: block.NewFetcherMetrics(reg, [][]string{{corruptedBucketIndex}, {noBucketIndex}, {parquetConvertedMeta}}, nil), } } diff --git a/pkg/storegateway/bucket_index_metadata_fetcher_test.go b/pkg/storegateway/bucket_index_metadata_fetcher_test.go index 859570f8ce4..db1a0b704e5 100644 --- a/pkg/storegateway/bucket_index_metadata_fetcher_test.go +++ b/pkg/storegateway/bucket_index_metadata_fetcher_test.go @@ -86,6 +86,7 @@ func TestBucketIndexMetadataFetcher_Fetch(t *testing.T) { blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -135,6 +136,7 @@ func TestBucketIndexMetadataFetcher_Fetch_KeyPermissionDenied(t *testing.T) { blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -187,6 +189,7 @@ func TestBucketIndexMetadataFetcher_Fetch_NoBucketIndex(t *testing.T) { blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 1 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -243,6 +246,7 @@ func TestBucketIndexMetadataFetcher_Fetch_CorruptedBucketIndex(t *testing.T) { blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -291,6 +295,7 @@ func TestBucketIndexMetadataFetcher_Fetch_ShouldResetGaugeMetrics(t *testing.T) blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -316,6 +321,7 @@ func TestBucketIndexMetadataFetcher_Fetch_ShouldResetGaugeMetrics(t *testing.T) blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 1 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -349,6 +355,7 @@ func TestBucketIndexMetadataFetcher_Fetch_ShouldResetGaugeMetrics(t *testing.T) blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 @@ -376,6 +383,7 @@ func TestBucketIndexMetadataFetcher_Fetch_ShouldResetGaugeMetrics(t *testing.T) blocks_meta_synced{state="marked-for-no-compact"} 0 blocks_meta_synced{state="no-bucket-index"} 0 blocks_meta_synced{state="no-meta-json"} 0 + blocks_meta_synced{state="parquet-converted"} 0 blocks_meta_synced{state="parquet-migrated"} 0 blocks_meta_synced{state="time-excluded"} 0 blocks_meta_synced{state="too-fresh"} 0 diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go index 20ef4b020e1..b8bea325db2 100644 --- a/pkg/storegateway/bucket_stores.go +++ b/pkg/storegateway/bucket_stores.go @@ -63,6 +63,10 @@ type ThanosBucketStores struct { metaFetcherMetrics *MetadataFetcherMetrics shardingStrategy ShardingStrategy + // ignoreParquetBlocks when true, excludes Parquet-converted blocks from + // syncing so they are not loaded by this TSDB store. + ignoreParquetBlocks bool + // Index cache shared across all tenants. indexCache storecache.IndexCache @@ -109,20 +113,37 @@ var ErrTooManyInflightRequests = status.Error(codes.ResourceExhausted, "too many func NewBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (BucketStores, error) { switch cfg.BucketStore.BucketStoreType { case string(tsdb.ParquetBucketStore): - return newParquetBucketStores(cfg, bucketClient, limits, logger, reg) + return newHybridBucketStores(cfg, shardingStrategy, bucketClient, limits, logLevel, logger, reg) case string(tsdb.TSDBBucketStore): - return newThanosBucketStores(cfg, shardingStrategy, bucketClient, limits, logLevel, logger, reg) + return newThanosBucketStores(cfg, shardingStrategy, bucketClient, nil, nil, false, limits, logLevel, logger, reg) default: return nil, fmt.Errorf("unsupported bucket store type: %s", cfg.BucketStore.BucketStoreType) } } +func newMatchersCache(cfg tsdb.BlocksStorageConfig, logger log.Logger, reg prometheus.Registerer) (storecache.MatchersCache, error) { + if cfg.BucketStore.MatchersCacheMaxItems <= 0 { + return storecache.NoopMatchersCache, nil + } + r := prometheus.NewRegistry() + reg.MustRegister(tsdb.NewMatchCacheMetrics("cortex_storegateway", r, logger)) + return storecache.NewMatchersCache(storecache.WithSize(cfg.BucketStore.MatchersCacheMaxItems), storecache.WithPromRegistry(r)) +} + // newThanosBucketStores creates a new TSDB-based bucket stores -func newThanosBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (*ThanosBucketStores, error) { - matchers := tsdb.NewMatchers() - cachingBucket, err := tsdb.CreateCachingBucket(cfg.BucketStore.ChunksCache, cfg.BucketStore.MetadataCache, tsdb.ParquetLabelsCacheConfig{}, matchers, bucketClient, logger, reg) - if err != nil { - return nil, errors.Wrapf(err, "create caching bucket") +func newThanosBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, cachingBucket objstore.InstrumentedBucket, matcherCache storecache.MatchersCache, ignoreParquetBlocks bool, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (*ThanosBucketStores, error) { + var err error + if cachingBucket == nil { + matchers := tsdb.NewMatchers() + cachingBucket, err = tsdb.CreateCachingBucket(cfg.BucketStore.ChunksCache, cfg.BucketStore.MetadataCache, tsdb.ParquetLabelsCacheConfig{}, matchers, bucketClient, logger, reg) + if err != nil { + return nil, errors.Wrapf(err, "create caching bucket") + } + } + if matcherCache == nil { + if matcherCache, err = newMatchersCache(cfg, logger, reg); err != nil { + return nil, err + } } // The number of concurrent queries against the tenants BucketStores are limited. @@ -134,20 +155,21 @@ func newThanosBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy Shardi }).Set(float64(cfg.BucketStore.MaxConcurrent)) u := &ThanosBucketStores{ - logger: logger, - cfg: cfg, - limits: limits, - bucket: cachingBucket, - shardingStrategy: shardingStrategy, - stores: map[string]*store.BucketStore{}, - storesErrors: map[string]error{}, - logLevel: logLevel, - bucketStoreMetrics: NewBucketStoreMetrics(), - metaFetcherMetrics: NewMetadataFetcherMetrics(), - queryGate: queryGate, - partitioner: newGapBasedPartitioner(cfg.BucketStore.PartitionerMaxGapBytes, reg), - userTokenBuckets: make(map[string]*util.TokenBucket), - inflightRequests: util.NewInflightRequestTracker(), + logger: logger, + cfg: cfg, + limits: limits, + bucket: cachingBucket, + shardingStrategy: shardingStrategy, + ignoreParquetBlocks: ignoreParquetBlocks, + stores: map[string]*store.BucketStore{}, + storesErrors: map[string]error{}, + logLevel: logLevel, + bucketStoreMetrics: NewBucketStoreMetrics(), + metaFetcherMetrics: NewMetadataFetcherMetrics(), + queryGate: queryGate, + partitioner: newGapBasedPartitioner(cfg.BucketStore.PartitionerMaxGapBytes, reg), + userTokenBuckets: make(map[string]*util.TokenBucket), + inflightRequests: util.NewInflightRequestTracker(), syncTimes: promauto.With(reg).NewHistogram(prometheus.HistogramOpts{ Name: "cortex_bucket_stores_blocks_sync_seconds", Help: "The total time it takes to perform a sync stores", @@ -174,16 +196,7 @@ func newThanosBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy Shardi return nil, errors.Wrap(err, "failed to create users scanner") } - u.matcherCache = storecache.NoopMatchersCache - - if cfg.BucketStore.MatchersCacheMaxItems > 0 { - r := prometheus.NewRegistry() - reg.MustRegister(tsdb.NewMatchCacheMetrics("cortex_storegateway", r, logger)) - u.matcherCache, err = storecache.NewMatchersCache(storecache.WithSize(cfg.BucketStore.MatchersCacheMaxItems), storecache.WithPromRegistry(r)) - if err != nil { - return nil, err - } - } + u.matcherCache = matcherCache // Init the index cache. if u.indexCache, err = tsdb.NewIndexCache(cfg.BucketStore.IndexCache, logger, reg); err != nil { @@ -604,6 +617,10 @@ func (u *ThanosBucketStores) getOrCreateStore(userID string) (*store.BucketStore filters = append(filters, NewIgnoreNonQueryableBlocksFilter(userLogger, u.cfg.BucketStore.IgnoreBlocksWithin)) } + if u.ignoreParquetBlocks { + filters = append(filters, NewIgnoreParquetBlocksFilter(userLogger, u.cfg.BucketStore.ParquetBlocksGracePeriod())) + } + // Instantiate a different blocks metadata fetcher based on whether bucket index is enabled or not. var fetcher block.MetadataFetcher if u.cfg.BucketStore.BucketIndex.Enabled { diff --git a/pkg/storegateway/bucket_stores_test.go b/pkg/storegateway/bucket_stores_test.go index 4bf3e7af065..60c0da2a849 100644 --- a/pkg/storegateway/bucket_stores_test.go +++ b/pkg/storegateway/bucket_stores_test.go @@ -803,6 +803,67 @@ func queryLabelsValues(stores BucketStores, userID, labelName, metricName string return stores.LabelValues(ctx, req) } +func queryLabelsNamesWithBlocks(stores BucketStores, userID, metricName string, start, end int64, blockIDs ...string) (*storepb.LabelNamesResponse, error) { + req := &storepb.LabelNamesRequest{ + Start: start, + End: end, + Matchers: []storepb.LabelMatcher{{ + Type: storepb.LabelMatcher_EQ, + Name: labels.MetricName, + Value: metricName, + }}, + PartialResponseStrategy: storepb.PartialResponseStrategy_ABORT, + } + if len(blockIDs) > 0 { + hints := &hintspb.LabelNamesRequestHints{ + BlockMatchers: []storepb.LabelMatcher{{ + Type: storepb.LabelMatcher_RE, + Name: block.BlockIDLabel, + Value: strings.Join(blockIDs, "|"), + }}, + } + anyHints, err := types.MarshalAny(hints) + if err != nil { + return nil, err + } + req.Hints = anyHints + } + + ctx := setUserIDToGRPCContext(context.Background(), userID) + return stores.LabelNames(ctx, req) +} + +func queryLabelsValuesWithBlocks(stores BucketStores, userID, labelName, metricName string, start, end int64, blockIDs ...string) (*storepb.LabelValuesResponse, error) { + req := &storepb.LabelValuesRequest{ + Start: start, + End: end, + Label: labelName, + Matchers: []storepb.LabelMatcher{{ + Type: storepb.LabelMatcher_EQ, + Name: labels.MetricName, + Value: metricName, + }}, + PartialResponseStrategy: storepb.PartialResponseStrategy_ABORT, + } + if len(blockIDs) > 0 { + hints := &hintspb.LabelValuesRequestHints{ + BlockMatchers: []storepb.LabelMatcher{{ + Type: storepb.LabelMatcher_RE, + Name: block.BlockIDLabel, + Value: strings.Join(blockIDs, "|"), + }}, + } + anyHints, err := types.MarshalAny(hints) + if err != nil { + return nil, err + } + req.Hints = anyHints + } + + ctx := setUserIDToGRPCContext(context.Background(), userID) + return stores.LabelValues(ctx, req) +} + func mockLoggingLevel() logging.Level { level := logging.Level{} err := level.Set("info") diff --git a/pkg/storegateway/hybrid_bucket_stores.go b/pkg/storegateway/hybrid_bucket_stores.go new file mode 100644 index 00000000000..5ab2d81739e --- /dev/null +++ b/pkg/storegateway/hybrid_bucket_stores.go @@ -0,0 +1,576 @@ +package storegateway + +import ( + "context" + "fmt" + "strings" + "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" + parquet_util "github.com/prometheus-community/parquet-common/util" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/prometheus/model/labels" + "github.com/thanos-io/objstore" + "github.com/thanos-io/thanos/pkg/block" + "github.com/thanos-io/thanos/pkg/store/hintspb" + "github.com/thanos-io/thanos/pkg/store/labelpb" + "github.com/thanos-io/thanos/pkg/store/storepb" + "github.com/weaveworks/common/httpgrpc" + "github.com/weaveworks/common/logging" + "github.com/weaveworks/common/user" + "golang.org/x/sync/errgroup" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/cortexproject/cortex/pkg/storage/bucket" + cortex_parquet "github.com/cortexproject/cortex/pkg/storage/parquet" + "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/multierror" + "github.com/cortexproject/cortex/pkg/util/spanlogger" + "github.com/cortexproject/cortex/pkg/util/validation" +) + +type HybridBucketStores struct { + logger log.Logger + cfg tsdb.BlocksStorageConfig + limits *validation.Overrides + bucket objstore.Bucket + + inflightRequests *cortex_util.InflightRequestTracker + + // indexLoader is the bucket index used to classify blocks as Parquet or TSDB. + indexLoader *bucketindex.Loader + + // parquetBlocksCache caches, per tenant, the set of Parquet-converted block IDs derived + // from the bucket index. + parquetBlocksCacheMu sync.Mutex + parquetBlocksCache map[string]parquetBlocksCacheEntry + + parquet *ParquetBucketStores + tsdb *ThanosBucketStores +} + +// parquetBlocksCacheEntry is a cached snapshot of the Parquet-converted block IDs for a tenant, +// valid for a specific bucket index version (updatedAt). +type parquetBlocksCacheEntry struct { + updatedAt int64 + parquetSet map[string]struct{} +} + +func newHybridBucketStores(cfg tsdb.BlocksStorageConfig, shardingStrategy ShardingStrategy, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logLevel logging.Level, logger log.Logger, reg prometheus.Registerer) (*HybridBucketStores, error) { + cachingBucket, err := createCachingBucketClientForParquet(cfg, bucketClient, "parquet-storegateway", logger, reg) + if err != nil { + return nil, err + } + matcherCache, err := newMatchersCache(cfg, logger, reg) + if err != nil { + return nil, err + } + + // The TSDB store only syncs blocks not yet converted to Parquet (when the bucket index + // is enabled, IgnoreParquetBlocksFilter excludes converted blocks). + tsdbStore, err := newThanosBucketStores(cfg, shardingStrategy, bucketClient, cachingBucket, matcherCache, cfg.BucketStore.BucketIndex.Enabled, limits, logLevel, logger, reg) + if err != nil { + return nil, errors.Wrap(err, "create TSDB store for hybrid bucket stores") + } + + parquetStore, err := newParquetBucketStores(cfg, bucketClient, cachingBucket, matcherCache, limits, logger, reg) + if err != nil { + return nil, errors.Wrap(err, "create parquet store for hybrid bucket stores") + } + + return &HybridBucketStores{ + logger: logger, + cfg: cfg, + limits: limits, + bucket: cachingBucket, + inflightRequests: cortex_util.NewInflightRequestTracker(), + indexLoader: parquetStore.indexLoader, + parquetBlocksCache: map[string]parquetBlocksCacheEntry{}, + parquet: parquetStore, + tsdb: tsdbStore, + }, nil +} + +// Series implements BucketStores. +func (h *HybridBucketStores) Series(req *storepb.SeriesRequest, srv storepb.Store_SeriesServer) error { + spanLog, spanCtx := spanlogger.New(srv.Context(), "HybridBucketStores.Series") + defer spanLog.Finish() + + userID := getUserIDFromGRPCContext(spanCtx) + if userID == "" { + return fmt.Errorf("no userID") + } + spanCtx = user.InjectOrgID(spanCtx, userID) + + if err := h.checkStoreError(userID); err != nil { + return err + } + + store, err := h.parquet.getOrCreateStore(userID) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if maxInflightRequests := h.cfg.BucketStore.MaxInflightRequests; maxInflightRequests > 0 { + if h.inflightRequests.Count() >= maxInflightRequests { + return ErrTooManyInflightRequests + } + h.inflightRequests.Inc() + defer h.inflightRequests.Dec() + } + + wrappedSrv := spanSeriesServer{ + Store_SeriesServer: srv, + ctx: spanCtx, + } + return h.seriesWithTSDBStore(spanCtx, userID, req, store, wrappedSrv) +} + +// LabelNames implements BucketStores. +func (h *HybridBucketStores) LabelNames(ctx context.Context, req *storepb.LabelNamesRequest) (*storepb.LabelNamesResponse, error) { + spanLog, spanCtx := spanlogger.New(ctx, "HybridBucketStores.LabelNames") + defer spanLog.Finish() + + userID := getUserIDFromGRPCContext(spanCtx) + if userID == "" { + return nil, fmt.Errorf("no userID") + } + spanCtx = user.InjectOrgID(spanCtx, userID) + + if err := h.checkStoreError(userID); err != nil { + return nil, err + } + + store, err := h.parquet.getOrCreateStore(userID) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + + return h.labelNamesWithTSDBStore(spanCtx, userID, req, store) +} + +// LabelValues implements BucketStores. +func (h *HybridBucketStores) LabelValues(ctx context.Context, req *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) { + spanLog, spanCtx := spanlogger.New(ctx, "HybridBucketStores.LabelValues") + defer spanLog.Finish() + + userID := getUserIDFromGRPCContext(spanCtx) + if userID == "" { + return nil, fmt.Errorf("no userID") + } + spanCtx = user.InjectOrgID(spanCtx, userID) + + if err := h.checkStoreError(userID); err != nil { + return nil, err + } + + store, err := h.parquet.getOrCreateStore(userID) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + + return h.labelValuesWithTSDBStore(spanCtx, userID, req, store) +} + +func (h *HybridBucketStores) checkStoreError(userID string) error { + err := h.parquet.getStoreError(userID) + if err == nil { + return nil + } + userBkt := bucket.NewUserBucketClient(userID, h.bucket, h.limits) + if cortex_errors.ErrorIs(err, userBkt.IsAccessDeniedErr) { + return httpgrpc.Errorf(int(codes.PermissionDenied), "store error: %s", err) + } + return err +} + +// SyncBlocks implements BucketStores +func (h *HybridBucketStores) SyncBlocks(ctx context.Context) error { + return h.tsdb.SyncBlocks(ctx) +} + +// InitialSync implements BucketStores +func (h *HybridBucketStores) InitialSync(ctx context.Context) error { + if err := h.parquet.InitialSync(ctx); err != nil { + return err + } + return h.tsdb.InitialSync(ctx) +} + +// Stop implements BucketStores +func (h *HybridBucketStores) Stop() error { + return multierror.New(h.parquet.Stop(), h.tsdb.Stop()).Err() +} + +// splitRequestedBlocks splits the block IDs encoded in the request block matchers into +// Parquet-converted and not-yet-converted (TSDB) groups. +// +// The classification source depends on whether the bucket index is enabled: +// - when enabled, the per-block Parquet converter mark recorded in the bucket index is used. +// - when disabled, the per-block converter mark object is read directly from the bucket. +func (h *HybridBucketStores) splitRequestedBlocks(ctx context.Context, userID string, blockMatchers []storepb.LabelMatcher) (parquetIDs, tsdbIDs []string, err error) { + if len(blockMatchers) != 1 || blockMatchers[0].Type != storepb.LabelMatcher_RE || blockMatchers[0].Name != block.BlockIDLabel { + return nil, nil, status.Error(codes.InvalidArgument, "only one block matcher is supported") + } + + blockIDs := strings.Split(blockMatchers[0].Value, "|") + filtered := blockIDs[:0] + for _, id := range blockIDs { + if id != "" { + filtered = append(filtered, id) + } + } + blockIDs = filtered + + isParquet, err := h.parquetBlocks(ctx, userID, blockIDs) + if err != nil { + return nil, nil, err + } + + for _, id := range blockIDs { + if isParquet[id] { + parquetIDs = append(parquetIDs, id) + } else { + tsdbIDs = append(tsdbIDs, id) + } + } + + return parquetIDs, tsdbIDs, nil +} + +// parquetBlocks returns, for each requested block ID, whether it has been converted to Parquet. +// +// When the bucket index is available it is used as the source of truth. +// If the bucket index is disabled or cannot be loaded, it falls back to reading the per-block converter mark directly. +func (h *HybridBucketStores) parquetBlocks(ctx context.Context, userID string, blockIDs []string) (map[string]bool, error) { + if h.indexLoader != nil { + idx, _, err := h.indexLoader.GetIndex(ctx, userID) + if err == nil { + parquetSet := h.parquetSetFromIndex(userID, idx) + result := make(map[string]bool, len(blockIDs)) + for _, id := range blockIDs { + _, ok := parquetSet[id] + result[id] = ok + } + return result, nil + } + level.Warn(h.logger).Log("msg", "failed to get bucket index, falling back to converter marks for parquet classification", "user", userID, "err", err) + } + + return h.parquetBlocksFromConverterMarks(ctx, userID, blockIDs) +} + +// parquetSetFromIndex returns the set of Parquet-converted block IDs for the tenant, built from +// the bucket index. The result is cached and rebuilt only when the index version (UpdatedAt) +// changes. +func (h *HybridBucketStores) parquetSetFromIndex(userID string, idx *bucketindex.Index) map[string]struct{} { + h.parquetBlocksCacheMu.Lock() + defer h.parquetBlocksCacheMu.Unlock() + + if e, ok := h.parquetBlocksCache[userID]; ok && e.updatedAt == idx.UpdatedAt { + return e.parquetSet + } + + parquetSet := make(map[string]struct{}, len(idx.Blocks)) + for _, b := range idx.Blocks { + if b.Parquet != nil { + parquetSet[b.ID.String()] = struct{}{} + } + } + + h.parquetBlocksCache[userID] = parquetBlocksCacheEntry{updatedAt: idx.UpdatedAt, parquetSet: parquetSet} + return parquetSet +} + +// parquetBlocksFromConverterMarks classifies each block by reading its converter mark directly. +// A missing mark (Version == 0) means the block has not been converted to Parquet yet. +func (h *HybridBucketStores) parquetBlocksFromConverterMarks(ctx context.Context, userID string, blockIDs []string) (map[string]bool, error) { + result := make(map[string]bool, len(blockIDs)) + userBkt := bucket.NewUserBucketClient(userID, h.bucket, h.limits) + for _, id := range blockIDs { + uid, err := ulid.Parse(id) + if err != nil { + return nil, errors.Wrapf(err, "failed to parse block ID %s", id) + } + marker, err := cortex_parquet.ReadConverterMark(ctx, uid, userBkt, h.logger) + if err != nil { + return nil, errors.Wrapf(err, "failed to read converter mark for block %s", id) + } + result[id] = marker.Version > 0 + } + + return result, nil +} + +// blockIDsMatcher builds the single regex block matcher understood by the stores. +func blockIDsMatcher(blockIDs []string) storepb.LabelMatcher { + return storepb.LabelMatcher{ + Type: storepb.LabelMatcher_RE, + Name: block.BlockIDLabel, + Value: strings.Join(blockIDs, "|"), + } +} + +// sliceSeriesSet implements storepb.SeriesSet over a slice of already-sorted series. +type sliceSeriesSet struct { + series []*storepb.Series + i int +} + +func newSliceSeriesSet(series []*storepb.Series) *sliceSeriesSet { + return &sliceSeriesSet{series: series, i: -1} +} + +func (s *sliceSeriesSet) Next() bool { + s.i++ + return s.i < len(s.series) +} + +func (s *sliceSeriesSet) At() (labels.Labels, []storepb.AggrChunk) { + return s.series[s.i].PromLabels(), s.series[s.i].Chunks +} + +func (s *sliceSeriesSet) Err() error { return nil } + +// seriesWithTSDBStore serves a Series request by splitting the requested blocks between the +// Parquet store and the TSDB store, then merging the (sorted) results. +func (h *HybridBucketStores) seriesWithTSDBStore(ctx context.Context, userID string, req *storepb.SeriesRequest, store *parquetBucketStore, srv storepb.Store_SeriesServer) error { + var blockMatchers []storepb.LabelMatcher + if req.Hints != nil { + reqHints := &hintspb.SeriesRequestHints{} + if err := types.UnmarshalAny(req.Hints, reqHints); err != nil { + return status.Error(codes.InvalidArgument, errors.Wrap(err, "unmarshal series request hints").Error()) + } + blockMatchers = reqHints.BlockMatchers + } + + parquetIDs, tsdbIDs, err := h.splitRequestedBlocks(ctx, userID, blockMatchers) + if err != nil { + return err + } + + // If only one store has blocks to serve, delegate directly without buffering/merging. + switch { + case len(tsdbIDs) == 0: + return store.Series(req, srv) + case len(parquetIDs) == 0: + return h.tsdb.Series(req, srv) + } + + // Both stores must be buffered so we can sort-merge their outputs before sending. + parquetSrv := newBucketStoreSeriesServer(ctx) + tsdbSrv := newBucketStoreSeriesServer(ctx) + + g, _ := errgroup.WithContext(ctx) + g.Go(func() error { return store.Series(rewriteSeriesReqBlocks(req, parquetIDs), parquetSrv) }) + g.Go(func() error { return h.tsdb.Series(rewriteSeriesReqBlocks(req, tsdbIDs), tsdbSrv) }) + if err := g.Wait(); err != nil { + return err + } + + // Both stores return series sorted by labels; MergeSeriesSets unions them and merges + // chunks of series that exist in both (e.g. the same series across a parquet and a TSDB block). + merged := storepb.MergeSeriesSets(newSliceSeriesSet(parquetSrv.SeriesSet), newSliceSeriesSet(tsdbSrv.SeriesSet)) + for merged.Next() { + lset, chks := merged.At() + if err := srv.Send(storepb.NewSeriesResponse(&storepb.Series{ + Labels: labelpb.ZLabelsFromPromLabels(lset), + Chunks: chks, + })); err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "send merged series response").Error()) + } + } + if err := merged.Err(); err != nil { + return err + } + + resHints := hintspb.SeriesResponseHints{ + QueriedBlocks: append(parquetSrv.Hints.QueriedBlocks, tsdbSrv.Hints.QueriedBlocks...), + } + anyHints, err := types.MarshalAny(&resHints) + if err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "marshal series response hints").Error()) + } + if err := srv.Send(storepb.NewHintsSeriesResponse(anyHints)); err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "send series response hints").Error()) + } + + return nil +} + +func rewriteSeriesReqBlocks(req *storepb.SeriesRequest, blockIDs []string) *storepb.SeriesRequest { + reqHints := &hintspb.SeriesRequestHints{} + if req.Hints != nil { + // Best effort: ignore unmarshal error, we always overwrite the block matchers below. + _ = types.UnmarshalAny(req.Hints, reqHints) + } + reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} + + clone := *req + if anyHints, err := types.MarshalAny(reqHints); err == nil { + clone.Hints = anyHints + } + return &clone +} + +// labelNamesWithTSDBStore serves a LabelNames request across the Parquet store and the TSDB store. +func (h *HybridBucketStores) labelNamesWithTSDBStore(ctx context.Context, userID string, req *storepb.LabelNamesRequest, store *parquetBucketStore) (*storepb.LabelNamesResponse, error) { + var blockMatchers []storepb.LabelMatcher + if req.Hints != nil { + reqHints := &hintspb.LabelNamesRequestHints{} + if err := types.UnmarshalAny(req.Hints, reqHints); err != nil { + return nil, status.Error(codes.InvalidArgument, errors.Wrap(err, "unmarshal label names request hints").Error()) + } + blockMatchers = reqHints.BlockMatchers + } + + parquetIDs, tsdbIDs, err := h.splitRequestedBlocks(ctx, userID, blockMatchers) + if err != nil { + return nil, err + } + + switch { + case len(tsdbIDs) == 0: + return store.LabelNames(ctx, req) + case len(parquetIDs) == 0: + return h.tsdb.LabelNames(ctx, req) + } + + var ( + parquetResp, tsdbResp *storepb.LabelNamesResponse + ) + g, gCtx := errgroup.WithContext(ctx) + g.Go(func() error { + r, err := store.LabelNames(gCtx, rewriteLabelNamesReqBlocks(req, parquetIDs)) + parquetResp = r + return err + }) + g.Go(func() error { + r, err := h.tsdb.LabelNames(gCtx, rewriteLabelNamesReqBlocks(req, tsdbIDs)) + tsdbResp = r + return err + }) + if err := g.Wait(); err != nil { + return nil, err + } + + names := parquet_util.MergeUnsortedSlices(int(req.Limit), parquetResp.Names, tsdbResp.Names) + anyHints, err := mergeLabelNamesHints(parquetResp, tsdbResp) + if err != nil { + return nil, err + } + return &storepb.LabelNamesResponse{Names: names, Hints: anyHints}, nil +} + +func rewriteLabelNamesReqBlocks(req *storepb.LabelNamesRequest, blockIDs []string) *storepb.LabelNamesRequest { + reqHints := &hintspb.LabelNamesRequestHints{} + if req.Hints != nil { + _ = types.UnmarshalAny(req.Hints, reqHints) + } + reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} + + clone := *req + if anyHints, err := types.MarshalAny(reqHints); err == nil { + clone.Hints = anyHints + } + return &clone +} + +func mergeLabelNamesHints(a, b *storepb.LabelNamesResponse) (*types.Any, error) { + merged := &hintspb.LabelNamesResponseHints{} + for _, resp := range []*storepb.LabelNamesResponse{a, b} { + if resp == nil || resp.Hints == nil { + continue + } + hints := hintspb.LabelNamesResponseHints{} + if err := types.UnmarshalAny(resp.Hints, &hints); err != nil { + return nil, errors.Wrap(err, "unmarshal label names response hints") + } + merged.QueriedBlocks = append(merged.QueriedBlocks, hints.QueriedBlocks...) + } + return types.MarshalAny(merged) +} + +// labelValuesWithTSDBStore serves a LabelValues request across the Parquet store and the TSDB store. +func (h *HybridBucketStores) labelValuesWithTSDBStore(ctx context.Context, userID string, req *storepb.LabelValuesRequest, store *parquetBucketStore) (*storepb.LabelValuesResponse, error) { + var blockMatchers []storepb.LabelMatcher + if req.Hints != nil { + reqHints := &hintspb.LabelValuesRequestHints{} + if err := types.UnmarshalAny(req.Hints, reqHints); err != nil { + return nil, status.Error(codes.InvalidArgument, errors.Wrap(err, "unmarshal label values request hints").Error()) + } + blockMatchers = reqHints.BlockMatchers + } + + parquetIDs, tsdbIDs, err := h.splitRequestedBlocks(ctx, userID, blockMatchers) + if err != nil { + return nil, err + } + + switch { + case len(tsdbIDs) == 0: + return store.LabelValues(ctx, req) + case len(parquetIDs) == 0: + return h.tsdb.LabelValues(ctx, req) + } + + var parquetResp, tsdbResp *storepb.LabelValuesResponse + g, gCtx := errgroup.WithContext(ctx) + g.Go(func() error { + r, err := store.LabelValues(gCtx, rewriteLabelValuesReqBlocks(req, parquetIDs)) + parquetResp = r + return err + }) + g.Go(func() error { + r, err := h.tsdb.LabelValues(gCtx, rewriteLabelValuesReqBlocks(req, tsdbIDs)) + tsdbResp = r + return err + }) + if err := g.Wait(); err != nil { + return nil, err + } + + values := parquet_util.MergeUnsortedSlices(int(req.Limit), parquetResp.Values, tsdbResp.Values) + anyHints, err := mergeLabelValuesHints(parquetResp, tsdbResp) + if err != nil { + return nil, err + } + return &storepb.LabelValuesResponse{Values: values, Hints: anyHints}, nil +} + +func rewriteLabelValuesReqBlocks(req *storepb.LabelValuesRequest, blockIDs []string) *storepb.LabelValuesRequest { + reqHints := &hintspb.LabelValuesRequestHints{} + if req.Hints != nil { + _ = types.UnmarshalAny(req.Hints, reqHints) + } + reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} + + clone := *req + if anyHints, err := types.MarshalAny(reqHints); err == nil { + clone.Hints = anyHints + } + return &clone +} + +func mergeLabelValuesHints(a, b *storepb.LabelValuesResponse) (*types.Any, error) { + merged := &hintspb.LabelValuesResponseHints{} + for _, resp := range []*storepb.LabelValuesResponse{a, b} { + if resp == nil || resp.Hints == nil { + continue + } + hints := hintspb.LabelValuesResponseHints{} + if err := types.UnmarshalAny(resp.Hints, &hints); err != nil { + return nil, errors.Wrap(err, "unmarshal label values response hints") + } + merged.QueriedBlocks = append(merged.QueriedBlocks, hints.QueriedBlocks...) + } + return types.MarshalAny(merged) +} diff --git a/pkg/storegateway/hybrid_bucket_stores_test.go b/pkg/storegateway/hybrid_bucket_stores_test.go new file mode 100644 index 00000000000..fcd0f086183 --- /dev/null +++ b/pkg/storegateway/hybrid_bucket_stores_test.go @@ -0,0 +1,603 @@ +package storegateway + +import ( + "bytes" + "context" + "fmt" + "math" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/go-kit/log" + ulidv2 "github.com/oklog/ulid/v2" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/prometheus/prometheus/model/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/thanos-io/objstore" + + "github.com/cortexproject/cortex/pkg/storage/bucket" + "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/validation" +) + +func TestHybridBucketStores_Series_ShouldReturnErrorIfMaxInflightRequestIsReached(t *testing.T) { + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + cfg.BucketStore.MaxInflightRequests = 10 + reg := prometheus.NewPedanticRegistry() + storageDir := t.TempDir() + generateStorageBlock(t, storageDir, "user_id", "series_1", 0, 100, 15) + bucket, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) + require.NoError(t, err) + + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bucket), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + hybridStores := stores.(*HybridBucketStores) + // Set inflight requests to the limit + for range 10 { + hybridStores.inflightRequests.Inc() + } + series, warnings, err := querySeries(stores, "user_id", "series_1", 0, 100) + assert.ErrorIs(t, err, ErrTooManyInflightRequests) + assert.Empty(t, series) + assert.Empty(t, warnings) +} + +func TestHybridBucketStores_Series_ShouldNotCheckMaxInflightRequestsIfTheLimitIsDisabled(t *testing.T) { + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + reg := prometheus.NewPedanticRegistry() + storageDir := t.TempDir() + userId := "user_id" + generateStorageBlock(t, storageDir, userId, "series_1", 0, 100, 15) + bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) + require.NoError(t, err) + + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + hybridStores := stores.(*HybridBucketStores) + // Set inflight requests to the limit (max_inflight_request is set to 0 by default = disabled) + for range 10 { + hybridStores.inflightRequests.Inc() + } + + userPath := fmt.Sprintf("%s/%s", storageDir, userId) + + limits := validation.Limits{} + overrides := validation.NewOverrides(limits, nil) + uBucket := bucket.NewUserBucketClient(userId, bkt, overrides) + blockIds, err := convertToParquetBlocksForTesting(userPath, uBucket) + require.NoError(t, err) + + series, _, err := querySeries(stores, userId, "series_1", 0, 100, blockIds...) + require.NoError(t, err) + assert.Equal(t, 1, len(series)) +} + +// TestHybridBucketStores_SharesCaches verifies that, in Parquet mode, the Parquet +// store and its TSDB fallback share the caching bucket and the matchers cache. +func TestHybridBucketStores_SharesCaches(t *testing.T) { + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + // Enable the caches that both stores would otherwise register independently. + cfg.BucketStore.ChunksCache.Backend = cortex_tsdb.CacheBackendInMemory + cfg.BucketStore.MetadataCache.Backend = cortex_tsdb.CacheBackendInMemory + cfg.BucketStore.MatchersCacheMaxItems = 100 + + bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: t.TempDir()}) + require.NoError(t, err) + + // A pedantic registry makes any duplicate metric registration fail. + reg := prometheus.NewPedanticRegistry() + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NotNil(t, stores) +} + +// setupHybridParquetTSDB creates one Parquet-converted block ([0,100) with metricParquet) and one +// plain TSDB block ([100,200) with metricTSDB) in Parquet store-gateway mode, then builds and +// initial-syncs the hybrid stores. When bucketIndexEnabled is true it also writes a bucket index +// recording the Parquet conversion (converted long ago, so the grace period does not apply). +func setupHybridParquetTSDB(t *testing.T, userID string, bucketIndexEnabled bool, metricParquet, metricTSDB string) (BucketStores, *prometheus.Registry, []string) { + t.Helper() + + storageDir := t.TempDir() + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + cfg.BucketStore.BucketIndex.Enabled = bucketIndexEnabled + + // Parquet block [0,100): create alone so only this block is converted. + generateStorageBlock(t, storageDir, userID, metricParquet, 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) + userPath := filepath.Join(storageDir, userID) + + parquetBlockIDs, err := convertToParquetBlocksForTesting(userPath, uBucket) + require.NoError(t, err) + require.Len(t, parquetBlockIDs, 1) + + // TSDB block [100,200): not converted. + generateStorageBlock(t, storageDir, userID, metricTSDB, 100, 200, 15) + + entries, err := os.ReadDir(userPath) + require.NoError(t, err) + var tsdbBlockIDs []string + for _, e := range entries { + if _, parseErr := ulidv2.Parse(e.Name()); parseErr == nil && e.Name() != parquetBlockIDs[0] { + tsdbBlockIDs = append(tsdbBlockIDs, e.Name()) + } + } + require.Len(t, tsdbBlockIDs, 1) + allBlockIDs := append(parquetBlockIDs, tsdbBlockIDs...) + + // Build bucket index with Parquet info so IgnoreParquetBlocksFilter works. + if bucketIndexEnabled { + parquetUID, parseErr := ulidv2.Parse(parquetBlockIDs[0]) + require.NoError(t, parseErr) + require.NoError(t, uBucket.Upload(context.Background(), bucketindex.ConverterMarkFilePath(parquetUID), bytes.NewReader([]byte("{}")))) + idx, _, _, idxErr := bucketindex.NewUpdater(bkt, userID, nil, log.NewNopLogger()).EnableParquet().UpdateIndex(context.Background(), nil) + require.NoError(t, idxErr) + // Mark the block as converted long ago so the grace period does not keep it in the TSDB + // store: these tests verify the immediate-drop (steady-state) path. + for _, b := range idx.Blocks { + if b.Parquet != nil { + b.Parquet.ConvertedAt = time.Now().Add(-24 * time.Hour).Unix() + } + } + require.NoError(t, bucketindex.WriteIndex(context.Background(), bkt, userID, nil, idx)) + } + + reg := prometheus.NewRegistry() + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + return stores, reg, allBlockIDs +} + +// TestHybridBucketStores_Series verifies that with one Parquet-converted block and one plain TSDB +// block, both are served correctly in Parquet store-gateway mode (bucket index enabled or +// disabled), and that the same series spanning both stores is merged into one. +func TestHybridBucketStores_Series(t *testing.T) { + const userID = "user-1" + + assertBothServed := func(t *testing.T, stores BucketStores, allBlockIDs []string) { + t.Helper() + seriesFromParquet, _, err := querySeries(stores, userID, "series_parquet", 0, 100, allBlockIDs...) + require.NoError(t, err) + require.Len(t, seriesFromParquet, 1, "series from Parquet-converted block must be returned") + + seriesFromTSDB, _, err := querySeries(stores, userID, "series_tsdb", 100, 200, allBlockIDs...) + require.NoError(t, err) + require.Len(t, seriesFromTSDB, 1, "series from TSDB block must be returned") + } + + t.Run("bucket index enabled: parquet and tsdb blocks both served", func(t *testing.T) { + stores, _, allBlockIDs := setupHybridParquetTSDB(t, userID, true, "series_parquet", "series_tsdb") + assertBothServed(t, stores, allBlockIDs) + }) + + t.Run("bucket index disabled: parquet and tsdb blocks both served", func(t *testing.T) { + stores, _, allBlockIDs := setupHybridParquetTSDB(t, userID, false, "series_parquet", "series_tsdb") + assertBothServed(t, stores, allBlockIDs) + }) + + t.Run("same series spanning both stores is merged into one", func(t *testing.T) { + stores, _, allBlockIDs := setupHybridParquetTSDB(t, userID, false, "same_series", "same_series") + + // Same label set in both blocks: MergeSeriesSets must yield one series covering [0,200). + series, _, err := querySeries(stores, userID, "same_series", 0, 200, allBlockIDs...) + require.NoError(t, err) + require.Len(t, series, 1, "same series across two blocks must be merged into one") + + require.NotEmpty(t, series[0].Chunks) + minT, maxT := int64(math.MaxInt64), int64(math.MinInt64) + for _, chk := range series[0].Chunks { + if chk.MinTime < minT { + minT = chk.MinTime + } + if chk.MaxTime > maxT { + maxT = chk.MaxTime + } + } + assert.Less(t, minT, int64(100), "chunks must cover the Parquet block [0, 100)") + assert.GreaterOrEqual(t, maxT, int64(100), "chunks must cover the TSDB block [100, 200)") + }) +} + +// TestHybridBucketStores_SyncedBlocksMetrics verifies the cortex_blocks_meta_synced metric of the +// TSDB sub-store in Parquet store-gateway mode: +// +// - bucket index enabled: IgnoreParquetBlocksFilter excludes the converted block from the TSDB +// sync (loaded=1, parquet-converted=1). +// - bucket index disabled: no such filter, so both blocks are loaded (loaded=2). +func TestHybridBucketStores_SyncedBlocksMetrics(t *testing.T) { + const userID = "user-1" + + tests := []struct { + name string + bucketIndexEnabled bool + expectedSyncedMetrics string + }{ + { + name: "bucket index enabled: converted block excluded from TSDB sync", + bucketIndexEnabled: true, + // IgnoreParquetBlocksFilter excludes the Parquet-converted block → loaded=1, parquet-converted=1. + expectedSyncedMetrics: ` + # HELP cortex_blocks_meta_synced Reflects current state of synced blocks (over all tenants). + # TYPE cortex_blocks_meta_synced gauge + cortex_blocks_meta_synced{state="corrupted-bucket-index"} 0 + cortex_blocks_meta_synced{state="corrupted-meta-json"} 0 + cortex_blocks_meta_synced{state="duplicate"} 0 + cortex_blocks_meta_synced{state="failed"} 0 + cortex_blocks_meta_synced{state="label-excluded"} 0 + cortex_blocks_meta_synced{state="loaded"} 1 + cortex_blocks_meta_synced{state="marked-for-deletion"} 0 + cortex_blocks_meta_synced{state="marked-for-no-compact"} 0 + cortex_blocks_meta_synced{state="no-bucket-index"} 0 + cortex_blocks_meta_synced{state="no-meta-json"} 0 + cortex_blocks_meta_synced{state="parquet-converted"} 1 + cortex_blocks_meta_synced{state="parquet-migrated"} 0 + cortex_blocks_meta_synced{state="time-excluded"} 0 + cortex_blocks_meta_synced{state="too-fresh"} 0 + `, + }, + { + name: "bucket index disabled: both blocks loaded by TSDB sub-store", + bucketIndexEnabled: false, + // No IgnoreParquetBlocksFilter (ignoreParquetBlocks=false) → both blocks loaded. + // parquet-converted state is not registered by the non-bucket-index MetaFetcher. + expectedSyncedMetrics: ` + # HELP cortex_blocks_meta_synced Reflects current state of synced blocks (over all tenants). + # TYPE cortex_blocks_meta_synced gauge + cortex_blocks_meta_synced{state="corrupted-meta-json"} 0 + cortex_blocks_meta_synced{state="duplicate"} 0 + cortex_blocks_meta_synced{state="failed"} 0 + cortex_blocks_meta_synced{state="label-excluded"} 0 + cortex_blocks_meta_synced{state="loaded"} 2 + cortex_blocks_meta_synced{state="marked-for-deletion"} 0 + cortex_blocks_meta_synced{state="marked-for-no-compact"} 0 + cortex_blocks_meta_synced{state="no-meta-json"} 0 + cortex_blocks_meta_synced{state="parquet-migrated"} 0 + cortex_blocks_meta_synced{state="time-excluded"} 0 + cortex_blocks_meta_synced{state="too-fresh"} 0 + `, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, reg, _ := setupHybridParquetTSDB(t, userID, tc.bucketIndexEnabled, "series_parquet", "series_tsdb") + require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(tc.expectedSyncedMetrics), + "cortex_blocks_meta_synced", + )) + }) + } +} + +// TestHybridBucketStores_parquetSetFromIndex_Caches verifies that the per-tenant Parquet block +// set derived from the bucket index is cached and only rebuilt when the index version (UpdatedAt) +// changes. +func TestHybridBucketStores_parquetSetFromIndex_Caches(t *testing.T) { + h := &HybridBucketStores{parquetBlocksCache: map[string]parquetBlocksCacheEntry{}} + + b1 := ulidv2.MustNew(1, nil) + b2 := ulidv2.MustNew(2, nil) + b3 := ulidv2.MustNew(3, nil) + + idx := &bucketindex.Index{ + UpdatedAt: 1000, + Blocks: bucketindex.Blocks{ + {ID: b1, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + {ID: b2}, // not converted -> not in the parquet set + }, + } + + set1 := h.parquetSetFromIndex("user-1", idx) + require.Contains(t, set1, b1.String()) + require.NotContains(t, set1, b2.String()) + require.Len(t, set1, 1) + + // Same UpdatedAt -> cache hit: even after mutating idx.Blocks, the cached set is returned unchanged. + idx.Blocks = append(idx.Blocks, &bucketindex.Block{ID: b3, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}) + set2 := h.parquetSetFromIndex("user-1", idx) + require.Len(t, set2, 1, "same index version must return the cached set (not rebuilt)") + require.NotContains(t, set2, b3.String()) + + // New UpdatedAt -> rebuild, now including b3. + idx.UpdatedAt = 2000 + set3 := h.parquetSetFromIndex("user-1", idx) + require.Len(t, set3, 2, "new index version must rebuild the set") + require.Contains(t, set3, b1.String()) + require.Contains(t, set3, b3.String()) + + // Different tenant is cached independently. + other := h.parquetSetFromIndex("user-2", &bucketindex.Index{UpdatedAt: 1}) + require.Empty(t, other) +} + +// TestHybridBucketStores_parquetBlocks_ConverterMarkFallback verifies that block classification +// falls back to reading the per-block converter mark directly when the bucket index is disabled +// or when loading the bucket index fails. +func TestHybridBucketStores_parquetBlocks_ConverterMarkFallback(t *testing.T) { + const userID = "user-1" + + ctx := context.Background() + bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: t.TempDir()}) + require.NoError(t, err) + overrides := validation.NewOverrides(validation.Limits{}, nil) + uBucket := bucket.NewUserBucketClient(userID, bkt, overrides) + + converted := ulidv2.MustNew(1, nil) // has a converter mark -> parquet + notConverted := ulidv2.MustNew(2, nil) // no mark -> tsdb + require.NoError(t, cortex_parquet.WriteConverterMark(ctx, converted, uBucket, 1)) + + blockIDs := []string{converted.String(), notConverted.String()} + + assertClassified := func(t *testing.T, h *HybridBucketStores) { + t.Helper() + result, err := h.parquetBlocks(ctx, userID, blockIDs) + require.NoError(t, err) + assert.True(t, result[converted.String()], "block with a converter mark must be classified as parquet") + assert.False(t, result[notConverted.String()], "block without a converter mark must be classified as tsdb") + } + + t.Run("bucket index disabled (indexLoader nil)", func(t *testing.T) { + h := &HybridBucketStores{ + logger: log.NewNopLogger(), + bucket: bkt, + limits: overrides, + indexLoader: nil, + parquetBlocksCache: map[string]parquetBlocksCacheEntry{}, + } + assertClassified(t, h) + }) + + t.Run("bucket index load fails (falls back to converter marks)", func(t *testing.T) { + // A loader over a bucket without a written index: GetIndex returns ErrIndexNotFound, so + // parquetBlocks must fall back to converter marks instead of failing. + loader := bucketindex.NewLoader(bucketindex.LoaderConfig{ + CheckInterval: time.Minute, + UpdateOnStaleInterval: time.Minute, + UpdateOnErrorInterval: time.Minute, + IdleTimeout: time.Minute, + }, bkt, overrides, log.NewNopLogger(), prometheus.NewRegistry()) + + h := &HybridBucketStores{ + logger: log.NewNopLogger(), + bucket: bkt, + limits: overrides, + indexLoader: loader, + parquetBlocksCache: map[string]parquetBlocksCacheEntry{}, + } + assertClassified(t, h) + }) +} + +// TestHybridBucketStores_Series_MergedSortedByLabels verifies that when distinct series come +// from both the Parquet store and the TSDB store in the same query, the hybrid layer returns +// them as a single label-sorted stream. +func TestHybridBucketStores_Series_MergedSortedByLabels(t *testing.T) { + const ( + userID = "user-1" + metricName = "merge_metric" + ) + + storageDir := t.TempDir() + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + // Bucket index disabled: the router classifies each block by reading its converter mark + // directly, so the Parquet block goes to the Parquet store and the TSDB block to the TSDB store. + cfg.BucketStore.BucketIndex.Enabled = false + + // Parquet block [0,100): series "a" and "c". + generateStorageBlockWithSeriesValues(t, storageDir, userID, metricName, []string{"a", "c"}, 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) + userPath := filepath.Join(storageDir, userID) + + parquetBlockIDs, err := convertToParquetBlocksForTesting(userPath, uBucket) + require.NoError(t, err) + require.Len(t, parquetBlockIDs, 1) + + // TSDB block [0,100): series "b" and "d", which interleave with the Parquet series once sorted. + generateStorageBlockWithSeriesValues(t, storageDir, userID, metricName, []string{"b", "d"}, 0, 100, 15) + + entries, err := os.ReadDir(userPath) + require.NoError(t, err) + var tsdbBlockIDs []string + for _, e := range entries { + if _, parseErr := ulidv2.Parse(e.Name()); parseErr == nil && e.Name() != parquetBlockIDs[0] { + tsdbBlockIDs = append(tsdbBlockIDs, e.Name()) + } + } + require.Len(t, tsdbBlockIDs, 1) + allBlockIDs := append(parquetBlockIDs, tsdbBlockIDs...) + + reg := prometheus.NewRegistry() + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + series, _, err := querySeries(stores, userID, metricName, 0, 100, allBlockIDs...) + require.NoError(t, err) + + // Union across both stores: a (parquet), b (tsdb), c (parquet), d (tsdb). + require.Len(t, series, 4, "hybrid must return the union of series from both stores") + + got := make([]string, 0, len(series)) + for i, s := range series { + lset := s.PromLabels() + got = append(got, lset.Get("series")) + if i > 0 { + require.Negative(t, labels.Compare(series[i-1].PromLabels(), lset), + "merged series must be strictly increasing by labels") + } + } + assert.Equal(t, []string{"a", "b", "c", "d"}, got, "series must be label-sorted and interleave both stores") +} + +// TestHybridBucketStores_LabelNamesAndValues_Merged verifies that when a query spans both a +// Parquet-converted block and a plain TSDB block, the hybrid layer merges label names and label +// values from both sub-stores (sorted and de-duplicated). +func TestHybridBucketStores_LabelNamesAndValues_Merged(t *testing.T) { + const ( + userID = "user-1" + metricName = "merge_metric" + ) + + storageDir := t.TempDir() + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + // Bucket index disabled: the router classifies each block by reading its converter mark + // directly, so the Parquet block goes to the Parquet store and the TSDB block to the TSDB store. + cfg.BucketStore.BucketIndex.Enabled = false + + // Parquet block: a Parquet-only label name "pk" and series values a, c. + generateStorageBlockWithLabelSets(t, storageDir, userID, []labels.Labels{ + labels.FromStrings(labels.MetricName, metricName, "series", "a", "pk", "1"), + labels.FromStrings(labels.MetricName, metricName, "series", "c", "pk", "1"), + }, 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) + userPath := filepath.Join(storageDir, userID) + + parquetBlockIDs, err := convertToParquetBlocksForTesting(userPath, uBucket) + require.NoError(t, err) + require.Len(t, parquetBlockIDs, 1) + + // TSDB block: a TSDB-only label name "tk" and series values b, c, d. The value "c" is shared + // with the Parquet block to exercise de-duplication in the merged LabelValues response. + generateStorageBlockWithLabelSets(t, storageDir, userID, []labels.Labels{ + labels.FromStrings(labels.MetricName, metricName, "series", "b", "tk", "1"), + labels.FromStrings(labels.MetricName, metricName, "series", "c", "tk", "1"), + labels.FromStrings(labels.MetricName, metricName, "series", "d", "tk", "1"), + }, 0, 100, 15) + + entries, err := os.ReadDir(userPath) + require.NoError(t, err) + var tsdbBlockIDs []string + for _, e := range entries { + if _, parseErr := ulidv2.Parse(e.Name()); parseErr == nil && e.Name() != parquetBlockIDs[0] { + tsdbBlockIDs = append(tsdbBlockIDs, e.Name()) + } + } + require.Len(t, tsdbBlockIDs, 1) + allBlockIDs := append(parquetBlockIDs, tsdbBlockIDs...) + + reg := prometheus.NewRegistry() + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + t.Run("LabelNames merges names from both stores", func(t *testing.T) { + resp, err := queryLabelsNamesWithBlocks(stores, userID, metricName, 0, 100, allBlockIDs...) + require.NoError(t, err) + // __name__ and series from both, pk only from Parquet, tk only from TSDB. + assert.Equal(t, []string{labels.MetricName, "pk", "series", "tk"}, resp.Names) + }) + + t.Run("LabelValues merges and de-duplicates values from both stores", func(t *testing.T) { + resp, err := queryLabelsValuesWithBlocks(stores, userID, "series", metricName, 0, 100, allBlockIDs...) + require.NoError(t, err) + // a,c (Parquet) + b,c,d (TSDB) -> deduplicated, sorted union. + assert.Equal(t, []string{"a", "b", "c", "d"}, resp.Values) + }) +} + +// TestHybridBucketStores_Series_StaleIndexRoutingGap guards against a data-gap that can happen +// right after a block is converted to Parquet, because the two bucket-index consumers refresh at +// different times: +// +// - The TSDB sub-store's metadata fetcher reads the bucket index fresh on every sync, so it +// observes the block as Parquet and drops it via IgnoreParquetBlocksFilter. +// - The query router (splitRequestedBlocks -> parquetBlocks) reads the bucket index through the +// cached indexLoader, which can still hold the previous snapshot where the block is a plain +// TSDB block, so it routes the block to the TSDB store. +// +// Without mitigation the block would be served by neither store (the Parquet store isn't asked for +// it, and the TSDB store already dropped it) and, since Thanos silently ignores requested-but-not- +// loaded blocks, the series would just disappear. The grace period keeps a freshly-converted block +// in the TSDB store for a while, so it is still served during this handover window. This test +// asserts the just-converted block's series is still returned. +func TestHybridBucketStores_Series_StaleIndexRoutingGap(t *testing.T) { + const userID = "user-1" + + storageDir := t.TempDir() + cfg := prepareStorageConfig(t) + cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) + cfg.BucketStore.BucketIndex.Enabled = true + // Keep the cached bucket index from being refreshed in the background during the test, + // so the query router keeps observing the stale (pre-conversion) snapshot. + cfg.BucketStore.SyncInterval = time.Hour + + // Block A [0,100): metric_a. Will be converted to Parquet. + generateStorageBlock(t, storageDir, userID, "metric_a", 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) + userPath := filepath.Join(storageDir, userID) + + parquetBlockIDs, err := convertToParquetBlocksForTesting(userPath, uBucket) + require.NoError(t, err) + require.Len(t, parquetBlockIDs, 1) + parquetUID, err := ulidv2.Parse(parquetBlockIDs[0]) + require.NoError(t, err) + + // v1 = the "before conversion" bucket index snapshot (A recorded as a plain TSDB block). + idxV1, _, _, 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, idxV1)) + + reg := prometheus.NewRegistry() + stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) + require.NoError(t, err) + require.NoError(t, stores.InitialSync(context.Background())) + + hStores := stores.(*HybridBucketStores) + + // Warm the query router's cached bucket index with the stale v1 snapshot (A = TSDB). + _, _, err = hStores.indexLoader.GetIndex(context.Background(), userID) + require.NoError(t, err) + + // v2 = the "after conversion" bucket index snapshot (A recorded as Parquet). It overwrites + // the index file in storage. The TSDB sub-store reads this fresh copy on the next sync and + // drops A, while the query router still holds the cached v1 snapshot above. + require.NoError(t, uBucket.Upload(context.Background(), bucketindex.ConverterMarkFilePath(parquetUID), bytes.NewReader([]byte("{}")))) + idxV2, _, _, 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, idxV2)) + + // Re-sync only the TSDB sub-store: it reads the fresh index and drops block A. + require.NoError(t, hStores.tsdb.SyncBlocks(context.Background())) + + // The query router still sees the stale snapshot and routes block A to the TSDB store, + // which no longer has it. Block A's series must still be returned — otherwise data has + // silently vanished during the TSDB -> Parquet handover. + series, _, err := querySeries(stores, userID, "metric_a", 0, 100, parquetBlockIDs...) + require.NoError(t, err) + require.Len(t, series, 1, "series from the just-converted block must not silently disappear due to a stale index snapshot") +} diff --git a/pkg/storegateway/metadata_fetcher_filters.go b/pkg/storegateway/metadata_fetcher_filters.go index 326c544946f..c192b9efe27 100644 --- a/pkg/storegateway/metadata_fetcher_filters.go +++ b/pkg/storegateway/metadata_fetcher_filters.go @@ -77,6 +77,56 @@ func (f *IgnoreDeletionMarkFilter) FilterWithBucketIndex(_ context.Context, meta return nil } +// IgnoreParquetBlocksFilter excludes Parquet-converted blocks from the TSDB +// bucket store sync, so they are served by the Parquet store instead. +// +// A just-converted block is kept for gracePeriod after its conversion time to +// bridge the window where the query router's cached bucket index still routes it +// to the TSDB store, avoiding a temporary data gap. +type IgnoreParquetBlocksFilter struct { + logger log.Logger + // gracePeriod how long a converted block is kept in the TSDB store after + // its conversion time before being dropped. 0 drops converted blocks immediately. + gracePeriod time.Duration +} + +func NewIgnoreParquetBlocksFilter(logger log.Logger, gracePeriod time.Duration) *IgnoreParquetBlocksFilter { + return &IgnoreParquetBlocksFilter{logger: logger, gracePeriod: gracePeriod} +} + +// Filter implements block.MetadataFilter. +// +// Without the bucket index we cannot tell whether a block has been converted to +// Parquet, so this is intentionally a no-op. +func (f *IgnoreParquetBlocksFilter) Filter(_ context.Context, _ map[ulid.ULID]*metadata.Meta, _ block.GaugeVec, _ block.GaugeVec) error { + return nil +} + +// FilterWithBucketIndex implements MetadataFilterWithBucketIndex. +func (f *IgnoreParquetBlocksFilter) FilterWithBucketIndex(_ context.Context, metas map[ulid.ULID]*metadata.Meta, idx *bucketindex.Index, synced block.GaugeVec) error { + now := time.Now() + + for _, b := range idx.ParquetBlocks() { + if _, ok := metas[b.ID]; !ok { + continue + } + + if f.gracePeriod > 0 && b.Parquet.ConvertedAt > 0 { + convertedAt := time.Unix(b.Parquet.ConvertedAt, 0) + if now.Sub(convertedAt) < f.gracePeriod { + level.Debug(f.logger).Log("msg", "keeping parquet-converted block in the TSDB store during grace period", "block", b.ID, "converted_at", convertedAt) + continue + } + } + + level.Debug(f.logger).Log("msg", "ignoring block because it has been converted to parquet", "block", b.ID) + synced.WithLabelValues(parquetConvertedMeta).Inc() + delete(metas, b.ID) + } + + return nil +} + func NewIgnoreNonQueryableBlocksFilter(logger log.Logger, ignoreWithin time.Duration) *IgnoreNonQueryableBlocksFilter { return &IgnoreNonQueryableBlocksFilter{ logger: logger, diff --git a/pkg/storegateway/metadata_fetcher_filters_test.go b/pkg/storegateway/metadata_fetcher_filters_test.go index 26832fb5854..d4a0e9cf091 100644 --- a/pkg/storegateway/metadata_fetcher_filters_test.go +++ b/pkg/storegateway/metadata_fetcher_filters_test.go @@ -22,6 +22,7 @@ import ( "github.com/prometheus/prometheus/tsdb" "github.com/cortexproject/cortex/pkg/storage/bucket" + cortex_parquet "github.com/cortexproject/cortex/pkg/storage/parquet" "github.com/cortexproject/cortex/pkg/storage/tsdb/bucketindex" cortex_testutil "github.com/cortexproject/cortex/pkg/util/testutil" ) @@ -174,3 +175,70 @@ func TestIgnoreNonQueryableBlocksFilter(t *testing.T) { require.NoError(t, f.Filter(ctx, inputMetas, synced, modified)) assert.Equal(t, expectedMetas, inputMetas) } + +func TestIgnoreParquetBlocksFilter_FilterWithBucketIndex_GracePeriod(t *testing.T) { + t.Parallel() + + now := time.Now() + ctx := context.Background() + logger := log.NewNopLogger() + + var ( + recentParquet = ulid.MustNew(1, nil) // converted just now -> within grace + oldParquet = ulid.MustNew(2, nil) // converted long ago -> past grace + unknownAt = ulid.MustNew(3, nil) // parquet without ConvertedAt -> dropped immediately + plainTSDB = ulid.MustNew(4, nil) // not converted -> always kept + ) + + buildIndex := func() *bucketindex.Index { + return &bucketindex.Index{ + Blocks: bucketindex.Blocks{ + {ID: recentParquet, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion, ConvertedAt: now.Add(-1 * time.Minute).Unix()}}, + {ID: oldParquet, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion, ConvertedAt: now.Add(-2 * time.Hour).Unix()}}, + {ID: unknownAt, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + {ID: plainTSDB}, + }, + } + } + + newMetas := func() map[ulid.ULID]*metadata.Meta { + return map[ulid.ULID]*metadata.Meta{ + recentParquet: {}, + oldParquet: {}, + unknownAt: {}, + plainTSDB: {}, + } + } + + t.Run("grace period keeps recently-converted blocks, drops the rest", func(t *testing.T) { + metas := newMetas() + synced := extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}) + + f := NewIgnoreParquetBlocksFilter(logger, time.Hour) + require.NoError(t, f.FilterWithBucketIndex(ctx, metas, buildIndex(), synced)) + + // Recently converted (within grace) and plain TSDB blocks are kept. + assert.Contains(t, metas, recentParquet, "recently-converted block must be kept during grace period") + assert.Contains(t, metas, plainTSDB, "non-parquet block must always be kept") + // Past-grace and unknown-conversion-time blocks are dropped. + assert.NotContains(t, metas, oldParquet, "block converted before the grace window must be dropped") + assert.NotContains(t, metas, unknownAt, "parquet block without a conversion time must be dropped immediately") + + assert.Equal(t, 2.0, promtest.ToFloat64(synced.WithLabelValues(parquetConvertedMeta))) + }) + + t.Run("grace period disabled drops all parquet blocks immediately", func(t *testing.T) { + metas := newMetas() + synced := extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}) + + f := NewIgnoreParquetBlocksFilter(logger, 0) + require.NoError(t, f.FilterWithBucketIndex(ctx, metas, buildIndex(), synced)) + + assert.NotContains(t, metas, recentParquet) + assert.NotContains(t, metas, oldParquet) + assert.NotContains(t, metas, unknownAt) + assert.Contains(t, metas, plainTSDB, "non-parquet block must always be kept") + + assert.Equal(t, 3.0, promtest.ToFloat64(synced.WithLabelValues(parquetConvertedMeta))) + }) +} diff --git a/pkg/storegateway/parquet_bucket_stores.go b/pkg/storegateway/parquet_bucket_stores.go index 9c6ac89bec5..f3f99446d94 100644 --- a/pkg/storegateway/parquet_bucket_stores.go +++ b/pkg/storegateway/parquet_bucket_stores.go @@ -33,7 +33,6 @@ import ( "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" @@ -62,19 +61,24 @@ type ParquetBucketStores struct { parquetShardCache parquetutil.CacheInterface[parquet_storage.ParquetShard] 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 -func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore.InstrumentedBucket, limits *validation.Overrides, logger log.Logger, reg prometheus.Registerer) (*ParquetBucketStores, error) { - // Create caching bucket client for parquet bucket stores - cachingBucket, err := createCachingBucketClientForParquet(cfg, bucketClient, "parquet-storegateway", logger, reg) - if err != nil { - return nil, err +// newParquetBucketStores creates a new multi-tenant parquet bucket stores. +func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore.InstrumentedBucket, cachingBucket objstore.InstrumentedBucket, matcherCache storecache.MatchersCache, limits *validation.Overrides, logger log.Logger, reg prometheus.Registerer) (*ParquetBucketStores, error) { + var err error + if cachingBucket == nil { + // Create caching bucket client for parquet bucket stores + if cachingBucket, err = createCachingBucketClientForParquet(cfg, bucketClient, "parquet-storegateway", logger, reg); err != nil { + return nil, err + } + } + if matcherCache == nil { + if matcherCache, err = newMatchersCache(cfg, logger, reg); err != nil { + return nil, err + } } parquetShardCache, err := parquetutil.NewParquetShardCache[parquet_storage.ParquetShard](&cfg.BucketStore.ParquetShardCache, "parquet-shards", reg) @@ -99,20 +103,9 @@ func newParquetBucketStores(cfg tsdb.BlocksStorageConfig, bucketClient objstore. stores: map[string]*parquetBucketStore{}, storesErrors: map[string]error{}, chunksDecoder: schema.NewPrometheusParquetChunksDecoder(chunkenc.NewPool()), - inflightRequests: cortex_util.NewInflightRequestTracker(), parquetShardCache: parquetShardCache, rowRangesCache: rowRangesCache, - } - - if cfg.BucketStore.MatchersCacheMaxItems > 0 { - r := prometheus.NewRegistry() - reg.MustRegister(tsdb.NewMatchCacheMetrics("cortex_storegateway", r, logger)) - u.matcherCache, err = storecache.NewMatchersCache(storecache.WithSize(cfg.BucketStore.MatchersCacheMaxItems), storecache.WithPromRegistry(r)) - if err != nil { - return nil, err - } - } else { - u.matcherCache = storecache.NoopMatchersCache + matcherCache: matcherCache, } if cfg.BucketStore.BucketIndex.Enabled { @@ -153,20 +146,12 @@ func (u *ParquetBucketStores) Series(req *storepb.SeriesRequest, srv storepb.Sto return status.Error(codes.Internal, err.Error()) } - maxInflightRequests := u.cfg.BucketStore.MaxInflightRequests - if maxInflightRequests > 0 { - if u.inflightRequests.Count() >= maxInflightRequests { - return ErrTooManyInflightRequests - } - - u.inflightRequests.Inc() - defer u.inflightRequests.Dec() - } - - return store.Series(req, spanSeriesServer{ + wrappedSrv := spanSeriesServer{ Store_SeriesServer: srv, ctx: spanCtx, - }) + } + + return store.Series(req, wrappedSrv) } // LabelNames implements BucketStores @@ -601,7 +586,7 @@ func chunkToStoreEncoding(in chunkenc.Encoding) storepb.Chunk_Encoding { } // createCachingBucketClientForParquet creates a caching bucket client for parquet bucket stores -func createCachingBucketClientForParquet(storageCfg tsdb.BlocksStorageConfig, bucketClient objstore.InstrumentedBucket, name string, logger log.Logger, reg prometheus.Registerer) (objstore.Bucket, error) { +func createCachingBucketClientForParquet(storageCfg tsdb.BlocksStorageConfig, bucketClient objstore.InstrumentedBucket, name string, logger log.Logger, reg prometheus.Registerer) (objstore.InstrumentedBucket, error) { // Create caching bucket using the existing infrastructure matchers := tsdb.NewMatchers() cachingBucket, err := tsdb.CreateCachingBucket(storageCfg.BucketStore.ChunksCache, storageCfg.BucketStore.MetadataCache, storageCfg.BucketStore.ParquetLabelsCache, matchers, bucketClient, logger, reg) diff --git a/pkg/storegateway/parquet_bucket_stores_test.go b/pkg/storegateway/parquet_bucket_stores_test.go index e7f3b8780c2..033b354b130 100644 --- a/pkg/storegateway/parquet_bucket_stores_test.go +++ b/pkg/storegateway/parquet_bucket_stores_test.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "errors" - "fmt" "os" "path/filepath" "strconv" @@ -276,7 +275,7 @@ func TestParquetBucketStoresWithCaching(t *testing.T) { limits := validation.NewOverrides(validation.Limits{}, nil) // Create parquet bucket stores with caching - parquetStores, err := newParquetBucketStores(storageCfg, bucketClient, limits, log.NewNopLogger(), prometheus.NewRegistry()) + parquetStores, err := newParquetBucketStores(storageCfg, bucketClient, nil, nil, limits, log.NewNopLogger(), prometheus.NewRegistry()) require.NoError(t, err) require.NotNil(t, parquetStores) require.NotNil(t, parquetStores.rowRangesCache) @@ -335,64 +334,6 @@ func TestCreateCachingBucketClientForParquet(t *testing.T) { require.NotEqual(t, bucketClient, cachingBucket) } -func TestParquetBucketStores_Series_ShouldReturnErrorIfMaxInflightRequestIsReached(t *testing.T) { - cfg := prepareStorageConfig(t) - cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) - cfg.BucketStore.MaxInflightRequests = 10 - reg := prometheus.NewPedanticRegistry() - storageDir := t.TempDir() - generateStorageBlock(t, storageDir, "user_id", "series_1", 0, 100, 15) - bucket, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) - require.NoError(t, err) - - stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bucket), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) - require.NoError(t, err) - require.NoError(t, stores.InitialSync(context.Background())) - - parquetStores := stores.(*ParquetBucketStores) - // Set inflight requests to the limit - for range 10 { - parquetStores.inflightRequests.Inc() - } - series, warnings, err := querySeries(stores, "user_id", "series_1", 0, 100) - assert.ErrorIs(t, err, ErrTooManyInflightRequests) - assert.Empty(t, series) - assert.Empty(t, warnings) -} - -func TestParquetBucketStores_Series_ShouldNotCheckMaxInflightRequestsIfTheLimitIsDisabled(t *testing.T) { - cfg := prepareStorageConfig(t) - cfg.BucketStore.BucketStoreType = string(cortex_tsdb.ParquetBucketStore) - reg := prometheus.NewPedanticRegistry() - storageDir := t.TempDir() - userId := "user_id" - generateStorageBlock(t, storageDir, userId, "series_1", 0, 100, 15) - bkt, err := filesystem.NewBucketClient(filesystem.Config{Directory: storageDir}) - require.NoError(t, err) - - stores, err := NewBucketStores(cfg, NewNoShardingStrategy(log.NewNopLogger(), nil), objstore.WithNoopInstr(bkt), defaultLimitsOverrides(t), mockLoggingLevel(), log.NewNopLogger(), reg) - require.NoError(t, err) - require.NoError(t, stores.InitialSync(context.Background())) - - parquetStores := stores.(*ParquetBucketStores) - // Set inflight requests to the limit (max_inflight_request is set to 0 by default = disabled)for range 10 { - for range 10 { - parquetStores.inflightRequests.Inc() - } - - userPath := fmt.Sprintf("%s/%s", storageDir, userId) - - limits := validation.Limits{} - overrides := validation.NewOverrides(limits, nil) - uBucket := bucket.NewUserBucketClient(userId, bkt, overrides) - blockIds, err := convertToParquetBlocksForTesting(userPath, uBucket) - require.NoError(t, err) - - series, _, err := querySeries(stores, userId, "series_1", 0, 100, blockIds...) - require.NoError(t, err) - assert.Equal(t, 1, len(series)) -} - func convertToParquetBlocksForTesting(userPath string, userBkt objstore.InstrumentedBucket) ([]string, error) { return convertToParquetBlocksWithShardsForTesting(userPath, userBkt, 0, 0) } @@ -473,6 +414,58 @@ func generateStorageBlockWithMultipleSeries(t *testing.T, storageDir, userID, me require.NoError(t, db.Snapshot(userDir, true)) } +// generateStorageBlockWithSeriesValues creates a single TSDB block containing one series per +// provided "series" label value (all sharing the same metric name). It lets tests control the +// exact label ordering across blocks. +func generateStorageBlockWithSeriesValues(t *testing.T, storageDir, userID, metricName string, seriesValues []string, minT, maxT int64, step int) { + t.Helper() + userDir := filepath.Join(storageDir, userID) + if _, err := os.Stat(userDir); os.IsNotExist(err) { + require.NoError(t, os.Mkdir(userDir, os.ModePerm)) + } + + tmpDir := t.TempDir() + db, err := tsdb.Open(tmpDir, promslog.NewNopLogger(), nil, tsdb.DefaultOptions(), nil) + require.NoError(t, err) + defer func() { require.NoError(t, db.Close()) }() + + app := db.Appender(context.Background()) + for i, v := range seriesValues { + lbls := labels.FromStrings(labels.MetricName, metricName, "series", v) + for ts := minT; ts < maxT; ts += int64(step) { + _, err = app.Append(0, lbls, ts, float64(i)) + require.NoError(t, err) + } + } + require.NoError(t, app.Commit()) + require.NoError(t, db.Snapshot(userDir, true)) +} + +// generateStorageBlockWithLabelSets creates a single TSDB block containing exactly the provided +// series label sets. It lets tests control both the label names and values present in each block. +func generateStorageBlockWithLabelSets(t *testing.T, storageDir, userID string, seriesLabels []labels.Labels, minT, maxT int64, step int) { + t.Helper() + userDir := filepath.Join(storageDir, userID) + if _, err := os.Stat(userDir); os.IsNotExist(err) { + require.NoError(t, os.Mkdir(userDir, os.ModePerm)) + } + + tmpDir := t.TempDir() + db, err := tsdb.Open(tmpDir, promslog.NewNopLogger(), nil, tsdb.DefaultOptions(), nil) + require.NoError(t, err) + defer func() { require.NoError(t, db.Close()) }() + + app := db.Appender(context.Background()) + for i, lbls := range seriesLabels { + for ts := minT; ts < maxT; ts += int64(step) { + _, err = app.Append(0, lbls, ts, float64(i)) + require.NoError(t, err) + } + } + require.NoError(t, app.Commit()) + require.NoError(t, db.Snapshot(userDir, true)) +} + // TestParquetBucketStores_Series_MultiShard verifies that when a parquet block is split into // multiple shards, the Store Gateway reads all shards and returns the complete series set. func TestParquetBucketStores_Series_MultiShard(t *testing.T) { @@ -628,7 +621,7 @@ func TestParquetBucketStores_Series_MultiShard_BucketIndexStale_FallbackToConver 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()) + stores, err := newParquetBucketStores(cfg, objstore.WithNoopInstr(bkt), nil, nil, overrides, log.NewNopLogger(), prometheus.NewRegistry()) require.NoError(t, err) series, _, err := querySeries(stores, userID, metricName, 0, 100, blockIDs...) From 7d661a6ac6090b65a9d45546526533d124020432 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 14 Jul 2026 16:59:42 +0900 Subject: [PATCH 2/5] Add pruning of Parquet blocks cache for unowned tenants Signed-off-by: SungJin1212 --- pkg/storegateway/bucket_stores.go | 12 +++++++++ pkg/storegateway/hybrid_bucket_stores.go | 25 +++++++++++++++++-- pkg/storegateway/hybrid_bucket_stores_test.go | 25 +++++++++++++++++++ 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go index b8bea325db2..dbbd6df7750 100644 --- a/pkg/storegateway/bucket_stores.go +++ b/pkg/storegateway/bucket_stores.go @@ -502,6 +502,18 @@ func (u *ThanosBucketStores) getStore(userID string) *store.BucketStore { return u.stores[userID] } +// ownedUserIDs returns the set of tenant IDs this instance currently holds a bucket store for. +// After a sync cycle these are the tenants owned by this store-gateway. +func (u *ThanosBucketStores) ownedUserIDs() map[string]struct{} { + u.storesMu.RLock() + defer u.storesMu.RUnlock() + ids := make(map[string]struct{}, len(u.stores)) + for userID := range u.stores { + ids[userID] = struct{}{} + } + return ids +} + func (u *ThanosBucketStores) getStoreError(userID string) error { u.storesErrorsMu.RLock() defer u.storesErrorsMu.RUnlock() diff --git a/pkg/storegateway/hybrid_bucket_stores.go b/pkg/storegateway/hybrid_bucket_stores.go index 5ab2d81739e..bb2803a3953 100644 --- a/pkg/storegateway/hybrid_bucket_stores.go +++ b/pkg/storegateway/hybrid_bucket_stores.go @@ -194,7 +194,12 @@ func (h *HybridBucketStores) checkStoreError(userID string) error { // SyncBlocks implements BucketStores func (h *HybridBucketStores) SyncBlocks(ctx context.Context) error { - return h.tsdb.SyncBlocks(ctx) + if err := h.tsdb.SyncBlocks(ctx); err != nil { + return err + } + // Keep the per-tenant Parquet blocks cache bounded to the tenants this instance owns. + h.pruneParquetBlocksCache(h.tsdb.ownedUserIDs()) + return nil } // InitialSync implements BucketStores @@ -202,7 +207,11 @@ func (h *HybridBucketStores) InitialSync(ctx context.Context) error { if err := h.parquet.InitialSync(ctx); err != nil { return err } - return h.tsdb.InitialSync(ctx) + if err := h.tsdb.InitialSync(ctx); err != nil { + return err + } + h.pruneParquetBlocksCache(h.tsdb.ownedUserIDs()) + return nil } // Stop implements BucketStores @@ -290,6 +299,18 @@ func (h *HybridBucketStores) parquetSetFromIndex(userID string, idx *bucketindex return parquetSet } +// pruneParquetBlocksCache removes cached Parquet block sets for tenants no longer owned by this instance. +func (h *HybridBucketStores) pruneParquetBlocksCache(includeUserIDs map[string]struct{}) { + h.parquetBlocksCacheMu.Lock() + defer h.parquetBlocksCacheMu.Unlock() + + for userID := range h.parquetBlocksCache { + if _, ok := includeUserIDs[userID]; !ok { + delete(h.parquetBlocksCache, userID) + } + } +} + // parquetBlocksFromConverterMarks classifies each block by reading its converter mark directly. // A missing mark (Version == 0) means the block has not been converted to Parquet yet. func (h *HybridBucketStores) parquetBlocksFromConverterMarks(ctx context.Context, userID string, blockIDs []string) (map[string]bool, error) { diff --git a/pkg/storegateway/hybrid_bucket_stores_test.go b/pkg/storegateway/hybrid_bucket_stores_test.go index fcd0f086183..70af9b76d5c 100644 --- a/pkg/storegateway/hybrid_bucket_stores_test.go +++ b/pkg/storegateway/hybrid_bucket_stores_test.go @@ -331,6 +331,31 @@ func TestHybridBucketStores_parquetSetFromIndex_Caches(t *testing.T) { require.Empty(t, other) } +// TestHybridBucketStores_pruneParquetBlocksCache verifies that cached Parquet block sets are +// removed for tenants no longer owned by this instance, while owned tenants are retained. +func TestHybridBucketStores_pruneParquetBlocksCache(t *testing.T) { + h := &HybridBucketStores{parquetBlocksCache: map[string]parquetBlocksCacheEntry{ + "user-1": {updatedAt: 1}, + "user-2": {updatedAt: 1}, + "user-3": {updatedAt: 1}, + }} + + // Only user-1 and user-3 are still owned by this instance. + h.pruneParquetBlocksCache(map[string]struct{}{ + "user-1": {}, + "user-3": {}, + }) + + require.Contains(t, h.parquetBlocksCache, "user-1") + require.NotContains(t, h.parquetBlocksCache, "user-2", "unowned tenant must be pruned") + require.Contains(t, h.parquetBlocksCache, "user-3") + require.Len(t, h.parquetBlocksCache, 2) + + // Empty owned set prunes everything. + h.pruneParquetBlocksCache(map[string]struct{}{}) + require.Empty(t, h.parquetBlocksCache) +} + // TestHybridBucketStores_parquetBlocks_ConverterMarkFallback verifies that block classification // falls back to reading the per-block converter mark directly when the bucket index is disabled // or when loading the bucket index fails. From 2aeae372cb739eb0cd749ec89a7256fd846bd93e Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 14 Jul 2026 17:12:57 +0900 Subject: [PATCH 3/5] handle error from rewriteSeriesReqBlocks Signed-off-by: SungJin1212 --- pkg/storegateway/hybrid_bucket_stores.go | 76 +++++++++++++++++------- 1 file changed, 55 insertions(+), 21 deletions(-) diff --git a/pkg/storegateway/hybrid_bucket_stores.go b/pkg/storegateway/hybrid_bucket_stores.go index bb2803a3953..c4d87c3003f 100644 --- a/pkg/storegateway/hybrid_bucket_stores.go +++ b/pkg/storegateway/hybrid_bucket_stores.go @@ -391,8 +391,20 @@ func (h *HybridBucketStores) seriesWithTSDBStore(ctx context.Context, userID str tsdbSrv := newBucketStoreSeriesServer(ctx) g, _ := errgroup.WithContext(ctx) - g.Go(func() error { return store.Series(rewriteSeriesReqBlocks(req, parquetIDs), parquetSrv) }) - g.Go(func() error { return h.tsdb.Series(rewriteSeriesReqBlocks(req, tsdbIDs), tsdbSrv) }) + g.Go(func() error { + r, err := rewriteSeriesReqBlocks(req, parquetIDs) + if err != nil { + return err + } + return store.Series(r, parquetSrv) + }) + g.Go(func() error { + r, err := rewriteSeriesReqBlocks(req, tsdbIDs) + if err != nil { + return err + } + return h.tsdb.Series(r, tsdbSrv) + }) if err := g.Wait(); err != nil { return err } @@ -427,7 +439,7 @@ func (h *HybridBucketStores) seriesWithTSDBStore(ctx context.Context, userID str return nil } -func rewriteSeriesReqBlocks(req *storepb.SeriesRequest, blockIDs []string) *storepb.SeriesRequest { +func rewriteSeriesReqBlocks(req *storepb.SeriesRequest, blockIDs []string) (*storepb.SeriesRequest, error) { reqHints := &hintspb.SeriesRequestHints{} if req.Hints != nil { // Best effort: ignore unmarshal error, we always overwrite the block matchers below. @@ -435,11 +447,13 @@ func rewriteSeriesReqBlocks(req *storepb.SeriesRequest, blockIDs []string) *stor } reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} - clone := *req - if anyHints, err := types.MarshalAny(reqHints); err == nil { - clone.Hints = anyHints + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten series request hints").Error()) } - return &clone + clone := *req + clone.Hints = anyHints + return &clone, nil } // labelNamesWithTSDBStore serves a LabelNames request across the Parquet store and the TSDB store. @@ -470,12 +484,20 @@ func (h *HybridBucketStores) labelNamesWithTSDBStore(ctx context.Context, userID ) g, gCtx := errgroup.WithContext(ctx) g.Go(func() error { - r, err := store.LabelNames(gCtx, rewriteLabelNamesReqBlocks(req, parquetIDs)) + req, err := rewriteLabelNamesReqBlocks(req, parquetIDs) + if err != nil { + return err + } + r, err := store.LabelNames(gCtx, req) parquetResp = r return err }) g.Go(func() error { - r, err := h.tsdb.LabelNames(gCtx, rewriteLabelNamesReqBlocks(req, tsdbIDs)) + req, err := rewriteLabelNamesReqBlocks(req, tsdbIDs) + if err != nil { + return err + } + r, err := h.tsdb.LabelNames(gCtx, req) tsdbResp = r return err }) @@ -491,18 +513,20 @@ func (h *HybridBucketStores) labelNamesWithTSDBStore(ctx context.Context, userID return &storepb.LabelNamesResponse{Names: names, Hints: anyHints}, nil } -func rewriteLabelNamesReqBlocks(req *storepb.LabelNamesRequest, blockIDs []string) *storepb.LabelNamesRequest { +func rewriteLabelNamesReqBlocks(req *storepb.LabelNamesRequest, blockIDs []string) (*storepb.LabelNamesRequest, error) { reqHints := &hintspb.LabelNamesRequestHints{} if req.Hints != nil { _ = types.UnmarshalAny(req.Hints, reqHints) } reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} - clone := *req - if anyHints, err := types.MarshalAny(reqHints); err == nil { - clone.Hints = anyHints + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten label names request hints").Error()) } - return &clone + clone := *req + clone.Hints = anyHints + return &clone, nil } func mergeLabelNamesHints(a, b *storepb.LabelNamesResponse) (*types.Any, error) { @@ -546,12 +570,20 @@ func (h *HybridBucketStores) labelValuesWithTSDBStore(ctx context.Context, userI var parquetResp, tsdbResp *storepb.LabelValuesResponse g, gCtx := errgroup.WithContext(ctx) g.Go(func() error { - r, err := store.LabelValues(gCtx, rewriteLabelValuesReqBlocks(req, parquetIDs)) + req, err := rewriteLabelValuesReqBlocks(req, parquetIDs) + if err != nil { + return err + } + r, err := store.LabelValues(gCtx, req) parquetResp = r return err }) g.Go(func() error { - r, err := h.tsdb.LabelValues(gCtx, rewriteLabelValuesReqBlocks(req, tsdbIDs)) + req, err := rewriteLabelValuesReqBlocks(req, tsdbIDs) + if err != nil { + return err + } + r, err := h.tsdb.LabelValues(gCtx, req) tsdbResp = r return err }) @@ -567,18 +599,20 @@ func (h *HybridBucketStores) labelValuesWithTSDBStore(ctx context.Context, userI return &storepb.LabelValuesResponse{Values: values, Hints: anyHints}, nil } -func rewriteLabelValuesReqBlocks(req *storepb.LabelValuesRequest, blockIDs []string) *storepb.LabelValuesRequest { +func rewriteLabelValuesReqBlocks(req *storepb.LabelValuesRequest, blockIDs []string) (*storepb.LabelValuesRequest, error) { reqHints := &hintspb.LabelValuesRequestHints{} if req.Hints != nil { _ = types.UnmarshalAny(req.Hints, reqHints) } reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} - clone := *req - if anyHints, err := types.MarshalAny(reqHints); err == nil { - clone.Hints = anyHints + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten label values request hints").Error()) } - return &clone + clone := *req + clone.Hints = anyHints + return &clone, nil } func mergeLabelValuesHints(a, b *storepb.LabelValuesResponse) (*types.Any, error) { From dede58a1eb11fc87884149b73f6ea8475a6c622b Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 14 Jul 2026 18:57:35 +0900 Subject: [PATCH 4/5] Update ParquetBlocksGracePeriod calculation to account for query router delays Signed-off-by: SungJin1212 --- pkg/storage/tsdb/config.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pkg/storage/tsdb/config.go b/pkg/storage/tsdb/config.go index 47d227c50e0..70334c98d01 100644 --- a/pkg/storage/tsdb/config.go +++ b/pkg/storage/tsdb/config.go @@ -450,9 +450,17 @@ func (cfg *BucketStoreConfig) Validate() error { // ParquetBlocksGracePeriod is how long the TSDB store keeps syncing a just-converted // Parquet block before dropping it, avoiding a data gap right after conversion. -// 2 * SyncInterval comfortably covers that window. +// +// It must cover the worst-case delay until the query router's cached bucket index +// (GetIndex) is refreshed and starts routing the block to the Parquet store: +// +// UpdateOnStaleInterval(=SyncInterval) + 1.2*CheckInterval(=1m) + readIndexTimeout(=15s) +// ≈ SyncInterval + 87s. +// +// A long SyncInterval is covered by 2*SyncInterval; a short one is covered by the +// SyncInterval + 2m. func (cfg *BucketStoreConfig) ParquetBlocksGracePeriod() time.Duration { - return 2 * cfg.SyncInterval + return max(2*cfg.SyncInterval, cfg.SyncInterval+2*time.Minute) } type BucketIndexConfig struct { From 2a08540a666701e8e3e6a75e5989c67673592ec4 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 14 Jul 2026 20:52:44 +0900 Subject: [PATCH 5/5] Fix test to bypass ParquetBlocksGracePeriod for steady-state drop path Signed-off-by: SungJin1212 --- integration/parquet_querier_test.go | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/integration/parquet_querier_test.go b/integration/parquet_querier_test.go index b46e1cf1e2f..2080c45db17 100644 --- a/integration/parquet_querier_test.go +++ b/integration/parquet_querier_test.go @@ -3,8 +3,11 @@ package integration import ( + "bytes" "context" + "encoding/json" "fmt" + "path" "path/filepath" "slices" "sort" @@ -745,7 +748,17 @@ func TestParquetStoreGateway_HybridMode(t *testing.T) { convert.WithName(idA.String())) require.NoError(t, tsdbBlock.Close()) require.NoError(t, convertErr) - require.NoError(t, cortex_parquet.WriteConverterMark(ctx, idA, userBkt, numShards)) + // Set ConvertedAt far in the past to bypass the just-converted grace period + // (ParquetBlocksGracePeriod), so this test exercises the steady-state drop path. + marker := cortex_parquet.ConverterMark{ + Version: cortex_parquet.CurrentVersion, + Shards: numShards, + ConvertedAt: time.Now().Add(-24 * time.Hour).Unix(), + } + markerBytes, marshalErr := json.Marshal(marker) + require.NoError(t, marshalErr) + markerPath := path.Join(idA.String(), cortex_parquet.ConverterMarkerFileName) + require.NoError(t, userBkt.Upload(ctx, markerPath, bytes.NewReader(markerBytes))) // Upload at the global marker path so the bucket index updater discovers it. require.NoError(t, userBkt.Upload(ctx, bucketindex.ConverterMarkFilePath(idA), strings.NewReader("{}"))) }