🍒 12295 - fix(netty-4.1): Treat Netty native client aborts as non-error responses

dd-java-agent/instrumentation/netty/netty-4.1/build.gradle

@@ -53,6 +53,17 @@ dependencies {
   latestDepTestImplementation group: 'io.netty', name: 'netty-codec-http', version: '4.+'
   latestDepTestImplementation group: 'io.netty', name: 'netty-codec-http2', version: '4.+'
   latestDepTestImplementation group: 'io.netty', name: 'netty-codec-socks', version: '4.+'
+  // Next native dependencies are required by NettyNativeClientAbortSpanTest
+  latestDepTestImplementation group: 'io.netty', name: 'netty-transport-classes-epoll', version: '4.+'
+  latestDepTestImplementation group: 'io.netty', name: 'netty-transport-classes-kqueue', version: '4.+'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-epoll', version: '4.+', classifier: 'linux-aarch_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-epoll', version: '4.+', classifier: 'linux-x86_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-kqueue', version: '4.+', classifier: 'osx-aarch_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-kqueue', version: '4.+', classifier: 'osx-x86_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-unix-common', version: '4.+', classifier: 'linux-aarch_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-unix-common', version: '4.+', classifier: 'linux-x86_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-unix-common', version: '4.+', classifier: 'osx-aarch_64'
+  latestDepTestRuntimeOnly group: 'io.netty', name: 'netty-transport-native-unix-common', version: '4.+', classifier: 'osx-x86_64'
 
   latestDepTestImplementation group: 'org.asynchttpclient', name: 'async-http-client', version: '2.+'
 }

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

@@ -126,7 +126,6 @@ private static void finishSpan(
       final Future<?> future) {
     if (!future.isSuccess()) {
       DECORATE.onError(span, future.cause());
-      span.setHttpStatusCode(500);
     }
     finishSpan(serverContext, storedContext, span);
   }

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

@@ -1,9 +1,11 @@
 package datadog.trace.instrumentation.netty41.server;
 
 import datadog.appsec.api.blocking.BlockingContentType;
+import datadog.trace.api.DDTags;
 import datadog.trace.api.gateway.BlockResponseFunction;
 import datadog.trace.api.internal.TraceSegment;
 import datadog.trace.bootstrap.instrumentation.api.AgentPropagation;
+import datadog.trace.bootstrap.instrumentation.api.AgentSpan;
 import datadog.trace.bootstrap.instrumentation.api.ContextVisitors;
 import datadog.trace.bootstrap.instrumentation.api.URIDataAdapter;
 import datadog.trace.bootstrap.instrumentation.api.URIDataAdapterBase;
@@ -34,6 +36,14 @@ public class NettyHttpServerDecorator
   public static final NettyHttpServerDecorator DECORATE = new NettyHttpServerDecorator();
   private static final CharSequence NETTY_REQUEST =
       UTF8BytesString.create(DECORATE.operationName());
