[kafka] bytecode instrumentation - sync #3055

lachmatt · 2023-10-30T08:50:50Z

Why

Towards #2854

What

Bytecode instrumentation for Confluent.Kafka client.

Instrumentation is based on datadog's instrumentation. @zacharycmontoya please take a look and approve if you accept it from datadog's side.

Implementation follows appropriate semantic conventions for messaging and kafka specifically.

publish spans are started at the beginning of Produce method, and finished at the end of it (or when delivery handler completes, if set).
~~process spans are started at the end of Consume method, and finished when next Consume method starts, consumer is closed, disposed, or unsubscribes from the topic.~~
receive spans are started and stopped at the end of Consume method (appropriate comment in code added)

Note:
peer.service (recommended) and server.address (conditionally required, if available) attributes are not being added right now.
As per spec, their value should be a name of the broker given message was sent to/received from.
I haven't found a good way to extract that information.

TODO:
Instrumentation of async API (i.e ProduceAsync), preferrably in a separate PR.

Tests

Included in PR.

Checklist

CHANGELOG.md is updated.
Documentation is updated.
New features are covered by tests.

test/test-applications/integrations/TestApplication.Kafka/TestApplication.Kafka.csproj

docs/config.md

src/OpenTelemetry.AutoInstrumentation/Configurations/TracerInstrumentation.cs

Kielek

In general LGTM.
I have some doubts if we should use attribtes or tags in our codebase.
.NET api is using diffferent names than OTel

test/IntegrationTests/KafkaCollection.cs

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/MessagingTags.cs

...elemetry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerCloseIntegration.cs

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/KafkaCommon.cs

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConstructorIntegration.cs

lachmatt · 2023-10-31T14:15:21Z

I will adjust duck types to follow best practices from the doc.

zacharycmontoya · 2023-11-01T01:15:42Z

Thanks for posting the PR! I can take a look later this week

docs/config.md

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/MessagingAttributes.cs

pjanotti · 2023-11-06T19:25:33Z

...elemetry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerCloseIntegration.cs

+
+    internal static CallTargetReturn OnMethodEnd<TTarget>(TTarget instance, Exception? exception, in CallTargetState state)
+    {
+        ConsumerCache.Remove(instance!);


Can we really guarantee that instance is not null here?

This is an instance method so it's safe to assume that the instance is not null

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/KafkaCommon.cs

zacharycmontoya · 2023-11-10T21:02:34Z

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConsumeSyncIntegration.cs

+        return CallTargetState.GetDefault();
+    }
+
+    internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state)


Since TTarget is always Confluent.Kafka.Consumer'2, you can always duck type (no chance of failure unless their API changes) and avoid doing it later. The only thing that changes is that now you'll need to change your call to the ConsumerCache so it's like: ConsumerCache.TryGet(instance.Instance!, out var groupId)

Suggested change

internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state)

internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state)

where TTarget : IClientName, IDuckType

It seems there is additional investigation needed as this approach doesn't work for me. I created an #3136 for tracking purposes.

zacharycmontoya · 2023-11-10T23:28:00Z

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ProducerProduceSyncIntegration.cs

+{
+    internal static CallTargetState OnMethodBegin<TTarget, TTopicPartition, TMessage, TDeliveryHandler>(
+        TTarget instance, TTopicPartition topicPartition, TMessage message, TDeliveryHandler deliveryHandler)
+        where TMessage : IKafkaMessage, IDuckType


Similar to my other suggestion, you can duck type immediately here instead of later on, since we know the type will always have the right structure.

Suggested change

where TMessage : IKafkaMessage, IDuckType

where TTarget : IClientName, IDuckType

where TMessage : IKafkaMessage, IDuckType

#3055 (comment)
Tracked under #3136

zacharycmontoya · 2023-11-10T23:31:01Z

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ProducerProduceSyncIntegration.cs

+            }
+
+            // Store as state information if delivery handler was set
+            return new CallTargetState(activity, deliveryHandler is null);


By passing a bool as the object state argument, you're unnecessarily doing a box operation. I suggest you pass the deliveryHandler object as-is and later do the null-check

fixed in fb2e35b

zacharycmontoya

LGTM with a couple of small suggestions

pyohannes

I gave this a review in the light of the very recent changes to messaging semantic conventions.

pyohannes · 2023-11-13T13:19:12Z

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConsumeSyncIntegration.cs

+            spanName = $"{consumeResult.Topic} {MessagingAttributes.Values.ProcessOperationName}";
+        }
+
+        spanName ??= MessagingAttributes.Values.ProcessOperationName;


The "process" operation was removed in favor of a "deliver" operation for push-based scenarios, and a "receive" operation for pull-based scenarios.

As this instruments a polling method, the operation name should be "receive".

https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md#operation-names

@pyohannes
Apart from adjusting operation name, I think where activity is stopped should be adjusted as well.
receive operation is defined to be connected with a consumer requesting one or more messages.
Taking that into account, I think activity created here should be stopped here as well.
Let me know if that makes sense.

