Skip to content

Commit 738201c

Browse files
xyyz150xyyz150
authored
add rocketmq-client-java-5.x plugin (#598)
1 parent afa2578 commit 738201c

File tree

25 files changed

+1168
-2
lines changed

25 files changed

+1168
-2
lines changed

‎.github/workflows/plugins-test.2.yaml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,7 @@ jobs:
8181
- jersey-2.26.x-2.39.x-scenario
8282
- websphere-liberty-23.x-scenario
8383
- nacos-client-2.x-scenario
84+
- rocketmq-5-grpc-scenario
8485
steps:
8586
- uses: actions/checkout@v2
8687
with:

‎CHANGES.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -152,7 +152,8 @@ Callable {
152152
* Merge two instrumentation classes to avoid duplicate enhancements in MySQL plugins.
153153
* Support asynchronous invocation in jetty client 9.0 and 9.x plugin
154154
* Add nacos-client 2.x plugin
155-
* Staticize the tags for preventing synchronization in JDK 8
155+
* Staticize the tags for preventing synchronization in JDK 8
156+
* Add RocketMQ-Client-Java 5.x plugin
156157
* Fix NullPointerException in lettuce-5.x-plugin.
157158

158159
#### Documentation

‎apm-sniffer/apm-sdk-plugin/pom.xml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,7 @@
134134
<module>rocketMQ-5.x-plugin</module>
135135
<module>websphere-liberty-23.x-plugin</module>
136136
<module>aerospike-plugin</module>
137+
<module>rocketMQ-client-java-5.x-plugin</module>
137138
</modules>
138139
<packaging>pom</packaging>
139140

‎apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/pom.xml‎

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Licensed to the Apache Software Foundation (ASF) under one or more
4+
~ contributor license agreements. See the NOTICE file distributed with
5+
~ this work for additional information regarding copyright ownership.
6+
~ The ASF licenses this file to You under the Apache License, Version 2.0
7+
~ (the "License"); you may not use this file except in compliance with
8+
~ the License. You may obtain a copy of the License at
9+
~
10+
~ http://www.apache.org/licenses/LICENSE-2.0
11+
~
12+
~ Unless required by applicable law or agreed to in writing, software
13+
~ distributed under the License is distributed on an "AS IS" BASIS,
14+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
~ See the License for the specific language governing permissions and
16+
~ limitations under the License.
17+
~
18+
-->
19+
20+
<project xmlns="http://maven.apache.org/POM/4.0.0"
21+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
22+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
23+
<parent>
24+
<artifactId>apm-sdk-plugin</artifactId>
25+
<groupId>org.apache.skywalking</groupId>
26+
<version>9.0.0-SNAPSHOT</version>
27+
</parent>
28+
<modelVersion>4.0.0</modelVersion>
29+
30+
<artifactId>apm-rocketmq-client-java-5.x-plugin</artifactId>
31+
<name>rocketMQ-client-java-5.x-plugin</name>
32+
33+
<properties>
34+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
35+
<rocketmq-client-java.version>5.0.5</rocketmq-client-java.version>
36+
</properties>
37+
38+
<dependencies>
39+
<dependency>
40+
<groupId>org.apache.rocketmq</groupId>
41+
<artifactId>rocketmq-client-java</artifactId>
42+
<version>${rocketmq-client-java.version}</version>
43+
<scope>provided</scope>
44+
</dependency>
45+
</dependencies>
46+
47+
<build>
48+
<plugins>
49+
<plugin>
50+
<artifactId>maven-deploy-plugin</artifactId>
51+
</plugin>
52+
</plugins>
53+
</build>
54+
</project>

‎apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageListenerInterceptor.java‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5;
20+
21+
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
22+
import org.apache.rocketmq.client.apis.message.MessageView;
23+
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
24+
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
25+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
26+
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
27+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
28+
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
29+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
30+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
31+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
32+
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
33+
import org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.ConsumerEnhanceInfos;
34+
35+
import java.lang.reflect.Method;
36+
37+
public class MessageListenerInterceptor implements InstanceMethodsAroundInterceptor {
38+
39+
public static final String CONSUMER_OPERATION_NAME_PREFIX = "RocketMQ/";
40+
41+
@Override
42+
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
43+
MessageView messageView = (MessageView) allArguments[0];
44+
45+
ContextCarrier contextCarrier = getContextCarrierFromMessage(messageView);
46+
47+
AbstractSpan span = ContextManager.createEntrySpan(CONSUMER_OPERATION_NAME_PREFIX + messageView.getTopic()
48+
+ "/Consumer", contextCarrier);
49+
Tags.MQ_TOPIC.set(span, messageView.getTopic());
50+
51+
Object skyWalkingDynamicField = objInst.getSkyWalkingDynamicField();
52+
if (skyWalkingDynamicField != null) {
53+
ConsumerEnhanceInfos consumerEnhanceInfos = (ConsumerEnhanceInfos) skyWalkingDynamicField;
54+
Tags.MQ_BROKER.set(span, consumerEnhanceInfos.getNamesrvAddr());
55+
span.setPeer(consumerEnhanceInfos.getNamesrvAddr());
56+
}
57+
58+
span.setComponent(ComponentsDefine.ROCKET_MQ_CONSUMER);
59+
SpanLayer.asMQ(span);
60+
}
61+
62+
@Override
63+
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
64+
ConsumeResult status = (ConsumeResult) ret;
65+
if (ConsumeResult.FAILURE.equals(status)) {
66+
AbstractSpan activeSpan = ContextManager.activeSpan();
67+
activeSpan.errorOccurred();
68+
Tags.MQ_STATUS.set(activeSpan, status.name());
69+
}
70+
ContextManager.stopSpan();
71+
return ret;
72+
}
73+
74+
@Override
75+
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
76+
ContextManager.activeSpan().log(t);
77+
}
78+
79+
private ContextCarrier getContextCarrierFromMessage(MessageView message) {
80+
ContextCarrier contextCarrier = new ContextCarrier();
81+
82+
CarrierItem next = contextCarrier.items();
83+
while (next.hasNext()) {
84+
next = next.next();
85+
next.setHeadValue(message.getProperties().get(next.getHeadKey()));
86+
}
87+
88+
return contextCarrier;
89+
}
90+
}

