traces/modules/ingester/flush.go

package ingester

import (
	"context"
	"fmt"
	"math/rand"
	"net/http"
	"strconv"
	"time"

	gklog "github.com/go-kit/log"
	"github.com/go-kit/log/level"
	"github.com/google/uuid"
	"github.com/grafana/dskit/services"
	"github.com/grafana/dskit/user"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"

	"github.com/grafana/tempo/pkg/util/log"
)

var (
	metricBlocksFlushed = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "tempo",
		Name:      "ingester_blocks_flushed_total",
		Help:      "The total number of blocks flushed",
	})
	metricFailedFlushes = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "tempo",
		Name:      "ingester_failed_flushes_total",
		Help:      "The total number of failed traces",
	})
	metricFlushRetries = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "tempo",
		Name:      "ingester_flush_retries_total",
		Help:      "The total number of retries after a failed flush",
	})
	metricFlushFailedRetries = promauto.NewCounter(prometheus.CounterOpts{
		Namespace: "tempo",
		Name:      "ingester_flush_failed_retries_total",
		Help:      "The total number of failed retries after a failed flush",
	})
	metricFlushDuration = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace:                       "tempo",
		Name:                            "ingester_flush_duration_seconds",
		Help:                            "Records the amount of time to flush a complete block.",
		Buckets:                         prometheus.ExponentialBuckets(1, 2, 10),
		NativeHistogramBucketFactor:     1.1,
		NativeHistogramMaxBucketNumber:  100,
		NativeHistogramMinResetDuration: 1 * time.Hour,
	})
	metricFlushSize = promauto.NewHistogram(prometheus.HistogramOpts{
		Namespace: "tempo",
		Name:      "ingester_flush_size_bytes",
		Help:      "Size in bytes of blocks flushed.",
		Buckets:   prometheus.ExponentialBuckets(1024*1024, 2, 10), // from 1MB up to 1GB
	})
)

const (
	initialBackoff      = 30 * time.Second
	flushJitter         = 10 * time.Second
	maxBackoff          = 120 * time.Second
	maxCompleteAttempts = 3
)

const (
	opKindComplete = iota
	opKindFlush
)

// ShutdownHandler handles a graceful shutdown for an ingester. It does the following things in order
// * Stop incoming writes by exiting from the ring
// * Flush all blocks to backend
func (i *Ingester) ShutdownHandler(w http.ResponseWriter, _ *http.Request) {
	go func() {
		level.Info(log.Logger).Log("msg", "shutdown handler called")

		// lifecycler should exit the ring on shutdown
		i.lifecycler.SetUnregisterOnShutdown(true)

		// stop accepting new writes
		i.markUnavailable()

		// flush any remaining traces
		i.flushRemaining()

		// stop ingester service
		_ = services.StopAndAwaitTerminated(context.Background(), i)

		level.Info(log.Logger).Log("msg", "shutdown handler complete")
	}()

	_, _ = w.Write([]byte("shutdown job acknowledged"))
}

// FlushHandler calls sweepAllInstances(true) which will force push all traces into the WAL and force
// mark all head blocks as ready to flush. It will either flush all instances or if an instance is specified,
// just that one.
func (i *Ingester) FlushHandler(w http.ResponseWriter, r *http.Request) {
	queryParamInstance := "tenant"

	if r.URL.Query().Has(queryParamInstance) {
		instance, ok := i.getInstanceByID(r.URL.Query().Get(queryParamInstance))
		if !ok {
			w.WriteHeader(http.StatusNotFound)
			return
		}
		level.Info(log.Logger).Log("msg", "flushing instance", "instance", instance.instanceID)
		i.cutOneInstanceToWal(instance, true)
	} else {
		i.cutAllInstancesToWal()
	}

	w.WriteHeader(http.StatusNoContent)
}

type flushOp struct {
	kind     int
	at       time.Time // When to execute
	attempts uint
	backoff  time.Duration
	userID   string
	blockID  uuid.UUID
}

func (o *flushOp) Key() string {
	return o.userID + "/" + strconv.Itoa(o.kind) + "/" + o.blockID.String()
}

// Priority orders entries in the queue. The larger the number the higher the priority, so inverted here to
// prioritize entries with earliest timestamps.
func (o *flushOp) Priority() int64 {
	return -o.at.Unix()
}

