Compactor: Upload Sparse Index Headers to Object Storage

CHANGELOG.md

@@ -57,6 +57,7 @@
   * `go_cpu_classes_gc_total_cpu_seconds_total`
   * `go_cpu_classes_total_cpu_seconds_total`
   * `go_cpu_classes_idle_cpu_seconds_total`
+* [ENCHANCEMENT] Compactor: Add experimental `-compactor.upload-sparse-index-headers` option. When enabled, the compactor will attempt to upload sparse index headers to object storage. This prevents latency spikes after adding store-gateway replicas. #10684
 * [BUGFIX] Distributor: Use a boolean to track changes while merging the ReplicaDesc components, rather than comparing the objects directly. #10185
 * [BUGFIX] Querier: fix timeout responding to query-frontend when response size is very close to `-querier.frontend-client.grpc-max-send-msg-size`. #10154
 * [BUGFIX] Query-frontend and querier: show warning/info annotations in some cases where they were missing (if a lazy querier was used). #10277

cmd/mimir/config-descriptor.json

@@ -11512,6 +11512,17 @@
           "fieldFlag": "compactor.max-lookback",
           "fieldType": "duration",
           "fieldCategory": "experimental"
+        },
+        {
+          "kind": "field",
+          "name": "upload_sparse_index_headers",
+          "required": false,
+          "desc": "If enabled, the compactor will construct and upload sparse index headers to object storage during each compaction cycle.",
+          "fieldValue": null,
+          "fieldDefaultValue": false,
+          "fieldFlag": "compactor.upload-sparse-index-headers",
+          "fieldType": "boolean",
+          "fieldCategory": "experimental"
         }
       ],
       "fieldValue": null,

cmd/mimir/help-all.txt.tmpl

@@ -1283,6 +1283,8 @@ Usage of ./cmd/mimir/mimir:
     	Number of symbols flushers used when doing split compaction. (default 1)
   -compactor.tenant-cleanup-delay duration
     	For tenants marked for deletion, this is the time between deletion of the last block, and doing final cleanup (marker files, debug files) of the tenant. (default 6h0m0s)
+  -compactor.upload-sparse-index-headers
+    	[experimental] If enabled, the compactor will construct and upload sparse index headers to object storage during each compaction cycle.
   -config.expand-env
     	Expands ${var} or $var in config according to the values of the environment variables.
   -config.file value

docs/sources/mimir/configure/about-versioning.md

@@ -71,6 +71,8 @@ The following features are currently experimental:
     - `-compactor.in-memory-tenant-meta-cache-size`
   - Limit blocks processed in each compaction cycle. Blocks uploaded prior to the maximum lookback aren't processed.
     - `-compactor.max-lookback`
+  - Enable the compactor to upload sparse index headers to object storage during compaction cycles.
+    - `-compactor.upload-sparse-index-headers`
 - Ruler
   - Aligning of evaluation timestamp on interval (`align_evaluation_time_on_interval`)
   - Allow defining limits on the maximum number of rules allowed in a rule group by namespace and the maximum number of rule groups by namespace. If set, this supersedes the `-ruler.max-rules-per-rule-group` and `-ruler.max-rule-groups-per-tenant` limits.

docs/sources/mimir/configure/configuration-parameters/index.md

@@ -4948,6 +4948,11 @@ sharding_ring:
 # blocks are considered regardless of their upload time.
 # CLI flag: -compactor.max-lookback
 [max_lookback: <duration> | default = 0s]
