Log channel ID when logging error (#2892)

* Log channel ID when logging error

* Review comments

* Review comments

* Review comments

Author: dagnir

.idea/inspectionProfiles/AWS_Java_SDK_2_0.xml

@@ -57,8 +57,6 @@
 import org.reactivestreams.Publisher;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.http.Protocol;
 import software.amazon.awssdk.http.nio.netty.internal.http2.FlushOnReadHandler;
@@ -68,12 +66,13 @@
 import software.amazon.awssdk.http.nio.netty.internal.nrs.HttpStreamsClientHandler;
 import software.amazon.awssdk.http.nio.netty.internal.nrs.StreamedHttpRequest;
 import software.amazon.awssdk.http.nio.netty.internal.utils.ChannelUtils;
+import software.amazon.awssdk.http.nio.netty.internal.utils.NettyClientLogger;
 import software.amazon.awssdk.http.nio.netty.internal.utils.NettyUtils;
 import software.amazon.awssdk.metrics.MetricCollector;
 
 @SdkInternalApi
 public final class NettyRequestExecutor {
-    private static final Logger log = LoggerFactory.getLogger(NettyRequestExecutor.class);
+    private static final NettyClientLogger log = NettyClientLogger.getLogger(NettyRequestExecutor.class);
     private static final RequestAdapter REQUEST_ADAPTER_HTTP2 = new RequestAdapter(Protocol.HTTP2);
     private static final RequestAdapter REQUEST_ADAPTER_HTTP1_1 = new RequestAdapter(Protocol.HTTP1_1);
     private static final AtomicLong EXECUTION_COUNTER = new AtomicLong(0L);
@@ -129,7 +128,7 @@ private CompletableFuture<Void> createExecutionFuture(Promise<Channel> channelPr
                         }
                     });
                 } catch (Throwable exc) {
-                    log.warn("Unable to add a task to cancel the request to channel's EventLoop", exc);
+                    log.warn(ch, () -> "Unable to add a task to cancel the request to channel's EventLoop", exc);
                 }
             }
         });
@@ -151,13 +150,13 @@ private void verifyMetricsWereCollected(CompletableFuture<Void> metricsFuture) {
         }
 
         if (!metricsFuture.isDone()) {
-            log.debug("HTTP request metric collection did not finish in time, so results may be incomplete.");
+            log.debug(null, () -> "HTTP request metric collection did not finish in time, so results may be incomplete.");
             metricsFuture.cancel(false);
             return;
         }
 
         metricsFuture.exceptionally(t -> {
-            log.debug("HTTP request metric collection failed, so results may be incomplete.", t);
+            log.debug(null, () -> "HTTP request metric collection failed, so results may be incomplete.", t);
             return null;
         });
     }
@@ -172,7 +171,7 @@ private void makeRequestListener(Future<Channel> channelFuture) {
                 }
             });
         } else {
-            handleFailure(() -> "Failed to create connection to " + endpoint(), channelFuture.cause());
+            handleFailure(channel, () -> "Failed to create connection to " + endpoint(), channelFuture.cause());
         }
     }
 
@@ -203,7 +202,7 @@ private boolean tryConfigurePipeline() {
             default:
                 String errorMsg = "Unknown protocol: " + protocol;
                 closeAndRelease(channel);
-                handleFailure(() -> errorMsg, new RuntimeException(errorMsg));
+                handleFailure(channel, () -> errorMsg, new RuntimeException(errorMsg));
                 return false;
         }
 
@@ -220,7 +219,7 @@ private boolean tryConfigurePipeline() {
         if (!channel.isActive()) {
             String errorMessage = "Channel was closed before it could be written to.";
             closeAndRelease(channel);
-            handleFailure(() -> errorMessage, new IOException(errorMessage));
+            handleFailure(channel, () -> errorMessage, new IOException(errorMessage));
             return false;
         }
 
@@ -254,7 +253,7 @@ private void writeRequest(HttpRequest request) {
                    } else {
                        // TODO: Are there cases where we can keep the channel open?
                        closeAndRelease(channel);
-                       handleFailure(() -> "Failed to make request to " + endpoint(), wireCall.cause());
+                       handleFailure(channel, () -> "Failed to make request to " + endpoint(), wireCall.cause());
                    }
                });
 
@@ -297,8 +296,8 @@ private URI endpoint() {
         return context.executeRequest().request().getUri();
     }
 
