Add an option to spread ingester flushes evenly over time

pkg/ingester/flush.go

@@ -187,6 +187,17 @@ func (i *Ingester) shouldFlushChunk(c *desc, fp model.Fingerprint) flushReason {
 		return noFlush
 	}
 
+	if i.cfg.SpreadFlushes {
+		now := model.Now()
+		// Map from the fingerprint hash to a fixed point in the cycle of period MaxChunkAge
+		startOfCycle := now.Add(-(now.Sub(model.Time(0)) % i.cfg.MaxChunkAge))
+		slot := startOfCycle.Add(time.Duration(fp) % i.cfg.MaxChunkAge)
+		// If that point is now, to the resolution of FlushCheckPeriod, flush the chunk.
+		if slot >= now && slot < now.Add(i.cfg.FlushCheckPeriod) {
+			return reasonAged
+		}
+		// If we missed the slot due to timing it will get caught by the MaxChunkAge check below, some time later.
+	}
 	// Adjust max age slightly to spread flushes out over time
 	var jitter time.Duration
 	if i.cfg.ChunkAgeJitter != 0 {

pkg/ingester/ingester.go

@@ -91,6 +91,7 @@ type Config struct {
 	MaxChunkAge       time.Duration
 	ChunkAgeJitter    time.Duration
 	ConcurrentFlushes int
+	SpreadFlushes     bool
 
 	RateUpdatePeriod time.Duration
 
@@ -109,6 +110,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.MaxChunkIdle, "ingester.max-chunk-idle", 5*time.Minute, "Maximum chunk idle time before flushing.")
 	f.DurationVar(&cfg.MaxChunkAge, "ingester.max-chunk-age", 12*time.Hour, "Maximum chunk age before flushing.")
 	f.DurationVar(&cfg.ChunkAgeJitter, "ingester.chunk-age-jitter", 20*time.Minute, "Range of time to subtract from MaxChunkAge to spread out flushes")
+	f.BoolVar(&cfg.SpreadFlushes, "ingester.spread-flushes", false, "If true, spread series flushes across the whole period of MaxChunkAge")
 	f.IntVar(&cfg.ConcurrentFlushes, "ingester.concurrent-flushes", 50, "Number of concurrent goroutines flushing to dynamodb.")
 	f.DurationVar(&cfg.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 }