From 793775770aaec7660b93d9be80708d2a1e5460fb Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 15:56:21 -0700 Subject: [PATCH 01/11] 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 5be33eb209e518bf3c4a05f93e18ef4c60fc9d68 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Mon, 13 Jul 2026 22:10:45 -0700 Subject: [PATCH 02/11] 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 | 66 ++++++++++-- 4 files changed, 201 insertions(+), 46 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 0d934772d56..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, headers.get("accept")); + 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 5819f940090..29b7df5ced4 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 @@ -4,6 +4,9 @@ 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; @@ -25,12 +28,27 @@ 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 String acceptHeader) { + 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, acceptHeader); + 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. @@ -94,10 +112,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(); - attributes.attr(BLOCKED_RESPONSE_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; @@ -172,7 +194,9 @@ private static boolean isPoisoned(final Deque contexts) { } private final Context tracingContext; - private final String acceptHeader; + private final HttpHeaders requestHeaders; + private final HttpMethod requestMethod; + private boolean responseStarted; private boolean responseAnalyzed; public Context tracingContext() { @@ -183,6 +207,22 @@ public String acceptHeader() { return acceptHeader; } + 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; } @@ -191,8 +231,20 @@ public void markResponseAnalyzed() { responseAnalyzed = true; } - private ServerRequestContext(final Context tracingContext, final String acceptHeader) { + public boolean isResponseBlocked() { + return responseBlocked; + } + + public void markResponseBlocked() { + responseBlocked = true; + } + + private ServerRequestContext( + final Context tracingContext, + final HttpHeaders requestHeaders, + final HttpMethod requestMethod) { this.tracingContext = tracingContext; - this.acceptHeader = acceptHeader; + this.requestHeaders = requestHeaders; + this.requestMethod = requestMethod; } } From aaaf77d791f65576db31e6285ada9bc04a38dc62 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 14 Jul 2026 08:57:24 -0700 Subject: [PATCH 03/11] 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 29b7df5ced4..68d6e8641e9 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 @@ -197,6 +197,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; public Context tracingContext() { @@ -223,6 +224,14 @@ public void markResponseStarted() { responseStarted = true; } + public boolean isBeforeFinishCalled() { + return beforeFinishCalled; + } + + public void markBeforeFinishCalled() { + beforeFinishCalled = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; } From 372e9a81ed7fd14a44d727898285d93f28b8325d Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 14 Jul 2026 16:49:28 -0700 Subject: [PATCH 04/11] Fix Netty 4.1 server span edge cases from PR review - Defer beforeFinish until the response actually completes. - Finish close-delimited responses without error. - Guard Content-Length parsing. - Skip AppSec response analysis for all interim 1xx responses in MaybeBlockResponseHandler, not just 100 Continue. Add a regression test for close-delimited responses. --- .../HttpServerRequestTracingHandler.java | 22 +++++++----- .../HttpServerResponseTracingHandler.java | 36 +++++++++++++++---- .../server/MaybeBlockResponseHandler.java | 9 ++++- .../server/NettyChunkedResponseSpanTest.java | 34 ++++++++++++++++++ .../netty41/ServerRequestContext.java | 9 +++++ 5 files changed, 94 insertions(+), 16 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 fc4efadda99..1e2ffa28ba1 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); + finishSpanOnChannelClose(storedContext); } else { publishSpanOnChannelClose(storedContext.tracingContext()); } @@ -109,17 +109,23 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception { } } - private static void finishSpanOnIncompleteResponse(final ServerRequestContext serverContext) { + private static void finishSpanOnChannelClose(final ServerRequestContext serverContext) { final Context storedContext = serverContext.tracingContext(); final AgentSpan span = AgentSpan.fromContext(storedContext); - if (span != null) { + if (span == null) { + return; + } + if (!serverContext.isResponseCloseDelimited()) { + // The response declared a body via Content-Length or chunked encoding, but the channel closed + // before that body completed. Close-delimited responses, in contrast, end normally when the + // connection closes. DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE)); - if (!serverContext.isBeforeFinishCalled()) { - serverContext.markBeforeFinishCalled(); - DECORATE.beforeFinish(storedContext); - } - span.finish(); } + if (!serverContext.isBeforeFinishCalled()) { + serverContext.markBeforeFinishCalled(); + DECORATE.beforeFinish(storedContext); + } + span.finish(); } private static void publishSpanOnChannelClose(final Context storedContext) { 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 14ca128f350..e11b814be28 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 @@ -50,9 +50,6 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann 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); @@ -85,9 +82,18 @@ private static boolean isTerminalResponse( } DECORATE.onResponse(span, response); serverContext.markResponseStarted(); - return msg instanceof LastHttpContent + if (msg instanceof LastHttpContent || isBodylessResponse(serverContext, response) - || isWebsocketUpgrade; + || isWebsocketUpgrade) { + return true; + } + // A response with neither a Content-Length nor chunked transfer-encoding is delimited by the + // connection closing, so a later channel close is the normal end of this response rather than + // an incomplete one. + if (!hasKnownBodyLength(response)) { + serverContext.markResponseCloseDelimited(); + } + return false; } return serverContext.isResponseStarted() && msg instanceof LastHttpContent; } @@ -112,8 +118,24 @@ private static boolean isBodylessResponse( || statusCode == 205 || statusCode == 304 || (serverContext.isConnectRequest() && statusCode >= 200 && statusCode < 300) - || (HttpUtil.getContentLength(response, -1) == 0 - && !HttpUtil.isTransferEncodingChunked(response)); + || (contentLength(response) == 0 && !HttpUtil.isTransferEncodingChunked(response)); + } + + private static boolean hasKnownBodyLength(final HttpResponse response) { + return contentLength(response) >= 0 || HttpUtil.isTransferEncodingChunked(response); + } + + /** + * Returns the response {@code Content-Length}, or {@code -1} when it is absent or malformed. A + * malformed value is left for Netty's encoder to reject rather than failing the write from the + * tracing handler. + */ + private static long contentLength(final HttpResponse response) { + try { + return HttpUtil.getContentLength(response, -1L); + } catch (final NumberFormatException e) { + return -1L; + } } private static void finishSpan( 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 5dab095e4e0..c1ab0de230f 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 @@ -63,7 +63,14 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr return; } HttpResponse origResponse = (HttpResponse) msg; - if (origResponse.status().code() == HttpResponseStatus.CONTINUE.code()) { + int statusCode = origResponse.status().code(); + // Interim 1xx responses (e.g. 100 Continue, 103 Early Hints) precede the final response. + // Analyzing one here would consume the one-shot response analysis before the final response is + // written, so its status and headers would never be inspected. Switching Protocols (101) is + // terminal, so it is still analyzed. + if (statusCode >= 100 + && statusCode < 200 + && statusCode != HttpResponseStatus.SWITCHING_PROTOCOLS.code()) { super.write(ctx, msg, prm); return; } 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 75d0eb1a69b..d37cb0bac2f 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 @@ -52,6 +52,7 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { 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 CLOSE_DELIMITED_PATH = "/close-delimited"; private static final String WEBSOCKET_PATH = "/websocket"; private static final String CONNECT_AUTHORITY = "example.com:443"; private static final String EARLY_HINTS_PATH = "/chunked/early-hints"; @@ -102,6 +103,27 @@ void keepsServerSpanOpenUntilConnectionDrops() throws Exception { assertTraces(trace(serverSpan(PATH).error())); } + @Test + void finishesServerSpanWithoutErrorForCloseDelimitedResponse() throws Exception { + boolean reportedBeforeConnectionClose; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(CLOSE_DELIMITED_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitFirstChunkWritten(); + reportedBeforeConnectionClose = writer.waitForTracesMax(1, 1); + + closeChannel(responseContext); + } + + assertFalse( + reportedBeforeConnectionClose, + "server span should not be reported before the connection closes"); + // The response has neither Content-Length nor chunked encoding, so closing the connection is + // its normal completion and the span must not be flagged as an error. + assertTraces(trace(serverSpan(CLOSE_DELIMITED_PATH))); + } + @Test void closesServerSpanWithoutErrorWhenConnectionDropsBeforeResponseStarts() throws Exception { boolean reportedBeforeConnectionDrop; @@ -352,6 +374,8 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { writeHeaderOnlyResponse(ctx, request.uri(), OK); } else if (NO_RESPONSE_PATH.equals(request.uri())) { requestsWithoutResponse.offer(ctx); + } else if (CLOSE_DELIMITED_PATH.equals(request.uri())) { + writeCloseDelimitedResponse(ctx); } else if (WEBSOCKET_PATH.equals(request.uri())) { writeHeaderOnlyWebSocketUpgrade(ctx, request.uri()); } else { @@ -367,6 +391,16 @@ private void writeChunkedResponse(ChannelHandlerContext ctx) { .addListener(future -> firstChunkWrites.offer(ctx)); } + private void writeCloseDelimitedResponse(ChannelHandlerContext ctx) { + // No Content-Length and no chunked transfer-encoding: the body is delimited by the connection + // closing. + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(CONNECTION, "close"); + 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)) 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 68d6e8641e9..69bcb8762b0 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 @@ -197,6 +197,7 @@ private static boolean isPoisoned(final Deque contexts) { private final HttpHeaders requestHeaders; private final HttpMethod requestMethod; private boolean responseStarted; + private boolean responseCloseDelimited; private boolean beforeFinishCalled; private boolean responseAnalyzed; @@ -224,6 +225,14 @@ public void markResponseStarted() { responseStarted = true; } + public boolean isResponseCloseDelimited() { + return responseCloseDelimited; + } + + public void markResponseCloseDelimited() { + responseCloseDelimited = true; + } + public boolean isBeforeFinishCalled() { return beforeFinishCalled; } From 0a6269fbd9d8323d31ffb897eb20aa25bdc9df79 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Tue, 14 Jul 2026 17:49:52 -0700 Subject: [PATCH 05/11] Restore request-ended callbacks for terminal Netty responses --- .../netty41/server/HttpServerResponseTracingHandler.java | 4 ++++ 1 file changed, 4 insertions(+) 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 e11b814be28..366d8508455 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 @@ -50,6 +50,10 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann future -> finishSpan(serverContext, storedContext, span, future)); } ctx.write(msg, writePromise); + if (terminalResponse) { + // Run the request-ended callbacks (AppSec/IAST) here, while the context is attached + beforeFinish(serverContext, storedContext); + } } catch (final Throwable throwable) { DECORATE.onError(span, throwable); span.setHttpStatusCode(500); From 344cbb46cf54d5163c8b98cddd3333f27dad5d3c Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 14:26:11 -0700 Subject: [PATCH 06/11] Resolved: terminal responses no longer call `beforeFinish(...)` immediately after `ctx.write(...)`. The write-promise listener is now the only normal terminal completion path, so `finishSpan(...)` runs after the write completes and invokes `beforeFinish(...)` there. The exception path is also guarded with `writePromise.isDone()`: if the terminal write promise has already completed, the listener has already handled finishing, so the catch block just rethrows and avoids a second finish/decorator pass. --- .../server/HttpServerResponseTracingHandler.java | 16 +++++++--------- .../netty41/ServerRequestContextTest.java | 9 +++++---- .../server/MaybeBlockResponseHandlerTest.java | 4 +++- .../netty41/ServerRequestContext.java | 6 ++++-- 4 files changed, 19 insertions(+), 16 deletions(-) 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 366d8508455..1a2f0fe2dc4 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 @@ -50,17 +50,15 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann future -> finishSpan(serverContext, storedContext, span, future)); } ctx.write(msg, writePromise); - if (terminalResponse) { - // Run the request-ended callbacks (AppSec/IAST) here, while the context is attached - beforeFinish(serverContext, storedContext); - } } catch (final Throwable throwable) { - DECORATE.onError(span, throwable); - span.setHttpStatusCode(500); - if (!terminalResponse) { - ServerRequestContext.remove(ctx.channel(), serverContext); + if (!terminalResponse || !writePromise.isDone()) { + DECORATE.onError(span, throwable); + span.setHttpStatusCode(500); + if (!terminalResponse) { + ServerRequestContext.remove(ctx.channel(), serverContext); + } + finishSpan(serverContext, storedContext, span); } - finishSpan(serverContext, storedContext, span); throw throwable; } } 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 index 4cac8e6c1e8..f51ad0fa8bb 100644 --- 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 @@ -19,13 +19,13 @@ void disablesTrackingWhenPipeliningLimitIsExceeded() { DefaultAttributeMap attributes = new DefaultAttributeMap(); for (int i = 0; i < PIPELINING_LIMIT; i++) { - assertNotNull(ServerRequestContext.add(attributes, Context.root(), null)); + 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(), null)); + assertNull(ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders())); } @Test @@ -35,7 +35,7 @@ void capturesAcceptHeaderValue() { headers.set("accept", "text/html"); ServerRequestContext serverContext = - ServerRequestContext.add(attributes, Context.root(), headers.get("accept")); + ServerRequestContext.add(attributes, Context.root(), headers); headers.set("accept", "application/json"); assertEquals("text/html", serverContext.acceptHeader()); @@ -44,7 +44,8 @@ void capturesAcceptHeaderValue() { @Test void tracksBlockedResponseUntilChannelClose() { DefaultAttributeMap attributes = new DefaultAttributeMap(); - ServerRequestContext serverContext = ServerRequestContext.add(attributes, Context.root(), null); + ServerRequestContext serverContext = + ServerRequestContext.add(attributes, Context.root(), new DefaultHttpHeaders()); ServerRequestContext.markResponseBlocked(attributes); ServerRequestContext.remove(attributes, serverContext); diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandlerTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandlerTest.java index 7ba3a2aabf1..658acc3f94d 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandlerTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/MaybeBlockResponseHandlerTest.java @@ -9,6 +9,7 @@ import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.handler.codec.http.DefaultHttpHeaders; import org.junit.jupiter.api.Test; class MaybeBlockResponseHandlerTest { @@ -16,7 +17,8 @@ class MaybeBlockResponseHandlerTest { @Test void dropsWritesAfterBlockedContextHasBeenRemoved() { EmbeddedChannel channel = new EmbeddedChannel(MaybeBlockResponseHandler.INSTANCE); - ServerRequestContext serverContext = ServerRequestContext.add(channel, Context.root(), null); + ServerRequestContext serverContext = + ServerRequestContext.add(channel, Context.root(), new DefaultHttpHeaders()); ServerRequestContext.markResponseBlocked(channel); ServerRequestContext.remove(channel, serverContext); ByteBuf lateResponseChunk = Unpooled.buffer().writeByte(1); 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 69bcb8762b0..9836d1112d3 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 @@ -112,6 +112,7 @@ public static void remove( /** Closes all pending request contexts on channel close. */ public static void closeAll(final AttributeMap attributes) { + attributes.attr(BLOCKED_RESPONSE_ATTRIBUTE_KEY).remove(); close(removeAll(attributes)); } @@ -194,12 +195,13 @@ private static boolean isPoisoned(final Deque contexts) { } private final Context tracingContext; - private final HttpHeaders requestHeaders; + private final String acceptHeader; private final HttpMethod requestMethod; private boolean responseStarted; private boolean responseCloseDelimited; private boolean beforeFinishCalled; private boolean responseAnalyzed; + private boolean responseBlocked; public Context tracingContext() { return tracingContext; @@ -262,7 +264,7 @@ private ServerRequestContext( final HttpHeaders requestHeaders, final HttpMethod requestMethod) { this.tracingContext = tracingContext; - this.requestHeaders = requestHeaders; + this.acceptHeader = requestHeaders == null ? null : requestHeaders.get("accept"); this.requestMethod = requestMethod; } } From c2e0b3fb7b1dfe2d245a9e4fdc346bec7d79d966 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 14:35:59 -0700 Subject: [PATCH 07/11] Changed HttpServerResponseTracingHandler.java so unknown-length responses are marked close-delimited before `LastHttpContent` can make them terminal. Close-delimited contexts now wait for `channelInactive`. Added a regression in NettyChunkedResponseSpanTest.java for `DefaultFullHttpResponse` with no `Content-Length` and no chunked encoding. --- .../HttpServerResponseTracingHandler.java | 12 +++-- .../server/NettyChunkedResponseSpanTest.java | 45 +++++++++++++++++++ 2 files changed, 53 insertions(+), 4 deletions(-) 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 1a2f0fe2dc4..ec68f6726fe 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 @@ -84,9 +84,7 @@ private static boolean isTerminalResponse( } DECORATE.onResponse(span, response); serverContext.markResponseStarted(); - if (msg instanceof LastHttpContent - || isBodylessResponse(serverContext, response) - || isWebsocketUpgrade) { + if (isBodylessResponse(serverContext, response) || isWebsocketUpgrade) { return true; } // A response with neither a Content-Length nor chunked transfer-encoding is delimited by the @@ -94,10 +92,16 @@ private static boolean isTerminalResponse( // an incomplete one. if (!hasKnownBodyLength(response)) { serverContext.markResponseCloseDelimited(); + return false; + } + if (msg instanceof LastHttpContent) { + return true; } return false; } - return serverContext.isResponseStarted() && msg instanceof LastHttpContent; + return serverContext.isResponseStarted() + && !serverContext.isResponseCloseDelimited() + && msg instanceof LastHttpContent; } private static boolean isInformationalResponse(final HttpResponse response) { 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 d37cb0bac2f..bf2e72f3a7e 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 @@ -53,6 +53,7 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { private static final String HEAD_PATH = "/bodyless/head"; private static final String NO_RESPONSE_PATH = "/no-response"; private static final String CLOSE_DELIMITED_PATH = "/close-delimited"; + private static final String CLOSE_DELIMITED_FULL_RESPONSE_PATH = "/close-delimited/full"; private static final String WEBSOCKET_PATH = "/websocket"; private static final String CONNECT_AUTHORITY = "example.com:443"; private static final String EARLY_HINTS_PATH = "/chunked/early-hints"; @@ -124,6 +125,27 @@ void finishesServerSpanWithoutErrorForCloseDelimitedResponse() throws Exception assertTraces(trace(serverSpan(CLOSE_DELIMITED_PATH))); } + @Test + void waitsForChannelCloseForCloseDelimitedFullResponse() throws Exception { + boolean reportedBeforeConnectionClose; + try (Socket socket = connect()) { + socket + .getOutputStream() + .write(request(CLOSE_DELIMITED_FULL_RESPONSE_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitCloseDelimitedFullResponseWritten(); + reportedBeforeConnectionClose = writer.waitForTracesMax(1, 1); + + closeChannel(responseContext); + } + + assertFalse( + reportedBeforeConnectionClose, + "server span should not be reported before the connection closes"); + assertTraces(trace(serverSpan(CLOSE_DELIMITED_FULL_RESPONSE_PATH))); + } + @Test void closesServerSpanWithoutErrorWhenConnectionDropsBeforeResponseStarts() throws Exception { boolean reportedBeforeConnectionDrop; @@ -348,6 +370,8 @@ private static final class ChunkedResponseHandler new LinkedBlockingQueue<>(); private final BlockingQueue firstChunkWrites = new LinkedBlockingQueue<>(); + private final BlockingQueue closeDelimitedFullResponseWrites = + new LinkedBlockingQueue<>(); private final BlockingQueue headerOnlyWrites = new LinkedBlockingQueue<>(); private final BlockingQueue requestsWithoutResponse = new LinkedBlockingQueue<>(); @@ -376,6 +400,8 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { requestsWithoutResponse.offer(ctx); } else if (CLOSE_DELIMITED_PATH.equals(request.uri())) { writeCloseDelimitedResponse(ctx); + } else if (CLOSE_DELIMITED_FULL_RESPONSE_PATH.equals(request.uri())) { + writeCloseDelimitedFullResponse(ctx); } else if (WEBSOCKET_PATH.equals(request.uri())) { writeHeaderOnlyWebSocketUpgrade(ctx, request.uri()); } else { @@ -401,6 +427,16 @@ private void writeCloseDelimitedResponse(ChannelHandlerContext ctx) { .addListener(future -> firstChunkWrites.offer(ctx)); } + private void writeCloseDelimitedFullResponse(ChannelHandlerContext ctx) { + // No Content-Length and no chunked transfer-encoding: the full response still has a + // close-delimited body on the wire. + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.copiedBuffer("first", UTF_8)); + response.headers().set(CONNECTION, "close"); + ctx.writeAndFlush(response) + .addListener(future -> closeDelimitedFullResponseWrites.offer(ctx)); + } + private void writeHeaderOnlyResponse( ChannelHandlerContext ctx, String path, HttpResponseStatus status) { ctx.writeAndFlush(new DefaultHttpResponse(HTTP_1_1, status)) @@ -444,6 +480,15 @@ private ChannelHandlerContext awaitEarlyHintsWritten() throws InterruptedExcepti return responseContext; } + private ChannelHandlerContext awaitCloseDelimitedFullResponseWritten() + throws InterruptedException { + ChannelHandlerContext responseContext = closeDelimitedFullResponseWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write the full close-delimited response"); + } + return responseContext; + } + private ChannelHandlerContext awaitRequestWithoutResponse() throws InterruptedException { ChannelHandlerContext responseContext = requestsWithoutResponse.poll(5, SECONDS); if (responseContext == null) { From 4f7d9245eecceac1ec5a776bc5f58fb007936741 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 14:52:56 -0700 Subject: [PATCH 08/11] Restrict close-delimited response handling to HTTP/1.x only. The Netty HTTP/2 server path now caches when `Http2StreamFrameToHttpObjectCodec` is present and avoids marking no-length responses as close-delimited there. This keeps HTTP/2 responses dependent on `LastHttpContent`/END_STREAM, so an early stream close is still reported as an incomplete-response error. --- .../NettyChannelPipelineInstrumentation.java | 2 ++ .../server/HttpServerResponseTracingHandler.java | 15 +++++++++++---- .../instrumentation/netty41/AttributeKeys.java | 3 +++ 3 files changed, 16 insertions(+), 4 deletions(-) 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 668077b87a6..52aa8aa36ef 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 @@ -8,6 +8,7 @@ import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.noopContinuation; import static datadog.trace.instrumentation.netty41.AttributeKeys.CONNECT_PARENT_CONTINUATION_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY; +import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_STREAM_CODEC_ATTRIBUTE_KEY; import static net.bytebuddy.matcher.ElementMatchers.isMethod; import static net.bytebuddy.matcher.ElementMatchers.returns; import static net.bytebuddy.matcher.ElementMatchers.takesArgument; @@ -237,6 +238,7 @@ else if (handler instanceof HttpClientCodec) { pipeline, handler, HttpClientResponseTracingHandler.INSTANCE); } else if (NettyHttp2Helper.isHttp2FrameCodec(handler)) { if (NettyHttp2Helper.isServer(handler)) { + pipeline.channel().attr(HTTP2_STREAM_CODEC_ATTRIBUTE_KEY).set(Boolean.TRUE); NettyPipelineHelper.addHandlerAfter( pipeline, handler, 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 ec68f6726fe..069205f7f67 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,5 +1,6 @@ package datadog.trace.instrumentation.netty41.server; +import static datadog.trace.instrumentation.netty41.AttributeKeys.HTTP2_STREAM_CODEC_ATTRIBUTE_KEY; import static datadog.trace.instrumentation.netty41.AttributeKeys.WEBSOCKET_SENDER_HANDLER_CONTEXT; import static datadog.trace.instrumentation.netty41.server.NettyHttpServerDecorator.DECORATE; @@ -87,10 +88,10 @@ private static boolean isTerminalResponse( if (isBodylessResponse(serverContext, response) || isWebsocketUpgrade) { return true; } - // A response with neither a Content-Length nor chunked transfer-encoding is delimited by the - // connection closing, so a later channel close is the normal end of this response rather than - // an incomplete one. - if (!hasKnownBodyLength(response)) { + // HTTP/1.x responses with neither a Content-Length nor chunked transfer-encoding are + // delimited by the connection closing. HTTP/2 streams translated through + // Http2StreamFrameToHttpObjectCodec are delimited by END_STREAM/LastHttpContent instead. + if (isCloseDelimitedHttp1Response(ctx, response)) { serverContext.markResponseCloseDelimited(); return false; } @@ -131,6 +132,12 @@ private static boolean hasKnownBodyLength(final HttpResponse response) { return contentLength(response) >= 0 || HttpUtil.isTransferEncodingChunked(response); } + private static boolean isCloseDelimitedHttp1Response( + final ChannelHandlerContext ctx, final HttpResponse response) { + return !hasKnownBodyLength(response) + && !Boolean.TRUE.equals(ctx.channel().attr(HTTP2_STREAM_CODEC_ATTRIBUTE_KEY).get()); + } + /** * Returns the response {@code Content-Length}, or {@code -1} when it is absent or malformed. A * malformed value is left for Netty's encoder to reject rather than failing the write from the 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 307002a68fb..b8790de68ef 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 @@ -29,6 +29,9 @@ public final class AttributeKeys { public static final AttributeKey HTTP2_CONNECTION_CODEC_ATTRIBUTE_KEY = attributeKey("datadog.http2.connection.codec"); + public static final AttributeKey HTTP2_STREAM_CODEC_ATTRIBUTE_KEY = + attributeKey("datadog.http2.stream.codec"); + public static final AttributeKey PARENT_CONTEXT_ATTRIBUTE_KEY = attributeKey("datadog.server.parent-context"); From 69ec7b0498b24abc836d74776c8454ba7471e5ee Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 15:06:17 -0700 Subject: [PATCH 09/11] The Netty 4.1 websocket upgrade check is now value-based for status `101`, and the regression test for a distinct `HttpResponseStatus(101, ...)` passes. --- .../HttpServerResponseTracingHandler.java | 2 +- .../server/NettyChunkedResponseSpanTest.java | 24 +++++++++++++++---- 2 files changed, 21 insertions(+), 5 deletions(-) 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 069205f7f67..29ce9f444ea 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 @@ -111,7 +111,7 @@ private static boolean isInformationalResponse(final HttpResponse response) { } private static boolean isWebsocketUpgrade(final HttpResponse response) { - return response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS + return response.status().code() == HttpResponseStatus.SWITCHING_PROTOCOLS.code() && response .headers() .containsValue(HttpHeaderNames.UPGRADE, HttpHeaderValues.WEBSOCKET, true); 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 bf2e72f3a7e..93ff7ae01cc 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 @@ -55,9 +55,12 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { private static final String CLOSE_DELIMITED_PATH = "/close-delimited"; private static final String CLOSE_DELIMITED_FULL_RESPONSE_PATH = "/close-delimited/full"; private static final String WEBSOCKET_PATH = "/websocket"; + private static final String CUSTOM_WEBSOCKET_STATUS_PATH = "/websocket/custom-status"; 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 static final HttpResponseStatus CUSTOM_SWITCHING_PROTOCOLS = + new HttpResponseStatus(SWITCHING_PROTOCOLS.code(), "Switching Protocols"); private final ChunkedResponseHandler handler = new ChunkedResponseHandler(); @@ -285,6 +288,12 @@ void finishesServerSpanForHeaderOnlyWebSocketUpgrade() throws Exception { assertHeaderOnlyResponseFinishes(WEBSOCKET_PATH, SWITCHING_PROTOCOLS); } + @Test + void finishesServerSpanForHeaderOnlyWebSocketUpgradeWithDistinctStatusInstance() + throws Exception { + assertHeaderOnlyResponseFinishes(CUSTOM_WEBSOCKET_STATUS_PATH, SWITCHING_PROTOCOLS); + } + private void assertHeaderOnlyResponseFinishes(String path, HttpResponseStatus status) throws Exception { assertHeaderOnlyResponseFinishes("GET", path, status, path); @@ -321,7 +330,7 @@ private static String request(String 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)) { + if (isWebSocketPath(path)) { headers += "Connection: Upgrade\r\n" + "Upgrade: websocket\r\n" @@ -331,6 +340,10 @@ private static String request(String method, String path) { return headers + "\r\n"; } + private static boolean isWebSocketPath(String path) { + return WEBSOCKET_PATH.equals(path) || CUSTOM_WEBSOCKET_STATUS_PATH.equals(path); + } + private static SpanMatcher serverSpan(String path) { return serverSpan("GET", path); } @@ -403,7 +416,9 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { } else if (CLOSE_DELIMITED_FULL_RESPONSE_PATH.equals(request.uri())) { writeCloseDelimitedFullResponse(ctx); } else if (WEBSOCKET_PATH.equals(request.uri())) { - writeHeaderOnlyWebSocketUpgrade(ctx, request.uri()); + writeHeaderOnlyWebSocketUpgrade(ctx, request.uri(), SWITCHING_PROTOCOLS); + } else if (CUSTOM_WEBSOCKET_STATUS_PATH.equals(request.uri())) { + writeHeaderOnlyWebSocketUpgrade(ctx, request.uri(), CUSTOM_SWITCHING_PROTOCOLS); } else { writeChunkedResponse(ctx); } @@ -449,8 +464,9 @@ private void writeContentLengthZeroResponse(ChannelHandlerContext ctx, String pa ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); } - private void writeHeaderOnlyWebSocketUpgrade(ChannelHandlerContext ctx, String path) { - DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS); + private void writeHeaderOnlyWebSocketUpgrade( + ChannelHandlerContext ctx, String path, HttpResponseStatus status) { + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, status); response.headers().set(UPGRADE, "WebSocket"); response.headers().set(CONNECTION, "Upgrade"); ctx.writeAndFlush(response).addListener(future -> headerOnlyWrites.offer(path)); From cde82d0068ff84666c3efe2547a932a540dee649 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 15:16:42 -0700 Subject: [PATCH 10/11] It now treats known-length responses as complete once the declared body bytes are successfully written, so closing the connection afterward no longer marks the span as an incomplete-response error. Chunked/incomplete known-length closes still error as expected. Tests passed for `NettyChunkedResponseSpanTest`; changes are not committed. --- .../HttpServerResponseTracingHandler.java | 32 +++- .../server/NettyChunkedResponseSpanTest.java | 144 +++++++++++++++++- .../netty41/ServerRequestContext.java | 19 +++ 3 files changed, 190 insertions(+), 5 deletions(-) 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 29ce9f444ea..58b3fe9fc23 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 @@ -9,10 +9,13 @@ import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.bootstrap.instrumentation.websocket.HandlerContext; import datadog.trace.instrumentation.netty41.ServerRequestContext; +import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufHolder; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; +import io.netty.channel.FileRegion; import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpHeaderValues; import io.netty.handler.codec.http.HttpResponse; @@ -43,19 +46,23 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann try (final ContextScope ignored = storedContext.attach()) { final boolean terminalResponse = isTerminalResponse(ctx, span, serverContext, msg); - final ChannelPromise writePromise = terminalResponse && prm.isVoid() ? ctx.newPromise() : prm; + final boolean knownLengthResponseComplete = + !terminalResponse && serverContext.recordResponseBodyBytes(responseBodyBytes(msg)); + final boolean finishResponseOnWrite = terminalResponse || knownLengthResponseComplete; + final ChannelPromise writePromise = + finishResponseOnWrite && prm.isVoid() ? ctx.newPromise() : prm; try { - if (terminalResponse) { + if (finishResponseOnWrite) { ServerRequestContext.remove(ctx.channel(), serverContext); writePromise.addListener( future -> finishSpan(serverContext, storedContext, span, future)); } ctx.write(msg, writePromise); } catch (final Throwable throwable) { - if (!terminalResponse || !writePromise.isDone()) { + if (!finishResponseOnWrite || !writePromise.isDone()) { DECORATE.onError(span, throwable); span.setHttpStatusCode(500); - if (!terminalResponse) { + if (!finishResponseOnWrite) { ServerRequestContext.remove(ctx.channel(), serverContext); } finishSpan(serverContext, storedContext, span); @@ -88,6 +95,10 @@ private static boolean isTerminalResponse( if (isBodylessResponse(serverContext, response) || isWebsocketUpgrade) { return true; } + final long contentLength = contentLength(response); + if (contentLength > 0 && !HttpUtil.isTransferEncodingChunked(response)) { + serverContext.setResponseContentLength(contentLength); + } // HTTP/1.x responses with neither a Content-Length nor chunked transfer-encoding are // delimited by the connection closing. HTTP/2 streams translated through // Http2StreamFrameToHttpObjectCodec are delimited by END_STREAM/LastHttpContent instead. @@ -138,6 +149,19 @@ private static boolean isCloseDelimitedHttp1Response( && !Boolean.TRUE.equals(ctx.channel().attr(HTTP2_STREAM_CODEC_ATTRIBUTE_KEY).get()); } + private static long responseBodyBytes(final Object msg) { + if (msg instanceof ByteBuf) { + return ((ByteBuf) msg).readableBytes(); + } + if (msg instanceof ByteBufHolder) { + return ((ByteBufHolder) msg).content().readableBytes(); + } + if (msg instanceof FileRegion) { + return ((FileRegion) msg).count(); + } + return 0; + } + /** * Returns the response {@code Content-Length}, or {@code -1} when it is absent or malformed. A * malformed value is left for Netty's encoder to reject rather than failing the write from the 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 93ff7ae01cc..da6547148e4 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 @@ -26,6 +26,7 @@ import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.DefaultFileRegion; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.DefaultHttpContent; @@ -34,7 +35,12 @@ import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.handler.codec.http.LastHttpContent; +import java.io.IOException; import java.net.Socket; +import java.nio.channels.FileChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.regex.Pattern; @@ -54,6 +60,9 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { private static final String NO_RESPONSE_PATH = "/no-response"; private static final String CLOSE_DELIMITED_PATH = "/close-delimited"; private static final String CLOSE_DELIMITED_FULL_RESPONSE_PATH = "/close-delimited/full"; + private static final String KNOWN_LENGTH_BYTE_BUF_PATH = "/known-length/bytebuf"; + private static final String KNOWN_LENGTH_FILE_REGION_PATH = "/known-length/file-region"; + private static final String INCOMPLETE_KNOWN_LENGTH_PATH = "/known-length/incomplete"; private static final String WEBSOCKET_PATH = "/websocket"; private static final String CUSTOM_WEBSOCKET_STATUS_PATH = "/websocket/custom-status"; private static final String CONNECT_AUTHORITY = "example.com:443"; @@ -149,6 +158,50 @@ void waitsForChannelCloseForCloseDelimitedFullResponse() throws Exception { assertTraces(trace(serverSpan(CLOSE_DELIMITED_FULL_RESPONSE_PATH))); } + @Test + void finishesServerSpanWithoutErrorForKnownLengthByteBufResponseClosedAfterBody() + throws Exception { + try (Socket socket = connect()) { + socket.getOutputStream().write(request(KNOWN_LENGTH_BYTE_BUF_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitKnownLengthByteBufWritten(); + closeChannel(responseContext); + assertEquals("first", readHttpResponseBody(socket.getInputStream())); + } + + assertTraces(trace(serverSpan(KNOWN_LENGTH_BYTE_BUF_PATH))); + } + + @Test + void finishesServerSpanWithoutErrorForKnownLengthFileRegionResponseClosedAfterBody() + throws Exception { + try (Socket socket = connect()) { + socket.getOutputStream().write(request(KNOWN_LENGTH_FILE_REGION_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitKnownLengthFileRegionWritten(); + closeChannel(responseContext); + assertEquals("first", readHttpResponseBody(socket.getInputStream())); + } + + assertTraces(trace(serverSpan(KNOWN_LENGTH_FILE_REGION_PATH))); + } + + @Test + void marksKnownLengthResponseErrorWhenConnectionClosesBeforeDeclaredBodyComplete() + throws Exception { + try (Socket socket = connect()) { + socket.getOutputStream().write(request(INCOMPLETE_KNOWN_LENGTH_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitIncompleteKnownLengthWritten(); + closeChannel(responseContext); + } + + assertTraces(trace(serverSpan(INCOMPLETE_KNOWN_LENGTH_PATH).error())); + } + @Test void closesServerSpanWithoutErrorWhenConnectionDropsBeforeResponseStarts() throws Exception { boolean reportedBeforeConnectionDrop; @@ -385,12 +438,18 @@ private static final class ChunkedResponseHandler new LinkedBlockingQueue<>(); private final BlockingQueue closeDelimitedFullResponseWrites = new LinkedBlockingQueue<>(); + private final BlockingQueue knownLengthByteBufWrites = + new LinkedBlockingQueue<>(); + private final BlockingQueue knownLengthFileRegionWrites = + new LinkedBlockingQueue<>(); + private final BlockingQueue incompleteKnownLengthWrites = + new LinkedBlockingQueue<>(); private final BlockingQueue headerOnlyWrites = new LinkedBlockingQueue<>(); private final BlockingQueue requestsWithoutResponse = new LinkedBlockingQueue<>(); @Override - protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { + protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) throws Exception { if (CONTINUE_PATH.equals(request.uri())) { ctx.writeAndFlush(new DefaultFullHttpResponse(HTTP_1_1, CONTINUE)) .addListener(future -> continueWrites.offer(ctx)); @@ -415,6 +474,12 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) { writeCloseDelimitedResponse(ctx); } else if (CLOSE_DELIMITED_FULL_RESPONSE_PATH.equals(request.uri())) { writeCloseDelimitedFullResponse(ctx); + } else if (KNOWN_LENGTH_BYTE_BUF_PATH.equals(request.uri())) { + writeKnownLengthByteBufResponse(ctx); + } else if (KNOWN_LENGTH_FILE_REGION_PATH.equals(request.uri())) { + writeKnownLengthFileRegionResponse(ctx); + } else if (INCOMPLETE_KNOWN_LENGTH_PATH.equals(request.uri())) { + writeIncompleteKnownLengthResponse(ctx); } else if (WEBSOCKET_PATH.equals(request.uri())) { writeHeaderOnlyWebSocketUpgrade(ctx, request.uri(), SWITCHING_PROTOCOLS); } else if (CUSTOM_WEBSOCKET_STATUS_PATH.equals(request.uri())) { @@ -452,6 +517,45 @@ private void writeCloseDelimitedFullResponse(ChannelHandlerContext ctx) { .addListener(future -> closeDelimitedFullResponseWrites.offer(ctx)); } + private void writeKnownLengthByteBufResponse(ChannelHandlerContext ctx) { + byte[] body = "first".getBytes(UTF_8); + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(CONNECTION, "close"); + response.headers().set(CONTENT_LENGTH, body.length); + ctx.write(response); + ctx.writeAndFlush(Unpooled.wrappedBuffer(body)) + .addListener(future -> knownLengthByteBufWrites.offer(ctx)); + } + + private void writeKnownLengthFileRegionResponse(ChannelHandlerContext ctx) throws IOException { + byte[] body = "first".getBytes(UTF_8); + Path path = Files.createTempFile("netty-known-length", ".body"); + Files.write(path, body); + FileChannel fileChannel = FileChannel.open(path, StandardOpenOption.READ); + + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(CONNECTION, "close"); + response.headers().set(CONTENT_LENGTH, body.length); + ctx.write(response); + ctx.writeAndFlush(new DefaultFileRegion(fileChannel, 0, body.length)) + .addListener( + future -> { + close(fileChannel); + delete(path); + knownLengthFileRegionWrites.offer(ctx); + }); + } + + private void writeIncompleteKnownLengthResponse(ChannelHandlerContext ctx) { + byte[] body = "first".getBytes(UTF_8); + DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + response.headers().set(CONNECTION, "close"); + response.headers().set(CONTENT_LENGTH, body.length + 1); + ctx.write(response); + ctx.writeAndFlush(Unpooled.wrappedBuffer(body)) + .addListener(future -> incompleteKnownLengthWrites.offer(ctx)); + } + private void writeHeaderOnlyResponse( ChannelHandlerContext ctx, String path, HttpResponseStatus status) { ctx.writeAndFlush(new DefaultHttpResponse(HTTP_1_1, status)) @@ -505,6 +609,30 @@ private ChannelHandlerContext awaitCloseDelimitedFullResponseWritten() return responseContext; } + private ChannelHandlerContext awaitKnownLengthByteBufWritten() throws InterruptedException { + ChannelHandlerContext responseContext = knownLengthByteBufWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write the known-length ByteBuf response"); + } + return responseContext; + } + + private ChannelHandlerContext awaitKnownLengthFileRegionWritten() throws InterruptedException { + ChannelHandlerContext responseContext = knownLengthFileRegionWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write the known-length FileRegion response"); + } + return responseContext; + } + + private ChannelHandlerContext awaitIncompleteKnownLengthWritten() throws InterruptedException { + ChannelHandlerContext responseContext = incompleteKnownLengthWrites.poll(5, SECONDS); + if (responseContext == null) { + throw new AssertionError("server did not write the incomplete known-length response"); + } + return responseContext; + } + private ChannelHandlerContext awaitRequestWithoutResponse() throws InterruptedException { ChannelHandlerContext responseContext = requestsWithoutResponse.poll(5, SECONDS); if (responseContext == null) { @@ -520,5 +648,19 @@ private void awaitHeaderOnlyResponseWritten(String path) throws InterruptedExcep } assertEquals(path, responsePath, "server wrote header-only response for unexpected path"); } + + private static void close(FileChannel fileChannel) { + try { + fileChannel.close(); + } catch (IOException ignored) { + } + } + + private static void delete(Path path) { + try { + Files.deleteIfExists(path); + } catch (IOException ignored) { + } + } } } 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 9836d1112d3..997d59abfc2 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 @@ -202,6 +202,8 @@ private static boolean isPoisoned(final Deque contexts) { private boolean beforeFinishCalled; private boolean responseAnalyzed; private boolean responseBlocked; + private long responseContentLength = -1; + private long responseBodyBytes; public Context tracingContext() { return tracingContext; @@ -235,6 +237,23 @@ public void markResponseCloseDelimited() { responseCloseDelimited = true; } + public void setResponseContentLength(final long contentLength) { + responseContentLength = contentLength; + responseBodyBytes = 0; + } + + public boolean recordResponseBodyBytes(final long bodyBytes) { + if (responseContentLength < 0 || bodyBytes <= 0) { + return false; + } + if (Long.MAX_VALUE - responseBodyBytes < bodyBytes) { + responseBodyBytes = Long.MAX_VALUE; + } else { + responseBodyBytes += bodyBytes; + } + return responseBodyBytes >= responseContentLength; + } + public boolean isBeforeFinishCalled() { return beforeFinishCalled; } From 3ebe9c3de3fb2dcc83d44e2c7a6839054ac84638 Mon Sep 17 00:00:00 2001 From: Yury Gribkov Date: Thu, 16 Jul 2026 22:54:39 -0700 Subject: [PATCH 11/11] Fix Netty IAST reporting on async response completion Reactivate the stored request context while finishing Netty server spans so IAST request-ended callbacks run with the request span active. This lets sink reports attach to the request span when response write promises complete asynchronously or on channel close. Add Netty regressions covering IAST request-ended callbacks for known-length full responses and delayed write-promise completion. --- .../HttpServerRequestTracingHandler.java | 22 ++- .../HttpServerResponseTracingHandler.java | 6 +- .../server/NettyChunkedResponseSpanTest.java | 187 ++++++++++++++++++ 3 files changed, 203 insertions(+), 12 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 1e2ffa28ba1..0f93dec3101 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 @@ -115,17 +115,19 @@ private static void finishSpanOnChannelClose(final ServerRequestContext serverCo if (span == null) { return; } - if (!serverContext.isResponseCloseDelimited()) { - // The response declared a body via Content-Length or chunked encoding, but the channel closed - // before that body completed. Close-delimited responses, in contrast, end normally when the - // connection closes. - DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE)); - } - if (!serverContext.isBeforeFinishCalled()) { - serverContext.markBeforeFinishCalled(); - DECORATE.beforeFinish(storedContext); + try (final ContextScope ignored = storedContext.attach()) { + if (!serverContext.isResponseCloseDelimited()) { + // The response declared a body via Content-Length or chunked encoding, but the channel + // closed before that body completed. Close-delimited responses, in contrast, end normally + // when the connection closes. + DECORATE.onError(span, new IllegalStateException(INCOMPLETE_RESPONSE_MESSAGE)); + } + if (!serverContext.isBeforeFinishCalled()) { + serverContext.markBeforeFinishCalled(); + DECORATE.beforeFinish(storedContext); + } + span.finish(); } - span.finish(); } private static void publishSpanOnChannelClose(final Context storedContext) { 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 58b3fe9fc23..78e1cf7bfd8 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 @@ -189,8 +189,10 @@ private static void finishSpan( 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 + try (final ContextScope ignored = storedContext.attach()) { + beforeFinish(serverContext, storedContext); + span.finish(); // Finish the span manually since finishSpanOnClose was false + } } private static void beforeFinish( 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 da6547148e4..c2952606c41 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 @@ -22,19 +22,35 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.api.function.TriConsumer; +import datadog.trace.api.gateway.Events; +import datadog.trace.api.gateway.Flow; +import datadog.trace.api.gateway.IGSpanInfo; +import datadog.trace.api.gateway.RequestContext; +import datadog.trace.api.gateway.RequestContextSlot; +import datadog.trace.api.gateway.Subscription; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; +import datadog.trace.bootstrap.instrumentation.api.TagContext; +import datadog.trace.instrumentation.netty41.ServerRequestContext; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; +import io.netty.channel.ChannelOutboundHandlerAdapter; +import io.netty.channel.ChannelPromise; import io.netty.channel.DefaultFileRegion; import io.netty.channel.SimpleChannelInboundHandler; +import io.netty.channel.embedded.EmbeddedChannel; import io.netty.handler.codec.http.DefaultFullHttpResponse; import io.netty.handler.codec.http.DefaultHttpContent; +import io.netty.handler.codec.http.DefaultHttpHeaders; 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 io.netty.util.ReferenceCountUtil; import java.io.IOException; import java.net.Socket; import java.nio.channels.FileChannel; @@ -43,6 +59,10 @@ import java.nio.file.StandardOpenOption; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.BiFunction; +import java.util.function.Supplier; import java.util.regex.Pattern; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; @@ -60,6 +80,7 @@ public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport { private static final String NO_RESPONSE_PATH = "/no-response"; private static final String CLOSE_DELIMITED_PATH = "/close-delimited"; private static final String CLOSE_DELIMITED_FULL_RESPONSE_PATH = "/close-delimited/full"; + private static final String KNOWN_LENGTH_FULL_RESPONSE_PATH = "/known-length/full"; private static final String KNOWN_LENGTH_BYTE_BUF_PATH = "/known-length/bytebuf"; private static final String KNOWN_LENGTH_FILE_REGION_PATH = "/known-length/file-region"; private static final String INCOMPLETE_KNOWN_LENGTH_PATH = "/known-length/incomplete"; @@ -173,6 +194,132 @@ void finishesServerSpanWithoutErrorForKnownLengthByteBufResponseClosedAfterBody( assertTraces(trace(serverSpan(KNOWN_LENGTH_BYTE_BUF_PATH))); } + @Test + void callsIastRequestEndedForKnownLengthFullResponse() throws Exception { + Object iastRequestData = new Object(); + AtomicReference authorization = new AtomicReference<>(); + AtomicBoolean requestEnded = new AtomicBoolean(); + AtomicBoolean requestSpanActive = new AtomicBoolean(); + + Events events = Events.get(); + Subscription requestStarted = + AgentTracer.get() + .getSubscriptionService(RequestContextSlot.IAST) + .registerCallback( + events.requestStarted(), + new Supplier>() { + @Override + public Flow get() { + return new Flow.ResultFlow<>(iastRequestData); + } + }); + Subscription requestHeader = + AgentTracer.get() + .getSubscriptionService(RequestContextSlot.IAST) + .registerCallback( + events.requestHeader(), + new TriConsumer() { + @Override + public void accept(RequestContext context, String key, String value) { + if (iastRequestData == context.getData(RequestContextSlot.IAST) + && "authorization".equalsIgnoreCase(key)) { + authorization.set(value); + } + } + }); + Subscription requestEnd = + AgentTracer.get() + .getSubscriptionService(RequestContextSlot.IAST) + .registerCallback( + events.requestEnded(), + new BiFunction>() { + @Override + public Flow apply(RequestContext context, IGSpanInfo span) { + if (iastRequestData == context.getData(RequestContextSlot.IAST)) { + requestEnded.set(true); + AgentSpan activeSpan = AgentTracer.activeSpan(); + requestSpanActive.set( + activeSpan != null && activeSpan.getRequestContext() == context); + } + return Flow.ResultFlow.empty(); + } + }); + + try { + try (Socket socket = connect()) { + socket + .getOutputStream() + .write( + requestWithAuthorization(KNOWN_LENGTH_FULL_RESPONSE_PATH, "Basic dXNlcjpwYXNz") + .getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + assertEquals("ok", readHttpResponseBody(socket.getInputStream())); + } + + assertTraces(trace(serverSpan(KNOWN_LENGTH_FULL_RESPONSE_PATH))); + assertEquals("Basic dXNlcjpwYXNz", authorization.get()); + assertTrue(requestEnded.get(), "IAST requestEnded callback was not called"); + assertTrue(requestSpanActive.get(), "request span was not active during requestEnded"); + } finally { + requestEnd.cancel(); + requestHeader.cancel(); + requestStarted.cancel(); + } + } + + @Test + void activatesRequestSpanWhenWritePromiseCompletesAfterWriteReturns() { + Object iastRequestData = new Object(); + AtomicBoolean requestSpanActive = new AtomicBoolean(); + + Events events = Events.get(); + Subscription requestEnd = + AgentTracer.get() + .getSubscriptionService(RequestContextSlot.IAST) + .registerCallback( + events.requestEnded(), + new BiFunction>() { + @Override + public Flow apply(RequestContext context, IGSpanInfo span) { + if (iastRequestData == context.getData(RequestContextSlot.IAST)) { + AgentSpan activeSpan = AgentTracer.activeSpan(); + requestSpanActive.set( + activeSpan != null && activeSpan.getRequestContext() == context); + } + return Flow.ResultFlow.empty(); + } + }); + + HoldingOutboundHandler holdingHandler = new HoldingOutboundHandler(); + EmbeddedChannel channel = + new EmbeddedChannel(holdingHandler, HttpServerResponseTracingHandler.INSTANCE); + AgentSpan span = + AgentTracer.get() + .startSpan( + "netty", + "netty.request", + new TagContext().withRequestContextDataIast(iastRequestData)); + ServerRequestContext.add(channel, span, new DefaultHttpHeaders()); + + try { + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.copiedBuffer("ok", UTF_8)); + response.headers().set(CONTENT_LENGTH, response.content().readableBytes()); + + channel.writeOutbound(response); + assertFalse(requestSpanActive.get(), "write promise completed synchronously"); + + holdingHandler.completeWrite(); + + assertTrue(requestSpanActive.get(), "request span was not active during requestEnded"); + } finally { + holdingHandler.completeWrite(); + requestEnd.cancel(); + channel.finishAndReleaseAll(); + } + } + @Test void finishesServerSpanWithoutErrorForKnownLengthFileRegionResponseClosedAfterBody() throws Exception { @@ -381,6 +528,14 @@ private static String request(String path) { return request("GET", path); } + private static String requestWithAuthorization(String path, String authorization) { + return "GET " + + path + + " HTTP/1.1\r\nHost: localhost\r\nAuthorization: " + + authorization + + "\r\n\r\n"; + } + private static String request(String method, String path) { String headers = method + " " + path + " HTTP/1.1\r\nHost: localhost\r\n"; if (isWebSocketPath(path)) { @@ -474,6 +629,8 @@ protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) thro writeCloseDelimitedResponse(ctx); } else if (CLOSE_DELIMITED_FULL_RESPONSE_PATH.equals(request.uri())) { writeCloseDelimitedFullResponse(ctx); + } else if (KNOWN_LENGTH_FULL_RESPONSE_PATH.equals(request.uri())) { + writeKnownLengthFullResponse(ctx); } else if (KNOWN_LENGTH_BYTE_BUF_PATH.equals(request.uri())) { writeKnownLengthByteBufResponse(ctx); } else if (KNOWN_LENGTH_FILE_REGION_PATH.equals(request.uri())) { @@ -517,6 +674,13 @@ private void writeCloseDelimitedFullResponse(ChannelHandlerContext ctx) { .addListener(future -> closeDelimitedFullResponseWrites.offer(ctx)); } + private void writeKnownLengthFullResponse(ChannelHandlerContext ctx) { + DefaultFullHttpResponse response = + new DefaultFullHttpResponse(HTTP_1_1, OK, Unpooled.copiedBuffer("ok", UTF_8)); + response.headers().set(CONTENT_LENGTH, response.content().readableBytes()); + ctx.writeAndFlush(response); + } + private void writeKnownLengthByteBufResponse(ChannelHandlerContext ctx) { byte[] body = "first".getBytes(UTF_8); DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); @@ -663,4 +827,27 @@ private static void delete(Path path) { } } } + + private static final class HoldingOutboundHandler extends ChannelOutboundHandlerAdapter { + private Object msg; + private ChannelPromise promise; + + @Override + public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) { + this.msg = msg; + this.promise = promise; + } + + private void completeWrite() { + if (promise == null || promise.isDone()) { + return; + } + try { + promise.setSuccess(); + } finally { + ReferenceCountUtil.release(msg); + msg = null; + } + } + } }