+  private static final String NETTY_NATIVE_IO_EXCEPTION_CLASS_NAME =
+      "io.netty.channel.unix.Errors$NativeIoException";
+  private static final String NETTY_NATIVE_WRITEV_ADDRESSES_FAILURE_PREFIX =
+      "writevAddresses(..) failed";
+  private static final String NETTY_NATIVE_WRITEV_SYSCALL_FAILURE_PREFIX =
+      "syscall:writev(..) failed";
+  private static final String BROKEN_PIPE_MESSAGE_SUFFIX = ": Broken pipe";
+  private static final String CONNECTION_RESET_MESSAGE_SUFFIX = ": Connection reset by peer";
 
   @Override
   protected String[] instrumentationNames() {
@@ -108,6 +118,37 @@ protected boolean isAppSecOnResponseSeparate() {
     return true;
   }
 
+  @Override
+  protected void doOnError(final AgentSpan span, final Throwable throwable, byte errorPriority) {
+    if (isNettyNativeClientAbort(throwable)) {
+      span.setTag(DDTags.ERROR_MSG, safeMessage(throwable));
+      span.setTag(DDTags.ERROR_TYPE, throwable.getClass().getName());
+      return;
+    }
+    super.doOnError(span, throwable, errorPriority);
+  }
+
+  private static boolean isNettyNativeClientAbort(final Throwable throwable) {
+    if (throwable == null
+        || !NETTY_NATIVE_IO_EXCEPTION_CLASS_NAME.equals(throwable.getClass().getName())) {
+      return false;
+    }
+    final String message = safeMessage(throwable);
+    return message != null
+        && (message.startsWith(NETTY_NATIVE_WRITEV_ADDRESSES_FAILURE_PREFIX)
+            || message.startsWith(NETTY_NATIVE_WRITEV_SYSCALL_FAILURE_PREFIX))
+        && (message.endsWith(BROKEN_PIPE_MESSAGE_SUFFIX)
+            || message.endsWith(CONNECTION_RESET_MESSAGE_SUFFIX));
+  }
+
+  private static String safeMessage(final Throwable throwable) {
+    try {
+      return throwable.getMessage();
+    } catch (Throwable ignored) {
+      return null;
+    }
+  }
+
   @Override
   protected BlockResponseFunction createBlockResponseFunction(
       HttpRequest httpRequest, Channel channel) {

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

@@ -376,7 +376,7 @@ public Flow<Void> apply(RequestContext context, IGSpanInfo span) {
       assertThrows(IOException.class, channel::checkException);
 
       assertTrue(requestSpanActive.get(), "request span was not active during requestEnded");
-      assertEquals(500, span.getTag(Tags.HTTP_STATUS));
+      assertEquals(200, span.getTag(Tags.HTTP_STATUS));
       assertTraces(trace(span().root().operationName("netty.request").error()));
     } finally {
       holdingHandler.failWrite(writeFailure);

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

@@ -186,6 +186,8 @@ void requestBlockOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse() throws
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   @Test
@@ -221,6 +223,8 @@ void additionalPipelinedRequestsBehindDeferredBlockAreIgnored() throws Exception
           handler.inboundException,
           "additional pipelined requests should be swallowed by the existing blocking handler");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   @Test
@@ -248,6 +252,8 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierChunkedResponseCompletion
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   @Test
@@ -282,6 +288,8 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeaderOnlyResponse() throw
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   @Test
@@ -310,6 +318,9 @@ void requestBlockOnLaterPipelinedRequestFollowsEarlierHeadResponse() throws Exce
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(
+        SORT_BY_START_TIME, serverTrace("HEAD", FIRST_PATH), serverTrace("GET", SECOND_PATH));
   }
 
   @Test
@@ -329,6 +340,8 @@ void lastContentAfterInterimResponseDoesNotCompleteServerSpan() throws Exception
           "first response should be the interim response");
       assertEquals("response " + FIRST_PATH, readHttpResponseBody(socket.getInputStream()));
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH));
   }
 
   @Test
@@ -359,6 +372,8 @@ void requestBlockOnLaterPipelinedRequestWaitsForEarlierEarlyHintsResponseComplet
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   @Test
@@ -387,6 +402,8 @@ void blockResponseFunctionOnLaterPipelinedRequestDoesNotOvertakeEarlierResponse(
           readHeaders(socket.getInputStream()).startsWith("HTTP/1.1 403 "),
           "second response should be the deferred blocking response");
     }
+
+    assertTraces(SORT_BY_START_TIME, serverTrace(FIRST_PATH), serverTrace(SECOND_PATH));
   }
 
   private static String pipelinedRequests() {
@@ -453,11 +470,15 @@ public Flow<Object> get() {
   }
 
   private static TraceMatcher serverTrace(String path) {
+    return serverTrace("GET", path);
+  }
+
+  private static TraceMatcher serverTrace(String method, String path) {
     return trace(
         span()
             .root()
             .operationName(Pattern.compile("netty\\.request"))
-            .resourceName(Pattern.compile("GET " + Pattern.quote(path)))
+            .resourceName(Pattern.compile(Pattern.quote(method + " " + path)))
             .type("web"));
   }