Adding a jitter on the head compaction and changing the max head compaction interval to 30 min

Signed-off-by: alanprot <[email protected]>

Author: alanprot

pkg/ingester/ingester.go

@@ -70,6 +70,7 @@ const (
 
 	// Jitter applied to the idle timeout to prevent compaction in all ingesters concurrently.
 	compactionIdleTimeoutJitter = 0.25
+	initialHeadCompactionJitter = 0.5
 
 	instanceIngestionRateTickInterval = time.Second
 
@@ -2404,13 +2405,20 @@ func (i *Ingester) shipBlocks(ctx context.Context, allowed *util.AllowedTenants)
 }
 
 func (i *Ingester) compactionLoop(ctx context.Context) error {
-	ticker := time.NewTicker(i.cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval)
+	// Apply a jitter on the first head compaction
+	firstHeadCompaction := true
+	ticker := time.NewTicker(util.DurationWithPositiveJitter(i.cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval, initialHeadCompactionJitter))
 	defer ticker.Stop()
 
 	for ctx.Err() == nil {
 		select {
 		case <-ticker.C:
 			i.compactBlocks(ctx, false, nil)
+			// Reset the ticker to run the configured interval on the first head compaction
+			if firstHeadCompaction {
+				ticker.Reset(i.cfg.BlocksStorageConfig.TSDB.HeadCompactionInterval)
+				firstHeadCompaction = false
+			}
 
 		case req := <-i.TSDBState.forceCompactTrigger:
 			i.compactBlocks(ctx, true, req.users)

pkg/storage/tsdb/config.go

@@ -173,7 +173,7 @@ func (cfg *TSDBConfig) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.ShipInterval, "blocks-storage.tsdb.ship-interval", 1*time.Minute, "How frequently the TSDB blocks are scanned and new ones are shipped to the storage. 0 means shipping is disabled.")
 	f.IntVar(&cfg.ShipConcurrency, "blocks-storage.tsdb.ship-concurrency", 10, "Maximum number of tenants concurrently shipping blocks to the storage.")
 	f.IntVar(&cfg.MaxTSDBOpeningConcurrencyOnStartup, "blocks-storage.tsdb.max-tsdb-opening-concurrency-on-startup", 10, "limit the number of concurrently opening TSDB's on startup")
-	f.DurationVar(&cfg.HeadCompactionInterval, "blocks-storage.tsdb.head-compaction-interval", 1*time.Minute, "How frequently does Cortex try to compact TSDB head. Block is only created if data covers smallest block range. Must be greater than 0 and max 5 minutes.")
+	f.DurationVar(&cfg.HeadCompactionInterval, "blocks-storage.tsdb.head-compaction-interval", 1*time.Minute, "How frequently does Cortex try to compact TSDB head. Block is only created if data covers smallest block range. Must be greater than 0 and max 30 minutes. Note that up to 50% jitter is added to the value to avoid ingesters compacting concurrently.")
 	f.IntVar(&cfg.HeadCompactionConcurrency, "blocks-storage.tsdb.head-compaction-concurrency", 5, "Maximum number of tenants concurrently compacting TSDB head into a new block")
 	f.DurationVar(&cfg.HeadCompactionIdleTimeout, "blocks-storage.tsdb.head-compaction-idle-timeout", 1*time.Hour, "If TSDB head is idle for this duration, it is compacted. Note that up to 25% jitter is added to the value to avoid ingesters compacting concurrently. 0 means disabled.")
 	f.IntVar(&cfg.HeadChunksWriteBufferSize, "blocks-storage.tsdb.head-chunks-write-buffer-size-bytes", chunks.DefaultWriteBufferSize, "The write buffer size used by the head chunks mapper. Lower values reduce memory utilisation on clusters with a large number of tenants at the cost of increased disk I/O operations.")
@@ -198,7 +198,7 @@ func (cfg *TSDBConfig) Validate() error {
 		return errInvalidOpeningConcurrency
 	}
 
-	if cfg.HeadCompactionInterval <= 0 || cfg.HeadCompactionInterval > 5*time.Minute {
+	if cfg.HeadCompactionInterval <= 0 || cfg.HeadCompactionInterval > 30*time.Minute {
 		return errInvalidCompactionInterval
 	}