diff --git a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryGateMonitor.java b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryGateMonitor.java index 37e20d7e7b797..c517f36096f3b 100644 --- a/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryGateMonitor.java +++ b/server/src/main/java/org/elasticsearch/indices/recovery/RecoveryGateMonitor.java @@ -9,6 +9,8 @@ package org.elasticsearch.indices.recovery; +import org.elasticsearch.common.settings.ClusterSettings; +import org.elasticsearch.common.settings.Setting; import org.elasticsearch.common.util.CachedSupplier; import org.elasticsearch.core.TimeValue; import org.elasticsearch.logging.LogManager; @@ -31,6 +33,13 @@ public final class RecoveryGateMonitor { private static final Logger logger = LogManager.getLogger(RecoveryGateMonitor.class); + public static final Setting ENABLE_RECOVERY_GATES_SETTING = Setting.boolSetting( + "indices.recovery.gates.enabled", + false, + Setting.Property.Dynamic, + Setting.Property.NodeScope + ); + /// How often to re-evaluate the gates while a callback is waiting. // TODO: make this configurable via a node setting private static final TimeValue RECHECK_INTERVAL = TimeValue.timeValueSeconds(1); @@ -39,21 +48,27 @@ public final class RecoveryGateMonitor { private final Supplier> gates; private final ThreadPool threadPool; + private volatile boolean gatesEnabled; + /// One-shot callbacks awaiting an outcome, fired and cleared by a [#check] that evaluates to it. Guarded by `this`. private final Map> outcomeCallbacks = new EnumMap<>(RecoveryGate.Outcome.class); /// Whether a recheck is scheduled; at most one is pending at a time. Guarded by `this`. private boolean recheckScheduled; - public RecoveryGateMonitor(Supplier> gatesSupplier, ThreadPool threadPool) { + public RecoveryGateMonitor(Supplier> gatesSupplier, ThreadPool threadPool, ClusterSettings clusterSettings) { this.gates = CachedSupplier.wrap(() -> List.copyOf(gatesSupplier.get())); this.threadPool = threadPool; + clusterSettings.initializeAndWatchIfRegistered(ENABLE_RECOVERY_GATES_SETTING, enabled -> this.gatesEnabled = enabled); } /// The current node-wide decision, most-restrictive-wins: the first blocking gate's decision, else [RecoveryGate.Decision#RUN]. /// A gate that throws is ignored (failing open, i.e. towards pre-gating behaviour) with a warning, so a buggy gate degrades to no /// gating rather than stalling recoveries indefinitely. public RecoveryGate.Decision evaluate() { + if (gatesEnabled == false) { + return RecoveryGate.Decision.RUN; + } for (RecoveryGate gate : gates.get()) { final RecoveryGate.Decision decision; try { diff --git a/server/src/main/java/org/elasticsearch/node/NodeConstruction.java b/server/src/main/java/org/elasticsearch/node/NodeConstruction.java index cec7ed5b44efc..9830a470bd353 100644 --- a/server/src/main/java/org/elasticsearch/node/NodeConstruction.java +++ b/server/src/main/java/org/elasticsearch/node/NodeConstruction.java @@ -944,7 +944,8 @@ public Map logFields() { // Recovery gates may be contributed by plugins and are resolved once on first use, by which point plugin components exist. final RecoveryGateMonitor recoveryGateMonitor = new RecoveryGateMonitor( () -> pluginsService.filterPlugins(RecoveryPlugin.class).flatMap(p -> p.getRecoveryGates().stream()).toList(), - threadPool + threadPool, + clusterService.getClusterSettings() ); final ThrottlingRecoveryService throttlingRecoveryService = new ThrottlingRecoveryService( threadPool, diff --git a/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryGateMonitorTests.java b/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryGateMonitorTests.java index 9f9ea2e9786d8..215802f403d1d 100644 --- a/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryGateMonitorTests.java +++ b/server/src/test/java/org/elasticsearch/indices/recovery/RecoveryGateMonitorTests.java @@ -9,15 +9,19 @@ package org.elasticsearch.indices.recovery; +import org.elasticsearch.common.settings.ClusterSettings; +import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.util.concurrent.DeterministicTaskQueue; import org.elasticsearch.indices.recovery.RecoveryGate.Decision; import org.elasticsearch.test.ESTestCase; import java.util.ArrayList; import java.util.List; +import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import static org.elasticsearch.indices.recovery.RecoveryGateMonitor.ENABLE_RECOVERY_GATES_SETTING; import static org.hamcrest.Matchers.equalTo; public class RecoveryGateMonitorTests extends ESTestCase { @@ -87,7 +91,7 @@ public void testGatesResolvedOnceOnFirstUse() { final var monitor = new RecoveryGateMonitor(() -> { resolutions.incrementAndGet(); return List.of(() -> Decision.RUN); - }, new DeterministicTaskQueue().getThreadPool()); + }, new DeterministicTaskQueue().getThreadPool(), clusterSettingsWithGatesEnabled()); assertThat("supplier must not be resolved at construction", resolutions.get(), equalTo(0)); for (int i = between(1, 3); i > 0; i--) { @@ -160,11 +164,71 @@ public void testCallbackFailureDoesNotPreventOtherCallbacks() { assertFalse(taskQueue.hasDeferredTasks()); } + public void testGatesAreNotConsultedByDefault() { + // Gates are off by default: recoveries are always allowed and the gates are not consulted, not even resolved. + final var monitor = new RecoveryGateMonitor( + () -> { throw new AssertionError("gates must not be resolved while the gates are disabled"); }, + new DeterministicTaskQueue().getThreadPool(), + ClusterSettings.createBuiltInClusterSettings() + ); + for (int i = between(1, 3); i > 0; i--) { + assertTrue(monitor.evaluate().mayRun()); + } + } + + public void testGatesEnabledSettingUpdatesDynamically() { + final var taskQueue = new DeterministicTaskQueue(); + final var clusterSettings = new ClusterSettings(Settings.EMPTY, Set.of(ENABLE_RECOVERY_GATES_SETTING)); + final var monitor = new RecoveryGateMonitor( + () -> List.of(() -> Decision.block(randomIdentifier(), randomAlphaOfLengthBetween(5, 30))), + taskQueue.getThreadPool(), + clusterSettings + ); + assertTrue("gates are disabled by default", monitor.evaluate().mayRun()); + + clusterSettings.applySettings(Settings.builder().put(ENABLE_RECOVERY_GATES_SETTING.getKey(), true).build()); + assertFalse(monitor.evaluate().mayRun()); + + // Resetting the setting disables the gates again. + clusterSettings.applySettings(Settings.EMPTY); + assertTrue(monitor.evaluate().mayRun()); + } + + public void testDisablingGatesReleasesWaitingCallbacks() { + final var taskQueue = new DeterministicTaskQueue(); + final var clusterSettings = clusterSettingsWithGatesEnabled(); + final var monitor = new RecoveryGateMonitor( + () -> List.of(() -> Decision.block(randomIdentifier(), randomAlphaOfLengthBetween(5, 30))), + taskQueue.getThreadPool(), + clusterSettings + ); + + final AtomicInteger fired = new AtomicInteger(); + monitor.addCallback(RecoveryGate.Outcome.RUN, fired::incrementAndGet); + taskQueue.runAllRunnableTasks(); + assertThat(fired.get(), equalTo(0)); + assertTrue("waiting callback starts the periodic recheck", taskQueue.hasDeferredTasks()); + + // Disabling the gates is noticed by the next periodic recheck, which fires the waiting callback and stops rescheduling. + clusterSettings.applySettings(Settings.builder().put(ENABLE_RECOVERY_GATES_SETTING.getKey(), false).build()); + taskQueue.advanceTime(); + taskQueue.runAllRunnableTasks(); + assertThat(fired.get(), equalTo(1)); + assertFalse(taskQueue.hasDeferredTasks()); + } + private static RecoveryGateMonitor newMonitor(DeterministicTaskQueue taskQueue, AtomicReference decision) { return newMonitor(taskQueue, List.of(decision::get)); } private static RecoveryGateMonitor newMonitor(DeterministicTaskQueue taskQueue, List gateList) { - return new RecoveryGateMonitor(() -> gateList, taskQueue.getThreadPool()); + return new RecoveryGateMonitor(() -> gateList, taskQueue.getThreadPool(), clusterSettingsWithGatesEnabled()); + } + + private static ClusterSettings clusterSettingsWithGatesEnabled() { + return new ClusterSettings( + Settings.builder().put(ENABLE_RECOVERY_GATES_SETTING.getKey(), true).build(), + Set.of(ENABLE_RECOVERY_GATES_SETTING) + ); } } diff --git a/server/src/test/java/org/elasticsearch/indices/recovery/ThrottlingRecoveryServiceTests.java b/server/src/test/java/org/elasticsearch/indices/recovery/ThrottlingRecoveryServiceTests.java index 57c96b254b40a..991218405199d 100644 --- a/server/src/test/java/org/elasticsearch/indices/recovery/ThrottlingRecoveryServiceTests.java +++ b/server/src/test/java/org/elasticsearch/indices/recovery/ThrottlingRecoveryServiceTests.java @@ -61,6 +61,7 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; +import static org.elasticsearch.indices.recovery.RecoveryGateMonitor.ENABLE_RECOVERY_GATES_SETTING; import static org.elasticsearch.indices.recovery.ThrottlingRecoveryService.INDICES_RECOVERY_MAX_CONCURRENT_RECOVERIES_SETTING; import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.lessThanOrEqualTo; @@ -1118,7 +1119,7 @@ public void testGateBlocksAllRecoveriesUntilItAllows() { DefaultProjectResolver.INSTANCE, newClusterService(Integer.MAX_VALUE), // plenty of slots, so only the gate can hold recoveries back RecoverySchedulingListener.NOOP, - new RecoveryGateMonitor(() -> List.of(gate), taskQueue.getThreadPool()) + new RecoveryGateMonitor(() -> List.of(gate), taskQueue.getThreadPool(), clusterSettingsWithGatesEnabled()) ); service.start(); @@ -1189,7 +1190,11 @@ public void onRecoveriesUnblocked(long blockedTimeMillis) { final String gateName = randomIdentifier(); final var gateDecision = new AtomicReference<>(RecoveryGate.Decision.block(gateName, randomAlphaOfLengthBetween(5, 30))); final RecoveryGate gate = gateDecision::get; - final var recoveryGateMonitor = new RecoveryGateMonitor(() -> List.of(gate), taskQueue.getThreadPool()); + final var recoveryGateMonitor = new RecoveryGateMonitor( + () -> List.of(gate), + taskQueue.getThreadPool(), + clusterSettingsWithGatesEnabled() + ); final var service = new ThrottlingRecoveryService( taskQueue.getThreadPool(), DefaultProjectResolver.INSTANCE, @@ -1235,6 +1240,61 @@ public void onRecoveriesUnblocked(long blockedTimeMillis) { assertFalse("No more scheduled tasks", taskQueue.hasAnyTasks()); } + /// The gating escape hatch: dynamically disabling the recovery gates must release recoveries held by a gate that never + /// unblocks by itself, via the next periodic recheck. + public void testDisablingGatesReleasesBlockedRecoveries() { + final var taskQueue = new DeterministicTaskQueue(); + + final var unblockedCount = new AtomicInteger(); + final var reportedBlockedMillis = new AtomicLong(-1); + final RecoverySchedulingListener listener = new RecoverySchedulingListener() { + @Override + public void onRecoveriesUnblocked(long blockedTimeMillis) { + unblockedCount.incrementAndGet(); + reportedBlockedMillis.set(blockedTimeMillis); + } + }; + final var clusterSettings = clusterSettingsWithGatesEnabled(); + // This gate never unblocks by itself: only disabling the gates can release the held recoveries. + final RecoveryGate gate = () -> RecoveryGate.Decision.block("stuck", "never unblocks"); + final var service = new ThrottlingRecoveryService( + taskQueue.getThreadPool(), + DefaultProjectResolver.INSTANCE, + newClusterService(Integer.MAX_VALUE), // plenty of slots, so only the gate can hold recoveries back + listener, + new RecoveryGateMonitor(() -> List.of(gate), taskQueue.getThreadPool(), clusterSettings) + ); + service.start(); + + final long blockedSince = taskQueue.getCurrentTimeMillis(); + final var started = new AtomicInteger(); + final int count = between(1, 100); + for (int i = 0; i < count; i++) { + service.enqueue(ProjectId.DEFAULT, RecoveryListener.NOOP, newRecoveryState(), UUIDs.randomBase64UUID(), stats, l -> { + started.incrementAndGet(); + l.onRecoveryDone(null, ShardLongFieldRange.EMPTY, ShardLongFieldRange.EMPTY); + }); + } + taskQueue.runAllRunnableTasks(); + // Stay blocked across a few periodic rechecks. + for (int i = between(0, 3); i > 0; i--) { + taskQueue.advanceTime(); + taskQueue.runAllRunnableTasks(); + } + assertThat(started.get(), equalTo(0)); + assertThat(unblockedCount.get(), equalTo(0)); + + // The next periodic recheck notices the gates are disabled: it dispatches every held recovery, reports the blocked + // duration and stops rescheduling. + clusterSettings.applySettings(Settings.builder().put(ENABLE_RECOVERY_GATES_SETTING.getKey(), false).build()); + taskQueue.advanceTime(); + taskQueue.runAllRunnableTasks(); + assertThat(started.get(), equalTo(count)); + assertThat(unblockedCount.get(), equalTo(1)); + assertThat(reportedBlockedMillis.get(), equalTo(taskQueue.getCurrentTimeMillis() - blockedSince)); + assertFalse("No more scheduled tasks", taskQueue.hasAnyTasks()); + } + /// Hammers the service from multiple real threads while the gate flaps, to catch races between dispatch, the monitor's /// evaluations, and the resume callback: a missed wake-up leaves recoveries queued (the latch below never opens) and a deadlock /// hangs the test. Unlike the deterministic tests above, this uses a real thread pool. @@ -1246,7 +1306,7 @@ public void testConcurrentEnqueuesWithFlappingGateEventuallyDispatchEverything() DefaultProjectResolver.INSTANCE, newClusterService(randomBoolean() ? Integer.MAX_VALUE : between(1, 5)), RecoverySchedulingListener.NOOP, - new RecoveryGateMonitor(() -> List.of(gate), threadPool) + new RecoveryGateMonitor(() -> List.of(gate), threadPool, clusterSettingsWithGatesEnabled()) ); service.start(); @@ -1318,7 +1378,14 @@ private static ThrottlingRecoveryService newStartedService( /// A [RecoveryGateMonitor] with no gates: the decision never transitions, so the change listener never fires. private static RecoveryGateMonitor monitorWithNoGates(ThreadPool threadPool) { - return new RecoveryGateMonitor(() -> List.of(), threadPool); + return new RecoveryGateMonitor(() -> List.of(), threadPool, ClusterSettings.createBuiltInClusterSettings()); + } + + private static ClusterSettings clusterSettingsWithGatesEnabled() { + return new ClusterSettings( + Settings.builder().put(ENABLE_RECOVERY_GATES_SETTING.getKey(), true).build(), + Set.of(ENABLE_RECOVERY_GATES_SETTING) + ); } private static RecoveryState newRecoveryState() { diff --git a/server/src/test/java/org/elasticsearch/indices/recovery/TransportCancelRecoveriesActionTests.java b/server/src/test/java/org/elasticsearch/indices/recovery/TransportCancelRecoveriesActionTests.java index 35d95eef01d94..36feb93bfe4e6 100644 --- a/server/src/test/java/org/elasticsearch/indices/recovery/TransportCancelRecoveriesActionTests.java +++ b/server/src/test/java/org/elasticsearch/indices/recovery/TransportCancelRecoveriesActionTests.java @@ -79,7 +79,7 @@ public void setupAction() { DefaultProjectResolver.INSTANCE, clusterService, RecoverySchedulingListener.NOOP, - new RecoveryGateMonitor(() -> List.of(), taskQueue.getThreadPool()) + new RecoveryGateMonitor(() -> List.of(), taskQueue.getThreadPool(), clusterSettings) ); throttlingRecoveryService.start(); action = new TransportCancelRecoveriesAction( diff --git a/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTestHelper.java b/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTestHelper.java index c7f4a8d8226f9..e5fab0cac39a2 100644 --- a/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTestHelper.java +++ b/server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTestHelper.java @@ -659,7 +659,7 @@ public RecyclerBytesStreamOutput newNetworkBytesStream(@Nullable CircuitBreaker projectResolver, clusterService, RecoverySchedulingListener.NOOP, - new RecoveryGateMonitor(List::of, threadPool) + new RecoveryGateMonitor(List::of, threadPool, clusterService.getClusterSettings()) ); indicesService = new IndicesServiceBuilder().settings(settings) diff --git a/test/framework/src/main/java/org/elasticsearch/test/InternalSettingsPlugin.java b/test/framework/src/main/java/org/elasticsearch/test/InternalSettingsPlugin.java index a4334164bc001..a2e1fc81dd8b7 100644 --- a/test/framework/src/main/java/org/elasticsearch/test/InternalSettingsPlugin.java +++ b/test/framework/src/main/java/org/elasticsearch/test/InternalSettingsPlugin.java @@ -18,6 +18,7 @@ import org.elasticsearch.index.IndexSettings; import org.elasticsearch.index.engine.EngineConfig; import org.elasticsearch.indices.recovery.PeerRecoverySourceService; +import org.elasticsearch.indices.recovery.RecoveryGateMonitor; import org.elasticsearch.indices.recovery.ThrottlingRecoveryService; import org.elasticsearch.monitor.fs.FsService; import org.elasticsearch.plugins.Plugin; @@ -77,7 +78,8 @@ public List> getSettings() { FsService.ALWAYS_REFRESH_SETTING, PeerRecoverySourceService.INDICES_RECOVERY_MAX_CONCURRENT_OUTGOING_RECOVERIES_SETTING, ThrottlingRecoveryService.INDICES_RECOVERY_MAX_CONCURRENT_RECOVERIES_SETTING, - RecoveryDirectCancellationService.ENABLE_DIRECT_RECOVERY_CANCELLATIONS_SETTING + RecoveryDirectCancellationService.ENABLE_DIRECT_RECOVERY_CANCELLATIONS_SETTING, + RecoveryGateMonitor.ENABLE_RECOVERY_GATES_SETTING ); } }