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

change the remote address of MQ exit span to naming service address

This commit is contained in:
ascrutae 2017-12-04 12:45:00 +08:00
parent bfbf7aebc9
commit 1fa81783f7
9 changed files with 81 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/AbstractMessageConsumeInterceptor.java → apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/AbstractMessageConsumeInterceptor.java
View File

@ -16,10 +16,11 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
package org.apache.rocketmq.common.message;
package org.skywalking.apm.plugin.rocketMQ.v4;
import java.lang.reflect.Method;
import java.util.List;
import org.apache.rocketmq.common.message.MessageExt;
import org.skywalking.apm.agent.core.context.CarrierItem;
import org.skywalking.apm.agent.core.context.ContextCarrier;
import org.skywalking.apm.agent.core.context.ContextManager;

2
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageConcurrentlyConsumeInterceptor.java → apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageConcurrentlyConsumeInterceptor.java
View File

@ -16,7 +16,7 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
package org.apache.rocketmq.common.message;
package org.skywalking.apm.plugin.rocketMQ.v4;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;

2
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageOrderlyConsumeInterceptor.java → apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageOrderlyConsumeInterceptor.java
View File

@ -16,7 +16,7 @@
* Project repository: https://github.com/OpenSkywalking/skywalking
*/
package org.apache.rocketmq.common.message;
package org.skywalking.apm.plugin.rocketMQ.v4;
import java.lang.reflect.Method;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;

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

@ -56,7 +56,8 @@ public class MessageSendInterceptor implements InstanceMethodsAroundInterceptor
MethodInterceptResult result) throws Throwable {
Message message = (Message)allArguments[2];
ContextCarrier contextCarrier = new ContextCarrier();
AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, (String)allArguments[0]);
String namingServiceAddress = String.valueOf(objInst.getSkyWalkingDynamicField());
AbstractSpan span = ContextManager.createExitSpan(buildOperationName(message.getTopic()), contextCarrier, namingServiceAddress);
span.setComponent(ComponentsDefine.ROCKET_MQ);
span.setLayer(SpanLayer.MQ);
span.tag("brokerName", (String)allArguments[1]);

43
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java Normal file
View File

@ -0,0 +1,43 @@
/*
* Copyright 2017, OpenSkywalking Organization All rights reserved.
*
* Licensed 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 repository: https://github.com/OpenSkywalking/skywalking
*/
package org.skywalking.apm.plugin.rocketMQ.v4;
import java.lang.reflect.Method;
import org.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
import org.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
import org.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) {
}
}

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

@ -31,7 +31,7 @@ import static org.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierar
/**
* {@link ConsumeMessageConcurrentlyInstrumentation} intercepts the {@link org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently#consumeMessage(java.util.List,
* org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext)} method by using {@link
* org.apache.rocketmq.common.message.MessageConcurrentlyConsumeInterceptor}.
* org.skywalking.apm.plugin.rocketMQ.v4.MessageConcurrentlyConsumeInterceptor}.
*
* @author zhang xin
*/

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

@ -31,7 +31,7 @@ import static org.skywalking.apm.agent.core.plugin.match.HierarchyMatch.byHierar
/**
* {@link ConsumeMessageOrderlyInstrumentation} intercepts the {@link org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly#consumeMessage(java.util.List,
* org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext)} method by using {@link
* org.apache.rocketmq.common.message.MessageConcurrentlyConsumeInterceptor}.
* org.skywalking.apm.plugin.rocketMQ.v4.MessageConcurrentlyConsumeInterceptor}.
*
* @author zhang xin
*/

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

@ -44,6 +44,8 @@ public class MQClientAPIImplInstrumentation extends ClassInstanceMethodsEnhanceP
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.skywalking.apm.plugin.rocketMQ.v4.MessageSendInterceptor";
public static final String UPDATE_NAME_SERVER_INTERCEPT_CLASS = "org.skywalking.apm.plugin.rocketMQ.v4.UpdateNameServerInterceptor";
public static final String UPDATE_NAME_SERVER_METHOD_NAME = "updateNameServerAddressList";
@Override protected ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
return new ConstructorInterceptPoint[0];
@ -60,6 +62,19 @@ public class MQClientAPIImplInstrumentation extends ClassInstanceMethodsEnhanceP
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;
}

19
apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java
View File

@ -74,9 +74,20 @@ public class MessageSendInterceptorTest {
@Mock
private EnhancedInstance callBack;
private EnhancedInstance enhancedInstance;
@Before
public void setUp() {
messageSendInterceptor = new MessageSendInterceptor();
enhancedInstance = new EnhancedInstance() {
@Override public Object getSkyWalkingDynamicField() {
return "127.0.0.1:6543";
}
@Override public void setSkyWalkingDynamicField(Object value) {
}
};
arguments = new Object[] {"127.0.0.1", "test", message, messageRequestHeader, null, CommunicationMode.ASYNC, callBack};
argumentsWithoutCallback = new Object[] {"127.0.0.1", "test", message, messageRequestHeader, null, CommunicationMode.ASYNC, null};
@ -86,8 +97,8 @@ public class MessageSendInterceptorTest {
@Test
public void testSendMessage() throws Throwable {
messageSendInterceptor.beforeMethod(null, null, arguments, null, null);
messageSendInterceptor.afterMethod(null, null, arguments, null, null);
messageSendInterceptor.beforeMethod(enhancedInstance, null, arguments, null, null);
messageSendInterceptor.afterMethod(enhancedInstance, null, arguments, null, null);
assertThat(segmentStorage.getTraceSegments().size(), is(1));
TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);
@ -106,8 +117,8 @@ public class MessageSendInterceptorTest {
@Test
public void testSendMessageWithoutCallBack() throws Throwable {
messageSendInterceptor.beforeMethod(null, null, argumentsWithoutCallback, null, null);
messageSendInterceptor.afterMethod(null, null, argumentsWithoutCallback, null, null);
messageSendInterceptor.beforeMethod(enhancedInstance, null, argumentsWithoutCallback, null, null);
messageSendInterceptor.afterMethod(enhancedInstance, null, argumentsWithoutCallback, null, null);
assertThat(segmentStorage.getTraceSegments().size(), is(1));
TraceSegment traceSegment = segmentStorage.getTraceSegments().get(0);