kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

add RocketMQ 5.x plugin (#536)

This commit is contained in:
weihubeats 2023-06-07 22:05:08 +08:00 committed by GitHub
parent 5002da5a13
commit 6621f82739
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
31 changed files with 1174 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.md
View File

@ -8,6 +8,7 @@ Release Notes.
* Support Jdk17 ZGC metric collect
* Support Jetty 11.x plugin
* Fix the scenario of using the HBase plugin with spring-data-hadoop.
* Add RocketMQ 5.x plugin
#### Documentation

1
apm-sniffer/apm-sdk-plugin/pom.xml
View File

@ -131,6 +131,7 @@
<module>jersey-3.x-plugin</module>
<module>grizzly-2.3.x-4.x-plugin</module>
<module>grizzly-2.3.x-4.x-work-threadpool-plugin</module>
<module>rocketMQ-5.x-plugin</module>
</modules>
<packaging>pom</packaging>

29
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/AbstractRocketMQInstrumentation.java Normal file
View File

@ -0,0 +1,29 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v4.define;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
public abstract class AbstractRocketMQInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
@Override
protected String[] witnessClasses() {
return new String[] {Constants.WITNESS_ROCKETMQ_4X_CLASS};
}
}

24
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/Constants.java Normal file
View File

@ -0,0 +1,24 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v4.define;
public class Constants {
public static final String WITNESS_ROCKETMQ_4X_CLASS = "org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader";
}

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java
View File

@ -22,13 +22,12 @@ import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
public class ConsumeMessageConcurrentlyInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public class ConsumeMessageConcurrentlyInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently";
private static final String CONSUMER_MESSAGE_METHOD = "consumeMessage";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageConcurrentlyConsumeInterceptor";

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java
View File

@ -22,13 +22,12 @@ import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch;
public class ConsumeMessageOrderlyInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public class ConsumeMessageOrderlyInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly";
private static final String ENHANCE_METHOD = "consumeMessage";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v4.MessageOrderlyConsumeInterceptor";

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/DefaultMQPushConsumerInstrumentation.java
View File

@ -22,13 +22,12 @@ import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class DefaultMQPushConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public class DefaultMQPushConsumerInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.DefaultMQPushConsumer";
private static final String REGISTER_MESSAGE_LISTENER_METHOD_NAME = "registerMessageListener";

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java
View File

@ -22,14 +22,13 @@ import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class MQClientAPIImplInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public class MQClientAPIImplInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.impl.MQClientAPIImpl";
private static final String SEND_MESSAGE_METHOD_NAME = "sendMessage";

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v4/define/SendCallbackInstrumentation.java
View File

@ -22,14 +22,13 @@ import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch;
public class SendCallbackInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
public class SendCallbackInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.producer.SendCallback";
private static final String ON_SUCCESS_ENHANCE_METHOD = "onSuccess";

53
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/pom.xml Normal file
View File

