Skip to content

Rewrite traces using rebatching #4690

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
stoewer merged 15 commits into from
Feb 20, 2025
Merged

Rewrite traces using rebatching #4690

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
fa7bf72
lol
joe-elliott Dec 3, 2024
db52e53
first attempt
joe-elliott Dec 3, 2024
7611253
Remove 4to4 converter from tempo-cli
stoewer Feb 7, 2025
281ecd1
Generated trace data with unique scope and resource spans
stoewer Feb 12, 2025
e1757b1
Fix tests in spanmetrics processor
stoewer Feb 12, 2025
6669e80
Move rebatch code out of nested set function
stoewer Feb 13, 2025
671d118
Make parquetTraceToTempopbTrace() private
stoewer Feb 13, 2025
706e29f
Add benchmark for testing traceToParquet()
stoewer Feb 13, 2025
76e9b73
make check-fmt
stoewer Feb 13, 2025
17c124d
rebatchTrace() without allocating new slices for ResourceSpans and Sc…
stoewer Feb 14, 2025
8ef32f6
Fix bug with duplicated rs ans ss in rebatchTrace()
stoewer Feb 16, 2025
d50ab3c
CHANGELOG.md
stoewer Feb 17, 2025
1349851
Merge remote-tracking branch 'upstream/main' into rewrite-traces
stoewer Feb 18, 2025
a4c7e90
Add finalizeTrace() function
stoewer Feb 20, 2025
cb19b4e
Call finalizeTrace() in combiner
stoewer Feb 20, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
* [CHANGE] Update to go 1.24.0 [#4704](https://github.com/grafana/tempo/pull/4704) (@ruslan-mikhailov)
* [FEATURE] Added most_recent=true query hint to TraceQL to return most recent results. [#4238](https://github.com/grafana/tempo/pull/4238) (@joe-elliott)
* [FEATURE] Add ability to add artificial delay to push requests [#4716](https://github.com/grafana/tempo/pull/4716) (@yvrhdn)
* [ENHANCEMENT] Rewrite traces using rebatching [#4690](https://github.com/grafana/tempo/pull/4690) (@stoewer @joe-elliott)
* [ENHANCEMENT] Update minio to version [#4341](https://github.com/grafana/tempo/pull/4568) (@javiermolinar)
* [ENHANCEMENT] Prevent queries in the ingester from blocking flushing traces to disk and memory spikes. [#4483](https://github.com/grafana/tempo/pull/4483) (@joe-elliott)
* [ENHANCEMENT] Update tempo operational dashboard for new block-builder and v2 traces api [#4559](https://github.com/grafana/tempo/pull/4559) (@mdisibio)
Expand Down
247 changes: 145 additions & 102 deletions modules/generator/processor/spanmetrics/spanmetrics_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -560,22 +560,27 @@ func TestTargetInfoEnabled(t *testing.T) {
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// add instance
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})

// add additional source attributes
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
})

batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
})
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
// add additional source attributes
{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
},
{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch}})

Expand Down Expand Up @@ -608,22 +613,27 @@ func TestTargetInfoDisabled(t *testing.T) {
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// add instance
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})

// add additional source attributes
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
})

batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
})
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
// add additional source attributes
{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
},
{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch}})

Expand Down Expand Up @@ -655,33 +665,36 @@ func TestTargetInfoWithExclusion(t *testing.T) {
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// add instance
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})

// add additional source attributes
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
})

batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
})

// add attribute for labels that we want to drop
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "container",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
})

batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "container.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "xyz123"}},
})
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
// add additional source attributes
{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
},
{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
},
// add attribute for labels that we want to drop
{
Key: "container",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
{
Key: "container.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "xyz123"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch}})

Expand Down Expand Up @@ -716,22 +729,27 @@ func TestTargetInfoSanitizeLabelName(t *testing.T) {
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// add instance
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})

// add additional source attributes
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "cluster-id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
})

batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "target.ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
})
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
// add additional source attributes
{
Key: "cluster-id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
},
{
Key: "target.ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch}})

Expand Down Expand Up @@ -766,11 +784,18 @@ func TestTargetInfoWithJobAndInstanceOnly(t *testing.T) {
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// add instance
batch.Resource.Attributes = append(batch.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch}})

Expand Down Expand Up @@ -846,38 +871,56 @@ func TestTargetInfoWithDifferentBatches(t *testing.T) {
require.NoError(t, err)
defer p.Shutdown(context.Background())

// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)

// first batch will not have name-space nor instance id
// this will NOT create target_info metrics since it only has job but no other attributes

batch2 := test.MakeBatch(10, nil)
// TODO give these spans some duration so we can verify latencies are recorded correctly, in fact we should also test with various span names etc.
batch := test.MakeBatch(10, nil)
batch.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
}

// batch 2 will have instance id & cluster
// this will create a target_info metric with job, instance, and cluster
batch2 := test.MakeBatch(10, nil)
batch2.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add instance
{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
},
// add cluster
{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
},
}

// add instance
batch2.Resource.Attributes = append(batch2.Resource.Attributes, &common_v1.KeyValue{
Key: "service.instance.id",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "abc-instance-id-test-def"}},
})

// add additional source attributes
batch2.Resource.Attributes = append(batch2.Resource.Attributes, &common_v1.KeyValue{
Key: "cluster",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "eu-west-0"}},
})

batch3 := test.MakeBatch(10, nil)
// batch 3 will have ip
// this will create a target_info metric with job and ip only // no cluster no instance
batch3 := test.MakeBatch(10, nil)

// add additional source attributes
batch3.Resource.Attributes = append(batch3.Resource.Attributes, &common_v1.KeyValue{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
})
batch3.Resource.Attributes = []*common_v1.KeyValue{
// add service name
{
Key: "service.name",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "test-service"}},
},
// add ip
{
Key: "ip",
Value: &common_v1.AnyValue{Value: &common_v1.AnyValue_StringValue{StringValue: "1.1.1.1"}},
},
}

p.PushSpans(context.Background(), &tempopb.PushSpansRequest{Batches: []*trace_v1.ResourceSpans{batch, batch2, batch3}})

Expand Down
Loading