Remove unnecessary iteration in request cache clear (#19263)

Signed-off-by: Peter Alfonsi <petealft@amazon.com>
Signed-off-by: Peter Alfonsi <peter.alfonsi@gmail.com>
Signed-off-by: Ankit Jain <jainankitk@apache.org>
Co-authored-by: Peter Alfonsi <petealft@amazon.com>
Co-authored-by: Ankit Jain <jainankitk@apache.org>

Author: 3 people

CHANGELOG.md

@@ -97,6 +97,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Cache serialised cluster state based on cluster state version and node version.([#19307](https://github.com/opensearch-project/OpenSearch/pull/19307))
 - Fix stats API in store-subdirectory module's SubdirectoryAwareStore ([#19470](https://github.com/opensearch-project/OpenSearch/pull/19470))
 - Handle negative search request nodes stats ([#19340](https://github.com/opensearch-project/OpenSearch/pull/19340))
+- Remove unnecessary iteration per-shard in request cache cleanup ([#19263](https://github.com/opensearch-project/OpenSearch/pull/19263))
 
 ### Dependencies
 - Bump `com.gradleup.shadow:shadow-gradle-plugin` from 8.3.5 to 8.3.9 ([#19400](https://github.com/opensearch-project/OpenSearch/pull/19400))

server/src/main/java/org/opensearch/action/admin/indices/cache/clear/TransportClearIndicesCacheAction.java

@@ -32,7 +32,10 @@
 
 package org.opensearch.action.admin.indices.cache.clear;
 
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 import org.opensearch.action.support.ActionFilters;
+import org.opensearch.action.support.broadcast.BroadcastShardOperationFailedException;
 import org.opensearch.action.support.broadcast.node.TransportBroadcastByNodeAction;
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.block.ClusterBlockException;
@@ -68,6 +71,7 @@ public class TransportClearIndicesCacheAction extends TransportBroadcastByNodeAc
     private final IndicesService indicesService;
 
     private final Node node;
+    private final Logger clearActionLogger = LogManager.getLogger(getClass());
 
     @Inject
     public TransportClearIndicesCacheAction(
@@ -135,6 +139,15 @@ protected EmptyResult shardOperation(ClearIndicesCacheRequest request, ShardRout
         return EmptyResult.INSTANCE;
     }
 
+    @Override
+    protected void nodeOperation(List<EmptyResult> results, List<BroadcastShardOperationFailedException> accumulatedExceptions) {
+        try {
+            indicesService.forceClearNodewideCaches();
+        } catch (Exception e) {
+            clearActionLogger.warn("Node-wide force cache clear failed; marked keys will be cleaned at next scheduled cache cleanup", e);
+        }
+    }
+
     /**
      * The refresh request works against *all* shards.
      */

server/src/main/java/org/opensearch/action/support/broadcast/node/TransportBroadcastByNodeAction.java

@@ -238,6 +238,13 @@ protected abstract Response newResponse(
      */
     protected abstract ShardsIterator shards(ClusterState clusterState, Request request, String[] concreteIndices);
 
+    /**
+     * Executes a node-level operation. This method is called one time per node, after all shard-level operations have completed.
+     * @param results List of results from the completed shard-level operations.
+     * @param accumulatedExceptions List of any exceptions thrown by the shard-level operations.
+     */
+    protected void nodeOperation(List<ShardOperationResult> results, List<BroadcastShardOperationFailedException> accumulatedExceptions) {}
+
     /**
      * Executes a global block check before polling the cluster state.
      *
@@ -479,6 +486,8 @@ public void messageReceived(final NodeRequest request, TransportChannel channel,
                 }
             }
 
+            nodeOperation(results, accumulatedExceptions);
+
             channel.sendResponse(new NodeResponse(request.getNodeId(), totalShards, results, accumulatedExceptions));
         }
 

server/src/main/java/org/opensearch/indices/IndicesRequestCache.java

@@ -266,6 +266,9 @@ void setStalenessThreshold(String threshold) {
 
     void clear(CacheEntity entity) {
         cacheCleanupManager.enqueueCleanupKey(new CleanupKey(entity, null));
+    }
+
+    void forceCleanCache() {
         cacheCleanupManager.forceCleanCache();
     }
 

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -2117,6 +2117,8 @@ private QueryRewriteContext getRewriteContext(LongSupplier nowInMillis, boolean
 
     /**
      * Clears the caches for the given shard id if the shard is still allocated on this node
+     * Query cache and field data cache keys are cleared immediately by this method; request cache keys are only marked for cleanup
+     * TODO: In a future PR field data cache keys will also only be marked for cleanup.
      */
     public void clearIndexShardCache(ShardId shardId, boolean queryCache, boolean fieldDataCache, boolean requestCache, String... fields) {
         final IndexService service = indexService(shardId.getIndex());
@@ -2129,6 +2131,14 @@ public void clearIndexShardCache(ShardId shardId, boolean queryCache, boolean fi
         }
     }
 
+    /**
+     * Force clear node-wide request cache, which will remove any keys which have been previously marked for cleanup.
+     */
+    public void forceClearNodewideCaches() {
+        indicesRequestCache.forceCleanCache();
+        // TODO: Field data cache will also be cleared here in a future PR.
+    }
+
     /**
      * Returns a function which given an index name, returns a predicate which fields must match in order to be returned by get mappings,
      * get index, get field mappings and field capabilities API. Useful to filter the fields that such API return.

server/src/test/java/org/opensearch/action/support/broadcast/node/TransportBroadcastByNodeActionTests.java

@@ -59,15 +59,18 @@
 import org.opensearch.cluster.routing.ShardsIterator;
 import org.opensearch.cluster.routing.TestShardRouting;
 import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.common.metrics.CounterMetric;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.concurrent.ThreadContext;
+import org.opensearch.core.action.ActionListener;
 import org.opensearch.core.action.support.DefaultShardOperationFailedException;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.Writeable;
 import org.opensearch.core.index.Index;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.core.rest.RestStatus;
 import org.opensearch.core.transport.TransportResponse;
+import org.opensearch.tasks.Task;
 import org.opensearch.telemetry.tracing.noop.NoopTracer;
 import org.opensearch.test.OpenSearchTestCase;
 import org.opensearch.test.transport.CapturingTransport;
@@ -99,6 +102,7 @@
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.object.HasToString.hasToString;
+import static org.mockito.Mockito.mock;
 
 public class TransportBroadcastByNodeActionTests extends OpenSearchTestCase {
 
@@ -137,6 +141,7 @@ class TestTransportBroadcastByNodeAction extends TransportBroadcastByNodeAction<
         Response,
         TransportBroadcastByNodeAction.EmptyResult> {
         private final Map<ShardRouting, Object> shards = new HashMap<>();
+        private final CounterMetric nodeOperationCount = new CounterMetric();
 
         TestTransportBroadcastByNodeAction(
             TransportService transportService,
@@ -206,6 +211,18 @@ protected ClusterBlockException checkRequestBlock(ClusterState state, Request re
             return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_WRITE, concreteIndices);
         }
 
+        @Override
+        protected void nodeOperation(
+            List<TransportBroadcastByNodeAction.EmptyResult> results,
+            List<BroadcastShardOperationFailedException> accumulatedExceptions
+        ) {
+            nodeOperationCount.inc();
+        }
+
+        public long getNodeOperationCount() {
+            return nodeOperationCount.count();
+        }
+
         public Map<ShardRouting, Object> getResults() {
             return shards;
         }
@@ -452,6 +469,33 @@ public void testOperationExecution() throws Exception {
             assertThat(exception.getMessage(), is("operation indices:admin/test failed"));
             assertThat(exception, hasToString(containsString("operation failed")));
         }
+        assertEquals(1, action.getNodeOperationCount());
+    }
+
+    public void testNodeLevelHookForMultiNode() throws Exception {
+        // Manually send the request from coordinator --> target nodes, action.doExecute() does not do that in this test case setup
+
+        Request request = new Request(new String[] { TEST_INDEX });
+        final PlainActionFuture<Response> listener = PlainActionFuture.newFuture();
+        action.doExecute(mock(Task.class), request, ActionListener.wrap(listener::onResponse, listener::onFailure));
+
+        // Get captured NodeRequest from coordinator for each target node
+        Map<String, List<CapturingTransport.CapturedRequest>> captured = transport.getCapturedRequestsByTargetNodeAndClear();
+        assertFalse("expected at least one target node", captured.isEmpty());
+
+        final TransportBroadcastByNodeAction<?, ?, ?>.BroadcastByNodeTransportRequestHandler handler =
+            action.new BroadcastByNodeTransportRequestHandler();
+
+        for (Map.Entry<String, List<CapturingTransport.CapturedRequest>> e : captured.entrySet()) {
+            CapturingTransport.CapturedRequest cr = e.getValue().get(0);
+            assertEquals(action.transportNodeBroadcastAction, cr.action);
+            TransportBroadcastByNodeAction.NodeRequest nodeReq = (TransportBroadcastByNodeAction.NodeRequest) cr.request;
+
+            final PlainActionFuture<TransportResponse> future = PlainActionFuture.newFuture();
+            TestTransportChannel channel = new TestTransportChannel(future);
+            handler.messageReceived(nodeReq, channel, null);
+        }
+        assertEquals("nodeOperation should run once per target node", captured.size(), action.getNodeOperationCount());
     }
 
     public void testResultAggregation() throws ExecutionException, InterruptedException {