@ -0,0 +1,53 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You under the Apache License, Version 2.0
~ (the "License"); you may not use this file except in compliance with
~ the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
~
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>apm-sdk-plugin</artifactId>
<groupId>org.apache.skywalking</groupId>
<version>8.17.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>apm-rocketMQ-5.x-plugin</artifactId>
<name>rocketMQ-5.x-plugin</name>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<rocketmq-client.version>5.1.1</rocketmq-client.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>${rocketmq-client.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<artifactId>maven-deploy-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>

91
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/AbstractMessageConsumeInterceptor.java Normal file
View File

@ -0,0 +1,91 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import java.util.List;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.rocketMQ.v5.define.ConsumerEnhanceInfos;
/**
* {@link AbstractMessageConsumeInterceptor} create entry span when the <code>consumeMessage</code> in the {@link
* org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently} and {@link
* org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly} class.
*/
public abstract class AbstractMessageConsumeInterceptor implements InstanceMethodsAroundInterceptor {
public static final String CONSUMER_OPERATION_NAME_PREFIX = "RocketMQ/";
@Override
public final void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
List<MessageExt> msgs = (List<MessageExt>) allArguments[0];
ContextCarrier contextCarrier = getContextCarrierFromMessage(msgs.get(0));
AbstractSpan span = ContextManager.createEntrySpan(CONSUMER_OPERATION_NAME_PREFIX + msgs.get(0)
.getTopic() + "/Consumer", contextCarrier);
Tags.MQ_TOPIC.set(span, msgs.get(0).getTopic());
if (msgs.get(0).getStoreHost() != null) {
String brokerAddress = msgs.get(0).getStoreHost().toString();
brokerAddress = StringUtils.removeStart(brokerAddress, "/");
Tags.MQ_BROKER.set(span, brokerAddress);
}
span.setComponent(ComponentsDefine.ROCKET_MQ_CONSUMER);
SpanLayer.asMQ(span);
for (int i = 1; i < msgs.size(); i++) {
ContextManager.extract(getContextCarrierFromMessage(msgs.get(i)));
}
Object skyWalkingDynamicField = objInst.getSkyWalkingDynamicField();
if (skyWalkingDynamicField != null) {
ConsumerEnhanceInfos consumerEnhanceInfos = (ConsumerEnhanceInfos) skyWalkingDynamicField;
span.setPeer(consumerEnhanceInfos.getNamesrvAddr());
}
}
@Override
public final void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
ContextManager.activeSpan().log(t);
}
private ContextCarrier getContextCarrierFromMessage(MessageExt message) {
ContextCarrier contextCarrier = new ContextCarrier();
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
next = next.next();
next.setHeadValue(message.getUserProperty(next.getHeadKey()));
}
return contextCarrier;
}
}

49
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/MessageConcurrentlyConsumeInterceptor.java Normal file
View File

@ -0,0 +1,49 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
/**
* {@link MessageConcurrentlyConsumeInterceptor} set the process status after the {@link
* org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently#consumeMessage(java.util.List,
* org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext)} method execute.
*/
public class MessageConcurrentlyConsumeInterceptor extends AbstractMessageConsumeInterceptor {
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
ConsumeConcurrentlyStatus status = (ConsumeConcurrentlyStatus) ret;
if (status == ConsumeConcurrentlyStatus.RECONSUME_LATER) {
AbstractSpan activeSpan = ContextManager.activeSpan();
activeSpan.errorOccurred();
Tags.MQ_STATUS.set(activeSpan, status.name());
}
ContextManager.stopSpan();
return ret;
}
}

50
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/MessageOrderlyConsumeInterceptor.java Normal file
View File

@ -0,0 +1,50 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
/**
* {@link MessageOrderlyConsumeInterceptor} set the process status after the {@link
* org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly#consumeMessage(java.util.List,
* org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext)} method execute.
*/
public class MessageOrderlyConsumeInterceptor extends AbstractMessageConsumeInterceptor {
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
ConsumeOrderlyStatus status = (ConsumeOrderlyStatus) ret;
if (status == ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT) {
AbstractSpan activeSpan = ContextManager.activeSpan();
activeSpan.errorOccurred();
Tags.MQ_STATUS.set(activeSpan, status.name());
}
ContextManager.stopSpan();
return ret;
}
}

114
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/MessageSendInterceptor.java Normal file
View File

