add signals for duplicate rf1 data (#4296)

Signed-off-by: Joe Elliott <[email protected]>

cmd/tempo-cli/cmd-rewrite-blocks.go

@@ -141,6 +141,7 @@ func rewriteBlock(ctx context.Context, r backend.Reader, w backend.Writer, meta
 		SpansDiscarded:    func(_, _, _ string, _ int) {},
 		DisconnectedTrace: func() {},
 		RootlessTrace:     func() {},
+		DedupedSpans:      func(_, _ int) {},
 	}
 
 	compactor := enc.NewCompactor(opts)

tempodb/compactor.go

@@ -65,6 +65,11 @@ var (
 		Name:      "compaction_outstanding_blocks",
 		Help:      "Number of blocks remaining to be compacted before next maintenance cycle",
 	}, []string{"tenant"})
+	metricDedupedSpans = promauto.NewGaugeVec(prometheus.GaugeOpts{
+		Namespace: "tempodb",
+		Name:      "compaction_spans_combined_total",
+		Help:      "Number of spans that are deduped per replication factor.",
+	}, []string{"replication_factor"})
 )
 
 func (rw *readerWriter) compactionLoop(ctx context.Context) {
@@ -166,6 +171,10 @@ func (rw *readerWriter) doCompaction(ctx context.Context) {
 				metricCompactionErrors.Inc()
 			}
 
+			if !rw.compactorSharder.Owns(hashString) {
+				level.Warn(rw.logger).Log("msg", "compaction complete but we no longer own the hash", "hashString", hashString)
+			}
+
 			// after a maintenance cycle bail out
 			if start.Add(rw.compactorCfg.MaxTimePerTenant).Before(time.Now()) {
 				measureOutstandingBlocks(tenantID, blockSelector, rw.compactorSharder.Owns)
@@ -263,6 +272,9 @@ func (rw *readerWriter) compact(ctx context.Context, blockMetas []*backend.Block
 		RootlessTrace: func() {
 			dataquality.WarnRootlessTrace(tenantID, dataquality.PhaseTraceCompactorCombine)
 		},
+		DedupedSpans: func(replFactor, dedupedSpans int) {
+			metricDedupedSpans.WithLabelValues(strconv.Itoa(replFactor)).Add(float64(dedupedSpans))
+		},
 	}
 
 	compactor := enc.NewCompactor(opts)

tempodb/encoding/common/interfaces.go

@@ -88,6 +88,7 @@ type CompactionOptions struct {
 	SpansDiscarded    func(traceID string, rootSpanName string, rootServiceName string, spans int)
 	DisconnectedTrace func()
 	RootlessTrace     func()
+	DedupedSpans      func(replFactor, dedupedSpans int)
 }
 
 type Iterator interface {

tempodb/encoding/vparquet4/combiner.go

@@ -42,7 +42,8 @@ func (c *Combiner) Consume(tr *Trace) (spanCount int) {
 }
 
 // ConsumeWithFinal consumes the trace, but allows for performance savings when
-// it is known that this is the last expected input trace.
+// it is known that this is the last expected input trace. the spanCount returned
+// is the number of duplicate spans between the two traces
 func (c *Combiner) ConsumeWithFinal(tr *Trace, final bool) (spanCount int) {
 	if tr == nil {
 		return
@@ -110,9 +111,10 @@ func (c *Combiner) ConsumeWithFinal(tr *Trace, final bool) (spanCount int) {
 
 			if len(notFoundSpans) > 0 {
 				ils.Spans = notFoundSpans
-				spanCount += len(notFoundSpans)
 				notFoundILS = append(notFoundILS, ils)
 			}
+
+			spanCount += len(ils.Spans) - len(notFoundSpans)
 		}
 
 		// if there were some spans not found in A, add everything left in the batch

tempodb/encoding/vparquet4/combiner_test.go

@@ -18,6 +18,12 @@ func TestCombiner(t *testing.T) {
 			c.Consume(b)
 			return c.Result()
 		},
+		func(a, b *Trace) (*Trace, int, bool) {
+			c := NewCombiner()
+			c.Consume(a)
+			c.ConsumeWithFinal(b, true)
+			return c.Result()
+		},
 	}
 
 	tests := []struct {
@@ -369,6 +375,147 @@ func TestCombiner(t *testing.T) {
 	}
 }
 
+func TestCombinerReturnsDuplicates(t *testing.T) {
+	tests := []struct {
+		name          string
+		traceA        *Trace
+		traceB        *Trace
+		expectedDupes int
+	}{
+		{
+			name:          "nil traceA",
+			traceA:        nil,
+			traceB:        &Trace{},
+			expectedDupes: 0,
+		},
+		{
+			name:          "nil traceB",
+			traceA:        &Trace{},
+			traceB:        nil,
+			expectedDupes: 0,
+		},
+		{
+			name:          "empty traces",
+			traceA:        &Trace{},
+			traceB:        &Trace{},
+			expectedDupes: 0,
+		},
+		{
+			name: "no dupes",
+			traceA: &Trace{
+				TraceID:         []byte{0x00, 0x01},
+				RootServiceName: "serviceNameA",
+				ResourceSpans: []ResourceSpans{
+					{
+						Resource: Resource{
+							ServiceName: "serviceNameA",
+						},
+						ScopeSpans: []ScopeSpans{
+							{
+								Spans: []Span{
+									{
+										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
+										StatusCode:     0,
+										NestedSetLeft:  1,
+										NestedSetRight: 2,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			traceB: &Trace{
+				TraceID:         []byte{0x00, 0x01},
+				RootServiceName: "serviceNameB",
+				ResourceSpans: []ResourceSpans{
+					{
+						Resource: Resource{
+							ServiceName: "serviceNameB",
+						},
+						ScopeSpans: []ScopeSpans{
+							{
+								Spans: []Span{
+									{
+										SpanID:       []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02},
+										ParentSpanID: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
+										StatusCode:   0,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedDupes: 0,
+		},
+		{
+			name: "one dupe",
+			traceA: &Trace{
+				TraceID:         []byte{0x00, 0x01},
+				RootServiceName: "serviceNameA",
+				ResourceSpans: []ResourceSpans{
+					{
+						Resource: Resource{
+							ServiceName: "serviceNameA",
+						},
+						ScopeSpans: []ScopeSpans{
+							{
+								Spans: []Span{
+									{
+										SpanID:         []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
+										StatusCode:     0,
+										NestedSetLeft:  1,
+										NestedSetRight: 2,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			traceB: &Trace{
+				TraceID:         []byte{0x00, 0x01},
+				RootServiceName: "serviceNameB",
+				ResourceSpans: []ResourceSpans{
+					{
+						Resource: Resource{
+							ServiceName: "serviceNameB",
+						},
+						ScopeSpans: []ScopeSpans{
+							{
+								Spans: []Span{
+									{
+										SpanID:     []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
+										StatusCode: 0,
+									},
+									{
+										SpanID:       []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x03},
+										ParentSpanID: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
+										StatusCode:   2,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			expectedDupes: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cmb := NewCombiner()
+
+			cmb.Consume(tt.traceA)
+			actualDupes := cmb.Consume(tt.traceB)
+
+			assert.Equal(t, tt.expectedDupes, actualDupes)
+		})
+	}
+}
+
 func BenchmarkCombine(b *testing.B) {
 	batchCount := 100
 	spanCounts := []int{

tempodb/encoding/vparquet4/compactor.go

@@ -67,6 +67,7 @@ func (c *Compactor) Compact(ctx context.Context, l log.Logger, r backend.Reader,
 	}
 
 	var (
+		replicationFactor   = inputs[0].ReplicationFactor
 		nextCompactionLevel = compactionLevel + 1
 		sch                 = parquet.SchemaOf(new(Trace))
 	)
@@ -110,15 +111,17 @@ func (c *Compactor) Compact(ctx context.Context, l log.Logger, r backend.Reader,
 
 		// Time to combine.
 		cmb := NewCombiner()
+		dedupedSpans := 0
 		for i, row := range rows {
 			tr := new(Trace)
 			err := sch.Reconstruct(tr, row)
 			if err != nil {
 				return nil, err
 			}
-			cmb.ConsumeWithFinal(tr, i == len(rows)-1)
+			dedupedSpans += cmb.ConsumeWithFinal(tr, i == len(rows)-1)
 			pool.Put(row)
 		}
+		c.opts.DedupedSpans(int(replicationFactor), dedupedSpans)
 		tr, _, connected := cmb.Result()
 		if !connected {
 			c.opts.DisconnectedTrace()
@@ -160,7 +163,7 @@ func (c *Compactor) Compact(ctx context.Context, l log.Logger, r backend.Reader,
 				TenantID:          inputs[0].TenantID,
 				CompactionLevel:   nextCompactionLevel,
 				TotalObjects:      recordsPerBlock, // Just an estimate
-				ReplicationFactor: inputs[0].ReplicationFactor,
+				ReplicationFactor: replicationFactor,
 				DedicatedColumns:  inputs[0].DedicatedColumns,
 			}