Skip to content

[#noissue]Multiple methods are supported to use consumer exclusively #8876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
aalinyu wants to merge 1 commit into
base: master
Choose a base branch
from
+95 −40
Open

[#noissue]Multiple methods are supported to use consumer exclusively #8876

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 17 additions & 5 deletions plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {

Expand All @@ -38,15 +42,15 @@ public class KafkaConfig {
private final boolean springConsumerEnable;
private final boolean headerEnable;
private final boolean headerRecorded;
private final String kafkaEntryPoint;
private final List<String> kafkaEntryPoints;

public KafkaConfig(ProfilerConfig config) {
this.producerEnable = config.readBoolean(PRODUCER_ENABLE, false);
this.consumerEnable = config.readBoolean(CONSUMER_ENABLE, false);
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() {
Expand All @@ -69,8 +73,16 @@ public boolean isHeaderRecorded() {
return headerRecorded;
}

public String getKafkaEntryPoint() {
return kafkaEntryPoint;
public List<String> getKafkaEntryPoints() {
return kafkaEntryPoints;
}

private List<String> split(String values) {
if (StringUtils.isEmpty(values)) {
return Collections.emptyList();
}

return StringUtils.tokenizeToStringList(values, ",");
}

@Override
Expand All @@ -80,7 +92,7 @@ public String toString() {
", consumerEnable=" + consumerEnable +
", springConsumerEnable=" + springConsumerEnable +
", headerEnable=" + headerEnable +
", kafkaEntryPoint='" + kafkaEntryPoint + '\'' +
", kafkaEntryPoints='" + kafkaEntryPoints.toString() + '\'' +
'}';
}
}
97 changes: 70 additions & 27 deletions plugins/kafka/src/main/java/com/navercorp/pinpoint/plugin/kafka/KafkaPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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<String, Set<String>> methods = parseEntryPointsMethods(config.getKafkaEntryPoints());
if (logger.isInfoEnabled()) {
logger.info("Kafka EntryPoints entry points={}", methods);
}

// add kafka consumer include methods
for (Map.Entry<String, Set<String>> 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<String> 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
Expand Down Expand Up @@ -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;
}

Expand All @@ -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) {
Expand All @@ -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 {

Expand Down Expand Up @@ -455,4 +463,39 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader classLoader,

}


private Map<String, Set<String>> parseEntryPointsMethods(List<String> fullyQualifiedMethodNames) {
Map<String, Set<String>> userMethods = new HashMap<String, Set<String>>();
for (String fullyQualifiedMethodName : fullyQualifiedMethodNames) {
try {
final String className = toClassName(fullyQualifiedMethodName);
final String methodName = toMethodName(fullyQualifiedMethodName);
Set<String> methodNames = userMethods.computeIfAbsent(className, k -> new HashSet<String>());
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);
}

}
16 changes: 8 additions & 8 deletions plugins/kafka/src/test/java/com/navercorp/pinpoint/plugin/kafka/KafkaConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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) {
Expand Down