Add ability to change Naming Mode for spring-rabbit-plugin

CHANGELOG.asciidoc

@@ -33,6 +33,7 @@ Use subheadings with the "=====" level for adding notes for unreleased changes:
 
 [float]
 ===== Features
+* Added a configuration option to use queues in names of spring-rabbit transactions - {pull}3424[#3424]
 * Added support for OpenTelementry Attributes db.statement and db.user - {pull}3475[#3475]
 
 [float]

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

@@ -52,6 +52,10 @@ protected static boolean captureHeaderKey(String key) {
         return !WildcardMatcher.isAnyMatch(coreConfiguration.getSanitizeFieldNames(), key);
     }
 
+    protected static MessagingConfiguration.RabbitMQNamingMode getRabbitMQNamingMode() {
+        return messagingConfiguration.getRabbitMQNamingMode();
+    }
+
     /**
      * Captures queue name and optional timestamp
      *

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-spring/src/main/java/co/elastic/apm/agent/rabbitmq/SpringAmqpTransactionHelper.java

@@ -22,6 +22,7 @@
 import co.elastic.apm.agent.tracer.Transaction;
 import co.elastic.apm.agent.rabbitmq.header.SpringRabbitMQTextHeaderGetter;
 import co.elastic.apm.agent.sdk.internal.util.PrivilegedActionUtils;
+import co.elastic.apm.agent.tracer.configuration.MessagingConfiguration;
 import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 
@@ -37,12 +38,12 @@ public SpringAmqpTransactionHelper(Tracer tracer) {
 
     @Nullable
     public Transaction<?> createTransaction(Message message, String transactionNamePrefix) {
-        String exchange = null;
+        String exchangeOrQueue = null;
         MessageProperties messageProperties = message.getMessageProperties();
         if (messageProperties != null) {
-            exchange = messageProperties.getReceivedExchange();
+            exchangeOrQueue = getExchangeOrQueue(messageProperties);
         }
-        if (exchange != null && AbstractBaseInstrumentation.isIgnored(exchange)) {
+        if (exchangeOrQueue != null && AbstractBaseInstrumentation.isIgnored(exchangeOrQueue)) {
             return null;
         }
 
@@ -58,7 +59,7 @@ public Transaction<?> createTransaction(Message message, String transactionNameP
         transaction.withType("messaging")
             .withName(transactionNamePrefix)
             .appendToName(" RECEIVE from ")
-            .appendToName(AbstractBaseInstrumentation.normalizeExchangeName(exchange));
+            .appendToName(AbstractBaseInstrumentation.normalizeExchangeName(exchangeOrQueue));
 
         transaction.setFrameworkName("Spring AMQP");
 
@@ -67,12 +68,20 @@ public Transaction<?> createTransaction(Message message, String transactionNameP
             String receivedRoutingKey = messageProperties.getReceivedRoutingKey();
             transaction.getContext().getMessage().withAge(timestamp).withRoutingKey(receivedRoutingKey);
         }
-        if (exchange != null) {
-            transaction.getContext().getMessage().withQueue(exchange);
+        if (exchangeOrQueue != null) {
+            transaction.getContext().getMessage().withQueue(exchangeOrQueue);
         }
 
         // only capture incoming messages headers for now (consistent with other messaging plugins)
         AbstractBaseInstrumentation.captureHeaders(messageProperties != null ? messageProperties.getHeaders() : null, transaction.getContext().getMessage());
         return transaction.activate();
     }
+
+    private static String getExchangeOrQueue(MessageProperties messageProperties) {
+        if (MessagingConfiguration.RabbitMQNamingMode.QUEUE == AbstractBaseInstrumentation.getRabbitMQNamingMode()) {
+            return messageProperties.getConsumerQueue();
+        } else {
+            return messageProperties.getReceivedExchange();
+        }
+    }
 }

apm-agent-plugins/apm-rabbitmq/apm-rabbitmq-spring/src/test/java/co/elastic/apm/agent/rabbitmq/AbstractRabbitMqTest.java

@@ -20,6 +20,7 @@
 
 import co.elastic.apm.agent.impl.transaction.Span;
 import co.elastic.apm.agent.impl.transaction.Transaction;
+import co.elastic.apm.agent.tracer.configuration.MessagingConfiguration;
 import org.junit.Test;
 
 import java.util.List;
@@ -28,8 +29,11 @@
 import static co.elastic.apm.agent.rabbitmq.RabbitMQIT.checkSendSpan;
 import static co.elastic.apm.agent.rabbitmq.RabbitMQIT.checkTransaction;
 import static co.elastic.apm.agent.rabbitmq.RabbitMQIT.getNonRootTransaction;
+import static co.elastic.apm.agent.rabbitmq.TestConstants.QUEUE_NAME;
 import static co.elastic.apm.agent.rabbitmq.TestConstants.TOPIC_EXCHANGE_NAME;
+import static co.elastic.apm.agent.tracer.configuration.MessagingConfiguration.JmsStrategy.POLLING;
 import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+import static org.mockito.Mockito.doReturn;
 
 public abstract class AbstractRabbitMqTest extends RabbitMqTestBase {
 
@@ -46,6 +50,20 @@ public void verifyThatTransactionWithSpanCreated() {
         checkParentChild(receiveTransaction, testSpan);
     }
 
+    @Test
+    public void verifyThatTransactionWithSpanCreated_NamingModeQueue() {
+        MessagingConfiguration messagingConfiguration = config.getConfig(MessagingConfiguration.class);
+        doReturn(MessagingConfiguration.RabbitMQNamingMode.QUEUE).when(messagingConfiguration).getRabbitMQNamingMode();
+
+        rabbitTemplate.convertAndSend(TOPIC_EXCHANGE_NAME, TestConstants.ROUTING_KEY, MESSAGE);
+        Transaction receiveTransaction = reporter.getFirstTransaction(1000);
+        checkTransaction(receiveTransaction, QUEUE_NAME, "Spring AMQP");
+        Span testSpan = reporter.getFirstSpan(1000);
+        assertThat(testSpan.getNameAsString()).isEqualTo("testSpan");
+        assertThat(testSpan.getType()).isEqualTo("custom");
+        checkParentChild(receiveTransaction, testSpan);
+    }
+
     @Test
     public void verifyThatTransactionWithSpanCreated_DistributedTracing() {
         Transaction rootTransaction = startTestRootTransaction("Rabbit-Test Root Transaction");

apm-agent-tracer/src/main/java/co/elastic/apm/agent/tracer/configuration/MessagingConfiguration.java

@@ -19,7 +19,6 @@
 package co.elastic.apm.agent.tracer.configuration;
 
 import co.elastic.apm.agent.common.util.WildcardMatcher;
-import co.elastic.apm.agent.tracer.configuration.WildcardMatcherValueConverter;
 import org.stagemonitor.configuration.ConfigurationOption;
 import org.stagemonitor.configuration.ConfigurationOptionProvider;
 import org.stagemonitor.configuration.converter.ListValueConverter;
@@ -33,7 +32,7 @@ public class MessagingConfiguration extends ConfigurationOptionProvider {
     private static final String MESSAGE_POLLING_TRANSACTION_STRATEGY = "message_polling_transaction_strategy";
     private static final String MESSAGE_BATCH_STRATEGY = "message_batch_strategy";
 
-    private ConfigurationOption<JmsStrategy> messagePollingTransactionStrategy = ConfigurationOption.enumOption(JmsStrategy.class)
+    private final ConfigurationOption<JmsStrategy> messagePollingTransactionStrategy = ConfigurationOption.enumOption(JmsStrategy.class)
         .key(MESSAGE_POLLING_TRANSACTION_STRATEGY)
         .configurationCategory(MESSAGING_CATEGORY)
         .tags("internal")
@@ -45,7 +44,7 @@ public class MessagingConfiguration extends ConfigurationOptionProvider {
         .dynamic(true)
         .buildWithDefault(JmsStrategy.HANDLING);
 
-    private ConfigurationOption<BatchStrategy> messageBatchStrategy = ConfigurationOption.enumOption(BatchStrategy.class)
+    private final ConfigurationOption<BatchStrategy> messageBatchStrategy = ConfigurationOption.enumOption(BatchStrategy.class)
         .key(MESSAGE_BATCH_STRATEGY)
         .configurationCategory(MESSAGING_CATEGORY)
         .tags("internal")
@@ -57,7 +56,7 @@ public class MessagingConfiguration extends ConfigurationOptionProvider {
         .dynamic(true)
         .buildWithDefault(BatchStrategy.BATCH_HANDLING);
 
-    private ConfigurationOption<Boolean> collectQueueAddress = ConfigurationOption.booleanOption()
+    private final ConfigurationOption<Boolean> collectQueueAddress = ConfigurationOption.booleanOption()
         .key("collect_queue_address")
         .configurationCategory(MESSAGING_CATEGORY)
         .tags("internal")
@@ -109,6 +108,14 @@ public class MessagingConfiguration extends ConfigurationOptionProvider {
         .dynamic(false)
         .buildWithDefault(Collections.<String>emptyList());
 
+    private final ConfigurationOption<RabbitMQNamingMode> rabbitMQNamingMode = ConfigurationOption.enumOption(RabbitMQNamingMode.class)
+        .key("rabbitmq_naming_mode")
+        .configurationCategory(MESSAGING_CATEGORY)
+        .description("Defines whether the agent should use the exchanges or the queue for the naming of RabbitMQ Transactions. Valid options are `QUEUE` and `EXCHANGE`")
+        .dynamic(true)
+        .tags("added[1.46.0]")
+        .buildWithDefault(RabbitMQNamingMode.EXCHANGE);
+
     public JmsStrategy getMessagePollingTransactionStrategy() {
         return messagePollingTransactionStrategy.get();
     }
@@ -133,6 +140,10 @@ public Collection<String> getJmsListenerPackages() {
         return jmsListenerPackages.get();
     }
 
+    public RabbitMQNamingMode getRabbitMQNamingMode() {
+        return rabbitMQNamingMode.get();
+    }
+
     public enum JmsStrategy {
         /**
          * Create a transaction capturing JMS {@code receive} invocations
@@ -164,4 +175,15 @@ public enum BatchStrategy {
          */
         BATCH_HANDLING
     }
+
+    public enum RabbitMQNamingMode {
+        /**
+         * Use exchange in transaction names
+         */
+        EXCHANGE,
+        /**
+         * Use queue in transaction names
+         */
+        QUEUE,
+    }
 }

