Add Spring RabbitMQ plugin

.github/workflows/plugins-jdk17-test.0.yaml

@@ -80,6 +80,7 @@ jobs:
           - jetty-11.x-scenario
           - jetty-10.x-scenario
           - spring-ai-1.x-scenario
+          - spring-rabbitmq-3.x-4.x-scenario
     steps:
       - uses: actions/checkout@v2
         with:

.github/workflows/plugins-test.3.yaml

@@ -96,6 +96,7 @@ jobs:
           - spring-kafka-1.3.x-scenario
           - spring-kafka-2.2.x-scenario
           - spring-kafka-2.3.x-scenario
+          - spring-rabbitmq-2.x-scenario
           - spring-scheduled-3.x-5.x-scenario
           - elasticjob-2.x-scenario
           - quartz-scheduler-2.x-scenario

CHANGES.md

@@ -7,6 +7,7 @@ Release Notes.
 
 * Added support for Lettuce reactive Redis commands.
 * Add Spring AI 1.x plugin and GenAI layer.
+* Add Spring RabbitMQ plugin.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/249?closed=1)
 

apm-sniffer/apm-sdk-plugin/rabbitmq-plugin/src/main/java/org/apache/skywalking/apm/plugin/rabbitmq/RabbitMQConsumerInterceptor.java

@@ -29,6 +29,9 @@ public class RabbitMQConsumerInterceptor implements InstanceMethodsAroundInterce
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
         MethodInterceptResult result) throws Throwable {
+        if (Thread.currentThread().getName().toLowerCase().contains("springframework")) {
+            return;
+        }
         Consumer consumer = (Consumer) allArguments[6];
         allArguments[6] = new TracerConsumer(consumer, (String) objInst.getSkyWalkingDynamicField());
     }

apm-sniffer/apm-sdk-plugin/spring-plugins/pom.xml

@@ -46,10 +46,11 @@
         <module>spring-webflux-6.x-webclient-plugin</module>
         <module>resttemplate-commons</module>
         <module>spring-ai-1.x-plugin</module>
+        <module>spring-rabbitmq-plugin</module>
     </modules>
     <packaging>pom</packaging>
 
-    <name>apm-sdk-plugin</name>
+    <name>spring-plugins</name>
     <url>http://maven.apache.org</url>
 
     <properties>

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-rabbitmq-plugin/pom.xml

@@ -0,0 +1,46 @@
+<?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">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.skywalking</groupId>
+        <artifactId>spring-plugins</artifactId>
+        <version>9.7.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>apm-spring-rabbitmq-plugin</artifactId>
+    <name>apm-spring-rabbitmq-plugin</name>
+    <packaging>jar</packaging>
+
+    <properties>
+        <spring-rabbit.version>2.2.1.RELEASE</spring-rabbit.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.amqp</groupId>
+            <artifactId>spring-rabbit</artifactId>
+            <version>${spring-rabbit.version}</version>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+</project>

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-rabbitmq-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/rabbitmq/SpringRabbitMQConsumerInterceptor.java

