From bc977094c956da9527d67151e631a0635a5a656f Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 12:11:36 -0700 Subject: [PATCH 1/3] test(netty-4.1): Add pipelining test --- .../server/NettyHttp11PipeliningTest.java | 223 ++++++++++++++++++ 1 file changed, 223 insertions(+) create mode 100644 dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttp11PipeliningTest.java diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttp11PipeliningTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttp11PipeliningTest.java new file mode 100644 index 00000000000..610fa95ddbe --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttp11PipeliningTest.java @@ -0,0 +1,223 @@ +package datadog.trace.instrumentation.netty41.server; + +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TraceAssertions.SORT_BY_START_TIME; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.assertions.TraceMatcher; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.FullHttpRequest; +import io.netty.handler.codec.http.HttpObjectAggregator; +import io.netty.handler.codec.http.HttpServerCodec; +import java.io.ByteArrayOutputStream; +import java.io.EOFException; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.regex.Pattern; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class NettyHttp11PipeliningTest extends AbstractInstrumentationTest { + + private static final String FIRST_PATH = "/pipelined/first"; + private static final String SECOND_PATH = "/pipelined/second"; + private static final String THIRD_PATH = "/pipelined/third"; + + private EventLoopGroup eventLoopGroup; + private PipeliningHandler handler; + private int port; + + @BeforeAll + void startServer() throws Exception { + eventLoopGroup = new NioEventLoopGroup(); + handler = new PipeliningHandler(3); + ServerBootstrap bootstrap = + new ServerBootstrap() + .group(eventLoopGroup) + .channel(NioServerSocketChannel.class) + .childHandler( + new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) { + ch.pipeline().addLast(new HttpServerCodec()); + ch.pipeline().addLast(new HttpObjectAggregator(65536)); + ch.pipeline().addLast(handler); + } + }); + Channel channel = bootstrap.bind(0).sync().channel(); + port = ((InetSocketAddress) channel.localAddress()).getPort(); + } + + @AfterAll + void stopServer() { + if (eventLoopGroup != null) { + eventLoopGroup.shutdownGracefully(); + } + } + + @Test + void createsServerSpanForEachPipelinedRequest() throws Exception { + try (Socket socket = new Socket("localhost", port)) { + socket.setSoTimeout(5000); + socket.getOutputStream().write(pipelinedRequests().getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + assertTrue( + handler.awaitAllRequestsReceived(), + "server did not receive all pipelined requests before responding"); + + handler.writeResponses(); + + assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); + assertEquals("response " + SECOND_PATH, readHttpResponseBody(socket.getInputStream())); + assertEquals("response " + THIRD_PATH, readHttpResponseBody(socket.getInputStream())); + } + + assertTraces( + SORT_BY_START_TIME, + serverTrace(FIRST_PATH), + serverTrace(SECOND_PATH), + serverTrace(THIRD_PATH)); + } + + private static String pipelinedRequests() { + return "GET " + + FIRST_PATH + + " HTTP/1.1\r\nHost: localhost\r\n\r\n" + + "GET " + + SECOND_PATH + + " HTTP/1.1\r\nHost: localhost\r\n\r\n" + + "GET " + + THIRD_PATH + + " HTTP/1.1\r\nHost: localhost\r\n\r\n"; + } + + private static TraceMatcher serverTrace(String path) { + return trace( + span() + .root() + .operationName(Pattern.compile("netty\\.request")) + .resourceName(Pattern.compile("GET " + Pattern.quote(path))) + .type("web")); + } + + private static String readHttpResponseBody(InputStream in) throws IOException { + String headers = readHeaders(in); + assertTrue(headers.startsWith("HTTP/1.1 200 "), "unexpected response: " + headers); + int contentLength = contentLength(headers); + byte[] body = new byte[contentLength]; + int read = 0; + while (read < contentLength) { + int count = in.read(body, read, contentLength - read); + if (count == -1) { + throw new EOFException("response ended before body was complete"); + } + read += count; + } + return new String(body, UTF_8); + } + + private static String readHeaders(InputStream in) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + int state = 0; + while (state < 4) { + int b = in.read(); + if (b == -1) { + throw new EOFException("response ended before headers were complete"); + } + out.write(b); + if ((state == 0 || state == 2) && b == '\r') { + state++; + } else if ((state == 1 || state == 3) && b == '\n') { + state++; + } else { + state = b == '\r' ? 1 : 0; + } + } + return out.toString(US_ASCII.name()); + } + + private static int contentLength(String headers) { + for (String line : headers.split("\r\n")) { + int separator = line.indexOf(':'); + if (separator > 0 && "content-length".equalsIgnoreCase(line.substring(0, separator))) { + return Integer.parseInt(line.substring(separator + 1).trim()); + } + } + throw new AssertionError("missing content-length header: " + headers); + } + + private static final class PipeliningHandler + extends SimpleChannelInboundHandler { + private final CountDownLatch receivedRequests; + private final List paths = new ArrayList<>(); + private volatile ChannelHandlerContext context; + + private PipeliningHandler(int expectedRequests) { + receivedRequests = new CountDownLatch(expectedRequests); + } + + @Override + protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) { + context = ctx; + synchronized (paths) { + paths.add(request.uri()); + } + receivedRequests.countDown(); + } + + private boolean awaitAllRequestsReceived() throws InterruptedException { + return receivedRequests.await(5, SECONDS); + } + + private void writeResponses() { + ChannelHandlerContext responseContext = context; + if (responseContext == null) { + throw new IllegalStateException("no request context captured"); + } + List responsePaths; + synchronized (paths) { + responsePaths = new ArrayList<>(paths); + } + responseContext + .executor() + .execute( + () -> { + for (String path : responsePaths) { + byte[] body = ("response " + path).getBytes(UTF_8); + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.wrappedBuffer(body)); + response.headers().set(CONTENT_LENGTH, body.length); + responseContext.write(response); + } + responseContext.flush(); + }); + } + } +} From 7f7fdf417c2bc8e60c646e038b8040aa78132211 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 15:21:23 -0700 Subject: [PATCH 2/3] fix(netty-4.1): pipelined response context tracking --- .../ChannelFutureListenerInstrumentation.java | 1 + ...yChannelHandlerContextInstrumentation.java | 6 +- .../NettyChannelPipelineInstrumentation.java | 1 + .../HttpServerRequestTracingHandler.java | 30 +++-- .../HttpServerResponseTracingHandler.java | 16 ++- .../server/MaybeBlockResponseHandler.java | 34 ++--- .../netty41/AttributeKeys.java | 12 +- .../netty41/ServerRequestContext.java | 121 ++++++++++++++++++ 8 files changed, 162 insertions(+), 59 deletions(-) create mode 100644 dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/ChannelFutureListenerInstrumentation.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/ChannelFutureListenerInstrumentation.java index 85fe3902fc0..26754d69a6f 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/ChannelFutureListenerInstrumentation.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/ChannelFutureListenerInstrumentation.java @@ -46,6 +46,7 @@ public ElementMatcher hierarchyMatcher() { public String[] helperClassNames() { return new String[] { packageName + ".AttributeKeys", + packageName + ".ServerRequestContext", // client helpers packageName + ".client.NettyHttpClientDecorator", packageName + ".client.NettyResponseInjectAdapter", diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelHandlerContextInstrumentation.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelHandlerContextInstrumentation.java index 62363fda0cc..b73e4e0fbf7 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelHandlerContextInstrumentation.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelHandlerContextInstrumentation.java @@ -14,7 +14,6 @@ import static net.bytebuddy.matcher.ElementMatchers.isPublic; import com.google.auto.service.AutoService; -import datadog.context.Context; import datadog.trace.agent.tooling.Instrumenter; import datadog.trace.agent.tooling.InstrumenterModule; import datadog.trace.bootstrap.instrumentation.api.AgentScope; @@ -48,6 +47,7 @@ public ElementMatcher hierarchyMatcher() { public String[] helperClassNames() { return new String[] { packageName + ".AttributeKeys", + packageName + ".ServerRequestContext", packageName + ".client.NettyHttpClientDecorator", packageName + ".server.ResponseExtractAdapter", packageName + ".server.NettyHttpServerDecorator", @@ -71,8 +71,8 @@ public void methodAdvice(MethodTransformer transformer) { public static class FireAdvice { @Advice.OnMethodEnter(suppress = Throwable.class) public static AgentScope scopeSpan(@Advice.This final ChannelHandlerContext ctx) { - final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get(); - final AgentSpan channelSpan = spanFromContext(storedContext); + final AgentSpan channelSpan = + spanFromContext(ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get()); if (channelSpan == null || channelSpan == activeSpan()) { // don't modify the scope return noopScope(); diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelPipelineInstrumentation.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelPipelineInstrumentation.java index 7fce2fd12eb..668077b87a6 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelPipelineInstrumentation.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/NettyChannelPipelineInstrumentation.java @@ -72,6 +72,7 @@ public ElementMatcher hierarchyMatcher() { public String[] helperClassNames() { return new String[] { packageName + ".AttributeKeys", + packageName + ".ServerRequestContext", // client helpers packageName + ".client.NettyHttpClientDecorator", packageName + ".client.NettyResponseInjectAdapter", diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java index ce58fac2ded..fbf2c689360 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java @@ -1,22 +1,21 @@ package datadog.trace.instrumentation.netty41.server; -import static datadog.trace.instrumentation.netty41.AttributeKeys.ANALYZED_RESPONSE_KEY; -import static datadog.trace.instrumentation.netty41.AttributeKeys.BLOCKED_RESPONSE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.PARENT_CONTEXT_ATTRIBUTE_KEY; -import static datadog.trace.instrumentation.netty41.AttributeKeys.REQUEST_HEADERS_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE; import datadog.context.Context; import datadog.context.ContextScope; import datadog.trace.api.gateway.Flow; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.instrumentation.netty41.ServerRequestContext; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.HttpRequest; +import java.util.Deque; @ChannelHandler.Sharable public class HttpServerRequestTracingHandler extends ChannelInboundHandlerAdapter { @@ -49,11 +48,8 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) { DECORATE.afterStart(span); DECORATE.onRequest(span, channel, request, parentContext); - channel.attr(ANALYZED_RESPONSE_KEY).set(null); - channel.attr(BLOCKED_RESPONSE_KEY).set(null); - - channel.attr(CONTEXT_ATTRIBUTE_KEY).set(context); - channel.attr(REQUEST_HEADERS_ATTRIBUTE_KEY).set(request.headers()); + final ServerRequestContext serverContext = + ServerRequestContext.add(channel, context, request.headers()); Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction(); if (rba != null) { @@ -76,7 +72,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) { DECORATE.onError(span, throwable); DECORATE.beforeFinish(ignored.context()); span.finish(); // Finish the span manually since finishSpanOnClose was false - ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove(); + ServerRequestContext.remove(ctx.channel(), serverContext); throw throwable; } } @@ -88,11 +84,17 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); } finally { try { - final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).getAndRemove(); - final AgentSpan span = AgentSpan.fromContext(storedContext); - if (span != null && span.phasedFinish()) { - // at this point we can just publish this span to avoid loosing the rest of the trace - span.publish(); + final Deque storedContexts = + ServerRequestContext.removeAll(ctx.channel()); + if (storedContexts != null) { + ServerRequestContext storedContext; + while ((storedContext = storedContexts.pollFirst()) != null) { + final AgentSpan span = AgentSpan.fromContext(storedContext.tracingContext()); + if (span != null && span.phasedFinish()) { + // at this point we can just publish this span to avoid losing the rest of the trace + span.publish(); + } + } } } catch (final Throwable ignored) { } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java index 04d7f8893fc..11ea5475e1e 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java @@ -1,6 +1,5 @@ package datadog.trace.instrumentation.netty41.server; -import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.WEBSOCKET_SENDER_HANDLER_CONTEXT; import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE; @@ -8,6 +7,7 @@ import datadog.context.ContextScope; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext; +import datadog.trace.instrumentation.netty41.ServerRequestContext; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; @@ -22,10 +22,16 @@ public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdap @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) { - final Context storedContext = ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).get(); + if (!(msg instanceof HttpResponse)) { + ctx.write(msg, prm); + return; + } + + final ServerRequestContext serverContext = ServerRequestContext.nextResponse(ctx.channel()); + final Context storedContext = serverContext == null ? null : serverContext.tracingContext(); final AgentSpan span = AgentSpan.fromContext(storedContext); - if (span == null || !(msg instanceof HttpResponse)) { + if (span == null) { ctx.write(msg, prm); return; } @@ -39,7 +45,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann DECORATE.onError(span, throwable); span.setHttpStatusCode(500); span.finish(); // Finish the span manually since finishSpanOnClose was false - ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove(); + ServerRequestContext.remove(ctx.channel(), serverContext); throw throwable; } final boolean isWebsocketUpgrade = @@ -55,7 +61,7 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann DECORATE.onResponse(span, response); DECORATE.beforeFinish(scope.context()); span.finish(); // Finish the span manually since finishSpanOnClose was false - ctx.channel().attr(CONTEXT_ATTRIBUTE_KEY).remove(); + ServerRequestContext.remove(ctx.channel(), serverContext); } } } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandler.java index 4f79bd6aea6..172c838b52f 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandler.java @@ -1,9 +1,5 @@ package datadog.trace.instrumentation.netty41.server; -import static datadog.trace.instrumentation.netty41.AttributeKeys.ANALYZED_RESPONSE_KEY; -import static datadog.trace.instrumentation.netty41.AttributeKeys.BLOCKED_RESPONSE_KEY; -import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; -import static datadog.trace.instrumentation.netty41.AttributeKeys.REQUEST_HEADERS_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE; import static io.netty.handler.codec.http.HttpHeaders.setContentLength; @@ -13,6 +9,7 @@ import datadog.trace.api.gateway.RequestContext; import datadog.trace.bootstrap.blocking.BlockingActionHelper; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.instrumentation.netty41.ServerRequestContext; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; @@ -36,27 +33,12 @@ public class MaybeBlockResponseHandler extends ChannelOutboundHandlerAdapter { private MaybeBlockResponseHandler() {} - private static boolean isAnalyzedResponse(Channel ch) { - return ch.attr(ANALYZED_RESPONSE_KEY).get() != null; - } - - private static void markAnalyzedResponse(Channel ch) { - ch.attr(ANALYZED_RESPONSE_KEY).set(Boolean.TRUE); - } - - private static boolean isBlockedResponse(Channel ch) { - return ch.attr(BLOCKED_RESPONSE_KEY).get() != null; - } - - private static void markBlockedResponse(Channel ch) { - ch.attr(BLOCKED_RESPONSE_KEY).set(Boolean.TRUE); - } - @Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) throws Exception { Channel channel = ctx.channel(); - Context storedContext = channel.attr(CONTEXT_ATTRIBUTE_KEY).get(); + ServerRequestContext serverContext = ServerRequestContext.nextResponse(channel); + Context storedContext = serverContext == null ? null : serverContext.tracingContext(); AgentSpan span = AgentSpan.fromContext(storedContext); RequestContext requestContext; if (span == null || (requestContext = span.getRequestContext()) == null) { @@ -64,8 +46,8 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr return; } - if (isAnalyzedResponse(channel)) { - if (isBlockedResponse(channel)) { + if (serverContext.isResponseAnalyzed()) { + if (serverContext.isResponseBlocked()) { // block further writes log.debug("Write suppressed, msg {} dropped", msg); ReferenceCountUtil.release(msg); @@ -88,14 +70,14 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr Flow flow = DECORATE.callIGCallbackResponseAndHeaders( span, origResponse, origResponse.getStatus().code(), ResponseExtractAdapter.GETTER); - markAnalyzedResponse(channel); + serverContext.markResponseAnalyzed(); Flow.Action action = flow.getAction(); if (!(action instanceof Flow.Action.RequestBlockingAction)) { super.write(ctx, msg, prm); return; } - markBlockedResponse(channel); + serverContext.markResponseBlocked(); Flow.Action.RequestBlockingAction rba = (Flow.Action.RequestBlockingAction) action; int httpCode = BlockingActionHelper.getHttpCode(rba.getStatusCode()); HttpResponseStatus httpResponseStatus = HttpResponseStatus.valueOf(httpCode); @@ -112,7 +94,7 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr BlockingContentType bct = rba.getBlockingContentType(); if (bct != BlockingContentType.NONE) { - HttpHeaders reqHeaders = ctx.attr(REQUEST_HEADERS_ATTRIBUTE_KEY).get(); + HttpHeaders reqHeaders = serverContext.requestHeaders(); String acceptHeader = reqHeaders != null ? reqHeaders.get("accept") : null; BlockingActionHelper.TemplateType type = BlockingActionHelper.determineTemplateType(bct, acceptHeader); diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java index 24e34d48ab5..307002a68fb 100644 --- a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/AttributeKeys.java @@ -7,7 +7,6 @@ import datadog.trace.bootstrap.instrumentation.api.AgentScope; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext; -import io.netty.handler.codec.http.HttpHeaders; import io.netty.util.AttributeKey; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -33,15 +32,6 @@ public final class AttributeKeys { public static final AttributeKey PARENT_CONTEXT_ATTRIBUTE_KEY = attributeKey("datadog.server.parent-context"); - public static final AttributeKey REQUEST_HEADERS_ATTRIBUTE_KEY = - attributeKey("datadog.server.request.headers"); - - public static final AttributeKey ANALYZED_RESPONSE_KEY = - attributeKey("datadog.server.analyzed_response"); - - public static final AttributeKey BLOCKED_RESPONSE_KEY = - attributeKey("datadog.server.blocked_response"); - public static final AttributeKey WEBSOCKET_SENDER_HANDLER_CONTEXT = attributeKey("datadog.server.websocket.sender.handler_context"); @@ -56,7 +46,7 @@ public final class AttributeKeys { * cassandra driver. */ @SuppressWarnings("unchecked") - private static AttributeKey attributeKey(final String key) { + static AttributeKey attributeKey(final String key) { ConcurrentMap> map = MAPS.get(AttributeKey.class); AttributeKey attributeKey = (AttributeKey) map.get(key); if (null == attributeKey) { diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java new file mode 100644 index 00000000000..1a780668312 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java @@ -0,0 +1,121 @@ +package datadog.trace.instrumentation.netty41; + +import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; + +import datadog.context.Context; +import io.netty.handler.codec.http.HttpHeaders; +import io.netty.util.AttributeKey; +import io.netty.util.AttributeMap; +import java.util.ArrayDeque; +import java.util.Deque; + +/** Per-request server state stored on the channel until the matching response is written. */ +public final class ServerRequestContext { + + /** Pending server request contexts for a channel. */ + private static final AttributeKey> + SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY = + AttributeKeys.attributeKey("datadog.server.request.contexts"); + + /** Adds a request context to the queue tail. */ + public static ServerRequestContext add( + final AttributeMap attributes, final Context context, final HttpHeaders requestHeaders) { + final ServerRequestContext serverContext = new ServerRequestContext(context, requestHeaders); + getOrCreate(attributes).addLast(serverContext); + // The deque is authoritative for server request/response matching. CONTEXT_ATTRIBUTE_KEY is a + // legacy mirror of the current inbound request used by generic fire* activation. + attributes.attr(CONTEXT_ATTRIBUTE_KEY).set(context); + return serverContext; + } + + /** Returns the server request context for the next response. */ + public static ServerRequestContext nextResponse(final AttributeMap attributes) { + final Deque contexts = + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); + // HTTP/1.1 responses are written in request order, including when requests are pipelined on one + // connection. + return contexts == null ? null : contexts.peekFirst(); + } + + /** Removes a completed or failed request context. */ + public static void remove( + final AttributeMap attributes, final ServerRequestContext serverContext) { + if (serverContext == null) { + return; + } + final Deque contexts = + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); + if (contexts != null) { + if (contexts.peekFirst() == serverContext) { + // Response completion consumes the queue head. + contexts.pollFirst(); + } else { + // Request-side failures normally remove the tail. Remove by value to cover later cleanup + // after additional pipelined requests were queued. + contexts.remove(serverContext); + } + final ServerRequestContext currentContext = contexts.peekLast(); + if (currentContext == null) { + attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); + } else { + // Keep the legacy mirror pointed at the current inbound request after removing an older + // response context. + attributes.attr(CONTEXT_ATTRIBUTE_KEY).set(currentContext.tracingContext()); + } + } + } + + /** Removes all pending request contexts on channel close. */ + public static Deque removeAll(final AttributeMap attributes) { + // The legacy mirror must not outlive the authoritative request queue. + attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); + return attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove(); + } + + /** Creates the per-channel server request context queue. */ + private static Deque getOrCreate(final AttributeMap attributes) { + Deque contexts = + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); + if (contexts == null) { + // Netty serializes handler callbacks for a channel on its EventLoop, so this queue does not + // need the allocation and atomic overhead of a concurrent collection. + contexts = new ArrayDeque<>(); + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).set(contexts); + } + return contexts; + } + + private final Context tracingContext; + private final HttpHeaders requestHeaders; + private boolean responseAnalyzed; + private boolean responseBlocked; + + public Context tracingContext() { + return tracingContext; + } + + public HttpHeaders requestHeaders() { + return requestHeaders; + } + + public boolean isResponseAnalyzed() { + return responseAnalyzed; + } + + public void markResponseAnalyzed() { + responseAnalyzed = true; + } + + public boolean isResponseBlocked() { + return responseBlocked; + } + + public void markResponseBlocked() { + responseBlocked = true; + } + + private ServerRequestContext(final Context tracingContext, final HttpHeaders requestHeaders) { + this.tracingContext = tracingContext; + this.requestHeaders = requestHeaders; + } +} From 57f9d65709b10c0ec26e7b39e0ee6a8463a6e8bc Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 19:32:26 -0700 Subject: [PATCH 3/3] Disable Netty server tracing on context queue overflow --- .../HttpServerRequestTracingHandler.java | 20 ++--- .../netty41/ServerRequestContextTest.java | 28 ++++++ .../netty41/ServerRequestContext.java | 86 ++++++++++++++++--- 3 files changed, 111 insertions(+), 23 deletions(-) create mode 100644 dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/ServerRequestContextTest.java diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java index fbf2c689360..7e7dfcb1e89 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java @@ -15,7 +15,6 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.HttpRequest; -import java.util.Deque; @ChannelHandler.Sharable public class HttpServerRequestTracingHandler extends ChannelInboundHandlerAdapter { @@ -37,6 +36,12 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) { } final HttpRequest request = (HttpRequest) msg; + if (!ServerRequestContext.canTrackRequest(channel)) { + channel.attr(PARENT_CONTEXT_ATTRIBUTE_KEY).remove(); + ctx.fireChannelRead(msg); + return; + } + final HttpHeaders headers = request.headers(); final Context storedParentContext = channel.attr(PARENT_CONTEXT_ATTRIBUTE_KEY).getAndRemove(); final Context parentContext = @@ -84,18 +89,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); } finally { try { - final Deque storedContexts = - ServerRequestContext.removeAll(ctx.channel()); - if (storedContexts != null) { - ServerRequestContext storedContext; - while ((storedContext = storedContexts.pollFirst()) != null) { - final AgentSpan span = AgentSpan.fromContext(storedContext.tracingContext()); - if (span != null && span.phasedFinish()) { - // at this point we can just publish this span to avoid losing the rest of the trace - span.publish(); - } - } - } + ServerRequestContext.closeAll(ctx.channel()); } catch (final Throwable ignored) { } } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/ServerRequestContextTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/ServerRequestContextTest.java new file mode 100644 index 00000000000..cc2896bde03 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/ServerRequestContextTest.java @@ -0,0 +1,28 @@ +package datadog.trace.instrumentation.netty41; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +import datadog.context.Context; +import io.netty.handler.codec.http.DefaultHttpHeaders; +import io.netty.util.DefaultAttributeMap; +import org.junit.jupiter.api.Test; + +class ServerRequestContextTest { + private static final int PIPELINING_LIMIT = 1000; + + @Test + void disablesTrackingWhenPipeliningLimitIsExceeded() { + DefaultAttributeMap attributes = new DefaultAttributeMap(); + + for (int i = 0; i < PIPELINING_LIMIT; i++) { + assertNotNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders())); + } + + assertFalse(ServerRequestContext.canTrackRequest(attributes)); + assertNull(ServerRequestContext.nextResponse(attributes)); + assertNull(attributes.attr(AttributeKeys.CONTEXT_ATTRIBUTE_KEY).get()); + assertNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders())); + } +} diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java index 1a780668312..4a8ea3e59e7 100644 --- a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java @@ -3,25 +3,33 @@ import static datadog.trace.instrumentation.netty41.AttributeKeys.CONTEXT_ATTRIBUTE_KEY; import datadog.context.Context; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import io.netty.handler.codec.http.HttpHeaders; import io.netty.util.AttributeKey; import io.netty.util.AttributeMap; import java.util.ArrayDeque; import java.util.Deque; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** Per-request server state stored on the channel until the matching response is written. */ public final class ServerRequestContext { - - /** Pending server request contexts for a channel. */ - private static final AttributeKey> - SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY = - AttributeKeys.attributeKey("datadog.server.request.contexts"); + /** Returns whether a new server request can be tracked on this channel. */ + public static boolean canTrackRequest(final AttributeMap attributes) { + final Deque contexts = + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); + return contexts == null || canAdd(attributes, contexts); + } /** Adds a request context to the queue tail. */ public static ServerRequestContext add( final AttributeMap attributes, final Context context, final HttpHeaders requestHeaders) { + final Deque contexts = getOrCreate(attributes); + if (!canAdd(attributes, contexts)) { + return null; + } final ServerRequestContext serverContext = new ServerRequestContext(context, requestHeaders); - getOrCreate(attributes).addLast(serverContext); + contexts.addLast(serverContext); // The deque is authoritative for server request/response matching. CONTEXT_ATTRIBUTE_KEY is a // legacy mirror of the current inbound request used by generic fire* activation. attributes.attr(CONTEXT_ATTRIBUTE_KEY).set(context); @@ -34,7 +42,7 @@ public static ServerRequestContext nextResponse(final AttributeMap attributes) { attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); // HTTP/1.1 responses are written in request order, including when requests are pipelined on one // connection. - return contexts == null ? null : contexts.peekFirst(); + return contexts == null || isPoisoned(contexts) ? null : contexts.peekFirst(); } /** Removes a completed or failed request context. */ @@ -46,6 +54,9 @@ public static void remove( final Deque contexts = attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).get(); if (contexts != null) { + if (isPoisoned(contexts)) { + return; + } if (contexts.peekFirst() == serverContext) { // Response completion consumes the queue head. contexts.pollFirst(); @@ -65,13 +76,24 @@ public static void remove( } } - /** Removes all pending request contexts on channel close. */ - public static Deque removeAll(final AttributeMap attributes) { + /** Closes all pending request contexts on channel close. */ + public static void closeAll(final AttributeMap attributes) { // The legacy mirror must not outlive the authoritative request queue. attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); - return attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove(); + close(attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove()); } + private static final int PIPELINING_LIMIT = 1000; + + private static final Logger log = LoggerFactory.getLogger(ServerRequestContext.class); + + /** Pending server request contexts for a channel. */ + private static final AttributeKey> + SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY = + AttributeKeys.attributeKey("datadog.server.request.contexts"); + + private static final Deque POISONED_CONTEXTS = new ArrayDeque<>(0); + /** Creates the per-channel server request context queue. */ private static Deque getOrCreate(final AttributeMap attributes) { Deque contexts = @@ -85,6 +107,50 @@ private static Deque getOrCreate(final AttributeMap attrib return contexts; } + private static void close(final Deque contexts) { + if (contexts == null || isPoisoned(contexts)) { + return; + } + ServerRequestContext context; + while ((context = contexts.pollFirst()) != null) { + try { + final AgentSpan span = AgentSpan.fromContext(context.tracingContext()); + if (span != null && span.phasedFinish()) { + // These contexts no longer have a response handler path that can finish the span. + span.publish(); + } + } catch (final Throwable ignored) { + } + } + } + + private static boolean canAdd( + final AttributeMap attributes, final Deque contexts) { + if (isPoisoned(contexts)) { + return false; + } + // If this limit is exceeded, stop tracing on the channel and drain the deque. This suggests + // contexts are not being removed, for example because the server stopped writing responses. + if (contexts.size() >= PIPELINING_LIMIT) { + final int pendingContexts = contexts.size(); + close(contexts); + attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).set(POISONED_CONTEXTS); + attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); + log.error( + "Too many pending Netty server request contexts on a channel; " + + "closing {} contexts and disabling Netty server tracing on that channel " + + "(limit: {})", + pendingContexts, + PIPELINING_LIMIT); + return false; + } + return true; + } + + private static boolean isPoisoned(final Deque contexts) { + return contexts == POISONED_CONTEXTS; + } + private final Context tracingContext; private final HttpHeaders requestHeaders; private boolean responseAnalyzed;