docs/configuration.asciidoc

@@ -171,6 +171,7 @@ Click on a key to get more information.
 * <<config-messaging>>
 ** <<config-ignore-message-queues>>
 ** <<config-jms-listener-packages>>
+** <<config-rabbitmq-naming-mode>>
 * <<config-metrics>>
 ** <<config-dedot-custom-metrics>>
 ** <<config-custom-metrics-histogram-boundaries>>
@@ -2504,6 +2505,30 @@ Starting from version 1.43.0, the classes that are part of the 'application_pack
 | `elastic.apm.jms_listener_packages` | `jms_listener_packages` | `ELASTIC_APM_JMS_LISTENER_PACKAGES`
 |============
 
+// This file is auto generated. Please make your changes in *Configuration.java (for example CoreConfiguration.java) and execute ConfigurationExporter
+[float]
+[[config-rabbitmq-naming-mode]]
+==== `rabbitmq_naming_mode` (added[1.46.0])
+
+Defines whether the agent should use the exchanges or the queue for the naming of RabbitMQ Transactions. Valid options are `QUEUE` and `EXCHANGE`
+
+<<configuration-dynamic, image:./images/dynamic-config.svg[] >>
+
+Valid options: `EXCHANGE`, `QUEUE`
+
+[options="header"]
+|============
+| Default                          | Type                | Dynamic
+| `EXCHANGE` | RabbitMQNamingMode | true
+|============
+
+
+[options="header"]
+|============
+| Java System Properties      | Property file   | Environment
+| `elastic.apm.rabbitmq_naming_mode` | `rabbitmq_naming_mode` | `ELASTIC_APM_RABBITMQ_NAMING_MODE`
+|============
+
 [[config-metrics]]
 === Metrics configuration options
 
@@ -4603,6 +4628,15 @@ Example: `5ms`.
 #
 # jms_listener_packages=
 
+# Defines whether the agent should use the exchanges or the queue for the naming of RabbitMQ Transactions. Valid options are `QUEUE` and `EXCHANGE`
+#
+# Valid options: EXCHANGE, QUEUE
+# This setting can be changed at runtime
+# Type: RabbitMQNamingMode
+# Default value: EXCHANGE
+#
+# rabbitmq_naming_mode=EXCHANGE
+
 ############################################
 # Metrics                                  #
 ############################################