// cutToWalLoop kicks off a goroutine for the passed instance that will periodically cut traces to WAL.
// it signals completion through cutToWalWg, waits for cutToWalStart and stops on cutToWalStop.
func (i *Ingester) cutToWalLoop(instance *instance) {
	i.cutToWalWg.Add(1)

	go func() {
		defer i.cutToWalWg.Done()

		// wait for the signal to start. we need the wal to be completely replayed
		// before we start cutting to WAL
		select {
		case <-i.cutToWalStart:
		case <-i.cutToWalStop:
			return
		}

		// ticker
		ticker := time.NewTicker(i.cfg.FlushCheckPeriod)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				i.cutOneInstanceToWal(instance, false)
			case <-i.cutToWalStop:
				return
			}
		}
	}()
}

// cutAllInstancesToWal periodically schedules series for flushing and garbage collects instances with no series
func (i *Ingester) cutAllInstancesToWal() {
	instances := i.getInstances()

	for _, instance := range instances {
		i.cutOneInstanceToWal(instance, true)
	}
}

func (i *Ingester) cutOneInstanceToWal(instance *instance, immediate bool) {
	// cut traces internally
	err := instance.CutCompleteTraces(i.cfg.MaxTraceIdle, i.cfg.MaxTraceLive, immediate)
	if err != nil {
		level.Error(log.WithUserID(instance.instanceID, log.Logger)).Log("msg", "failed to cut traces", "err", err)
		return
	}

	// see if it's ready to cut a block
	blockID, err := instance.CutBlockIfReady(i.cfg.MaxBlockDuration, i.cfg.MaxBlockBytes, immediate)
	if err != nil {
		level.Error(log.WithUserID(instance.instanceID, log.Logger)).Log("msg", "failed to cut block", "err", err)
		return
	}

	if blockID != uuid.Nil {
		level.Info(log.Logger).Log("msg", "enqueueing flush op", "tenant", instance.instanceID, "block", blockID)
		// jitter to help when flushing many instances at the same time
		// no jitter if immediate (initiated via /flush handler for example)
		i.enqueue(&flushOp{
			kind:    opKindComplete,
			userID:  instance.instanceID,
			blockID: blockID,
		}, !immediate)
	}

	// dump any blocks that have been flushed for awhile
	err = instance.ClearOldBlocks(i.cfg.FlushObjectStorage, i.cfg.CompleteBlockTimeout)
	if err != nil {
		level.Error(log.WithUserID(instance.instanceID, log.Logger)).Log("msg", "failed to complete block", "err", err)
	}
}

func (i *Ingester) flushLoop(j int) {
	defer func() {
		level.Debug(log.Logger).Log("msg", "Ingester.flushLoop() exited")
		i.flushQueuesDone.Done()
	}()

	for {
		op := i.flushQueues.Dequeue(j)
		if op == nil {
			return
		}

		op.attempts++

		var retry bool
		var err error

		if op.kind == opKindComplete {
			retry, err = i.handleComplete(context.Background(), op)
		} else {
			retry, err = i.handleFlush(context.Background(), op.userID, op.blockID)
		}

		if err != nil {
			handleFailedOp(op, err)
		}

		if retry {
			i.requeue(op)
		} else {
			i.flushQueues.Clear(op)
		}
	}
}

func handleFailedOp(op *flushOp, err error) {
	level.Error(log.WithUserID(op.userID, log.Logger)).Log("msg", "error performing op in flushQueue",
		"op", op.kind, "block", op.blockID.String(), "attempts", op.attempts, "err", err)
	metricFailedFlushes.Inc()

	if op.attempts > 1 {
		metricFlushFailedRetries.Inc()
	}
}

func handleAbandonedOp(op *flushOp) {
	level.Info(log.WithUserID(op.userID, log.Logger)).Log("msg", "Abandoning op in flush queue because ingester is shutting down",
		"op", op.kind, "block", op.blockID.String(), "attempts", op.attempts)
}

