Add kafka instrumentation by andrewlock · Pull Request #1444 · DataDog/dd-trace-dotnet

andrewlock · 2021-05-04T12:13:09Z

Adds support for instrumenting Confluent.Kafka v1.4.0+

Producers inject the span context into Kafka record headers. Consumers generate a span when a message is received using the extracted propagated context, and leave it open until the next call to Consumer.Consume() (or Consumer.Close()/Dispose()/Unsubscribe()).

This approach is required due to the design of the Confluent.Kafka Consumer API, and assumes that you use the "standard" consumer setup of a while(true) loop, as shown in examples, e.g.:

using var consumer = new ConsumerBuilder<Ignore, string>(config).Build();
consumer.Subscribe(topics);
while (true)
{
    // blocks until a message is available, closes existing consume span if open
    var consumeResult = consumer.Consume(cancellationToken);
    // on completion, context propagated from producer and new span created
    // ...use consumerResult in handler
}

// Close final span
consumer.Close();

If customers are using a different setup, then they can disable the consumer behaviour using DD_TRACE_KAFKA_CREATE_CONSUMER_SCOPE_ENABLED=false. This will still create the span for the ConsumeResult, but will immediately close it, rather than leaving it open. Customers can then manually extract the headers from the ConsumeResult to reconstruct the context for manual instrumentation as they see fit

Metadata

Service = {service}-kafka
Span kind = producer | consumer
Span type = queue
Span name = kafka.produce | kafka.consume
Resource Name = {Produce|Consume} Topic {TopicName}
Tags
- component=kafka
- kafka.partition
- kafka.offset
- kafka.tombstone
Metrics
- message.queue_time_ms

@DataDog/apm-dotnet

colin-higgins · 2021-05-10T17:08:25Z

 wait-for-it mongo:27017 -- \
 wait-for-it postgres:5432 -- \
+wait-for-it -t 60 kafka-zookeeper:2181 -- \
+wait-for-it -t 60 kafka-broker:9092 -- \


I like that you added a timeout, we should probably consider adding a more reasonable one for each

The default is 15s I think, kafka is just slooooow. Though because of the way we start the dependencies using docker-compose, in the vast majority of cases these all immediately succeed after 0 seconds

zacharycmontoya · 2021-05-12T22:03:35Z

+                    return;
+                }
+
+                // TODO: record end-to-end time?


The end-to-end time is already recorded as the Span duration. Did you have something else in mind?

The Java implementation records the trace end-to-end time. That requires passing the trace start time around as baggage (as the start will be a distributed trace!) Seems like a lot of work, and to my mind would require sending that data in all distributed spans, which touches a lot of places. I'm not convinced of the usefulness, what do you think?

https://github.com/DataDog/dd-trace-java/blob/e5f2514a90b04f4f7620d1497207b5a0224d0a3e/dd-java-agent/instrumentation/kafka-clients-0.11/src/main/java/datadog/trace/instrumentation/kafka_clients/KafkaDecorator.java#L93-L105

passing the trace start time around as baggage

If baggage is a requirement, the .NET Tracer doesn't have baggage support yet.
(edit: aside from a few specific fields like sampling priority or origin)

the .NET Tracer doesn't have baggage support yet

Well that solves that then 😄

zacharycmontoya · 2021-05-13T00:37:40Z

+                {
+                    try
+                    {
+                        inst.Handler = newAction;


I don't understand why we need to re-set the Handler field inside of a new Action instead of just executing the code. Is there a specific reason for that? If we can avoid that, then that's one fewer allocation

I think we have to do it this way.

We need to set the handler in MethodEnd, not in MethodBegin. But we have to generate the newAction variable in MethodBegin as that's where we have access to the TActionOfDeliveryReport type.

With the approach you're suggesting, we would need to pass the newAction as state in CallTargetState. But then we wouldn't be able to cast it to the correct type in MethodEnd, as we don't have the TActionOfDeliveryReport generic parameter there.

So I think this approach is the only one that is really doable - I couldn't find a way to set it directly in MethodEnd without either losing the benefits of CachedWrapperDelegate, or having to set the Handler field through reflection?

You're right, given all the constraints this does seem like the best way to do it.

However, what if instead of instrumenting the constructor of Confluent.Kafka.Producer'2+TypedDeliveryHandlerShim_Action we instead just instrument the HandleDeliveryReport method itself? Looking at the decompilation for Confluent.Kafka 1.4.3.0, all the fields we are reading come straight from the TopicPartitionOffsetError property of the DeliveryReport argument and these values look to be the same when they're eventually passed to the handler. That could remove a lot of this complexity and possibly some allocations. Have you already tried this approach and, if so, were there some gotcha's there?

Absolutely agree this would be an easier approach, and we did consider it. The main problems with this approach are:

You lose the correct "start time" for the produce span

More critically, we'd lose the correct parent span, given that the callback runs once the Produce() method has finished, and potentially a different span is active..

We don't have access to the message Value in the callback, so wouldn't be able to tag Tombstones any more

No. 2 is the big problem, and the main reason we went for the current, more complex, approach. Does that make sense to you?

Got it, I didn't realize how crazy this instrumentation scenario would be. Thanks for taking the time to respond to everything 😄

This is a Kafka design issue though, rather than an instrumentation issue - those consumes are correct, so we expect _at least_ x spans to be generated for consumers.

Also renamed record -> message as seems to be more common nomenclature

- Test multiple packages - sln file issues

andrewlock force-pushed the andrew/kafka branch 2 times, most recently from f647672 to eb156f2 Compare May 4, 2021 17:23

andrewlock added area:automatic-instrumentation Automatic instrumentation managed C# code (Datadog.Trace.ClrProfiler.Managed) type:new-feature labels May 4, 2021

andrewlock force-pushed the andrew/kafka branch 5 times, most recently from a8e5906 to 41e2fbf Compare May 7, 2021 15:55

andrewlock marked this pull request as ready for review May 7, 2021 18:10

andrewlock requested a review from a team as a code owner May 7, 2021 18:10

andrewlock force-pushed the andrew/kafka branch from 41e2fbf to 95b787d Compare May 10, 2021 16:03

colin-higgins reviewed May 10, 2021

View reviewed changes

andrewlock requested a review from a team May 11, 2021 09:53

andrewlock force-pushed the andrew/kafka branch from 95b787d to 61847f0 Compare May 11, 2021 10:08