@ -0,0 +1,114 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.rocketMQ.v5.define.SendCallBackEnhanceInfo;
import org.apache.skywalking.apm.util.StringUtil;
import static org.apache.rocketmq.common.message.MessageDecoder.NAME_VALUE_SEPARATOR;
import static org.apache.rocketmq.common.message.MessageDecoder.PROPERTY_SEPARATOR;
/**
* {@link MessageSendInterceptor} create exit span when the method {@link org.apache.rocketmq.client.impl.MQClientAPIImpl#sendMessage(String,
* String, Message, org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader, long,
* org.apache.rocketmq.client.impl.CommunicationMode, org.apache.rocketmq.client.producer.SendCallback,
* org.apache.rocketmq.client.impl.producer.TopicPublishInfo, org.apache.rocketmq.client.impl.factory.MQClientInstance,
* int, org.apache.rocketmq.client.hook.SendMessageContext, org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl)}
* execute.
*/
public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor {
public static final String ASYNC_SEND_OPERATION_NAME_PREFIX = "RocketMQ/";
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
Message message = (Message) allArguments[2];
ContextCarrier contextCarrier = new ContextCarrier();
String namingServiceAddress = String.valueOf(objInst.getSkyWalkingDynamicField());
AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress);
span.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
Tags.MQ_BROKER.set(span, (String) allArguments[0]);
Tags.MQ_TOPIC.set(span, message.getTopic());
String keys = message.getKeys();
if (StringUtil.isNotBlank(keys)) {
span.tag(Tags.ofKey("mq.message.keys"), keys);
}
String tags = message.getTags();
if (StringUtil.isNotBlank(tags)) {
span.tag(Tags.ofKey("mq.message.tags"), tags);
}
contextCarrier.extensionInjector().injectSendingTimestamp();
SpanLayer.asMQ(span);
SendMessageRequestHeader requestHeader = (SendMessageRequestHeader) allArguments[3];
StringBuilder properties = new StringBuilder(requestHeader.getProperties());
CarrierItem next = contextCarrier.items();
while (next.hasNext()) {
next = next.next();
if (!StringUtil.isEmpty(next.getHeadValue())) {
if (properties.length() > 0 && properties.charAt(properties.length() - 1) != PROPERTY_SEPARATOR) {
// adapt for RocketMQ 4.9.x or later
properties.append(PROPERTY_SEPARATOR);
}
properties.append(next.getHeadKey());
properties.append(NAME_VALUE_SEPARATOR);
properties.append(next.getHeadValue());
}
}
requestHeader.setProperties(properties.toString());
if (allArguments[6] != null) {
((EnhancedInstance) allArguments[6]).setSkyWalkingDynamicField(new SendCallBackEnhanceInfo(message.getTopic(), ContextManager
.capture()));
}
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
ContextManager.stopSpan();
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
ContextManager.activeSpan().log(t);
}
private String buildOperationName(String topicName) {
return ASYNC_SEND_OPERATION_NAME_PREFIX + topicName + "/Producer";
}
}

70
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnExceptionInterceptor.java Normal file
View File

@ -0,0 +1,70 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.rocketMQ.v5.define.SendCallBackEnhanceInfo;
/**
* {@link OnExceptionInterceptor} create local span when the method {@link org.apache.rocketmq.client.producer.SendCallback#onException(Throwable)}
* execute.
*/
public class OnExceptionInterceptor implements InstanceMethodsAroundInterceptor {
public static final String CALLBACK_OPERATION_NAME_PREFIX = "RocketMQ/";
private static final String DEFAULT_TOPIC = "no_topic";
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
SendCallBackEnhanceInfo enhanceInfo = (SendCallBackEnhanceInfo) objInst.getSkyWalkingDynamicField();
String topicId = DEFAULT_TOPIC;
// The SendCallBackEnhanceInfo could be null when there is an internal exception in the client API,
// such as MQClientException("no route info of this topic")
if (enhanceInfo != null) {
topicId = enhanceInfo.getTopicId();
}
AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + topicId + "/Producer/Callback");
activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
activeSpan.log((Throwable) allArguments[0]);
if (enhanceInfo != null && enhanceInfo.getContextSnapshot() != null) {
ContextManager.continued(enhanceInfo.getContextSnapshot());
}
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
ContextManager.stopSpan();
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
ContextManager.activeSpan().log(t);
}
}

68
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/OnSuccessInterceptor.java Normal file
View File