+
+# (experimental) If enabled, the compactor will construct and upload sparse
+# index headers to object storage during each compaction cycle.
+# CLI flag: -compactor.upload-sparse-index-headers
+[upload_sparse_index_headers: <boolean> | default = false]
 ```
 
 ### store_gateway

pkg/compactor/bucket_compactor.go

@@ -29,8 +29,10 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/tsdb"
 	"github.com/thanos-io/objstore"
+	"github.com/thanos-io/objstore/providers/filesystem"
 	"go.uber.org/atomic"
 
+	"github.com/grafana/mimir/pkg/storage/indexheader"
 	"github.com/grafana/mimir/pkg/storage/sharding"
 	mimir_tsdb "github.com/grafana/mimir/pkg/storage/tsdb"
 	"github.com/grafana/mimir/pkg/storage/tsdb/block"
@@ -393,13 +395,28 @@ func (c *BucketCompactor) runCompactionJob(ctx context.Context, job *Job) (shoul
 		c.metrics.compactionBlocksVerificationFailed.Inc()
 	}
 
+	var uploadSparseIndexHeaders = c.uploadSparseIndexHeaders
+	var fsbkt objstore.Bucket
+	if uploadSparseIndexHeaders {
+		// Create a bucket backed by the local compaction directory. This allows calls to NewStreamBinaryReader to
+		// construct sparse-index-headers without making requests to object storage. Building and uploading
+		// sparse-index-headers is best effort, and we do not skip uploading a compacted block if there's an
+		// error affecting the sparse-index-header upload.
+		fsbkt, err = filesystem.NewBucket(subDir)
+		if err != nil {
+			level.Warn(jobLogger).Log("msg", "failed to create filesystem bucket, skipping sparse header upload", "err", err)
+			uploadSparseIndexHeaders = false
+		}
+	}
+
 	blocksToUpload := convertCompactionResultToForEachJobs(compIDs, job.UseSplitting(), jobLogger)
 	err = concurrency.ForEachJob(ctx, len(blocksToUpload), c.blockSyncConcurrency, func(ctx context.Context, idx int) error {
 		blockToUpload := blocksToUpload[idx]
 
 		uploadedBlocks.Inc()
 
-		bdir := filepath.Join(subDir, blockToUpload.ulid.String())
+		blockID := blockToUpload.ulid.String()
+		bdir := filepath.Join(subDir, blockID)
 
 		// When splitting is enabled, we need to inject the shard ID as an external label.
 		newLabels := job.Labels().Map()
@@ -431,6 +448,32 @@ func (c *BucketCompactor) runCompactionJob(ctx context.Context, job *Job) (shoul
 			return errors.Wrapf(err, "upload of %s failed", blockToUpload.ulid)
 		}
 
+		if uploadSparseIndexHeaders {
+			// Calling NewStreamBinaryReader reads a block's index and writes a sparse-index-header to disk. Because we
+			// don't use the writer, we pass a default indexheader.Config and don't register metrics.
+			if _, err := indexheader.NewStreamBinaryReader(
+				ctx,
+				jobLogger,
+				fsbkt,
+				subDir,
+				blockToUpload.ulid,
+				mimir_tsdb.DefaultPostingOffsetInMemorySampling,
+				indexheader.NewStreamBinaryReaderMetrics(nil),
+				indexheader.Config{},
+			); err != nil {
+				level.Warn(jobLogger).Log("msg", "failed to create sparse index headers", "block", blockID, "err", err)
+				return nil
+			}
+
+			// upload local sparse-index-header to object storage
+			src := path.Join(bdir, block.SparseIndexHeaderFilename)
+			dst := path.Join(blockID, block.SparseIndexHeaderFilename)
+			if err := objstore.UploadFile(ctx, jobLogger, c.bkt, src, dst); err != nil {
+				level.Warn(jobLogger).Log("msg", "failed to upload sparse index headers", "block", blockID, "err", err)
+				return nil
+			}
+		}
+
 		elapsed := time.Since(begin)
 		level.Info(jobLogger).Log("msg", "uploaded block", "result_block", blockToUpload.ulid, "duration", elapsed, "duration_ms", elapsed.Milliseconds(), "external_labels", labels.FromMap(newLabels))
 		return nil
@@ -747,20 +790,21 @@ var ownAllJobs = func(*Job) (bool, error) {
 
 // BucketCompactor compacts blocks in a bucket.
 type BucketCompactor struct {
-	logger               log.Logger
-	sy                   *metaSyncer
-	grouper              Grouper
-	comp                 Compactor
-	planner              Planner
-	compactDir           string
-	bkt                  objstore.Bucket
-	concurrency          int
-	skipUnhealthyBlocks  bool
-	ownJob               ownCompactionJobFunc
-	sortJobs             JobsOrderFunc
-	waitPeriod           time.Duration
-	blockSyncConcurrency int
-	metrics              *BucketCompactorMetrics
+	logger                   log.Logger
+	sy                       *metaSyncer
+	grouper                  Grouper
+	comp                     Compactor
+	planner                  Planner
+	compactDir               string
+	bkt                      objstore.Bucket
+	concurrency              int
+	skipUnhealthyBlocks      bool
+	uploadSparseIndexHeaders bool
+	ownJob                   ownCompactionJobFunc
+	sortJobs                 JobsOrderFunc
+	waitPeriod               time.Duration
+	blockSyncConcurrency     int
+	metrics                  *BucketCompactorMetrics
 }
 
 // NewBucketCompactor creates a new bucket compactor.
@@ -779,25 +823,27 @@ func NewBucketCompactor(
 	waitPeriod time.Duration,
 	blockSyncConcurrency int,
 	metrics *BucketCompactorMetrics,
+	uploadSparseIndexHeaders bool,
 ) (*BucketCompactor, error) {
 	if concurrency <= 0 {
 		return nil, errors.Errorf("invalid concurrency level (%d), concurrency level must be > 0", concurrency)
 	}
 	return &BucketCompactor{
-		logger:               logger,
-		sy:                   sy,
-		grouper:              grouper,
-		planner:              planner,
-		comp:                 comp,
-		compactDir:           compactDir,
-		bkt:                  bkt,
-		concurrency:          concurrency,
-		skipUnhealthyBlocks:  skipUnhealthyBlocks,
-		ownJob:               ownJob,
-		sortJobs:             sortJobs,
-		waitPeriod:           waitPeriod,
-		blockSyncConcurrency: blockSyncConcurrency,
-		metrics:              metrics,
+		logger:                   logger,
+		sy:                       sy,
+		grouper:                  grouper,
+		planner:                  planner,
+		comp:                     comp,
+		compactDir:               compactDir,
+		bkt:                      bkt,
+		concurrency:              concurrency,
+		skipUnhealthyBlocks:      skipUnhealthyBlocks,
+		ownJob:                   ownJob,
+		sortJobs:                 sortJobs,
+		waitPeriod:               waitPeriod,
+		blockSyncConcurrency:     blockSyncConcurrency,
+		metrics:                  metrics,
+		uploadSparseIndexHeaders: uploadSparseIndexHeaders,
 	}, nil
 }
 

pkg/compactor/bucket_compactor_e2e_test.go

@@ -240,7 +240,7 @@ func TestGroupCompactE2E(t *testing.T) {
 		planner := NewSplitAndMergePlanner([]int64{1000, 3000})
 		grouper := NewSplitAndMergeGrouper("user-1", []int64{1000, 3000}, 0, 0, logger)
 		metrics := NewBucketCompactorMetrics(blocksMarkedForDeletion, prometheus.NewPedanticRegistry())
-		bComp, err := NewBucketCompactor(logger, sy, grouper, planner, comp, dir, bkt, 2, true, ownAllJobs, sortJobsByNewestBlocksFirst, 0, 4, metrics)
+		bComp, err := NewBucketCompactor(logger, sy, grouper, planner, comp, dir, bkt, 2, true, ownAllJobs, sortJobsByNewestBlocksFirst, 0, 4, metrics, true)
 		require.NoError(t, err)
 
 		// Compaction on empty should not fail.
@@ -374,6 +374,21 @@ func TestGroupCompactE2E(t *testing.T) {
 			return nil
 		}))
 
+		// expect the blocks that are compacted to have sparse-index-headers in object storage.
+		require.NoError(t, bkt.Iter(ctx, "", func(n string) error {
+			id, ok := block.IsBlockDir(n)
+			if !ok {
+				return nil
+			}
+
+			if _, ok := others[id.String()]; ok {
+				p := path.Join(id.String(), block.SparseIndexHeaderFilename)
+				exists, _ := bkt.Exists(ctx, p)
+				assert.True(t, exists, "expected sparse index headers not found %s", p)
+			}
+			return nil
+		}))
+
 		for id, found := range nonCompactedExpected {
 			assert.True(t, found, "not found expected block %s", id.String())
 		}

pkg/compactor/bucket_compactor_test.go

@@ -120,7 +120,7 @@ func TestBucketCompactor_FilterOwnJobs(t *testing.T) {
 	m := NewBucketCompactorMetrics(promauto.With(nil).NewCounter(prometheus.CounterOpts{}), nil)
 	for testName, testCase := range tests {
 		t.Run(testName, func(t *testing.T) {
-			bc, err := NewBucketCompactor(log.NewNopLogger(), nil, nil, nil, nil, "", nil, 2, false, testCase.ownJob, nil, 0, 4, m)
+			bc, err := NewBucketCompactor(log.NewNopLogger(), nil, nil, nil, nil, "", nil, 2, false, testCase.ownJob, nil, 0, 4, m, false)
 			require.NoError(t, err)
 
 			res, err := bc.filterOwnJobs(jobsFn())
@@ -156,7 +156,7 @@ func TestBlockMaxTimeDeltas(t *testing.T) {
 
 	metrics := NewBucketCompactorMetrics(promauto.With(nil).NewCounter(prometheus.CounterOpts{}), nil)
 	now := time.UnixMilli(1500002900159)
-	bc, err := NewBucketCompactor(log.NewNopLogger(), nil, nil, nil, nil, "", nil, 2, false, nil, nil, 0, 4, metrics)
+	bc, err := NewBucketCompactor(log.NewNopLogger(), nil, nil, nil, nil, "", nil, 2, false, nil, nil, 0, 4, metrics, true)
 	require.NoError(t, err)
 
 	deltas := bc.blockMaxTimeDeltas(now, []*Job{j1, j2})

pkg/compactor/compactor.go

@@ -130,6 +130,9 @@ type Config struct {
 	// Allow downstream projects to customise the blocks compactor.
 	BlocksGrouperFactory   BlocksGrouperFactory   `yaml:"-"`
 	BlocksCompactorFactory BlocksCompactorFactory `yaml:"-"`
+
+	// Allow compactor to upload sparse-index-header files
+	UploadSparseIndexHeaders bool `yaml:"upload_sparse_index_headers" category:"experimental"`
 }
 
 // RegisterFlags registers the MultitenantCompactor flags.
@@ -158,6 +161,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet, logger log.Logger) {
 	f.DurationVar(&cfg.TenantCleanupDelay, "compactor.tenant-cleanup-delay", 6*time.Hour, "For tenants marked for deletion, this is the time between deletion of the last block, and doing final cleanup (marker files, debug files) of the tenant.")
 	f.BoolVar(&cfg.NoBlocksFileCleanupEnabled, "compactor.no-blocks-file-cleanup-enabled", false, "If enabled, will delete the bucket-index, markers and debug files in the tenant bucket when there are no blocks left in the index.")
 	f.DurationVar(&cfg.MaxLookback, "compactor.max-lookback", 0*time.Second, "Blocks uploaded before the lookback aren't considered in compactor cycles. If set, this value should be larger than all values in `-blocks-storage.tsdb.block-ranges-period`. A value of 0s means that all blocks are considered regardless of their upload time.")
+	f.BoolVar(&cfg.UploadSparseIndexHeaders, "compactor.upload-sparse-index-headers", false, "If enabled, the compactor will construct and upload sparse index headers to object storage during each compaction cycle.")
 
 	// compactor concurrency options
 	f.IntVar(&cfg.MaxOpeningBlocksConcurrency, "compactor.max-opening-blocks-concurrency", 1, "Number of goroutines opening blocks before compaction.")
@@ -834,6 +838,7 @@ func (c *MultitenantCompactor) compactUser(ctx context.Context, userID string) e
 		c.compactorCfg.CompactionWaitPeriod,
 		c.compactorCfg.BlockSyncConcurrency,
 		c.bucketCompactorMetrics,
+		c.compactorCfg.UploadSparseIndexHeaders,
 	)
 	if err != nil {
 		return errors.Wrap(err, "failed to create bucket compactor")

pkg/storegateway/indexheader/header.go → pkg/storage/indexheader/header.go

@@ -13,7 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/tsdb/index"
 
-	streamindex "github.com/grafana/mimir/pkg/storegateway/indexheader/index"
+	streamindex "github.com/grafana/mimir/pkg/storage/indexheader/index"
 )
 
 const (

pkg/storegateway/indexheader/header_test.go → pkg/storage/indexheader/header_test.go

@@ -129,6 +129,68 @@ func TestReadersComparedToIndexHeader(t *testing.T) {
 
 }
 
+func Test_DownsampleSparseIndexHeader(t *testing.T) {
+	tests := map[string]struct {
+		protoRate         int
+		inMemSamplingRate int
+	}{
+		"downsample_1_to_32":       {1, 32},
+		"downsample_2_to_32":       {2, 32},
+		"downsample_8_to_32":       {8, 32},
+		"downsample_4_to_16":       {4, 16},
+		"downsample_4_to_48":       {4, 48},
+		"downsample_4_to_4096":     {4, 4096},
+		"downsample_noop_32_to_32": {32, 32},
+	}
+
+	for name, tt := range tests {
+		t.Run(name, func(t *testing.T) {
+			m, err := block.ReadMetaFromDir("./testdata/index_format_v2")
+			require.NoError(t, err)
+
+			tmpDir := t.TempDir()
+			test.Copy(t, "./testdata/index_format_v2", filepath.Join(tmpDir, m.ULID.String()))
+
+			bkt, err := filesystem.NewBucket(tmpDir)
+			require.NoError(t, err)
+
+			ctx := context.Background()
+			noopMetrics := NewStreamBinaryReaderMetrics(nil)
+
+			// write a sparse index-header file to disk
+			br1, err := NewStreamBinaryReader(ctx, log.NewNopLogger(), bkt, tmpDir, m.ULID, tt.protoRate, noopMetrics, Config{})
+			require.NoError(t, err)
+			require.Equal(t, tt.protoRate, br1.postingsOffsetTable.PostingOffsetInMemSampling())
+
+			origLabelNames, err := br1.postingsOffsetTable.LabelNames()
+			require.NoError(t, err)
+
+			// a second call to NewStreamBinaryReader loads the previously written sparse index-header and downsamples
+			// the header from tt.protoRate to tt.inMemSamplingRate entries for each posting
+			br2, err := NewStreamBinaryReader(ctx, log.NewNopLogger(), bkt, tmpDir, m.ULID, tt.inMemSamplingRate, noopMetrics, Config{})
+			require.NoError(t, err)
+			require.Equal(t, tt.inMemSamplingRate, br2.postingsOffsetTable.PostingOffsetInMemSampling())
+
+			downsampleLabelNames, err := br2.postingsOffsetTable.LabelNames()
+			require.NoError(t, err)
+
+			// label names are equal between original and downsampled sparse index-headers
+			require.ElementsMatch(t, downsampleLabelNames, origLabelNames)
+
+			// downsampled postings are a subset of the original sparse index-header postings
+			origIdxpbTbl := br1.postingsOffsetTable.NewSparsePostingOffsetTable()
+			downsampleIdxpbTbl := br2.postingsOffsetTable.NewSparsePostingOffsetTable()
+
+			rr := tt.inMemSamplingRate / tt.protoRate
+			for name, vals := range origIdxpbTbl.Postings {
+				downsampledOffsets := downsampleIdxpbTbl.Postings[name].Offsets
+				require.Equal(t, (len(vals.Offsets)+rr-1)/rr, len(downsampledOffsets))
+				require.Subset(t, vals.Offsets, downsampledOffsets)
+			}
+		})
+	}
+}
+
 func compareIndexToHeader(t *testing.T, indexByteSlice index.ByteSlice, headerReader Reader) {
 	ctx := context.Background()