Merge pull request #182 from weaveworks/127-flush-chunks-older-than-12hr

Flush chunks older than 12hrs

Author: tomwilkie

cmd/cortex/main.go

@@ -108,6 +108,7 @@ func main() {
 	flag.DurationVar(&cfg.ingesterConfig.FlushCheckPeriod, "ingester.flush-period", 1*time.Minute, "Period with which to attempt to flush chunks.")
 	flag.DurationVar(&cfg.ingesterConfig.RateUpdatePeriod, "ingester.rate-update-period", 15*time.Second, "Period with which to update the per-user ingestion rates.")
 	flag.DurationVar(&cfg.ingesterConfig.MaxChunkIdle, "ingester.max-chunk-idle", 1*time.Hour, "Maximum chunk idle time before flushing.")
+	flag.DurationVar(&cfg.ingesterConfig.MaxChunkAge, "ingester.max-chunk-age", 12*time.Hour, "Maximum chunk age time before flushing.")
 	flag.IntVar(&cfg.ingesterConfig.ConcurrentFlushes, "ingester.concurrent-flushes", 25, "Number of concurrent goroutines flushing to dynamodb.")
 	flag.IntVar(&cfg.numTokens, "ingester.num-tokens", 128, "Number of tokens for each ingester.")
 	flag.IntVar(&cfg.ingesterConfig.GRPCListenPort, "ingester.grpc.listen-port", 9095, "gRPC server listen port.")

ingester/ingester.go

@@ -88,6 +88,7 @@ type Ingester struct {
 type Config struct {
 	FlushCheckPeriod  time.Duration
 	MaxChunkIdle      time.Duration
+	MaxChunkAge       time.Duration
 	RateUpdatePeriod  time.Duration
 	ConcurrentFlushes int
 	GRPCListenPort    int
@@ -199,29 +200,6 @@ func (i *Ingester) Ready() bool {
 	return i.cfg.Ring.Ready()
 }
 
-func (i *Ingester) getStateFor(ctx context.Context) (*userState, error) {
-	userID, err := user.GetID(ctx)
-	if err != nil {
-		return nil, fmt.Errorf("no user id")
-	}
-
-	i.userStateLock.Lock()
-	defer i.userStateLock.Unlock()
-	state, ok := i.userState[userID]
-	if !ok {
-		state = &userState{
-			userID:          userID,
-			fpToSeries:      newSeriesMap(),
-			fpLocker:        newFingerprintLocker(16),
-			index:           newInvertedIndex(),
-			ingestedSamples: newEWMARate(0.2, i.cfg.RateUpdatePeriod),
-		}
-		state.mapper = newFPMapper(state.fpToSeries)
-		i.userState[userID] = state
-	}
-	return state, nil
-}
-
 // Push implements cortex.IngesterServer
 func (i *Ingester) Push(ctx context.Context, req *remote.WriteRequest) (*cortex.WriteResponse, error) {
 	for _, sample := range util.FromWriteRequest(req) {
@@ -290,6 +268,29 @@ func (i *Ingester) append(ctx context.Context, sample *model.Sample) error {
 	return err
 }
 
+func (i *Ingester) getStateFor(ctx context.Context) (*userState, error) {
+	userID, err := user.GetID(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("no user id")
+	}
+
+	i.userStateLock.Lock()
+	defer i.userStateLock.Unlock()
+	state, ok := i.userState[userID]
+	if !ok {
+		state = &userState{
+			userID:          userID,
+			fpToSeries:      newSeriesMap(),
+			fpLocker:        newFingerprintLocker(16),
+			index:           newInvertedIndex(),
+			ingestedSamples: newEWMARate(0.2, i.cfg.RateUpdatePeriod),
+		}
+		state.mapper = newFPMapper(state.fpToSeries)
+		i.userState[userID] = state
+	}
+	return state, nil
+}
+
 func (u *userState) getOrCreateSeries(metric model.Metric) (model.Fingerprint, *memorySeries, error) {
 	rawFP := metric.FastFingerprint()
 	u.fpLocker.Lock(rawFP)
@@ -448,7 +449,7 @@ func (i *Ingester) Stop() {
 
 func (i *Ingester) loop() {
 	defer func() {
-		i.flushAllUsers(true)
+		i.sweepUsers(true)
 
 		// We close flush queue here to ensure the flushLoops pick
 		// up all the flushes triggered by the last run
@@ -465,7 +466,7 @@ func (i *Ingester) loop() {
 	for {
 		select {
 		case <-flushTick:
-			i.flushAllUsers(false)
+			i.sweepUsers(false)
 		case <-rateUpdateTick:
 			i.updateRates()
 		case <-i.quit:
@@ -474,7 +475,8 @@ func (i *Ingester) loop() {
 	}
 }
 
-func (i *Ingester) flushAllUsers(immediate bool) {
+// sweepUsers periodically schedules series for flushing and garbage collects users with no series
+func (i *Ingester) sweepUsers(immediate bool) {
 	if i.chunkStore == nil {
 		return
 	}
@@ -487,42 +489,68 @@ func (i *Ingester) flushAllUsers(immediate bool) {
 	i.userStateLock.Unlock()
 
 	for id, state := range userState {
-		i.flushUser(id, state, immediate)
-	}
-}
-
-func (i *Ingester) flushUser(userID string, userState *userState, immediate bool) {
-	for pair := range userState.fpToSeries.iter() {
-		i.flushSeries(userState, pair.fp, pair.series, immediate)
+		for pair := range state.fpToSeries.iter() {
+			state.fpLocker.Lock(pair.fp)
+			i.sweepSeries(id, pair.fp, pair.series, immediate)
+			state.fpLocker.Unlock(pair.fp)
+		}
 	}
 
-	// TODO: this is probably slow, and could be done in a better way.
 	i.userStateLock.Lock()
-	if userState.fpToSeries.length() == 0 {
-		delete(i.userState, userID)
+	for id, state := range userState {
+		if state.fpToSeries.length() == 0 {
+			delete(i.userState, id)
+		}
 	}
 	i.userStateLock.Unlock()
 }
 
-func (i *Ingester) flushSeries(u *userState, fp model.Fingerprint, series *memorySeries, immediate bool) {
-	// Enqueue this series flushing if the oldest chunk is older than the threshold
-
-	u.fpLocker.Lock(fp)
+// sweepSeries schedules a series for flushing based on a set of criteria
+//
+// NB we don't close the head chunk here, as the series could wait in the queue
+// for some time, and we want to encourage chunks to be as full as possible.
+func (i *Ingester) sweepSeries(userID string, fp model.Fingerprint, series *memorySeries, immediate bool) {
 	if len(series.chunkDescs) <= 0 {
-		u.fpLocker.Unlock(fp)
 		return
 	}
 
 	lastTime := series.lastTime
-	flush := immediate || len(series.chunkDescs) > 1 || model.Now().Sub(lastTime) > i.cfg.MaxChunkIdle
-	u.fpLocker.Unlock(fp)
+	flush := i.shouldFlushSeries(series, immediate)
 
 	if flush {
 		flushQueueIndex := int(uint64(fp) % uint64(i.cfg.ConcurrentFlushes))
-		i.flushQueues[flushQueueIndex].Enqueue(&flushOp{lastTime, u.userID, fp, immediate})
+		i.flushQueues[flushQueueIndex].Enqueue(&flushOp{lastTime, userID, fp, immediate})
 	}
 }
 
+func (i *Ingester) shouldFlushSeries(series *memorySeries, immediate bool) bool {
+	// Series should be scheduled for flushing if they have more than one chunk
+	if immediate || len(series.chunkDescs) > 1 {
+		return true
+	}
+
+	// Or if the only existing chunk need flushing
+	if len(series.chunkDescs) > 0 {
+		return i.shouldFlushChunk(series.chunkDescs[0])
+	}
+
+	return false
+}
+
+func (i *Ingester) shouldFlushChunk(c *prom_chunk.Desc) bool {
+	// Chunks should be flushed if their oldest entry is older than MaxChunkAge
+	if model.Now().Sub(c.ChunkFirstTime) > i.cfg.MaxChunkAge {
+		return true
+	}
+
+	// Chunk should be flushed if their last entry is older then MaxChunkIdle
+	if model.Now().Sub(c.ChunkLastTime) > i.cfg.MaxChunkIdle {
+		return true
+	}
+
+	return false
+}
+
 func (i *Ingester) flushLoop(j int) {
 	defer func() {
 		log.Info("Ingester.flushLoop() exited")
@@ -536,55 +564,61 @@ func (i *Ingester) flushLoop(j int) {
 		}
 		op := o.(*flushOp)
 
-		// get the user
-		i.userStateLock.Lock()
-		userState, ok := i.userState[op.userID]
-		i.userStateLock.Unlock()
-		if !ok {
-			continue
+		if err := i.flushUserSeries(op.userID, op.fp, op.immediate); err != nil {
+			log.Errorf("Failed to flush user: %v", err)
 		}
-		ctx := user.WithID(context.Background(), op.userID)
+	}
+}
 
-		// Decide what chunks to flush
-		series, ok := userState.fpToSeries.get(op.fp)
-		if !ok {
-			continue
-		}
+func (i *Ingester) flushUserSeries(userID string, fp model.Fingerprint, immediate bool) error {
+	i.userStateLock.Lock()
+	userState, ok := i.userState[userID]
+	i.userStateLock.Unlock()
+	if !ok {
+		return nil
+	}
 
-		userState.fpLocker.Lock(op.fp)
+	series, ok := userState.fpToSeries.get(fp)
+	if !ok {
+		return nil
+	}
 
-		// Assume we're going to flush everything
-		chunks := series.chunkDescs
+	userState.fpLocker.Lock(fp)
+	if !i.shouldFlushSeries(series, immediate) {
+		userState.fpLocker.Unlock(fp)
+		return nil
+	}
 
-		// If the head chunk is old enough, close it
-		if op.immediate || model.Now().Sub(series.lastTime) > i.cfg.MaxChunkIdle {
-			series.closeHead()
-		} else {
-			chunks = chunks[:len(chunks)-1]
-		}
-		userState.fpLocker.Unlock(op.fp)
+	// Assume we're going to flush everything, and maybe don't flush the head chunk if it doesn't need it.
+	chunks := series.chunkDescs
+	if immediate || (len(chunks) > 0 && i.shouldFlushChunk(chunks[0])) {
+		series.closeHead()
+	} else {
+		chunks = chunks[:len(chunks)-1]
+	}
+	userState.fpLocker.Unlock(fp)
 
-		if len(chunks) == 0 {
-			continue
-		}
+	if len(chunks) == 0 {
+		return nil
+	}
 
-		// flush the chunks without locking the series
-		err := i.flushChunks(ctx, op.fp, series.metric, chunks)
-		if err != nil {
-			log.Errorf("Failed to flush series: %v", err)
-			continue
-		}
+	// flush the chunks without locking the series, as we don't want to hold the series lock for the duration of the dynamo/s3 rpcs.
+	ctx := user.WithID(context.Background(), userID)
+	err := i.flushChunks(ctx, fp, series.metric, chunks)
+	if err != nil {
+		return err
+	}
 
-		// now remove the chunks
-		userState.fpLocker.Lock(op.fp)
-		series.chunkDescs = series.chunkDescs[len(chunks):]
-		i.memoryChunks.Sub(float64(len(chunks)))
-		if len(series.chunkDescs) == 0 {
-			userState.fpToSeries.del(op.fp)
-			userState.index.delete(series.metric, op.fp)
-		}
-		userState.fpLocker.Unlock(op.fp)
+	// now remove the chunks
+	userState.fpLocker.Lock(fp)
+	series.chunkDescs = series.chunkDescs[len(chunks):]
+	i.memoryChunks.Sub(float64(len(chunks)))
+	if len(series.chunkDescs) == 0 {
+		userState.fpToSeries.del(fp)
+		userState.index.delete(series.metric, fp)
 	}
+	userState.fpLocker.Unlock(fp)
+	return nil
 }
 
 func (i *Ingester) flushChunks(ctx context.Context, fp model.Fingerprint, metric model.Metric, chunkDescs []*prom_chunk.Desc) error {