Add Kafka consumer group tag by andrewlock · Pull Request #3099 · DataDog/dd-trace-dotnet

andrewlock · 2022-08-18T16:12:47Z

Summary of changes

Adds the consumer group as a Tag to Kafka spans

Reason for change

We need to capture the consumer group for data streams monitoring, so may as well tag it in the span as it's available.

Implementation details

This is annoyingly hard, as it's only exposed indirectly in the Consumer constructor, in the config provided by ConsumerBuilder, and is not accessible later in the Consume method when we're actually creating the Span.

To work around that, we're using a ConditionalWeakTable to create a map from a Consumer instance to a group ID. We then look up the group ID again when Consume is called. We explicitly remove the entry from the table when the consumer is closed or disposed.

Test coverage

Added extra asserts to the KafkaTests to ensure the tags are there

Other details

We should really convert those tests to snapshot tests... any volunteers? :hide:

pierotibou · 2022-08-19T07:05:09Z

+internal class ConsumerGroupHelper
+{
+    // A map between Kafka Consumer<TKey,TValue> and the corresponding consumer group
+    private static readonly ConditionalWeakTable<object, string> ConsumerToGroupIdMap = new();


pierotibou · 2022-08-19T07:20:37Z

+        /// <summary>
+        /// The consumer group that consumed the message
+        /// </summary>
+        internal const string KafkaConsumerGroup = "kafka.group";


FYI, each time we will add a new tag in the near future, we need to make sure it is already handled by the TagRemapping done on the backend side. This one already existed in Java so we're good. Here's the code for later reference

tonyredondo

LGTM, just a question.

tonyredondo · 2022-08-19T20:36:33Z

+
+    public static void SetConsumerGroup(object consumer, string groupId)
+    {
+        ConsumerToGroupIdMap.GetValue(consumer, x => groupId);


why not ConsumerToGroupIdMap.AddOrUpdate(consumer, groupId); to avoid the lambda?

AddOrUpdate doesn't exist < netcoreapp3.1. Will add an #if, thanks 🙂

zacharycmontoya · 2022-08-19T21:08:53Z

        [Tag(Trace.Tags.KafkaTombstone)]
        public string Tombstone { get; set; }

+        [Tag(Trace.Tags.KafkaConsumerGroup)]


Looks like we should make the SpanMetadataRules more strict...I really would have liked for the existing integration tests to fail because the tag key kafka.group exists on some spans now, even though it's not specified in SpanMetadataRules.cs

Yeah, I guess we don't check that that tags don't exist currently. I agree we potentially want to add that functionality, though we also probably also want/need to make it easy to add global tags to all spans too. I added the optional kafka.group span in this PR anyway, and we can look into it down the line

We need to capture the consumer group for data streams monitoring This is annoyingly hard, as it's only exposed indirectly in the `Consumer` constructor, in the config provided by `ConsumerBuilder`, and is not accessible later in the `Consume` method. To work around that, using a `ConditionalWeakTable` to create a map from `Consumer` instance to a group ID. We then look up the group ID again when `Consume` is called. We explicitly remove the entry from the table when the consumer is closed or disposed.

andrewlock · 2022-08-22T10:53:06Z

Benchmarks Report 🐌

Benchmarks for #3099 compared to master:

1 benchmarks are faster, with geometric mean 1.259
All benchmarks have the same allocations

The following thresholds were used for comparing the benchmark speeds:

Mann–Whitney U test with statistical test for significance of 5%
Only results indicating a difference greater than 10% and 0.3 ns are considered.

Allocation changes below 0.5% are ignored.

Benchmark details

Benchmarks.Trace.AgentWriterBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`WriteAndFlushEnrichedTraces`	net472	765μs	353ns	1.37μs	0.365	3.18 KB
master	`WriteAndFlushEnrichedTraces`	netcoreapp3.1	455μs	197ns	709ns	0	2.58 KB
#3099	`WriteAndFlushEnrichedTraces`	net472	718μs	513ns	1.99μs	0.361	3.18 KB
#3099	`WriteAndFlushEnrichedTraces`	netcoreapp3.1	458μs	134ns	482ns	0	2.58 KB

Benchmarks.Trace.AppSecBodyBenchmark - Faster 🎉 Same allocations ✔️

Faster 🎉 in #3099

Benchmark	base/diff	Base Median (ns)	Diff Median (ns)	Modality
Benchmarks.Trace.AppSecBodyBenchmark.AllCycleMoreComplexBody‑net472	1.259	226.60	179.95

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Gen 1	Allocated
master	`AllCycleSimpleBody`	net472	182ns	0.16ns	0.6ns	0.0675	0	425 B
master	`AllCycleSimpleBody`	netcoreapp3.1	237ns	0.272ns	1.05ns	0.00578	0	424 B
master	`AllCycleMoreComplexBody`	net472	227ns	0.22ns	0.825ns	0.0637	0	401 B
master	`AllCycleMoreComplexBody`	netcoreapp3.1	235ns	0.251ns	0.972ns	0.00551	0	400 B
master	`BodyExtractorSimpleBody`	net472	255ns	0.294ns	1.1ns	0.0574	0	361 B
master	`BodyExtractorSimpleBody`	netcoreapp3.1	230ns	0.216ns	0.779ns	0.00369	0	272 B
master	`BodyExtractorMoreComplexBody`	net472	14.2μs	13.4ns	50.1ns	1.2	0.0214	7.62 KB
master	`BodyExtractorMoreComplexBody`	netcoreapp3.1	12.2μs	16.7ns	64.8ns	0.0904	0	6.75 KB
#3099	`AllCycleSimpleBody`	net472	181ns	0.195ns	0.755ns	0.0676	0	425 B
#3099	`AllCycleSimpleBody`	netcoreapp3.1	235ns	0.204ns	0.789ns	0.00588	0	424 B
#3099	`AllCycleMoreComplexBody`	net472	180ns	0.196ns	0.76ns	0.0637	0	401 B
#3099	`AllCycleMoreComplexBody`	netcoreapp3.1	236ns	0.348ns	1.35ns	0.00544	0	400 B
#3099	`BodyExtractorSimpleBody`	net472	252ns	0.295ns	1.14ns	0.0573	0	361 B
#3099	`BodyExtractorSimpleBody`	netcoreapp3.1	233ns	0.18ns	0.696ns	0.00374	0	272 B
#3099	`BodyExtractorMoreComplexBody`	net472	14.4μs	12.5ns	46.8ns	1.2	0.0216	7.62 KB
#3099	`BodyExtractorMoreComplexBody`	netcoreapp3.1	12.4μs	17.6ns	68ns	0.0924	0	6.75 KB

Benchmarks.Trace.AspNetCoreBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`SendRequest`	net472	0ns	0ns	0ns	0	0 b
master	`SendRequest`	netcoreapp3.1	181μs	103ns	400ns	0.18	20.33 KB
#3099	`SendRequest`	net472	0ns	0ns	0ns	0	0 b
#3099	`SendRequest`	netcoreapp3.1	179μs	178ns	689ns	0.267	20.33 KB

Benchmarks.Trace.DbCommandBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Gen 1	Allocated
master	`ExecuteNonQuery`	net472	1.53μs	0.697ns	2.51ns	0.126	0.000762	794 B
master	`ExecuteNonQuery`	netcoreapp3.1	1.25μs	0.413ns	1.6ns	0.0107	0	824 B
#3099	`ExecuteNonQuery`	net472	1.57μs	0.539ns	2.02ns	0.126	0.00079	794 B
#3099	`ExecuteNonQuery`	netcoreapp3.1	1.21μs	0.554ns	2ns	0.0109	0	824 B

Benchmarks.Trace.ElasticsearchBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`CallElasticsearch`	net472	2.24μs	0.601ns	2.25ns	0.159	1 KB
master	`CallElasticsearch`	netcoreapp3.1	1.4μs	0.521ns	2.02ns	0.0133	984 B
master	`CallElasticsearchAsync`	net472	2.37μs	0.957ns	3.71ns	0.18	1.14 KB
master	`CallElasticsearchAsync`	netcoreapp3.1	1.42μs	0.434ns	1.68ns	0.015	1.1 KB
#3099	`CallElasticsearch`	net472	2.25μs	0.769ns	2.77ns	0.159	1 KB
#3099	`CallElasticsearch`	netcoreapp3.1	1.39μs	0.74ns	2.77ns	0.0132	984 B
#3099	`CallElasticsearchAsync`	net472	2.33μs	1.82ns	7.05ns	0.18	1.14 KB
#3099	`CallElasticsearchAsync`	netcoreapp3.1	1.51μs	0.391ns	1.41ns	0.015	1.1 KB

Benchmarks.Trace.GraphQLBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`ExecuteAsync`	net472	2.37μs	3.41ns	13.2ns	0.2	1.26 KB
master	`ExecuteAsync`	netcoreapp3.1	1.55μs	2.76ns	10.7ns	0.0163	1.22 KB
#3099	`ExecuteAsync`	net472	2.41μs	8.14ns	30.5ns	0.2	1.26 KB
#3099	`ExecuteAsync`	netcoreapp3.1	1.51μs	3.95ns	14.2ns	0.0163	1.22 KB

Benchmarks.Trace.HttpClientBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`SendAsync`	net472	5μs	9.14ns	35.4ns	0.392	2.48 KB
master	`SendAsync`	netcoreapp3.1	3.17μs	4.91ns	19ns	0.0311	2.36 KB
#3099	`SendAsync`	net472	4.85μs	8.35ns	31.2ns	0.393	2.48 KB
#3099	`SendAsync`	netcoreapp3.1	3.28μs	15.2ns	59ns	0.0322	2.36 KB

Benchmarks.Trace.ILoggerBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`EnrichedLog`	net472	2.85μs	1.07ns	4.14ns	0.263	1.66 KB
master	`EnrichedLog`	netcoreapp3.1	2.31μs	0.835ns	2.89ns	0.0231	1.73 KB
#3099	`EnrichedLog`	net472	2.82μs	8.99ns	33.6ns	0.263	1.66 KB
#3099	`EnrichedLog`	netcoreapp3.1	2.31μs	2.16ns	8.36ns	0.0234	1.73 KB

Benchmarks.Trace.Log4netBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Gen 1	Allocated
master	`EnrichedLog`	net472	148μs	51.8ns	201ns	0.67	0.223	4.5 KB
master	`EnrichedLog`	netcoreapp3.1	113μs	107ns	415ns	0.0563	0	4.38 KB
#3099	`EnrichedLog`	net472	151μs	93.6ns	363ns	0.669	0.223	4.5 KB
#3099	`EnrichedLog`	netcoreapp3.1	115μs	115ns	446ns	0	0	4.38 KB

Benchmarks.Trace.NLogBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Gen 1	Allocated
master	`EnrichedLog`	net472	5.42μs	4.87ns	18.2ns	0.543	0.00272	3.43 KB
master	`EnrichedLog`	netcoreapp3.1	4.14μs	8.88ns	34.4ns	0.0517	0	3.8 KB
#3099	`EnrichedLog`	net472	5.21μs	14.2ns	54.8ns	0.544	0.00255	3.43 KB
#3099	`EnrichedLog`	netcoreapp3.1	4.14μs	10.1ns	39.2ns	0.0528	0	3.8 KB

Benchmarks.Trace.RedisBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`SendReceive`	net472	2.01μs	0.768ns	2.88ns	0.194	1.22 KB
master	`SendReceive`	netcoreapp3.1	1.61μs	0.319ns	1.2ns	0.0161	1.21 KB
#3099	`SendReceive`	net472	2.02μs	1.14ns	4.26ns	0.194	1.22 KB
#3099	`SendReceive`	netcoreapp3.1	1.69μs	0.494ns	1.85ns	0.0163	1.21 KB

Benchmarks.Trace.SerilogBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`EnrichedLog`	net472	4.66μs	1.71ns	6.64ns	0.329	2.08 KB
master	`EnrichedLog`	netcoreapp3.1	4.13μs	1.4ns	5.22ns	0.0227	1.69 KB
#3099	`EnrichedLog`	net472	4.69μs	1.42ns	4.93ns	0.33	2.08 KB
#3099	`EnrichedLog`	netcoreapp3.1	4.11μs	1.64ns	6.34ns	0.0226	1.69 KB

Benchmarks.Trace.SpanBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`StartFinishSpan`	net472	871ns	1.03ns	3.99ns	0.105	658 B
master	`StartFinishSpan`	netcoreapp3.1	749ns	0.616ns	2.3ns	0.00894	648 B
master	`StartFinishScope`	net472	1.13μs	1.21ns	4.51ns	0.117	738 B
master	`StartFinishScope`	netcoreapp3.1	884ns	0.542ns	2.1ns	0.0106	768 B
#3099	`StartFinishSpan`	net472	831ns	0.814ns	3.15ns	0.105	658 B
#3099	`StartFinishSpan`	netcoreapp3.1	742ns	0.402ns	1.56ns	0.00887	648 B
#3099	`StartFinishScope`	net472	1.03μs	1.11ns	4.15ns	0.117	738 B
#3099	`StartFinishScope`	netcoreapp3.1	892ns	0.447ns	1.67ns	0.0107	768 B

Benchmarks.Trace.TraceAnnotationsBenchmark - Same speed ✔️ Same allocations ✔️

Raw results

Branch	Method	Toolchain	Mean	StdError	StdDev	Gen 0	Allocated
master	`RunOnMethodBegin`	net472	1.22μs	1.38ns	5.35ns	0.117	738 B
master	`RunOnMethodBegin`	netcoreapp3.1	1.01μs	0.778ns	2.91ns	0.0101	768 B
#3099	`RunOnMethodBegin`	net472	1.2μs	0.998ns	3.87ns	0.117	738 B
#3099	`RunOnMethodBegin`	netcoreapp3.1	1.01μs	0.52ns	1.87ns	0.0104	768 B

andrewlock · 2022-08-22T11:15:15Z

Code Coverage Report 📊

✔️ Merging #3099 into master will not change line coverage
✔️ Merging #3099 into master will not change branch coverage
⛔ Merging #3099 into master will will increase complexity by 35

	master	#3099	Change
Lines	`16876` / `23280`	`16906` / `23314`
Lines %	`72%`	`73%`	`0%` ✔️
Branches	`10071` / `14362`	`10095` / `14392`
Branches %	`70%`	`70%`	`0%` ✔️
Complexity	`15458`	`15493`	`35` ⛔

View the full report for further details:

Datadog.Trace Breakdown ✔️

	master	#3099	Change
Lines %	`72%`	`73%`	`0%` ✔️
Branches %	`70%`	`70%`	`0%` ✔️
Complexity	`15458`	`15493`	`35` ⛔

The following classes have significant coverage changes.

File	Line coverage change	Branch coverage change	Complexity change
Datadog.Trace.Ci.GitInfo	`-17%` ⛔	`-11%` ⛔	`0` ✔️
Datadog.Trace.Ci.CIVisibility	`5%` ✔️	`6%` ✔️	`0` ✔️

The following classes were added in #3099:

File	Line coverage	Branch coverage	Complexity
Datadog.Trace.ClrProfiler.AutoInstrumentation.Kafka.ConsumerGroupHelper	`100%`	`100%`	`5`
Datadog.Trace.ClrProfiler.AutoInstrumentation.Kafka.KafkaConsumerConstructorIntegration	`90%`	`83%`	`12`

View the full reports for further details:

andrewlock added type:enhancement Improvement to an existing feature area:automatic-instrumentation Automatic instrumentation managed C# code (Datadog.Trace.ClrProfiler.Managed) labels Aug 18, 2022

andrewlock requested a review from a team as a code owner August 18, 2022 16:12

This comment has been minimized.

Sign in to view

pierotibou approved these changes Aug 19, 2022

View reviewed changes

tonyredondo approved these changes Aug 19, 2022

View reviewed changes

zacharycmontoya reviewed Aug 19, 2022

View reviewed changes

andrewlock added 3 commits August 22, 2022 11:09

Add KafkaConsumerGroup tag

bbdeac0

PR feedback

dcce137

andrewlock force-pushed the andrew/add-kafka-consumer-group branch from bbc2185 to dcce137 Compare August 22, 2022 10:14

zacharycmontoya approved these changes Aug 22, 2022

View reviewed changes

andrewlock merged commit 8250fa8 into master Aug 23, 2022

andrewlock deleted the andrew/add-kafka-consumer-group branch August 23, 2022 13:41

github-actions Bot added this to the vNext milestone Aug 23, 2022

andrewlock added the area:data-streams-monitoring label Aug 31, 2022

This was referenced Sep 8, 2022

Add Kafka data streams monitoring implementation #3192

Merged

Add support for IDiscoveryService to DataStreamsManager #3197

Closed

This was referenced Sep 15, 2022

Added support for IDiscoveryService to DataStreamsManager #3220

Merged

GZip the data streams monitoring data sent to the agent #3249

Merged

Uh oh!

Conversation

andrewlock commented Aug 18, 2022

Summary of changes

Reason for change

Implementation details

Test coverage

Other details

Uh oh!

This comment has been minimized.

This comment has been minimized.

pierotibou Aug 19, 2022

Choose a reason for hiding this comment

Uh oh!

pierotibou Aug 19, 2022

Choose a reason for hiding this comment

Uh oh!

tonyredondo left a comment

Choose a reason for hiding this comment

Uh oh!

tonyredondo Aug 19, 2022

Choose a reason for hiding this comment

Uh oh!

andrewlock Aug 22, 2022

Choose a reason for hiding this comment

Uh oh!

zacharycmontoya Aug 19, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

andrewlock Aug 22, 2022

Choose a reason for hiding this comment

Uh oh!

andrewlock commented Aug 22, 2022

Benchmarks Report 🐌

Benchmark details

Raw results

Faster 🎉 in #3099

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Raw results

Uh oh!

andrewlock commented Aug 22, 2022

Code Coverage Report 📊

Datadog.Trace Breakdown ✔️

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

zacharycmontoya Aug 19, 2022 •

edited

Loading