@ -0,0 +1,68 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.skywalking.apm.agent.core.context.ContextManager;
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
import org.apache.skywalking.apm.plugin.rocketMQ.v5.define.SendCallBackEnhanceInfo;
/**
* {@link OnSuccessInterceptor} create local span when the method {@link org.apache.rocketmq.client.producer.SendCallback#onSuccess(SendResult)}
* execute.
*/
public class OnSuccessInterceptor implements InstanceMethodsAroundInterceptor {
public static final String CALLBACK_OPERATION_NAME_PREFIX = "RocketMQ/";
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
SendCallBackEnhanceInfo enhanceInfo = (SendCallBackEnhanceInfo) objInst.getSkyWalkingDynamicField();
AbstractSpan activeSpan = ContextManager.createLocalSpan(CALLBACK_OPERATION_NAME_PREFIX + enhanceInfo.getTopicId() + "/Producer/Callback");
activeSpan.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
SendStatus sendStatus = ((SendResult) allArguments[0]).getSendStatus();
if (sendStatus != SendStatus.SEND_OK) {
activeSpan.errorOccurred();
Tags.MQ_STATUS.set(activeSpan, sendStatus.name());
}
ContextManager.continued(enhanceInfo.getContextSnapshot());
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
ContextManager.stopSpan();
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
ContextManager.activeSpan().log(t);
}
}

51
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/RegisterMessageListenerInterceptor.java Normal file
View File

@ -0,0 +1,51 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
import org.apache.skywalking.apm.plugin.rocketMQ.v5.define.ConsumerEnhanceInfos;
public class RegisterMessageListenerInterceptor implements InstanceMethodsAroundInterceptor {
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
DefaultMQPushConsumer defaultMQPushConsumer = (DefaultMQPushConsumer) objInst;
String namesrvAddr = defaultMQPushConsumer.getNamesrvAddr();
ConsumerEnhanceInfos consumerEnhanceInfos = new ConsumerEnhanceInfos(namesrvAddr);
if (allArguments[0] instanceof EnhancedInstance) {
EnhancedInstance enhancedMessageListener = (EnhancedInstance) allArguments[0];
enhancedMessageListener.setSkyWalkingDynamicField(consumerEnhanceInfos);
}
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
}
}

45
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/UpdateNameServerInterceptor.java Normal file
View File

@ -0,0 +1,45 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5;
import java.lang.reflect.Method;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
public class UpdateNameServerInterceptor implements InstanceMethodsAroundInterceptor {
@Override
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
MethodInterceptResult result) throws Throwable {
objInst.setSkyWalkingDynamicField(allArguments[0]);
}
@Override
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
Object ret) throws Throwable {
return ret;
}
@Override
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,
Class<?>[] argumentsTypes, Throwable t) {
}
}

29
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/AbstractRocketMQInstrumentation.java Normal file
View File

@ -0,0 +1,29 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.ClassInstanceMethodsEnhancePluginDefine;
public abstract class AbstractRocketMQInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
@Override
protected String[] witnessClasses() {
return new String[] {Constants.WITNESS_ROCKETMQ_5X_CLASS};
}
}

24
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/Constants.java Normal file
View File

@ -0,0 +1,24 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
public class Constants {
public static final String WITNESS_ROCKETMQ_5X_CLASS = "org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader";
}

66
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/ConsumeMessageConcurrentlyInstrumentation.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
public class ConsumeMessageConcurrentlyInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently";
private static final String CONSUMER_MESSAGE_METHOD = "consumeMessage";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v5.MessageConcurrentlyConsumeInterceptor";
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(CONSUMER_MESSAGE_METHOD);
}
@Override
public String getMethodsInterceptor() {
return INTERCEPTOR_CLASS;
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
protected ClassMatch enhanceClass() {
return HierarchyMatch.byHierarchyMatch(new String[] {ENHANCE_CLASS});
}
}

66
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/ConsumeMessageOrderlyInstrumentation.java Normal file
View File

@ -0,0 +1,66 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch;
public class ConsumeMessageOrderlyInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly";
private static final String ENHANCE_METHOD = "consumeMessage";
private static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v5.MessageOrderlyConsumeInterceptor";
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(ENHANCE_METHOD);
}
@Override
public String getMethodsInterceptor() {
return INTERCEPTOR_CLASS;
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
protected ClassMatch enhanceClass() {
return byHierarchyMatch(new String[] {ENHANCE_CLASS});
}
}

