-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Rd 12757 java tracer support kafka instrumentation (#67)
- Loading branch information
Showing
25 changed files
with
1,228 additions
and
83 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
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
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
71 changes: 71 additions & 0 deletions
71
src/main/java/io/lumigo/core/instrumentation/impl/ApacheKafkaConsumerInstrumentation.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,71 @@ | ||
| package io.lumigo.core.instrumentation.impl; | ||
|
|
||
| import static net.bytebuddy.matcher.ElementMatchers.*; | ||
|
|
||
| import io.lumigo.core.SpansContainer; | ||
| import io.lumigo.core.instrumentation.LumigoInstrumentationApi; | ||
| import io.lumigo.core.instrumentation.agent.Loader; | ||
| import io.lumigo.core.utils.LRUCache; | ||
| import net.bytebuddy.agent.builder.AgentBuilder; | ||
| import net.bytebuddy.asm.Advice; | ||
| import net.bytebuddy.description.type.TypeDescription; | ||
| import net.bytebuddy.matcher.ElementMatcher; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecords; | ||
| import org.apache.kafka.clients.consumer.KafkaConsumer; | ||
| import org.apache.kafka.clients.consumer.internals.ConsumerMetadata; | ||
| import org.pmw.tinylog.Logger; | ||
|
|
||
| public class ApacheKafkaConsumerInstrumentation implements LumigoInstrumentationApi { | ||
|
|
||
| public static final String INSTRUMENTATION_PACKAGE_PREFIX = "org.apache.kafka.clients.consumer"; | ||
|
|
||
| @Override | ||
| public ElementMatcher<TypeDescription> getTypeMatcher() { | ||
| return named("org.apache.kafka.clients.consumer.KafkaConsumer"); | ||
| } | ||
|
|
||
| @Override | ||
| public AgentBuilder.Transformer.ForAdvice getTransformer() { | ||
| return new AgentBuilder.Transformer.ForAdvice() | ||
| .include(Loader.class.getClassLoader()) | ||
| .advice( | ||
| isMethod() | ||
| .and(isPublic()) | ||
| .and(named("poll")) | ||
| .and(takesArguments(1)) | ||
| .and( | ||
| returns( | ||
| named( | ||
| "org.apache.kafka.clients.consumer.ConsumerRecords"))), | ||
| ApacheKafkaConsumerAdvice.class.getName()); | ||
| } | ||
|
|
||
| public static class ApacheKafkaConsumerAdvice { | ||
| public static final SpansContainer spansContainer = SpansContainer.getInstance(); | ||
| public static final LRUCache<String, Long> startTimeMap = new LRUCache<>(1000); | ||
|
|
||
| @Advice.OnMethodEnter(suppress = Throwable.class) | ||
| public static void methodEnter(@Advice.FieldValue("clientId") String clientId) { | ||
| try { | ||
| startTimeMap.put(clientId, System.currentTimeMillis()); | ||
| } catch (Exception e) { | ||
| Logger.error(e); | ||
| } | ||
| } | ||
|
|
||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void methodExit( | ||
| @Advice.This KafkaConsumer<?, ?> consumer, | ||
| @Advice.FieldValue("metadata") ConsumerMetadata metadata, | ||
| @Advice.FieldValue("clientId") String clientId, | ||
| @Advice.Return ConsumerRecords<?, ?> consumerRecords) { | ||
| try { | ||
| Logger.info("Handling kafka request {}", consumerRecords.hashCode()); | ||
| spansContainer.addKafkaConsumeSpan( | ||
| startTimeMap.get(clientId), consumer, metadata, consumerRecords); | ||
| } catch (Throwable error) { | ||
| Logger.error(error, "Failed to add kafka span"); | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.