From 8839cde493a51dfa4ea98525cd33d61a27a9da27 Mon Sep 17 00:00:00 2001 From: SungJin1212 Date: Tue, 21 Jul 2026 19:08:19 +0900 Subject: [PATCH] Implement hybrid mode Store Gateway Signed-off-by: SungJin1212 --- CHANGELOG.md | 1 + integration/parquet_querier_test.go | 412 ++++++++++ .../bucket_index_metadata_fetcher.go | 3 +- .../bucket_index_metadata_fetcher_test.go | 8 + .../bucket_store_inmemory_server.go | 39 +- .../bucket_store_streaming_server.go | 126 +++ pkg/storegateway/bucket_stores.go | 113 ++- pkg/storegateway/bucket_stores_test.go | 5 + pkg/storegateway/hybrid_bucket_stores.go | 605 ++++++++++++++ pkg/storegateway/hybrid_bucket_stores_test.go | 742 ++++++++++++++++++ pkg/storegateway/metadata_fetcher_filters.go | 55 ++ .../metadata_fetcher_filters_test.go | 46 ++ pkg/storegateway/parquet_bucket_stores.go | 53 +- .../parquet_bucket_stores_test.go | 63 +- 14 files changed, 2133 insertions(+), 138 deletions(-) create mode 100644 pkg/storegateway/bucket_store_streaming_server.go 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 d53c4aaf220..b0d9b696e39 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..0cd9fa837b5 100644 --- a/integration/parquet_querier_test.go +++ b/integration/parquet_querier_test.go @@ -3,17 +3,26 @@ package integration import ( + "bytes" "context" + "encoding/json" "fmt" + "path" "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" @@ -610,6 +619,12 @@ func TestParquetMultiShardQuery(t *testing.T) { return len(labelSets) == totalSeries }) + if tc.viaStoreGateway { + // wait until the parquet block is converted + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(1), []string{"cortex_blocks_meta_synced"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "state", "parquet-converted")))) + } + rangeRes, err := c.QueryRange(`test_series_a`, start, end, scrapeInterval) require.NoError(t, err) rangeMatrix, ok := rangeRes.(model.Matrix) @@ -637,3 +652,400 @@ 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) + marker := cortex_parquet.ConverterMark{ + Version: cortex_parquet.CurrentVersion, + Shards: numShards, + } + 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("{}"))) + } + + 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). + 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")))) +} + +func TestParquetStoreGateway_HybridModeFuzz(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: the second block will never be auto-converted, so the + // store-gateway keeps serving it from the TSDB sub-store (hybrid). + "-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", + "-blocks-storage.bucket-store.index-cache.backend": tsdb.IndexCacheBackendInMemory, + // Route reads through the store-gateway Parquet (hybrid) 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", + // Keep the queried range older than this so all data is served from blocks, not ingesters. + "-limits.query-ingesters-within": "2h", + "-alertmanager.web.external-url": "http://localhost/alertmanager", + // Enables EnableParquet() in the compactor's bucket index updater so the manually-created + // Parquet marker is honoured in the bucket index. + "-parquet-converter.enabled": "true", + "-frontend.query-vertical-shard-size": "3", + }) + + require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs", []byte{})) + + const ( + userID = "user-1" + numShared = 6 // series present in both blocks + numPOnly = 4 // series present only in the Parquet block + numTOnly = 4 // series present only in the TSDB block + numSamples = 60 + ) + + ctx := context.Background() + rnd := newFuzzRand(t) + dir := filepath.Join(s.SharedDir(), "data") + scrapeInterval := time.Minute + statusCodes := []string{"200", "400", "404", "500", "502"} + + now := time.Now() + // The whole range must be older than -limits.query-ingesters-within (2h). Block A covers + // [start, mid) and is converted to Parquet; block B covers [mid, end) and stays TSDB. + start := now.Add(-time.Hour * 24) + mid := now.Add(-time.Hour * 13) + end := now.Add(-time.Hour * 3) + + // shared: same series in both blocks. In the store-gateway these are served by the Parquet + // store (block A) and the TSDB store (block B) and their chunks are merged for the same + // series. + sharedLbls := make([]labels.Labels, 0, numShared) + for i := 0; i < numShared; i++ { + sharedLbls = append(sharedLbls, labels.FromStrings( + labels.MetricName, "test_shared", "job", "test", + "series", strconv.Itoa(i%3), "status_code", statusCodes[i%5])) + } + // parquet-only: only in block A. Carries a Parquet-only label name "pk". + parquetOnlyLbls := make([]labels.Labels, 0, numPOnly) + for i := 0; i < numPOnly; i++ { + parquetOnlyLbls = append(parquetOnlyLbls, labels.FromStrings( + labels.MetricName, "test_parquet_only", "job", "test", + "series", strconv.Itoa(i%3), "pk", strconv.Itoa(i))) + } + // tsdb-only: only in block B. Carries a TSDB-only label name "tk". + tsdbOnlyLbls := make([]labels.Labels, 0, numTOnly) + for i := 0; i < numTOnly; i++ { + tsdbOnlyLbls = append(tsdbOnlyLbls, labels.FromStrings( + labels.MetricName, "test_tsdb_only", "job", "test", + "series", strconv.Itoa(i%3), "tk", strconv.Itoa(i))) + } + + // Block A (Parquet) gets shared + parquet-only, block B (TSDB) gets shared + tsdb-only. + lblsA := append(append([]labels.Labels{}, sharedLbls...), parquetOnlyLbls...) + lblsB := append(append([]labels.Labels{}, sharedLbls...), tsdbOnlyLbls...) + lblsAll := append(append(append([]labels.Labels{}, sharedLbls...), parquetOnlyLbls...), tsdbOnlyLbls...) + + // Block A [start, mid): converted to Parquet. Block B [mid, end): stays TSDB. + idA, err := e2e.CreateBlock(ctx, rnd, dir, lblsA, numSamples, start.UnixMilli(), mid.UnixMilli(), scrapeInterval.Milliseconds(), 10) + require.NoError(t, err) + idB, err := e2e.CreateBlock(ctx, rnd, dir, lblsB, numSamples, mid.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 (block B is left as TSDB). + { + 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) + marker := cortex_parquet.ConverterMark{ + Version: cortex_parquet.CurrentVersion, + Shards: numShards, + } + 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("{}"))) + } + + cortex := e2ecortex.NewSingleBinary("cortex", flags, "") + require.NoError(t, s.StartAndWaitReady(cortex)) + + // Wait until the bucket index reflects the hybrid state: block A tagged Parquet, block B 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 + }) + + 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")))) + + c1, err := e2ecortex.NewClient("", cortex.HTTPEndpoint(), "", "", userID) + require.NoError(t, err) + + // Prometheus reads both blocks directly from the shared data dir, giving it the same data. + require.NoError(t, writeFileToSharedDir(s, "prometheus.yml", []byte(""))) + prom := e2edb.NewPrometheus("", nil) + require.NoError(t, s.StartAndWaitReady(prom)) + + c2, err := e2ecortex.NewPromQueryClient(prom.HTTPEndpoint()) + require.NoError(t, err) + waitUntilReady(t, ctx, c1, c2, `{job="test"}`, start, end) + + opts := []promqlsmith.Option{ + promqlsmith.WithEnabledFunctions(enabledFunctions), + promqlsmith.WithEnabledAggrs(enabledAggrs), + } + ps := promqlsmith.New(rnd, lblsAll, opts...) + + runQueryFuzzTestCases(t, ps, c1, c2, end, start, end, scrapeInterval, 500, true) + + // Confirm the hybrid path actually routed blocks to both sub-stores. + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_hybrid_bucket_stores_blocks_routed_total"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "store", "parquet")))) + require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Greater(0), []string{"cortex_hybrid_bucket_stores_blocks_routed_total"}, + e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "store", "tsdb")))) +} 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_store_inmemory_server.go b/pkg/storegateway/bucket_store_inmemory_server.go index d3902d5b234..528f42c2fbf 100644 --- a/pkg/storegateway/bucket_store_inmemory_server.go +++ b/pkg/storegateway/bucket_store_inmemory_server.go @@ -41,23 +41,40 @@ func (s *bucketStoreSeriesServer) Send(r *storepb.SeriesResponse) error { } if recvSeries := r.GetSeries(); recvSeries != nil { - // Thanos uses a pool for the chunks and may use other pools in the future. - // Given we need to retain the reference after the pooled slices are recycled, - // we need to do a copy here. We prefer to stay on the safest side at this stage - // so we do a marshal+unmarshal to copy the whole series. - recvSeriesData, err := recvSeries.Marshal() - if err != nil { - return errors.Wrap(err, "marshal received series") + if err := s.appendSeries(recvSeries); err != nil { + return err } + } - copiedSeries := &storepb.Series{} - if err = copiedSeries.Unmarshal(recvSeriesData); err != nil { - return errors.Wrap(err, "unmarshal received series") + // When the request's ResponseBatchSize is >= 2, series are delivered as a Batch response + // instead of individual Series responses. + if recvBatch := r.GetBatch(); recvBatch != nil { + for _, recvSeries := range recvBatch.Series { + if err := s.appendSeries(recvSeries); err != nil { + return err + } } + } + + return nil +} + +func (s *bucketStoreSeriesServer) appendSeries(recvSeries *storepb.Series) error { + // Thanos uses a pool for the chunks and may use other pools in the future. + // Given we need to retain the reference after the pooled slices are recycled, + // we need to do a copy here. We prefer to stay on the safest side at this stage + // so we do a marshal+unmarshal to copy the whole series. + recvSeriesData, err := recvSeries.Marshal() + if err != nil { + return errors.Wrap(err, "marshal received series") + } - s.SeriesSet = append(s.SeriesSet, copiedSeries) + copiedSeries := &storepb.Series{} + if err = copiedSeries.Unmarshal(recvSeriesData); err != nil { + return errors.Wrap(err, "unmarshal received series") } + s.SeriesSet = append(s.SeriesSet, copiedSeries) return nil } diff --git a/pkg/storegateway/bucket_store_streaming_server.go b/pkg/storegateway/bucket_store_streaming_server.go new file mode 100644 index 00000000000..acabf4bbfdc --- /dev/null +++ b/pkg/storegateway/bucket_store_streaming_server.go @@ -0,0 +1,126 @@ +package storegateway + +import ( + "context" + "sync" + + "github.com/gogo/protobuf/types" + "github.com/pkg/errors" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/util/annotations" + "github.com/thanos-io/thanos/pkg/store/hintspb" + "github.com/thanos-io/thanos/pkg/store/storepb" +) + +// channelSeriesServer adapts a push-based storepb.Store_SeriesServer into a pull-based +// storepb.SeriesSet backed by a bounded channel. A producer goroutine runs a sub-store's +// Series() with this server, and Send pushes each received series onto the channel. The +// consumer pulls them via Next/At/Err so storepb.MergeSeriesSets can stream-merge multiple +// stores without buffering full results. +type channelSeriesServer struct { + storepb.Store_SeriesServer + + ctx context.Context + ch chan *storepb.Series + + warnings annotations.Annotations + hints hintspb.SeriesResponseHints + + mu sync.Mutex + err error + + // cur holds the series returned by the most recent successful Next call. + cur *storepb.Series +} + +func newChannelSeriesServer(ctx context.Context, bufferSize int) *channelSeriesServer { + if bufferSize < 1 { + bufferSize = 1 + } + return &channelSeriesServer{ + ctx: ctx, + ch: make(chan *storepb.Series, bufferSize), + } +} + +// Context implements storepb.Store_SeriesServer. +func (s *channelSeriesServer) Context() context.Context { return s.ctx } + +// Send implements storepb.Store_SeriesServer. It handles single Series, batched Series, warning +// and hints responses. +func (s *channelSeriesServer) Send(r *storepb.SeriesResponse) error { + if w := r.GetWarning(); w != "" { + s.warnings.Add(errors.New(w)) + } + + if rawHints := r.GetHints(); rawHints != nil { + if err := types.UnmarshalAny(rawHints, &s.hints); err != nil { + return errors.Wrap(err, "failed to unmarshal series hints") + } + } + + if series := r.GetSeries(); series != nil { + return s.push(series) + } + + if batch := r.GetBatch(); batch != nil { + for _, series := range batch.Series { + if err := s.push(series); err != nil { + return err + } + } + } + + return nil +} + +func (s *channelSeriesServer) push(series *storepb.Series) error { + // Thanos uses pools for the chunks and may use other pools in the future. Given we need to + // retain the reference after the pooled slices are recycled, we copy via marshal+unmarshal. + data, err := series.Marshal() + if err != nil { + return errors.Wrap(err, "marshal received series") + } + copied := &storepb.Series{} + if err := copied.Unmarshal(data); err != nil { + return errors.Wrap(err, "unmarshal received series") + } + + select { + case s.ch <- copied: + return nil + case <-s.ctx.Done(): + return s.ctx.Err() + } +} + +// Close records the producer's terminal error (if any) and closes the channel. It must be called +// exactly once by the producer goroutine after Series() returns. +func (s *channelSeriesServer) Close(err error) { + s.mu.Lock() + s.err = err + s.mu.Unlock() + close(s.ch) +} + +// Next implements storepb.SeriesSet. +func (s *channelSeriesServer) Next() bool { + series, ok := <-s.ch + if !ok { + return false + } + s.cur = series + return true +} + +// At implements storepb.SeriesSet. +func (s *channelSeriesServer) At() (labels.Labels, []storepb.AggrChunk) { + return s.cur.PromLabels(), s.cur.Chunks +} + +// Err implements storepb.SeriesSet. It is safe to call concurrently while the producer is running. +func (s *channelSeriesServer) Err() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.err +} diff --git a/pkg/storegateway/bucket_stores.go b/pkg/storegateway/bucket_stores.go index 20ef4b020e1..6a56ceba3cd 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 @@ -86,6 +90,11 @@ type ThanosBucketStores struct { storesErrorsMu sync.RWMutex storesErrors map[string]error + // Per-tenant Parquet filter (set only when ignoreParquetBlocks). Exposes the + // blocks dropped from the TSDB store, used for hybrid routing. + parquetFiltersMu sync.RWMutex + parquetFilters map[string]*IgnoreParquetBlocksFilter + instanceTokenBucket *util.TokenBucket userTokenBucketsMu sync.RWMutex @@ -109,20 +118,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 +160,22 @@ 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{}, + parquetFilters: map[string]*IgnoreParquetBlocksFilter{}, + 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 +202,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 { @@ -495,6 +514,22 @@ func (u *ThanosBucketStores) getStoreError(userID string) error { return u.storesErrors[userID] } +// droppedParquetBlocks returns the block IDs dropped from the TSDB store for the +// user (served by Parquet). ok is false when no drop set is available. +func (u *ThanosBucketStores) droppedParquetBlocks(userID string) (map[string]struct{}, bool) { + u.parquetFiltersMu.RLock() + f := u.parquetFilters[userID] + u.parquetFiltersMu.RUnlock() + if f == nil { + return nil, false + } + dropped := f.DroppedBlocks() + if dropped == nil { + return nil, false + } + return dropped, true +} + var ( errBucketStoreNotEmpty = errors.New("bucket store not empty") errBucketStoreNotFound = errors.New("bucket store not found") @@ -533,6 +568,10 @@ func (u *ThanosBucketStores) closeEmptyBucketStore(userID string) error { u.userTokenBucketsMu.Unlock() } + u.parquetFiltersMu.Lock() + delete(u.parquetFilters, userID) + u.parquetFiltersMu.Unlock() + u.metaFetcherMetrics.RemoveUserRegistry(userID) u.bucketStoreMetrics.RemoveUserRegistry(userID) return bs.Close() @@ -604,6 +643,12 @@ func (u *ThanosBucketStores) getOrCreateStore(userID string) (*store.BucketStore filters = append(filters, NewIgnoreNonQueryableBlocksFilter(userLogger, u.cfg.BucketStore.IgnoreBlocksWithin)) } + var parquetFilter *IgnoreParquetBlocksFilter + if u.ignoreParquetBlocks { + parquetFilter = NewIgnoreParquetBlocksFilter(userLogger) + filters = append(filters, parquetFilter) + } + // 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 { @@ -719,6 +764,12 @@ func (u *ThanosBucketStores) getOrCreateStore(userID string) (*store.BucketStore } u.stores[userID] = bs + // Register the Parquet filter only after the store is successfully created. + if parquetFilter != nil { + u.parquetFiltersMu.Lock() + u.parquetFilters[userID] = parquetFilter + u.parquetFiltersMu.Unlock() + } u.metaFetcherMetrics.AddUserRegistry(userID, fetcherReg) u.bucketStoreMetrics.AddUserRegistry(userID, bucketStoreReg) diff --git a/pkg/storegateway/bucket_stores_test.go b/pkg/storegateway/bucket_stores_test.go index 4bf3e7af065..435859e63be 100644 --- a/pkg/storegateway/bucket_stores_test.go +++ b/pkg/storegateway/bucket_stores_test.go @@ -730,6 +730,10 @@ func generateStorageBlock(t *testing.T, storageDir, userID string, metricName st } func querySeries(stores BucketStores, userID, metricName string, minT, maxT int64, blockIDs ...string) ([]*storepb.Series, annotations.Annotations, error) { + return querySeriesWithBatchSize(stores, userID, metricName, minT, maxT, 0, blockIDs...) +} + +func querySeriesWithBatchSize(stores BucketStores, userID, metricName string, minT, maxT, batchSize int64, blockIDs ...string) ([]*storepb.Series, annotations.Annotations, error) { var ( anyHints *types.Any err error @@ -760,6 +764,7 @@ func querySeries(stores BucketStores, userID, metricName string, minT, maxT int6 }}, PartialResponseStrategy: storepb.PartialResponseStrategy_ABORT, Hints: anyHints, + ResponseBatchSize: batchSize, } ctx := setUserIDToGRPCContext(context.Background(), userID) diff --git a/pkg/storegateway/hybrid_bucket_stores.go b/pkg/storegateway/hybrid_bucket_stores.go new file mode 100644 index 00000000000..e5fb599e0e6 --- /dev/null +++ b/pkg/storegateway/hybrid_bucket_stores.go @@ -0,0 +1,605 @@ +package storegateway + +import ( + "context" + "fmt" + "strings" + + "github.com/go-kit/log" + "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/client_golang/prometheus/promauto" + "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/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" + cortex_util "github.com/cortexproject/cortex/pkg/util" + "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 + + parquet *ParquetBucketStores + tsdb *ThanosBucketStores + + metrics *hybridBucketStoresMetrics +} + +type hybridBucketStoresMetrics struct { + blocksRoutedTotal *prometheus.CounterVec + operationsTotal *prometheus.CounterVec +} + +func newHybridBucketStoresMetrics(reg prometheus.Registerer) *hybridBucketStoresMetrics { + return &hybridBucketStoresMetrics{ + blocksRoutedTotal: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_hybrid_bucket_stores_blocks_routed_total", + Help: "Total number of requested blocks routed to each sub-store.", + }, []string{"store"}), + operationsTotal: promauto.With(reg).NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_hybrid_bucket_stores_operations_total", + Help: "Total number of operations by which sub-store(s) served them.", + }, []string{"store", "method"}), + } +} + +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(), + parquet: parquetStore, + tsdb: tsdbStore, + metrics: newHybridBucketStoresMetrics(reg), + }, 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 { + if err := h.parquet.getStoreError(userID); err != nil { + return err + } + return h.tsdb.getStoreError(userID) +} + +// 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. +func (h *HybridBucketStores) splitRequestedBlocks(ctx context.Context, userID string, blockMatchers []storepb.LabelMatcher, method string) (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) + } + } + + h.metrics.blocksRoutedTotal.WithLabelValues("parquet").Add(float64(len(parquetIDs))) + h.metrics.blocksRoutedTotal.WithLabelValues("tsdb").Add(float64(len(tsdbIDs))) + switch { + case len(parquetIDs) > 0 && len(tsdbIDs) > 0: + h.metrics.operationsTotal.WithLabelValues("mixed", method).Inc() + case len(parquetIDs) > 0: + h.metrics.operationsTotal.WithLabelValues("parquet", method).Inc() + case len(tsdbIDs) > 0: + h.metrics.operationsTotal.WithLabelValues("tsdb", method).Inc() + } + + return parquetIDs, tsdbIDs, nil +} + +// parquetBlocks returns, for each requested block ID, whether it is served by the Parquet store. +func (h *HybridBucketStores) parquetBlocks(ctx context.Context, userID string, blockIDs []string) (map[string]bool, error) { + if dropped, ok := h.tsdb.droppedParquetBlocks(userID); ok { + result := make(map[string]bool, len(blockIDs)) + for _, id := range blockIDs { + _, isParquet := dropped[id] + result[id] = isParquet + } + return result, nil + } + + return h.parquetBlocksFromConverterMarks(ctx, userID, blockIDs) +} + +// 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, "|"), + } +} + +// 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, "Series") + 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 return series sorted by labels. We run them concurrently, each pushing into a + // channel-backed SeriesSet, and stream-merge their outputs so we never buffer either store's + // full result set in memory. + mergeCtx, cancel := context.WithCancel(ctx) + defer cancel() + + g, gCtx := errgroup.WithContext(mergeCtx) + + batchSize := int(req.ResponseBatchSize) + parquetSet := newChannelSeriesServer(gCtx, batchSize) + tsdbSet := newChannelSeriesServer(gCtx, batchSize) + g.Go(func() error { + r, err := rewriteSeriesReqBlocks(req, parquetIDs) + if err == nil { + err = store.Series(r, parquetSet) + } + parquetSet.Close(err) + return err + }) + g.Go(func() error { + r, err := rewriteSeriesReqBlocks(req, tsdbIDs) + if err == nil { + err = h.tsdb.Series(r, tsdbSet) + } + tsdbSet.Close(err) + return err + }) + + outSrv := newFlushableServer(newBatchableServer(srv, batchSize)) + + var sendErr error + merged := storepb.MergeSeriesSets(parquetSet, tsdbSet) + for merged.Next() { + lset, chks := merged.At() + if err := outSrv.Send(storepb.NewSeriesResponse(&storepb.Series{ + Labels: labelpb.ZLabelsFromPromLabels(lset), + Chunks: chks, + })); err != nil { + sendErr = status.Error(codes.Unknown, errors.Wrap(err, "send merged series response").Error()) + break + } + } + + // Unblock and wait for the producers before inspecting their errors. + cancel() + waitErr := g.Wait() + + producerErr := merged.Err() + if producerErr == nil { + producerErr = waitErr + } + + switch { + case sendErr != nil && producerErr != nil: + return multierror.New(producerErr, sendErr).Err() + case producerErr != nil: + return producerErr + case sendErr != nil: + return sendErr + } + + // Forward accumulated warnings from both stores. + warnings := parquetSet.warnings + warnings.Merge(tsdbSet.warnings) + for _, w := range warnings { + if err := outSrv.Send(storepb.NewWarnSeriesResponse(w)); err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "send merged series warning").Error()) + } + } + + resHints := hintspb.SeriesResponseHints{ + QueriedBlocks: append(parquetSet.hints.QueriedBlocks, tsdbSet.hints.QueriedBlocks...), + } + if parquetSet.hints.QueryStats != nil || tsdbSet.hints.QueryStats != nil { + stats := &hintspb.QueryStats{} + if s := parquetSet.hints.QueryStats; s != nil { + stats.Merge(s) + } + if s := tsdbSet.hints.QueryStats; s != nil { + stats.Merge(s) + } + resHints.QueryStats = stats + } + anyHints, err := types.MarshalAny(&resHints) + if err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "marshal series response hints").Error()) + } + if err := outSrv.Send(storepb.NewHintsSeriesResponse(anyHints)); err != nil { + return status.Error(codes.Unknown, errors.Wrap(err, "send series response hints").Error()) + } + + return outSrv.Flush() +} + +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. + _ = types.UnmarshalAny(req.Hints, reqHints) + } + reqHints.BlockMatchers = []storepb.LabelMatcher{blockIDsMatcher(blockIDs)} + + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten series request hints").Error()) + } + clone := *req + clone.Hints = anyHints + return &clone, nil +} + +// 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, "LabelNames") + 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 := rewriteLabelNamesReqBlocks(req, parquetIDs) + if err != nil { + return err + } + parquetResp, err = store.LabelNames(gCtx, r) + return err + }) + g.Go(func() error { + r, err := rewriteLabelNamesReqBlocks(req, tsdbIDs) + if err != nil { + return err + } + tsdbResp, err = h.tsdb.LabelNames(gCtx, 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, + Warnings: append(parquetResp.Warnings, tsdbResp.Warnings...), + Hints: anyHints, + }, nil +} + +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)} + + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten label names request hints").Error()) + } + clone := *req + clone.Hints = anyHints + return &clone, nil +} + +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, "LabelValues") + 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 := rewriteLabelValuesReqBlocks(req, parquetIDs) + if err != nil { + return err + } + parquetResp, err = store.LabelValues(gCtx, r) + return err + }) + g.Go(func() error { + r, err := rewriteLabelValuesReqBlocks(req, tsdbIDs) + if err != nil { + return err + } + tsdbResp, err = h.tsdb.LabelValues(gCtx, 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, + Warnings: append(parquetResp.Warnings, tsdbResp.Warnings...), + Hints: anyHints, + }, nil +} + +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)} + + anyHints, err := types.MarshalAny(reqHints) + if err != nil { + return nil, status.Error(codes.Internal, errors.Wrap(err, "marshal rewritten label values request hints").Error()) + } + clone := *req + clone.Hints = anyHints + return &clone, nil +} + +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..29df0ebb76d --- /dev/null +++ b/pkg/storegateway/hybrid_bucket_stores_test.go @@ -0,0 +1,742 @@ +package storegateway + +import ( + "bytes" + "context" + "fmt" + "math" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/go-kit/log" + "github.com/gogo/protobuf/types" + ulidv2 "github.com/oklog/ulid/v2" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/prometheus/common/promslog" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/tsdb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/thanos-io/objstore" + "github.com/thanos-io/thanos/pkg/block" + thanos_metadata "github.com/thanos-io/thanos/pkg/block/metadata" + "github.com/thanos-io/thanos/pkg/extprom" + "github.com/thanos-io/thanos/pkg/store/hintspb" + "github.com/thanos-io/thanos/pkg/store/storepb" + + "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. +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) + 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_parquetBlocks_UsesDropSet verifies that routing classification +// is driven by the TSDB store's Parquet filter drop set, so a block is classified as +// Parquet iff the filter dropped it from the TSDB store. +func TestHybridBucketStores_parquetBlocks_UsesDropSet(t *testing.T) { + const userID = "user-1" + + dropped := ulidv2.MustNew(1, nil) // converted -> dropped from TSDB -> parquet + kept := ulidv2.MustNew(2, nil) // not converted -> stays in TSDB -> tsdb + unknown := ulidv2.MustNew(3, nil) // not in the index -> tsdb + + // Run the filter over a fresh index so it records its drop set. + filter := NewIgnoreParquetBlocksFilter(log.NewNopLogger()) + metas := map[ulidv2.ULID]*thanos_metadata.Meta{ + dropped: {}, + kept: {}, + } + idx := &bucketindex.Index{ + Blocks: bucketindex.Blocks{ + {ID: dropped, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + {ID: kept}, + }, + } + require.NoError(t, filter.FilterWithBucketIndex(context.Background(), metas, idx, extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}))) + + h := &HybridBucketStores{ + tsdb: &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{userID: filter}}, + } + + result, err := h.parquetBlocks(context.Background(), userID, []string{dropped.String(), kept.String(), unknown.String()}) + require.NoError(t, err) + assert.True(t, result[dropped.String()], "dropped (converted) block must be classified as parquet") + assert.False(t, result[kept.String()], "kept (non-converted) block must be classified as tsdb") + assert.False(t, result[unknown.String()], "block not in the index must be classified as tsdb") +} + +// TestHybridBucketStores_parquetBlocks_ConverterMarkFallback verifies that block classification +// falls back to reading the per-block converter mark directly when no drop set is available +// (e.g. the bucket index is disabled, or the user has not been synced yet). +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("no drop set available (user not synced / parquet filtering disabled)", func(t *testing.T) { + // The TSDB store has no Parquet filter for this user, so droppedParquetBlocks returns + // (nil, false) and parquetBlocks must fall back to reading converter marks. + h := &HybridBucketStores{ + logger: log.NewNopLogger(), + bucket: bkt, + limits: overrides, + tsdb: &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{}}, + } + assertClassified(t, h) + }) +} + +func TestThanosBucketStores_droppedParquetBlocks(t *testing.T) { + const userID = "user-1" + + t.Run("no filter for user -> ok=false", func(t *testing.T) { + u := &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{}} + got, ok := u.droppedParquetBlocks(userID) + assert.False(t, ok) + assert.Nil(t, got) + }) + + t.Run("filter registered but never run -> ok=false", func(t *testing.T) { + u := &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{ + userID: NewIgnoreParquetBlocksFilter(log.NewNopLogger()), + }} + got, ok := u.droppedParquetBlocks(userID) + assert.False(t, ok) + assert.Nil(t, got) + }) + + t.Run("filter run -> ok=true with recorded set", func(t *testing.T) { + parquetBlock := ulidv2.MustNew(1, nil) + filter := NewIgnoreParquetBlocksFilter(log.NewNopLogger()) + require.NoError(t, filter.FilterWithBucketIndex(context.Background(), + map[ulidv2.ULID]*thanos_metadata.Meta{parquetBlock: {}}, + &bucketindex.Index{Blocks: bucketindex.Blocks{ + {ID: parquetBlock, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + }}, + extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}), + )) + + u := &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{userID: filter}} + got, ok := u.droppedParquetBlocks(userID) + assert.True(t, ok) + assert.Contains(t, got, parquetBlock.String()) + }) +} + +func TestHybridBucketStores_parquetBlocks_EmptyDropSetRoutesTSDB(t *testing.T) { + const userID = "user-1" + + b1 := ulidv2.MustNew(1, nil) + b2 := ulidv2.MustNew(2, nil) + + filter := NewIgnoreParquetBlocksFilter(log.NewNopLogger()) + require.NoError(t, filter.FilterWithBucketIndex(context.Background(), + map[ulidv2.ULID]*thanos_metadata.Meta{b1: {}, b2: {}}, + &bucketindex.Index{Blocks: bucketindex.Blocks{{ID: b1}, {ID: b2}}}, // no Parquet blocks + extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}), + )) + + h := &HybridBucketStores{ + tsdb: &ThanosBucketStores{parquetFilters: map[string]*IgnoreParquetBlocksFilter{userID: filter}}, + } + + result, err := h.parquetBlocks(context.Background(), userID, []string{b1.String(), b2.String()}) + require.NoError(t, err) + assert.False(t, result[b1.String()], "block must be routed to TSDB when the drop set is empty") + assert.False(t, result[b2.String()], "block must be routed to TSDB when the drop set is empty") +} + +// 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())) + + // Exercise both the unbatched path (batchSize 0/1 => individual Series responses) and the + // batched path (batchSize >= 2 => Batch responses). + for _, batchSize := range []int64{0, 1, 2, 3} { + t.Run(fmt.Sprintf("batchSize=%d", batchSize), func(t *testing.T) { + series, _, err := querySeriesWithBatchSize(stores, userID, metricName, 0, 100, batchSize, 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: the TSDB sub-store reads the bucket index fresh on +// every sync and drops the converted block via IgnoreParquetBlocksFilter. The hybrid layer must +// still serve that block. +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 + // A long sync interval keeps background refreshes from interfering with the manual re-sync below. + 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) + + // v2 = the "after conversion" bucket index snapshot (A recorded as Parquet). It overwrites + // the index file in storage, so the TSDB sub-store drops A on the next sync. + 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 the TSDB sub-store: it reads the fresh index, drops block A, and records A in its + // drop set so hybrid routing sends A to the Parquet store. + require.NoError(t, hStores.tsdb.SyncBlocks(context.Background())) + + // Block A was dropped from the TSDB store; the drop set makes hybrid route it to the Parquet + // store instead. 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 after the TSDB store drops it") +} + +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) +} + +// 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.TB, 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)) +} diff --git a/pkg/storegateway/metadata_fetcher_filters.go b/pkg/storegateway/metadata_fetcher_filters.go index 326c544946f..7d95a6ff0de 100644 --- a/pkg/storegateway/metadata_fetcher_filters.go +++ b/pkg/storegateway/metadata_fetcher_filters.go @@ -2,6 +2,7 @@ package storegateway import ( "context" + "sync" "time" "github.com/go-kit/log" @@ -77,6 +78,60 @@ func (f *IgnoreDeletionMarkFilter) FilterWithBucketIndex(_ context.Context, meta return nil } +// IgnoreParquetBlocksFilter drops Parquet-converted blocks from the TSDB store +// sync so they are served by the Parquet store instead. +// +// It records the dropped blocks and exposes them via DroppedBlocks. The hybrid +// bucket store routes a block to the Parquet store iff it was dropped here, so the +// drop and routing decisions always use the same bucket index snapshot. +type IgnoreParquetBlocksFilter struct { + logger log.Logger + + // dropped is the set of block IDs dropped on the last sync (served by Parquet). + droppedMu sync.RWMutex + dropped map[string]struct{} +} + +func NewIgnoreParquetBlocksFilter(logger log.Logger) *IgnoreParquetBlocksFilter { + return &IgnoreParquetBlocksFilter{logger: logger} +} + +// 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 { + dropped := make(map[string]struct{}) + for _, b := range idx.ParquetBlocks() { + if _, ok := metas[b.ID]; ok { + 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) + } + + // Record every Parquet block for routing. + dropped[b.ID.String()] = struct{}{} + } + + f.droppedMu.Lock() + f.dropped = dropped + f.droppedMu.Unlock() + + return nil +} + +// DroppedBlocks returns the block IDs served by the Parquet store as of the last sync. +func (f *IgnoreParquetBlocksFilter) DroppedBlocks() map[string]struct{} { + f.droppedMu.RLock() + defer f.droppedMu.RUnlock() + return f.dropped +} + 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..3170aaabf4c 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,48 @@ func TestIgnoreNonQueryableBlocksFilter(t *testing.T) { require.NoError(t, f.Filter(ctx, inputMetas, synced, modified)) assert.Equal(t, expectedMetas, inputMetas) } + +func TestIgnoreParquetBlocksFilter_FilterWithBucketIndex_DropsAndRecords(t *testing.T) { + t.Parallel() + + ctx := context.Background() + logger := log.NewNopLogger() + + var ( + parquetInMetas = ulid.MustNew(1, nil) // parquet + in metas -> dropped, recorded, counted + parquetNotInMetas = ulid.MustNew(2, nil) // parquet but not in metas -> recorded only + plainTSDB = ulid.MustNew(3, nil) // not converted -> always kept, not recorded + ) + + idx := &bucketindex.Index{ + Blocks: bucketindex.Blocks{ + {ID: parquetInMetas, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + {ID: parquetNotInMetas, Parquet: &cortex_parquet.ConverterMarkMeta{Version: cortex_parquet.CurrentVersion}}, + {ID: plainTSDB}, + }, + } + + metas := map[ulid.ULID]*metadata.Meta{ + parquetInMetas: {}, + plainTSDB: {}, + } + synced := extprom.NewTxGaugeVec(nil, prometheus.GaugeOpts{Name: "synced"}, []string{"state"}) + + f := NewIgnoreParquetBlocksFilter(logger) + require.NoError(t, f.FilterWithBucketIndex(ctx, metas, idx, synced)) + + // All Parquet blocks are dropped from the TSDB metas; the plain TSDB block is kept. + assert.NotContains(t, metas, parquetInMetas, "parquet block must be dropped from the TSDB store") + assert.Contains(t, metas, plainTSDB, "non-parquet block must always be kept") + + // Only blocks actually present in metas increment the synced counter. + assert.Equal(t, 1.0, promtest.ToFloat64(synced.WithLabelValues(parquetConvertedMeta))) + + // The drop set records every Parquet block, even one that was never + // in the TSDB metas, and excludes non-parquet blocks. + dropped := f.DroppedBlocks() + assert.Contains(t, dropped, parquetInMetas.String()) + assert.Contains(t, dropped, parquetNotInMetas.String()) + assert.NotContains(t, dropped, plainTSDB.String()) + assert.Len(t, dropped, 2) +} 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..624bc02fad9 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) } @@ -628,7 +569,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...)