From b3f85fbecfd3de61176387420a36f1da8bd0dcaa Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 15:56:21 -0700 Subject: [PATCH 1/3] Add Netty chunked response span timing tests --- .../server/NettyChunkedResponseSpanTest.java | 417 ++++++++++++++++++ .../server/NettyHttp11PipeliningTest.java | 99 +---- .../server/NettyHttpServerTestSupport.java | 159 +++++++ 3 files changed, 584 insertions(+), 91 deletions(-) create mode 100644 dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java create mode 100644 dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttpServerTestSupport.java diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java new file mode 100644 index 00000000000..a48a9b2e9d2 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -0,0 +1,417 @@ +package datadog.trace.instrumentation.netty41.server; + +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static io.netty.handler.codec.http.HttpHeaderNames.CONNECTION; +import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH; +import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING; +import static io.netty.handler.codec.http.HttpHeaderNames.UPGRADE; +import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED; +import static io.netty.handler.codec.http.HttpResponseStatus.CONTINUE; +import static io.netty.handler.codec.http.HttpResponseStatus.NOT_MODIFIED; +import static io.netty.handler.codec.http.HttpResponseStatus.NO_CONTENT; +import static io.netty.handler.codec.http.HttpResponseStatus.OK; +import static io.netty.handler.codec.http.HttpResponseStatus.RESET_CONTENT; +import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS; +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.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.agent.test.assertions.SpanMatcher; +import io.netty.buffer.Unpooled; +import io.netty.channel.Channel; +import io.netty.channel.ChannelHandler; +import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.handler.codec.http.DefaultFullHttpResponse; +import io.netty.handler.codec.http.DefaultHttpContent; +import io.netty.handler.codec.http.DefaultHttpResponse; +import io.netty.handler.codec.http.HttpRequest; +import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpServerCodec; +import io.netty.handler.codec.http.LastHttpContent; +import java.net.Socket; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.regex.Pattern; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; + +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { + + private static final String PATH = "/chunked"; + private static final String CONTINUE_PATH = "/chunked/continue"; + private static final String NO_CONTENT_PATH = "/bodyless/no-content"; + private static final String RESET_CONTENT_PATH = "/bodyless/reset-content"; + private static final String NOT_MODIFIED_PATH = "/bodyless/not-modified"; + private static final String CONTENT_LENGTH_ZERO_PATH = "/bodyless/content-length-zero"; + private static final String HEAD_PATH = "/bodyless/head"; + private static final String WEBSOCKET_PATH = "/websocket"; + private static final String UPGRADE_PATH = "/upgrade"; + private static final String CONNECT_AUTHORITY = "example.com:443"; + private static final String EARLY_HINTS_PATH = "/chunked/early-hints"; + private static final HttpResponseStatus EARLY_HINTS = new HttpResponseStatus(103, "Early Hints"); + + private final ChunkedResponseHandler handler = new ChunkedResponseHandler(); + + @Override + protected void configurePipeline(Channel ch) { + ch.pipeline().addLast(new HttpServerCodec()); + ch.pipeline().addLast(handler); + } + + @Test + void keepsServerSpanOpenUntilLastResponseChunk() throws Exception { + boolean reportedBeforeLastChunk; + try (Socket socket = connect()) { + socket.getOutputStream().write(request().getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitFirstChunkWritten(); + reportedBeforeLastChunk = writer.waitForTracesMax(1, 1); + + writeLastChunk(responseContext); + assertEquals("first", readChunkedHttpResponseBody(socket.getInputStream())); + } + + assertFalse( + reportedBeforeLastChunk, "server span should not be reported before LastHttpContent"); + assertTraces(trace(serverSpan(PATH))); + } + + @Test + void keepsServerSpanOpenUntilConnectionDrops() throws Exception { + boolean reportedBeforeConnectionDrop; + try (Socket socket = connect()) { + socket.getOutputStream().write(request().getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitFirstChunkWritten(); + reportedBeforeConnectionDrop = writer.waitForTracesMax(1, 1); + + closeChannel(responseContext); + } + + assertFalse( + reportedBeforeConnectionDrop, "server span should not be reported before connection drop"); + assertTraces(trace(serverSpan(PATH).error())); + } + + @Test + void keepsServerSpansSeparateForSequentialKeepAliveChunkedResponses() throws Exception { + boolean firstReportedBeforeLastChunk; + boolean secondReportedBeforeLastChunk; + try (Socket socket = connect()) { + socket.getOutputStream().write(request().getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext firstResponseContext = handler.awaitFirstChunkWritten(); + firstReportedBeforeLastChunk = writer.waitForTracesMax(1, 1); + + writeLastChunk(firstResponseContext); + assertEquals("first", readChunkedHttpResponseBody(socket.getInputStream())); + assertTrue(writer.waitForTracesMax(1, 5), "first keep-alive response was not reported"); + + socket.getOutputStream().write(request().getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext secondResponseContext = handler.awaitFirstChunkWritten(); + secondReportedBeforeLastChunk = writer.waitForTracesMax(2, 1); + + writeLastChunk(secondResponseContext); + assertEquals("first", readChunkedHttpResponseBody(socket.getInputStream())); + } + + assertFalse( + firstReportedBeforeLastChunk, + "first server span should not be reported before LastHttpContent"); + assertFalse( + secondReportedBeforeLastChunk, + "second server span should not be reported before LastHttpContent"); + assertTraces(trace(serverSpan(PATH)), trace(serverSpan(PATH))); + } + + @Test + void keepsServerSpanOpenAfter100ContinueUntilLastResponseChunk() throws Exception { + boolean reportedAfter100Continue; + boolean reportedBeforeLastChunk; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(CONTINUE_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.await100ContinueWritten(); + assertTrue( + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 100 "), + "server did not write 100 Continue"); + reportedAfter100Continue = writer.waitForTracesMax(1, 1); + + writeFirstChunk(responseContext); + responseContext = handler.awaitFirstChunkWritten(); + reportedBeforeLastChunk = writer.waitForTracesMax(1, 1); + + writeLastChunk(responseContext); + assertEquals("first", readChunkedHttpResponseBody(socket.getInputStream())); + } + + assertFalse(reportedAfter100Continue, "server span should not be reported after 100 Continue"); + assertFalse( + reportedBeforeLastChunk, "server span should not be reported before LastHttpContent"); + assertTraces(trace(serverSpan(CONTINUE_PATH))); + } + + @Test + void keepsServerSpanOpenAfter103EarlyHintsUntilLastResponseChunk() throws Exception { + boolean reportedAfterEarlyHints; + boolean reportedBeforeLastChunk; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(EARLY_HINTS_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitEarlyHintsWritten(); + assertResponseStatus(readHeaders(socket.getInputStream()), EARLY_HINTS); + reportedAfterEarlyHints = writer.waitForTracesMax(1, 1); + + writeFirstChunk(responseContext); + responseContext = handler.awaitFirstChunkWritten(); + reportedBeforeLastChunk = writer.waitForTracesMax(1, 1); + + writeLastChunk(responseContext); + assertEquals("first", readChunkedHttpResponseBody(socket.getInputStream())); + } + + assertFalse(reportedAfterEarlyHints, "server span should not be reported after 103"); + assertFalse( + reportedBeforeLastChunk, "server span should not be reported before LastHttpContent"); + assertTraces(trace(serverSpan(EARLY_HINTS_PATH))); + } + + @Test + void finishesServerSpanForHeaderOnlyNoContentResponse() throws Exception { + assertHeaderOnlyResponseFinishes(NO_CONTENT_PATH, NO_CONTENT); + } + + @Test + void finishesServerSpanForHeaderOnlyResetContentResponse() throws Exception { + assertHeaderOnlyResponseFinishes(RESET_CONTENT_PATH, RESET_CONTENT); + } + + @Test + void finishesServerSpanForHeaderOnlyNotModifiedResponse() throws Exception { + assertHeaderOnlyResponseFinishes(NOT_MODIFIED_PATH, NOT_MODIFIED); + } + + @Test + void finishesServerSpanForHeaderOnlyContentLengthZeroResponse() throws Exception { + assertHeaderOnlyResponseFinishes(CONTENT_LENGTH_ZERO_PATH, OK); + } + + @Test + void finishesServerSpanForHeaderOnlyHeadResponse() throws Exception { + assertHeaderOnlyResponseFinishes("HEAD", HEAD_PATH, OK); + } + + @Test + void finishesServerSpanForHeaderOnlyConnectResponse() throws Exception { + assertHeaderOnlyResponseFinishes("CONNECT", CONNECT_AUTHORITY, OK, "/"); + } + + @Test + void finishesServerSpanForHeaderOnlyWebSocketUpgrade() throws Exception { + assertHeaderOnlyResponseFinishes(WEBSOCKET_PATH, SWITCHING_PROTOCOLS); + } + + @Test + void finishesServerSpanForHeaderOnlyProtocolUpgrade() throws Exception { + assertHeaderOnlyResponseFinishes(UPGRADE_PATH, SWITCHING_PROTOCOLS); + } + + private void assertHeaderOnlyResponseFinishes(String path, HttpResponseStatus status) + throws Exception { + assertHeaderOnlyResponseFinishes("GET", path, status, path); + } + + private void assertHeaderOnlyResponseFinishes( + String method, String path, HttpResponseStatus status) throws Exception { + assertHeaderOnlyResponseFinishes(method, path, status, path); + } + + private void assertHeaderOnlyResponseFinishes( + String method, String path, HttpResponseStatus status, String resourcePath) throws Exception { + try (Socket socket = connect()) { + socket.getOutputStream().write(request(method, path).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + handler.awaitHeaderOnlyResponseWritten(path); + assertResponseStatus(readHeaders(socket.getInputStream()), status); + assertTrue( + writer.waitForTracesMax(1, 5), + "server span should be reported after header-only response " + status.code()); + } + + assertTraces(trace(serverSpan(method, resourcePath))); + } + + private static String request() { + return request(PATH); + } + + private static String request(String path) { + return request("GET", path); + } + + private static String request(String method, String path) { + String headers = method + " " + path + " HTTP/1.1\r\nHost: localhost\r\n"; + if (WEBSOCKET_PATH.equals(path)) { + headers += + "Connection: Upgrade\r\n" + + "Upgrade: websocket\r\n" + + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + + "Sec-WebSocket-Version: 13\r\n"; + } else if (UPGRADE_PATH.equals(path)) { + headers += "Connection: Upgrade\r\n" + "Upgrade: h2c\r\n"; + } + return headers + "\r\n"; + } + + private static SpanMatcher serverSpan(String path) { + return serverSpan("GET", path); + } + + private static SpanMatcher serverSpan(String method, String path) { + return span() + .root() + .operationName(Pattern.compile("netty\\.request")) + .resourceName(Pattern.compile(method + " " + Pattern.quote(path))) + .type("web"); + } + + private static void assertResponseStatus(String headers, HttpResponseStatus status) { + assertTrue( + headers.startsWith("HTTP/1.1 " + status.code() + " "), "unexpected response: " + headers); + } + + private void writeFirstChunk(ChannelHandlerContext responseContext) { + responseContext.executor().execute(() -> handler.writeChunkedResponse(responseContext)); + } + + private static void writeLastChunk(ChannelHandlerContext responseContext) { + responseContext + .executor() + .execute(() -> responseContext.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT)); + } + + private static void closeChannel(ChannelHandlerContext responseContext) { + responseContext.executor().execute(() -> responseContext.channel().close()); + } + + @ChannelHandler.Sharable + private static final class ChunkedResponseHandler + extends SimpleChannelInboundHandler { + private final BlockingQueue continueWrites = new LinkedBlockingQueue<>(); + private final BlockingQueue earlyHintsWrites = + new LinkedBlockingQueue<>(); + private final BlockingQueue firstChunkWrites = + new LinkedBlockingQueue<>(); + private final BlockingQueue headerOnlyWrites = new LinkedBlockingQueue<>(); + + @Override + protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { + if (CONTINUE_PATH.equals(request.uri())) { + ctx.writeAndFlush(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE)) + .addListener(future -> continueWrites.offer(ctx)); + } else if (EARLY_HINTS_PATH.equals(request.uri())) { + ctx.writeAndFlush(new DefaultFullHttpResponse(HTTP_1_1, EARLY_HINTS)) + .addListener(future -> earlyHintsWrites.offer(ctx)); + } else if (NO_CONTENT_PATH.equals(request.uri())) { + writeHeaderOnlyResponse(ctx, request.uri(), NO_CONTENT); + } else if (RESET_CONTENT_PATH.equals(request.uri())) { + writeHeaderOnlyResponse(ctx, request.uri(), RESET_CONTENT); + } else if (NOT_MODIFIED_PATH.equals(request.uri())) { + writeHeaderOnlyResponse(ctx, request.uri(), NOT_MODIFIED); + } else if (CONTENT_LENGTH_ZERO_PATH.equals(request.uri())) { + writeContentLengthZeroResponse(ctx, request.uri()); + } else if (HEAD_PATH.equals(request.uri())) { + writeHeaderOnlyResponse(ctx, request.uri(), OK); + } else if (CONNECT_AUTHORITY.equals(request.uri())) { + writeHeaderOnlyResponse(ctx, request.uri(), OK); + } else if (WEBSOCKET_PATH.equals(request.uri())) { + writeHeaderOnlyWebSocketUpgrade(ctx, request.uri()); + } else if (UPGRADE_PATH.equals(request.uri())) { + writeHeaderOnlyProtocolUpgrade(ctx, request.uri()); + } else { + writeChunkedResponse(ctx); + } + } + + private void writeChunkedResponse(ChannelHandlerContext ctx) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(TRANSFER_ENCODING, CHUNKED); + ctx.write(response); + ctx.writeAndFlush(new DefaultHttpContent(Unpooled.copiedBuffer("first", UTF_8))) + .addListener(future -> firstChunkWrites.offer(ctx)); + } + + private void writeHeaderOnlyResponse( + ChannelHandlerContext ctx, String path, HttpResponseStatus status) { + ctx.writeAndFlush(new DefaultHttpResponse(HTTP_1_1, status)) + .addListener(future -> headerOnlyWrites.offer(path)); + } + + private void writeContentLengthZeroResponse(ChannelHandlerContext ctx, String path) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(CONTENT_LENGTH, 0); + ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); + } + + private void writeHeaderOnlyWebSocketUpgrade(ChannelHandlerContext ctx, String path) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS); + response.headers().set(UPGRADE, "WebSocket"); + response.headers().set(CONNECTION, "Upgrade"); + ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); + } + + private void writeHeaderOnlyProtocolUpgrade(ChannelHandlerContext ctx, String path) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS); + response.headers().set(UPGRADE, "h2c"); + response.headers().set(CONNECTION, "Upgrade"); + ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); + } + + private ChannelHandlerContext awaitFirstChunkWritten() throws InterruptedException { + ChannelHandlerContext responseContext = firstChunkWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write the first chunk"); + } + return responseContext; + } + + private ChannelHandlerContext await100ContinueWritten() throws InterruptedException { + ChannelHandlerContext responseContext = continueWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write 100 Continue"); + } + return responseContext; + } + + private ChannelHandlerContext awaitEarlyHintsWritten() throws InterruptedException { + ChannelHandlerContext responseContext = earlyHintsWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write 103 Early Hints"); + } + return responseContext; + } + + private void awaitHeaderOnlyResponseWritten(String path) throws InterruptedException { + String responsePath = headerOnlyWrites.poll(5, SECONDS); + if (responsePath == null) { + throw new AssertionError("server did not write the header-only response"); + } + assertEquals(path, responsePath, "server wrote header-only response for unexpected path"); + } + } +} 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 index 610fa95ddbe..d40cbed8270 100644 --- 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 @@ -12,79 +12,42 @@ 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 { +public class NettyHttp11PipeliningTest extends NettyHttpServerTestSupport { 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(); - } + private final PipeliningHandler handler = new PipeliningHandler(3); - @AfterAll - void stopServer() { - if (eventLoopGroup != null) { - eventLoopGroup.shutdownGracefully(); - } + @Override + protected void configurePipeline(Channel ch) { + ch.pipeline().addLast(new HttpServerCodec()); + ch.pipeline().addLast(new HttpObjectAggregator(65536)); + ch.pipeline().addLast(handler); } @Test void createsServerSpanForEachPipelinedRequest() throws Exception { - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(pipelinedRequests().getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -127,52 +90,6 @@ private static TraceMatcher serverTrace(String 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; diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttpServerTestSupport.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttpServerTestSupport.java new file mode 100644 index 00000000000..f946f5cb01b --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttpServerTestSupport.java @@ -0,0 +1,159 @@ +package datadog.trace.instrumentation.netty41.server; + +import static java.nio.charset.StandardCharsets.US_ASCII; +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import io.netty.bootstrap.ServerBootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioServerSocketChannel; +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 org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; + +abstract class NettyHttpServerTestSupport extends AbstractInstrumentationTest { + + private EventLoopGroup eventLoopGroup; + private Channel serverChannel; + private int port; + + @BeforeAll + void startServer() throws Exception { + eventLoopGroup = new NioEventLoopGroup(); + ServerBootstrap bootstrap = + new ServerBootstrap() + .group(eventLoopGroup) + .channel(NioServerSocketChannel.class) + .childHandler( + new ChannelInitializer() { + @Override + protected void initChannel(Channel ch) { + configurePipeline(ch); + } + }); + serverChannel = bootstrap.bind(0).sync().channel(); + port = ((InetSocketAddress) serverChannel.localAddress()).getPort(); + } + + @AfterAll + void stopServer() { + if (serverChannel != null) { + serverChannel.close(); + } + if (eventLoopGroup != null) { + eventLoopGroup.shutdownGracefully(); + } + } + + protected abstract void configurePipeline(Channel ch); + + protected Socket connect() throws IOException { + Socket socket = new Socket("localhost", port); + socket.setSoTimeout(5000); + return socket; + } + + protected static String readHttpResponseBody(InputStream in) throws IOException { + String headers = readHeaders(in); + assertOkResponse(headers); + byte[] body = new byte[contentLength(headers)]; + readFully(in, body); + return new String(body, UTF_8); + } + + protected static String readChunkedHttpResponseBody(InputStream in) throws IOException { + String headers = readHeaders(in); + assertOkResponse(headers); + + ByteArrayOutputStream body = new ByteArrayOutputStream(); + while (true) { + int chunkSize = Integer.parseInt(readLine(in), 16); + if (chunkSize == 0) { + while (!readLine(in).isEmpty()) {} + return body.toString(UTF_8.name()); + } + byte[] chunk = new byte[chunkSize]; + readFully(in, chunk); + body.write(chunk); + assertEquals("", readLine(in), "chunk was not followed by CRLF"); + } + } + + private static void assertOkResponse(String headers) { + assertTrue(headers.startsWith("HTTP/1.1 200 "), "unexpected response: " + headers); + } + + protected 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 String readLine(InputStream in) throws IOException { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + boolean seenCarriageReturn = false; + while (true) { + int b = in.read(); + if (b == -1) { + throw new EOFException("response ended before line was complete"); + } + if (seenCarriageReturn && b == '\n') { + return out.toString(US_ASCII.name()); + } + if (seenCarriageReturn) { + out.write('\r'); + seenCarriageReturn = false; + } + if (b == '\r') { + seenCarriageReturn = true; + } else { + out.write(b); + } + } + } + + 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 void readFully(InputStream in, byte[] bytes) throws IOException { + int read = 0; + while (read < bytes.length) { + int count = in.read(bytes, read, bytes.length - read); + if (count == -1) { + throw new EOFException("response ended before body was complete"); + } + read += count; + } + } +} From f65d4a4537cb742125aaa98617fa8a46073df475 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 22:10:45 -0700 Subject: [PATCH 2/3] Fix Netty 4.1 server span completion for streaming responses Keep server spans open until terminal response writes, finish incomplete responses on channel close, and use queued request contexts to avoid keep-alive/pipelining misattribution. Cover chunked, bodyless, HEAD, CONNECT, interim, and WebSocket upgrade response cases. --- .../HttpServerRequestTracingHandler.java | 35 +++++- .../HttpServerResponseTracingHandler.java | 100 ++++++++++++++---- .../server/NettyChunkedResponseSpanTest.java | 46 +++++--- .../netty41/ServerRequestContext.java | 50 ++++++++- 4 files changed, 189 insertions(+), 42 deletions(-) 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 7e7dfcb1e89..74fb3755581 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,10 +15,13 @@ 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 { public static HttpServerRequestTracingHandler INSTANCE = new HttpServerRequestTracingHandler(); + private static final String INCOMPLETE_RESPONSE_MESSAGE = + "Channel closed before response completed"; @Override public void channelRead(final ChannelHandlerContext ctx, final Object msg) { @@ -54,7 +57,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) { DECORATE.onRequest(span, channel, request, parentContext); final ServerRequestContext serverContext = - ServerRequestContext.add(channel, context, request.headers()); + ServerRequestContext.add(channel, context, request); Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction(); if (rba != null) { @@ -89,9 +92,37 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { super.channelInactive(ctx); } finally { try { - ServerRequestContext.closeAll(ctx.channel()); + final Deque storedContexts = + ServerRequestContext.removeAll(ctx.channel()); + if (storedContexts != null) { + ServerRequestContext storedContext; + while ((storedContext = storedContexts.pollFirst()) != null) { + if (storedContext.isResponseStarted()) { + finishSpanOnIncompleteResponse(storedContext.tracingContext()); + } else { + publishSpanOnChannelClose(storedContext.tracingContext()); + } + } + } } catch (final Throwable ignored) { } } } + + private static void finishSpanOnIncompleteResponse(final Context storedContext) { + final AgentSpan span = AgentSpan.fromContext(storedContext); + if (span != null) { + DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE)); + DECORATE.beforeFinish(storedContext); + span.finish(); + } + } + + private static void publishSpanOnChannelClose(final Context storedContext) { + final AgentSpan span = AgentSpan.fromContext(storedContext); + if (span != null && span.phasedFinish()) { + // At this point we can just publish this span to avoid losing the rest of the trace. + span.publish(); + } + } } 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 11ea5475e1e..dd30238c810 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 @@ -13,8 +13,12 @@ import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.handler.codec.http.HttpHeaderNames; +import io.netty.handler.codec.http.HttpHeaderValues; import io.netty.handler.codec.http.HttpResponse; import io.netty.handler.codec.http.HttpResponseStatus; +import io.netty.handler.codec.http.HttpUtil; +import io.netty.handler.codec.http.LastHttpContent; +import io.netty.util.concurrent.Future; @ChannelHandler.Sharable public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter { @@ -22,11 +26,6 @@ public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdap @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) { - 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); @@ -36,33 +35,94 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann return; } - try (final ContextScope scope = storedContext.attach()) { - final HttpResponse response = (HttpResponse) msg; + if (!(msg instanceof HttpResponse) && !serverContext.isResponseStarted()) { + ctx.write(msg, prm); + return; + } + try (final ContextScope ignored = storedContext.attach()) { + final boolean terminalResponse = isTerminalResponse(ctx, span, serverContext, msg); + final ChannelPromise writePromise = terminalResponse && prm.isVoid() ? ctx.newPromise() : prm; try { - ctx.write(msg, prm); + if (terminalResponse) { + ServerRequestContext.remove(ctx.channel(), serverContext); + writePromise.addListener(future -> finishSpan(storedContext, span, future)); + } + ctx.write(msg, writePromise); } catch (final Throwable throwable) { DECORATE.onError(span, throwable); span.setHttpStatusCode(500); - span.finish(); // Finish the span manually since finishSpanOnClose was false - ServerRequestContext.remove(ctx.channel(), serverContext); + if (!terminalResponse) { + ServerRequestContext.remove(ctx.channel(), serverContext); + } + finishSpan(storedContext, span); throw throwable; } - final boolean isWebsocketUpgrade = - response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS - && "websocket".equals(response.headers().get(HttpHeaderNames.UPGRADE)); + } + } + + private static boolean isTerminalResponse( + final ChannelHandlerContext ctx, + final AgentSpan span, + final ServerRequestContext serverContext, + final Object msg) { + if (msg instanceof HttpResponse) { + final HttpResponse response = (HttpResponse) msg; + + final boolean isWebsocketUpgrade = isWebsocketUpgrade(response); + if (isInformationalResponse(response) && !isWebsocketUpgrade) { + return false; + } + if (isWebsocketUpgrade) { ctx.channel() .attr(WEBSOCKET_SENDER_HANDLER_CONTEXT) .set(new HandlerContext.Sender(span, ctx.channel().id().asShortText())); } - if (response.status() != HttpResponseStatus.CONTINUE - && (response.status() != HttpResponseStatus.SWITCHING_PROTOCOLS || isWebsocketUpgrade)) { - DECORATE.onResponse(span, response); - DECORATE.beforeFinish(scope.context()); - span.finish(); // Finish the span manually since finishSpanOnClose was false - ServerRequestContext.remove(ctx.channel(), serverContext); - } + DECORATE.onResponse(span, response); + serverContext.markResponseStarted(); + return msg instanceof LastHttpContent + || isBodylessResponse(serverContext, response) + || isWebsocketUpgrade; } + return serverContext.isResponseStarted() && msg instanceof LastHttpContent; + } + + private static boolean isInformationalResponse(final HttpResponse response) { + final int statusCode = response.status().code(); + return statusCode >= 100 && statusCode < 200; + } + + private static boolean isWebsocketUpgrade(final HttpResponse response) { + return response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS + && response + .headers() + .containsValue(HttpHeaderNames.UPGRADE, HttpHeaderValues.WEBSOCKET, true); + } + + private static boolean isBodylessResponse( + final ServerRequestContext serverContext, final HttpResponse response) { + final int statusCode = response.status().code(); + return serverContext.isHeadRequest() + || statusCode == 204 + || statusCode == 205 + || statusCode == 304 + || (serverContext.isConnectRequest() && statusCode >= 200 && statusCode < 300) + || (HttpUtil.getContentLength(response, -1) == 0 + && !HttpUtil.isTransferEncodingChunked(response)); + } + + private static void finishSpan( + final Context storedContext, final AgentSpan span, final Future future) { + if (!future.isSuccess()) { + DECORATE.onError(span, future.cause()); + span.setHttpStatusCode(500); + } + finishSpan(storedContext, span); + } + + private static void finishSpan(final Context storedContext, final AgentSpan span) { + DECORATE.beforeFinish(storedContext); + span.finish(); // Finish the span manually since finishSpanOnClose was false } } diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java index a48a9b2e9d2..75d0eb1a69b 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -51,8 +51,8 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { private static final String NOT_MODIFIED_PATH = "/bodyless/not-modified"; private static final String CONTENT_LENGTH_ZERO_PATH = "/bodyless/content-length-zero"; private static final String HEAD_PATH = "/bodyless/head"; + private static final String NO_RESPONSE_PATH = "/no-response"; private static final String WEBSOCKET_PATH = "/websocket"; - private static final String UPGRADE_PATH = "/upgrade"; private static final String CONNECT_AUTHORITY = "example.com:443"; private static final String EARLY_HINTS_PATH = "/chunked/early-hints"; private static final HttpResponseStatus EARLY_HINTS = new HttpResponseStatus(103, "Early Hints"); @@ -102,6 +102,22 @@ void keepsServerSpanOpenUntilConnectionDrops() throws Exception { assertTraces(trace(serverSpan(PATH).error())); } + @Test + void closesServerSpanWithoutErrorWhenConnectionDropsBeforeResponseStarts() throws Exception { + boolean reportedBeforeConnectionDrop; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(NO_RESPONSE_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + handler.awaitRequestWithoutResponse(); + reportedBeforeConnectionDrop = writer.waitForTracesMax(1, 1); + } + + assertFalse( + reportedBeforeConnectionDrop, "server span should not be reported before connection drop"); + assertTraces(trace(serverSpan(NO_RESPONSE_PATH))); + } + @Test void keepsServerSpansSeparateForSequentialKeepAliveChunkedResponses() throws Exception { boolean firstReportedBeforeLastChunk; @@ -225,11 +241,6 @@ void finishesServerSpanForHeaderOnlyWebSocketUpgrade() throws Exception { assertHeaderOnlyResponseFinishes(WEBSOCKET_PATH, SWITCHING_PROTOCOLS); } - @Test - void finishesServerSpanForHeaderOnlyProtocolUpgrade() throws Exception { - assertHeaderOnlyResponseFinishes(UPGRADE_PATH, SWITCHING_PROTOCOLS); - } - private void assertHeaderOnlyResponseFinishes(String path, HttpResponseStatus status) throws Exception { assertHeaderOnlyResponseFinishes("GET", path, status, path); @@ -272,8 +283,6 @@ private static String request(String method, String path) { + "Upgrade: websocket\r\n" + "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n" + "Sec-WebSocket-Version: 13\r\n"; - } else if (UPGRADE_PATH.equals(path)) { - headers += "Connection: Upgrade\r\n" + "Upgrade: h2c\r\n"; } return headers + "\r\n"; } @@ -318,6 +327,8 @@ private static final class ChunkedResponseHandler private final BlockingQueue firstChunkWrites = new LinkedBlockingQueue<>(); private final BlockingQueue headerOnlyWrites = new LinkedBlockingQueue<>(); + private final BlockingQueue requestsWithoutResponse = + new LinkedBlockingQueue<>(); @Override protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { @@ -339,10 +350,10 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { writeHeaderOnlyResponse(ctx, request.uri(), OK); } else if (CONNECT_AUTHORITY.equals(request.uri())) { writeHeaderOnlyResponse(ctx, request.uri(), OK); + } else if (NO_RESPONSE_PATH.equals(request.uri())) { + requestsWithoutResponse.offer(ctx); } else if (WEBSOCKET_PATH.equals(request.uri())) { writeHeaderOnlyWebSocketUpgrade(ctx, request.uri()); - } else if (UPGRADE_PATH.equals(request.uri())) { - writeHeaderOnlyProtocolUpgrade(ctx, request.uri()); } else { writeChunkedResponse(ctx); } @@ -375,13 +386,6 @@ private void writeHeaderOnlyWebSocketUpgrade(ChannelHandlerContext ctx, String p ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); } - private void writeHeaderOnlyProtocolUpgrade(ChannelHandlerContext ctx, String path) { - DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS); - response.headers().set(UPGRADE, "h2c"); - response.headers().set(CONNECTION, "Upgrade"); - ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); - } - private ChannelHandlerContext awaitFirstChunkWritten() throws InterruptedException { ChannelHandlerContext responseContext = firstChunkWrites.poll(5, SECONDS); if (responseContext == null) { @@ -406,6 +410,14 @@ private ChannelHandlerContext awaitEarlyHintsWritten() throws InterruptedExcepti return responseContext; } + private ChannelHandlerContext awaitRequestWithoutResponse() throws InterruptedException { + ChannelHandlerContext responseContext = requestsWithoutResponse.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not receive request without response"); + } + return responseContext; + } + private void awaitHeaderOnlyResponseWritten(String path) throws InterruptedException { String responsePath = headerOnlyWrites.poll(5, SECONDS); if (responsePath == null) { 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 4a8ea3e59e7..d6ce554a947 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 @@ -5,6 +5,8 @@ import datadog.context.Context; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import io.netty.handler.codec.http.HttpHeaders; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpRequest; import io.netty.util.AttributeKey; import io.netty.util.AttributeMap; import java.util.ArrayDeque; @@ -24,11 +26,26 @@ public static boolean canTrackRequest(final AttributeMap attributes) { /** Adds a request context to the queue tail. */ public static ServerRequestContext add( final AttributeMap attributes, final Context context, final HttpHeaders requestHeaders) { + return add(attributes, context, requestHeaders, null); + } + + /** Adds a request context to the queue tail. */ + public static ServerRequestContext add( + final AttributeMap attributes, final Context context, final HttpRequest request) { + return add(attributes, context, request.headers(), request.method()); + } + + private static ServerRequestContext add( + final AttributeMap attributes, + final Context context, + final HttpHeaders requestHeaders, + final HttpMethod requestMethod) { final Deque contexts = getOrCreate(attributes); if (!canAdd(attributes, contexts)) { return null; } - final ServerRequestContext serverContext = new ServerRequestContext(context, requestHeaders); + final ServerRequestContext serverContext = + new ServerRequestContext(context, requestHeaders, requestMethod); 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. @@ -78,9 +95,14 @@ public static void remove( /** Closes all pending request contexts on channel close. */ public static void closeAll(final AttributeMap attributes) { + close(removeAll(attributes)); + } + + /** Removes all pending request contexts. */ + public static Deque removeAll(final AttributeMap attributes) { // The legacy mirror must not outlive the authoritative request queue. attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); - close(attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove()); + return attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove(); } private static final int PIPELINING_LIMIT = 1000; @@ -153,6 +175,8 @@ private static boolean isPoisoned(final Deque contexts) { private final Context tracingContext; private final HttpHeaders requestHeaders; + private final HttpMethod requestMethod; + private boolean responseStarted; private boolean responseAnalyzed; private boolean responseBlocked; @@ -164,6 +188,22 @@ public HttpHeaders requestHeaders() { return requestHeaders; } + public boolean isHeadRequest() { + return HttpMethod.HEAD.equals(requestMethod); + } + + public boolean isConnectRequest() { + return HttpMethod.CONNECT.equals(requestMethod); + } + + public boolean isResponseStarted() { + return responseStarted; + } + + public void markResponseStarted() { + responseStarted = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; } @@ -180,8 +220,12 @@ public void markResponseBlocked() { responseBlocked = true; } - private ServerRequestContext(final Context tracingContext, final HttpHeaders requestHeaders) { + private ServerRequestContext( + final Context tracingContext, + final HttpHeaders requestHeaders, + final HttpMethod requestMethod) { this.tracingContext = tracingContext; this.requestHeaders = requestHeaders; + this.requestMethod = requestMethod; } } From de233160c2374d3e9ddd0fe4a558c3757565875a Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 14 Jul 2026 08:57:24 -0700 Subject: [PATCH 3/3] Fix Netty server span lifecycle for streaming responses Keep Netty 4.1 server spans open until terminal response writes while preserving request-ended callback timing for AppSec/IAST. Track one-shot beforeFinish execution separately from span finish so streaming spans get correct duration without dropping request-end security tags. --- .../HttpServerRequestTracingHandler.java | 10 +++++-- .../HttpServerResponseTracingHandler.java | 28 +++++++++++++++---- .../netty41/ServerRequestContext.java | 9 ++++++ 3 files changed, 38 insertions(+), 9 deletions(-) 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 74fb3755581..fc4efadda99 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 @@ -98,7 +98,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { ServerRequestContext storedContext; while ((storedContext = storedContexts.pollFirst()) != null) { if (storedContext.isResponseStarted()) { - finishSpanOnIncompleteResponse(storedContext.tracingContext()); + finishSpanOnIncompleteResponse(storedContext); } else { publishSpanOnChannelClose(storedContext.tracingContext()); } @@ -109,11 +109,15 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { } } - private static void finishSpanOnIncompleteResponse(final Context storedContext) { + private static void finishSpanOnIncompleteResponse(final ServerRequestContext serverContext) { + final Context storedContext = serverContext.tracingContext(); final AgentSpan span = AgentSpan.fromContext(storedContext); if (span != null) { DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE)); - DECORATE.beforeFinish(storedContext); + if (!serverContext.isBeforeFinishCalled()) { + serverContext.markBeforeFinishCalled(); + DECORATE.beforeFinish(storedContext); + } span.finish(); } } 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 dd30238c810..14ca128f350 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 @@ -46,16 +46,20 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann try { if (terminalResponse) { ServerRequestContext.remove(ctx.channel(), serverContext); - writePromise.addListener(future -> finishSpan(storedContext, span, future)); + writePromise.addListener( + future -> finishSpan(serverContext, storedContext, span, future)); } ctx.write(msg, writePromise); + if (serverContext.isResponseStarted()) { + beforeFinish(serverContext, storedContext); + } } catch (final Throwable throwable) { DECORATE.onError(span, throwable); span.setHttpStatusCode(500); if (!terminalResponse) { ServerRequestContext.remove(ctx.channel(), serverContext); } - finishSpan(storedContext, span); + finishSpan(serverContext, storedContext, span); throw throwable; } } @@ -113,16 +117,28 @@ private static boolean isBodylessResponse( } private static void finishSpan( - final Context storedContext, final AgentSpan span, final Future future) { + final ServerRequestContext serverContext, + final Context storedContext, + final AgentSpan span, + final Future future) { if (!future.isSuccess()) { DECORATE.onError(span, future.cause()); span.setHttpStatusCode(500); } - finishSpan(storedContext, span); + finishSpan(serverContext, storedContext, span); } - private static void finishSpan(final Context storedContext, final AgentSpan span) { - DECORATE.beforeFinish(storedContext); + private static void finishSpan( + final ServerRequestContext serverContext, final Context storedContext, final AgentSpan span) { + beforeFinish(serverContext, storedContext); span.finish(); // Finish the span manually since finishSpanOnClose was false } + + private static void beforeFinish( + final ServerRequestContext serverContext, final Context storedContext) { + if (!serverContext.isBeforeFinishCalled()) { + serverContext.markBeforeFinishCalled(); + DECORATE.beforeFinish(storedContext); + } + } } 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 d6ce554a947..82a67ae85f1 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 @@ -177,6 +177,7 @@ private static boolean isPoisoned(final Deque contexts) { private final HttpHeaders requestHeaders; private final HttpMethod requestMethod; private boolean responseStarted; + private boolean beforeFinishCalled; private boolean responseAnalyzed; private boolean responseBlocked; @@ -204,6 +205,14 @@ public void markResponseStarted() { responseStarted = true; } + public boolean isBeforeFinishCalled() { + return beforeFinishCalled; + } + + public void markBeforeFinishCalled() { + beforeFinishCalled = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; }