Skip to content

Commit e984025

Browse files
mccullsclaude
andcommitted
Fix W3C baggage propagation across reactor-netty connect hand-off
The connect-span path carried only the active AgentSpan across the subscription -> I/O thread hand-off, dropping the rest of the Datadog Context (including baggage). Capture the full Context instead and build the continuation directly via Context.capture(). Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent 1358e8e commit e984025

3 files changed

Lines changed: 122 additions & 24 deletions

File tree

dd-java-agent/instrumentation/reactor-netty-1.0/src/main/java/datadog/trace/instrumentation/reactor/netty/CaptureConnectSpan.java

Lines changed: 7 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,24 @@
11
package datadog.trace.instrumentation.reactor.netty;
22

3-
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan;
4-
5-
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
3+
import datadog.context.Context;
64
import java.util.function.Function;
75
import reactor.core.publisher.Mono;
86
import reactor.netty.Connection;
97

108
public class CaptureConnectSpan
119
implements Function<Mono<? extends Connection>, Mono<? extends Connection>> {
1210

13-
static final String CONNECT_SPAN = "datadog.connect.span";
11+
static final String CONNECT_CONTEXT = "datadog.connect.context";
1412

1513
@Override
1614
public Mono<? extends Connection> apply(Mono<? extends Connection> mono) {
1715
return mono.contextWrite(
18-
context -> {
19-
final AgentSpan span = activeSpan();
20-
if (null != span) {
21-
return context.put(CONNECT_SPAN, span);
16+
reactorCtx -> {
17+
final Context context = Context.current();
18+
if (context != Context.root()) {
19+
return reactorCtx.put(CONNECT_CONTEXT, context);
2220
} else {
23-
return context;
21+
return reactorCtx;
2422
}
2523
});
2624
}
Original file line numberDiff line numberDiff line change
@@ -1,29 +1,29 @@
11
package datadog.trace.instrumentation.reactor.netty;
22

3-
import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.captureSpan;
43
import static datadog.trace.instrumentation.netty41.AttributeKeys.CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY;
5-
import static datadog.trace.instrumentation.reactor.netty.CaptureConnectSpan.CONNECT_SPAN;
4+
import static datadog.trace.instrumentation.reactor.netty.CaptureConnectSpan.CONNECT_CONTEXT;
65

6+
import datadog.context.Context;
77
import datadog.context.ContextContinuation;
8-
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
98
import java.util.function.BiConsumer;
109
import reactor.netty.Connection;
1110
import reactor.netty.http.client.HttpClientRequest;
1211

1312
public class TransferConnectSpan implements BiConsumer<HttpClientRequest, Connection> {
1413
@Override
15-
public void accept(HttpClientRequest httpClientRequest, Connection connection) {
16-
final AgentSpan span = httpClientRequest.currentContextView().getOrDefault(CONNECT_SPAN, null);
17-
final ContextContinuation continuation = null == span ? null : captureSpan(span);
18-
if (null != continuation) {
19-
ContextContinuation current =
20-
connection
21-
.channel()
22-
.attr(CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY)
23-
.getAndSet(continuation);
24-
if (null != current) {
25-
current.release();
26-
}
14+
public void accept(HttpClientRequest clientRequest, Connection connection) {
15+
final Context context = clientRequest.currentContextView().getOrDefault(CONNECT_CONTEXT, null);
16+
if (null == context) {
17+
return;
18+
}
19+
final ContextContinuation newContinuation = context.capture();
20+
ContextContinuation oldContinuation =
21+
connection
22+
.channel()
23+
.attr(CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY)
24+
.getAndSet(newContinuation);
25+
if (null != oldContinuation) {
26+
oldContinuation.release();
2727
}
2828
}
2929
}
Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
import static org.junit.jupiter.api.Assertions.assertNotNull;
2+
import static org.junit.jupiter.api.Assertions.assertTrue;
3+
4+
import com.sun.net.httpserver.HttpServer;
5+
import datadog.context.Context;
6+
import datadog.context.ContextScope;
7+
import datadog.trace.agent.test.AbstractInstrumentationTest;
8+
import datadog.trace.bootstrap.instrumentation.api.AgentScope;
9+
import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
10+
import datadog.trace.bootstrap.instrumentation.api.AgentTracer;
11+
import datadog.trace.bootstrap.instrumentation.api.Baggage;
12+
import java.io.IOException;
13+
import java.net.InetSocketAddress;
14+
import java.time.Duration;
15+
import java.util.Collections;
16+
import java.util.concurrent.Executors;
17+
import java.util.concurrent.atomic.AtomicReference;
18+
import org.junit.jupiter.api.AfterAll;
19+
import org.junit.jupiter.api.BeforeAll;
20+
import org.junit.jupiter.api.Test;
21+
import reactor.netty.http.client.HttpClient;
22+
23+
/**
24+
* Regression test for the W3C baggage header not propagating on outgoing Reactor Netty requests.
25+
*
26+
* <p>The bug: the connect-span path carried only {@code activeSpan()} across the subscription ->
27+
* I/O thread hand-off, dropping the rest of the Datadog {@code Context} (including baggage). The
28+
* outgoing request then had no baggage to inject, so the {@code baggage} header was silently
29+
* skipped. This test sets a baggage item in the active context, makes one outgoing request, and
30+
* asserts the server received the {@code baggage} header.
31+
*
32+
* <p>It is a <em>coupling</em> test: producer (sets baggage) -> Reactor Netty carrier -> consumer
33+
* (injects the header). The failure it guards against lives in the hand-off between integrations,
34+
* not in any one of them, so per-integration tests would not catch it.
35+
*/
36+
class ReactorNettyBaggagePropagationTest extends AbstractInstrumentationTest {
37+
38+
private static HttpServer mockServer;
39+
private static String baseUrl;
40+
private static final AtomicReference<String> capturedBaggage = new AtomicReference<>();
41+
42+
@BeforeAll
43+
static void startServer() throws IOException {
44+
mockServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0);
45+
mockServer.createContext(
46+
"/capture",
47+
exchange -> {
48+
capturedBaggage.set(exchange.getRequestHeaders().getFirst("baggage"));
49+
byte[] body = "ok".getBytes();
50+
exchange.sendResponseHeaders(200, body.length);
51+
exchange.getResponseBody().write(body);
52+
exchange.close();
53+
});
54+
mockServer.setExecutor(Executors.newCachedThreadPool());
55+
mockServer.start();
56+
baseUrl =
57+
"http://"
58+
+ mockServer.getAddress().getHostString()
59+
+ ":"
60+
+ mockServer.getAddress().getPort();
61+
}
62+
63+
@AfterAll
64+
static void stopServer() {
65+
if (mockServer != null) {
66+
mockServer.stop(0);
67+
mockServer = null;
68+
}
69+
}
70+
71+
@Test
72+
void baggageHeaderPropagatedOnOutgoingRequest() {
73+
capturedBaggage.set(null);
74+
Baggage baggage = Baggage.create(Collections.singletonMap("user.id", "abc123"));
75+
76+
AgentSpan span = AgentTracer.startSpan("test", "parent");
77+
try (AgentScope spanScope = AgentTracer.activateSpan(span)) {
78+
// Active context now carries both the span and the baggage — the exact shape the connect-span
79+
// path must carry across the subscription -> I/O thread hand-off.
80+
try (ContextScope baggageScope = Context.current().with(baggage).attach()) {
81+
HttpClient.create()
82+
.get()
83+
.uri(baseUrl + "/capture")
84+
.response()
85+
.block(Duration.ofSeconds(10));
86+
}
87+
} finally {
88+
span.finish();
89+
}
90+
91+
String header = capturedBaggage.get();
92+
assertNotNull(
93+
header,
94+
"outgoing request must carry a W3C 'baggage' header when baggage is in the active context;"
95+
+ " null means the connect-span path dropped the context (carried only the span)");
96+
assertTrue(
97+
header.contains("user.id=abc123"),
98+
"baggage header should contain the propagated item, was: " + header);
99+
}
100+
}

0 commit comments

Comments
 (0)