That's correct, good catch.

Your current logic will result in correct "process" span durations only if the following holds true:

// Attempting to consume another message marks the end
// of a previous message processing

However, there are many consumption scenarios where this is not the case. For example, think about fetching messages and then distributing them to different threads for processing.

You should create "Receive" spans here measuring the duration of the Consume operation and linking to the context attached to the message.

changed in fb2e35b

pyohannes · 2023-11-13T13:22:19Z

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConsumeSyncIntegration.cs

+        }
+
+        spanName ??= MessagingAttributes.Values.ProcessOperationName;
+        var activity = KafkaCommon.Source.StartActivity(spanName, ActivityKind.Consumer, propagatedContext.ActivityContext);


It is now required to link to the propagated context (instead of parenting).

I'd recommend to link to the extracted context, and parent with the currently activated context, if there's any.

https://github.com/open-telemetry/semantic-conventions/blob/main/docs/messaging/messaging-spans.md#consumer-spans

changed in fb2e35b

There is additional section in OTEP that I think applies here:

For single-message scenarios, and if the "Deliver" or "Receive" spans would be
root spans of a new trace, the creation context may also be used as a parent on
those operations in addition to being added as a link. Keeping single-messages
operations in the same trace can greatly improve the user experience.

Situation described above will often be the case here.

I noticed this was discussed in a PR updating semantic conventions for messaging, but finally removed.

Would your recommendation be to not do it at all, or maybe add a configuration option to allow to opt-in for this behavior?

Yes, we didn't want to give a clear recommendation for this in the specification.

However, when you think that in your case it's beneficial for the user experience, it's perfectly fine to set the remote context as parent (in addition to setting it as link). I would only do this when there's no current activity.

pyohannes · 2023-11-13T13:27:20Z

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/KafkaCommon.cs