@@ -0,0 +1,85 @@
+/*
+ * 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.spring.rabbitmq;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+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.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+public class SpringRabbitMQConsumerInterceptor implements InstanceMethodsAroundInterceptor {
+    public static final String OPERATE_NAME_PREFIX = "RabbitMQ/";
+    public static final String CONSUMER_OPERATE_NAME_SUFFIX = "/Consumer";
+
+    @Override
+    public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
+        MethodInterceptResult result) throws Throwable {
+        Channel channel = (Channel) allArguments[0];
+        Message message = (Message) allArguments[1];
+        MessageProperties messageProperties = message.getMessageProperties();
+        Map<String, Object> headers = messageProperties.getHeaders();
+        ContextCarrier contextCarrier = new ContextCarrier();
+        CarrierItem next = contextCarrier.items();
+        while (next.hasNext()) {
+            next = next.next();
+            Object value = headers.get(next.getHeadKey());
+            if (value != null) {
+                next.setHeadValue(value.toString());
+            }
+        }
+        String operationName = OPERATE_NAME_PREFIX + "Topic/" + messageProperties.getReceivedExchange()
+                               + "Queue/" + messageProperties.getReceivedRoutingKey() + CONSUMER_OPERATE_NAME_SUFFIX;
+        AbstractSpan activeSpan = ContextManager.createEntrySpan(operationName, contextCarrier);
+        Connection connection = channel.getConnection();
+        String serverUrl = connection.getAddress().getHostAddress() + ":" + connection.getPort();
+        Tags.MQ_BROKER.set(activeSpan, serverUrl);
+        Tags.MQ_TOPIC.set(activeSpan, messageProperties.getReceivedExchange());
+        Tags.MQ_QUEUE.set(activeSpan, messageProperties.getReceivedRoutingKey());
+        activeSpan.setComponent(ComponentsDefine.RABBITMQ_CONSUMER);
+        activeSpan.setPeer(serverUrl);
+        SpanLayer.asMQ(activeSpan);
+    }
+
+    @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);
+    }
+}

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-rabbitmq-plugin/src/main/java/org/apache/skywalking/apm/plugin/spring/rabbitmq/define/SpringRabbitMQConsumerInstrumentation.java

@@ -0,0 +1,69 @@
+/*
+ * 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.spring.rabbitmq.define;
+
+import static net.bytebuddy.matcher.ElementMatchers.named;
+
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.ConstructorInterceptPoint;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.DeclaredInstanceMethodsInterceptPoint;
+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.NameMatch;
+
+import net.bytebuddy.description.method.MethodDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+public class SpringRabbitMQConsumerInstrumentation extends ClassInstanceMethodsEnhancePluginDefine {
+    public static final String ENHANCE_CLASS = "org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer";
+    public static final String ENHANCE_METHOD = "executeListener";
+    public static final String INTERCEPTOR_CLASS = "org.apache.skywalking.apm.plugin.spring.rabbitmq.SpringRabbitMQConsumerInterceptor";
+
+    @Override
+    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
+        return null;
+    }
+
+    @Override
+    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
+        return new InstanceMethodsInterceptPoint[] {
+            new DeclaredInstanceMethodsInterceptPoint() {
+                @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 NameMatch.byName(ENHANCE_CLASS);
+    }
+}

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-rabbitmq-plugin/src/main/resources/skywalking-plugin.def

@@ -0,0 +1,17 @@
+# 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.
+
+spring-rabbitmq=org.apache.skywalking.apm.plugin.spring.rabbitmq.define.SpringRabbitMQConsumerInstrumentation

apm-sniffer/apm-sdk-plugin/spring-plugins/spring-rabbitmq-plugin/src/test/java/org/apache/skywalking/apm/plugin/spring/rabbitmq/RabbitMQSpringConsumerInterceptorTest.java

@@ -0,0 +1,111 @@
+/*
+ * 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.spring.rabbitmq;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.skywalking.apm.agent.core.context.SW8CarrierItem;
+import org.apache.skywalking.apm.agent.core.context.trace.TraceSegment;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.apache.skywalking.apm.agent.test.tools.AgentServiceRule;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStorage;
+import org.apache.skywalking.apm.agent.test.tools.SegmentStoragePoint;
+import org.apache.skywalking.apm.agent.test.tools.TracingSegmentRunner;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+
+@RunWith(TracingSegmentRunner.class)
+public class RabbitMQSpringConsumerInterceptorTest {
+
+    @SegmentStoragePoint
+    private SegmentStorage segmentStorage;
+
+    private SpringRabbitMQConsumerInterceptor rabbitMQConsumerInterceptor;
+
+    @Mock
+    private EnhancedInstance enhancedInstance;
+
+    @Rule
+    public AgentServiceRule serviceRule = new AgentServiceRule();
+
+    @Before
+    public void setUp() throws Exception {
+        rabbitMQConsumerInterceptor = new SpringRabbitMQConsumerInterceptor();
+    }
+
+    @Test
+    public void testRabbitMQConsumerInterceptorWithNilHeaders() throws Throwable {
+        Object[] args = prepareMockData(false);
+        rabbitMQConsumerInterceptor.beforeMethod(enhancedInstance, null, args, new Class[0], null);
+        rabbitMQConsumerInterceptor.afterMethod(enhancedInstance, null, args, new Class[0], null);
+        List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
+        Assert.assertThat(traceSegments.size(), is(1));
+    }
+
+    @Test
+    public void testRabbitMQConsumerInterceptor() throws Throwable {
+        Object[] args = prepareMockData(true);
+        rabbitMQConsumerInterceptor.beforeMethod(enhancedInstance, null, args, new Class[0], null);
+        rabbitMQConsumerInterceptor.afterMethod(enhancedInstance, null, args, new Class[0], null);
+        List<TraceSegment> traceSegments = segmentStorage.getTraceSegments();
+        Assert.assertThat(traceSegments.size(), is(1));
+    }
+
+    private Object[] prepareMockData(boolean withHeaders) throws Exception {
+        Channel channel = mock(Channel.class);
+        Connection connection = mock(Connection.class);
+        InetAddress address = mock(InetAddress.class);
+        Message message = mock(Message.class);
+        MessageProperties messageProperties = mock(MessageProperties.class);
+
+        when(channel.getConnection()).thenReturn(connection);
+        when(connection.getAddress()).thenReturn(address);
+        when(address.getHostAddress()).thenReturn("127.0.0.1");
+        when(connection.getPort()).thenReturn(5672);
+        when(message.getMessageProperties()).thenReturn(messageProperties);
+        when(messageProperties.getReceivedExchange()).thenReturn("test-exchange");
+        when(messageProperties.getReceivedRoutingKey()).thenReturn("test-routing-key");
+
+        if (withHeaders) {
+            Map<String, Object> headers = new HashMap<>();
+            headers.put(SW8CarrierItem.HEADER_NAME,
+                "1-My40LjU=-MS4yLjM=-3-c2VydmljZQ==-aW5zdGFuY2U=-L2FwcA==-MTI3LjAuMC4xOjgwODA=");
+            when(messageProperties.getHeader(SW8CarrierItem.HEADER_NAME))
+                .thenReturn(headers.get(SW8CarrierItem.HEADER_NAME));
+        }
+
+        return new Object[] {channel, message};
+    }
+}

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

@@ -114,6 +114,7 @@
 - spring-core-patch
 - spring-kafka-1.x
 - spring-kafka-2.x
+- spring-rabbitmq
 - spring-mvc-annotation
 - spring-mvc-annotation-3.x
 - spring-mvc-annotation-4.x