15290: Log Unexpected Disconnects

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4TcpChannel.java

@@ -34,15 +34,22 @@ public class Netty4TcpChannel implements TcpChannel {
     private final ListenableFuture<Void> closeContext = new ListenableFuture<>();
     private final ChannelStats stats = new ChannelStats();
     private final boolean rstOnClose;
+    private volatile Exception channelError = null;
 
     Netty4TcpChannel(Channel channel, boolean isServer, String profile, boolean rstOnClose, ChannelFuture connectFuture) {
         this.channel = channel;
         this.isServer = isServer;
         this.profile = profile;
         this.connectContext = new ListenableFuture<>();
         this.rstOnClose = rstOnClose;
-        addListener(this.channel.closeFuture(), closeContext);
         addListener(connectFuture, connectContext);
+        addListener(this.channel.closeFuture(), ActionListener.running(() -> {
+            if (channelError != null) {
+                closeContext.onFailure(channelError);
+            } else {
+                closeContext.onResponse(null);
+            }
+        }));
     }
 
     @Override
@@ -95,6 +102,11 @@ public void addConnectListener(ActionListener<Void> listener) {
         connectContext.addListener(listener);
     }
 
+    @Override
+    public void onException(Exception e) {
+        channelError = e;
+    }
+
     @Override
     public ChannelStats getChannelStats() {
         return stats;

modules/transport-netty4/src/main/java/org/elasticsearch/transport/netty4/Netty4Transport.java

@@ -28,6 +28,7 @@
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.TransportVersion;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.network.NetworkService;
@@ -281,8 +282,8 @@ protected Netty4TcpChannel initiateChannel(DiscoveryNode node, ConnectionProfile
             rstOnClose,
             connectFuture
         );
+        addClosedExceptionLogger(nettyChannel);
         channel.attr(CHANNEL_KEY).set(nettyChannel);
-
         return nettyChannel;
     }
 
@@ -312,14 +313,19 @@ protected class ClientChannelInitializer extends ChannelInitializer<Channel> {
 
         @Override
         protected void initChannel(Channel ch) throws Exception {
-            addClosedExceptionLogger(ch);
             assert ch instanceof Netty4NioSocketChannel;
             NetUtils.tryEnsureReasonableKeepAliveConfig(((Netty4NioSocketChannel) ch).javaChannel());
             setupPipeline(ch, false);
         }
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            Netty4TcpChannel channel = ctx.channel().attr(CHANNEL_KEY).get();
+            if (cause instanceof Error) {
+                channel.onException(new Exception(cause));
+            } else {
+                channel.onException((Exception) cause);
+            }
             ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
@@ -337,17 +343,23 @@ protected ServerChannelInitializer(String name) {
 
         @Override
         protected void initChannel(Channel ch) throws Exception {
-            addClosedExceptionLogger(ch);
             assert ch instanceof Netty4NioSocketChannel;
             NetUtils.tryEnsureReasonableKeepAliveConfig(((Netty4NioSocketChannel) ch).javaChannel());
             Netty4TcpChannel nettyTcpChannel = new Netty4TcpChannel(ch, true, name, rstOnClose, ch.newSucceededFuture());
+            addClosedExceptionLogger(nettyTcpChannel);
             ch.attr(CHANNEL_KEY).set(nettyTcpChannel);
             setupPipeline(ch, isRemoteClusterServerChannel);
             serverAcceptedChannel(nettyTcpChannel);
         }
 
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            Netty4TcpChannel channel = ctx.channel().attr(CHANNEL_KEY).get();
+            if (cause instanceof Error) {
+                channel.onException(new Exception(cause));
+            } else {
+                channel.onException((Exception) cause);
+            }
             ExceptionsHelper.maybeDieOnAnotherThread(cause);
             super.exceptionCaught(ctx, cause);
         }
@@ -383,12 +395,12 @@ protected InboundPipeline getInboundPipeline(Channel ch, boolean isRemoteCluster
         );
     }
 
-    private static void addClosedExceptionLogger(Channel channel) {
-        Netty4Utils.addListener(channel.closeFuture(), channelFuture -> {
-            if (channelFuture.isSuccess() == false && logger.isDebugEnabled()) {
-                logger.debug(format("exception while closing channel: %s", channelFuture.channel()), channelFuture.cause());
+    private static void addClosedExceptionLogger(Netty4TcpChannel channel) {
+        channel.addCloseListener(ActionListener.wrap((ignored) -> {}, (e) -> {
+            if (logger.isDebugEnabled()) {
+                logger.debug(format("exception while closing channel: %s", channel), e);
             }
-        });
+        }));
     }
 
     @ChannelHandler.Sharable

server/src/main/java/org/elasticsearch/common/network/CloseableChannel.java

@@ -38,6 +38,9 @@ public interface CloseableChannel extends Closeable {
      * channel. If the channel is already closed when the listener is added the listener will immediately be
      * executed by the thread that is attempting to add the listener.
      *
+     * When the close completes but an exception prompted the closure, the exception will be passed to the
+     * listener's onFailure method.
+     *
      * @param listener to be executed
      */
     void addCloseListener(ActionListener<Void> listener);

server/src/main/java/org/elasticsearch/tasks/TaskManager.java

@@ -736,11 +736,11 @@ private ChannelPendingTaskTracker startTrackingChannel(TcpChannel channel, Consu
             return curr;
         });
         if (tracker.registered.compareAndSet(false, true)) {
-            channel.addCloseListener(ActionListener.wrap(r -> {
+            channel.addCloseListener(ActionListener.running(() -> {
                 final ChannelPendingTaskTracker removedTracker = channelPendingTaskTrackers.remove(channel);
                 assert removedTracker == tracker;
                 onChannelClosed(tracker);
-            }, e -> { assert false : new AssertionError("must not be here", e); }));
+            }));
         }
         return tracker;
     }

server/src/main/java/org/elasticsearch/transport/CloseableConnection.java

@@ -48,6 +48,12 @@ public void close() {
         }
     }
 
+    public void closeAndFail(Exception e) {
+        if (closed.compareAndSet(false, true)) {
+            closeContext.onFailure(e);
+        }
+    }
+
     @Override
     public void onRemoved() {
         if (removed.compareAndSet(false, true)) {

server/src/main/java/org/elasticsearch/transport/ClusterConnectionManager.java

@@ -47,6 +47,12 @@ public class ClusterConnectionManager implements ConnectionManager {
         .newConcurrentMap();
     private final AbstractRefCounted connectingRefCounter = AbstractRefCounted.of(this::pendingConnectionsComplete);
 
+    record NodeConnectionHistory(String ephemeralId, long disconnectTime) {}
+
+    // map of nodeId -> NodeConnectionHistory entries updated on connection close, with any error
+    private final ConcurrentMap<String, NodeConnectionHistory> nodeHistory = ConcurrentCollections.newConcurrentMap();
+    private long nodeHistoryLastGC = 0;
+
     private final Transport transport;
     private final ThreadContext threadContext;
     private final ConnectionProfile defaultProfile;
@@ -226,6 +232,19 @@ private void connectToNodeOrRetry(
                         } else {
                             logger.debug("connected to node [{}]", node);
                             managerRefs.mustIncRef();
+
+                            // log case where remote has same ephemeralId as previous connection (the network was disrupted, but not the
+                            // remote process), and update history with removal (we just connected successfully)
+                            final DiscoveryNode connNode = conn.getNode();
+                            NodeConnectionHistory hist = nodeHistory.remove(connNode.getId());
+                            if (hist != null && hist.ephemeralId.equals(connNode.getEphemeralId())) {
+                                logger.warn(
+                                    """
+                                        transport connection to [{}] reopened, with same ephemeralId found.""",
+                                    node.descriptionWithoutAttributes()
+                                );
+                            }
+
                             try {
                                 connectionListener.onNodeConnected(node, conn);
                             } finally {
@@ -235,25 +254,75 @@ private void connectToNodeOrRetry(
                                     managerRefs.decRef();
                                 }));
 
-                                conn.addCloseListener(ActionListener.running(() -> {
-                                    if (connectingRefCounter.hasReferences() == false) {
-                                        logger.trace("connection manager shut down, closing transport connection to [{}]", node);
-                                    } else if (conn.hasReferences()) {
-                                        logger.info(
-                                            """
-                                                transport connection to [{}] closed by remote; \
-                                                if unexpected, see [{}] for troubleshooting guidance""",
-                                            node.descriptionWithoutAttributes(),
-                                            ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
+                                conn.addCloseListener(new ActionListener<Void>() {
+                                    @Override
+                                    public void onResponse(Void ignored) {
+                                        final NodeConnectionHistory hist = new NodeConnectionHistory(
+                                            node.getEphemeralId(),
+                                            System.currentTimeMillis()
                                         );
-                                        // In production code we only close connections via ref-counting, so this message confirms that a
-                                        // 'node-left ... reason: disconnected' event was caused by external factors. Put differently, if a
-                                        // node leaves the cluster with "reason: disconnected" but without this message being logged then
-                                        // that's a bug.
-                                    } else {
-                                        logger.debug("closing unused transport connection to [{}]", node);
+                                        nodeHistory.put(conn.getNode().getId(), hist);
                                     }
-                                }));
+
+                                    @Override
+                                    public void onFailure(Exception e) {
+                                        final NodeConnectionHistory hist = new NodeConnectionHistory(
+                                            node.getEphemeralId(),
+                                            System.currentTimeMillis()
+                                        );
+                                        nodeHistory.put(conn.getNode().getId(), hist);
+                                    }
+                                });
+
+                                conn.addCloseListener(ActionListener.running(ClusterConnectionManager.this::collectHistoryGarbage));
+
+                                conn.addCloseListener(new ActionListener<Void>() {
+                                    @Override
+                                    public void onResponse(Void ignored) {
+                                        if (connectingRefCounter.hasReferences() == false) {
+                                            logger.trace("connection manager shut down, closing transport connection to [{}]", node);
+                                        } else if (conn.hasReferences()) {
+                                            // the connection is closing down, but hasn't been released by the client/library side
+                                            // this is an event coming up from the network side
+                                            logger.info(
+                                                """
+                                                    transport connection to [{}] closed by remote; \
+                                                    if unexpected, see [{}] for troubleshooting guidance""",
+                                                node.descriptionWithoutAttributes(),
+                                                ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
+                                            );
+                                            // In production code we only close connections via ref-counting, so this message confirms that
+                                            // a 'node-left ... reason: disconnected' event was caused by external factors. Put
+                                            // differently, if a node leaves the cluster with "reason: disconnected" but without this
+                                            // message being logged then that's a bug.
+                                        } else {
+                                            logger.debug("closing unused transport connection to [{}]", node);
+                                        }
+                                    }
+
+                                    @Override
+                                    public void onFailure(Exception e) {
+                                        if (conn.hasReferences()) {
+                                            logger.warn(
+                                                """
+                                                    transport connection to [{}] closed by remote with exception [{}]; \
+                                                    if unexpected, see [{}] for troubleshooting guidance""",
+                                                node.descriptionWithoutAttributes(),
+                                                e,
+                                                ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
+                                            );
+                                        } else {
+                                            logger.warn(
+                                                """
+                                                    transport connection to [{}] closed with exception [{}]; \
+                                                    if unexpected, see [{}] for troubleshooting guidance""",
+                                                node.descriptionWithoutAttributes(),
+                                                e,
+                                                ReferenceDocs.NETWORK_DISCONNECT_TROUBLESHOOTING
+                                            );
+                                        }
+                                    }
+                                });
                             }
                         }
                     } finally {
@@ -276,6 +345,26 @@ private void connectToNodeOrRetry(
         );
     }
 
+    /**
+     * Removes entries in the nodeHistory table that are too old
+     */
+    private void collectHistoryGarbage() {
+        final long now = System.currentTimeMillis();
+        final long hour = 60 * 60 * 1000;
+
+        if (now - hour > nodeHistoryLastGC) {
+            final int startSize = nodeHistory.size();
+            nodeHistoryLastGC = now;
+            final long expire = now - hour;
+            for (Map.Entry<String, NodeConnectionHistory> entry : nodeHistory.entrySet()) {
+                if (expire > entry.getValue().disconnectTime) {
+                    nodeHistory.remove(entry.getKey(), entry.getValue());
+                }
+            }
+            logger.trace("ClusterConnectionManager GCed connection history from {} to {} entries", startSize, nodeHistory.size());
+        }
+    }
+
     /**
      * Returns a connection for the given node if the node is connected.
      * Connections returned from this method must not be closed. The lifecycle of this connection is

server/src/main/java/org/elasticsearch/transport/InboundHandler.java

@@ -390,6 +390,7 @@ private void handleHandshakeRequest(TcpChannel channel, InboundMessage message)
                 () -> "error processing handshake version [" + header.getVersion() + "] received on [" + channel + "], closing channel",
                 e
             );
+            channel.onException(e);
             channel.close();
         }
     }

server/src/main/java/org/elasticsearch/transport/OutboundHandler.java

@@ -168,6 +168,7 @@ void sendResponse(
                     ),
                     ex
                 );
+                channel.onException(ex);
                 channel.close();
             } else {
                 sendErrorResponse(transportVersion, channel, requestId, action, responseStatsConsumer, ex);
@@ -204,6 +205,7 @@ void sendErrorResponse(
         } catch (Exception sendException) {
             sendException.addSuppressed(error);
             logger.error(() -> format("Failed to send error response on channel [%s], closing channel", channel), sendException);
+            channel.onException(sendException);
             channel.close();
         }
     }
@@ -431,6 +433,7 @@ private void maybeLogSlowMessage(boolean success) {
                 }
             });
         } catch (RuntimeException ex) {
+            channel.onException(ex);
             Releasables.closeExpectNoException(() -> listener.onFailure(ex), () -> CloseableChannel.closeChannel(channel));
             throw ex;
         }

server/src/main/java/org/elasticsearch/transport/TcpChannel.java

@@ -66,6 +66,13 @@ public interface TcpChannel extends CloseableChannel {
      */
     void addConnectListener(ActionListener<Void> listener);
 
+    /**
+     * Report an exception on this channel
+     *
+     * @param e the exception
+     */
+    void onException(Exception e);
+
     /**
      * Returns stats about this channel
      */