ingester: use a 64-bit hash to avoid collisions (#5276) (#5302)

* ingester: use a 64-bit hash to avoid collisions

We had this issue in the distributor, it it can happen in the
ingester too with live traces.

* Update changelog

* Add test for live traces collision

Author: carles-grafana

CHANGELOG.md

@@ -1,5 +1,7 @@
 ## main / unreleased
 
+* [BUGFIX] Fix ingester issue where a hash collision could lead to spans stored incorrectly [#5276](https://github.com/grafana/tempo/pull/5276) (@carles-grafana)
+
 # v2.8.0
 
 * [CHANGE] **BREAKING CHANGE** Change default http-listen-port from 80 to 3200 [#4960](https://github.com/grafana/tempo/pull/4960) (@martialblog)

modules/ingester/instance.go

@@ -6,8 +6,6 @@ import (
 	"encoding/hex"
 	"errors"
 	"fmt"
-	"hash"
-	"hash/fnv"
 	"sort"
 	"sync"
 	"time"
@@ -25,6 +23,7 @@ import (
 	"github.com/grafana/tempo/pkg/model/trace"
 	"github.com/grafana/tempo/pkg/tempopb"
 	"github.com/grafana/tempo/pkg/tracesizes"
+	"github.com/grafana/tempo/pkg/util"
 	"github.com/grafana/tempo/pkg/util/log"
 	"github.com/grafana/tempo/pkg/validation"
 	"github.com/grafana/tempo/tempodb"
@@ -79,7 +78,7 @@ var (
 
 type instance struct {
 	tracesMtx      sync.Mutex
-	traces         map[uint32]*liveTrace
+	traces         map[uint64]*liveTrace
 	traceSizes     *tracesizes.Tracker
 	traceSizeBytes uint64
 
@@ -105,16 +104,14 @@ type instance struct {
 	localReader backend.Reader
 	localWriter backend.Writer
 
-	hash hash.Hash32
-
 	logger         kitlog.Logger
 	maxTraceLogger *log.RateLimitedLogger
 }
 
 func newInstance(instanceID string, limiter Limiter, overrides ingesterOverrides, writer tempodb.Writer, l *local.Backend, dedicatedColumns backend.DedicatedColumns) (*instance, error) {
 	logger := kitlog.With(log.Logger, "tenant", instanceID)
 	i := &instance{
-		traces:     map[uint32]*liveTrace{},
+		traces:     map[uint64]*liveTrace{},
 		traceSizes: tracesizes.New(),
 
 		instanceID:         instanceID,
@@ -130,8 +127,6 @@ func newInstance(instanceID string, limiter Limiter, overrides ingesterOverrides
 		localReader: backend.NewReader(l),
 		localWriter: backend.NewWriter(l),
 
-		hash: fnv.New32(),
-
 		logger:         logger,
 		maxTraceLogger: log.NewRateLimitedLogger(maxTraceLogLinesPerSecond, level.Warn(logger)),
 	}
@@ -213,7 +208,7 @@ func (i *instance) push(ctx context.Context, id, traceBytes []byte) error {
 		return errTraceTooLarge
 	}
 
-	tkn := i.tokenForTraceID(id)
+	tkn := util.HashForTraceID(id)
 	trace := i.getOrCreateTrace(id, tkn)
 
 	err = trace.Push(ctx, i.instanceID, traceBytes)
@@ -418,7 +413,7 @@ func (i *instance) FindTraceByID(ctx context.Context, id []byte, allowPartialTra
 
 	// live traces
 	i.tracesMtx.Lock()
-	if liveTrace, ok := i.traces[i.tokenForTraceID(id)]; ok {
+	if liveTrace, ok := i.traces[util.HashForTraceID(id)]; ok {
 		completeTrace, err = model.MustNewSegmentDecoder(model.CurrentEncoding).PrepareForRead(liveTrace.batches)
 		for _, b := range liveTrace.batches {
 			metrics.InspectedBytes += uint64(len(b))
@@ -514,7 +509,7 @@ func (i *instance) AddCompletingBlock(b common.WALBlock) {
 // getOrCreateTrace will return a new trace object for the given request
 //
 //	It must be called under the i.tracesMtx lock
-func (i *instance) getOrCreateTrace(traceID []byte, fp uint32) *liveTrace {
+func (i *instance) getOrCreateTrace(traceID []byte, fp uint64) *liveTrace {
 	trace, ok := i.traces[fp]
 	if ok {
 		return trace
@@ -526,13 +521,6 @@ func (i *instance) getOrCreateTrace(traceID []byte, fp uint32) *liveTrace {
 	return trace
 }
 
-// tokenForTraceID hash trace ID, should be called under lock
-func (i *instance) tokenForTraceID(id []byte) uint32 {
-	i.hash.Reset()
-	_, _ = i.hash.Write(id)
-	return i.hash.Sum32()
-}
-
 // resetHeadBlock() should be called under lock
 func (i *instance) resetHeadBlock() error {
 	dedicatedColumns := i.getDedicatedColumns()

modules/ingester/instance_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/grafana/tempo/pkg/model/trace"
 	"github.com/grafana/tempo/pkg/tempopb"
 	v1_trace "github.com/grafana/tempo/pkg/tempopb/trace/v1"
+	"github.com/grafana/tempo/pkg/util"
 	"github.com/grafana/tempo/pkg/util/test"
 )
 
@@ -409,7 +410,7 @@ func TestInstanceCutCompleteTraces(t *testing.T) {
 			instance, _ := defaultInstance(t)
 
 			for _, trace := range tc.input {
-				fp := instance.tokenForTraceID(trace.traceID)
+				fp := util.HashForTraceID(trace.traceID)
 				instance.traces[fp] = trace
 			}
 
@@ -418,12 +419,12 @@ func TestInstanceCutCompleteTraces(t *testing.T) {
 
 			require.Equal(t, len(tc.expectedExist), len(instance.traces))
 			for _, expectedExist := range tc.expectedExist {
-				_, ok := instance.traces[instance.tokenForTraceID(expectedExist.traceID)]
+				_, ok := instance.traces[util.HashForTraceID(expectedExist.traceID)]
 				require.True(t, ok)
 			}
 
 			for _, expectedNotExist := range tc.expectedNotExist {
-				_, ok := instance.traces[instance.tokenForTraceID(expectedNotExist.traceID)]
+				_, ok := instance.traces[util.HashForTraceID(expectedNotExist.traceID)]
 				require.False(t, ok)
 			}
 		})
@@ -684,6 +685,72 @@ func TestInstancePartialSuccess(t *testing.T) {
 	require.Equal(t, expected, result)
 }
 
+func TestFindTraceByIDNoCollisions(t *testing.T) {
+	ctx := context.Background()
+	ingester, _, _ := defaultIngester(t, t.TempDir())
+	instance, err := ingester.getOrCreateInstance(testTenantID)
+	require.NoError(t, err, "unexpected error creating new instance")
+
+	// Verify that Trace IDs that collide with fnv-32 hash do not collide with fnv-64 hash
+	hexIDs := []string{
+		"fd5980503add11f09f80f77608c1b2da",
+		"091ea7803ade11f0998a055186ee1243",
+		"9e0d446036dc11f09ac04988d2097052",
+		"a61ed97036dc11f0883771db3b51b1ec",
+		"6b27f5501eda11f09e99db1b2c23c542",
+		"6b4149b01eda11f0b0e2a966cf7ebbc8",
+		"3e9582202f9a11f0afb01b7c06024bd6",
+		"370db6802f9a11f0a9a212dff3125239",
+		"978d70802a7311f0991f350653ef0ab4",
+		"9b66da202a7311f09d292db17ccfd31a",
+		"de567f703bb711f0b8c377682d1667e6",
+		"dc2d0fc03bb711f091de732fcf93048c",
+	}
+
+	ids := make([][]byte, len(hexIDs))
+	for i, hexID := range hexIDs {
+		traceID, err := util.HexStringToTraceID(hexID)
+		require.NoError(t, err)
+		ids[i] = traceID
+	}
+
+	multiMaxBytes := make([]int, len(ids))
+	for j := range multiMaxBytes {
+		multiMaxBytes[j] = 1000
+	}
+	req := makePushBytesRequestMultiTraces(ids, multiMaxBytes)
+	require.Equal(t, len(ids), len(req.Traces))
+	response := instance.PushBytesRequest(ctx, req)
+	errored, maxLiveCount, traceTooLargeCount := CheckPushBytesError(response)
+
+	require.False(t, errored)
+	require.Equal(t, 0, maxLiveCount)
+	require.Equal(t, 0, traceTooLargeCount)
+
+	traceResults := make([]*tempopb.Trace, len(ids))
+	for i := range ids {
+		result, err := instance.FindTraceByID(ctx, ids[i], false)
+		require.NoError(t, err, "error finding trace by id")
+		require.NotNil(t, result)
+		require.NotNil(t, result.Trace)
+		traceResults[i] = result.Trace
+	}
+
+	// Verify that spans do not appear in multiple traces
+	spanIDs := map[string]struct{}{}
+	for _, trace := range traceResults {
+		for _, resourceSpan := range trace.ResourceSpans {
+			for _, scopeSpan := range resourceSpan.ScopeSpans {
+				for _, span := range scopeSpan.Spans {
+					_, found := spanIDs[string(span.SpanId)]
+					require.False(t, found, "span %s appears in multiple traces", span.SpanId)
+					spanIDs[string(span.SpanId)] = struct{}{}
+				}
+			}
+		}
+	}
+}
+
 func TestSortByteSlices(t *testing.T) {
 	numTraces := 100