From 1e68089526c93e5a03d72f54ead886e85e8b4228 Mon Sep 17 00:00:00 2001 From: Eric Firth Date: Tue, 30 Jun 2026 09:09:08 -0400 Subject: [PATCH 1/2] 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) --- .../amqp/RabbitChannelInstrumentation.java | 15 ++++++++++++--- .../src/test/groovy/RabbitMQTest.groovy | 4 ++-- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java index c8beb8c4863..98a194fbeca 100644 --- a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java +++ b/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; @@ -219,9 +220,17 @@ 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 ((exchange == null || exchange.isEmpty()) && hasRoutingKey) { + // Publishing to the default exchange: the routing key is the destination queue + // name, so record it as the topic (matching the consumer checkpoint). Without + // this the producer has neither a topic nor an exchange and shows up disconnected + // in the data streams map. + 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 = diff --git a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy index 407195ed08c..53a9545af6c 100644 --- a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy +++ b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy @@ -225,7 +225,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 +493,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 } From 9b275feafc8119925f7ed7b6f175e12214e16537 Mon Sep 17 00:00:00 2001 From: Eric Firth Date: Mon, 6 Jul 2026 09:34:44 -0400 Subject: [PATCH 2/2] 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 --- .../amqp/RabbitChannelInstrumentation.java | 10 ++--- .../src/test/groovy/RabbitMQTest.groovy | 44 +++++++++++++++++++ 2 files changed, 48 insertions(+), 6 deletions(-) diff --git a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java index 98a194fbeca..2780015a5bf 100644 --- a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java +++ b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/main/java/datadog/trace/instrumentation/rabbitmq/amqp/RabbitChannelInstrumentation.java @@ -208,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; @@ -222,11 +224,7 @@ public static void onEnter( } final boolean hasRoutingKey = routingKey != null && !routingKey.isEmpty(); DataStreamsTags tags; - if ((exchange == null || exchange.isEmpty()) && hasRoutingKey) { - // Publishing to the default exchange: the routing key is the destination queue - // name, so record it as the topic (matching the consumer checkpoint). Without - // this the producer has neither a topic nor an exchange and shows up disconnected - // in the data streams map. + if (isDefaultExchange && hasRoutingKey) { tags = create("rabbitmq", OUTBOUND, routingKey); } else { tags = createWithExchange("rabbitmq", OUTBOUND, exchange, hasRoutingKey); diff --git a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy b/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy index 53a9545af6c..966ab9d4b80 100644 --- a/dd-java-agent/instrumentation/rabbitmq-amqp-2.7/src/test/groovy/RabbitMQTest.groovy +++ b/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 @@ -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 -> ", false, span(0), operationForProducer()) + } + trace(1) { + rabbitSpan(it, "basic.get ", false, null, operationForConsumer()) + } + } + + cleanup: + removeSysConfig(RABBIT_PROPAGATION_DISABLED_QUEUES) + } + def rabbitSpan( TraceAssert trace, String resource,