Adds resiliency to read-only filesystems #45286 #52680
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -22,6 +22,7 @@ | |
| import org.apache.logging.log4j.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.apache.logging.log4j.message.ParameterizedMessage; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.cluster.coordination.Coordinator.Mode; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.cluster.node.DiscoveryNodes; | ||
|
|
@@ -39,10 +40,10 @@ | |
| import org.elasticsearch.transport.TransportException; | ||
| import org.elasticsearch.transport.TransportRequest; | ||
| import org.elasticsearch.transport.TransportRequestOptions; | ||
| import org.elasticsearch.transport.TransportService; | ||
| import org.elasticsearch.transport.TransportRequestOptions.Type; | ||
| import org.elasticsearch.transport.TransportResponse.Empty; | ||
| import org.elasticsearch.transport.TransportResponseHandler; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.HashSet; | ||
|
|
@@ -52,6 +53,7 @@ | |
| import java.util.function.BiConsumer; | ||
| import java.util.function.Consumer; | ||
| import java.util.function.Predicate; | ||
| import java.util.function.Supplier; | ||
|
|
||
| import static org.elasticsearch.common.util.concurrent.ConcurrentCollections.newConcurrentMap; | ||
|
|
||
|
|
@@ -94,12 +96,13 @@ public class FollowersChecker { | |
| private final Set<DiscoveryNode> faultyNodes = new HashSet<>(); | ||
|
|
||
| private final TransportService transportService; | ||
| private final NodeFsHealthChecker nodeFsHealthChecker; | ||
|
|
||
| private volatile FastResponseState fastResponseState; | ||
|
|
||
| public FollowersChecker(Settings settings, TransportService transportService, | ||
| Consumer<FollowerCheckRequest> handleRequestAndUpdateState, | ||
| BiConsumer<DiscoveryNode, String> onNodeFailure, Client nodeClient) { | ||
| this.settings = settings; | ||
| this.transportService = transportService; | ||
| this.handleRequestAndUpdateState = handleRequestAndUpdateState; | ||
|
|
@@ -118,6 +121,7 @@ public void onNodeDisconnected(DiscoveryNode node, Transport.Connection connecti | |
| handleDisconnectedNode(node); | ||
| } | ||
| }); | ||
| nodeFsHealthChecker = new NodeFsHealthChecker(settings, transportService, nodeClient, this::failFollower, this::followers); | ||
| } | ||
|
|
||
| /** | ||
|
|
@@ -139,9 +143,37 @@ public void setCurrentNodes(DiscoveryNodes discoveryNodes) { | |
| followerChecker.start(); | ||
| } | ||
| }); | ||
| nodeFsHealthChecker.start(); | ||
| } | ||
| } | ||
|
|
||
| private Set<DiscoveryNode> followers(){ | ||
| return followerCheckers.keySet(); | ||
| } | ||
|
|
||
| private void failFollower(DiscoveryNode discoveryNode, String reason, Supplier<Boolean> supplier) { | ||
| transportService.getThreadPool().generic().execute(new Runnable() { | ||
| @Override | ||
| public void run() { | ||
| synchronized (mutex) { | ||
| if (supplier.get() == false) { | ||
| logger.trace("{} no longer running, not marking faulty", discoveryNode); | ||
| return; | ||
| } | ||
| logger.debug("{} marking node as faulty", discoveryNode); | ||
| faultyNodes.add(discoveryNode); | ||
| followerCheckers.remove(discoveryNode); | ||
| } | ||
| onNodeFailure.accept(discoveryNode, reason); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "detected failure of " + discoveryNode; | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| /** | ||
| * Clear the set of known nodes, stopping all checks. | ||
| */ | ||
|
|
@@ -351,28 +383,10 @@ public String executor() { | |
| } | ||
|
|
||
| void failNode(String reason) { | ||
| failFollower(discoveryNode, reason, () -> running()); | ||
| } | ||
|
|
||
|
|
||
| private void scheduleNextWakeUp() { | ||
| transportService.getThreadPool().schedule(new Runnable() { | ||
| @Override | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -42,6 +42,7 @@ | |||||
| import org.elasticsearch.common.settings.Settings; | ||||||
| import org.elasticsearch.common.unit.TimeValue; | ||||||
| import org.elasticsearch.discovery.DiscoveryModule; | ||||||
| import org.elasticsearch.monitor.fs.FsService; | ||||||
| import org.elasticsearch.threadpool.ThreadPool; | ||||||
| import org.elasticsearch.threadpool.ThreadPool.Names; | ||||||
| import org.elasticsearch.transport.TransportChannel; | ||||||
|
|
@@ -88,6 +89,8 @@ public class JoinHelper { | |||||
|
|
||||||
| @Nullable // if using single-node discovery | ||||||
| private final TimeValue joinTimeout; | ||||||
| private final FsService fsService; | ||||||
|
|
||||||
|
|
||||||
| private final Set<Tuple<DiscoveryNode, JoinRequest>> pendingOutgoingJoins = Collections.synchronizedSet(new HashSet<>()); | ||||||
|
|
||||||
|
|
@@ -96,9 +99,10 @@ public class JoinHelper { | |||||
| JoinHelper(Settings settings, AllocationService allocationService, MasterService masterService, | ||||||
| TransportService transportService, LongSupplier currentTermSupplier, Supplier<ClusterState> currentStateSupplier, | ||||||
| BiConsumer<JoinRequest, JoinCallback> joinHandler, Function<StartJoinRequest, Join> joinLeaderInTerm, | ||||||
| Collection<BiConsumer<DiscoveryNode, ClusterState>> joinValidators, RerouteService rerouteService, FsService fsService) { | ||||||
| this.masterService = masterService; | ||||||
| this.transportService = transportService; | ||||||
| this.fsService = fsService; | ||||||
| this.joinTimeout = DiscoveryModule.isSingleNodeDiscovery(settings) ? null : JOIN_TIMEOUT_SETTING.get(settings); | ||||||
| this.joinTaskExecutor = new JoinTaskExecutor(allocationService, logger, rerouteService) { | ||||||
|
|
||||||
|
|
@@ -232,6 +236,10 @@ void logLastFailedJoinAttempt() { | |||||
|
|
||||||
| public void sendJoinRequest(DiscoveryNode destination, long term, Optional<Join> optionalJoin) { | ||||||
| assert destination.isMasterNode() : "trying to join master-ineligible " + destination; | ||||||
| if (fsService.stats().getTotal().isWritable() == Boolean.FALSE) { | ||||||
| logger.warn("All paths are not writable. Blocking join request"); | ||||||
|
There was a problem hiding this comment. A warning here isn't helpful, we should be logging the failure elsewhere so this will simply result in confusion.
Suggested change
Also, we have this generic |
||||||
| return; | ||||||
| } | ||||||
| final JoinRequest joinRequest = new JoinRequest(transportService.getLocalNode(), term, optionalJoin); | ||||||
| final Tuple<DiscoveryNode, JoinRequest> dedupKey = Tuple.tuple(destination, joinRequest); | ||||||
| if (pendingOutgoingJoins.add(dedupKey)) { | ||||||
|
|
||||||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| /* | ||
| * 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.LogManager; | ||
| import org.apache.logging.log4j.Logger; | ||
| import org.elasticsearch.action.admin.cluster.node.stats.NodeStats; | ||
| import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsRequest; | ||
| import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse; | ||
| import org.elasticsearch.client.Client; | ||
| import org.elasticsearch.cluster.block.ClusterBlockException; | ||
| import org.elasticsearch.cluster.node.DiscoveryNode; | ||
| import org.elasticsearch.common.TriConsumer; | ||
| import org.elasticsearch.common.settings.Setting; | ||
| import org.elasticsearch.common.settings.Settings; | ||
| import org.elasticsearch.common.unit.TimeValue; | ||
| import org.elasticsearch.monitor.fs.FsHealthService; | ||
| import org.elasticsearch.threadpool.ThreadPool; | ||
| import org.elasticsearch.transport.ReceiveTimeoutTransportException; | ||
| import org.elasticsearch.transport.TransportService; | ||
|
|
||
| import java.util.Set; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public class NodeFsHealthChecker { | ||
|
There was a problem hiding this comment. This doesn't seem necessary, it's enough for followers to reject the today's health checks. There was a problem hiding this comment. Agree. That simplifies a great deal |
||
|
|
||
| private static final Logger logger = LogManager.getLogger(FollowersChecker.class); | ||
|
|
||
| public static final Setting<TimeValue> FS_HEALTH_CHECK_INTERVAL_SETTING = | ||
| Setting.timeSetting("cluster.fault_detection.fs_health_check.interval", | ||
| TimeValue.timeValueMillis(5000), TimeValue.timeValueMillis(100), Setting.Property.NodeScope); | ||
|
|
||
| private final TimeValue fsHealthCheckInterval; | ||
| private final boolean fsHealthCheckEnabled; | ||
| private final TriConsumer<DiscoveryNode, String, Supplier<Boolean>> failFollower; | ||
| private final TransportService transportService; | ||
| private final Supplier<Set<DiscoveryNode>> followerNodesSupplier; | ||
| private final Client nodeClient; | ||
|
|
||
| public NodeFsHealthChecker(Settings settings, TransportService transportService, Client nodeClient, TriConsumer<DiscoveryNode, | ||
| String, Supplier<Boolean>> failFollower, Supplier<Set<DiscoveryNode>> followerNodesSupplier){ | ||
| fsHealthCheckEnabled = FsHealthService.ENABLED_SETTING.get(settings); | ||
| fsHealthCheckInterval = FS_HEALTH_CHECK_INTERVAL_SETTING.get(settings); | ||
| this.followerNodesSupplier = followerNodesSupplier; | ||
| this.failFollower = failFollower; | ||
| this.transportService = transportService; | ||
| this.nodeClient = nodeClient; | ||
| } | ||
|
|
||
|
|
||
| void start() { | ||
| handleWakeUp(); | ||
| } | ||
|
|
||
|
|
||
| private void scheduleNextWakeUp() { | ||
| if (followerNodesSupplier.get().isEmpty() == false) { | ||
| transportService.getThreadPool().schedule(new Runnable() { | ||
| @Override | ||
| public void run() { | ||
| handleWakeUp(); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return NodeFsHealthChecker.this + "::handleWakeUp"; | ||
| } | ||
| }, fsHealthCheckInterval, ThreadPool.Names.SAME); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private void handleWakeUp() { | ||
| if (fsHealthCheckEnabled && followerNodesSupplier.get().isEmpty() == false) { | ||
| NodesStatsRequest nodesStatsRequest = new NodesStatsRequest().clear().fs(true).timeout(fsHealthCheckInterval); | ||
| NodesStatsResponse nodesStatsResponse = fetchNodeStats(nodesStatsRequest); | ||
| if(nodesStatsResponse == null){ | ||
| return; | ||
| } | ||
| for (NodeStats nodeStats : nodesStatsResponse.getNodes()) { | ||
| if (nodeStats.getFs() == null) { | ||
| logger.warn("Unable to retrieve node FS stats for {}", nodeStats.getNode().getName()); | ||
| } else { | ||
| if (nodeStats.getFs().getTotal().isWritable() == Boolean.FALSE) { | ||
| failFollower.apply(nodeStats.getNode(), "read-only-file-system", () -> | ||
| followerNodesSupplier.get().contains(nodeStats.getNode())); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| scheduleNextWakeUp(); | ||
| } | ||
|
|
||
| private NodesStatsResponse fetchNodeStats(NodesStatsRequest nodeStatsRequest) { | ||
| NodesStatsResponse nodesStatsResponse = null; | ||
| try { | ||
| nodesStatsResponse = nodeClient.admin().cluster().nodesStats(nodeStatsRequest).actionGet(); | ||
| } catch (Exception e){ | ||
| if (e instanceof ReceiveTimeoutTransportException) { | ||
| logger.error("NodeStatsRequest timed out for FollowerChecker", e); | ||
| } else { | ||
| if (e instanceof ClusterBlockException) { | ||
| if (logger.isTraceEnabled()) { | ||
| logger.trace("Failed to execute NodeStatsRequest for FollowerChecker", e); | ||
| } | ||
| } else { | ||
| logger.warn("Failed to execute NodeStatsRequest for FollowerChecker", e); | ||
| } | ||
| } | ||
| } | ||
| return nodesStatsResponse; | ||
| } | ||
| } | ||
|
|
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I have left out spaces assuming
checkStyleswould catch. But unfortunate. I'll fix white spacing