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..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 @@ -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 = 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..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 @@ -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 -> ", 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,