store-gateway: receive bucket index metadata from queriers

pkg/storegateway/bucket.go

@@ -49,6 +49,7 @@ import (
 	"github.com/grafana/mimir/pkg/storage/tsdb"
 	"github.com/grafana/mimir/pkg/storage/tsdb/block"
 	"github.com/grafana/mimir/pkg/storage/tsdb/bucketcache"
+	"github.com/grafana/mimir/pkg/storage/tsdb/bucketindex"
 	"github.com/grafana/mimir/pkg/storegateway/hintspb"
 	"github.com/grafana/mimir/pkg/storegateway/indexcache"
 	"github.com/grafana/mimir/pkg/storegateway/storegatewaypb"
@@ -88,6 +89,7 @@ type BucketStore struct {
 	logger          log.Logger
 	metrics         *BucketStoreMetrics
 	bkt             objstore.InstrumentedBucketReader
+	bucketIndexMeta BucketIndexMetadataReader
 	fetcher         block.MetadataFetcher
 	dir             string
 	indexCache      indexcache.IndexCache
@@ -203,6 +205,7 @@ func WithLazyLoadingGate(lazyLoadingGate gate.Gate) BucketStoreOption {
 func NewBucketStore(
 	userID string,
 	bkt objstore.InstrumentedBucketReader,
+	bucketIndexMeta BucketIndexMetadataReader,
 	fetcher block.MetadataFetcher,
 	dir string,
 	bucketStoreConfig tsdb.BucketStoreConfig,
@@ -217,6 +220,7 @@ func NewBucketStore(
 	s := &BucketStore{
 		logger:                      log.NewNopLogger(),
 		bkt:                         bkt,
+		bucketIndexMeta:             bucketIndexMeta,
 		fetcher:                     fetcher,
 		dir:                         dir,
 		indexCache:                  noopCache{},
@@ -600,6 +604,8 @@ func (s *BucketStore) Series(req *storepb.SeriesRequest, srv storegatewaypb.Stor
 
 	logSeriesRequestToSpan(spanLogger, req.MinTime, req.MaxTime, matchers, reqBlockMatchers, shardSelector, req.StreamingChunksBatchSize)
 
+	defer s.recordBucketIndexDiscoveryDiff(ctx)
+
 	blocks, indexReaders, chunkReaders := s.openBlocksForReading(ctx, req.SkipChunks, req.MinTime, req.MaxTime, reqBlockMatchers, stats)
 	// We must keep the readers open until all their data has been sent.
 	defer func() {
@@ -1205,6 +1211,27 @@ func (s *BucketStore) recordSeriesHashCacheStats(stats *queryStats) {
 	s.metrics.seriesHashCacheHits.Add(float64(stats.seriesHashCacheHits))
 }
 
+func (s *BucketStore) recordBucketIndexDiscoveryDiff(ctx context.Context) {
+	reqUpdatedAt := getBucketIndexUpdatedAtFromGRPCContext(ctx)
+	meta := s.bucketIndexMeta.Metadata()
+
+	diff := meta.UpdatedAt - reqUpdatedAt
+
+	level.Debug(spanlogger.FromContext(ctx, s.logger)).Log("msg", "bucket index versions (updated_at)", "ours", meta.UpdatedAt, "requested", reqUpdatedAt, "diff", diff)
+
+	if diff == 0 {
+		// Skip recording the difference in metrics when versions are the same.
+		return
+	}
+	var group string
+	if diff > 0 {
+		group = labelDiscoveryDiffNewer
+	} else {
+		group = labelDiscoveryDiffOlder
+	}
+	s.metrics.bucketIndexDiscoveryDiffs.WithLabelValues(group).Inc()
+}
+
 func (s *BucketStore) openBlocksForReading(ctx context.Context, skipChunks bool, minT, maxT int64, blockMatchers []*labels.Matcher, stats *safeQueryStats) ([]*bucketBlock, map[ulid.ULID]*bucketIndexReader, map[ulid.ULID]chunkReader) {
 	spanCtx, span := tracer.Start(ctx, "bucket_store_open_blocks_for_reading")
 	defer span.End()
@@ -1270,6 +1297,8 @@ func (s *BucketStore) LabelNames(ctx context.Context, req *storepb.LabelNamesReq
 		}
 	}
 
+	defer s.recordBucketIndexDiscoveryDiff(ctx)
+
 	g, gctx := errgroup.WithContext(ctx)
 
 	var setsMtx sync.Mutex
@@ -1463,6 +1492,8 @@ func (s *BucketStore) LabelValues(ctx context.Context, req *storepb.LabelValuesR
 		}
 	}
 
+	defer s.recordBucketIndexDiscoveryDiff(ctx)
+
 	var setsMtx sync.Mutex
 	var sets [][]string
 	s.blockSet.filter(req.Start, req.End, reqBlockMatchers, func(b *bucketBlock) {
@@ -2112,3 +2143,7 @@ func maybeNilShard(shard *sharding.ShardSelector) sharding.ShardSelector {
 	}
 	return *shard
 }
+
+type BucketIndexMetadataReader interface {
+	Metadata() *bucketindex.Metadata
+}

pkg/storegateway/bucket_e2e_test.go

@@ -207,6 +207,7 @@ func prepareStoreWithTestBlocks(t testing.TB, bkt objstore.Bucket, cfg *prepareS
 	store, err := NewBucketStore(
 		userID,
 		objstore.WithNoopInstr(bkt),
+		newTestBucketIndexMetadataReader(t, bkt, userID),
 		metaFetcher,
 		cfg.tempDir,
 		cfg.bucketStoreConfig,

pkg/storegateway/bucket_index_metadata.go

@@ -193,3 +193,24 @@ func (f *BucketIndexBlockMetadataFetcher) Fetch(ctx context.Context) (metas map[
 
 	return metas, nil, nil
 }
+
+type bucketIndexMetadataReader struct {
+	indexReader interface {
+		Index() *bucketindex.Index
+	}
+}
+
+// newBucketIndexMetadataReaderFromLoader is an adapter from BucketIndexLoader to BucketIndexMetadataReader.
+func newBucketIndexMetadataReaderFromLoader(loader *BucketIndexLoader) BucketIndexMetadataReader {
+	return &bucketIndexMetadataReader{
+		indexReader: loader,
+	}
+}
+
+func (r *bucketIndexMetadataReader) Metadata() *bucketindex.Metadata {
+	idx := r.indexReader.Index()
+	if idx == nil {
+		return &bucketindex.Metadata{}
+	}
+	return idx.Metadata()
+}

pkg/storegateway/bucket_store_metrics.go

@@ -17,25 +17,31 @@ import (
 	"github.com/grafana/mimir/pkg/storage/indexheader"
 )
 
+const (
+	labelDiscoveryDiffOlder = "older"
+	labelDiscoveryDiffNewer = "newer"
+)
+
 // BucketStoreMetrics holds all the metrics tracked by BucketStore. These metrics
 // MUST be monotonic (counter, summary, histogram) because a single metrics instance
 // can be passed to multiple BucketStore and metrics MUST be correct even after a
 // BucketStore is offloaded.
 type BucketStoreMetrics struct {
-	blockLoads            prometheus.Counter
-	blockLoadFailures     prometheus.Counter
-	blockDrops            prometheus.Counter
-	blockDropFailures     prometheus.Counter
-	blockDiscoveryLatency prometheus.Histogram
-	seriesDataTouched     *prometheus.SummaryVec
-	seriesDataFetched     *prometheus.SummaryVec
-	seriesDataSizeTouched *prometheus.SummaryVec
-	seriesDataSizeFetched *prometheus.SummaryVec
-	seriesBlocksQueried   *prometheus.SummaryVec
-	resultSeriesCount     prometheus.Summary
-	chunkSizeBytes        prometheus.Histogram
-	queriesDropped        *prometheus.CounterVec
-	seriesRefetches       prometheus.Counter
+	blockLoads                prometheus.Counter
+	blockLoadFailures         prometheus.Counter
+	blockDrops                prometheus.Counter
+	blockDropFailures         prometheus.Counter
+	blockDiscoveryLatency     prometheus.Histogram
+	bucketIndexDiscoveryDiffs *prometheus.CounterVec
+	seriesDataTouched         *prometheus.SummaryVec
+	seriesDataFetched         *prometheus.SummaryVec
+	seriesDataSizeTouched     *prometheus.SummaryVec
+	seriesDataSizeFetched     *prometheus.SummaryVec
+	seriesBlocksQueried       *prometheus.SummaryVec
+	resultSeriesCount         prometheus.Summary
+	chunkSizeBytes            prometheus.Histogram
+	queriesDropped            *prometheus.CounterVec
+	seriesRefetches           prometheus.Counter
 
 	// Metrics tracked when streaming store-gateway is enabled.
 	streamingSeriesRequestDurationByStage      *prometheus.HistogramVec
@@ -85,6 +91,13 @@ func NewBucketStoreMetrics(reg prometheus.Registerer) *BucketStoreMetrics {
 		NativeHistogramMaxBucketNumber:  100,
 		NativeHistogramMinResetDuration: 1 * time.Hour,
 	})
+	m.bucketIndexDiscoveryDiffs = promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		Name: "cortex_bucket_store_bucket_index_discovery_difference_total",
+		Help: "Total number of cases of difference between store-gateway and querier versions of bucket index UpdatedAt. diff=older|newer is from the store-gateway side, meaning its version of bucket index is behind/ahead of querier.",
+	}, []string{"diff"})
+	m.bucketIndexDiscoveryDiffs.WithLabelValues(labelDiscoveryDiffOlder)
+	m.bucketIndexDiscoveryDiffs.WithLabelValues(labelDiscoveryDiffNewer)
+
 	m.seriesDataTouched = promauto.With(reg).NewSummaryVec(prometheus.SummaryOpts{
 		Name: "cortex_bucket_store_series_data_touched",
 		Help: "How many items of a data type in a block were touched for a single Series/LabelValues/LabelNames request.",

pkg/storegateway/bucket_stores.go

@@ -10,6 +10,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
 	"sync"
 	"time"
 
@@ -538,6 +539,7 @@ func (u *BucketStores) getOrCreateStore(ctx context.Context, userID string) (*Bu
 	bs, err := NewBucketStore(
 		userID,
 		userBkt,
+		newBucketIndexMetadataReaderFromLoader(loader),
 		fetcher,
 		u.syncDirForUser(userID),
 		u.cfg.BucketStore,
@@ -645,6 +647,16 @@ func getUserIDFromGRPCContext(ctx context.Context) string {
 	return values[0]
 }
 
+func getBucketIndexUpdatedAtFromGRPCContext(ctx context.Context) int64 {
+	values := metadata.ValueFromIncomingContext(ctx, GrpcContextMetadataBucketIndexUpdatedAt)
+	if len(values) != 1 {
+		return 0
+	}
+	// Ignore any parsing errors because BucketIndexUpdatedAt is best-effort.
+	value, _ := strconv.ParseInt(values[0], 10, 64)
+	return value
+}
+
 type spanSeriesServer struct {
 	storegatewaypb.StoreGateway_SeriesServer
 

pkg/storegateway/bucket_test.go

@@ -1456,6 +1456,7 @@ func benchBucketSeries(t test.TB, skipChunk bool, samplesPerSeries, totalSeries
 		st, err := NewBucketStore(
 			"test",
 			ibkt,
+			newTestBucketIndexMetadataReader(t, bkt, "test"),
 			f,
 			tmpDir,
 			mimir_tsdb.BucketStoreConfig{
@@ -1588,6 +1589,7 @@ func TestBucketStore_Series_Concurrency(t *testing.T) {
 					store, err := NewBucketStore(
 						"tenant",
 						instrumentedBucket,
+						newTestBucketIndexMetadataReader(t, instrumentedBucket, "tenant"),
 						metaFetcher,
 						tmpDir,
 						mimir_tsdb.BucketStoreConfig{
@@ -1757,10 +1759,11 @@ func TestBucketStore_Series_OneBlock_InMemIndexCacheSegfault(t *testing.T) {
 	}
 
 	store := &BucketStore{
-		userID:     "test",
-		bkt:        objstore.WithNoopInstr(bkt),
-		logger:     logger,
-		indexCache: indexCache,
+		userID:          "test",
+		bkt:             objstore.WithNoopInstr(bkt),
+		bucketIndexMeta: newTestBucketIndexMetadataReader(t, bkt, "test"),
+		logger:          logger,
+		indexCache:      indexCache,
 		indexReaderPool: indexheader.NewReaderPool(log.NewNopLogger(), indexheader.Config{
 			LazyLoadingEnabled:     false,
 			LazyLoadingIdleTimeout: 0,
@@ -1920,6 +1923,7 @@ func TestBucketStore_Series_ErrorUnmarshallingRequestHints(t *testing.T) {
 	store, err := NewBucketStore(
 		"test",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, bkt, "test"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{
@@ -1977,6 +1981,7 @@ func TestBucketStore_Series_CanceledRequest(t *testing.T) {
 	store, err := NewBucketStore(
 		"test",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, instrBkt, "test"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{
@@ -2049,6 +2054,7 @@ func TestBucketStore_Series_TimeoutGate(t *testing.T) {
 	store, err := NewBucketStore(
 		"test",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, instrBkt, "test"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{
@@ -2127,6 +2133,7 @@ func TestBucketStore_Series_InvalidRequest(t *testing.T) {
 	store, err := NewBucketStore(
 		"test",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, instrBkt, "test"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{
@@ -2253,6 +2260,7 @@ func testBucketStoreSeriesBlockWithMultipleChunks(
 	store, err := NewBucketStore(
 		"tenant",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, instrBkt, "tenant"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{
@@ -2414,6 +2422,7 @@ func TestBucketStore_Series_Limits(t *testing.T) {
 					store, err := NewBucketStore(
 						"tenant",
 						instrBkt,
+						newTestBucketIndexMetadataReader(t, instrBkt, "tenant"),
 						fetcher,
 						tmpDir,
 						mimir_tsdb.BucketStoreConfig{
@@ -2533,6 +2542,7 @@ func setupStoreForHintsTest(t *testing.T, maxSeriesPerBatch int, opts ...BucketS
 	store, err := NewBucketStore(
 		"tenant",
 		instrBkt,
+		newTestBucketIndexMetadataReader(t, instrBkt, "tenant"),
 		fetcher,
 		tmpDir,
 		mimir_tsdb.BucketStoreConfig{

pkg/storegateway/gateway_test.go

@@ -1675,3 +1675,22 @@ func createBucketIndex(t testing.TB, bkt objstore.Bucket, userID string) *bucket
 
 	return idx
 }
+
+type testBucketIndexMetadataReader struct {
+	t      testing.TB
+	bkt    objstore.Bucket
+	userID string
+}
+
+func newTestBucketIndexMetadataReader(t testing.TB, bkt objstore.Bucket, userID string) *testBucketIndexMetadataReader {
+	return &testBucketIndexMetadataReader{
+		t:      t,
+		bkt:    bkt,
+		userID: userID,
+	}
+}
+
+func (t testBucketIndexMetadataReader) Metadata() *bucketindex.Metadata {
+	idx := createBucketIndex(t.t, t.bkt, t.userID)
+	return idx.Metadata()
+}