Parquet WAL

CHANGELOG.md

@@ -71,6 +71,7 @@ Internal types are updated to use `scope` instead of `instrumentation_library`.
 * [ENHANCEMENT] metrics-generator: extract `status_message` field from spans [#1786](https:/grafana/tempo/pull/1786), [#1794](https:/grafana/tempo/pull/1794) (@stoewer)
 * [ENHANCEMENT] metrics-generator: handle collisions between user defined and default dimensions [#1794](https:/grafana/tempo/pull/1794) (@stoewer)
 * [ENHANCEMENT] distributor: Log span names when `distributor.log_received_spans.include_all_attributes` is on [#1790](https:/grafana/tempo/pull/1790) (@suraciii)
+* [ENHANCEMENT] Add parquet WAL [#1878](https:/grafana/tempo/pull/1878) (@joe-elliott, @mdisibio)
 * [BUGFIX] Honor caching and buffering settings when finding traces by id [#1697](https:/grafana/tempo/pull/1697) (@joe-elliott)
 * [BUGFIX] Correctly propagate errors from the iterator layer up through the queriers [#1723](https:/grafana/tempo/pull/1723) (@joe-elliott)
 * [BUGFIX] Make multitenancy work with HTTP [#1781](https:/grafana/tempo/pull/1781) (@gouthamve)

modules/ingester/ingester_test.go

@@ -73,6 +73,7 @@ func TestPushQueryAllEncodings(t *testing.T) {
  TraceID: traceID,
  })
  require.NoError(t, err, "unexpected error querying")
+ trace.SortTrace(foundTrace.Trace)
  equal := proto.Equal(traces[i], foundTrace.Trace)
  require.True(t, equal)
  }
@@ -152,8 +153,8 @@ func TestWal(t *testing.T) {
  TraceID: traceID,
  })
  require.NoError(t, err, "unexpected error querying")
- equal := proto.Equal(traces[i], foundTrace.Trace)
- require.True(t, equal)
+ trace.SortTrace(foundTrace.Trace)
+ test.TracesEqual(t, traces[i], foundTrace.Trace)
  }
 
  // a block that has been replayed should have a flush queue entry to complete it
@@ -173,8 +174,7 @@ func TestWal(t *testing.T) {
  require.NoError(t, err, "unexpected error querying")
 
  trace.SortTrace(foundTrace.Trace)
- equal := proto.Equal(traces[i], foundTrace.Trace)
- require.True(t, equal)
+ test.TracesEqual(t, traces[i], foundTrace.Trace)
  }
 }
 
@@ -187,20 +187,23 @@ func TestSearchWAL(t *testing.T) {
  inst, _ := i.getOrCreateInstance("test")
  require.NotNil(t, inst)
 
+ dec := model.MustNewSegmentDecoder(model.CurrentEncoding)
+
  // create some search data
  id := make([]byte, 16)
  _, err = rand.Read(id)
  require.NoError(t, err)
  trace := test.MakeTrace(10, id)
- traceBytes, err := trace.Marshal()
+ b1, err := dec.PrepareForWrite(trace, 0, 0)
  require.NoError(t, err)
+
  entry := &tempofb.SearchEntryMutable{}
  entry.TraceID = id
  entry.AddTag("foo", "bar")
  searchBytes := entry.ToBytes()
 
  // push to instance
- require.NoError(t, inst.PushBytes(context.Background(), id, traceBytes, searchBytes))
+ require.NoError(t, inst.PushBytes(context.Background(), id, b1, searchBytes))
 
  // Write wal
  require.NoError(t, inst.CutCompleteTraces(0, true))
@@ -307,8 +310,8 @@ func TestFlush(t *testing.T) {
  TraceID: traceID,
  })
  require.NoError(t, err, "unexpected error querying")
- equal := proto.Equal(traces[i], foundTrace.Trace)
- require.True(t, equal)
+ trace.SortTrace(foundTrace.Trace)
+ test.TracesEqual(t, traces[i], foundTrace.Trace)
  }
 }
 
@@ -333,6 +336,7 @@ func defaultIngesterModule(t testing.TB, tmpDir string) *Ingester {
  },
  WAL: &wal.Config{
  Filepath: tmpDir,
+ Version: "v2",
  },
  },
  }, log.NewNopLogger())

modules/ingester/instance.go

