Ingester block lifetime changes

CHANGELOG.md

@@ -19,6 +19,7 @@
 * [BUGFIX] Fixes issue where wal was deleted before successful flush and adds exponential backoff for flush errors [#593](https://github.com/grafana/tempo/pull/593)
 * [BUGFIX] Fixes issue where Tempo would not parse odd length trace ids [#605](https://github.com/grafana/tempo/pull/605)
 * [BUGFIX] Sort traces on flush to reduce unexpected recombination work by compactors [#606](https://github.com/grafana/tempo/pull/606)
+* [BUGFIX] Ingester fully persists blocks locally to reduce amount of work done after restart [#628](https://github.com/grafana/tempo/pull/628)
 * [ENHANCEMENT] Add kafka receiver. [#613](https://github.com/grafana/tempo/pull/613)
 
 ## v0.6.0

modules/compactor/config.go

@@ -21,7 +21,7 @@ type Config struct {
 func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet) {
 	cfg.Compactor = tempodb.CompactorConfig{
 		ChunkSizeBytes:          10 * 1024 * 1024, // 10 MiB
-		FlushSizeBytes:          30 * 1024 * 1024, // 30 MiB
+		FlushSizeBytes:          tempodb.DefaultFlushSizeBytes,
 		CompactedBlockRetention: time.Hour,
 		RetentionConcurrency:    tempodb.DefaultRetentionConcurrency,
 	}

modules/ingester/flush.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/pkg/errors"
 
 	"github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
@@ -200,20 +201,26 @@ func (i *Ingester) flushLoop(j int) {
 
 					err = instance.ClearCompletingBlock(op.blockID)
 					if err != nil {
-						// Failure to delete the WAL doesn't prevent flushing the bloc
+						// Failure to delete the WAL doesn't prevent flushing the block
 						handleFailedOp(op, err)
 					}
 				} else {
 					retry = true
 				}
 			} else {
-				// add a flushOp for the block we just completed
-				// No delay
-				i.enqueue(&flushOp{
-					kind:    opKindFlush,
-					userID:  instance.instanceID,
-					blockID: op.blockID,
-				}, false)
+				err = instance.ClearCompletingBlock(op.blockID)
+				if err != nil {
+					err = errors.Wrap(err, "error clearing completing block")
+					handleFailedOp(op, err)
+				} else {
+					// add a flushOp for the block we just completed
+					// No delay
+					i.enqueue(&flushOp{
+						kind:    opKindFlush,
+						userID:  instance.instanceID,
+						blockID: op.blockID,
+					}, false)
+				}
 			}
 
 		} else {
@@ -268,12 +275,12 @@ func (i *Ingester) flushBlock(userID string, blockID uuid.UUID) error {
 		}
 
 		// Delete original wal only after successful flush
-		err = instance.ClearCompletingBlock(blockID)
-		if err != nil {
-			// Error deleting wal doesn't fail the flush
-			level.Error(log.Logger).Log("msg", "Error clearing wal", "userID", userID, "blockID", blockID.String(), "err", err)
-			metricFailedFlushes.Inc()
-		}
+		//err = instance.ClearCompletingBlock(blockID)
+		//if err != nil {
+		//	// Error deleting wal doesn't fail the flush
+		//	level.Error(log.Logger).Log("msg", "Error clearing wal", "userID", userID, "blockID", blockID.String(), "err", err)
+		//	metricFailedFlushes.Inc()
+		//}
 
 		metricBlocksFlushed.Inc()
 	} else {

modules/ingester/ingester.go

@@ -2,27 +2,27 @@ package ingester
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"sync"
 	"time"
 
 	"github.com/go-kit/kit/log/level"
 	"github.com/opentracing/opentracing-go"
 	ot_log "github.com/opentracing/opentracing-go/log"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/weaveworks/common/user"
 
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
-
 	"github.com/grafana/tempo/modules/overrides"
 	"github.com/grafana/tempo/modules/storage"
 	"github.com/grafana/tempo/pkg/flushqueues"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/validation"
+	"github.com/grafana/tempo/tempodb/backend/local"
 	tempodb_wal "github.com/grafana/tempo/tempodb/wal"
 )
 
@@ -48,6 +48,10 @@ type Ingester struct {
 
 	lifecycler *ring.Lifecycler
 	store      storage.Store
+	local      *local.Backend
+	//localReader    backend.Reader
+	//localWriter    backend.Writer
+	//localCompactor backend.Compactor
 
 	flushQueues     *flushqueues.ExclusiveQueues
 	flushQueuesDone sync.WaitGroup
@@ -66,12 +70,19 @@ func New(cfg Config, store storage.Store, limits *overrides.Overrides) (*Ingeste
 		flushQueues: flushqueues.New(cfg.ConcurrentFlushes, metricFlushQueueLength),
 	}
 
+	l, err := local.NewBackend(&local.Config{
+		Path: store.WAL().BlocksFilePath(),
+	})
+	if err != nil {
+		return nil, err
+	}
+	i.local = l
+
 	i.flushQueuesDone.Add(cfg.ConcurrentFlushes)
 	for j := 0; j < cfg.ConcurrentFlushes; j++ {
 		go i.flushLoop(j)
 	}
 
-	var err error
 	i.lifecycler, err = ring.NewLifecycler(cfg.LifecyclerConfig, i, "ingester", cfg.OverrideRingKey, true, prometheus.DefaultRegisterer)
 	if err != nil {
 		return nil, fmt.Errorf("NewLifecycler failed %w", err)
@@ -94,6 +105,11 @@ func (i *Ingester) starting(ctx context.Context) error {
 		return fmt.Errorf("failed to replay wal %w", err)
 	}
 
+	err = i.rediscoverLocalBlocks()
+	if err != nil {
+		return fmt.Errorf("failed to rediscover local blocks %w", err)
+	}
+
 	// Now that user states have been created, we can start the lifecycler.
 	// Important: we want to keep lifecycler running until we ask it to stop, so we need to give it independent context
 	if err := i.lifecycler.StartAsync(context.Background()); err != nil {
@@ -236,7 +252,7 @@ func (i *Ingester) getOrCreateInstance(instanceID string) (*instance, error) {
 	inst, ok = i.instances[instanceID]
 	if !ok {
 		var err error
-		inst, err = newInstance(instanceID, i.limiter, i.store)
+		inst, err = newInstance(instanceID, i.limiter, i.store, i.local)
 		if err != nil {
 			return nil, err
 		}
@@ -351,3 +367,39 @@ func (i *Ingester) replayBlock(b *tempodb_wal.ReplayBlock) error {
 
 	return nil
 }
+
+func (i *Ingester) rediscoverLocalBlocks() error {
+	ctx := context.TODO()
+
+	tenants, err := i.local.Tenants(ctx)
+	if err != nil {
+		return errors.Wrap(err, "getting local tenants")
+	}
+
+	level.Info(log.Logger).Log("msg", "reloading local blocks", "tenants", len(tenants))
+
+	for _, t := range tenants {
+		inst, err := i.getOrCreateInstance(t)
+		if err != nil {
+			return err
+		}
+
+		err = inst.rediscoverLocalBlocks(ctx)
+		if err != nil {
+			return errors.Wrapf(err, "getting local blocks for tenant %v", t)
+		}
+
+		// Requeue needed flushes
+		for _, b := range inst.completeBlocks {
+			if b.FlushedTime().IsZero() {
+				i.enqueue(&flushOp{
+					kind:    opKindFlush,
+					userID:  t,
+					blockID: b.BlockMeta().BlockID,
+				}, true)
+			}
+		}
+	}
+
+	return nil
+}

modules/ingester/instance.go

@@ -22,6 +22,8 @@ import (
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/util"
 	"github.com/grafana/tempo/tempodb"
+	"github.com/grafana/tempo/tempodb/backend"
+	"github.com/grafana/tempo/tempodb/backend/local"
 	"github.com/grafana/tempo/tempodb/encoding"
 	"github.com/grafana/tempo/tempodb/encoding/common"
 	"github.com/grafana/tempo/tempodb/wal"
@@ -57,7 +59,7 @@ type instance struct {
 	blocksMtx        sync.RWMutex
 	headBlock        *wal.AppendBlock
 	completingBlocks []*wal.AppendBlock
-	completeBlocks   []*encoding.CompleteBlock
+	completeBlocks   []*LocalBlock
 
 	lastBlockCut time.Time
 
@@ -66,11 +68,12 @@ type instance struct {
 	bytesWrittenTotal  prometheus.Counter
 	limiter            *Limiter
 	writer             tempodb.Writer
+	local              *local.Backend
 
 	hash hash.Hash32
 }
 
-func newInstance(instanceID string, limiter *Limiter, writer tempodb.Writer) (*instance, error) {
+func newInstance(instanceID string, limiter *Limiter, writer tempodb.Writer, l *local.Backend) (*instance, error) {
 	i := &instance{
 		traces: map[uint32]*trace{},
 
@@ -79,6 +82,7 @@ func newInstance(instanceID string, limiter *Limiter, writer tempodb.Writer) (*i
 		bytesWrittenTotal:  metricBytesWrittenTotal.WithLabelValues(instanceID),
 		limiter:            limiter,
 		writer:             writer,
+		local:              l,
 
 		hash: fnv.New32(),
 	}
@@ -180,16 +184,20 @@ func (i *instance) CompleteBlock(blockID uuid.UUID) error {
 		return fmt.Errorf("error finding completingBlock")
 	}
 
-	// potentially long running operation placed outside blocksMtx
-	completeBlock, err := i.writer.CompleteBlock(completingBlock, i)
+	ctx := context.Background()
+
+	backendBlock, err := i.writer.CompleteBlockWithBackend(ctx, completingBlock, i, i.local, i.local)
 	if err != nil {
-		metricFailedFlushes.Inc()
-		level.Error(log.Logger).Log("msg", "unable to complete block.", "tenantID", i.instanceID, "err", err)
-		return err
+		return errors.Wrap(err, "error completing wal block with local backend")
+	}
+
+	ingesterBlock, err := NewIngesterBlock(ctx, backendBlock, i.local)
+	if err != nil {
+		return errors.Wrap(err, "error creating ingester block")
 	}
 
 	i.blocksMtx.Lock()
-	i.completeBlocks = append(i.completeBlocks, completeBlock)
+	i.completeBlocks = append(i.completeBlocks, ingesterBlock)
 	i.blocksMtx.Unlock()
 
 	return nil
@@ -216,7 +224,7 @@ func (i *instance) ClearCompletingBlock(blockID uuid.UUID) error {
 }
 
 // GetBlockToBeFlushed gets a list of blocks that can be flushed to the backend
-func (i *instance) GetBlockToBeFlushed(blockID uuid.UUID) *encoding.CompleteBlock {
+func (i *instance) GetBlockToBeFlushed(blockID uuid.UUID) *LocalBlock {
 	i.blocksMtx.Lock()
 	defer i.blocksMtx.Unlock()
 
@@ -243,7 +251,8 @@ func (i *instance) ClearFlushedBlocks(completeBlockTimeout time.Duration) error
 
 		if flushedTime.Add(completeBlockTimeout).Before(time.Now()) {
 			i.completeBlocks = append(i.completeBlocks[:idx], i.completeBlocks[idx+1:]...)
-			err = b.Clear() // todo: don't remove from complete blocks slice until after clear succeeds?
+			//err = b.Clear() // todo: don't remove from complete blocks slice until after clear succeeds?
+			err = i.local.ClearBlock(b.BlockMeta().BlockID, i.instanceID)
 			if err == nil {
 				metricBlocksClearedTotal.Inc()
 			}
@@ -291,7 +300,7 @@ func (i *instance) FindTraceByID(id []byte) (*tempopb.Trace, error) {
 
 	// completeBlock
 	for _, c := range i.completeBlocks {
-		foundBytes, err = c.Find(id, i)
+		foundBytes, err = c.Find(context.TODO(), id)
 		if err != nil {
 			return nil, fmt.Errorf("completeBlock.Find failed: %w", err)
 		}
@@ -404,3 +413,45 @@ func pushRequestTraceID(req *tempopb.PushRequest) ([]byte, error) {
 
 	return req.Batch.InstrumentationLibrarySpans[0].Spans[0].TraceId, nil
 }
+
+func (i *instance) rediscoverLocalBlocks(ctx context.Context) error {
+	ids, err := i.local.Blocks(ctx, i.instanceID)
+	if err != nil {
+		return err
+	}
+
+	for _, id := range ids {
+		meta, err := i.local.BlockMeta(ctx, id, i.instanceID)
+		if err != nil {
+			if err == backend.ErrMetaDoesNotExist {
+				// Partial/incomplete block found, remove, it will be recreated from data in the wal.
+				level.Warn(log.Logger).Log("msg", "Unable to reload meta for local block. This indicates an incomplete block and will be deleted", "tenant", i.instanceID, "block", id.String())
+				err = i.local.ClearBlock(id, i.instanceID)
+				if err != nil {
+					return errors.Wrapf(err, "deleting bad local block tenant %v block %v", i.instanceID, id.String())
+				}
+				continue
+			}
+
+			return err
+		}
+
+		b, err := encoding.NewBackendBlock(meta, i.local)
+		if err != nil {
+			return err
+		}
+
+		ib, err := NewIngesterBlock(ctx, b, i.local)
+		if err != nil {
+			return err
+		}
+
+		i.blocksMtx.Lock()
+		i.completeBlocks = append(i.completeBlocks, ib)
+		i.blocksMtx.Unlock()
+
+		level.Info(log.Logger).Log("msg", "reloaded local block", "tenantID", i.instanceID, "block", id.String(), "flushed", ib.FlushedTime())
+	}
+
+	return nil
+}