Remove Cluster Manager Routing

Author: tqranjan

server/src/main/java/org/opensearch/action/admin/cluster/cache/PruneCacheAction.java

@@ -11,16 +11,16 @@
 import org.opensearch.action.ActionType;
 
 /**
- * Transport action to prune remote file cache
+ * Transport action to prune file cache
  *
  * @opensearch.internal
  */
 public class PruneCacheAction extends ActionType<PruneCacheResponse> {
 
     public static final PruneCacheAction INSTANCE = new PruneCacheAction();
-    public static final String NAME = "cluster:admin/cache/remote/prune";
+    public static final String NAME = "cluster:admin/filecache/prune";
 
-    public PruneCacheAction() {
+    private PruneCacheAction() {
         super(NAME, PruneCacheResponse::new);
     }
 }

server/src/main/java/org/opensearch/action/admin/cluster/cache/PruneCacheRequest.java

@@ -8,19 +8,24 @@
 
 package org.opensearch.action.admin.cluster.cache;
 
+import org.opensearch.action.ActionRequest;
 import org.opensearch.action.ActionRequestValidationException;
-import org.opensearch.action.support.clustermanager.ClusterManagerNodeRequest;
+import org.opensearch.common.unit.TimeValue;
 import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.core.common.io.stream.StreamOutput;
 
 import java.io.IOException;
 
 /**
- * Request for pruning remote file cache
+ * Request for pruning file cache.
+ * Extends ActionRequest for direct node execution.
  *
  * @opensearch.internal
  */
-public class PruneCacheRequest extends ClusterManagerNodeRequest<PruneCacheRequest> {
+public class PruneCacheRequest extends ActionRequest {
+
+    private static final TimeValue DEFAULT_TIMEOUT = TimeValue.timeValueSeconds(30);
+    private TimeValue timeout = DEFAULT_TIMEOUT;
 
     /**
      * Default constructor
@@ -35,11 +40,37 @@ public PruneCacheRequest() {}
      */
     public PruneCacheRequest(StreamInput in) throws IOException {
         super(in);
+        timeout = in.readTimeValue();
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
+        out.writeTimeValue(timeout);
+    }
+
+    /**
+     * Sets the timeout for this request.
+     * Note: Currently preserved for API compatibility but not actively enforced
+     * in HandledTransportAction implementation.
+     *
+     * @param timeout operation timeout (passed through but not enforced by transport layer)
+     * @return this request
+     */
+    public PruneCacheRequest timeout(TimeValue timeout) {
+        this.timeout = timeout;
+        return this;
+    }
+
+    /**
+     * Gets the timeout for this request.
+     * Note: Currently preserved for API compatibility but not actively enforced
+     * in HandledTransportAction implementation.
+     *
+     * @return operation timeout (passed through but not enforced by transport layer)
+     */
+    public TimeValue timeout() {
+        return timeout;
     }
 
     @Override

server/src/main/java/org/opensearch/action/admin/cluster/cache/TransportPruneCacheAction.java

@@ -9,80 +9,47 @@
 package org.opensearch.action.admin.cluster.cache;
 
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.action.support.clustermanager.TransportClusterManagerNodeAction;
-import org.opensearch.cluster.ClusterState;
-import org.opensearch.cluster.block.ClusterBlockException;
-import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
-import org.opensearch.cluster.service.ClusterService;
+import org.opensearch.action.support.HandledTransportAction;
 import org.opensearch.common.Nullable;
 import org.opensearch.common.inject.Inject;
 import org.opensearch.core.action.ActionListener;
-import org.opensearch.core.common.io.stream.StreamInput;
 import org.opensearch.index.store.remote.filecache.FileCache;
+import org.opensearch.tasks.Task;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.TransportService;
 
-import java.io.IOException;
-
 /**
- * Transport action for pruning remote file cache
+ * Transport action for pruning file cache.
+ * Uses HandledTransportAction for direct node execution instead of cluster manager routing.
  *
  * @opensearch.internal
  */
-public class TransportPruneCacheAction extends TransportClusterManagerNodeAction<PruneCacheRequest, PruneCacheResponse> {
+public class TransportPruneCacheAction extends HandledTransportAction<PruneCacheRequest, PruneCacheResponse> {
 
     private final FileCache fileCache;
 
     @Inject
-    public TransportPruneCacheAction(
-        ThreadPool threadPool,
-        TransportService transportService,
-        ClusterService clusterService,
-        ActionFilters actionFilters,
-        IndexNameExpressionResolver indexNameExpressionResolver,
-        @Nullable FileCache fileCache
-    ) {
-        super(
-            PruneCacheAction.NAME,
-            transportService,
-            clusterService,
-            threadPool,
-            actionFilters,
-            PruneCacheRequest::new,
-            indexNameExpressionResolver
-        );
+    public TransportPruneCacheAction(TransportService transportService, ActionFilters actionFilters, @Nullable FileCache fileCache) {
+        super(PruneCacheAction.NAME, transportService, actionFilters, PruneCacheRequest::new, ThreadPool.Names.GENERIC);
         this.fileCache = fileCache;
     }
 
     @Override
-    protected String executor() {
-        return ThreadPool.Names.SAME;
-    }
-
-    @Override
-    protected PruneCacheResponse read(StreamInput in) throws IOException {
-        return new PruneCacheResponse(in);
-    }
-
-    @Override
-    protected void clusterManagerOperation(PruneCacheRequest request, ClusterState state, ActionListener<PruneCacheResponse> listener)
-        throws Exception {
+    protected void doExecute(Task task, PruneCacheRequest request, ActionListener<PruneCacheResponse> listener) {
         try {
             if (fileCache == null) {
-                listener.onResponse(new PruneCacheResponse(true, 0));
+                listener.onResponse(new PruneCacheResponse(true, 0L));
                 return;
             }
 
             long prunedBytes = fileCache.prune();
             listener.onResponse(new PruneCacheResponse(true, prunedBytes));
+        } catch (RuntimeException e) {
+            // Catch runtime exceptions from FileCache operations for better error diagnostics
+            listener.onFailure(e);
         } catch (Exception e) {
+            // Fallback for any other unexpected exceptions
             listener.onFailure(e);
         }
     }
-
-    @Override
-    protected ClusterBlockException checkBlock(PruneCacheRequest request, ClusterState state) {
-        // Prune cache operation doesn't require any cluster blocks
-        return null;
-    }
 }

server/src/main/java/org/opensearch/rest/action/admin/cluster/RestPruneCacheAction.java

@@ -49,7 +49,7 @@ protected RestChannelConsumer prepareRequest(RestRequest request, NodeClient cli
         PruneCacheRequest pruneCacheRequest = new PruneCacheRequest();
 
         // Handle timeout parameter
-        pruneCacheRequest.clusterManagerNodeTimeout(request.paramAsTime("timeout", pruneCacheRequest.clusterManagerNodeTimeout()));
+        pruneCacheRequest.timeout(request.paramAsTime("timeout", pruneCacheRequest.timeout()));
 
         // Delegate to Transport Action with standard response handling
         return channel -> client.execute(PruneCacheAction.INSTANCE, pruneCacheRequest, new RestToXContentListener<>(channel));

server/src/test/java/org/opensearch/action/admin/cluster/cache/PruneCacheRequestResponseTests.java

@@ -28,7 +28,7 @@ public class PruneCacheRequestResponseTests extends OpenSearchTestCase {
      */
     public void testPruneCacheRequestSerialization() throws IOException {
         PruneCacheRequest originalRequest = new PruneCacheRequest();
-        originalRequest.clusterManagerNodeTimeout(TimeValue.timeValueSeconds(30));
+        originalRequest.timeout(TimeValue.timeValueSeconds(30));
 
         // Serialize the request
         BytesStreamOutput out = new BytesStreamOutput();
@@ -39,7 +39,7 @@ public void testPruneCacheRequestSerialization() throws IOException {
         PruneCacheRequest deserializedRequest = new PruneCacheRequest(in);
 
         // Assert that timeout is properly serialized/deserialized
-        assertEquals(originalRequest.clusterManagerNodeTimeout(), deserializedRequest.clusterManagerNodeTimeout());
+        assertEquals(originalRequest.timeout(), deserializedRequest.timeout());
     }
 
     /**
@@ -58,7 +58,7 @@ public void testPruneCacheRequestDefaultTimeout() throws IOException {
         PruneCacheRequest deserializedRequest = new PruneCacheRequest(in);
 
         // Assert that default timeout is maintained
-        assertEquals(originalRequest.clusterManagerNodeTimeout(), deserializedRequest.clusterManagerNodeTimeout());
+        assertEquals(originalRequest.timeout(), deserializedRequest.timeout());
     }
 
     /**

server/src/test/java/org/opensearch/action/admin/cluster/cache/TransportPruneCacheActionTests.java

@@ -9,9 +9,6 @@
 package org.opensearch.action.admin.cluster.cache;
 
 import org.opensearch.action.support.ActionFilters;
-import org.opensearch.cluster.ClusterState;
-import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
-import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.core.action.ActionListener;
 import org.opensearch.index.store.remote.filecache.FileCache;
 import org.opensearch.test.OpenSearchTestCase;
@@ -27,15 +24,13 @@
 import static org.mockito.Mockito.when;
 
 /**
- * Tests for {@link TransportPruneCacheAction}.
+ * Tests for {@link TransportPruneCacheAction} using HandledTransportAction pattern.
  */
 public class TransportPruneCacheActionTests extends OpenSearchTestCase {
 
     private ThreadPool threadPool;
     private TransportService transportService;
-    private ClusterService clusterService;
     private ActionFilters actionFilters;
-    private IndexNameExpressionResolver indexNameExpressionResolver;
     private FileCache fileCache;
     private TransportPruneCacheAction action;
 
@@ -45,19 +40,10 @@ public void setUp() throws Exception {
 
         threadPool = new TestThreadPool("test");
         transportService = mock(TransportService.class);
-        clusterService = mock(ClusterService.class);
         actionFilters = mock(ActionFilters.class);
-        indexNameExpressionResolver = mock(IndexNameExpressionResolver.class);
         fileCache = mock(FileCache.class);
 
-        action = new TransportPruneCacheAction(
-            threadPool,
-            transportService,
-            clusterService,
-            actionFilters,
-            indexNameExpressionResolver,
-            fileCache
-        );
+        action = new TransportPruneCacheAction(transportService, actionFilters, fileCache);
     }
 
     @After
@@ -75,7 +61,6 @@ public void testSuccessfulPruneOperation() throws Exception {
         when(fileCache.prune()).thenReturn(expectedPrunedBytes);
 
         PruneCacheRequest request = new PruneCacheRequest();
-        ClusterState clusterState = mock(ClusterState.class);
 
         ActionListener<PruneCacheResponse> listener = new ActionListener<PruneCacheResponse>() {
             @Override
@@ -90,7 +75,7 @@ public void onFailure(Exception e) {
             }
         };
 
-        action.clusterManagerOperation(request, clusterState, listener);
+        action.doExecute(null, request, listener);
 
         verify(fileCache).prune();
     }
@@ -101,16 +86,12 @@ public void onFailure(Exception e) {
     public void testNullFileCache() throws Exception {
         // Create action with null FileCache
         TransportPruneCacheAction nullCacheAction = new TransportPruneCacheAction(
-            threadPool,
             transportService,
-            clusterService,
             actionFilters,
-            indexNameExpressionResolver,
             null // null FileCache
         );
 
         PruneCacheRequest request = new PruneCacheRequest();
-        ClusterState clusterState = mock(ClusterState.class);
 
         ActionListener<PruneCacheResponse> listener = new ActionListener<PruneCacheResponse>() {
             @Override
@@ -125,7 +106,7 @@ public void onFailure(Exception e) {
             }
         };
 
-        nullCacheAction.clusterManagerOperation(request, clusterState, listener);
+        nullCacheAction.doExecute(null, request, listener);
     }
 
     /**
@@ -137,7 +118,6 @@ public void testPruneException() throws Exception {
         when(fileCache.prune()).thenThrow(expectedException);
 
         PruneCacheRequest request = new PruneCacheRequest();
-        ClusterState clusterState = mock(ClusterState.class);
 
         ActionListener<PruneCacheResponse> listener = new ActionListener<PruneCacheResponse>() {
             @Override
@@ -151,36 +131,18 @@ public void onFailure(Exception e) {
             }
         };
 
-        action.clusterManagerOperation(request, clusterState, listener);
+        action.doExecute(null, request, listener);
 
         verify(fileCache).prune();
     }
 
-    /**
-     * Tests that the executor is correctly set to SAME.
-     */
-    public void testExecutor() {
-        assertEquals("Executor should be SAME for non-blocking operations", ThreadPool.Names.SAME, action.executor());
-    }
-
-    /**
-     * Tests that checkBlock returns null (no blocking).
-     */
-    public void testCheckBlock() {
-        PruneCacheRequest request = new PruneCacheRequest();
-        ClusterState clusterState = mock(ClusterState.class);
-
-        assertNull("Prune cache operation should not be blocked", action.checkBlock(request, clusterState));
-    }
-
     /**
      * Tests response deserialization.
      */
     public void testResponseDeserialization() throws Exception {
-        // This tests the read method implementation
-        // In a real test, you'd create a StreamInput with serialized response data
-        // For this test, we'll just verify the method exists and can be called
-        assertNotNull("Action should have read method", action);
+        // This tests the basic action functionality
+        // For HandledTransportAction, we verify the action instance is created correctly
+        assertNotNull("Action should be created successfully", action);
     }
 
     /**
@@ -193,7 +155,6 @@ public void testDifferentPruneReturnValues() throws Exception {
             when(fileCache.prune()).thenReturn(expectedBytes);
 
             PruneCacheRequest request = new PruneCacheRequest();
-            ClusterState clusterState = mock(ClusterState.class);
 
             ActionListener<PruneCacheResponse> listener = new ActionListener<PruneCacheResponse>() {
                 @Override
@@ -208,7 +169,7 @@ public void onFailure(Exception e) {
                 }
             };
 
-            action.clusterManagerOperation(request, clusterState, listener);
+            action.doExecute(null, request, listener);
         }
     }
 
@@ -225,7 +186,6 @@ public void testAsyncBehavior() throws Exception {
         }).when(fileCache).prune();
 
         PruneCacheRequest request = new PruneCacheRequest();
-        ClusterState clusterState = mock(ClusterState.class);
 
         final boolean[] callbackInvoked = { false };
 
@@ -243,7 +203,7 @@ public void onFailure(Exception e) {
             }
         };
 
-        action.clusterManagerOperation(request, clusterState, listener);
+        action.doExecute(null, request, listener);
 
         assertTrue("Callback should have been invoked", callbackInvoked[0]);
         verify(fileCache).prune();

server/src/test/java/org/opensearch/rest/action/admin/cluster/RestPruneCacheActionTests.java

@@ -82,7 +82,7 @@ public void testTimeoutParameterHandling() throws Exception {
             assertEquals(PruneCacheAction.INSTANCE, actionType);
             assertThat(request, instanceOf(PruneCacheRequest.class));
             PruneCacheRequest pruneCacheRequest = (PruneCacheRequest) request;
-            assertEquals(30000, pruneCacheRequest.clusterManagerNodeTimeout().getMillis());
+            assertEquals(30000, pruneCacheRequest.timeout().getMillis());
             return new PruneCacheResponse(true, 1024);
         });