Skip to content

Commit e910ab5

Browse files
mccullsdevflow.devflow-routing-intake
mcculls
and
devflow.devflow-routing-intake
authored
Support W3C baggage propagation across reactor-netty connect hand-off (#12027)
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> Review feedback Merge remote-tracking branch 'origin/master' into mcculls/fix-reactor-w3c-propagation Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>
1 parent 77b35db commit e910ab5

3 files changed

Lines changed: 130 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
}

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

Lines changed: 15 additions & 15 deletions
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+
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
}

‎dd-java-agent/instrumentation/reactor-netty-1.0/src/test/java/ReactorNettyBaggagePropagationTest.java‎

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

0 commit comments

Comments
 (0)