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/HttpServerRequestTracingHandler.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerRequestTracingHandler.java index 0d934772d56..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 @@ -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,49 @@ 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()) { + finishSpanOnChannelClose(storedContext); + } else { + publishSpanOnChannelClose(storedContext.tracingContext()); + } + } + } } catch (final Throwable ignored) { } } } + + private static void finishSpanOnChannelClose(final ServerRequestContext serverContext) { + final Context storedContext = serverContext.tracingContext(); + final AgentSpan span = AgentSpan.fromContext(storedContext); + if (span == null) { + return; + } + 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(); + } + } + + 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..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 @@ -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; @@ -8,13 +9,20 @@ 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; 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 +30,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 +39,167 @@ 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 boolean knownLengthResponseComplete = + !terminalResponse && serverContext.recordResponseBodyBytes(responseBodyBytes(msg)); + final boolean finishResponseOnWrite = terminalResponse || knownLengthResponseComplete; + final ChannelPromise writePromise = + finishResponseOnWrite && prm.isVoid() ? ctx.newPromise() : prm; try { - ctx.write(msg, prm); + if (finishResponseOnWrite) { + ServerRequestContext.remove(ctx.channel(), serverContext); + writePromise.addListener( + future -> finishSpan(serverContext, 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 (!finishResponseOnWrite || !writePromise.isDone()) { + DECORATE.onError(span, throwable); + span.setHttpStatusCode(500); + if (!finishResponseOnWrite) { + 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(); + 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. + if (isCloseDelimitedHttp1Response(ctx, response)) { + serverContext.markResponseCloseDelimited(); + return false; + } + if (msg instanceof LastHttpContent) { + return true; } + return false; + } + return serverContext.isResponseStarted() + && !serverContext.isResponseCloseDelimited() + && 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().code() == HttpResponseStatus.SWITCHING_PROTOCOLS.code() + && 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) + || (contentLength(response) == 0 && !HttpUtil.isTransferEncodingChunked(response)); + } + + 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()); + } + + 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 + * 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( + 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) { + try (final ContextScope ignored = storedContext.attach()) { + 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/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/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-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..c2952606c41 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -0,0 +1,853 @@ +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 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; +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.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; + +@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 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"; + 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(); + + @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 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 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 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 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 { + 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; + 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); + } + + @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); + } + + 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 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)) { + 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 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); + } + + 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 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) throws Exception { + 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 (CLOSE_DELIMITED_PATH.equals(request.uri())) { + 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())) { + 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())) { + writeHeaderOnlyWebSocketUpgrade(ctx, request.uri(), CUSTOM_SWITCHING_PROTOCOLS); + } 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 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 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 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); + 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)) + .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, 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)); + } + + 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 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 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) { + 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"); + } + + 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) { + } + } + } + + 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; + } + } + } +} 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/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"); 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..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 @@ -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,15 @@ 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)); + } + + /** 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; @@ -173,7 +196,14 @@ private static boolean isPoisoned(final Deque contexts) { private final Context tracingContext; private final String acceptHeader; + private final HttpMethod requestMethod; + private boolean responseStarted; + private boolean responseCloseDelimited; + private boolean beforeFinishCalled; private boolean responseAnalyzed; + private boolean responseBlocked; + private long responseContentLength = -1; + private long responseBodyBytes; public Context tracingContext() { return tracingContext; @@ -183,6 +213,55 @@ 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 isResponseCloseDelimited() { + return responseCloseDelimited; + } + + 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; + } + + public void markBeforeFinishCalled() { + beforeFinishCalled = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; } @@ -191,8 +270,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.acceptHeader = requestHeaders == null ? null : requestHeaders.get("accept"); + this.requestMethod = requestMethod; } }