Use routing key as DSM topic for RabbitMQ default-exchange publishes

dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java

@@ -8,6 +8,7 @@
 import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
 import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.namedOneOf;
 import static datadog.trace.api.datastreams.DataStreamsTags.Direction.OUTBOUND;
+import static datadog.trace.api.datastreams.DataStreamsTags.create;
 import static datadog.trace.api.datastreams.DataStreamsTags.createWithExchange;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan;
 import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan;
@@ -207,8 +208,10 @@ public static void onEnter(
       AgentSpan span = activeSpan();
       if (span == null) return;
       Config config = Config.get();
+      final boolean isDefaultExchange = exchange == null || exchange.isEmpty();
+      final String destination = isDefaultExchange ? routingKey : exchange;
       if (!config.isRabbitPropagationEnabled()
-          || config.isRabbitPropagationDisabledForDestination(exchange)) return;
+          || config.isRabbitPropagationDisabledForDestination(destination)) return;
       // This is the internal behavior when props are null.  We're just doing it earlier now.
       if (props == null) {
         props = MessageProperties.MINIMAL_BASIC;
@@ -219,9 +222,13 @@ public static void onEnter(
       if (TIME_IN_QUEUE_ENABLED) {
         RabbitDecorator.injectTimeInQueueStart(headers);
       }
-      DataStreamsTags tags =
-          createWithExchange(
-              "rabbitmq", OUTBOUND, exchange, routingKey != null && !routingKey.isEmpty());
+      final boolean hasRoutingKey = routingKey != null && !routingKey.isEmpty();
+      DataStreamsTags tags;
+      if (isDefaultExchange && hasRoutingKey) {
+        tags = create("rabbitmq", OUTBOUND, routingKey);
+      } else {
+        tags = createWithExchange("rabbitmq", OUTBOUND, exchange, hasRoutingKey);
+      }
       DataStreamsContext dsmContext = DataStreamsContext.fromTags(tags);
       defaultPropagator().inject(span.with(dsmContext), headers, SETTER);
       props =

dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy

@@ -14,6 +14,7 @@ import datadog.trace.agent.test.utils.PortUtils
 import datadog.trace.api.Config
 import datadog.trace.api.DDSpanTypes
 import datadog.trace.api.DDTags
+import datadog.trace.api.datastreams.PathwayContext
 import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags
 import datadog.trace.bootstrap.instrumentation.api.Tags
 import datadog.trace.core.DDSpan
@@ -225,7 +226,7 @@ abstract class RabbitMQTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags.hasAllTags("direction:out", "exchange:", "has_routing_key:true", "type:rabbitmq")
+        tags.hasAllTags("direction:out", "topic:" + queueName, "type:rabbitmq")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
@@ -493,7 +494,7 @@ abstract class RabbitMQTestBase extends VersionedNamingTestBase {
     if (isDataStreamsEnabled()) {
       StatsGroup first = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == 0 }
       verifyAll(first) {
-        tags.hasAllTags("direction:out", "exchange:", "has_routing_key:true", "type:rabbitmq")
+        tags.hasAllTags("direction:out", "topic:some-routing-queue", "type:rabbitmq")
       }
 
       StatsGroup second = TEST_DATA_STREAMS_WRITER.groups.find { it.parentHash == first.hash }
@@ -689,6 +690,49 @@ abstract class RabbitMQTestBase extends VersionedNamingTestBase {
     "deliver" | "some-exchange" | "some-routing-key" | "queueNameTest" | ""              | false
   }
 
+  def "test rabbit publish to default exchange with queue name in disabled queues (producer side)"() {
+    setup:
+    removeSysConfig(RABBIT_PROPAGATION_DISABLED_QUEUES)
+    def queueName = channel.queueDeclare().getQueue()
+    injectSysConfig(RABBIT_PROPAGATION_DISABLED_QUEUES, queueName)
+
+    when:
+    runUnderTrace("parent") {
+      channel.basicPublish("", queueName, null, "Hello, world!".bytes)
+    }
+    GetResponse response = channel.basicGet(queueName, true)
+    String body = new String(response.body)
+
+    then:
+    body == "Hello, world!"
+
+    and:
+    // Publishing to the default exchange uses the routing key (i.e. the queue name) as the
+    // DSM destination, so a queue name in RABBIT_PROPAGATION_DISABLED_QUEUES must suppress
+    // the pathway header, the same way it does for named-exchange publishes.
+    if (isDataStreamsEnabled()) {
+      def headers = response.getProps().getHeaders()
+      assert headers == null || !headers.containsKey(PathwayContext.PROPAGATION_KEY_BASE64)
+    }
+
+    and:
+    assertTraces(3, SORT_TRACES_BY_ID) {
+      trace(1) {
+        rabbitSpan(it, "queue.declare")
+      }
+      trace(2) {
+        basicSpan(it, "parent")
+        rabbitSpan(it, "basic.publish <default> -> <generated>", false, span(0), operationForProducer())
+      }
+      trace(1) {
+        rabbitSpan(it, "basic.get <generated>", false, null, operationForConsumer())
+      }
+    }
+
+    cleanup:
+    removeSysConfig(RABBIT_PROPAGATION_DISABLED_QUEUES)
+  }
+
   def rabbitSpan(
     TraceAssert trace,
     String resource,