Fix Netty client disconnects being reported as trace errors

dd-java-agent/instrumentation/netty/netty-4.1/src/main/java/datadog/trace/instrumentation/netty41/server/HttpServerResponseTracingHandler.java

@@ -19,6 +19,7 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.LastHttpContent;
 import io.netty.util.concurrent.Future;
+import java.nio.channels.ClosedChannelException;
 
 @ChannelHandler.Sharable
 public class HttpServerResponseTracingHandler extends ChannelOutboundHandlerAdapter {
@@ -124,12 +125,24 @@ private static void finishSpan(
       final Context storedContext,
       final AgentSpan span,
       final Future<?> future) {
-    if (!future.isSuccess()) {
+    // A client disconnect can fail the terminal write after the response has started. Leave any
+    // existing span error intact, and retain failures with an underlying cause or another subtype.
+    if (!future.isSuccess() && !isClosedChannelWithoutCause(future.cause())) {
       DECORATE.onError(span, future.cause());
     }
     finishSpan(serverContext, storedContext, span);
   }
 
+  private static boolean isClosedChannelWithoutCause(final Throwable failure) {
+    // StacklessClosedChannelException is package-private and absent from older Netty versions,
+    // so match it by class name.
+    return failure != null
+        && failure.getCause() == null
+        && (failure.getClass() == ClosedChannelException.class
+            || "io.netty.channel.StacklessClosedChannelException"
+                .equals(failure.getClass().getName()));
+  }
+
   private static void finishSpan(
       final ServerRequestContext serverContext, final Context storedContext, final AgentSpan span) {
     try (final ContextScope ignored = storedContext.attach()) {

dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyChunkedResponseSpanTest.java

@@ -2,6 +2,7 @@
 
 import static datadog.trace.agent.test.assertions.SpanMatcher.span;
 import static datadog.trace.agent.test.assertions.TraceMatcher.trace;
+import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan;
 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;
@@ -24,6 +25,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import datadog.trace.agent.test.assertions.SpanMatcher;
+import datadog.trace.api.DDTags;
 import datadog.trace.api.function.TriConsumer;
 import datadog.trace.api.gateway.Events;
 import datadog.trace.api.gateway.Flow;
@@ -55,6 +57,8 @@
 import io.netty.util.ReferenceCountUtil;
 import java.io.IOException;
 import java.net.Socket;
+import java.nio.channels.ClosedByInterruptException;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.FileChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -68,6 +72,8 @@
 import java.util.regex.Pattern;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.TestInstance;
+import org.tabletest.junit.TableTest;
+import org.tabletest.junit.TypeConverter;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 public class NettyChunkedResponseSpanTest extends NettyHttpServerTestSupport {
@@ -385,6 +391,63 @@ public Flow<Void> apply(RequestContext context, IGSpanInfo span) {
     }
   }
 
+  @TableTest({
+    "Scenario                               | Write failure     | Status | Application error | Expected error",
+    "plain closed channel                   | closed            | 200    | false             | false         ",
+    "existing application error             | closed            | 200    | true              | true          ",
+    "HTTP error response                    | closed            | 500    | false             | true          ",
+    "closed channel with underlying failure | closed with cause | 200    | false             | true          ",
+    "interrupted operation                  | interrupted       | 200    | false             | true          ",
+    "wrapped closed channel                 | wrapped closed    | 200    | false             | true          "
+  })
+  void preservesImportantErrorsOnTerminalWriteFailure(
+      Throwable writeFailure, int status, boolean applicationError, boolean expectedError) {
+    HoldingOutboundHandler holdingHandler = new HoldingOutboundHandler();
+    EmbeddedChannel channel =
+        new EmbeddedChannel(holdingHandler, HttpServerResponseTracingHandler.INSTANCE);
+    AgentSpan span = startSpan("netty", "netty.request");
+    ServerRequestContext.add(channel, span, null);
+    if (applicationError) {
+      span.addThrowable(new IllegalStateException("application failure"));
+    }
+
+    try {
+      DefaultFullHttpResponse response =
+          new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(status));
+      response.headers().set(CONTENT_LENGTH, 0);
+      channel.writeOutbound(response);
+
+      holdingHandler.failWrite(writeFailure);
+      assertThrows(writeFailure.getClass(), channel::checkException);
+
+      assertEquals(status, span.getTag(Tags.HTTP_STATUS));
+      assertTraces(trace(span().root().operationName("netty.request").error(expectedError)));
+      if (applicationError) {
+        assertEquals(IllegalStateException.class.getName(), span.getTag(DDTags.ERROR_TYPE));
+        assertEquals("application failure", span.getTag(DDTags.ERROR_MSG));
+      }
+    } finally {
+      holdingHandler.failWrite(writeFailure);
+      channel.finishAndReleaseAll();
+    }
+  }
+
+  @TypeConverter
+  public static Throwable toWriteFailure(String value) {
+    switch (value) {
+      case "closed":
+        return new ClosedChannelException();
+      case "closed with cause":
+        return new ClosedChannelException().initCause(new IOException("transport failure"));
+      case "interrupted":
+        return new ClosedByInterruptException();
+      case "wrapped closed":
+        return new IOException("response failure", new ClosedChannelException());
+      default:
+        throw new IllegalArgumentException("Unknown write failure: " + value);
+    }
+  }
+
   @Test
   void knownLengthFileRegionResponseWaitsForLastContent() throws Exception {
     boolean reportedBeforeLastContent;

dd-java-agent/instrumentation/netty/netty-4.1/src/test/java/datadog/trace/instrumentation/netty41/server/NettyClosedChannelClientAbortSpanTest.java

@@ -0,0 +1,134 @@
+package datadog.trace.instrumentation.netty41.server;
+
+import static io.netty.handler.codec.http.HttpHeaderNames.TRANSFER_ENCODING;
+import static io.netty.handler.codec.http.HttpHeaderValues.CHUNKED;
+import static io.netty.handler.codec.http.HttpResponseStatus.OK;
+import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;
+import static java.nio.charset.StandardCharsets.US_ASCII;
+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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.trace.bootstrap.instrumentation.api.Tags;
+import datadog.trace.core.DDSpan;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+import io.netty.channel.SimpleChannelInboundHandler;
+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.HttpServerCodec;
+import io.netty.handler.codec.http.LastHttpContent;
+import java.net.Socket;
+import java.nio.channels.ClosedChannelException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public class NettyClosedChannelClientAbortSpanTest extends NettyHttpServerTestSupport {
+
+  private final BlockingQueue<DelayedWrite> delayedWrites = new LinkedBlockingQueue<>();
+
+  @Override
+  protected void configurePipeline(Channel channel) {
+    DelayedWriteHandler delayedWriteHandler = new DelayedWriteHandler();
+    channel.pipeline().addLast(delayedWriteHandler);
+    channel.pipeline().addLast(new HttpServerCodec());
+    channel
+        .pipeline()
+        .addLast(
+            new SimpleChannelInboundHandler<HttpRequest>() {
+              @Override
+              protected void channelRead0(ChannelHandlerContext ctx, HttpRequest request) {
+                DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, OK);
+                response.headers().set(TRANSFER_ENCODING, CHUNKED);
+                ctx.write(response);
+                ctx.writeAndFlush(new DefaultHttpContent(ctx.alloc().buffer(1).writeByte('a')));
+
+                // Hold the encoded terminal write after the tracing handler attaches its listener.
+                delayedWriteHandler.delayWrites = true;
+                ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
+              }
+            });
+  }
+
+  @Test
+  void closedChannelFromCancelledResponseDoesNotMarkServerSpanError() throws Exception {
+    DelayedWrite delayedWrite = null;
+    try {
+      try (Socket socket = connect()) {
+        socket
+            .getOutputStream()
+            .write("GET /closed-channel HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes(US_ASCII));
+        socket.getOutputStream().flush();
+        assertTrue(readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 200 "));
+
+        delayedWrite = delayedWrites.poll(5, SECONDS);
+        assertNotNull(delayedWrite, "server did not attempt the terminal response write");
+        assertFalse(delayedWrite.promise.isDone());
+        assertTrue(writer.isEmpty(), "span finished before the terminal response write");
+      }
+
+      assertTrue(
+          delayedWrite.context.channel().closeFuture().await(5, SECONDS),
+          "server did not observe the client disconnect");
+      delayedWrite.resume();
+      assertTrue(delayedWrite.promise.await(5, SECONDS), "terminal response write did not finish");
+      // Recent Netty versions produce StacklessClosedChannelException; older ones use its base
+      // type.
+      Throwable failure = delayedWrite.promise.cause();
+      assertInstanceOf(ClosedChannelException.class, failure);
+
+      writer.waitForTraces(1);
+      assertEquals(1, writer.firstTrace().size());
+      DDSpan span = writer.firstTrace().get(0);
+      assertEquals("netty.request", span.getOperationName().toString());
+      assertEquals(200, span.getTag(Tags.HTTP_STATUS));
+      assertEquals(
+          0,
+          span.getError(),
+          () -> "client abort must not count as a trace error: " + failure.getClass().getName());
+    } finally {
+      if (delayedWrite != null && !delayedWrite.promise.isDone()) {
+        delayedWrite.resume();
+      }
+    }
+  }
+
+  private final class DelayedWriteHandler extends ChannelOutboundHandlerAdapter {
+    private boolean delayWrites;
+
+    @Override
+    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
+      if (delayWrites) {
+        delayedWrites.add(new DelayedWrite(ctx, msg, promise));
+      } else {
+        ctx.write(msg, promise);
+      }
+    }
+  }
+
+  private static final class DelayedWrite {
+    private final ChannelHandlerContext context;
+    private final Object message;
+    private final ChannelPromise promise;
+
+    private DelayedWrite(ChannelHandlerContext context, Object message, ChannelPromise promise) {
+      this.context = context;
+      this.message = message;
+      this.promise = promise;
+    }
+
+    private void resume() throws Exception {
+      // Let Netty fail the actual write to the closed channel and release the encoded buffer.
+      context.executor().submit(() -> context.writeAndFlush(message, promise)).sync();
+    }
+  }
+}