Implement hybrid mode for Store Gateway

CHANGELOG.md

@@ -15,6 +15,7 @@
 * [FEATURE] Querier: Add timeout classification to classify query timeouts as 4XX (user error) or 5XX (system error) based on phase timing. When enabled, queries that spend most of their time in PromQL evaluation return `422 Unprocessable Entity` instead of `503 Service Unavailable`. #7374
 * [FEATURE] Querier: Implement Resource Based Throttling in Querier. #7442
 * [FEATURE] Querier: Add resource-based query eviction that automatically cancels the heaviest running query when CPU or heap utilization exceeds configured thresholds. #7488
+* [FEATURE] Store Gateway: Implement a hybrid mode of the Store Gateway. #7689
 * [ENHANCEMENT] Upgrade prometheus alertmanager version to v0.32.1. #7462
 * [ENHANCEMENT] Tenant Federation: Avoid purging the regex resolver LRU cache on user-sync ticks when the set of known users has not changed. #7489
 * [ENHANCEMENT] Memberlist: Add `-memberlist.packet-read-timeout`, `-memberlist.max-packet-size`, and `-memberlist.max-concurrent-connections` flags to bound inbound gossip TCP connections, preventing slow-read, OOM, and connection-flood attacks on the gossip port. #7518

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"
@@ -637,3 +646,216 @@ 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)
+		// Set ConvertedAt far in the past to bypass the just-converted grace period
+		// (ParquetBlocksGracePeriod), so this test exercises the steady-state drop path.
+		marker := cortex_parquet.ConverterMark{
+			Version:     cortex_parquet.CurrentVersion,
+			Shards:      numShards,
+			ConvertedAt: time.Now().Add(-24 * time.Hour).Unix(),
+		}
+		markerBytes, marshalErr := json.Marshal(marker)
+		require.NoError(t, marshalErr)
+		markerPath := path.Join(idA.String(), cortex_parquet.ConverterMarkerFileName)
+		require.NoError(t, userBkt.Upload(ctx, markerPath, bytes.NewReader(markerBytes)))
+		// Upload at the global marker path so the bucket index updater discovers it.
+		require.NoError(t, userBkt.Upload(ctx, bucketindex.ConverterMarkFilePath(idA), strings.NewReader("{}")))
+	}
+
+	cortex := e2ecortex.NewSingleBinary("cortex", flags, "")
+	require.NoError(t, s.StartAndWaitReady(cortex))
+
+	c, err := e2ecortex.NewClient("", cortex.HTTPEndpoint(), "", "", userID)
+	require.NoError(t, err)
+
+	// Wait until the compactor has built the bucket index with the correct state:
+	// block A must be tagged as Parquet, block B must be TSDB.
+	cortex_testutil.Poll(t, 60*time.Second, true, func() any {
+		idx, idxErr := bucketindex.ReadIndex(ctx, storage.GetBucket(), userID, nil, log.Logger)
+		if idxErr != nil {
+			return false
+		}
+		foundParquetBlock, foundTSDBBlock := false, false
+		for _, b := range idx.Blocks {
+			switch b.ID {
+			case idA:
+				if b.Parquet != nil {
+					foundParquetBlock = true
+				}
+			case idB:
+				if b.Parquet == nil {
+					foundTSDBBlock = true
+				}
+			}
+		}
+		return foundParquetBlock && foundTSDBBlock
+	})
+
+	// Wait until both metrics are queryable via the store-gateway.
+	cortex_testutil.Poll(t, 120*time.Second, true, func() any {
+		labelSets, err := c.Series([]string{`{job="test"}`}, start, end)
+		if err != nil {
+			return false
+		}
+		foundParquet, foundTSDB := false, false
+		for _, ls := range labelSets {
+			switch string(ls[model.MetricNameLabel]) {
+			case metricParquet:
+				foundParquet = true
+			case metricTSDB:
+				foundTSDB = true
+			}
+		}
+		return foundParquet && foundTSDB
+	})
+
+	// series_parquet must be served by the Parquet store.
+	resParquet, err := c.QueryRange(metricParquet, start, midPoint, scrapeInterval)
+	require.NoError(t, err)
+	matrixParquet, ok := resParquet.(model.Matrix)
+	require.True(t, ok)
+	require.Len(t, matrixParquet, 1, "series_parquet must return one series (served from Parquet store)")
+
+	// series_tsdb must be served by the TSDB store.
+	resTSDB, err := c.QueryRange(metricTSDB, midPoint, end, scrapeInterval)
+	require.NoError(t, err)
+	matrixTSDB, ok := resTSDB.(model.Matrix)
+	require.True(t, ok)
+	require.Len(t, matrixTSDB, 1, "series_tsdb must return one series (served from TSDB store)")
+
+	// Series() must return the union of series_merge series from both the Parquet block
+	// (pk-labeled: a, c) and the TSDB block (tk-labeled: b, c, d). No extra poll is needed here:
+	// series_merge lives in the same blocks A/B already confirmed synced by the poll above.
+	mergeMatcher := fmt.Sprintf(`{__name__=%q}`, metricMerge)
+	mergedSeries, err := c.Series([]string{mergeMatcher}, start, end)
+	require.NoError(t, err)
+	require.Len(t, mergedSeries, 5, "series_merge must return the union of series from both stores")
+	var gotSeriesValues []string
+	for _, ls := range mergedSeries {
+		gotSeriesValues = append(gotSeriesValues, string(ls["series"]))
+	}
+	sort.Strings(gotSeriesValues)
+	require.Equal(t, []string{"a", "b", "c", "c", "d"}, gotSeriesValues,
+		"series_merge series values must include both blocks' series (c appears twice, once per block)")
+
+	// LabelNames() must merge label names from both stores: "pk" only exists in the Parquet
+	// block, "tk" only in the TSDB block.
+	names, err := c.LabelNames(start, end, mergeMatcher)
+	require.NoError(t, err)
+	require.Equal(t, []string{labels.MetricName, "job", "pk", "series", "tk"}, names,
+		"LabelNames must merge Parquet-only and TSDB-only label names")
+
+	// LabelValues() must merge and de-duplicate values from both stores: "c" is present in
+	// both blocks but must appear only once in the merged, sorted result.
+	values, err := c.LabelValues("series", start, end, []string{mergeMatcher})
+	require.NoError(t, err)
+	require.Equal(t, model.LabelValues{"a", "b", "c", "d"}, values,
+		"LabelValues must merge and de-duplicate values across both stores")
+
+	// TSDB sub-store must have loaded only block B (loaded=1) and excluded block A (parquet-converted=1).
+	require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_blocks_meta_synced"},
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "state", "loaded"))))
+	require.NoError(t, cortex.WaitSumMetricsWithOptions(e2e.Equals(1), []string{"cortex_blocks_meta_synced"},
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "state", "parquet-converted"))))
+}

