-
Notifications
You must be signed in to change notification settings - Fork 6
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
199 additions
and
7 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
163 changes: 163 additions & 0 deletions
163
remora-kafka/src/main/java/com/jkoolcloud/remora/advices/KafkaConsumerClientAdvice.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,163 @@ | ||
| package com.jkoolcloud.remora.advices; | ||
|
|
||
| import static com.jkoolcloud.remora.core.utils.ReflectionUtils.getFieldValue; | ||
| import static net.bytebuddy.matcher.ElementMatchers.*; | ||
|
|
||
| import java.lang.instrument.Instrumentation; | ||
| import java.lang.reflect.Method; | ||
| import java.text.MessageFormat; | ||
|
|
||
| import org.apache.kafka.clients.consumer.Consumer; | ||
| import org.tinylog.Logger; | ||
| import org.tinylog.TaggedLogger; | ||
|
|
||
| import com.jkoolcloud.remora.RemoraConfig; | ||
| import com.jkoolcloud.remora.core.EntryDefinition; | ||
|
|
||
| import net.bytebuddy.agent.builder.AgentBuilder; | ||
| import net.bytebuddy.asm.Advice; | ||
| import net.bytebuddy.description.method.MethodDescription; | ||
| import net.bytebuddy.description.type.TypeDescription; | ||
| import net.bytebuddy.matcher.ElementMatcher; | ||
|
|
||
| public class KafkaConsumerClientAdvice extends BaseTransformers { | ||
|
|
||
| public static final String ADVICE_NAME = "KafkaConsumerClientAdvice"; | ||
| public static String[] INTERCEPTING_CLASS = { "org.apache.kafka.clients.consumer.Consumer" }; | ||
| public static String INTERCEPTING_METHOD = "poll"; | ||
|
|
||
| @RemoraConfig.Configurable | ||
| public static boolean load = true; | ||
| @RemoraConfig.Configurable | ||
| public static boolean logging = false; | ||
| public static TaggedLogger logger; | ||
| static AgentBuilder.Transformer.ForAdvice advice = new AgentBuilder.Transformer.ForAdvice() | ||
| .include(KafkaConsumerClientAdvice.class.getClassLoader()).include(RemoraConfig.INSTANCE.classLoader)// | ||
| .advice(methodMatcher(), KafkaConsumerClientAdvice.class.getName()); | ||
|
|
||
| /** | ||
| * Method matcher intended to match intercepted class method/s to instrument. See (@ElementMatcher) for available | ||
| * method matches. | ||
| */ | ||
|
|
||
| private static ElementMatcher<? super MethodDescription> methodMatcher() { | ||
| return named(INTERCEPTING_METHOD); | ||
| } | ||
|
|
||
| /** | ||
| * Advices before method is called before instrumented method code | ||
| * | ||
| * @param thiz | ||
| * reference to method object | ||
| * @param method | ||
| * instrumented method description | ||
| * @param ed | ||
| * {@link EntryDefinition} for collecting ant passing values to | ||
| * {@link com.jkoolcloud.remora.core.output.OutputManager} | ||
| * @param startTime | ||
| * method startTime | ||
| * | ||
| */ | ||
|
|
||
| @Advice.OnMethodEnter | ||
| public static void before(@Advice.This Consumer thiz, // | ||
| @Advice.Origin Method method, // | ||
| @Advice.Local("ed") EntryDefinition ed, // | ||
| @Advice.Local("startTime") long startTime) { | ||
| try { | ||
| if (ed == null) { | ||
| ed = new EntryDefinition(KafkaConsumerClientAdvice.class); | ||
| } | ||
| if (logging) { | ||
| logger.info(format("Entering: {0} {1}", KafkaConsumerClientAdvice.class.getName(), "before")); | ||
| } | ||
| ed.setTransparent(); | ||
| startTime = fillDefaultValuesBefore(ed, stackThreadLocal, thiz, method, logging ? logger : null); | ||
|
|
||
| ed.setEventType(EntryDefinition.EventType.CALL); | ||
| String clientId = getFieldValue(thiz, String.class, "clientId"); | ||
| String groupId = getFieldValue(thiz, String.class, "groupId"); | ||
| ed.setApplication(MessageFormat.format("clientId={}, groupId={}", clientId, groupId)); | ||
| } catch (Throwable t) { | ||
| handleAdviceException(t, ADVICE_NAME, logging ? logger : null); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Method called on instrumented method finished. | ||
| * | ||
| * @param method | ||
| * instrumented method description | ||
| * @param arguments | ||
| * arguments provided for method | ||
| * @param exception | ||
| * exception thrown in method exit (not caught) | ||
| * @param ed | ||
| * {@link EntryDefinition} passed along the method (from before method) | ||
| * @param startTime | ||
| * startTime passed along the method | ||
| */ | ||
|
|
||
| @Advice.OnMethodExit(onThrowable = Throwable.class) | ||
| public static void after(@Advice.This Consumer producer, // | ||
| @Advice.Origin Method method, // | ||
| @Advice.Thrown Throwable exception, @Advice.Local("ed") EntryDefinition ed, // | ||
| @Advice.Local("startTime") long startTime) { | ||
| boolean doFinally = true; | ||
| try { | ||
| if (ed == null) { // ed expected to be null if not created by entry, that's for duplicates | ||
| if (logging) { | ||
| logger.info("EntryDefinition not exist, entry might be filtered out as duplicate or ran on test"); | ||
| } | ||
| doFinally = false; | ||
| return; | ||
| } | ||
| if (logging) { | ||
| logger.info(format("Exiting: {0} {1}", KafkaConsumerClientAdvice.class.getName(), "after")); | ||
| } | ||
| fillDefaultValuesAfter(ed, startTime, exception, logging ? logger : null); | ||
| } catch (Throwable t) { | ||
| handleAdviceException(t, ADVICE_NAME, logging ? logger : null); | ||
| } finally { | ||
| if (doFinally) { | ||
| doFinally(); | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| /** | ||
| * Type matcher should find the class intended for instrumentation See (@ElementMatcher) for available matches. | ||
| */ | ||
|
|
||
| @Override | ||
| public ElementMatcher<TypeDescription> getTypeMatcher() { | ||
| return hasSuperType(named(INTERCEPTING_CLASS[0])).and(not(isInterface())); | ||
| } | ||
|
|
||
| @Override | ||
| public AgentBuilder.Transformer getAdvice() { | ||
| return advice; | ||
| } | ||
|
|
||
| @Override | ||
| protected AgentBuilder.Listener getListener() { | ||
| return new TransformationLoggingListener(logger); | ||
| } | ||
|
|
||
| @Override | ||
| public void install(Instrumentation instrumentation) { | ||
| logger = Logger.tag(ADVICE_NAME); | ||
| if (load) { | ||
| getTransform().with(getListener()).installOn(instrumentation); | ||
| } else { | ||
| logger.info("Advice {0} not enabled", getName()); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return ADVICE_NAME; | ||
| } | ||
|
|
||
| } |
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
3 changes: 2 additions & 1 deletion
3
remora-kafka/src/main/resources/META-INF/services/com.jkoolcloud.remora.advices.RemoraAdvice
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 |
|---|---|---|
| @@ -1,2 +1,3 @@ | ||
| com.jkoolcloud.remora.advices.KafkaConsumerAdvice | ||
| com.jkoolcloud.remora.advices.KafkaProducerAdvice | ||
| com.jkoolcloud.remora.advices.KafkaProducerAdvice | ||
| com.jkoolcloud.remora.advices.KafkaConsumerClientAdvice |
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