[Zen2] Introduce LeaderChecker (#33024)

It is important that follower nodes periodically check that their leader is
still healthy and that they remain part of its cluster. If these checks fail
repeatedly then followers should attempt to find and join a new leader,
possibly electing one in the process. The LeaderChecker, introduced in this
commit, performs these periodic checks and deals with retries.

server/src/main/java/org/elasticsearch/cluster/coordination/LeaderChecker.java

@@ -0,0 +1,282 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License 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 org.elasticsearch.cluster.coordination;
+
+import org.apache.logging.log4j.message.ParameterizedMessage;
+import org.elasticsearch.cluster.node.DiscoveryNode;
+import org.elasticsearch.cluster.node.DiscoveryNodes;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.lease.Releasable;
+import org.elasticsearch.common.settings.Setting;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.threadpool.ThreadPool.Names;
+import org.elasticsearch.transport.ConnectTransportException;
+import org.elasticsearch.transport.TransportChannel;
+import org.elasticsearch.transport.TransportException;
+import org.elasticsearch.transport.TransportRequest;
+import org.elasticsearch.transport.TransportRequestOptions;
+import org.elasticsearch.transport.TransportRequestOptions.Type;
+import org.elasticsearch.transport.TransportResponse;
+import org.elasticsearch.transport.TransportResponse.Empty;
+import org.elasticsearch.transport.TransportResponseHandler;
+import org.elasticsearch.transport.TransportService;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * The LeaderChecker is responsible for allowing followers to check that the currently elected leader is still connected and healthy. We are
+ * fairly lenient, possibly allowing multiple checks to fail before considering the leader to be faulty, to allow for the leader to
+ * temporarily stand down on occasion, e.g. if it needs to move to a higher term. On deciding that the leader has failed a follower will
+ * become a candidate and attempt to become a leader itself.
+ */
+public class LeaderChecker extends AbstractComponent {
+
+    public static final String LEADER_CHECK_ACTION_NAME = "internal:coordination/fault_detection/leader_check";
+
+    // the time between checks sent to the leader
+    public static final Setting<TimeValue> LEADER_CHECK_INTERVAL_SETTING =
+        Setting.timeSetting("cluster.fault_detection.leader_check.interval",
+            TimeValue.timeValueMillis(10000), TimeValue.timeValueMillis(100), Setting.Property.NodeScope);
+
+    // the timeout for each check sent to the leader
+    public static final Setting<TimeValue> LEADER_CHECK_TIMEOUT_SETTING =
+        Setting.timeSetting("cluster.fault_detection.leader_check.timeout",
+            TimeValue.timeValueMillis(10000), TimeValue.timeValueMillis(1), Setting.Property.NodeScope);
+
+    // the number of failed checks that must happen before the leader is considered to have failed.
+    public static final Setting<Integer> LEADER_CHECK_RETRY_COUNT_SETTING =
+        Setting.intSetting("cluster.fault_detection.leader_check.retry_count", 3, 1, Setting.Property.NodeScope);
+
+    private final TimeValue leaderCheckInterval;
+    private final TimeValue leaderCheckTimeout;
+    private final int leaderCheckRetryCount;
+    private final TransportService transportService;
+    private final Runnable onLeaderFailure;
+
+    private volatile DiscoveryNodes lastPublishedDiscoveryNodes;
+
+    public LeaderChecker(final Settings settings, final TransportService transportService, final Runnable onLeaderFailure) {
+        super(settings);
+        leaderCheckInterval = LEADER_CHECK_INTERVAL_SETTING.get(settings);
+        leaderCheckTimeout = LEADER_CHECK_TIMEOUT_SETTING.get(settings);
+        leaderCheckRetryCount = LEADER_CHECK_RETRY_COUNT_SETTING.get(settings);
+        this.transportService = transportService;
+        this.onLeaderFailure = onLeaderFailure;
+
+        transportService.registerRequestHandler(LEADER_CHECK_ACTION_NAME, Names.SAME, LeaderCheckRequest::new, this::handleLeaderCheck);
+    }
+
+    /**
+     * Start a leader checker for the given leader. Should only be called after successfully joining this leader.
+     *
+     * @param leader the node to be checked as leader
+     * @return a `Releasable` that can be used to stop this checker.
+     */
+    public Releasable startLeaderChecker(final DiscoveryNode leader) {
+        assert transportService.getLocalNode().equals(leader) == false;
+        CheckScheduler checkScheduler = new CheckScheduler(leader);
+        checkScheduler.handleWakeUp();
+        return checkScheduler;
+    }
+
+    /**
+     * Update the "known" discovery nodes. Should be called on the leader before a new cluster state is published to reflect the new
+     * publication targets, and also called if a leader becomes a non-leader.
+     * TODO if heartbeats can make nodes become followers then this needs to be called before a heartbeat is sent to a new node too.
+     *
+     * isLocalNodeElectedMaster() should reflect whether this node is a leader, and nodeExists()
+     * should indicate whether nodes are known publication targets or not.
+     */
+    public void setLastPublishedDiscoveryNodes(DiscoveryNodes discoveryNodes) {
+        logger.trace("updating last-published nodes: {}", discoveryNodes);
+        lastPublishedDiscoveryNodes = discoveryNodes;
+    }
+
+    private void handleLeaderCheck(LeaderCheckRequest request, TransportChannel transportChannel, Task task) throws IOException {
+        final DiscoveryNodes lastPublishedDiscoveryNodes = this.lastPublishedDiscoveryNodes;
+        assert lastPublishedDiscoveryNodes != null;
+
+        if (lastPublishedDiscoveryNodes.isLocalNodeElectedMaster() == false) {
+            logger.debug("non-master handling {}", request);
+            transportChannel.sendResponse(new CoordinationStateRejectedException("non-leader rejecting leader check"));
+        } else if (lastPublishedDiscoveryNodes.nodeExists(request.getSender()) == false) {
+            logger.debug("leader check from unknown node: {}", request);
+            transportChannel.sendResponse(new CoordinationStateRejectedException("leader check from unknown node"));
+        } else {
+            logger.trace("handling {}", request);
+            transportChannel.sendResponse(Empty.INSTANCE);
+        }
+    }
+
+    private class CheckScheduler implements Releasable {
+
+        private final AtomicBoolean isClosed = new AtomicBoolean();
+        private final AtomicLong failureCountSinceLastSuccess = new AtomicLong();
+        private final DiscoveryNode leader;
+
+        CheckScheduler(final DiscoveryNode leader) {
+            this.leader = leader;
+        }
+
+        @Override
+        public void close() {
+            if (isClosed.compareAndSet(false, true) == false) {
+                logger.debug("already closed");
+            } else {
+                logger.debug("closed");
+            }
+        }
+
+        void handleWakeUp() {
+            if (isClosed.get()) {
+                logger.debug("closed check scheduler woken up, doing nothing");
+                return;
+            }
+
+            logger.trace("checking {} with [{}] = {}", leader, LEADER_CHECK_TIMEOUT_SETTING.getKey(), leaderCheckTimeout);
+
+            // TODO lag detection:
+            // In the PoC, the leader sent its current version to the follower in the response to a LeaderCheck, so the follower
+            // could detect if it was lagging. We'd prefer this to be implemented on the leader, so the response is just
+            // TransportResponse.Empty here.
+            transportService.sendRequest(leader, LEADER_CHECK_ACTION_NAME, new LeaderCheckRequest(transportService.getLocalNode()),
+                TransportRequestOptions.builder().withTimeout(leaderCheckTimeout).withType(Type.PING).build(),
+
+                new TransportResponseHandler<TransportResponse.Empty>() {
+                    @Override
+                    public void handleResponse(Empty response) {
+                        if (isClosed.get()) {
+                            logger.debug("closed check scheduler received a response, doing nothing");
+                            return;
+                        }
+
+                        failureCountSinceLastSuccess.set(0);
+                        scheduleNextWakeUp(); // logs trace message indicating success
+                    }
+
+                    @Override
+                    public void handleException(TransportException exp) {
+                        if (isClosed.get()) {
+                            logger.debug("closed check scheduler received a response, doing nothing");
+                            return;
+                        }
+
+                        if (exp instanceof ConnectTransportException || exp.getCause() instanceof ConnectTransportException) {
+                            logger.debug(new ParameterizedMessage("leader [{}] disconnected, failing immediately", leader), exp);
+                            leaderFailed();
+                            return;
+                        }
+
+                        long failureCount = failureCountSinceLastSuccess.incrementAndGet();
+                        if (failureCount >= leaderCheckRetryCount) {
+                            logger.debug(new ParameterizedMessage("{} consecutive failures (limit [{}] is {}) so leader [{}] has failed",
+                                failureCount, LEADER_CHECK_RETRY_COUNT_SETTING.getKey(), leaderCheckRetryCount, leader), exp);
+                            leaderFailed();
+                            return;
+                        }
+
+                        logger.debug(new ParameterizedMessage("{} consecutive failures (limit [{}] is {}) with leader [{}]",
+                            failureCount, LEADER_CHECK_RETRY_COUNT_SETTING.getKey(), leaderCheckRetryCount, leader), exp);
+                        scheduleNextWakeUp();
+                    }
+
+                    @Override
+                    public String executor() {
+                        return Names.SAME;
+                    }
+                });
+        }
+
+        private void leaderFailed() {
+            if (isClosed.compareAndSet(false, true)) {
+                transportService.getThreadPool().generic().execute(onLeaderFailure);
+            } else {
+                logger.debug("already closed, not failing leader");
+            }
+        }
+
+        private void scheduleNextWakeUp() {
+            logger.trace("scheduling next check of {} for [{}] = {}", leader, LEADER_CHECK_INTERVAL_SETTING.getKey(), leaderCheckInterval);
+            transportService.getThreadPool().schedule(leaderCheckInterval, Names.SAME, new Runnable() {
+                @Override
+                public void run() {
+                    handleWakeUp();
+                }
+
+                @Override
+                public String toString() {
+                    return "scheduled check of leader " + leader;
+                }
+            });
+        }
+    }
+
+    public static class LeaderCheckRequest extends TransportRequest {
+
+        private final DiscoveryNode sender;
+
+        public LeaderCheckRequest(final DiscoveryNode sender) {
+            this.sender = sender;
+        }
+
+        public LeaderCheckRequest(final StreamInput in) throws IOException {
+            super(in);
+            sender = new DiscoveryNode(in);
+        }
+
+        @Override
+        public void writeTo(final StreamOutput out) throws IOException {
+            super.writeTo(out);
+            sender.writeTo(out);
+        }
+
+        public DiscoveryNode getSender() {
+            return sender;
+        }
+
+        @Override
+        public boolean equals(final Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            final LeaderCheckRequest that = (LeaderCheckRequest) o;
+            return Objects.equals(sender, that.sender);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(sender);
+        }
+
+        @Override
+        public String toString() {
+            return "LeaderCheckRequest{" +
+                "sender=" + sender +
+                '}';
+        }
+    }
+}
+

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

@@ -951,7 +951,7 @@ private void checkForTimeout(long requestId) {
         assert responseHandlers.contains(requestId) == false;
         TimeoutInfoHolder timeoutInfoHolder = timeoutInfoHandlers.remove(requestId);
         if (timeoutInfoHolder != null) {
-            long time = System.currentTimeMillis();
+            long time = threadPool.relativeTimeInMillis();
             logger.warn("Received response for a request that has timed out, sent [{}ms] ago, timed out [{}ms] ago, " +
                     "action [{}], node [{}], id [{}]", time - timeoutInfoHolder.sentTime(), time - timeoutInfoHolder.timeoutTime(),
                 timeoutInfoHolder.action(), timeoutInfoHolder.node(), requestId);
@@ -1014,7 +1014,7 @@ protected void traceRequestSent(DiscoveryNode node, long requestId, String actio
     final class TimeoutHandler implements Runnable {
 
         private final long requestId;
-        private final long sentTime = System.currentTimeMillis();
+        private final long sentTime = threadPool.relativeTimeInMillis();
         private final String action;
         private final DiscoveryNode node;
         volatile ScheduledFuture future;
@@ -1028,7 +1028,7 @@ final class TimeoutHandler implements Runnable {
         @Override
         public void run() {
             if (responseHandlers.contains(requestId)) {
-                long timeoutTime = System.currentTimeMillis();
+                long timeoutTime = threadPool.relativeTimeInMillis();
                 timeoutInfoHandlers.put(requestId, new TimeoutInfoHolder(node, action, sentTime, timeoutTime));
                 // now that we have the information visible via timeoutInfoHandlers, we try to remove the request id
                 final Transport.ResponseContext holder = responseHandlers.remove(requestId);