[v2][adjuster] Implement adjuster for deduplicating spans #6391
+405
−0
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // Copyright (c) 2024 The Jaeger Authors. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package adjuster | ||
|
|
||
| import ( | ||
| "hash/fnv" | ||
|
|
||
| "go.opentelemetry.io/collector/pdata/pcommon" | ||
| "go.opentelemetry.io/collector/pdata/ptrace" | ||
| ) | ||
|
|
||
| var _ Adjuster = (*SpanHashDeduper)(nil) | ||
|
|
||
| // SpanHash creates an adjuster that deduplicates spans by removing all but one span | ||
| // with the same hash code. This is particularly useful for scenarios where spans | ||
| // may be duplicated during archival, such as with ElasticSearch archival. | ||
| // | ||
| // The hash code is generated by serializing the span into protobuf bytes and applying | ||
| // the FNV hashing algorithm to the serialized data. | ||
| // | ||
| // To ensure consistent hash codes, this adjuster should be executed after | ||
| // SortAttributesAndEvents, which normalizes the order of collections within the span. | ||
| func SpanHash() SpanHashDeduper { | ||
| return SpanHashDeduper{ | ||
| marshaler: &ptrace.ProtoMarshaler{}, | ||
| } | ||
| } | ||
|
|
||
| type SpanHashDeduper struct { | ||
| marshaler ptrace.Marshaler | ||
yurishkuro marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } | ||
|
|
||
| func (s *SpanHashDeduper) Adjust(traces ptrace.Traces) { | ||
| spansByHash := make(map[uint64]ptrace.Span) | ||
| resourceSpans := traces.ResourceSpans() | ||
|
There was a problem hiding this comment. I'd recommend going forward to use terms resources and scopes. Makes the code more readable There was a problem hiding this comment. sounds good - I can open a cleanup PR |
||
| for i := 0; i < resourceSpans.Len(); i++ { | ||
| rs := resourceSpans.At(i) | ||
| rs.Resource().Attributes() | ||
mahadzaryab1 marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| scopeSpans := rs.ScopeSpans() | ||
| for j := 0; j < scopeSpans.Len(); j++ { | ||
| ss := scopeSpans.At(j) | ||
| spans := ss.Spans() | ||
| newSpans := ptrace.NewSpanSlice() | ||
| for k := 0; k < spans.Len(); k++ { | ||
| span := spans.At(k) | ||
| h, err := s.computeHashCode( | ||
| span, | ||
| rs.Resource().Attributes(), | ||
| ss.Scope().Attributes(), | ||
| ) | ||
| if err != nil { | ||
| // TODO: Add Warning | ||
| continue | ||
| } | ||
| if _, ok := spansByHash[h]; !ok { | ||
| spansByHash[h] = span | ||
| span.CopyTo(newSpans.AppendEmpty()) | ||
| } | ||
| } | ||
| newSpans.CopyTo(spans) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (s *SpanHashDeduper) computeHashCode( | ||
| span ptrace.Span, | ||
| resourceAttributes pcommon.Map, | ||
| scopeAttributes pcommon.Map, | ||
| ) (uint64, error) { | ||
| traces := ptrace.NewTraces() | ||
| rs := traces.ResourceSpans().AppendEmpty() | ||
| resourceAttributes.CopyTo(rs.Resource().Attributes()) | ||
| ss := rs.ScopeSpans().AppendEmpty() | ||
| scopeAttributes.CopyTo(ss.Scope().Attributes()) | ||
|
There was a problem hiding this comment. I would rather do this outside of the loop for spans and only replace the span before hashing |
||
| newSpan := ss.Spans().AppendEmpty() | ||
| span.CopyTo(newSpan) | ||
| b, err := s.marshaler.MarshalTraces(traces) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| hasher := fnv.New64a() | ||
| _, err = hasher.Write(b) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| return hasher.Sum64(), nil | ||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,163 @@ | ||
| // Copyright (c) 2024 The Jaeger Authors. | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package adjuster | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
| "go.opentelemetry.io/collector/pdata/ptrace" | ||
| ) | ||
|
|
||
| func TestSpanHash_EmptySpans(t *testing.T) { | ||
| adjuster := SpanHash() | ||
| input := ptrace.NewTraces() | ||
| expected := ptrace.NewTraces() | ||
| adjuster.Adjust(input) | ||
| assert.Equal(t, expected, input) | ||
| } | ||
|
|
||
| func TestSpanHash_RemovesDuplicateSpans(t *testing.T) { | ||
| adjuster := SpanHash() | ||
| input := func() ptrace.Traces { | ||
| traces := ptrace.NewTraces() | ||
| rs := traces.ResourceSpans().AppendEmpty() | ||
| ss := rs.ScopeSpans().AppendEmpty() | ||
| spans := ss.Spans() | ||
|
|
||
| span1 := spans.AppendEmpty() | ||
| span1.SetName("span1") | ||
| span1.SetTraceID([16]byte{1}) | ||
| span1.SetSpanID([8]byte{1}) | ||
|
|
||
| span2 := spans.AppendEmpty() | ||
| span2.SetName("span2") | ||
| span2.SetTraceID([16]byte{1}) | ||
| span2.SetSpanID([8]byte{2}) | ||
|
|
||
| span3 := spans.AppendEmpty() | ||
| span3.SetName("span1") | ||
| span3.SetTraceID([16]byte{1}) | ||
| span3.SetSpanID([8]byte{1}) | ||
|
|
||
| span4 := spans.AppendEmpty() | ||
| span4.SetName("span2") | ||
| span4.SetTraceID([16]byte{1}) | ||
| span4.SetSpanID([8]byte{2}) | ||
|
|
||
| span5 := spans.AppendEmpty() | ||
| span5.SetName("span3") | ||
| span5.SetTraceID([16]byte{3}) | ||
| span5.SetSpanID([8]byte{4}) | ||
|
|
||
| rs2 := traces.ResourceSpans().AppendEmpty() | ||
| ss2 := rs2.ScopeSpans().AppendEmpty() | ||
| spans2 := ss2.Spans() | ||
|
|
||
| span6 := spans2.AppendEmpty() | ||
| span6.SetName("span4") | ||
| span6.SetTraceID([16]byte{5}) | ||
| span6.SetSpanID([8]byte{6}) | ||
|
|
||
| span7 := spans2.AppendEmpty() | ||
| span7.SetName("span3") | ||
| span7.SetTraceID([16]byte{3}) | ||
| span7.SetSpanID([8]byte{4}) | ||
|
|
||
| return traces | ||
| } | ||
| expected := func() ptrace.Traces { | ||
| traces := ptrace.NewTraces() | ||
| rs := traces.ResourceSpans().AppendEmpty() | ||
| ss := rs.ScopeSpans().AppendEmpty() | ||
| spans := ss.Spans() | ||
|
|
||
| span1 := spans.AppendEmpty() | ||
| span1.SetName("span1") | ||
| span1.SetTraceID([16]byte{1}) | ||
| span1.SetSpanID([8]byte{1}) | ||
|
|
||
| span2 := spans.AppendEmpty() | ||
| span2.SetName("span2") | ||
| span2.SetTraceID([16]byte{1}) | ||
| span2.SetSpanID([8]byte{2}) | ||
|
|
||
| span3 := spans.AppendEmpty() | ||
| span3.SetName("span3") | ||
| span3.SetTraceID([16]byte{3}) | ||
| span3.SetSpanID([8]byte{4}) | ||
|
|
||
| rs2 := traces.ResourceSpans().AppendEmpty() | ||
| ss2 := rs2.ScopeSpans().AppendEmpty() | ||
| spans2 := ss2.Spans() | ||
|
|
||
| span4 := spans2.AppendEmpty() | ||
| span4.SetName("span4") | ||
| span4.SetTraceID([16]byte{5}) | ||
| span4.SetSpanID([8]byte{6}) | ||
|
|
||
| return traces | ||
| } | ||
|
|
||
| i := input() | ||
| adjuster.Adjust(i) | ||
| assert.Equal(t, expected(), i) | ||
| } | ||
|
|
||
| func TestSpanHash_NoDuplicateSpans(t *testing.T) { | ||
| adjuster := SpanHash() | ||
| input := func() ptrace.Traces { | ||
| traces := ptrace.NewTraces() | ||
| rs := traces.ResourceSpans().AppendEmpty() | ||
| ss := rs.ScopeSpans().AppendEmpty() | ||
| spans := ss.Spans() | ||
|
|
||
| span1 := spans.AppendEmpty() | ||
| span1.SetName("span1") | ||
| span1.SetTraceID([16]byte{1}) | ||
| span1.SetSpanID([8]byte{1}) | ||
|
|
||
| span2 := spans.AppendEmpty() | ||
| span2.SetName("span2") | ||
| span2.SetTraceID([16]byte{1}) | ||
| span2.SetSpanID([8]byte{2}) | ||
|
|
||
| span3 := spans.AppendEmpty() | ||
| span3.SetName("span3") | ||
| span3.SetTraceID([16]byte{3}) | ||
| span3.SetSpanID([8]byte{4}) | ||
|
|
||
| return traces | ||
| } | ||
| expected := func() ptrace.Traces { | ||
| traces := ptrace.NewTraces() | ||
| rs := traces.ResourceSpans().AppendEmpty() | ||
| ss := rs.ScopeSpans().AppendEmpty() | ||
| spans := ss.Spans() | ||
|
|
||
| span1 := spans.AppendEmpty() | ||
| span1.SetName("span1") | ||
| span1.SetTraceID([16]byte{1}) | ||
| span1.SetSpanID([8]byte{1}) | ||
|
|
||
| span2 := spans.AppendEmpty() | ||
| span2.SetName("span2") | ||
| span2.SetTraceID([16]byte{1}) | ||
| span2.SetSpanID([8]byte{2}) | ||
|
|
||
| span3 := spans.AppendEmpty() | ||
| span3.SetName("span3") | ||
| span3.SetTraceID([16]byte{3}) | ||
| span3.SetSpanID([8]byte{4}) | ||
|
|
||
| return traces | ||
| } | ||
|
|
||
| i := input() | ||
| adjuster.Adjust(i) | ||
| assert.Equal(t, expected(), i) | ||
| } | ||
|
|
||
| // TODO: write tests for duplicate spans with different outer attributes | ||
| // TODO: write tests for error cases |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
A couple of thoughts on this:
Some storage backends (Cassandra, in particular), perform similar deduping by computing a hash before the span is saved and using it as part of the partition key (it creates tombstones if identical span is saved 2 times or more but no dups on read). So we could make this hashing process to be a part of the ingestion pipeline (e.g. in sanitizers) and simply store it as an attribute on the span. Then this adjuster would be "lazy", it will only recompute the hash if it doesn't already exist in the storage.
If we do this on the write path, we would want this to be as efficient as possible, so we would need to implement manual hashing by iterating through the attributes (and pre-sorting them to avoid dependencies) and but manually going through all fields of the Span / SpanEvent / SpanLink. The reason I was reluctant to do that in the past was to avoid unintended bugs if the data model was changed, like a new field added that we'd forget to add to the hash function. To protect against that we probably could use some fuzzing tests, by setting / unsetting each field individually and making sure the hash code changes as a result.
We don't have to do it now, but let's open a ticket for future improvement (I think it could be a
good-first-issue)