@@ -32,6 +32,7 @@ import (
  "github.com/grafana/tempo/tempodb/backend/local"
  "github.com/grafana/tempo/tempodb/encoding"
  "github.com/grafana/tempo/tempodb/encoding/common"
+ v2 "github.com/grafana/tempo/tempodb/encoding/v2"
  "github.com/grafana/tempo/tempodb/search"
 )
 
@@ -259,7 +260,9 @@ func (i *instance) CutCompleteTraces(cutoff time.Duration, immediate bool) error
  tempopb.ReuseByteSlices(t.batches)
  }
 
- return nil
+ i.blocksMtx.Lock()
+ defer i.blocksMtx.Unlock()
+ return i.headBlock.Flush()
 }
 
 // CutBlockIfReady cuts a completingBlock from the HeadBlock if ready.
@@ -274,11 +277,18 @@ func (i *instance) CutBlockIfReady(maxBlockLifetime time.Duration, maxBlockBytes
 
  now := time.Now()
  if i.lastBlockCut.Add(maxBlockLifetime).Before(now) || i.headBlock.DataLength() >= maxBlockBytes || immediate {
+
+ // Final flush
+ err := i.headBlock.Flush()
+ if err != nil {
+ return uuid.Nil, fmt.Errorf("failed to flush head block: %w", err)
+ }
+
  completingBlock := i.headBlock
 
  i.completingBlocks = append(i.completingBlocks, completingBlock)
 
- err := i.resetHeadBlock()
+ err = i.resetHeadBlock()
  if err != nil {
  return uuid.Nil, fmt.Errorf("failed to resetHeadBlock: %w", err)
  }
@@ -451,15 +461,15 @@ func (i *instance) FindTraceByID(ctx context.Context, id []byte) (*tempopb.Trace
  combiner.Consume(completeTrace)
 
  // headBlock
- tr, err := i.headBlock.FindTraceByID(ctx, id, common.SearchOptions{})
+ tr, err := i.headBlock.FindTraceByID(ctx, id, common.DefaultSearchOptions())
  if err != nil {
  return nil, fmt.Errorf("headBlock.FindTraceByID failed: %w", err)
  }
  combiner.Consume(tr)
 
  // completingBlock
  for _, c := range i.completingBlocks {
- tr, err = c.FindTraceByID(ctx, id, common.SearchOptions{})
+ tr, err = c.FindTraceByID(ctx, id, common.DefaultSearchOptions())
  if err != nil {
  return nil, fmt.Errorf("completingBlock.FindTraceByID failed: %w", err)
  }
@@ -468,7 +478,7 @@ func (i *instance) FindTraceByID(ctx context.Context, id []byte) (*tempopb.Trace
 
  // completeBlock
  for _, c := range i.completeBlocks {
- found, err := c.FindTraceByID(ctx, id, common.SearchOptions{})
+ found, err := c.FindTraceByID(ctx, id, common.DefaultSearchOptions())
  if err != nil {
  return nil, fmt.Errorf("completeBlock.FindTraceByID failed: %w", err)
  }
@@ -538,21 +548,23 @@ func (i *instance) resetHeadBlock() error {
  i.headBlock = newHeadBlock
  i.lastBlockCut = time.Now()
 
- // Create search data wal file
- f, enc, err := i.writer.WAL().NewFile(i.headBlock.BlockMeta().BlockID, i.instanceID, searchDir)
- if err != nil {
- return err
- }
+ // Create search data wal file if needed
+ if i.useFlatbufferSearch || i.headBlock.BlockMeta().Version == v2.VersionString {
+ f, enc, err := i.writer.WAL().NewFile(i.headBlock.BlockMeta().BlockID, i.instanceID, searchDir)
+ if err != nil {
+ return err
+ }
 
- b, err := search.NewStreamingSearchBlockForFile(f, i.headBlock.BlockMeta().BlockID, enc)
- if err != nil {
- return err
- }
- if i.searchHeadBlock != nil {
- i.searchAppendBlocks[oldHeadBlock.BlockMeta().BlockID.String()] = i.searchHeadBlock
- }
- i.searchHeadBlock = &searchStreamingBlockEntry{
- b: b,
+ b, err := search.NewStreamingSearchBlockForFile(f, i.headBlock.BlockMeta().BlockID, enc)
+ if err != nil {
+ return err
+ }
+ if i.searchHeadBlock != nil {
+ i.searchAppendBlocks[oldHeadBlock.BlockMeta().BlockID.String()] = i.searchHeadBlock
+ }
+ i.searchHeadBlock = &searchStreamingBlockEntry{
+ b: b,
+ }
  }
  return nil
 }

modules/ingester/instance_search.go

@@ -51,7 +51,7 @@ func (i *instance) Search(ctx context.Context, req *tempopb.SearchRequest) (*tem
  // deadlocking with other activity (ingest, flushing), caused by releasing
  // and then attempting to retake the lock.
  i.blocksMtx.RLock()
- i.searchWAL(ctx, p, sr)
+ i.searchWAL(ctx, req, p, sr)
  i.searchLocalBlocks(ctx, req, p, sr)
  i.blocksMtx.RUnlock()
 
@@ -143,7 +143,7 @@ func (i *instance) searchLiveTraces(ctx context.Context, p search.Pipeline, sr *
 }
 
 // searchWAL starts a search task for every WAL block. Must be called under lock.
-func (i *instance) searchWAL(ctx context.Context, p search.Pipeline, sr *search.Results) {
+func (i *instance) searchWAL(ctx context.Context, req *tempopb.SearchRequest, p search.Pipeline, sr *search.Results) {
  searchFunc := func(e *searchStreamingBlockEntry) {
  span, ctx := opentracing.StartSpanFromContext(ctx, "instance.searchWAL")
  defer span.Finish()
@@ -162,11 +162,43 @@ func (i *instance) searchWAL(ctx context.Context, p search.Pipeline, sr *search.
  }
  }
 
+ searchWalBlock := func(b common.WALBlock) {
+ span, ctx := opentracing.StartSpanFromContext(ctx, "instance.searchWALBlock", opentracing.Tags{
+ "blockID": b.BlockMeta().BlockID,
+ })
+ defer span.Finish()
+ defer sr.FinishWorker()
+
+ resp, err := b.Search(ctx, req, common.DefaultSearchOptions())
+ if err != nil {
+ level.Error(log.Logger).Log("msg", "error searching wal block", "blockID", b.BlockMeta().BlockID.String(), "err", err)
+ return
+ }
+
+ sr.AddBlockInspected()
+ sr.AddBytesInspected(resp.Metrics.InspectedBytes)
+ for _, r := range resp.Traces {
+ sr.AddResult(ctx, r)
+ }
+ }
+
  // head block
- sr.StartWorker()
- go searchFunc(i.searchHeadBlock)
+ if i.headBlock != nil {
+ sr.StartWorker()
+ go searchWalBlock(i.headBlock)
+ }
+
+ if i.searchHeadBlock != nil {
+ sr.StartWorker()
+ go searchFunc(i.searchHeadBlock)
+ }
 
  // completing blocks
+ for _, b := range i.completingBlocks {
+ sr.StartWorker()
+ go searchWalBlock(b)
+ }
+
  for _, e := range i.searchAppendBlocks {
  sr.StartWorker()
  go searchFunc(e)
@@ -223,10 +255,7 @@ func (i *instance) searchLocalBlocks(ctx context.Context, req *tempopb.SearchReq
  span.LogFields(ot_log.Event("local block entry mtx acquired"))
  span.SetTag("blockID", blockID)
 
- resp, err := e.Search(ctx, req, common.SearchOptions{
- ReadBufferCount: 32,
- ReadBufferSize: 1024 * 1024,
- })
+ resp, err := e.Search(ctx, req, common.DefaultSearchOptions())
  if err != nil {
  level.Error(log.Logger).Log("msg", "error searching local block", "blockID", blockID, "err", err)
  return
@@ -286,10 +315,7 @@ func (i *instance) SearchTags(ctx context.Context) (*tempopb.SearchTagsResponse,
  continue
  }
 
- err = b.SearchTags(ctx, distinctValues.Collect, common.SearchOptions{
- ReadBufferCount: 32,
- ReadBufferSize: 1024 * 1024,
- })
+ err = b.SearchTags(ctx, distinctValues.Collect, common.DefaultSearchOptions())
  if err == common.ErrUnsupported {
  level.Warn(log.Logger).Log("msg", "block does not support tag search", "blockID", b.BlockMeta().BlockID)
  continue
@@ -358,10 +384,7 @@ func (i *instance) SearchTagValues(ctx context.Context, tagName string) (*tempop
  continue
  }
 
- err = b.SearchTagValues(ctx, tagName, distinctValues.Collect, common.SearchOptions{
- ReadBufferCount: 32,
- ReadBufferSize: 1024 * 1024,
- })
+ err = b.SearchTagValues(ctx, tagName, distinctValues.Collect, common.DefaultSearchOptions())
  if err == common.ErrUnsupported {
  level.Warn(log.Logger).Log("msg", "block does not support tag value search", "blockID", b.BlockMeta().BlockID)
  continue
@@ -429,12 +452,14 @@ func (i *instance) visitSearchableBlocksWAL(ctx context.Context, visitFn func(bl
  return visitFn(entry.b)
  }
 
- err := visitUnderLock(i.searchHeadBlock)
- if err != nil {
- return err
- }
- if err := ctx.Err(); err != nil {
- return err
+ if i.searchHeadBlock != nil {
+ err := visitUnderLock(i.searchHeadBlock)
+ if err != nil {
+ return err
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
  }
 
  for _, b := range i.searchAppendBlocks {

modules/ingester/instance_search_test.go

@@ -88,6 +88,9 @@ func TestInstanceSearch(t *testing.T) {
  assert.NoError(t, err)
  assert.Len(t, sr.Traces, len(ids))
  checkEqual(t, ids, sr)
+
+ err = ingester.stopping(nil)
+ require.NoError(t, err)
  })
  }
 }
@@ -153,7 +156,6 @@ func testSearchTagsAndValues(t *testing.T, ctx context.Context, i *instance, tag
 
  sort.Strings(srv.TagValues)
  assert.Contains(t, sr.TagNames, tagName)
- assert.Equal(t, tagName, sr.TagNames[0])
  assert.Equal(t, expectedTagValues, srv.TagValues)
 }
 

modules/ingester/instance_test.go

@@ -374,6 +374,8 @@ func TestInstanceCutCompleteTraces(t *testing.T) {
 }
 
 func TestInstanceCutBlockIfReady(t *testing.T) {
+ dec := model.MustNewSegmentDecoder(model.CurrentEncoding)
+
  tt := []struct {
  name string
  maxBlockLifetime time.Duration
@@ -416,14 +418,16 @@ func TestInstanceCutBlockIfReady(t *testing.T) {
  instance, _ := defaultInstance(t)
 
  for i := 0; i < tc.pushCount; i++ {
- request := makeRequest([]byte{})
- err := instance.PushBytesRequest(context.Background(), request)
+ tr := test.MakeTrace(1, uuid.Nil[:])
+ bytes, err := dec.PrepareForWrite(tr, 0, 0)
+ require.NoError(t, err)
+ err = instance.PushBytes(context.Background(), uuid.Nil[:], bytes, nil)
  require.NoError(t, err)
  }
 
  // Defaults
  if tc.maxBlockBytes == 0 {
- tc.maxBlockBytes = 1000
+ tc.maxBlockBytes = 100000
  }
  if tc.maxBlockLifetime == 0 {
  tc.maxBlockLifetime = time.Hour
@@ -688,7 +692,7 @@ func makeRequestWithByteLimit(maxBytes int, traceID []byte) *tempopb.PushBytesRe
  batch := test.MakeBatch(1, traceID)
 
  for batch.Size() < maxBytes {
- batch.ScopeSpans[0].Spans = append(batch.ScopeSpans[0].Spans, test.MakeSpan(traceID))
+ batch.ScopeSpans[0].Spans = append(batch.ScopeSpans[0].Spans, test.MakeSpanWithAttributeCount(traceID, 0))
  }
 
  return makePushBytesRequest(traceID, batch)

modules/storage/config.go

@@ -15,6 +15,7 @@ import (
  "github.com/grafana/tempo/tempodb/backend/s3"
  "github.com/grafana/tempo/tempodb/encoding"
  "github.com/grafana/tempo/tempodb/encoding/common"
+ v2 "github.com/grafana/tempo/tempodb/encoding/v2"
  "github.com/grafana/tempo/tempodb/pool"
  "github.com/grafana/tempo/tempodb/wal"
 )
@@ -36,6 +37,7 @@ func (cfg *Config) RegisterFlagsAndApplyDefaults(prefix string, f *flag.FlagSet)
 
  cfg.Trace.WAL = &wal.Config{}
  f.StringVar(&cfg.Trace.WAL.Filepath, util.PrefixConfig(prefix, "trace.wal.path"), "/var/tempo/wal", "Path at which store WAL blocks.")
+ cfg.Trace.WAL.Version = v2.VersionString
  cfg.Trace.WAL.Encoding = backend.EncSnappy
  cfg.Trace.WAL.SearchEncoding = backend.EncNone
  cfg.Trace.WAL.IngestionSlack = 2 * time.Minute