-    private void handleFailure(Supplier<String> msg, Throwable cause) {
-        log.debug(msg.get(), cause);
+    private void handleFailure(Channel channel, Supplier<String> msgSupplier, Throwable cause) {
+        log.debug(channel, msgSupplier, cause);
         cause = decorateException(cause);
         context.handler().onError(cause);
         executeFuture.completeExceptionally(cause);
@@ -379,7 +378,7 @@ private String getMessageForTooManyAcquireOperationsError() {
      * @param channel The channel.
      */
     private void closeAndRelease(Channel channel) {
-        log.trace("closing and releasing channel {}", channel.id().asLongText());
+        log.trace(channel, () -> String.format("closing and releasing channel %s", channel.id().asLongText()));
         channel.attr(KEEP_ALIVE).set(false);
         channel.close();
         context.channelPool().release(channel);
@@ -472,7 +471,7 @@ public String toString() {
     /**
      * Decorator around {@link StreamedHttpRequest} to adapt a publisher of {@link ByteBuffer} (i.e. {@link
      * software.amazon.awssdk.http.async.SdkHttpContentPublisher}) to a publisher of {@link HttpContent}.
-     * <p />
+     * <p>
      * This publisher also prevents the adapted publisher from publishing more content to the subscriber than
      * the specified 'Content-Length' of the request.
      */
@@ -565,7 +564,7 @@ private static Optional<Long> contentLength(HttpRequest request) {
                 try {
                     return Optional.of(Long.parseLong(value));
                 } catch (NumberFormatException e) {
-                    log.warn("Unable  to parse 'Content-Length' header. Treating it as non existent.");
+                    log.warn(null, () -> "Unable  to parse 'Content-Length' header. Treating it as non existent.");
                 }
             }
             return Optional.empty();

http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/NettyRequestExecutor.java

@@ -0,0 +1,132 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.http.nio.netty.internal.utils;
+
+import io.netty.channel.Channel;
+import java.util.function.Supplier;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.annotations.SdkInternalApi;
+import software.amazon.awssdk.annotations.SdkTestInternalApi;
+
+/**
+ * Logger facade similar to {@link software.amazon.awssdk.utils.Logger}, that also includes channel information in the message
+ * when provided. When the logger has at least DEBUG level enabled, the logger uses {@link Channel#toString()} to provide the
+ * complete information about the channel. If only less verbose levels are available, then only the channel's ID is logged.
+ * <p>
+ * Having the channel information associated with the log message whenever available makes correlating messages that are all
+ * logged within the context of that channel possible; this is impossible to do otherwise because there is a 1:M mapping from
+ * event loops to channels.
+ * <p>
+ * <b>NOTE:</b> The absence of overrides that don't take a {@code Channel} parameter is deliberate. This is done to lessen the
+ * chances that a {code Channel} is omitted from the log by accident.
+ */
+@SdkInternalApi
+public final class NettyClientLogger {
+    private final Logger delegateLogger;
+
+    @SdkTestInternalApi
+    NettyClientLogger(Logger delegateLogger) {
+        this.delegateLogger = delegateLogger;
+    }
+
+    public static NettyClientLogger getLogger(Class<?> clzz) {
+        Logger delegate = LoggerFactory.getLogger(clzz);
+        return new NettyClientLogger(delegate);
+    }
+
+    /**
+     * Log a DEBUG level message including the channel information.
+     *
+     * @param channel The channel for this message is being logged
+     * @param msgSupplier Supplier for the log message
+     */
+    public void debug(Channel channel, Supplier<String> msgSupplier) {
+        debug(channel, msgSupplier, null);
+    }
+
+    /**
+     * Log a DEBUG level message with the given exception and including the channel information.
+     *
+     * @param channel The channel for this message is being logged
+     * @param msgSupplier Supplier for the log message
+     * @param t The throwable to log
+     */
+    public void debug(Channel channel, Supplier<String> msgSupplier, Throwable t) {
+        if (!delegateLogger.isDebugEnabled()) {
+            return;
+        }
+
+        String finalMessage = prependChannelInfo(msgSupplier, channel);
+        delegateLogger.debug(finalMessage, t);
+    }
+
+    /**
+     * Log a WARN level message and including the channel information.
+     *
+     * @param channel The channel for this message is being logged
+     * @param msgSupplier Supplier for the log message
+     */
+    public void warn(Channel channel, Supplier<String> msgSupplier) {
+        warn(channel, msgSupplier, null);
+    }
+
+    /**
+     * Log a WARN level message with the given exception and including the channel information.
+     *
+     * @param channel The channel for this message is being logged
+     * @param msgSupplier Supplier for the log message
+     * @param t The throwable to log
+     */
+    public void warn(Channel channel, Supplier<String> msgSupplier, Throwable t) {
+        if (!delegateLogger.isWarnEnabled()) {
+            return;
+        }
+
+        String finalMessage = prependChannelInfo(msgSupplier, channel);
+        delegateLogger.warn(finalMessage, t);
+    }
+
+    /**
+     * Log a TRACE level message including the channel information.
+     *
+     * @param channel The channel for this message is being logged
+     * @param msgSupplier Supplier for the log message
+     */
+    public void trace(Channel channel, Supplier<String> msgSupplier) {
+        if (!delegateLogger.isTraceEnabled()) {
+            return;
+        }
+
+        String finalMessage = prependChannelInfo(msgSupplier, channel);
+        delegateLogger.trace(finalMessage);
+    }
+
+    private String prependChannelInfo(Supplier<String> msgSupplier, Channel channel) {
+        if (channel == null) {
+            return msgSupplier.get();
+        }
+
+        String id;
+        if (!delegateLogger.isDebugEnabled()) {
+            id = channel.id().asShortText();
+        } else {
+            id = channel.toString();
+        }
+
+        return String.format("[Channel: %s] %s", id, msgSupplier.get());
+    }
+}