transport: log network reconnects with same peer process

ClusterConnectionManager now caches the previous ephemeralId (created on
process-start) of peer nodes on disconnect in a connection history table. On
reconnect, when a peer has the same ephemeralId as it did previously, this is
logged to indicate a network failure. The connectionHistory is trimmed to the
current set of peers by NodeConnectionsService.

Author: schase-es

modules/transport-netty4/src/internalClusterTest/java/org/elasticsearch/transport/netty4/ESLoggingHandlerIT.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.MockLog;
 import org.elasticsearch.test.junit.annotations.TestLogging;
+import org.elasticsearch.transport.ClusterConnectionManager;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.TransportLogger;
 
@@ -27,7 +28,7 @@ public class ESLoggingHandlerIT extends ESNetty4IntegTestCase {
 
     public void setUp() throws Exception {
         super.setUp();
-        mockLog = MockLog.capture(ESLoggingHandler.class, TransportLogger.class, TcpTransport.class);
+        mockLog = MockLog.capture(ESLoggingHandler.class, TransportLogger.class, TcpTransport.class, ClusterConnectionManager.class);
     }
 
     public void tearDown() throws Exception {
@@ -125,4 +126,24 @@ public void testExceptionalDisconnectLogging() throws Exception {
 
         mockLog.assertAllExpectationsMatched();
     }
+
+    @TestLogging(
+        value = "org.elasticsearch.transport.ClusterConnectionManager:WARN",
+        reason = "to ensure we log cluster manager disconnect events on WARN level"
+    )
+    public void testExceptionalDisconnectLoggingInClusterConnectionManager() throws Exception {
+        mockLog.addExpectation(
+            new MockLog.PatternSeenEventExpectation(
+                "cluster connection manager exceptional disconnect log",
+                ClusterConnectionManager.class.getCanonicalName(),
+                Level.WARN,
+                "transport connection to \\[.*\\] closed (by remote )?with exception .*"
+            )
+        );
+
+        final String nodeName = internalCluster().startNode();
+        internalCluster().restartNode(nodeName);
+
+        mockLog.assertAllExpectationsMatched();
+    }
 }

server/src/main/java/org/elasticsearch/cluster/NodeConnectionsService.java

@@ -99,11 +99,13 @@ public void connectToNodes(DiscoveryNodes discoveryNodes, Runnable onCompletion)
         }
 
         final List<Runnable> runnables = new ArrayList<>(discoveryNodes.getSize());
+        final List<DiscoveryNode> nodes = new ArrayList<>(discoveryNodes.getSize());
         try (var refs = new RefCountingRunnable(onCompletion)) {
             synchronized (mutex) {
                 // Ugly hack: when https://github.com/elastic/elasticsearch/issues/94946 is fixed, just iterate over discoveryNodes here
                 for (final Iterator<DiscoveryNode> iterator = discoveryNodes.mastersFirstStream().iterator(); iterator.hasNext();) {
                     final DiscoveryNode discoveryNode = iterator.next();
+                    nodes.add(discoveryNode);
                     ConnectionTarget connectionTarget = targetsByNode.get(discoveryNode);
                     final boolean isNewNode = connectionTarget == null;
                     if (isNewNode) {
@@ -120,6 +122,7 @@ public void connectToNodes(DiscoveryNodes discoveryNodes, Runnable onCompletion)
                         runnables.add(connectionTarget.connect(null));
                     }
                 }
+                transportService.retainConnectionHistory(nodes);
             }
         }
         runnables.forEach(Runnable::run);

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

@@ -27,11 +27,15 @@
 
 import java.util.Collections;
 import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+