32
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/ConsumerEnhanceInfos.java Normal file
View File

@ -0,0 +1,32 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
public class ConsumerEnhanceInfos {
private String namesrvAddr;
public ConsumerEnhanceInfos(String namesrvAddr) {
this.namesrvAddr = namesrvAddr;
}
public String getNamesrvAddr() {
return namesrvAddr;
}
}

68
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/DefaultMQPushConsumerInstrumentation.java Normal file
View File

@ -0,0 +1,68 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class DefaultMQPushConsumerInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.consumer.DefaultMQPushConsumer";
private static final String REGISTER_MESSAGE_LISTENER_METHOD_NAME = "registerMessageListener";
public static final String REGISTER_MESSAGE_LISTENER_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v5.RegisterMessageListenerInterceptor";
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(REGISTER_MESSAGE_LISTENER_METHOD_NAME);
}
@Override
public String getMethodsInterceptor() {
return REGISTER_MESSAGE_LISTENER_INTERCEPT_CLASS;
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
protected ClassMatch enhanceClass() {
return byName(ENHANCE_CLASS);
}
}

86
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/MQClientAPIImplInstrumentation.java Normal file
View File

@ -0,0 +1,86 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
import static org.apache.skywalking.apm.agent.core.plugin.match.NameMatch.byName;
public class MQClientAPIImplInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.impl.MQClientAPIImpl";
private static final String SEND_MESSAGE_METHOD_NAME = "sendMessage";
private static final String ASYNC_METHOD_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v5.MessageSendInterceptor";
public static final String UPDATE_NAME_SERVER_INTERCEPT_CLASS = "org.apache.skywalking.apm.plugin.rocketMQ.v5.UpdateNameServerInterceptor";
public static final String UPDATE_NAME_SERVER_METHOD_NAME = "updateNameServerAddressList";
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(SEND_MESSAGE_METHOD_NAME).and(takesArguments(12));
}
@Override
public String getMethodsInterceptor() {
return ASYNC_METHOD_INTERCEPTOR;
}
@Override
public boolean isOverrideArgs() {
return false;
}
},
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(UPDATE_NAME_SERVER_METHOD_NAME);
}
@Override
public String getMethodsInterceptor() {
return UPDATE_NAME_SERVER_INTERCEPT_CLASS;
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
protected ClassMatch enhanceClass() {
return byName(ENHANCE_CLASS);
}
}

42
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/SendCallBackEnhanceInfo.java Normal file
View File

@ -0,0 +1,42 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import org.apache.skywalking.apm.agent.core.context.ContextSnapshot;
/**
* {@link SendCallBackEnhanceInfo} saves the topic Id and {@link ContextSnapshot} instance for trace.
*/
public class SendCallBackEnhanceInfo {
private String topicId;
private ContextSnapshot contextSnapshot;
public SendCallBackEnhanceInfo(String topicId, ContextSnapshot contextSnapshot) {
this.topicId = topicId;
this.contextSnapshot = contextSnapshot;
}
public String getTopicId() {
return topicId;
}
public ContextSnapshot getContextSnapshot() {
return contextSnapshot;
}
}

86
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/v5/define/SendCallbackInstrumentation.java Normal file
View File

