From 8262112ab2652af6fe28e2d87a49b2d64b53e645 Mon Sep 17 00:00:00 2001 From: aalinyu Date: Fri, 20 May 2022 16:10:22 +0800 Subject: [PATCH] [#noissue]Multiple methods are supported to use consumer exclusively --- .../pinpoint/plugin/kafka/KafkaConfig.java | 22 ++++- .../pinpoint/plugin/kafka/KafkaPlugin.java | 97 +++++++++++++------ .../plugin/kafka/KafkaConfigTest.java | 16 +-- 3 files changed, 95 insertions(+), 40 deletions(-) diff --git a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfig.java b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfig.java index ab95449e562e..36b7b795a025 100644 --- a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfig.java +++ b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfig.java @@ -17,6 +17,10 @@ package com.navercorp.pinpoint.plugin.kafka; import com.navercorp.pinpoint.bootstrap.config.ProfilerConfig; +import com.navercorp.pinpoint.common.util.StringUtils; + +import java.util.Collections; +import java.util.List; public class KafkaConfig { @@ -38,7 +42,7 @@ public class KafkaConfig { private final boolean springConsumerEnable; private final boolean headerEnable; private final boolean headerRecorded; - private final String kafkaEntryPoint; + private final List kafkaEntryPoints; public KafkaConfig(ProfilerConfig config) { this.producerEnable = config.readBoolean(PRODUCER_ENABLE, false); @@ -46,7 +50,7 @@ public KafkaConfig(ProfilerConfig config) { this.springConsumerEnable = config.readBoolean(SPRING_CONSUMER_ENABLE, false); this.headerEnable = config.readBoolean(HEADER_ENABLE, true); this.headerRecorded = config.readBoolean(HEADER_RECORD, true); - this.kafkaEntryPoint = config.readString(CONSUMER_ENTRY_POINT, ""); + this.kafkaEntryPoints = split(config.readString(CONSUMER_ENTRY_POINT, "")); } public boolean isProducerEnable() { @@ -69,8 +73,16 @@ public boolean isHeaderRecorded() { return headerRecorded; } - public String getKafkaEntryPoint() { - return kafkaEntryPoint; + public List getKafkaEntryPoints() { + return kafkaEntryPoints; + } + + private List split(String values) { + if (StringUtils.isEmpty(values)) { + return Collections.emptyList(); + } + + return StringUtils.tokenizeToStringList(values, ","); } @Override @@ -80,7 +92,7 @@ public String toString() { ", consumerEnable=" + consumerEnable + ", springConsumerEnable=" + springConsumerEnable + ", headerEnable=" + headerEnable + - ", kafkaEntryPoint='" + kafkaEntryPoint + '\'' + + ", kafkaEntryPoints='" + kafkaEntryPoints.toString() + '\'' + '}'; } } diff --git a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java index 10c345ea33c1..361b8a96960c 100644 --- a/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java +++ b/plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java @@ -52,6 +52,10 @@ import java.security.ProtectionDomain; import java.util.List; +import java.util.Set; +import java.util.Map; +import java.util.HashMap; +import java.util.HashSet; import static com.navercorp.pinpoint.common.util.VarArgs.va; @@ -101,12 +105,34 @@ public void setup(ProfilerPluginSetupContext context) { transformTemplate.transform("org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter$IntegrationBatchMessageListener", BatchMessagingMessageListenerAdapterTransform.class); } - if (StringUtils.hasText(config.getKafkaEntryPoint())) { - transformEntryPoint(config.getKafkaEntryPoint()); + if (config.getKafkaEntryPoints() != null && config.getKafkaEntryPoints().size() > 0) { + // merge + final Map> methods = parseEntryPointsMethods(config.getKafkaEntryPoints()); + if (logger.isInfoEnabled()) { + logger.info("Kafka EntryPoints entry points={}", methods); + } + + // add kafka consumer include methods + for (Map.Entry> entry : methods.entrySet()) { + try { + addkafkaConsumerEntryPointClass(entry.getKey(), entry.getValue()); + if (logger.isDebugEnabled()) { + logger.debug("Add kafka consumer include class interceptor {}.{}", entry.getKey(), entry.getValue()); + } + } catch (Exception e) { + logger.warn("Failed to add kafka consumer include class(" + entry.getKey() + "." + entry.getValue() + ").", e); + } + } } } } + private void addkafkaConsumerEntryPointClass(final String className, final Set methodNames) { + final String[] methodNameArray = methodNames.toArray(new String[0]); + transformTemplate.transform(className, EntryPointTransform.class, new Object[]{methodNameArray}, new Class[]{String[].class}); + } + + public static class KafkaProducerTransform implements TransformCallback { @Override @@ -286,7 +312,7 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, } private boolean enableConsumerTransform(KafkaConfig config) { - if (config.isConsumerEnable() && StringUtils.hasText(config.getKafkaEntryPoint())) { + if (config.isConsumerEnable()) { return true; } @@ -298,22 +324,20 @@ public void setTransformTemplate(TransformTemplate transformTemplate) { this.transformTemplate = transformTemplate; } - public void transformEntryPoint(String entryPoint) { - final String clazzName = toClassName(entryPoint); - - transformTemplate.transform(clazzName, EntryPointTransform.class); - } public static class EntryPointTransform implements TransformCallback { private final PLogger logger = PLoggerFactory.getLogger(this.getClass()); + private final String[] methodNames; + + public EntryPointTransform(String[] methodNames) { + this.methodNames = methodNames; + } @Override public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, String className, Class classBeingRedefined, ProtectionDomain protectionDomain, byte[] classfileBuffer) throws InstrumentException { final InstrumentClass target = instrumentor.getInstrumentClass(classLoader, className, classfileBuffer); - final KafkaConfig config = new KafkaConfig(instrumentor.getProfilerConfig()); - final String methodName = toMethodName(config.getKafkaEntryPoint()); - for (InstrumentMethod method : target.getDeclaredMethods(MethodFilters.name(methodName))) { + for (InstrumentMethod method : target.getDeclaredMethods(MethodFilters.name(methodNames))) { try { String[] parameterTypes = method.getParameterTypes(); if (parameterTypes == null) { @@ -340,25 +364,9 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, return target.toBytecode(); } - private String toMethodName(String fullQualifiedMethodName) { - final int methodBeginPosition = fullQualifiedMethodName.lastIndexOf('.'); - if (methodBeginPosition <= 0 || methodBeginPosition + 1 >= fullQualifiedMethodName.length()) { - throw new IllegalArgumentException("invalid full qualified method name(" + fullQualifiedMethodName + "). not found method"); - } - - return fullQualifiedMethodName.substring(methodBeginPosition + 1); - } } - private String toClassName(String fullQualifiedMethodName) { - final int classEndPosition = fullQualifiedMethodName.lastIndexOf('.'); - if (classEndPosition <= 0) { - throw new IllegalArgumentException("invalid full qualified method name(" + fullQualifiedMethodName + "). not found method"); - } - - return fullQualifiedMethodName.substring(0, classEndPosition); - } public static class KafkaSelectorTransform implements TransformCallback { @@ -455,4 +463,39 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader, } + + private Map> parseEntryPointsMethods(List fullyQualifiedMethodNames) { + Map> userMethods = new HashMap>(); + for (String fullyQualifiedMethodName : fullyQualifiedMethodNames) { + try { + final String className = toClassName(fullyQualifiedMethodName); + final String methodName = toMethodName(fullyQualifiedMethodName); + Set methodNames = userMethods.computeIfAbsent(className, k -> new HashSet()); + methodNames.add(methodName); + } catch (Exception e) { + logger.warn("Failed to parse user method(" + fullyQualifiedMethodName + ").", e); + } + } + return userMethods; + } + + + private String toClassName(String fullQualifiedMethodName) { + final int classEndPosition = fullQualifiedMethodName.lastIndexOf('.'); + if (classEndPosition <= 0) { + throw new IllegalArgumentException("invalid full qualified method name(" + fullQualifiedMethodName + "). not found method"); + } + + return fullQualifiedMethodName.substring(0, classEndPosition); + } + + private String toMethodName(String fullQualifiedMethodName) { + final int methodBeginPosition = fullQualifiedMethodName.lastIndexOf('.'); + if (methodBeginPosition <= 0 || methodBeginPosition + 1 >= fullQualifiedMethodName.length()) { + throw new IllegalArgumentException("invalid full qualified method name(" + fullQualifiedMethodName + "). not found method"); + } + + return fullQualifiedMethodName.substring(methodBeginPosition + 1); + } + } diff --git a/plugins/kafka/src/test/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfigTest.java b/plugins/kafka/src/test/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfigTest.java index c5bbc76fa6e8..eb9c9a8f039c 100644 --- a/plugins/kafka/src/test/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfigTest.java +++ b/plugins/kafka/src/test/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfigTest.java @@ -30,11 +30,11 @@ public class KafkaConfigTest { @Test public void configTest1() throws Exception { - KafkaConfig config = createConfig("true", "true", "entryPoint"); + KafkaConfig config = createConfig("true", "true", "entryPoint1,entryPoint2"); Assert.assertTrue(config.isProducerEnable()); Assert.assertTrue(config.isConsumerEnable()); - Assert.assertEquals("entryPoint", config.getKafkaEntryPoint()); + Assert.assertEquals("[entryPoint1, entryPoint2]", config.getKafkaEntryPoints().toString()); } @Test @@ -43,7 +43,7 @@ public void configTest2() throws Exception { Assert.assertTrue(config.isProducerEnable()); Assert.assertFalse(config.isConsumerEnable()); - Assert.assertEquals("", config.getKafkaEntryPoint()); + Assert.assertEquals("[]", config.getKafkaEntryPoints().toString()); } @Test @@ -52,7 +52,7 @@ public void configTest3() throws Exception { Assert.assertFalse(config.isProducerEnable()); Assert.assertTrue(config.isConsumerEnable()); - Assert.assertEquals("", config.getKafkaEntryPoint()); + Assert.assertEquals("[]", config.getKafkaEntryPoints().toString()); } @Test @@ -61,7 +61,7 @@ public void configTest4() throws Exception { Assert.assertFalse(config.isProducerEnable()); Assert.assertFalse(config.isConsumerEnable()); - Assert.assertEquals("", config.getKafkaEntryPoint()); + Assert.assertEquals("[]", config.getKafkaEntryPoints().toString()); } @Test @@ -71,7 +71,7 @@ public void configTest5() throws Exception { Assert.assertTrue(config.isProducerEnable()); Assert.assertFalse(config.isConsumerEnable()); Assert.assertTrue(config.isHeaderEnable()); - Assert.assertEquals("entryPoint1", config.getKafkaEntryPoint()); + Assert.assertEquals("[entryPoint1]", config.getKafkaEntryPoints().toString()); } @Test @@ -81,7 +81,7 @@ public void configTest6() throws Exception { Assert.assertTrue(config.isProducerEnable()); Assert.assertFalse(config.isConsumerEnable()); Assert.assertFalse(config.isHeaderEnable()); - Assert.assertEquals("entryPoint2", config.getKafkaEntryPoint()); + Assert.assertEquals("[entryPoint2]", config.getKafkaEntryPoints().toString()); } @Test @@ -91,7 +91,7 @@ public void configTest7() throws Exception { Assert.assertFalse(config.isProducerEnable()); Assert.assertTrue(config.isConsumerEnable()); Assert.assertFalse(config.isHeaderEnable()); - Assert.assertEquals("entryPoint3", config.getKafkaEntryPoint()); + Assert.assertEquals("[entryPoint3]", config.getKafkaEntryPoints().toString()); } private KafkaConfig createConfig(String producerEnable, String consumerEnable) {