+import static org.elasticsearch.core.Strings.format;
 
 /**
  * This class manages node connections within a cluster. The connection is opened by the underlying transport.
@@ -47,6 +51,9 @@ public class ClusterConnectionManager implements ConnectionManager {
         .newConcurrentMap();
     private final AbstractRefCounted connectingRefCounter = AbstractRefCounted.of(this::pendingConnectionsComplete);
 
+    record NodeConnectionHistory(String ephemeralId, Exception disconnectCause) {}
+    private final ConcurrentMap<String, NodeConnectionHistory> nodeHistory = ConcurrentCollections.newConcurrentMap();
+
     private final Transport transport;
     private final ThreadContext threadContext;
     private final ConnectionProfile defaultProfile;
@@ -226,6 +233,29 @@ private void connectToNodeOrRetry(
                         } else {
                             logger.debug("connected to node [{}]", node);
                             managerRefs.mustIncRef();
+
+                            // log case where the remote node has same ephemeralId as its previous connection
+                            // (the network was disrupted, but not the remote process)
+                            final DiscoveryNode connNode = conn.getNode();
+                            NodeConnectionHistory hist = nodeHistory.remove(connNode.getId());
+                            if (hist != null && hist.ephemeralId.equals(connNode.getEphemeralId())) {
+                                if (hist.disconnectCause != null) {
+                                    logger.warn(
+                                        () -> format(
+                                            "transport connection reopened to node with same ephemeralId [%s], close exception:",
+                                            node.descriptionWithoutAttributes()
+                                        ),
+                                        hist.disconnectCause
+                                    );
+                                } else {
+                                    logger.warn(
+                                        """
+                                            transport connection reopened to node with same ephemeralId [{}]""",
+                                        node.descriptionWithoutAttributes()
+                                    );
+                                }
+                            }
+
                             try {
                                 connectionListener.onNodeConnected(node, conn);
                             } finally {
@@ -235,25 +265,65 @@ 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
-                                        );
-                                        // 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);
+                                conn.addCloseListener(new ActionListener<Void>() {
+                                    @Override
+                                    public void onResponse(Void ignored) {
+                                        final NodeConnectionHistory hist = new NodeConnectionHistory(node.getEphemeralId(), null);
+                                        nodeHistory.put(conn.getNode().getId(), hist);
                                     }
-                                }));
+
+                                    @Override
+                                    public void onFailure(Exception e) {
+                                        final NodeConnectionHistory hist = new NodeConnectionHistory(node.getEphemeralId(), e);
+                                        nodeHistory.put(conn.getNode().getId(), hist);
+                                    }
+                                });
+
+                                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()) {
+                                            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 +346,21 @@ private void connectToNodeOrRetry(
         );
     }
 
+    @Override
+    public void retainConnectionHistory(List<DiscoveryNode> nodes) {
+        List<String> nodeIds = nodes.stream().map(node -> node.getId()).collect(Collectors.toList());
+
+        final int startSize = nodeHistory.size();
+        // the keyset propagates changes to the underlying map
+        nodeHistory.keySet().retainAll(nodeIds);
+        logger.trace("Connection history garbage-collected from {} to {} entries", startSize, nodeHistory.size());
+    }
+
+    @Override
+    public int connectionHistorySize() {
+        return 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/ConnectionManager.java

@@ -15,6 +15,7 @@
 import org.elasticsearch.core.Releasable;
 
 import java.io.Closeable;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -50,6 +51,16 @@ void connectToNode(
 
     ConnectionProfile getConnectionProfile();
 
+    /**
+     * Keep the connection history for the nodes listed
+     */
+    void retainConnectionHistory(List<DiscoveryNode> nodes);
+
+    /**
+     * Exposed for tests
+     */
+    int connectionHistorySize();
+
     @FunctionalInterface
     interface ConnectionValidator {
         void validate(Transport.Connection connection, ConnectionProfile profile, ActionListener<Void> listener);

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

@@ -205,6 +205,15 @@ public void closeNoBlock() {
         delegate.closeNoBlock();
     }
 
+    @Override
+    public void retainConnectionHistory(List<DiscoveryNode> nodes) {
+        delegate.retainConnectionHistory(nodes);
+    }
+
+    public int connectionHistorySize() {
+        return delegate.connectionHistorySize();
+    }
+
     /**
      * This method returns a remote cluster alias for the given transport connection if it targets a node in the remote cluster.
      * This method will return an optional empty in case the connection targets the local node or the node in the local cluster.

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

@@ -483,6 +483,10 @@ public boolean nodeConnected(DiscoveryNode node) {
         return isLocalNode(node) || connectionManager.nodeConnected(node);
     }
 
+    public void retainConnectionHistory(List<DiscoveryNode> nodes) {
+        connectionManager.retainConnectionHistory(nodes);
+    }
+
     /**
      * Connect to the specified node with the given connection profile.
      * The ActionListener will be called on the calling thread or the generic thread pool.