Suspend Index throttling when relocating

server/src/internalClusterTest/java/org/elasticsearch/recovery/RelocationIT.java

@@ -43,6 +43,8 @@
 import org.elasticsearch.index.shard.IndexShard;
 import org.elasticsearch.index.shard.IndexShardState;
 import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.indices.IndexingMemoryController;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.recovery.PeerRecoveryTargetService;
 import org.elasticsearch.indices.recovery.RecoveryFileChunkRequest;
 import org.elasticsearch.plugins.Plugin;
@@ -162,10 +164,82 @@ public void testSimpleRelocationNoIndexing() {
         assertHitCount(prepareSearch("test").setSize(0), 20);
     }
 
+    // This tests that relocation can successfully suspend index throttling to grab
+    // indexing permits required for relocation to succeed.
+    public void testSimpleRelocationWithIndexingPaused() throws Exception {
+        logger.info("--> starting [node1] ...");
+        // Start node with PAUSE_INDEXING_ON_THROTTLE setting set to true. This means that if we activate
+        // index throttling for a shard on this node, it will pause indexing for that shard until throttling
+        // is deactivated.
+        final String node_1 = internalCluster().startNode(
+            Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true)
+        );
+
+        logger.info("--> creating test index ...");
+        prepareCreate("test", indexSettings(1, 0)).get();
+
+        logger.info("--> index 10 docs");
+        for (int i = 0; i < 10; i++) {
+            prepareIndex("test").setId(Integer.toString(i)).setSource("field", "value" + i).get();
+        }
+        logger.info("--> flush so we have an actual index");
+        indicesAdmin().prepareFlush().get();
+        logger.info("--> index more docs so we have something in the translog");
+        for (int i = 10; i < 20; i++) {
+            prepareIndex("test").setId(Integer.toString(i)).setSource("field", "value" + i).get();
+        }
+
+        logger.info("--> verifying count");
+        indicesAdmin().prepareRefresh().get();
+        assertHitCount(prepareSearch("test").setSize(0), 20L);
+
+        logger.info("--> start another node");
+        final String node_2 = internalCluster().startNode();
+        ClusterHealthResponse clusterHealthResponse = clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT)
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNodes("2")
+            .get();
+        assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
+
+        // Activate index throttling on "test" index primary shard
+        IndicesService indicesService = internalCluster().getInstance(IndicesService.class, node_1);
+        IndexShard shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
+        shard.activateThrottling();
+        // Verify that indexing is paused for the throttled shard
+        assertBusy(() -> { assertThat(shard.isIndexingPaused(), equalTo(true)); });
+        // Try to index a document into the "test" index which is currently throttled
+        logger.info("--> Try to index a doc while indexing is paused");
+        IndexRequestBuilder indexRequestBuilder = prepareIndex("test").setId(Integer.toString(20)).setSource("field", "value" + 20);
+        var future = indexRequestBuilder.execute();
+        // Verify that the new document has not been indexed indicating that the indexing thread is paused.
+        logger.info("--> verifying count is unchanged...");
+        indicesAdmin().prepareRefresh().get();
+        assertHitCount(prepareSearch("test").setSize(0), 20);
+
+        logger.info("--> relocate the shard from node1 to node2");
+        ClusterRerouteUtils.reroute(client(), new MoveAllocationCommand("test", 0, node_1, node_2));
+
+        // Relocation will suspend throttling for the paused shard, allow the indexing thread to proceed, thereby releasing
+        // the indexing permit it holds, in turn allowing relocation to acquire the permits and proceed.
+        clusterHealthResponse = clusterAdmin().prepareHealth(TEST_REQUEST_TIMEOUT)
+            .setWaitForEvents(Priority.LANGUID)
+            .setWaitForNoRelocatingShards(true)
+            .setTimeout(ACCEPTABLE_RELOCATION_TIME)
+            .get();
+        assertThat(clusterHealthResponse.isTimedOut(), equalTo(false));
+
+        // Relocated shard is not throttled
+        assertThat(shard.isIndexingPaused(), equalTo(false));
+        logger.info("--> verifying count after relocation ...");
+        indicesAdmin().prepareRefresh().get();
+        assertHitCount(prepareSearch("test").setSize(0), 21);
+    }
+
     public void testRelocationWhileIndexingRandom() throws Exception {
         int numberOfRelocations = scaledRandomIntBetween(1, rarely() ? 10 : 4);
         int numberOfReplicas = randomBoolean() ? 0 : 1;
         int numberOfNodes = numberOfReplicas == 0 ? 2 : 3;
+        boolean throttleIndexing = randomBoolean();
 
         logger.info(
             "testRelocationWhileIndexingRandom(numRelocations={}, numberOfReplicas={}, numberOfNodes={})",
@@ -174,9 +248,12 @@ public void testRelocationWhileIndexingRandom() throws Exception {
             numberOfNodes
         );
 
+        // Start node with PAUSE_INDEXING_ON_THROTTLE setting set to true. This means that if we activate
+        // index throttling for a shard on this node, it will pause indexing for that shard until throttling
+        // is deactivated.
         String[] nodes = new String[numberOfNodes];
         logger.info("--> starting [node1] ...");
-        nodes[0] = internalCluster().startNode();
+        nodes[0] = internalCluster().startNode(Settings.builder().put(IndexingMemoryController.PAUSE_INDEXING_ON_THROTTLE.getKey(), true));
 
         logger.info("--> creating test index ...");
         prepareCreate("test", indexSettings(1, numberOfReplicas)).get();
@@ -200,6 +277,14 @@ public void testRelocationWhileIndexingRandom() throws Exception {
             waitForDocs(numDocs, indexer);
             logger.info("--> {} docs indexed", numDocs);
 
+            if (throttleIndexing) {
+                // Activate index throttling on "test" index primary shard
+                IndicesService indicesService = internalCluster().getInstance(IndicesService.class, nodes[0]);
+                IndexShard shard = indicesService.indexServiceSafe(resolveIndex("test")).getShard(0);
+                shard.activateThrottling();
+                // Verify that indexing is paused for the throttled shard
+                assertBusy(() -> { assertThat(shard.isIndexingPaused(), equalTo(true)); });
+            }
             logger.info("--> starting relocations...");
             int nodeShiftBased = numberOfReplicas; // if we have replicas shift those
             for (int i = 0; i < numberOfRelocations; i++) {

server/src/main/java/org/elasticsearch/index/engine/Engine.java

@@ -460,7 +460,10 @@ protected static final class IndexThrottle {
         private final Condition pauseCondition = pauseIndexingLock.newCondition();
         private final ReleasableLock pauseLockReference = new ReleasableLock(pauseIndexingLock);
         private volatile AtomicBoolean suspendThrottling = new AtomicBoolean();
-        private final boolean pauseWhenThrottled; // Should throttling pause indexing ?
+
+        // Should throttling pause indexing ? This is decided by the
+        // IndexingMemoryController#PAUSE_INDEXING_ON_THROTTLE setting for this node.
+        private final boolean pauseWhenThrottled;
         private volatile ReleasableLock lock = NOOP_LOCK;
 
         public IndexThrottle(boolean pause) {
@@ -491,7 +494,6 @@ public Releasable acquireThrottle() {
         /** Activate throttling, which switches the lock to be a real lock */
         public void activate() {
             assert lock == NOOP_LOCK : "throttling activated while already active";
-
             startOfThrottleNS = System.nanoTime();
             if (pauseWhenThrottled) {
                 lock = pauseLockReference;
@@ -539,10 +541,14 @@ boolean isThrottled() {
             return lock != NOOP_LOCK;
         }
 
+        boolean isIndexingPaused() {
+            return (lock == pauseLockReference);
+        }
+
         /** Suspend throttling to allow another task such as relocation to acquire all indexing permits */
         public void suspendThrottle() {
             if (pauseWhenThrottled) {
-                try (Releasable releasableLock = pauseLockReference.acquire()) {
+                try (Releasable ignored = pauseLockReference.acquire()) {
                     suspendThrottling.setRelease(true);
                     pauseCondition.signalAll();
                 }
@@ -552,7 +558,7 @@ public void suspendThrottle() {
         /** Reverse what was done in {@link #suspendThrottle()} */
         public void resumeThrottle() {
             if (pauseWhenThrottled) {
-                try (Releasable releasableLock = pauseLockReference.acquire()) {
+                try (Releasable ignored = pauseLockReference.acquire()) {
                     suspendThrottling.setRelease(false);
                     pauseCondition.signalAll();
                 }
@@ -2258,6 +2264,20 @@ public interface Warmer {
      */
     public abstract void deactivateThrottling();
 
+    /**
+     * If indexing is throttled to the point where it is paused completely,
+     * another task trying to get indexing permits might want to pause throttling
+     * by letting one thread pass at a time so that it does not get starved.
+     */
+    public abstract void suspendThrottling();
+
+    /**
+     * Reverses a previous {@link #suspendThrottling} call.
+     */
+    public abstract void resumeThrottling();
+
+    public abstract boolean isIndexingPaused();
+
     /**
      * This method replays translog to restore the Lucene index which might be reverted previously.
      * This ensures that all acknowledged writes are restored correctly when this engine is promoted.

server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java

@@ -2852,11 +2852,26 @@ public void deactivateThrottling() {
         }
     }
 
+    @Override
+    public void suspendThrottling() {
+        throttle.suspendThrottle();
+    }
+
+    @Override
+    public void resumeThrottling() {
+        throttle.resumeThrottle();
+    }
+
     @Override
     public boolean isThrottled() {
         return throttle.isThrottled();
     }
 
+    @Override
+    public boolean isIndexingPaused() {
+        return throttle.isIndexingPaused();
+    }
+
     boolean throttleLockIsHeldByCurrentThread() {  // to be used in assertions and tests only
         return throttle.throttleLockIsHeldByCurrentThread();
     }

server/src/main/java/org/elasticsearch/index/engine/ReadOnlyEngine.java

@@ -507,6 +507,17 @@ public void activateThrottling() {}
     @Override
     public void deactivateThrottling() {}
 
+    @Override
+    public void suspendThrottling() {}
+
+    @Override
+    public void resumeThrottling() {}
+
+    @Override
+    public boolean isIndexingPaused() {
+        return (false);
+    }
+
     @Override
     public void trimUnreferencedTranslogFiles() {}
 

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -879,8 +879,14 @@ public void onFailure(Exception e) {
                         listener.onFailure(e);
                     }
                 }
-            }, 30L, TimeUnit.MINUTES, EsExecutors.DIRECT_EXECUTOR_SERVICE); // Wait on current thread because this execution is wrapped by
-                                                                            // CancellableThreads and we want to be able to interrupt it
+            },
+                30L,
+                TimeUnit.MINUTES,
+                // Wait on current thread because this execution is wrapped by CancellableThreads and we want to be able to interrupt it
+                EsExecutors.DIRECT_EXECUTOR_SERVICE,
+                this
+            );
+
         }
     }
 
@@ -2752,6 +2758,40 @@ public void deactivateThrottling() {
         }
     }
 
+    public boolean isIndexingPaused() {
+        Engine engine = getEngineOrNull();
+        final boolean indexingPaused;
+        if (engine == null) {
+            indexingPaused = false;
+        } else {
+            indexingPaused = engine.isIndexingPaused();
+        }
+        return (indexingPaused);
+    }
+
+    public boolean suspendThrottling() {
+        Engine engine = getEngineOrNull();
+        final boolean indexingPaused;
+        if (engine == null) {
+            indexingPaused = false;
+        } else {
+            indexingPaused = engine.isIndexingPaused();
+        }
+        if (indexingPaused) {
+            engine.suspendThrottling();
+            return (true);
+        }
+        return (false);
+    }
+
+    public void resumeThrottling() {
+        try {
+            getEngine().resumeThrottling();
+        } catch (AlreadyClosedException ex) {
+            // ignore
+        }
+    }
+
     private void handleRefreshException(Exception e) {
         if (e instanceof AlreadyClosedException) {
             // ignore
@@ -3811,7 +3851,7 @@ private void asyncBlockOperations(ActionListener<Releasable> onPermitAcquired, l
             onPermitAcquired.onFailure(e);
         });
         try {
-            indexShardOperationPermits.blockOperations(wrappedListener, timeout, timeUnit, threadPool.generic());
+            indexShardOperationPermits.blockOperations(wrappedListener, timeout, timeUnit, threadPool.generic(), this);
         } catch (Exception e) {
             forceRefreshes.close();
             throw e;

server/src/main/java/org/elasticsearch/index/shard/IndexShardOperationPermits.java

@@ -84,10 +84,16 @@ public void blockOperations(
         final ActionListener<Releasable> onAcquired,
         final long timeout,
         final TimeUnit timeUnit,
-        final Executor executor
+        final Executor executor,
+        @Nullable IndexShard indexShard
     ) {
         delayOperations();
+        // In case indexing is paused on the shard, suspend throttling so that any currently paused task can
+        // go ahead and release the indexing permit it holds.
+        indexShard.suspendThrottling();
         waitUntilBlocked(ActionListener.assertOnce(onAcquired), timeout, timeUnit, executor);
+        // TODO: Does this do anything ? Looks like the relocated shard does not have throttling enabled
+        indexShard.resumeThrottling();
     }
 
     private void waitUntilBlocked(ActionListener<Releasable> onAcquired, long timeout, TimeUnit timeUnit, Executor executor) {