TSDB: Option to configure TSDB Block Reload Interval (#16728)

Add --storage.tsdb.block-reload-interval flag to configure TSDB block reload interval.

---------

Signed-off-by: Naman-B-Parlecha <namanparlecha@gmail.com>
Signed-off-by: NamanParlecha <namanparlecha@gmail.com>
Co-authored-by: Arve Knudsen <arve.knudsen@gmail.com>

cmd/prometheus/main.go

@@ -486,6 +486,9 @@ func main() {
 	serverOnlyFlag(a, "storage.tsdb.delay-compact-file.path", "Path to a JSON file with uploaded TSDB blocks e.g. Thanos shipper meta file. If set TSDB will only compact 1 level blocks that are marked as uploaded in that file, improving external storage integrations e.g. with Thanos sidecar. 1+ level compactions won't be delayed.").
 		Default("").StringVar(&tsdbDelayCompactFilePath)
 
+	serverOnlyFlag(a, "storage.tsdb.block-reload-interval", "Interval at which to check for new or removed blocks in storage. Users who manually backfill or drop blocks must wait up to this duration before changes become available.").
+		Default("1m").Hidden().SetValue(&cfg.tsdb.BlockReloadInterval)
+
 	agentOnlyFlag(a, "storage.agent.path", "Base path for metrics storage.").
 		Default("data-agent/").StringVar(&cfg.agentStoragePath)
 
@@ -677,6 +680,10 @@ func main() {
 		}
 		cfg.tsdb.MaxExemplars = cfgFile.StorageConfig.ExemplarsConfig.MaxExemplars
 	}
+	if cfg.tsdb.BlockReloadInterval < model.Duration(1*time.Second) {
+		logger.Warn("The option --storage.tsdb.block-reload-interval is set to a value less than 1s. Setting it to 1s to avoid overload.")
+		cfg.tsdb.BlockReloadInterval = model.Duration(1 * time.Second)
+	}
 	if cfgFile.StorageConfig.TSDBConfig != nil {
 		cfg.tsdb.OutOfOrderTimeWindow = cfgFile.StorageConfig.TSDBConfig.OutOfOrderTimeWindow
 		if cfgFile.StorageConfig.TSDBConfig.Retention != nil {
@@ -1353,6 +1360,7 @@ func main() {
 					"RetentionDuration", cfg.tsdb.RetentionDuration,
 					"WALSegmentSize", cfg.tsdb.WALSegmentSize,
 					"WALCompressionType", cfg.tsdb.WALCompressionType,
+					"BlockReloadInterval", cfg.tsdb.BlockReloadInterval,
 				)
 
 				startTimeMargin := int64(2 * time.Duration(cfg.tsdb.MinBlockDuration).Seconds() * 1000)
@@ -1910,6 +1918,7 @@ type tsdbOptions struct {
 	EnableOverlappingCompaction    bool
 	UseUncachedIO                  bool
 	BlockCompactionExcludeFunc     tsdb.BlockExcludeFilterFunc
+	BlockReloadInterval            model.Duration
 }
 
 func (opts tsdbOptions) ToTSDBOptions() tsdb.Options {
@@ -1934,6 +1943,7 @@ func (opts tsdbOptions) ToTSDBOptions() tsdb.Options {
 		EnableOverlappingCompaction:    opts.EnableOverlappingCompaction,
 		UseUncachedIO:                  opts.UseUncachedIO,
 		BlockCompactionExcludeFunc:     opts.BlockCompactionExcludeFunc,
+		BlockReloadInterval:            time.Duration(opts.BlockReloadInterval),
 		FeatureRegistry:                features.DefaultRegistry,
 	}
 }

tsdb/db.go

@@ -94,6 +94,7 @@ func DefaultOptions() *Options {
 		CompactionDelayMaxPercent:   DefaultCompactionDelayMaxPercent,
 		CompactionDelay:             time.Duration(0),
 		PostingsDecoderFactory:      DefaultPostingsDecoderFactory,
+		BlockReloadInterval:         1 * time.Minute,
 	}
 }
 
@@ -239,6 +240,9 @@ type Options struct {
 	// It's passed down to the TSDB compactor.
 	BlockCompactionExcludeFunc BlockExcludeFilterFunc
 
+	// BlockReloadInterval is the interval at which blocks are reloaded.
+	BlockReloadInterval time.Duration
+
 	// FeatureRegistry is used to register TSDB features.
 	FeatureRegistry features.Collector
 }
@@ -844,6 +848,9 @@ func validateOpts(opts *Options, rngs []int64) (*Options, []int64) {
 	if opts.OutOfOrderTimeWindow < 0 {
 		opts.OutOfOrderTimeWindow = 0
 	}
+	if opts.BlockReloadInterval < 1*time.Second {
+		opts.BlockReloadInterval = 1 * time.Second
+	}
 
 	if len(rngs) == 0 {
 		// Start with smallest block duration and create exponential buckets until the exceed the
@@ -1131,7 +1138,7 @@ func (db *DB) run(ctx context.Context) {
 		}
 
 		select {
-		case <-time.After(1 * time.Minute):
+		case <-time.After(db.opts.BlockReloadInterval):
 			db.cmtx.Lock()
 			if err := db.reloadBlocks(); err != nil {
 				db.logger.Error("reloadBlocks", "err", err)

tsdb/db_test.go

@@ -9284,3 +9284,42 @@ func TestBlockClosingBlockedDuringRemoteRead(t *testing.T) {
 	case <-blockClosed:
 	}
 }
+
+func TestBlockReloadInterval(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name            string
+		reloadInterval  time.Duration
+		expectedReloads float64
+	}{
+		{
+			name:            "extremely small interval",
+			reloadInterval:  1 * time.Millisecond,
+			expectedReloads: 5,
+		},
+		{
+			name:            "one second interval",
+			reloadInterval:  1 * time.Second,
+			expectedReloads: 5,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			t.Parallel()
+			db := newTestDB(t, withOpts(&Options{
+				BlockReloadInterval: c.reloadInterval,
+			}))
+			if c.reloadInterval < 1*time.Second {
+				require.Equal(t, 1*time.Second, db.opts.BlockReloadInterval, "interval should be clamped to minimum of 1 second")
+			}
+			require.Equal(t, float64(1), prom_testutil.ToFloat64(db.metrics.reloads), "there should be one initial reload")
+			require.Eventually(t, func() bool {
+				return prom_testutil.ToFloat64(db.metrics.reloads) == c.expectedReloads
+			},
+				5*time.Second,
+				100*time.Millisecond,
+			)
+		})
+	}
+}