From 1fa81783f79483ea95ef8113cf812578f1d003a3 Mon Sep 17 00:00:00 2001 From: ascrutae Date: Mon, 4 Dec 2017 12:45:00 +0800 Subject: [PATCH] change the remote address of MQ exit span to naming service address --- .../AbstractMessageConsumeInterceptor.java | 3 +- ...MessageConcurrentlyConsumeInterceptor.java | 2 +- .../v4}/MessageOrderlyConsumeInterceptor.java | 2 +- .../rocketMQ/v4/MessageSendInterceptor.java | 3 +- .../v4/UpdateNameServerInterceptor.java | 43 +++++++++++++++++++ ...umeMessageConcurrentlyInstrumentation.java | 2 +- .../ConsumeMessageOrderlyInstrumentation.java | 2 +- .../MQClientAPIImplInstrumentation.java | 15 +++++++ .../v4/MessageSendInterceptorTest.java | 19 ++++++-- 9 files changed, 81 insertions(+), 10 deletions(-) rename apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/{apache/rocketmq/common/message => skywalking/apm/plugin/rocketMQ/v4}/AbstractMessageConsumeInterceptor.java (96%) rename apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/{apache/rocketmq/common/message => skywalking/apm/plugin/rocketMQ/v4}/MessageConcurrentlyConsumeInterceptor.java (97%) rename apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/{apache/rocketmq/common/message => skywalking/apm/plugin/rocketMQ/v4}/MessageOrderlyConsumeInterceptor.java (97%) create mode 100644 apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/AbstractMessageConsumeInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/AbstractMessageConsumeInterceptor.java similarity index 96% rename from apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/AbstractMessageConsumeInterceptor.java rename to apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/AbstractMessageConsumeInterceptor.java index 04af5c96a..b428759a8 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/AbstractMessageConsumeInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/AbstractMessageConsumeInterceptor.java @@ -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; diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageConcurrentlyConsumeInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageConcurrentlyConsumeInterceptor.java similarity index 97% rename from apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageConcurrentlyConsumeInterceptor.java rename to apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageConcurrentlyConsumeInterceptor.java index fb5725228..2e786e651 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageConcurrentlyConsumeInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageConcurrentlyConsumeInterceptor.java @@ -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; diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageOrderlyConsumeInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageOrderlyConsumeInterceptor.java similarity index 97% rename from apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageOrderlyConsumeInterceptor.java rename to apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageOrderlyConsumeInterceptor.java index dcf94717e..151bc8788 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/apache/rocketmq/common/message/MessageOrderlyConsumeInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageOrderlyConsumeInterceptor.java @@ -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; diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java index b85165856..d8f9eae70 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptor.java @@ -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]); diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java new file mode 100644 index 000000000..59d755e5e --- /dev/null +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/UpdateNameServerInterceptor.java @@ -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) { + + } +} diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java index 225471878..18214fe3e 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageConcurrentlyInstrumentation.java @@ -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 */ diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java index 78984cd66..968259ab6 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/ConsumeMessageOrderlyInstrumentation.java @@ -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 */ diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java index 582e31134..5cf5d8a9f 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/main/java/org/skywalking/apm/plugin/rocketMQ/v4/define/MQClientAPIImplInstrumentation.java @@ -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 getMethodsMatcher() { + return named(UPDATE_NAME_SERVER_METHOD_NAME); + } + + @Override public String getMethodsInterceptor() { + return UPDATE_NAME_SERVER_INTERCEPT_CLASS; + } + @Override public boolean isOverrideArgs() { return false; } diff --git a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java index 8bc0a0a72..d4d783712 100644 --- a/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java +++ b/apm-sniffer/apm-sdk-plugin/rocketMQ-4.x-plugin/src/test/java/org/skywalking/apm/plugin/rocketMQ/v4/MessageSendInterceptorTest.java @@ -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);