support Spring AMQP's SimpleMessageListenerContainer (#1657)

Author: videnkz

CHANGELOG.asciidoc

@@ -43,6 +43,7 @@ to assist with such efforts.
 * Adding capability to automatically create ECS-JSON-formatted version of the original application log files, through
 the <<config-log-ecs-reformatting>> config option. This allows effortless ingestion of logs to Elasticsearch without
 any further configuration. Supports log4j1, log4j2 and Logback. {pull}1261[#1261]
+* Add support to Spring AMQP - {pull}1657[#1657]
 
 [float]
 ===== Bug fixes

apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/MetricsConfiguration.java

@@ -1,3 +1,27 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2021 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.configuration;
 
 import org.stagemonitor.configuration.ConfigurationOption;

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/BaseInstrumentation.java renamed to apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/AbstractBaseInstrumentation.java

@@ -27,29 +27,20 @@
 import co.elastic.apm.agent.bci.TracerAwareInstrumentation;
 import co.elastic.apm.agent.configuration.CoreConfiguration;
 import co.elastic.apm.agent.configuration.MessagingConfiguration;
-import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.impl.GlobalTracer;
 import co.elastic.apm.agent.impl.context.Message;
 import co.elastic.apm.agent.impl.transaction.AbstractSpan;
 import co.elastic.apm.agent.matcher.WildcardMatcher;
-import com.rabbitmq.client.AMQP;
 
 import javax.annotation.Nullable;
-import java.util.Collection;
-import java.util.Collections;
 import java.util.Date;
 import java.util.Map;
 
-public abstract class BaseInstrumentation extends TracerAwareInstrumentation {
+public abstract class AbstractBaseInstrumentation extends TracerAwareInstrumentation {
 
  private static final CoreConfiguration coreConfiguration = GlobalTracer.requireTracerImpl().getConfig(CoreConfiguration.class);
  private static final MessagingConfiguration messagingConfiguration = GlobalTracer.requireTracerImpl().getConfig(MessagingConfiguration.class);
 
- @Override
- public Collection<String> getInstrumentationGroupNames() {
- return Collections.singletonList("rabbitmq");
- }
-
  /**
  * @param name name of the exchange or queue
  * @return {@literal true} when exchange or queue is ignored, {@literal false otherwise}
@@ -58,26 +49,49 @@ protected static boolean isIgnored(String name) {
  return WildcardMatcher.isAnyMatch(messagingConfiguration.getIgnoreMessageQueues(), name);
  }
 
- private static boolean isCaptureHeaders() {
+ protected static boolean isCaptureHeaders() {
  return coreConfiguration.isCaptureHeaders();
  }
 
- private static boolean captureHeaderKey(String key) {
+ protected static boolean captureHeaderKey(String key) {
  return !WildcardMatcher.isAnyMatch(coreConfiguration.getSanitizeFieldNames(), key);
  }
 
  /**
- * Captures que name and optional timestamp
+ * Captures queue name and optional timestamp
  *
  * @param queueOrExchange queue or exchange name to use in message.queue.name
- * @param properties properties (if any)
+ * @param age  age
  * @param context span/transaction context
  * @return captured message
  */
- protected static Message captureMessage(String queueOrExchange, @Nullable AMQP.BasicProperties properties, AbstractSpan<?> context) {
+ protected static Message captureMessage(String queueOrExchange, long age, AbstractSpan<?> context) {
  return context.getContext().getMessage()
  .withQueue(queueOrExchange)
- .withAge(getTimestamp(properties));
+ .withAge(age);
+ }
+
+ protected static long getTimestamp(@Nullable Date timestamp) {
+ long age = -1L;
+ if (timestamp != null) {
+ long now = System.currentTimeMillis();
+ long time = timestamp.getTime();
+ age = time <= now ? (now - time) : 0;
+ }
+ return age;
+ }
+
+ protected static void captureHeaders(@Nullable Map<String, Object> headers, Message message) {
+ if (!isCaptureHeaders() || headers == null || headers.size() <= 0) {
+ return;
+ }
+
+ for (Map.Entry<String, Object> entry : headers.entrySet()) {
+ if (captureHeaderKey(entry.getKey())) {
+ // headers aren't stored as String instances here
+ message.addHeader(entry.getKey(), String.valueOf(entry.getValue()));
+ }
+ }
  }
 
  protected static String normalizeExchangeName(@Nullable String exchange) {
@@ -100,33 +114,4 @@ protected static String normalizeQueueName(@Nullable String queue) {
  }
  return queue;
  }
-
- private static long getTimestamp(@Nullable AMQP.BasicProperties properties) {
- long age = -1L;
- if (null != properties) {
-
- Date timestamp = properties.getTimestamp();
- if (timestamp != null) {
- long now = System.currentTimeMillis();
- long time = timestamp.getTime();
- age = time <= now ? (now - time) : 0;
- }
- }
- return age;
- }
-
- protected static void captureHeaders(@Nullable AMQP.BasicProperties properties, Message message) {
- Map<String, Object> headers = properties != null ? properties.getHeaders() : null;
- if (!isCaptureHeaders() || headers == null || headers.size() <= 0) {
- return;
- }
-
- for (Map.Entry<String, Object> entry : headers.entrySet()) {
- if (captureHeaderKey(entry.getKey())) {
- // headers aren't stored as String instances here
- message.addHeader(entry.getKey(), String.valueOf(entry.getValue()));
- }
- }
-
- }
 }

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/ChannelInstrumentation.java

@@ -24,7 +24,6 @@
  */
 package co.elastic.apm.agent.rabbitmq;
 
-import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.impl.context.Destination;
 import co.elastic.apm.agent.impl.transaction.AbstractSpan;
 import co.elastic.apm.agent.impl.transaction.Span;
@@ -63,7 +62,7 @@
 /**
  * Instruments implementations of {@link com.rabbitmq.client.Channel}
  */
-public abstract class ChannelInstrumentation extends BaseInstrumentation {
+public abstract class ChannelInstrumentation extends RabbitmqBaseInstrumentation {
 
  @Override
  public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
@@ -148,7 +147,7 @@ public static Object[] onBasicPublish(@Advice.This Channel channel,
 
  properties = propagateTraceContext(exitSpan, properties);
 
- captureMessage(exchange, properties, exitSpan);
+ captureMessage(exchange, getTimestamp(properties.getTimestamp()), exitSpan);
  captureDestination(exchange, channel, exitSpan);
 
  return new Object[]{properties, exitSpan};
@@ -247,7 +246,7 @@ public static void onExit(@Advice.This Channel channel,
  span.requestDiscarding();
  }
 
- captureMessage(queue, properties, span);
+ captureMessage(queue, getTimestamp(properties != null ? properties.getTimestamp() : null), span);
  captureDestination(exchange, channel, span);
 
  span.captureException(thrown)

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/ConsumerInstrumentation.java

@@ -24,14 +24,14 @@
  */
 package co.elastic.apm.agent.rabbitmq;
 
-import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.impl.context.Message;
 import co.elastic.apm.agent.impl.transaction.Transaction;
 import co.elastic.apm.agent.rabbitmq.header.RabbitMQTextHeaderGetter;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.Envelope;
 import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.NamedElement;
 import net.bytebuddy.description.method.MethodDescription;
 import net.bytebuddy.description.type.TypeDescription;
 import net.bytebuddy.matcher.ElementMatcher;
@@ -41,6 +41,7 @@
 import static co.elastic.apm.agent.bci.bytebuddy.CustomElementMatchers.classLoaderCanLoadClass;
 import static net.bytebuddy.matcher.ElementMatchers.any;
 import static net.bytebuddy.matcher.ElementMatchers.isBootstrapClassLoader;
+import static net.bytebuddy.matcher.ElementMatchers.nameStartsWith;
 import static net.bytebuddy.matcher.ElementMatchers.named;
 import static net.bytebuddy.matcher.ElementMatchers.not;
 
@@ -50,7 +51,13 @@
  * <li>{@link com.rabbitmq.client.Consumer#handleDelivery}</li>
  * </ul>
  */
-public class ConsumerInstrumentation extends BaseInstrumentation {
+public class ConsumerInstrumentation extends RabbitmqBaseInstrumentation {
+
+ @Override
+ public ElementMatcher<? super NamedElement> getTypeMatcherPreFilter() {
+ // Spring RabbitMQ is supported through Spring interfaces, rather than the RabbitMQ API (apm-rabbitmq-spring module)
+ return not(nameStartsWith("org.springframework."));
+ }
 
  @Override
  public ElementMatcher<? super TypeDescription> getTypeMatcher() {
@@ -88,7 +95,6 @@ public static Object onHandleDelivery(@Advice.Origin Class<?> originClazz,
  if (!tracer.isRunning()) {
  return null;
  }
-
  String exchange = envelope != null ? envelope.getExchange() : null;
 
  if (null == exchange || isIgnored(exchange)) {
@@ -110,13 +116,12 @@ public static Object onHandleDelivery(@Advice.Origin Class<?> originClazz,
 
  transaction.setFrameworkName("RabbitMQ");
 
- Message message = captureMessage(exchange, properties, transaction);
+ Message message = captureMessage(exchange, getTimestamp(properties != null ? properties.getTimestamp() : null), transaction);
  // only capture incoming messages headers for now (consistent with other messaging plugins)
- captureHeaders(properties, message);
-
+ if (properties != null) {
+ captureHeaders(properties.getHeaders(), message);
+ }
  return transaction.activate();
-
-
  }
 
  @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class, inline = false)

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/RabbitmqBaseInstrumentation.java

@@ -0,0 +1,36 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2020 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.rabbitmq;
+
+import java.util.Collection;
+import java.util.Collections;
+
+public abstract class RabbitmqBaseInstrumentation extends AbstractBaseInstrumentation {
+
+ @Override
+ public Collection<String> getInstrumentationGroupNames() {
+ return Collections.singletonList("rabbitmq");
+ }
+}

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/header/AbstractTextHeaderGetter.java

@@ -0,0 +1,60 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2021 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.rabbitmq.header;
+
+import co.elastic.apm.agent.impl.transaction.TextHeaderGetter;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+
+public abstract class AbstractTextHeaderGetter<T> implements TextHeaderGetter<T> {
+
+ @Nullable
+ @Override
+ public String getFirstHeader(String headerName, T carrier) {
+ Map<String, Object> headers = getHeaders(carrier);
+ if (headers == null || headers.isEmpty()) {
+ return null;
+ }
+ Object headerValue = headers.get(headerName);
+ if (headerValue != null) {
+ // com.rabbitmq.client.impl.LongStringHelper.ByteArrayLongString
+ return headerValue.toString();
+ }
+ return null;
+ }
+
+ @Override
+ public <S> void forEach(String headerName, T carrier, S state, HeaderConsumer<String, S> consumer) {
+ String header = getFirstHeader(headerName, carrier);
+ if (header != null) {
+ consumer.accept(header, state);
+ }
+ }
+
+ @Nullable
+ protected abstract Map<String, Object> getHeaders(T carrier);
+
+}

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-plugin/src/main/java/co/elastic/apm/agent/rabbitmq/header/RabbitMQTextHeaderGetter.java

@@ -24,39 +24,19 @@
  */
 package co.elastic.apm.agent.rabbitmq.header;
 
-import co.elastic.apm.agent.impl.transaction.TextHeaderGetter;
 import com.rabbitmq.client.AMQP;
 
-import javax.annotation.Nullable;
 import java.util.Map;
 
-public class RabbitMQTextHeaderGetter implements TextHeaderGetter<AMQP.BasicProperties> {
+public class RabbitMQTextHeaderGetter extends AbstractTextHeaderGetter<AMQP.BasicProperties> {
 
  public static final RabbitMQTextHeaderGetter INSTANCE = new RabbitMQTextHeaderGetter();
 
  private RabbitMQTextHeaderGetter() {
  }
 
- @Nullable
  @Override
- public String getFirstHeader(String headerName, AMQP.BasicProperties carrier) {
- Map<String, Object> headers = carrier.getHeaders();
- if (headers == null || headers.isEmpty()) {
- return null;
- }
- Object headerValue = headers.get(headerName);
- if (headerValue != null) {
- // com.rabbitmq.client.impl.LongStringHelper.ByteArrayLongString
- return headerValue.toString();
- }
- return null;
- }
-
- @Override
- public <S> void forEach(String headerName, AMQP.BasicProperties carrier, S state, HeaderConsumer<String, S> consumer) {
- String header = getFirstHeader(headerName, carrier);
- if (header != null) {
- consumer.accept(header, state);
- }
+ protected Map<String, Object> getHeaders(AMQP.BasicProperties carrier) {
+ return carrier.getHeaders();
  }
 }