func (i *Ingester) handleComplete(ctx context.Context, op *flushOp) (retry bool, err error) {
	ctx, sp := tracer.Start(ctx, "ingester.Complete", trace.WithAttributes(attribute.String("tenant", op.userID), attribute.String("blockID", op.blockID.String())))
	defer sp.End()
	withSpan(level.Info(log.Logger), sp).Log("msg", "completing block", "tenant", op.userID, "block", op.blockID.String())

	// No point in proceeding if shutdown has been initiated since
	// we won't be able to queue up the next flush op
	if i.flushQueues.IsStopped() {
		handleAbandonedOp(op)
		return false, nil
	}

	start := time.Now()
	instance, err := i.getOrCreateInstance(op.userID)
	if err != nil {
		return false, err
	}

	err = instance.CompleteBlock(ctx, op.blockID)
	level.Info(log.Logger).Log("msg", "block completed", "tenant", op.userID, "blockID", op.blockID, "duration", time.Since(start))
	if err != nil {
		handleFailedOp(op, err)

		if op.attempts >= maxCompleteAttempts {
			level.Error(log.WithUserID(op.userID, log.Logger)).Log("msg", "Block exceeded max completion errors. Deleting. POSSIBLE DATA LOSS",
				"tenant", op.userID, "attempts", op.attempts, "block", op.blockID.String())

			// Delete WAL and move on
			err = instance.ClearCompletingBlock(op.blockID)
			return false, err
		}

		return true, nil
	}

	err = instance.ClearCompletingBlock(op.blockID)
	if err != nil {
		return false, fmt.Errorf("error clearing completing block: %w", err)
	}

	// add a flushOp for the block we just completed
	// No delay
	if i.cfg.FlushObjectStorage {
		i.enqueue(&flushOp{
			kind:    opKindFlush,
			userID:  instance.instanceID,
			blockID: op.blockID,
		}, false)
	}

	return false, nil
}

// withSpan adds traceID to a logger, if span is sampled
// TODO: move into some central trace/log package
func withSpan(logger gklog.Logger, sp trace.Span) gklog.Logger {
	if sp == nil {
		return logger
	}
	sctx := sp.SpanContext()
	if !sctx.IsSampled() {
		return logger
	}

	return gklog.With(logger, "traceID", sctx.TraceID().String())
}

func (i *Ingester) handleFlush(ctx context.Context, userID string, blockID uuid.UUID) (retry bool, err error) {
	ctx, sp := tracer.Start(ctx, "ingester.Flush", trace.WithAttributes(attribute.String("tenant", userID), attribute.String("blockID", blockID.String())))
	defer sp.End()
	withSpan(level.Info(log.Logger), sp).Log("msg", "flushing block", "tenant", userID, "block", blockID.String())

	instance, err := i.getOrCreateInstance(userID)
	if err != nil {
		return true, err
	}

	if instance == nil {
		return false, fmt.Errorf("instance id %s not found", userID)
	}

	if block := instance.GetBlockToBeFlushed(blockID); block != nil {
		ctx := user.InjectOrgID(ctx, userID)
		ctx, cancel := context.WithTimeout(ctx, i.cfg.FlushOpTimeout)
		defer cancel()

		start := time.Now()
		err = i.store.WriteBlock(ctx, block)
		metricFlushDuration.Observe(time.Since(start).Seconds())
		metricFlushSize.Observe(float64(block.BlockMeta().Size_))
		if err != nil {
			sp.SetStatus(codes.Error, "")
			sp.RecordError(err)
			return true, err
		}

		metricBlocksFlushed.Inc()
	} else {
		return false, fmt.Errorf("error getting block to flush")
	}

	return false, nil
}

func (i *Ingester) enqueueExec(op *flushOp) {
	// Check if shutdown initiated
	if i.flushQueues.IsStopped() {
		handleAbandonedOp(op)
		return
	}

	err := i.flushQueues.Enqueue(op)
	if err != nil {
		handleFailedOp(op, err)
	}
}

func (i *Ingester) enqueue(op *flushOp, jitter bool) {
	delay := time.Duration(0)

	if jitter {
		delay = time.Duration(rand.Float32() * float32(flushJitter))
	}

	op.at = time.Now().Add(delay)

	if !jitter {
		// Execute synchronously to make sure we can flush during shutdown
		i.enqueueExec(op)
		return
	}

	go func() {
		time.Sleep(delay)
		i.enqueueExec(op)
	}()
}

func (i *Ingester) requeue(op *flushOp) {
	op.backoff *= 2
	if op.backoff < initialBackoff {
		op.backoff = initialBackoff
	}
	if op.backoff > maxBackoff {
		op.backoff = maxBackoff
	}

	op.at = time.Now().Add(op.backoff)

	level.Info(log.WithUserID(op.userID, log.Logger)).Log("msg", "retrying op in flushQueue",
		"op", op.kind, "block", op.blockID.String(), "backoff", op.backoff)

	go func() {
		time.Sleep(op.backoff)

		// Check if shutdown initiated
		if i.flushQueues.IsStopped() {
			handleAbandonedOp(op)
			return
		}

		metricFlushRetries.Inc()

		err := i.flushQueues.Requeue(op)
		if err != nil {
			handleFailedOp(op, err)
		}
	}()
}