@ -0,0 +1,86 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.plugin.rocketMQ.v5.define;
import net.bytebuddy.description.method.MethodDescription;
import net.bytebuddy.matcher.ElementMatcher;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.interceptor.InstanceMethodsInterceptPoint;
import org.apache.skywalking.apm.agent.core.plugin.match.ClassMatch;
import static net.bytebuddy.matcher.ElementMatchers.named;
import static org.apache.skywalking.apm.agent.core.plugin.bytebuddy.ArgumentTypeNameMatch.takesArgumentWithType;
import static org.apache.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierarchyMatch;
public class SendCallbackInstrumentation extends AbstractRocketMQInstrumentation {
private static final String ENHANCE_CLASS = "org.apache.rocketmq.client.producer.SendCallback";
private static final String ON_SUCCESS_ENHANCE_METHOD = "onSuccess";
private static final String ON_SUCCESS_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v5.OnSuccessInterceptor";
private static final String ON_EXCEPTION_METHOD = "onException";
private static final String ON_EXCEPTION_INTERCEPTOR = "org.apache.skywalking.apm.plugin.rocketMQ.v5.OnExceptionInterceptor";
@Override
public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
}
@Override
public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
return new InstanceMethodsInterceptPoint[] {
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(ON_SUCCESS_ENHANCE_METHOD).and(takesArgumentWithType(0, "org.apache.rocketmq.client.producer.SendResult"));
}
@Override
public String getMethodsInterceptor() {
return ON_SUCCESS_INTERCEPTOR;
}
@Override
public boolean isOverrideArgs() {
return false;
}
},
new InstanceMethodsInterceptPoint() {
@Override
public ElementMatcher<MethodDescription> getMethodsMatcher() {
return named(ON_EXCEPTION_METHOD).and(takesArgumentWithType(0, "java.lang.Throwable"));
}
@Override
public String getMethodsInterceptor() {
return ON_EXCEPTION_INTERCEPTOR;
}
@Override
public boolean isOverrideArgs() {
return false;
}
}
};
}
@Override
protected ClassMatch enhanceClass() {
return byHierarchyMatch(new String[] {ENHANCE_CLASS});
}
}

21
apm-sniffer/apm-sdk-plugin/rocketMQ-5.x-plugin/src/main/resources/skywalking-plugin.def Normal file
View File

@ -0,0 +1,21 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
rocketMQ-5.x=org.apache.skywalking.apm.plugin.rocketMQ.v5.define.ConsumeMessageConcurrentlyInstrumentation
rocketMQ-5.x=org.apache.skywalking.apm.plugin.rocketMQ.v5.define.ConsumeMessageOrderlyInstrumentation
rocketMQ-5.x=org.apache.skywalking.apm.plugin.rocketMQ.v5.define.MQClientAPIImplInstrumentation
rocketMQ-5.x=org.apache.skywalking.apm.plugin.rocketMQ.v5.define.SendCallbackInstrumentation
rocketMQ-5.x=org.apache.skywalking.apm.plugin.rocketMQ.v5.define.DefaultMQPushConsumerInstrumentation

1
docs/en/setup/service-agent/java-agent/Plugin-list.md
View File

@ -83,6 +83,7 @@
- resteasy-server-4.x
- rocketMQ-3.x
- rocketMQ-4.x
- rocketMQ-5.x
- sentinel-1.x
- servicecomb-2.x
- sharding-sphere-3.x

2
docs/en/setup/service-agent/java-agent/Supported-list.md
View File

@ -69,7 +69,7 @@ metrics based on the tracing data.
* [Apache CXF](https://github.com/apache/cxf) 3.x
* [JSONRPC4J](https://github.com/briandilley/jsonrpc4j) 1.2.0 -> 1.6
* MQ
* [RocketMQ](https://github.com/apache/rocketmq) 4.x
* [RocketMQ](https://github.com/apache/rocketmq) 3.x-> 5.x
* [Kafka](http://kafka.apache.org) 0.11.0.0 -> 3.2.3
* [Spring-Kafka](https://github.com/spring-projects/spring-kafka) Spring Kafka Consumer 1.3.x -> 2.3.x (2.0.x and 2.1.x not tested and not recommended by [the official document](https://spring.io/projects/spring-kafka))
* [ActiveMQ](https://github.com/apache/activemq) 5.10.0 -> 5.15.4

1
test/plugin/scenarios/rocketmq-scenario/support-version.list
View File

@ -16,6 +16,7 @@
# lists your version here (Contains only the last version number of each minor version.)
5.1.0
4.9.4
4.8.0
4.7.1