Smooth out spikes in rate of chunk flush ops

Ingester chunk flushes run periodically, by default every minute.
Add a rate-limiter so we spread out calls to the DB across the period,
avoiding spikes of intense activity which can slow down other operations
such as incoming Push() calls.

Signed-off-by: Bryan Boreham <[email protected]>

Author: bboreham

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master / unreleased
 
+* [ENHANCEMENT] Smooth out spikes in rate of chunk flush operations. #3191
+
 ## 1.4.0-rc.0 in progress
 
 * [CHANGE] Cassandra backend support is now GA (stable). #3180

pkg/ingester/flush.go

@@ -10,6 +10,7 @@ import (
 	ot "github.com/opentracing/opentracing-go"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/pkg/labels"
+	"golang.org/x/time/rate"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/util"
@@ -19,6 +20,8 @@ const (
 	// Backoff for retrying 'immediate' flushes. Only counts for queue
 	// position, not wallclock time.
 	flushBackoff = 1 * time.Second
+	// Slowest per-second we will rate-limit flush calls to the DB
+	minFlushRate = 10
 )
 
 // Flush triggers a flush of all the chunks and closes the flush queues.
@@ -94,6 +97,25 @@ func (i *Ingester) sweepUsers(immediate bool) {
 	}
 
 	i.metrics.oldestUnflushedChunkTimestamp.Set(float64(oldest.Unix()))
+	i.setFlushRate()
+}
+
+// Compute a rate such to spread DB calls over nearly all of the flush period,
+// for example if we have 600 items in the queue and period 1 min we will send 10.5 per second.
+// Note if the DB can't keep up with this rate then it doesn't make any difference.
+func (i *Ingester) setFlushRate() {
+	totalQueueLength := 0
+	for _, q := range i.flushQueues {
+		totalQueueLength += q.Length()
+	}
+	const fudge = 1.05 // aim to finish a little bit before the end of the period
+	flushesPerSecond := float64(totalQueueLength) / i.cfg.FlushCheckPeriod.Seconds() * fudge
+	// Avoid going very slowly with tiny queues
+	if flushesPerSecond < minFlushRate {
+		flushesPerSecond = minFlushRate
+	}
+	level.Debug(util.Logger).Log("msg", "computed flush rate", "rate", flushesPerSecond)
+	i.flushRateLimiter.SetLimit(rate.Limit(flushesPerSecond))
 }
 
 type flushReason int8
@@ -229,6 +251,7 @@ func (i *Ingester) flushLoop(j int) {
 	}()
 
 	for {
+		_ = i.flushRateLimiter.Wait(context.Background())
 		o := i.flushQueues[j].Dequeue()
 		if o == nil {
 			return

pkg/ingester/ingester.go

@@ -18,6 +18,7 @@ import (
 	tsdb_record "github.com/prometheus/prometheus/tsdb/record"
 	"github.com/weaveworks/common/httpgrpc"
 	"github.com/weaveworks/common/user"
+	"golang.org/x/time/rate"
 	"google.golang.org/grpc/codes"
 
 	cortex_chunk "github.com/cortexproject/cortex/pkg/chunk"
@@ -134,6 +135,9 @@ type Ingester struct {
 	flushQueues     []*util.PriorityQueue
 	flushQueuesDone sync.WaitGroup
 
+	// Spread out calls to the DB over the flush period
+	flushRateLimiter *rate.Limiter
+
 	// This should never be nil.
 	wal WAL
 	// To be passed to the WAL.
@@ -189,11 +193,12 @@ func New(cfg Config, clientConfig client.Config, limits *validation.Overrides, c
 		clientConfig: clientConfig,
 		metrics:      newIngesterMetrics(registerer, true),
 
-		limits:        limits,
-		chunkStore:    chunkStore,
-		flushQueues:   make([]*util.PriorityQueue, cfg.ConcurrentFlushes),
-		usersMetadata: map[string]*userMetricsMetadata{},
-		registerer:    registerer,
+		limits:           limits,
+		chunkStore:       chunkStore,
+		flushQueues:      make([]*util.PriorityQueue, cfg.ConcurrentFlushes),
+		flushRateLimiter: rate.NewLimiter(minFlushRate, 1),
+		usersMetadata:    map[string]*userMetricsMetadata{},
+		registerer:       registerer,
 	}
 
 	var err error
@@ -268,12 +273,13 @@ func NewForFlusher(cfg Config, chunkStore ChunkStore, limits *validation.Overrid
 	}
 
 	i := &Ingester{
-		cfg:         cfg,
-		metrics:     newIngesterMetrics(registerer, true),
-		chunkStore:  chunkStore,
-		flushQueues: make([]*util.PriorityQueue, cfg.ConcurrentFlushes),
-		wal:         &noopWAL{},
-		limits:      limits,
+		cfg:              cfg,
+		metrics:          newIngesterMetrics(registerer, true),
+		chunkStore:       chunkStore,
+		flushQueues:      make([]*util.PriorityQueue, cfg.ConcurrentFlushes),
+		flushRateLimiter: rate.NewLimiter(rate.Inf, 1),
+		wal:              &noopWAL{},
+		limits:           limits,
 	}
 
 	i.BasicService = services.NewBasicService(i.startingForFlusher, i.loopForFlusher, i.stopping)