Add enabled setting for recovery gates

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<Boolean> 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<List<RecoveryGate>> 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<RecoveryGate.Outcome, List<Runnable>> 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<Collection<RecoveryGate>> gatesSupplier, ThreadPool threadPool) {
+    public RecoveryGateMonitor(Supplier<Collection<RecoveryGate>> 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 {

server/src/main/java/org/elasticsearch/node/NodeConstruction.java

@@ -944,7 +944,8 @@ public Map<String, String> 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,

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> decision) {
         return newMonitor(taskQueue, List.of(decision::get));
     }
 
     private static RecoveryGateMonitor newMonitor(DeterministicTaskQueue taskQueue, List<RecoveryGate> 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)
+        );
     }
 }

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() {

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(

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)

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<Setting<?>> 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
         );
     }
 }