From 30851ffbbfdf97f89cf861fa76c8b9e603f8447e Mon Sep 17 00:00:00 2001 From: Rajiv Kumar Vaidyanathan Date: Thu, 13 Jun 2024 20:20:26 +0530 Subject: [PATCH] enabling term version check on local state for all admin read actions Signed-off-by: Rajiv Kumar Vaidyanathan --- .../opensearch/action/IndicesRequestIT.java | 35 +++++++++++- .../AdmissionForClusterManagerIT.java | 26 ++++++++- .../health/TransportClusterHealthAction.java | 5 ++ .../state/TransportClusterStateAction.java | 5 -- .../TransportPendingClusterTasksAction.java | 5 ++ .../TransportClusterManagerNodeAction.java | 56 +++++++++++-------- ...TransportClusterManagerNodeReadAction.java | 5 ++ 7 files changed, 106 insertions(+), 31 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/action/IndicesRequestIT.java b/server/src/internalClusterTest/java/org/opensearch/action/IndicesRequestIT.java index 84d833569edcb..fb0bfb7ba27ce 100644 --- a/server/src/internalClusterTest/java/org/opensearch/action/IndicesRequestIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/action/IndicesRequestIT.java @@ -84,6 +84,8 @@ import org.opensearch.action.search.SearchResponse; import org.opensearch.action.search.SearchTransportService; import org.opensearch.action.search.SearchType; +import org.opensearch.action.support.clustermanager.term.GetTermVersionAction; +import org.opensearch.action.support.clustermanager.term.GetTermVersionResponse; import org.opensearch.action.support.replication.TransportReplicationActionTests; import org.opensearch.action.termvectors.MultiTermVectorsAction; import org.opensearch.action.termvectors.MultiTermVectorsRequest; @@ -93,6 +95,8 @@ import org.opensearch.action.update.UpdateRequest; import org.opensearch.action.update.UpdateResponse; import org.opensearch.client.Requests; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.coordination.ClusterStateTermVersion; import org.opensearch.cluster.metadata.IndexNameExpressionResolver; import org.opensearch.common.settings.Settings; import org.opensearch.common.util.concurrent.ThreadContext; @@ -124,6 +128,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Function; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; @@ -195,6 +200,7 @@ public void cleanUp() { } public void testGetFieldMappings() { + String getFieldMappingsShardAction = GetFieldMappingsAction.NAME + "[index][s]"; interceptTransportActions(getFieldMappingsShardAction); @@ -546,6 +552,7 @@ public void testDeleteIndex() { public void testGetMappings() { interceptTransportActions(GetMappingsAction.NAME); + stubClusterTermResponse(internalCluster().getClusterManagerName()); GetMappingsRequest getMappingsRequest = new GetMappingsRequest().indices(randomIndicesOrAliases()); internalCluster().coordOnlyNodeClient().admin().indices().getMappings(getMappingsRequest).actionGet(); @@ -565,8 +572,9 @@ public void testPutMapping() { } public void testGetSettings() { - interceptTransportActions(GetSettingsAction.NAME); + interceptTransportActions(GetSettingsAction.NAME); + stubClusterTermResponse(internalCluster().getClusterManagerName()); GetSettingsRequest getSettingsRequest = new GetSettingsRequest().indices(randomIndicesOrAliases()); internalCluster().coordOnlyNodeClient().admin().indices().getSettings(getSettingsRequest).actionGet(); @@ -781,6 +789,7 @@ public List getTransportInterceptors( } private final Set actions = new HashSet<>(); + private final Map stubHandlers = new ConcurrentHashMap<>(); private final Map> requests = new HashMap<>(); @@ -804,6 +813,11 @@ synchronized void interceptTransportActions(String... actions) { synchronized void clearInterceptedActions() { actions.clear(); + stubHandlers.clear(); + } + + synchronized void stub(String action, TransportRequestHandler handler) { + stubHandlers.put(action, handler); } private class InterceptingRequestHandler implements TransportRequestHandler { @@ -830,8 +844,25 @@ public void messageReceived(T request, TransportChannel channel, Task task) thro } } } - requestHandler.messageReceived(request, channel, task); + if (!stubHandlers.containsKey(action)) { + requestHandler.messageReceived(request, channel, task); + } else { + stubHandlers.get(action).messageReceived(request, channel, task); + } + } } } + + private void stubClusterTermResponse(String master) { + PluginsService pluginsService = internalCluster().getInstance(PluginsService.class, master); + pluginsService.filterPlugins(InterceptingTransportService.TestPlugin.class).stream().findFirst().get().instance.stub( + GetTermVersionAction.NAME, + (request, channel, task) -> channel.sendResponse( + new GetTermVersionResponse(new ClusterStateTermVersion(new ClusterName("test"), "1", -1, -1)) + ) + ); + + } + } diff --git a/server/src/internalClusterTest/java/org/opensearch/ratelimitting/admissioncontrol/AdmissionForClusterManagerIT.java b/server/src/internalClusterTest/java/org/opensearch/ratelimitting/admissioncontrol/AdmissionForClusterManagerIT.java index b9da5ffb86af0..21d89771e5fa6 100644 --- a/server/src/internalClusterTest/java/org/opensearch/ratelimitting/admissioncontrol/AdmissionForClusterManagerIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/ratelimitting/admissioncontrol/AdmissionForClusterManagerIT.java @@ -12,7 +12,11 @@ import org.apache.logging.log4j.Logger; import org.opensearch.action.admin.indices.alias.get.GetAliasesRequest; import org.opensearch.action.admin.indices.alias.get.GetAliasesResponse; +import org.opensearch.action.support.clustermanager.term.GetTermVersionAction; +import org.opensearch.action.support.clustermanager.term.GetTermVersionResponse; import org.opensearch.client.node.NodeClient; +import org.opensearch.cluster.ClusterName; +import org.opensearch.cluster.coordination.ClusterStateTermVersion; import org.opensearch.common.settings.Settings; import org.opensearch.common.unit.TimeValue; import org.opensearch.core.concurrency.OpenSearchRejectedExecutionException; @@ -20,6 +24,7 @@ import org.opensearch.node.IoUsageStats; import org.opensearch.node.ResourceUsageCollectorService; import org.opensearch.node.resource.tracker.ResourceTrackerSettings; +import org.opensearch.plugins.Plugin; import org.opensearch.ratelimitting.admissioncontrol.controllers.CpuBasedAdmissionController; import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlActionType; import org.opensearch.ratelimitting.admissioncontrol.enums.AdmissionControlMode; @@ -29,9 +34,13 @@ import org.opensearch.rest.action.admin.indices.RestGetAliasesAction; import org.opensearch.test.OpenSearchIntegTestCase; import org.opensearch.test.rest.FakeRestRequest; +import org.opensearch.test.transport.MockTransportService; +import org.opensearch.transport.TransportService; import org.junit.Before; +import java.util.Collection; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicReference; @@ -62,6 +71,10 @@ public class AdmissionForClusterManagerIT extends OpenSearchIntegTestCase { .put(CLUSTER_ADMIN_CPU_USAGE_LIMIT.getKey(), 50) .build(); + protected Collection> nodePlugins() { + return List.of(MockTransportService.TestPlugin.class); + } + @Before public void init() { String clusterManagerNode = internalCluster().startClusterManagerOnlyNode( @@ -82,12 +95,14 @@ public void init() { } public void testAdmissionControlEnforced() throws Exception { + stubClusterTermResponse(internalCluster().getClusterManagerName()); cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 99, new IoUsageStats(98)); // Write API on ClusterManager assertAcked(prepareCreate("test").setMapping("field", "type=text").setAliases("{\"alias1\" : {}}")); - + stubClusterTermResponse(internalCluster().getClusterManagerName()); // Read API on ClusterManager + GetAliasesRequest aliasesRequest = new GetAliasesRequest(); aliasesRequest.aliases("alias1"); try { @@ -156,6 +171,7 @@ public void admissionControlDisabledOnBreach(Settings admission) throws Interrup } public void testAdmissionControlResponseStatus() throws Exception { + stubClusterTermResponse(internalCluster().getClusterManagerName()); cMResourceCollector.collectNodeResourceUsageStats(clusterManagerNodeId, System.currentTimeMillis(), 97, 99, new IoUsageStats(98)); // Write API on ClusterManager @@ -195,4 +211,12 @@ Map getAdmissionControlStats(AdmissionControlS } return acStats; } + + private void stubClusterTermResponse(String master) { + MockTransportService primaryService = (MockTransportService) internalCluster().getInstance(TransportService.class, master); + primaryService.addRequestHandlingBehavior(GetTermVersionAction.NAME, (handler, request, channel, task) -> { + channel.sendResponse(new GetTermVersionResponse(new ClusterStateTermVersion(new ClusterName("test"), "1", -1, -1))); + }); + } + } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java index 1cc357a4c20f4..f69f462372888 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/health/TransportClusterHealthAction.java @@ -534,4 +534,9 @@ private ClusterHealthResponse clusterHealth( pendingTaskTimeInQueue ); } + + @Override + protected boolean localExecuteSupportedByAction() { + return false; + } } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/state/TransportClusterStateAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/state/TransportClusterStateAction.java index cae465a90446e..cdb00b584c5dd 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/state/TransportClusterStateAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/state/TransportClusterStateAction.java @@ -233,9 +233,4 @@ private ClusterStateResponse buildResponse(final ClusterStateRequest request, fi return new ClusterStateResponse(currentState.getClusterName(), builder.build(), false); } - - @Override - protected boolean localExecuteSupportedByAction() { - return true; - } } diff --git a/server/src/main/java/org/opensearch/action/admin/cluster/tasks/TransportPendingClusterTasksAction.java b/server/src/main/java/org/opensearch/action/admin/cluster/tasks/TransportPendingClusterTasksAction.java index 5d5053cc80738..01846ef46c1ed 100644 --- a/server/src/main/java/org/opensearch/action/admin/cluster/tasks/TransportPendingClusterTasksAction.java +++ b/server/src/main/java/org/opensearch/action/admin/cluster/tasks/TransportPendingClusterTasksAction.java @@ -110,4 +110,9 @@ protected void clusterManagerOperation( logger.trace("done fetching pending tasks from cluster service"); listener.onResponse(new PendingClusterTasksResponse(pendingTasks)); } + + @Override + protected boolean localExecuteSupportedByAction() { + return false; + } } diff --git a/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java b/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java index 080b0d607e991..e33ec7673d660 100644 --- a/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java +++ b/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeAction.java @@ -267,24 +267,7 @@ protected void doStart(ClusterState clusterState) { final DiscoveryNodes nodes = clusterState.nodes(); if (nodes.isLocalNodeElectedClusterManager() || localExecute(request)) { // check for block, if blocked, retry, else, execute locally - final ClusterBlockException blockException = checkBlock(request, clusterState); - if (blockException != null) { - if (!blockException.retryable()) { - listener.onFailure(blockException); - } else { - logger.debug("can't execute due to a cluster block, retrying", blockException); - retry(clusterState, blockException, newState -> { - try { - ClusterBlockException newException = checkBlock(request, newState); - return (newException == null || !newException.retryable()); - } catch (Exception e) { - // accept state as block will be rechecked by doStart() and listener.onFailure() then called - logger.trace("exception occurred during cluster block checking, accepting state", e); - return true; - } - }); - } - } else { + if (!checkForBlock(request, clusterState)) { threadPool.executor(executor) .execute( ActionRunnable.wrap( @@ -422,12 +405,39 @@ public GetTermVersionResponse read(StreamInput in) throws IOException { }; } + private boolean checkForBlock(Request request, ClusterState localClusterState) { + final ClusterBlockException blockException = checkBlock(request, localClusterState); + if (blockException != null) { + if (!blockException.retryable()) { + listener.onFailure(blockException); + } else { + logger.debug("can't execute due to a cluster block, retrying", blockException); + retry(localClusterState, blockException, newState -> { + try { + ClusterBlockException newException = checkBlock(request, newState); + return (newException == null || !newException.retryable()); + } catch (Exception e) { + // accept state as block will be rechecked by doStart() and listener.onFailure() then called + logger.trace("exception occurred during cluster block checking, accepting state", e); + return true; + } + }); + } + return true; + } else { + return false; + } + } + private void executeOnLocalNode(ClusterState localClusterState) { - Runnable runTask = ActionRunnable.wrap( - getDelegateForLocalExecute(localClusterState), - l -> clusterManagerOperation(task, request, localClusterState, l) - ); - threadPool.executor(executor).execute(runTask); + // check for block, if blocked, retry, else, execute locally + if (!checkForBlock(request, localClusterState)) { + Runnable runTask = ActionRunnable.wrap( + getDelegateForLocalExecute(localClusterState), + l -> clusterManagerOperation(task, request, localClusterState, l) + ); + threadPool.executor(executor).execute(runTask); + } } private void executeOnClusterManager(DiscoveryNode clusterManagerNode, ClusterState clusterState) { diff --git a/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeReadAction.java b/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeReadAction.java index d58487a475bcf..98b18c79a3c8d 100644 --- a/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeReadAction.java +++ b/server/src/main/java/org/opensearch/action/support/clustermanager/TransportClusterManagerNodeReadAction.java @@ -124,4 +124,9 @@ protected TransportClusterManagerNodeReadAction( protected final boolean localExecute(Request request) { return request.local(); } + + protected boolean localExecuteSupportedByAction() { + return true; + } + }