Use routing key as DSM topic for RabbitMQ default-exchange publishes (#11805)

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

When a producer publishes to the default exchange (exchange == ""), the
routing key is the destination queue name. The DSM checkpoint previously
recorded an empty exchange and no topic, so the producer had no destination
and showed up disconnected in the Data Streams Monitoring map. Record the
routing key as the topic in that case (matching the consumer checkpoint and
the JS/.NET tracers). Named-exchange publishes are unchanged.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Fix propagation-disabled check for default-exchange RabbitMQ publishes

The disabled-destination check still looked up propagation state by
exchange name, which is always empty for default-exchange publishes.
Use the routing key (queue name) instead, matching the DSM topic tag
and the consumer-side check, so disabling a queue by name also
suppresses producer-side propagation for default-exchange publishes.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Merge branch 'master' into eric.firth/dsm-rabbitmq-default-exchange-topic

Merge branch 'master' into eric.firth/dsm-rabbitmq-default-exchange-topic

Merge branch 'master' into eric.firth/dsm-rabbitmq-default-exchange-topic

Merge branch 'master' into eric.firth/dsm-rabbitmq-default-exchange-topic

Merge branch 'master' into eric.firth/dsm-rabbitmq-default-exchange-topic

Co-authored-by: eric.firth <eric.firth@datadoghq.com>

Author: ericfirth

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,