‎apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/MessageSendInterceptor.java‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,125 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5;
20+
21+
import org.apache.rocketmq.client.apis.message.Message;
22+
import org.apache.rocketmq.client.apis.message.MessageBuilder;
23+
import org.apache.rocketmq.client.apis.producer.SendReceipt;
24+
import org.apache.rocketmq.client.apis.producer.Transaction;
25+
import org.apache.rocketmq.client.java.impl.ClientImpl;
26+
import org.apache.rocketmq.client.java.message.MessageBuilderImpl;
27+
import org.apache.skywalking.apm.agent.core.context.CarrierItem;
28+
import org.apache.skywalking.apm.agent.core.context.ContextCarrier;
29+
import org.apache.skywalking.apm.agent.core.context.ContextManager;
30+
import org.apache.skywalking.apm.agent.core.context.tag.Tags;
31+
import org.apache.skywalking.apm.agent.core.context.trace.AbstractSpan;
32+
import org.apache.skywalking.apm.agent.core.context.trace.SpanLayer;
33+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
34+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
35+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
36+
import org.apache.skywalking.apm.agent.core.util.CollectionUtil;
37+
import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
38+
import org.apache.skywalking.apm.util.StringUtil;
39+
40+
import java.lang.reflect.Method;
41+
import java.util.Collection;
42+
import java.util.Map;
43+
import java.util.Optional;
44+
import java.util.stream.Collectors;
45+
46+
/**
47+
* {@link MessageSendInterceptor} create exit span when the method {@link org.apache.rocketmq.client.java.impl.producer.ProducerImpl#send(Message)}
48+
* and {@link org.apache.rocketmq.client.java.impl.producer.ProducerImpl#send(Message, Transaction)} execute.
49+
*/
50+
public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor {
51+
52+
public static final String ASYNC_SEND_OPERATION_NAME_PREFIX = "RocketMQ/";
53+
54+
@Override
55+
public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, MethodInterceptResult result) throws Throwable {
56+
Message message = (Message) allArguments[0];
57+
ClientImpl producerImpl = (ClientImpl) objInst;
58+
59+
ContextCarrier contextCarrier = new ContextCarrier();
60+
String namingServiceAddress = producerImpl.getClientConfiguration().getEndpoints();
61+
AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress);
62+
span.setComponent(ComponentsDefine.ROCKET_MQ_PRODUCER);
63+
Tags.MQ_BROKER.set(span, namingServiceAddress);
64+
Tags.MQ_TOPIC.set(span, message.getTopic());
65+
Collection<String> keys = message.getKeys();
66+
if (!CollectionUtil.isEmpty(keys)) {
67+
span.tag(Tags.ofKey("mq.message.keys"), keys.stream().collect(Collectors.joining(",")));
68+
}
69+
Optional<String> tag = message.getTag();
70+
if (tag.isPresent()) {
71+
span.tag(Tags.ofKey("mq.message.tags"), tag.get());
72+
}
73+
74+
contextCarrier.extensionInjector().injectSendingTimestamp();
75+
SpanLayer.asMQ(span);
76+
77+
Map<String, String> properties = message.getProperties();
78+
CarrierItem next = contextCarrier.items();
79+
while (next.hasNext()) {
80+
next = next.next();
81+
if (!StringUtil.isEmpty(next.getHeadValue())) {
82+
properties.put(next.getHeadKey(), next.getHeadValue());
83+
}
84+
}
85+
86+
MessageBuilder messageBuilder = new MessageBuilderImpl();
87+
messageBuilder.setTopic(message.getTopic());
88+
if (message.getTag().isPresent()) {
89+
messageBuilder.setTag(message.getTag().get());
90+
}
91+
messageBuilder.setKeys(message.getKeys().toArray(new String[0]));
92+
if (message.getMessageGroup().isPresent()) {
93+
messageBuilder.setMessageGroup(message.getMessageGroup().get());
94+
}
95+
96+
byte[] body = new byte[message.getBody().limit()];
97+
message.getBody().get(body);
98+
messageBuilder.setBody(body);
99+
if (message.getDeliveryTimestamp().isPresent()) {
100+
messageBuilder.setDeliveryTimestamp(message.getDeliveryTimestamp().get());
101+
}
102+
properties.entrySet().forEach(item -> messageBuilder.addProperty(item.getKey(), item.getValue()));
103+
allArguments[0] = messageBuilder.build();
104+
}
105+
106+
@Override
107+
public Object afterMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Object ret) throws Throwable {
108+
SendReceipt sendReceipt = (SendReceipt) ret;
109+
if (sendReceipt != null && sendReceipt.getMessageId() != null) {
110+
AbstractSpan activeSpan = ContextManager.activeSpan();
111+
activeSpan.tag(Tags.ofKey("mq.message.id"), sendReceipt.getMessageId().toString());
112+
}
113+
ContextManager.stopSpan();
114+
return ret;
115+
}
116+
117+
@Override
118+
public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes, Throwable t) {
119+
ContextManager.activeSpan().log(t);
120+
}
121+
122+
private String buildOperationName(String topicName) {
123+
return ASYNC_SEND_OPERATION_NAME_PREFIX + topicName + "/Producer";
124+
}
125+
}