+        }
+    }
+
+    public static void SetCommonAttributes(


I strongly recommend to also set server.address, as this will be crucial to differentiate between different Kafka instances.

The semantic conventions mandate it as "required if available".

Unfortunately, I haven't found a good way to extract that information, as mentioned in PR's description. I created #3137 for tracking.

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConsumeSyncIntegration.cs

…ntegrations/ConsumerConsumeSyncIntegration.cs

src/OpenTelemetry.AutoInstrumentation/.publicApi/net462/PublicAPI.Shipped.txt

lachmatt added 3 commits October 30, 2023 09:38

kafka instrumentation - sync

bbf764a

remove bom

6e3df61

remove BOM in additional file

8471219

Kielek reviewed Oct 30, 2023

View reviewed changes

test/test-applications/integrations/TestApplication.Kafka/TestApplication.Kafka.csproj Outdated Show resolved Hide resolved

docs/config.md Outdated Show resolved Hide resolved

src/OpenTelemetry.AutoInstrumentation/Configurations/TracerInstrumentation.cs Show resolved Hide resolved

lachmatt added 2 commits October 30, 2023 10:27

initial pr feedback

eb63351

duck types cleanup

9f5bb92

Kielek reviewed Oct 30, 2023

View reviewed changes

lachmatt added 4 commits October 30, 2023 14:01

tag to attribute rename

2393a60

fix header

5d92a7d

Merge branch 'main' into kafka-instrumentation

c57b6d9

dockerfiles for kafka and zookeeper

a58ffdf

lachmatt marked this pull request as ready for review October 31, 2023 11:20

lachmatt requested a review from a team October 31, 2023 11:20

set only groupid conditionally

e326123

lachmatt force-pushed the kafka-instrumentation branch from 2473c32 to e326123 Compare October 31, 2023 15:38

lachmatt added 2 commits November 2, 2023 09:42

Merge branch 'main' into kafka-instrumentation

59128d7

duck typing improvements

7ca6ee1

Kielek assigned zacharycmontoya Nov 6, 2023

Merge branch 'main' into kafka-instrumentation

3712a5b

nrcventura reviewed Nov 6, 2023

View reviewed changes

docs/config.md Show resolved Hide resolved

pjanotti reviewed Nov 6, 2023

View reviewed changes

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/MessagingAttributes.cs Show resolved Hide resolved

pjanotti reviewed Nov 6, 2023

View reviewed changes

lachmatt added 2 commits November 7, 2023 07:44

remove duplicated line

cb209c3

add links to semantic conventions repo

f528de7

lachmatt commented Nov 7, 2023

View reviewed changes

src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/KafkaCommon.cs Outdated Show resolved Hide resolved

lachmatt added 3 commits November 8, 2023 13:39

fixed headers instantiation

5f84b9a

Merge branch 'main' into kafka-instrumentation

47e64ba

cleanup

d8c429d

zacharycmontoya reviewed Nov 10, 2023

View reviewed changes

zacharycmontoya approved these changes Nov 10, 2023

View reviewed changes

pyohannes reviewed Nov 13, 2023

View reviewed changes

Merge branch 'main' into kafka-instrumentation

9ae0c78

lachmatt marked this pull request as draft November 17, 2023 15:02

pr feedback

fb2e35b

lachmatt marked this pull request as ready for review November 20, 2023 08:02

Merge branch 'main' into kafka-instrumentation

6d898f7

lachmatt commented Nov 20, 2023

View reviewed changes

...ry.AutoInstrumentation/Instrumentations/Kafka/Integrations/ConsumerConsumeSyncIntegration.cs Outdated Show resolved Hide resolved

Update src/OpenTelemetry.AutoInstrumentation/Instrumentations/Kafka/I…

ded1dcc

…ntegrations/ConsumerConsumeSyncIntegration.cs

This was referenced Nov 20, 2023

[kafka] duck typing improvements #3136

Closed

[kafka] add server.address attribute #3137

Closed

lachmatt added 2 commits November 20, 2023 12:47

Merge branch 'main' into kafka-instrumentation

0cbe0f4

Merge branch 'main' into kafka-instrumentation

a494a95

Kielek reviewed Nov 22, 2023

View reviewed changes

src/OpenTelemetry.AutoInstrumentation/.publicApi/net462/PublicAPI.Shipped.txt Outdated Show resolved Hide resolved

lachmatt added 2 commits November 22, 2023 09:30

reflect api changes correctly

e7ccb7a

Merge branch 'main' into kafka-instrumentation

357dc48

Kielek approved these changes Nov 22, 2023

View reviewed changes

Merge branch 'main' into kafka-instrumentation

1fbfe86

Kielek enabled auto-merge (squash) November 22, 2023 09:29

Merge branch 'main' into kafka-instrumentation

2a9183f

Kielek merged commit a894ff7 into open-telemetry:main Nov 22, 2023

lachmatt deleted the kafka-instrumentation branch November 24, 2023 12:49

lachmatt mentioned this pull request Nov 28, 2023

[kafka] ProduceAsync instrumentation #3165

Merged

1 task

lachmatt mentioned this pull request Dec 5, 2023

[kafka] set remote context as a parent for some consumer spans #3175

Merged

1 task

Kielek mentioned this pull request Dec 6, 2023

Add kafka client bytecode instrumentation #2854

Closed

pyohannes mentioned this pull request Dec 21, 2023

Create prototypes to validate the proposed semantic conventions for messaging open-telemetry/semantic-conventions#653

Open

lachmatt mentioned this pull request May 15, 2024

Change AccountService from Go to DotNet (auto) open-telemetry/opentelemetry-demo#1538

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[kafka] bytecode instrumentation - sync #3055

[kafka] bytecode instrumentation - sync #3055

lachmatt commented Oct 30, 2023 •

edited

Loading

Kielek left a comment

lachmatt commented Oct 31, 2023

zacharycmontoya commented Nov 1, 2023

pjanotti Nov 6, 2023

zacharycmontoya Nov 10, 2023

zacharycmontoya Nov 10, 2023

lachmatt Nov 20, 2023

zacharycmontoya Nov 10, 2023

lachmatt Nov 20, 2023 •

edited

Loading

zacharycmontoya Nov 10, 2023

lachmatt Nov 20, 2023

zacharycmontoya left a comment

pyohannes left a comment

pyohannes Nov 13, 2023 •

edited

Loading

lachmatt Nov 16, 2023 •

edited

Loading

pyohannes Nov 16, 2023

lachmatt Nov 20, 2023

pyohannes Nov 13, 2023 •

edited

Loading

lachmatt Nov 20, 2023

lachmatt Nov 20, 2023 •

edited

Loading

pyohannes Nov 20, 2023

pyohannes Nov 13, 2023

lachmatt Nov 20, 2023

	internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state)
	internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTarget instance, TResponse response, Exception? exception, in CallTargetState state)
	where TTarget : IClientName, IDuckType

	where TMessage : IKafkaMessage, IDuckType
	where TTarget : IClientName, IDuckType
	where TMessage : IKafkaMessage, IDuckType

[kafka] bytecode instrumentation - sync #3055

[kafka] bytecode instrumentation - sync #3055

Conversation

lachmatt commented Oct 30, 2023 • edited Loading

Why

What

Checklist

Kielek left a comment

Choose a reason for hiding this comment

lachmatt commented Oct 31, 2023

zacharycmontoya commented Nov 1, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lachmatt Nov 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zacharycmontoya left a comment

Choose a reason for hiding this comment

pyohannes left a comment

Choose a reason for hiding this comment

pyohannes Nov 13, 2023 • edited Loading

Choose a reason for hiding this comment

lachmatt Nov 16, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pyohannes Nov 13, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lachmatt Nov 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

lachmatt commented Oct 30, 2023 •

edited

Loading

lachmatt Nov 20, 2023 •

edited

Loading

pyohannes Nov 13, 2023 •

edited

Loading

lachmatt Nov 16, 2023 •

edited

Loading

pyohannes Nov 13, 2023 •

edited

Loading

lachmatt Nov 20, 2023 •

edited

Loading