Supporting SpringMapMessageListener

CHANGELOG.md

@@ -1,4 +1,12 @@
-# next (1.7.0)
+# next (1.8.0)
+
+## Features
+ * Add support for Spring's JMS flavor - instrumenting `org.springframework.jms.listener.SessionAwareMessageListener`
+
+## Bug Fixes
+ * Some JMS Consumers and Producers are filtered due to class name filtering in instrumentation matching
+
+# 1.7.0
 
 ## Features
  * Added the `trace_methods_duration_threshold` config option. When using the `trace_methods` config option with wild cards, this 

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/TraceContext.java

@@ -199,6 +199,8 @@ public boolean asChildOf(String traceParentHeader) {
         } catch (IllegalArgumentException e) {
             logger.warn(e.getMessage());
             return false;
+        } finally {
+            onMutation();
         }
     }
 

apm-agent-core/src/test/java/co/elastic/apm/agent/impl/transaction/TraceContextTest.java

@@ -96,6 +96,14 @@ void testResetState() {
         assertThat(traceContext.getIncomingTraceParentHeader()).isEqualTo("00-00000000000000000000000000000000-0000000000000000-00");
     }
 
+    @Test
+    void testResetOutgoingHeader() {
+        final TraceContext traceContext = TraceContext.with64BitId(mock(ElasticApmTracer.class));
+        String traceParentHeader = traceContext.getOutgoingTraceParentHeader().toString();
+        traceContext.asChildOf("00-0af7651916cd43dd8448eb211c80319c-b9c7c989f97918e1-00");
+        assertThat(traceContext.getOutgoingTraceParentHeader().toString()).isNotEqualTo(traceParentHeader);
+    }
+
     @Test
     void testRandomValue() {
         final TraceContext traceContext = TraceContext.with64BitId(mock(ElasticApmTracer.class));

apm-agent-plugins/apm-jms-plugin/pom.xml

@@ -9,6 +9,9 @@
 
     <properties>
         <apm-agent-parent.base.dir>${project.basedir}/../..</apm-agent-parent.base.dir>
+        <spring-framework.version>5.1.8.RELEASE</spring-framework.version>
+        <activemq.version>5.15.9</activemq.version>
+        <artemis.version>2.8.1</artemis.version>
     </properties>
 
     <artifactId>apm-jms-plugin</artifactId>
@@ -24,19 +27,31 @@
         <dependency>
             <groupId>org.apache.activemq</groupId>
             <artifactId>activemq-broker</artifactId>
-            <version>5.15.9</version>
+            <version>${activemq.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.activemq</groupId>
             <artifactId>artemis-jms-client</artifactId>
-            <version>2.8.1</version>
+            <version>${artemis.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.activemq</groupId>
             <artifactId>artemis-jms-server</artifactId>
-            <version>2.8.1</version>
+            <version>${artemis.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jms</artifactId>
+            <version>${spring-framework.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-context</artifactId>
+            <version>${spring-framework.version}</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

apm-agent-plugins/apm-jms-plugin/src/main/java/co/elastic/apm/agent/jms/JmsMessageConsumerInstrumentation.java

@@ -69,7 +69,10 @@ public abstract class JmsMessageConsumerInstrumentation extends BaseJmsInstrumen
 
     @Override
     public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
-        return nameContains("Message").or(nameContains("Consumer"));
+        return nameContains("Message")
+            .or(nameContains("Consumer"))
+            .or(nameContains("Receiver"))
+            .or(nameContains("Subscriber"));
     }
 
     @Override

apm-agent-plugins/apm-jms-plugin/src/main/java/co/elastic/apm/agent/jms/JmsMessageListenerInstrumentation.java

@@ -63,14 +63,17 @@ public JmsMessageListenerInstrumentation(ElasticApmTracer tracer) {
 
     @Override
     public ElementMatcher<? super TypeDescription> getTypeMatcher() {
-        return not(isInterface()).and(hasSuperType(named("javax.jms.MessageListener")));
+        return not(isInterface())
+            .and(
+                hasSuperType(named("javax.jms.MessageListener"))
+                    .or(hasSuperType(named("org.springframework.jms.listener.SessionAwareMessageListener")))
+            );
     }
 
     @Override
     public ElementMatcher<? super MethodDescription> getMethodMatcher() {
         return named("onMessage")
-            .and(takesArguments(1))
-            .and(takesArgument(0, named("javax.jms.Message"))).and(isPublic());
+            .and(takesArgument(0, hasSuperType(named("javax.jms.Message")))).and(isPublic());
     }
 
     @Override

apm-agent-plugins/apm-jms-plugin/src/main/java/co/elastic/apm/agent/jms/JmsMessageProducerInstrumentation.java

@@ -68,7 +68,10 @@ public abstract class JmsMessageProducerInstrumentation extends BaseJmsInstrumen
 
     @Override
     public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
-        return nameContains("Message").or(nameContains("Producer"));
+        return nameContains("Message")
+            .or(nameContains("Producer"))
+            .or(nameContains("Sender"))
+            .or(nameContains("Publisher"));
     }
 
     @Override

apm-agent-plugins/apm-jms-plugin/src/test/java/co/elastic/apm/agent/jms/spring/SpringJmsTest.java

@@ -0,0 +1,134 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2019 Elastic and contributors
+ * %%
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.
+ * #L%
+ */
+package co.elastic.apm.agent.jms.spring;
+
+import co.elastic.apm.agent.AbstractInstrumentationTest;
+import co.elastic.apm.agent.impl.transaction.Id;
+import co.elastic.apm.agent.impl.transaction.Span;
+import co.elastic.apm.agent.impl.transaction.TraceContext;
+import co.elastic.apm.agent.impl.transaction.Transaction;
+import org.apache.activemq.ActiveMQConnectionFactory;
+import org.apache.activemq.command.ActiveMQMapMessage;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class SpringJmsTest extends AbstractInstrumentationTest {
+
+    private static final String SPRING_TEST_QUEUE = "Spring-Test-Queue";
+    static final BlockingQueue<Map> resultQueue = new ArrayBlockingQueue<>(5);
+
+    private static Connection connection;
+    private static ClassPathXmlApplicationContext ctx;
+
+    @BeforeClass
+    public static void setup() throws JMSException {
+        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("vm://localhost?broker.persistent=false");
+        connection = connectionFactory.createConnection();
+        connection.start();
+        ctx = new ClassPathXmlApplicationContext("app-context.xml");
+    }
+
+    @AfterClass
+    public static void teardown() throws JMSException {
+        ctx.close();
+        connection.stop();
+    }
+
+    @Test
+    public void testSendListenSpringQueue() throws JMSException, InterruptedException {
+        reporter.reset();
+        try (Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE)) {
+
+            Transaction transaction = tracer.startTransaction(TraceContext.asRoot(), null, null).activate();
+            transaction.setName("JMS-Spring-Test Transaction");
+            transaction.withType("request");
+            transaction.withResult("success");
+
+            final String key1 = "key1";
+            final String key2 = "key2";
+            final String value1 = UUID.randomUUID().toString();
+            final String value2 = UUID.randomUUID().toString();
+            MapMessage mapMessage = new ActiveMQMapMessage();
+            mapMessage.setString(key1, value1);
+            mapMessage.setString(key2, value2);
+
+            Queue queue = session.createQueue(SPRING_TEST_QUEUE);
+            MessageProducer producer = session.createProducer(queue);
+            producer.send(mapMessage);
+
+            // Let the onMessage instrumentation end the transaction then end the base transaction
+            Thread.sleep(500);
+            transaction.deactivate().end();
+
+            Map result = resultQueue.poll(1, TimeUnit.SECONDS);
+
+            assertThat(result).isNotNull();
+            assertThat(result.size()).isEqualTo(2);
+            assertThat(result.get(key1).toString()).isEqualTo(value1);
+            assertThat(result.get(key2).toString()).isEqualTo(value2);
+
+            List<Transaction> transactions = reporter.getTransactions();
+            // The way the Spring framework works is polling through a standard JMS receive API in order to get the
+            // message, with which it then invokes the SpringMapMessageListener.onMessage() implementation, so we expect
+            // two JMS receive transactions (one for the receive and one for the onMessage), both with same parent and traceId
+            assertThat(transactions).hasSize(3);
+            Transaction baseTransaction = transactions.get(2);
+            Id traceId = baseTransaction.getTraceContext().getTraceId();
+
+            List<Span> spans = reporter.getSpans();
+            assertThat(spans).hasSize(1);
+            Span sendSpan = spans.get(0);
+            assertThat(sendSpan.getName().toString()).isEqualTo("JMS SEND to queue " + SPRING_TEST_QUEUE);
+            assertThat(sendSpan.getTraceContext().getTraceId()).isEqualTo(traceId);
+
+            Transaction receiveTransaction = transactions.get(0);
+            verifyReceiveTransaction(traceId, sendSpan, receiveTransaction);
+        }
+    }
+
+    private void verifyReceiveTransaction(Id traceId, Span sendSpan, Transaction receiveTransaction) {
+        assertThat(receiveTransaction.getName().toString()).isEqualTo("JMS RECEIVE from queue " + SPRING_TEST_QUEUE);
+        assertThat(receiveTransaction.getTraceContext().getTraceId()).isEqualTo(traceId);
+        assertThat(receiveTransaction.getTraceContext().getParentId()).isEqualTo(sendSpan.getTraceContext().getId());
+    }
+}

apm-agent-plugins/apm-jms-plugin/src/test/java/co/elastic/apm/agent/jms/spring/SpringMapMessageListener.java

@@ -0,0 +1,46 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2019 Elastic and contributors
+ * %%
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. 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.
+ * #L%
+ */
+package co.elastic.apm.agent.jms.spring;
+
+import org.apache.activemq.command.ActiveMQMapMessage;
+import org.springframework.jms.listener.SessionAwareMessageListener;
+import org.springframework.stereotype.Service;
+
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Session;
+import java.util.Map;
+
+@Service
+public class SpringMapMessageListener implements SessionAwareMessageListener<MapMessage> {
+
+    @Override
+    public void onMessage(MapMessage mapMessage, Session session) throws JMSException {
+        Map map = ((ActiveMQMapMessage) mapMessage).getContentMap();
+        System.out.println("Received map message: ");
+        map.forEach((key, value) -> System.out.println("   " + key + " --> " + value));
+        SpringJmsTest.resultQueue.offer(map);
+    }
+}

apm-agent-plugins/apm-jms-plugin/src/test/resources/app-context.xml

@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+       xmlns:context="http://www.springframework.org/schema/context"
+       xsi:schemaLocation="
+    http://www.springframework.org/schema/beans
+    http://www.springframework.org/schema/beans/spring-beans.xsd
+    http://www.springframework.org/schema/context
+    http://www.springframework.org/schema/context/spring-context.xsd">
+
+
+    <context:component-scan base-package="co.elastic.apm.agent.jms"/>
+
+    <!-- =============================================== -->
+    <!-- JMS Common，Define JMS connection Factory       -->
+    <!-- =============================================== -->
+    <!-- Activemq connection factory -->
+    <bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
+        <!-- brokerURL -->
+        <constructor-arg index="0" value="vm://localhost?broker.persistent=false"/>
+    </bean>
+
+    <!-- Pooled Spring connection factory -->
+    <bean id="connectionFactory"
+          class="org.springframework.jms.connection.CachingConnectionFactory">
+        <constructor-arg ref="amqConnectionFactory"/>
+    </bean>
+
+
+    <!-- ============================================================= -->
+    <!-- JMS Receive，Define MessageListenerContainer                  -->
+    <!-- ============================================================= -->
+    <bean id="messageListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
+        <property name="connectionFactory" ref="connectionFactory"/>
+        <property name="destinationName" value="Spring-Test-Queue"/>
+        <property name="messageListener" ref="springMapMessageListener"/>
+    </bean>
+
+</beans>