Add storage to the local-blocks processor and flush RF1 blocks

cmd/tempo/app/modules.go

@@ -263,7 +263,7 @@ func (t *App) initIngester() (services.Service, error) {
 
 func (t *App) initGenerator() (services.Service, error) {
 	t.cfg.Generator.Ring.ListenPort = t.cfg.Server.GRPCListenPort
-	genSvc, err := generator.New(&t.cfg.Generator, t.Overrides, prometheus.DefaultRegisterer, log.Logger)
+	genSvc, err := generator.New(&t.cfg.Generator, t.Overrides, prometheus.DefaultRegisterer, t.store, log.Logger)
 	if errors.Is(err, generator.ErrUnconfigured) && t.cfg.Target != MetricsGenerator { // just warn if we're not running the metrics-generator
 		level.Warn(log.Logger).Log("msg", "metrics-generator is not configured.", "err", err)
 		return services.NewIdleService(nil, nil), nil
@@ -576,7 +576,7 @@ func (t *App) setupModuleManager() error {
 		QueryFrontend:    {Common, Store, OverridesAPI},
 		Distributor:      {Common, IngesterRing, MetricsGeneratorRing},
 		Ingester:         {Common, Store, MemberlistKV},
-		MetricsGenerator: {Common, MemberlistKV},
+		MetricsGenerator: {Common, Store, MemberlistKV},
 		Querier:          {Common, Store, IngesterRing, MetricsGeneratorRing, SecondaryIngesterRing},
 		Compactor:        {Common, Store, MemberlistKV},
 		// composite targets

modules/generator/generator.go

@@ -19,6 +19,7 @@ import (
 	"go.uber.org/atomic"
 
 	"github.com/grafana/tempo/modules/generator/storage"
+	objStorage "github.com/grafana/tempo/modules/storage"
 	"github.com/grafana/tempo/pkg/tempopb"
 	tempodb_wal "github.com/grafana/tempo/tempodb/wal"
 )
@@ -52,6 +53,8 @@ type Generator struct {
 	subservices        *services.Manager
 	subservicesWatcher *services.FailureWatcher
 
+	store objStorage.Store
+
 	// When set to true, the generator will refuse incoming pushes
 	// and will flush any remaining metrics.
 	readOnly atomic.Bool
@@ -61,7 +64,7 @@ type Generator struct {
 }
 
 // New makes a new Generator.
-func New(cfg *Config, overrides metricsGeneratorOverrides, reg prometheus.Registerer, logger log.Logger) (*Generator, error) {
+func New(cfg *Config, overrides metricsGeneratorOverrides, reg prometheus.Registerer, store objStorage.Store, logger log.Logger) (*Generator, error) {
 	if cfg.Storage.Path == "" {
 		return nil, ErrUnconfigured
 	}
@@ -77,6 +80,8 @@ func New(cfg *Config, overrides metricsGeneratorOverrides, reg prometheus.Regist
 
 		instances: map[string]*instance{},
 
+		store: store,
+
 		reg:    reg,
 		logger: logger,
 	}
@@ -264,20 +269,20 @@ func (g *Generator) createInstance(id string) (*instance, error) {
 
 		tracesWAL, err = tempodb_wal.New(&tracesWALCfg)
 		if err != nil {
-			wal.Close()
+			_ = wal.Close()
 			return nil, err
 		}
 	}
 
-	inst, err := newInstance(g.cfg, id, g.overrides, wal, reg, g.logger, tracesWAL)
+	inst, err := newInstance(g.cfg, id, g.overrides, wal, reg, g.logger, tracesWAL, g.store)
 	if err != nil {
-		wal.Close()
+		_ = wal.Close()
 		return nil, err
 	}
 
 	err = g.reg.Register(reg)
 	if err != nil {
-		wal.Close()
+		inst.shutdown()
 		return nil, err
 	}
 

modules/generator/generator_test.go

@@ -55,7 +55,7 @@ overrides:
 	generatorConfig.Storage.Path = t.TempDir()
 	generatorConfig.Ring.KVStore.Store = "inmemory"
 	generatorConfig.Processor.SpanMetrics.RegisterFlagsAndApplyDefaults("", nil)
-	g, err := New(generatorConfig, o, prometheus.NewRegistry(), newTestLogger(t))
+	g, err := New(generatorConfig, o, prometheus.NewRegistry(), nil, newTestLogger(t))
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), g))
 

modules/generator/instance.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/grafana/tempo/tempodb"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"golang.org/x/exp/maps"
@@ -75,6 +76,7 @@ type instance struct {
 	wal      storage.Storage
 
 	traceWAL *wal.WAL
+	writer   tempodb.Writer
 
 	// processorsMtx protects the processors map, not the processors itself
 	processorsMtx sync.RWMutex
@@ -88,7 +90,7 @@ type instance struct {
 	logger log.Logger
 }
 
-func newInstance(cfg *Config, instanceID string, overrides metricsGeneratorOverrides, wal storage.Storage, reg prometheus.Registerer, logger log.Logger, traceWAL *wal.WAL) (*instance, error) {
+func newInstance(cfg *Config, instanceID string, overrides metricsGeneratorOverrides, wal storage.Storage, reg prometheus.Registerer, logger log.Logger, traceWAL *wal.WAL, writer tempodb.Writer) (*instance, error) {
 	logger = log.With(logger, "tenant", instanceID)
 
 	i := &instance{
@@ -99,6 +101,7 @@ func newInstance(cfg *Config, instanceID string, overrides metricsGeneratorOverr
 		registry: registry.New(&cfg.Registry, overrides, instanceID, wal, logger),
 		wal:      wal,
 		traceWAL: traceWAL,
+		writer:   writer,
 
 		processors: make(map[string]processor.Processor),
 
@@ -296,7 +299,7 @@ func (i *instance) addProcessor(processorName string, cfg ProcessorConfig) error
 	case servicegraphs.Name:
 		newProcessor = servicegraphs.New(cfg.ServiceGraphs, i.instanceID, i.registry, i.logger)
 	case localblocks.Name:
-		p, err := localblocks.New(cfg.LocalBlocks, i.instanceID, i.traceWAL, i.overrides)
+		p, err := localblocks.New(cfg.LocalBlocks, i.instanceID, i.traceWAL, i.writer, i.overrides)
 		if err != nil {
 			return err
 		}

modules/generator/instance_test.go

@@ -35,10 +35,10 @@ func Test_instance_concurrency(t *testing.T) {
 		servicegraphs.Name: {},
 	}
 
-	instance1, err := newInstance(&Config{}, "test", overrides, &noopStorage{}, prometheus.DefaultRegisterer, log.NewNopLogger(), nil)
+	instance1, err := newInstance(&Config{}, "test", overrides, &noopStorage{}, prometheus.DefaultRegisterer, log.NewNopLogger(), nil, nil)
 	assert.NoError(t, err)
 
-	instance2, err := newInstance(&Config{}, "test", overrides, &noopStorage{}, prometheus.DefaultRegisterer, log.NewNopLogger(), nil)
+	instance2, err := newInstance(&Config{}, "test", overrides, &noopStorage{}, prometheus.DefaultRegisterer, log.NewNopLogger(), nil, nil)
 	assert.NoError(t, err)
 
 	end := make(chan struct{})
@@ -89,7 +89,7 @@ func Test_instance_updateProcessors(t *testing.T) {
 	logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stdout))
 	overrides := mockOverrides{}
 
-	instance, err := newInstance(&cfg, "test", &overrides, &noopStorage{}, prometheus.DefaultRegisterer, logger, nil)
+	instance, err := newInstance(&cfg, "test", &overrides, &noopStorage{}, prometheus.DefaultRegisterer, logger, nil, nil)
 	assert.NoError(t, err)
 
 	// stop the update goroutine
@@ -293,7 +293,7 @@ func Test_instanceQueryRangeTraceQLToProto(t *testing.T) {
 	logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stdout))
 	overrides := mockOverrides{}
 
-	instance, err := newInstance(&cfg, "test", &overrides, &noopStorage{}, prometheus.DefaultRegisterer, logger, nil)
+	instance, err := newInstance(&cfg, "test", &overrides, &noopStorage{}, prometheus.DefaultRegisterer, logger, nil, nil)
 	assert.NoError(t, err)
 
 	req := &tempopb.QueryRangeRequest{

modules/generator/processor/localblocks/processor.go

@@ -12,6 +12,8 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/golang/groupcache/lru"
 	"github.com/google/uuid"
+	"github.com/grafana/tempo/modules/ingester"
+	"github.com/grafana/tempo/tempodb"
 	"github.com/opentracing/opentracing-go"
 	"go.uber.org/atomic"
 
@@ -52,17 +54,19 @@ type Processor struct {
 	blocksMtx      sync.RWMutex
 	headBlock      common.WALBlock
 	walBlocks      map[uuid.UUID]common.WALBlock
-	completeBlocks map[uuid.UUID]common.BackendBlock
+	completeBlocks map[uuid.UUID]*ingester.LocalBlock
 	lastCutTime    time.Time
 
 	liveTracesMtx sync.Mutex
 	liveTraces    *liveTraces
 	traceSizes    *traceSizes
+
+	writer tempodb.Writer
 }
 
 var _ gen.Processor = (*Processor)(nil)
 
-func New(cfg Config, tenant string, wal *wal.WAL, overrides ProcessorOverrides) (p *Processor, err error) {
+func New(cfg Config, tenant string, wal *wal.WAL, writer tempodb.Writer, overrides ProcessorOverrides) (p *Processor, err error) {
 	if wal == nil {
 		return nil, errors.New("local blocks processor requires traces wal")
 	}
@@ -82,23 +86,25 @@ func New(cfg Config, tenant string, wal *wal.WAL, overrides ProcessorOverrides)
 		overrides:      overrides,
 		enc:            enc,
 		walBlocks:      map[uuid.UUID]common.WALBlock{},
-		completeBlocks: map[uuid.UUID]common.BackendBlock{},
+		completeBlocks: map[uuid.UUID]*ingester.LocalBlock{},
 		liveTraces:     newLiveTraces(),
 		traceSizes:     newTraceSizes(),
 		closeCh:        make(chan struct{}),
 		wg:             sync.WaitGroup{},
 		cache:          lru.New(100),
+		writer:         writer,
 	}
 
 	err = p.reloadBlocks()
 	if err != nil {
 		return nil, fmt.Errorf("replaying blocks: %w", err)
 	}
 
-	p.wg.Add(4)
+	p.wg.Add(5)
+	go p.cutLoop()
+	go p.completeLoop()
 	go p.flushLoop()
 	go p.deleteLoop()
-	go p.completeLoop()
 	go p.metricLoop()
 
 	return p, nil
@@ -173,7 +179,7 @@ func (p *Processor) Shutdown(context.Context) {
 	}
 }
 
-func (p *Processor) flushLoop() {
+func (p *Processor) cutLoop() {
 	defer p.wg.Done()
 
 	flushTicker := time.NewTicker(p.Cfg.FlushCheckPeriod)
@@ -238,6 +244,26 @@ func (p *Processor) completeLoop() {
 	}
 }
 
+func (p *Processor) flushLoop() {
+	defer p.wg.Done()
+
+	ticker := time.NewTicker(10 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+			err := p.flushBlock()
+			if err != nil {
+				level.Error(log.WithUserID(p.tenant, log.Logger)).Log("msg", "local blocks processor failed to flush a block", "err", err)
+			}
+
+		case <-p.closeCh:
+			return
+		}
+	}
+}
+
 func (p *Processor) metricLoop() {
 	defer p.wg.Done()
 
@@ -299,7 +325,7 @@ func (p *Processor) completeBlock() error {
 	p.blocksMtx.Lock()
 	defer p.blocksMtx.Unlock()
 
-	p.completeBlocks[newMeta.BlockID] = newBlock
+	p.completeBlocks[newMeta.BlockID] = ingester.NewLocalBlock(ctx, newBlock, p.wal.LocalBackend())
 
 	err = b.Clear()
 	if err != nil {
@@ -310,6 +336,27 @@ func (p *Processor) completeBlock() error {
 	return nil
 }
 
+func (p *Processor) flushBlock() error {
+	var firstCompleteBlock *ingester.LocalBlock
+	p.blocksMtx.RLock()
+	for _, e := range p.completeBlocks {
+		firstCompleteBlock = e
+		break
+	}
+	p.blocksMtx.RUnlock()
+
+	if firstCompleteBlock == nil {
+		return nil
+	}
+
+	ctx := context.Background()
+	err := p.writer.WriteBlock(ctx, firstCompleteBlock)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
 func (p *Processor) GetMetrics(ctx context.Context, req *tempopb.SpanMetricsRequest) (*tempopb.SpanMetricsResponse, error) {
 	p.blocksMtx.RLock()
 	defer p.blocksMtx.RUnlock()
@@ -558,13 +605,21 @@ func (p *Processor) deleteOldBlocks() (err error) {
 	}
 
 	for id, b := range p.completeBlocks {
-		if b.BlockMeta().EndTime.Before(before) {
+		flushedTime := b.FlushedTime()
+		if flushedTime.IsZero() {
+			continue
+		}
+
+		if flushedTime.Add(p.Cfg.CompleteBlockTimeout).Before(time.Now()) {
 			err = p.wal.LocalBackend().ClearBlock(id, p.tenant)
 			if err != nil {
 				return err
 			}
 			delete(p.completeBlocks, id)
 		}
+
+		//if b.BlockMeta().EndTime.Before(before) {
+		//}
 	}
 
 	return
@@ -738,7 +793,8 @@ func (p *Processor) reloadBlocks() error {
 			return err
 		}
 
-		p.completeBlocks[id] = blk
+		lb := ingester.NewLocalBlock(ctx, blk, l)
+		p.completeBlocks[id] = lb
 	}
 
 	return nil

modules/generator/processor/localblocks/processor_test.go

@@ -6,6 +6,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/tempo/tempodb"
+
 	"github.com/grafana/tempo/pkg/tempopb"
 	v1 "github.com/grafana/tempo/pkg/tempopb/trace/v1"
 	"github.com/grafana/tempo/pkg/util/test"
@@ -32,6 +34,22 @@ func (m *mockOverrides) UnsafeQueryHints(string) bool {
 	return false
 }
 
+var _ tempodb.Writer = (*mockWriter)(nil)
+
+type mockWriter struct{}
+
+func (m *mockWriter) WriteBlock(context.Context, tempodb.WriteableBlock) error { return nil }
+
+func (m *mockWriter) CompleteBlock(context.Context, common.WALBlock) (common.BackendBlock, error) {
+	return nil, nil
+}
+
+func (m *mockWriter) CompleteBlockWithBackend(context.Context, common.WALBlock, backend.Reader, backend.Writer) (common.BackendBlock, error) {
+	return nil, nil
+}
+
+func (m *mockWriter) WAL() *wal.WAL { return nil }
+
 func TestProcessorDoesNotRace(t *testing.T) {
 	wal, err := wal.New(&wal.Config{
 		Filepath: t.TempDir(),
@@ -59,7 +77,7 @@ func TestProcessorDoesNotRace(t *testing.T) {
 		overrides = &mockOverrides{}
 	)
 
-	p, err := New(cfg, tenant, wal, overrides)
+	p, err := New(cfg, tenant, wal, &mockWriter{}, overrides)
 	require.NoError(t, err)
 
 	var (
@@ -112,6 +130,11 @@ func TestProcessorDoesNotRace(t *testing.T) {
 		require.NoError(t, err, "completing block")
 	})
 
+	go concurrent(func() {
+		err := p.flushBlock()
+		require.NoError(t, err, "flushing blocks")
+	})
+
 	go concurrent(func() {
 		err := p.deleteOldBlocks()
 		require.NoError(t, err, "deleting old blocks")
@@ -176,7 +199,7 @@ func TestReplicationFactor(t *testing.T) {
 		FilterServerSpans: false,
 	}
 
-	p, err := New(cfg, "fake", wal, &mockOverrides{})
+	p, err := New(cfg, "fake", wal, &mockWriter{}, &mockOverrides{})
 	require.NoError(t, err)
 
 	tr := test.MakeTrace(10, []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9})