-
Notifications
You must be signed in to change notification settings - Fork 879
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Make CONSUMER receive span a parent of CONSUMER process spans in kafk…
…a-streams (#4151)
- Loading branch information
Mateusz Rzeszutek
authored
Sep 18, 2021
1 parent
3bc0e07
commit 1a6294e
Showing
12 changed files
with
274 additions
and
87 deletions.
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
17 changes: 17 additions & 0 deletions
17
...n/java/io/opentelemetry/javaagent/instrumentation/kafka/KafkaConsumerIterableWrapper.java
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,17 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.kafka; | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerRecord; | ||
|
|
||
| public interface KafkaConsumerIterableWrapper<K, V> { | ||
|
|
||
| /** | ||
| * Returns the actual, non-tracing iterable. This method is only supposed to be used by other | ||
| * Kafka consumer instrumentations that want to suppress the kafka-clients one. | ||
| */ | ||
| Iterable<ConsumerRecord<K, V>> unwrap(); | ||
| } |
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
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
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
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
66 changes: 66 additions & 0 deletions
66
...entelemetry/javaagent/instrumentation/kafkastreams/RecordDeserializerInstrumentation.java
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,66 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.kafkastreams; | ||
|
|
||
| import static io.opentelemetry.javaagent.extension.matcher.AgentElementMatchers.hasClassesNamed; | ||
| import static net.bytebuddy.matcher.ElementMatchers.isInterface; | ||
| import static net.bytebuddy.matcher.ElementMatchers.isMethod; | ||
| import static net.bytebuddy.matcher.ElementMatchers.isPackagePrivate; | ||
| import static net.bytebuddy.matcher.ElementMatchers.named; | ||
| import static net.bytebuddy.matcher.ElementMatchers.not; | ||
| import static net.bytebuddy.matcher.ElementMatchers.returns; | ||
| import static net.bytebuddy.matcher.ElementMatchers.takesArgument; | ||
|
|
||
| import io.opentelemetry.api.trace.SpanContext; | ||
| import io.opentelemetry.javaagent.extension.instrumentation.TypeInstrumentation; | ||
| import io.opentelemetry.javaagent.extension.instrumentation.TypeTransformer; | ||
| import io.opentelemetry.javaagent.instrumentation.api.ContextStore; | ||
| import io.opentelemetry.javaagent.instrumentation.api.InstrumentationContext; | ||
| import net.bytebuddy.asm.Advice; | ||
| import net.bytebuddy.description.type.TypeDescription; | ||
| import net.bytebuddy.matcher.ElementMatcher; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecord; | ||
|
|
||
| // at some point in time SourceNodeRecordDeserializer was refactored into RecordDeserializer | ||
| public class RecordDeserializerInstrumentation implements TypeInstrumentation { | ||
|
|
||
| @Override | ||
| public ElementMatcher<ClassLoader> classLoaderOptimization() { | ||
| return hasClassesNamed("org.apache.kafka.streams.processor.internals.RecordDeserializer"); | ||
| } | ||
|
|
||
| @Override | ||
| public ElementMatcher<TypeDescription> typeMatcher() { | ||
| return named("org.apache.kafka.streams.processor.internals.RecordDeserializer") | ||
| .and(not(isInterface())); | ||
| } | ||
|
|
||
| @Override | ||
| public void transform(TypeTransformer transformer) { | ||
| transformer.applyAdviceToMethod( | ||
| isMethod() | ||
| .and(isPackagePrivate()) | ||
| .and(named("deserialize")) | ||
| .and(takesArgument(1, named("org.apache.kafka.clients.consumer.ConsumerRecord"))) | ||
| .and(returns(named("org.apache.kafka.clients.consumer.ConsumerRecord"))), | ||
| RecordDeserializerInstrumentation.class.getName() + "$DeserializeAdvice"); | ||
| } | ||
|
|
||
| @SuppressWarnings("unused") | ||
| public static class DeserializeAdvice { | ||
|
|
||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void onExit( | ||
| @Advice.Argument(1) ConsumerRecord<?, ?> incoming, | ||
| @Advice.Return(readOnly = false) ConsumerRecord<?, ?> result) { | ||
|
|
||
| // copy the receive CONSUMER span association | ||
| ContextStore<ConsumerRecord, SpanContext> singleRecordReceiveSpan = | ||
| InstrumentationContext.get(ConsumerRecord.class, SpanContext.class); | ||
| singleRecordReceiveSpan.put(result, singleRecordReceiveSpan.get(incoming)); | ||
| } | ||
| } | ||
| } |
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.