‎apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/PushConsumerImplInterceptor.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5;
20+
21+
import org.apache.rocketmq.client.apis.ClientConfiguration;
22+
import org.apache.rocketmq.client.apis.consumer.MessageListener;
23+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
24+
import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceConstructorInterceptor;
25+
import org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define.ConsumerEnhanceInfos;
26+
27+
import java.util.Map;
28+
29+
/**
30+
* {@link PushConsumerImplInterceptor} create exit span when the method {@link org.apache.rocketmq.client.java.impl.consumer.PushConsumerImpl#PushConsumerImpl(ClientConfiguration, String, Map, MessageListener, int, int, int)} execute.
31+
*/
32+
public class PushConsumerImplInterceptor implements InstanceConstructorInterceptor {
33+
34+
@Override
35+
public void onConstruct(EnhancedInstance objInst, Object[] allArguments) {
36+
ClientConfiguration clientConfiguration = (ClientConfiguration) allArguments[0];
37+
String namesrvAddr = clientConfiguration.getEndpoints();
38+
ConsumerEnhanceInfos consumerEnhanceInfos = new ConsumerEnhanceInfos(namesrvAddr);
39+
40+
if (allArguments[3] instanceof EnhancedInstance) {
41+
EnhancedInstance enhancedMessageListener = (EnhancedInstance) allArguments[3];
42+
enhancedMessageListener.setSkyWalkingDynamicField(consumerEnhanceInfos);
43+
}
44+
}
45+
}

‎apm-sniffer/apm-sdk-plugin/rocketMQ-client-java-5.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/rocketMQ/client/java/v5/define/ConsumerEnhanceInfos.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*
17+
*/
18+
19+
package org.apache.skywalking.apm.plugin.rocketMQ.client.java.v5.define;
20+
21+
public class ConsumerEnhanceInfos {
22+
23+
private String namesrvAddr;
24+
25+
public ConsumerEnhanceInfos(String namesrvAddr) {
26+
this.namesrvAddr = namesrvAddr;
27+
}
28+
29+
public String getNamesrvAddr() {
30+
return namesrvAddr;
31+
}
32+
}

0 commit comments

Comments
 (0)