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 edbd549fe70..2d61271ddc0 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,6 +15,7 @@ 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 { @@ -95,9 +96,43 @@ 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.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 47aa25f7015..9f330301f28 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 @@ -14,9 +14,11 @@ 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.LastHttpContent; +import io.netty.util.concurrent.Future; @ChannelHandler.Sharable public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter { @@ -25,7 +27,8 @@ public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdap @Override public void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise prm) { final boolean isResponse = msg instanceof HttpResponse; - if (!isResponse && !(msg instanceof LastHttpContent)) { + final boolean isLastContent = msg instanceof LastHttpContent; + if (!isResponse && !isLastContent) { ctx.write(msg, prm); return; } @@ -50,43 +53,99 @@ public void write(final ChannelHandlerContext ctx, final Object msg, final Chann return; } - try (final ContextScope scope = storedContext.attach()) { + try (final ContextScope ignored = storedContext.attach()) { final HttpResponse response = isResponse ? (HttpResponse) msg : null; - final boolean responseComplete = msg instanceof LastHttpContent; - + final boolean websocketUpgrade = response != null && isWebsocketUpgrade(response); + final boolean informationalResponse = + response != null && isInformationalResponse(response) && !websocketUpgrade; + final boolean finishResponseOnWrite = isLastContent && !informationalResponse; + final ChannelPromise writePromise = + finishResponseOnWrite && prm.isVoid() ? ctx.newPromise() : prm; try { - ctx.write(msg, prm); - } catch (final Throwable throwable) { - DECORATE.onError(span, throwable); - span.setHttpStatusCode(500); - span.finish(); // Finish the span manually since finishSpanOnClose was false - removeServerContext(ctx, serverContext); - throw throwable; - } - if (response != null) { - final boolean isWebsocketUpgrade = - response.status() == HttpResponseStatus.SWITCHING_PROTOCOLS - && "websocket".equals(response.headers().get(HttpHeaderNames.UPGRADE)); - if (isWebsocketUpgrade) { - ctx.channel() - .attr(WEBSOCKET_SENDER_HANDLER_CONTEXT) - .set(new HandlerContext.Sender(span, ctx.channel().id().asShortText())); + if (response != null && !informationalResponse) { + onResponse(ctx, span, serverContext, response, websocketUpgrade); } - if (isInformational(response) && !isWebsocketUpgrade) { - return; + if (finishResponseOnWrite) { + removeServerContext(ctx, serverContext); + writePromise.addListener( + future -> finishSpan(serverContext, storedContext, span, future)); } - if (serverContext != null) { - serverContext.markResponseStarted(); + ctx.write(msg, writePromise); + if (finishResponseOnWrite && (!writePromise.isDone() || writePromise.isSuccess())) { + final ServerRequestContext nextResponse = + ServerRequestContext.nextResponse(ctx.channel()); + BlockingResponseHandler.maybeWriteDeferredBlockResponse(ctx, nextResponse); } - DECORATE.onResponse(span, response); + } catch (final Throwable throwable) { + if (!finishResponseOnWrite || !writePromise.isDone()) { + DECORATE.onError(span, throwable); + span.setHttpStatusCode(500); + if (!finishResponseOnWrite) { + removeServerContext(ctx, serverContext); + } + finishSpan(serverContext, storedContext, span); + } + throw throwable; } - if (responseComplete) { - DECORATE.beforeFinish(scope.context()); - span.finish(); // Finish the span manually since finishSpanOnClose was false - removeServerContext(ctx, serverContext); - final ServerRequestContext nextResponse = ServerRequestContext.nextResponse(ctx.channel()); - BlockingResponseHandler.maybeWriteDeferredBlockResponse(ctx, nextResponse); + } + } + + private static void onResponse( + final ChannelHandlerContext ctx, + final AgentSpan span, + final ServerRequestContext serverContext, + final HttpResponse response, + final boolean websocketUpgrade) { + if (websocketUpgrade) { + ctx.channel() + .attr(WEBSOCKET_SENDER_HANDLER_CONTEXT) + .set(new HandlerContext.Sender(span, ctx.channel().id().asShortText())); + } + DECORATE.onResponse(span, response); + if (serverContext != null) { + serverContext.markResponseStarted(); + } + } + + 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 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 == null || !serverContext.isBeforeFinishCalled()) { + if (serverContext != null) { + serverContext.markBeforeFinishCalled(); } + DECORATE.beforeFinish(storedContext); } } @@ -98,9 +157,4 @@ private static void removeServerContext( ServerRequestContext.remove(ctx.channel(), serverContext); } } - - private static boolean isInformational(final HttpResponse response) { - final int statusCode = response.status().code(); - return statusCode >= 100 && statusCode < 200; - } } 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 c9e18635dfe..11782d59e02 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 @@ -70,6 +70,10 @@ public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise prm) thr } HttpResponse origResponse = (HttpResponse) msg; 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()) { diff --git a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandlerTest.java b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandlerTest.java index 656a082fcdc..d9d90f4d83a 100644 --- a/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandlerTest.java +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandlerTest.java @@ -13,6 +13,7 @@ import static io.netty.handler.codec.http.HttpResponseStatus.OK; import static io.netty.handler.codec.http.HttpResponseStatus.SWITCHING_PROTOCOLS; import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertSame; @@ -39,16 +40,22 @@ class HttpServerResponseTracingHandlerTest extends AbstractInstrumentationTest { @Test - void finishesMirroredContextWhenRequestQueueIsAbsent() { + void finishesMirroredContextOnLastContentWhenRequestQueueIsAbsent() { EmbeddedChannel channel = new EmbeddedChannel(HttpServerResponseTracingHandler.INSTANCE); AgentSpan span = startSpan("netty", "mirrored-http2-server"); channel.attr(CONTEXT_ATTRIBUTE_KEY).set(span); - assertTrue(channel.writeOutbound(new DefaultFullHttpResponse(HTTP_1_1, OK))); + HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); + assertTrue(channel.writeOutbound(response)); - FullHttpResponse response = channel.readOutbound(); - assertNotNull(response); - response.release(); + assertSame(span, channel.attr(CONTEXT_ATTRIBUTE_KEY).get()); + HttpResponse forwarded = channel.readOutbound(); + assertSame(response, forwarded); + ReferenceCountUtil.release(forwarded); + + assertTrue(channel.writeOutbound(LastHttpContent.EMPTY_LAST_CONTENT)); + + ReferenceCountUtil.release(channel.readOutbound()); assertNull(channel.attr(CONTEXT_ATTRIBUTE_KEY).get()); channel.finishAndReleaseAll(); assertTraces(trace(span().root().operationName("mirrored-http2-server"))); @@ -70,7 +77,7 @@ void forwardsLastContentBeforeFinalResponseWithoutCompletingContext() { } @Test - void headerOnlyResponseDoesNotCompleteContextBeforeLastContent() { + void headerOnlyResponseWaitsForLastContent() { EmbeddedChannel channel = new EmbeddedChannel(HttpServerResponseTracingHandler.INSTANCE); AgentSpan span = startSpan("netty", "header-only-server"); ServerRequestContext serverContext = ServerRequestContext.add(channel, span, null); @@ -84,6 +91,7 @@ void headerOnlyResponseDoesNotCompleteContextBeforeLastContent() { HttpResponse forwarded = channel.readOutbound(); assertSame(response, forwarded); ReferenceCountUtil.release(forwarded); + assertNull(channel.readOutbound()); assertTrue(channel.writeOutbound(LastHttpContent.EMPTY_LAST_CONTENT)); @@ -95,10 +103,12 @@ void headerOnlyResponseDoesNotCompleteContextBeforeLastContent() { } @Test - void rawFixedLengthBodyDoesNotCompleteContextBeforeLastContent() { + void rawFixedLengthBodyWaitsForLastContent() { EmbeddedChannel channel = new EmbeddedChannel(HttpServerResponseTracingHandler.INSTANCE); AgentSpan span = startSpan("netty", "raw-body-server"); ServerRequestContext serverContext = ServerRequestContext.add(channel, span, null); + ServerRequestContext nextServerContext = + ServerRequestContext.add(channel, Context.root(), null); HttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK); response.headers().set(CONTENT_LENGTH, 4); @@ -108,15 +118,34 @@ void rawFixedLengthBodyDoesNotCompleteContextBeforeLastContent() { assertSame(serverContext, ServerRequestContext.nextResponse(channel)); ReferenceCountUtil.release(channel.readOutbound()); + assertNull(channel.readOutbound()); assertTrue(channel.writeOutbound(LastHttpContent.EMPTY_LAST_CONTENT)); - assertNull(ServerRequestContext.nextResponse(channel)); + assertSame(nextServerContext, ServerRequestContext.nextResponse(channel)); ReferenceCountUtil.release(channel.readOutbound()); + ServerRequestContext.remove(channel, nextServerContext); channel.finishAndReleaseAll(); assertTraces(trace(span().root().operationName("raw-body-server"))); } + @Test + void doesNotThrowOnMalformedContentLength() { + EmbeddedChannel channel = new EmbeddedChannel(HttpServerResponseTracingHandler.INSTANCE); + AgentSpan span = startSpan("netty", "malformed-content-length-server"); + ServerRequestContext.add(channel, span, null); + FullHttpResponse response = new DefaultFullHttpResponse(HTTP_1_1, OK); + response.headers().set(CONTENT_LENGTH, "malformed"); + + assertDoesNotThrow(() -> assertTrue(channel.writeOutbound(response))); + + FullHttpResponse forwarded = channel.readOutbound(); + assertSame(response, forwarded); + forwarded.release(); + channel.finishAndReleaseAll(); + assertTraces(trace(span().root().operationName("malformed-content-length-server"))); + } + @Test void nettyEncoderRequiresLastContentBeforeNextKeepAliveResponse() { EmbeddedChannel channel = new EmbeddedChannel(new HttpResponseEncoder()); @@ -159,4 +188,27 @@ void fullWebsocketUpgradeCompletesContextAndPreservesHandshakeSpan() { channel.finishAndReleaseAll(); assertTraces(trace(span().root().operationName("websocket-handshake-server"))); } + + @Test + void fullNonWebsocketUpgradeWaitsForFinalResponse() { + EmbeddedChannel channel = new EmbeddedChannel(HttpServerResponseTracingHandler.INSTANCE); + AgentSpan span = startSpan("netty", "h2c-upgrade-server"); + ServerRequestContext serverContext = ServerRequestContext.add(channel, span, null); + FullHttpResponse upgradeResponse = new DefaultFullHttpResponse(HTTP_1_1, SWITCHING_PROTOCOLS); + upgradeResponse.headers().set(UPGRADE, "h2c"); + + assertTrue(channel.writeOutbound(upgradeResponse)); + + assertSame(serverContext, ServerRequestContext.nextResponse(channel)); + assertNull(channel.attr(WEBSOCKET_SENDER_HANDLER_CONTEXT).get()); + ReferenceCountUtil.release(channel.readOutbound()); + + assertTrue(channel.writeOutbound(new DefaultFullHttpResponse(HTTP_1_1, OK))); + + assertNull(ServerRequestContext.nextResponse(channel)); + ReferenceCountUtil.release(channel.readOutbound()); + channel.finishAndReleaseAll(); + + assertTraces(trace(span().root().operationName("h2c-upgrade-server"))); + } } 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..a68de06d005 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java @@ -0,0 +1,938 @@ +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.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +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.bootstrap.instrumentation.api.Tags; +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.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 finishesServerSpanWithoutFramingErrorWhenConnectionDrops() 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))); + } + + @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 finishesCloseDelimitedFullResponseOnLastContent() 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, 5); + + closeChannel(responseContext); + } + + assertTrue( + reportedBeforeConnectionClose, + "a FullHttpResponse should finish the server span before the connection closes"); + assertTraces(trace(serverSpan(CLOSE_DELIMITED_FULL_RESPONSE_PATH))); + } + + @Test + void knownLengthByteBufResponseWaitsForLastContent() throws Exception { + boolean reportedBeforeLastContent; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(KNOWN_LENGTH_BYTE_BUF_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitKnownLengthByteBufWritten(); + reportedBeforeLastContent = writer.waitForTracesMax(1, 1); + writeLastChunk(responseContext); + assertEquals("first", readHttpResponseBody(socket.getInputStream())); + assertTrue( + writer.waitForTracesMax(1, 5), "server span should be reported after LastHttpContent"); + } + + assertFalse( + reportedBeforeLastContent, + "Content-Length body bytes should not finish the span before LastHttpContent"); + 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, null); + + 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 marksSpanErrorWhenWritePromiseFailsAfterWriteReturns() { + 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, null); + IOException writeFailure = new IOException("delayed write failure"); + + 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"); + assertNull(ServerRequestContext.nextResponse(channel)); + + holdingHandler.failWrite(writeFailure); + assertThrows(IOException.class, channel::checkException); + + assertTrue(requestSpanActive.get(), "request span was not active during requestEnded"); + assertEquals(500, span.getTag(Tags.HTTP_STATUS)); + assertTraces(trace(span().root().operationName("netty.request").error())); + } finally { + holdingHandler.failWrite(writeFailure); + requestEnd.cancel(); + channel.finishAndReleaseAll(); + } + } + + @Test + void knownLengthFileRegionResponseWaitsForLastContent() throws Exception { + boolean reportedBeforeLastContent; + try (Socket socket = connect()) { + socket.getOutputStream().write(request(KNOWN_LENGTH_FILE_REGION_PATH).getBytes(US_ASCII)); + socket.getOutputStream().flush(); + + ChannelHandlerContext responseContext = handler.awaitKnownLengthFileRegionWritten(); + reportedBeforeLastContent = writer.waitForTracesMax(1, 1); + writeLastChunk(responseContext); + assertEquals("first", readHttpResponseBody(socket.getInputStream())); + assertTrue( + writer.waitForTracesMax(1, 5), "server span should be reported after LastHttpContent"); + } + + assertFalse( + reportedBeforeLastContent, + "Content-Length body bytes should not finish the span before LastHttpContent"); + assertTraces(trace(serverSpan(KNOWN_LENGTH_FILE_REGION_PATH))); + } + + @Test + void doesNotInferFramingErrorFromContentLengthWhenConnectionCloses() 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))); + } + + @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.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT).syncUninterruptibly(); + } + + 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.write(new DefaultHttpResponse(HTTP_1_1, status)); + ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT) + .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.write(response); + ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT) + .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.write(response); + ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT) + .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() { + finishWrite(null); + } + + private void failWrite(Throwable failure) { + finishWrite(failure); + } + + private void finishWrite(Throwable failure) { + if (promise == null || promise.isDone()) { + return; + } + try { + if (failure == null) { + promise.setSuccess(); + } else { + promise.setFailure(failure); + } + } 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 70825de6453..a611fbb8cb8 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 @@ -18,11 +18,11 @@ import static java.util.Collections.emptyMap; 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.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; import datadog.appsec.api.blocking.BlockingContentType; -import datadog.trace.agent.test.AbstractInstrumentationTest; import datadog.trace.agent.test.assertions.TraceMatcher; import datadog.trace.api.function.TriFunction; import datadog.trace.api.gateway.BlockResponseFunction; @@ -34,17 +34,12 @@ import datadog.trace.bootstrap.instrumentation.api.AgentSpan; import datadog.trace.bootstrap.instrumentation.api.AgentTracer; import datadog.trace.bootstrap.instrumentation.api.URIDataAdapter; -import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelHandler; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; -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.DefaultHttpContent; import io.netty.handler.codec.http.DefaultHttpResponse; @@ -55,11 +50,6 @@ import io.netty.handler.codec.http.HttpResponseStatus; import io.netty.handler.codec.http.HttpServerCodec; import io.netty.util.ReferenceCountUtil; -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.time.Duration; import java.util.ArrayList; @@ -68,9 +58,7 @@ import java.util.concurrent.CountDownLatch; import java.util.function.Supplier; import java.util.regex.Pattern; -import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.AfterEach; -import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.TestInstance; import reactor.core.publisher.Mono; @@ -78,53 +66,29 @@ import reactor.netty.http.server.HttpServer; @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 static final HttpResponseStatus EARLY_HINTS = new HttpResponseStatus(103, "Early Hints"); - private EventLoopGroup eventLoopGroup; - private DecodedRequestObserver decodedRequestObserver; - private PipeliningHandler handler; - private int port; + private final DecodedRequestObserver decodedRequestObserver = new DecodedRequestObserver(); + private final PipeliningHandler handler = new PipeliningHandler(); private Object appSecSubscriptions; private boolean originalAppSecActive; - @BeforeAll - void startServer() throws Exception { - eventLoopGroup = new NioEventLoopGroup(); - decodedRequestObserver = new DecodedRequestObserver(); - handler = new PipeliningHandler(); - ServerBootstrap bootstrap = - new ServerBootstrap() - .group(eventLoopGroup) - .channel(NioServerSocketChannel.class) - .childHandler( - new ChannelInitializer() { - @Override - protected void initChannel(Channel ch) { - HttpServerCodec codec = new HttpServerCodec(); - ch.pipeline().addLast(codec); - ch.pipeline() - .addAfter( - ch.pipeline().context(codec).name(), - "decoded-request-observer", - decodedRequestObserver); - ch.pipeline().addLast(new HttpObjectAggregator(65536)); - ch.pipeline().addLast(handler); - } - }); - Channel channel = bootstrap.bind(0).sync().channel(); - port = ((InetSocketAddress) channel.localAddress()).getPort(); - } - - @AfterAll - void stopServer() { - if (eventLoopGroup != null) { - eventLoopGroup.shutdownGracefully(); - } + @Override + protected void configurePipeline(Channel ch) { + HttpServerCodec codec = new HttpServerCodec(); + ch.pipeline().addLast(codec); + ch.pipeline() + .addAfter( + ch.pipeline().context(codec).name(), + "decoded-request-observer", + decodedRequestObserver); + ch.pipeline().addLast(new HttpObjectAggregator(65536)); + ch.pipeline().addLast(handler); } @AfterEach @@ -140,8 +104,7 @@ void resetAppSec() { @Test void createsServerSpanForEachPipelinedRequest() throws Exception { handler.expectRequests(3); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(pipelinedRequests().getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -203,8 +166,7 @@ void reactorNettyCompletesPipelinedFixedLengthResponses() throws Exception { void requestBlockOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse() throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -221,7 +183,7 @@ void requestBlockOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse() throws assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -231,8 +193,7 @@ void additionalPipelinedRequestsBehindDeferredBlockAreIgnored() throws Exception decodedRequestObserver.expectRequests(3); handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -254,7 +215,7 @@ void additionalPipelinedRequestsBehindDeferredBlockAreIgnored() throws Exception assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); assertNull( handler.inboundException, @@ -267,8 +228,7 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierChunkedResponseCompletion throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -283,9 +243,9 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierChunkedResponseCompletion handler.writeChunkedResponse(); - assertEquals("response " + FIRST_PATH, readHttpChunkedResponseBody(socket.getInputStream())); + assertEquals("response " + FIRST_PATH, readChunkedHttpResponseBody(socket.getInputStream())); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -294,8 +254,7 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierChunkedResponseCompletion void requestBlockOnLaterPipelinedRequestFollowsEarlierHeaderOnlyResponse() throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -308,13 +267,19 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeaderOnlyResponse() throw assertTrue( blockedRequestSeen.await(5, SECONDS), "server did not block second pipelined request"); - handler.writeHeaderOnlyResponse(); + handler.writeHeaderOnlyResponseHeaders(); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 204 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 204 "), "first response should be the header-only response"); + assertFalse( + writer.waitForTracesMax(1, 1), + "header-only response should remain pending before LastHttpContent"); + + handler.writeLastContent(); + assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -323,8 +288,7 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeaderOnlyResponse() throw void requestBlockOnLaterPipelinedRequestFollowsEarlierHeadResponse() throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(headRequest(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -340,10 +304,10 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeadResponse() throws Exce handler.writeHeadResponse(); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 200 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 200 "), "first response should be the HEAD response"); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -352,8 +316,7 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeadResponse() throws Exce void lastContentAfterInterimResponseDoesNotCompleteServerSpan() throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -362,7 +325,7 @@ void lastContentAfterInterimResponseDoesNotCompleteServerSpan() throws Exception handler.writeInterimResponseWithTerminatorThenResponse(); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 100 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 100 "), "first response should be the interim response"); assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); } @@ -373,8 +336,7 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierEarlyHintsResponseComplet throws Exception { handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -390,11 +352,11 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierEarlyHintsResponseComplet handler.writeEarlyHintsWithTerminatorThenResponse(); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 103 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 103 "), "first response should be the early hints response"); assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -405,8 +367,7 @@ void blockResponseFunctionOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse( enableAppSec(); handler.expectRequests(1); - try (Socket socket = new Socket("localhost", port)) { - socket.setSoTimeout(5000); + try (Socket socket = connect()) { socket.getOutputStream().write(request(FIRST_PATH).getBytes(US_ASCII)); socket.getOutputStream().flush(); @@ -423,7 +384,7 @@ void blockResponseFunctionOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse( assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream())); assertTrue( - readHttpResponseHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), + readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "), "second response should be the deferred blocking response"); } } @@ -500,99 +461,6 @@ private static TraceMatcher serverTrace(String path) { .type("web")); } - private static String readHttpResponseBody(InputStream in) throws IOException { - String headers = readHttpResponseHeaders(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 readHttpChunkedResponseBody(InputStream in) throws IOException { - String headers = readHttpResponseHeaders(in); - assertTrue(headers.startsWith("HTTP/1.1 200 "), "unexpected response: " + headers); - - ByteArrayOutputStream body = new ByteArrayOutputStream(); - while (true) { - String chunkSizeLine = readHttpLine(in); - int chunkSize = Integer.parseInt(chunkSizeLine, 16); - if (chunkSize == 0) { - String trailer; - do { - trailer = readHttpLine(in); - } while (!trailer.isEmpty()); - return body.toString(UTF_8.name()); - } - byte[] chunk = new byte[chunkSize]; - int read = 0; - while (read < chunkSize) { - int count = in.read(chunk, read, chunkSize - read); - if (count == -1) { - throw new EOFException("response ended before chunk was complete"); - } - read += count; - } - body.write(chunk); - String chunkTerminator = readHttpLine(in); - assertEquals("", chunkTerminator, "chunk was not followed by CRLF"); - } - } - - private static String readHttpResponseHeaders(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 readHttpLine(InputStream in) throws IOException { - ByteArrayOutputStream out = new ByteArrayOutputStream(); - int previous = -1; - while (true) { - int current = in.read(); - if (current == -1) { - throw new EOFException("response ended before line was complete"); - } - if (previous == '\r' && current == '\n') { - byte[] line = out.toByteArray(); - return new String(line, 0, line.length - 1, US_ASCII); - } - out.write(current); - previous = current; - } - } - - 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); - } - @ChannelHandler.Sharable private static final class DecodedRequestObserver extends ChannelInboundHandlerAdapter { private volatile CountDownLatch receivedRequests = new CountDownLatch(0); @@ -736,7 +604,7 @@ private void writeChunkedResponse() { }); } - private void writeHeaderOnlyResponse() { + private void writeHeaderOnlyResponseHeaders() { ChannelHandlerContext responseContext = context; if (responseContext == null) { throw new IllegalStateException("no request context captured"); @@ -747,12 +615,22 @@ private void writeHeaderOnlyResponse() { () -> { DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, NO_CONTENT); response.headers().set(CONNECTION, KEEP_ALIVE); - responseContext.write(response); - responseContext.write(new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER)); - responseContext.flush(); + responseContext.writeAndFlush(response); }); } + private void writeLastContent() { + ChannelHandlerContext responseContext = context; + if (responseContext == null) { + throw new IllegalStateException("no request context captured"); + } + responseContext + .executor() + .execute( + () -> + responseContext.writeAndFlush(new DefaultLastHttpContent(Unpooled.EMPTY_BUFFER))); + } + private void writeHeadResponse() { ChannelHandlerContext responseContext = context; if (responseContext == null) { 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..d75bab18d14 --- /dev/null +++ b/dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyHttpServerTestSupport.java @@ -0,0 +1,162 @@ +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(1); + 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() { + try { + if (serverChannel != null) { + serverChannel.close().syncUninterruptibly(); + } + } finally { + if (eventLoopGroup != null) { + eventLoopGroup.shutdownGracefully().syncUninterruptibly(); + } + } + } + + 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 57c1f54e12e..654bf24548f 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 @@ -129,11 +129,16 @@ 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 context mirror must not outlive the authoritative request queue. attributes.attr(CONTEXT_ATTRIBUTE_KEY).remove(); attributes.attr(BLOCKED_REQUEST_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; @@ -215,6 +220,7 @@ private static boolean isPoisoned(final Deque contexts) { private final Context tracingContext; private final String acceptHeader; private boolean responseStarted; + private boolean beforeFinishCalled; private boolean responseAnalyzed; private Object deferredBlockResponse; @@ -234,6 +240,14 @@ public void markResponseStarted() { responseStarted = true; } + public boolean isBeforeFinishCalled() { + return beforeFinishCalled; + } + + public void markBeforeFinishCalled() { + beforeFinishCalled = true; + } + public boolean isResponseAnalyzed() { return responseAnalyzed; }