pkg/storage/parquet/converter_marker.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"io"
 	"path"
+	"time"
 
 	"github.com/efficientgo/core/errors"
 	"github.com/go-kit/log"
@@ -32,6 +33,8 @@ type ConverterMark struct {
 	// Shards is the number of parquet shards created for this block.
 	// This field is optional for backward compatibility.
 	Shards int `json:"shards,omitempty"`
+	// ConvertedAt is the unix timestamp (seconds) when the block was converted.
+	ConvertedAt int64 `json:"converted_at,omitempty"`
 }
 
 func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*ConverterMark, error) {
@@ -58,8 +61,9 @@ func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Instr
 
 func WriteConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.Bucket, shards int) error {
 	marker := ConverterMark{
-		Version: CurrentVersion,
-		Shards:  shards,
+		Version:     CurrentVersion,
+		Shards:      shards,
+		ConvertedAt: time.Now().Unix(),
 	}
 	markerPath := path.Join(id.String(), ConverterMarkerFileName)
 	b, err := json.Marshal(marker)
@@ -75,6 +79,8 @@ type ConverterMarkMeta struct {
 	// Shards is the number of parquet shards created for this block.
 	// This field is optional for backward compatibility.
 	Shards int `json:"shards,omitempty"`
+	// ConvertedAt is the unix timestamp (seconds) when the block was converted.
+	ConvertedAt int64 `json:"converted_at,omitempty"`
 }
 
 func ValidConverterMarkVersion(version int) bool {

pkg/storage/tsdb/bucketindex/updater.go

@@ -210,8 +210,9 @@ func (w *Updater) updateParquetBlockIndexEntry(ctx context.Context, id ulid.ULID
 	}
 
 	block.Parquet = &parquet.ConverterMarkMeta{
-		Version: marker.Version,
-		Shards:  marker.Shards,
+		Version:     marker.Version,
+		Shards:      marker.Shards,
+		ConvertedAt: marker.ConvertedAt,
 	}
 	return nil
 }

pkg/storage/tsdb/config.go

@@ -448,6 +448,21 @@ func (cfg *BucketStoreConfig) Validate() error {
 	return nil
 }
 
+// ParquetBlocksGracePeriod is how long the TSDB store keeps syncing a just-converted
+// Parquet block before dropping it, avoiding a data gap right after conversion.
+//
+// It must cover the worst-case delay until the query router's cached bucket index
+// (GetIndex) is refreshed and starts routing the block to the Parquet store:
+//
+//	UpdateOnStaleInterval(=SyncInterval) + 1.2*CheckInterval(=1m) + readIndexTimeout(=15s)
+//	≈ SyncInterval + 87s.
+//
+// A long SyncInterval is covered by 2*SyncInterval; a short one is covered by the
+// SyncInterval + 2m.
+func (cfg *BucketStoreConfig) ParquetBlocksGracePeriod() time.Duration {
+	return max(2*cfg.SyncInterval, cfg.SyncInterval+2*time.Minute)
+}
+
 type BucketIndexConfig struct {
 	Enabled               bool          `yaml:"enabled"`
 	UpdateOnErrorInterval time.Duration `yaml:"update_on_error_interval"`

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),
 	}
 }