diff --git a/CHANGELOG.md b/CHANGELOG.md index e2ecad82c36..be840f8acbd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,7 @@ * [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] Storage: Add support for Oracle Cloud Infrastructure (OCI) Object Storage as a backend for blocks, ruler, and alertmanager storage. Configured via `-.oci.*` flags with `backend: oci`. #7718 +* [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] Parquet Converter: Add a ring status page to expose the ring status. #7455 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 9fdf2b87b16..6da1523b25c 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 { @@ -718,6 +763,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...)