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

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

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

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
 }
 

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
+}