diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java index 7e7dfcb1e89..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 @@ -15,10 +15,13 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.http.HttpHeaders; import io.netty.handler.codec.http.HttpRequest; +import java.util.Deque; @ChannelHandler.Sharable public class HttpServerRequestTracingHandler extends ChannelInboundHandlerAdapter { public static HttpServerRequestTracingHandler INSTANCE = new HttpServerRequestTracingHandler(); + private static final String INCOMPLETE_RESPONSE_MESSAGE = + "Channel closed before response completed"; @Override public void channelRead(final ChannelHandlerContext ctx, final Object msg) { @@ -54,7 +57,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) { DECORATE.onRequest(span, channel, request, parentContext); final ServerRequestContext serverContext = - ServerRequestContext.add(channel, context, request.headers()); + ServerRequestContext.add(channel, context, request); Flow.Action.RequestBlockingAction rba = span.getRequestBlockingAction(); if (rba != null) { @@ -89,9 +92,41 @@ 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); + } else { + publishSpanOnChannelClose(storedContext.tracingContext()); + } + } + } } catch (final Throwable ignored) { } } } + + 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)); + if (!serverContext.isBeforeFinishCalled()) { + serverContext.markBeforeFinishCalled(); + 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..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 @@ -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,110 @@ 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(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); - span.finish(); // Finish the span manually since finishSpanOnClose was false - ServerRequestContext.remove(ctx.channel(), serverContext); + if (!terminalResponse) { + ServerRequestContext.remove(ctx.channel(), serverContext); + } + finishSpan(serverContext, 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 ServerRequestContext serverContext, + final Context storedContext, + final AgentSpan span, + final Future future) { + if (!future.isSuccess()) { + DECORATE.onError(span, future.cause()); + span.setHttpStatusCode(500); + } + finishSpan(serverContext, storedContext, span); + } + + 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-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..75d0eb1a69b --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -0,0 +1,429 @@ +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 NO_RESPONSE_PATH = "/no-response"; + 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"; + 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 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; + 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); + } + + 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"; + } + 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<>(); + private final BlockingQueue requestsWithoutResponse = + 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 (NO_RESPONSE_PATH.equals(request.uri())) { + requestsWithoutResponse.offer(ctx); + } else if (WEBSOCKET_PATH.equals(request.uri())) { + writeHeaderOnlyWebSocketUpgrade(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 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 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) { + 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; + } + } +} diff --git a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java index 4a8ea3e59e7..82a67ae85f1 100644 --- a/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java +++ b/dd-java-agent/instrumentation/netty/netty-common/src/main/java/datadog/trace/instrumentation/netty41/ServerRequestContext.java @@ -5,6 +5,8 @@ import datadog.context.Context; import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import io.netty.handler.codec.http.HttpHeaders; +import io.netty.handler.codec.http.HttpMethod; +import io.netty.handler.codec.http.HttpRequest; import io.netty.util.AttributeKey; import io.netty.util.AttributeMap; import java.util.ArrayDeque; @@ -24,11 +26,26 @@ public static boolean canTrackRequest(final AttributeMap attributes) { /** Adds a request context to the queue tail. */ public static ServerRequestContext add( final AttributeMap attributes, final Context context, final HttpHeaders requestHeaders) { + return add(attributes, context, requestHeaders, null); + } + + /** Adds a request context to the queue tail. */ + public static ServerRequestContext add( + final AttributeMap attributes, final Context context, final HttpRequest request) { + return add(attributes, context, request.headers(), request.method()); + } + + private static ServerRequestContext add( + final AttributeMap attributes, + final Context context, + final HttpHeaders requestHeaders, + final HttpMethod requestMethod) { final Deque contexts = getOrCreate(attributes); if (!canAdd(attributes, contexts)) { return null; } - final ServerRequestContext serverContext = new ServerRequestContext(context, requestHeaders); + final ServerRequestContext serverContext = + new ServerRequestContext(context, requestHeaders, requestMethod); contexts.addLast(serverContext); // The deque is authoritative for server request/response matching. CONTEXT_ATTRIBUTE_KEY is a // legacy mirror of the current inbound request used by generic fire* activation. @@ -78,9 +95,14 @@ public static void remove( /** Closes all pending request contexts on channel close. */ public static void closeAll(final AttributeMap attributes) { + close(removeAll(attributes)); + } + + /** Removes all pending request contexts. */ + public static Deque removeAll(final AttributeMap attributes) { // The legacy mirror must not outlive the authoritative request queue. attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); - close(attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove()); + return attributes.attr(SERVER_REQUEST_CONTEXTS_ATTRIBUTE_KEY).getAndRemove(); } private static final int PIPELINING_LIMIT = 1000; @@ -153,6 +175,9 @@ private static boolean isPoisoned(final Deque contexts) { private final Context tracingContext; private final HttpHeaders requestHeaders; + private final HttpMethod requestMethod; + private boolean responseStarted; + private boolean beforeFinishCalled; private boolean responseAnalyzed; private boolean responseBlocked; @@ -164,6 +189,30 @@ public HttpHeaders requestHeaders() { return requestHeaders; } + public boolean isHeadRequest() { + return HttpMethod.HEAD.equals(requestMethod); + } + + public boolean isConnectRequest() { + return HttpMethod.CONNECT.equals(requestMethod); + } + + public boolean isResponseStarted() { + return responseStarted; + } + + public void markResponseStarted() { + responseStarted = true; + } + + public boolean isBeforeFinishCalled() { + return beforeFinishCalled; + } + + public void markBeforeFinishCalled() { + beforeFinishCalled = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; } @@ -180,8 +229,12 @@ public void markResponseBlocked() { responseBlocked = true; } - private ServerRequestContext(final Context tracingContext, final HttpHeaders requestHeaders) { + private ServerRequestContext( + final Context tracingContext, + final HttpHeaders requestHeaders, + final HttpMethod requestMethod) { this.tracingContext = tracingContext; this.requestHeaders = requestHeaders; + this.requestMethod = requestMethod; } }