Enabled default throttling for all tasks submitted to cluster manager (opensearch-project#17711)

Signed-off-by: Manik Garg <gargmanik1317@gmail.com>
(cherry picked from commit 6ce0628)

Author: gargmanik13

CHANGELOG.md

@@ -7,6 +7,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Added
 - Add multi-threaded writer support in pull-based ingestion ([#17912](https://github.com/opensearch-project/OpenSearch/pull/17912))
 - Unset discovery nodes for every transport node actions request ([#17682](https://github.com/opensearch-project/OpenSearch/pull/17682))
+- Enabled default throttling for all tasks submitted to cluster manager ([#17711](https://github.com/opensearch-project/OpenSearch/pull/17711))
 
 ### Changed
 - Change the default max header size from 8KB to 16KB. ([#18024](https://github.com/opensearch-project/OpenSearch/pull/18024))

plugins/workload-management/src/main/java/org/opensearch/plugin/wlm/service/WorkloadGroupPersistenceService.java

@@ -41,15 +41,15 @@
 import java.util.stream.Collectors;
 
 import static org.opensearch.cluster.metadata.WorkloadGroup.updateExistingWorkloadGroup;
+import static org.opensearch.cluster.service.ClusterManagerTask.CREATE_QUERY_GROUP;
+import static org.opensearch.cluster.service.ClusterManagerTask.DELETE_QUERY_GROUP;
+import static org.opensearch.cluster.service.ClusterManagerTask.UPDATE_QUERY_GROUP;
 
 /**
  * This class defines the functions for WorkloadGroup persistence
  */
 public class WorkloadGroupPersistenceService {
     static final String SOURCE = "query-group-persistence-service";
-    private static final String CREATE_QUERY_GROUP_THROTTLING_KEY = "create-query-group";
-    private static final String DELETE_QUERY_GROUP_THROTTLING_KEY = "delete-query-group";
-    private static final String UPDATE_QUERY_GROUP_THROTTLING_KEY = "update-query-group";
     private static final Logger logger = LogManager.getLogger(WorkloadGroupPersistenceService.class);
     /**
      *  max WorkloadGroup count setting name
@@ -94,9 +94,9 @@ public WorkloadGroupPersistenceService(
         final ClusterSettings clusterSettings
     ) {
         this.clusterService = clusterService;
-        this.createWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(CREATE_QUERY_GROUP_THROTTLING_KEY, true);
-        this.deleteWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(DELETE_QUERY_GROUP_THROTTLING_KEY, true);
-        this.updateWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(UPDATE_QUERY_GROUP_THROTTLING_KEY, true);
+        this.createWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(CREATE_QUERY_GROUP, true);
+        this.deleteWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(DELETE_QUERY_GROUP, true);
+        this.updateWorkloadGroupThrottlingKey = clusterService.registerClusterManagerTask(UPDATE_QUERY_GROUP, true);
         setMaxWorkloadGroupCount(MAX_QUERY_GROUP_COUNT.get(settings));
         clusterSettings.addSettingsUpdateConsumer(MAX_QUERY_GROUP_COUNT, this::setMaxWorkloadGroupCount);
     }

server/src/main/java/org/opensearch/action/admin/cluster/reroute/TransportClusterRerouteAction.java

@@ -53,7 +53,6 @@
 import org.opensearch.cluster.routing.allocation.command.AbstractAllocateAllocationCommand;
 import org.opensearch.cluster.routing.allocation.command.AllocateStalePrimaryAllocationCommand;
 import org.opensearch.cluster.routing.allocation.command.AllocationCommand;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -70,6 +69,8 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.CLUSTER_REROUTE_API;
+
 /**
  * Transport action for rerouting cluster allocation commands
  *
@@ -102,7 +103,7 @@ public TransportClusterRerouteAction(
         );
         this.allocationService = allocationService;
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        clusterRerouteTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CLUSTER_REROUTE_API_KEY, true);
+        clusterRerouteTaskKey = clusterService.registerClusterManagerTask(CLUSTER_REROUTE_API, true);
     }
 
     @Override

server/src/main/java/org/opensearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java

@@ -47,7 +47,6 @@
 import org.opensearch.cluster.node.DiscoveryNode;
 import org.opensearch.cluster.node.DiscoveryNodes;
 import org.opensearch.cluster.routing.allocation.AllocationService;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Nullable;
@@ -65,6 +64,7 @@
 import java.io.IOException;
 
 import static org.opensearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider.CLUSTER_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING;
+import static org.opensearch.cluster.service.ClusterManagerTask.CLUSTER_UPDATE_SETTINGS;
 import static org.opensearch.index.remote.RemoteStoreUtils.checkAndFinalizeRemoteStoreMigration;
 
 /**
@@ -108,7 +108,7 @@ public TransportClusterUpdateSettingsAction(
         this.clusterSettings = clusterSettings;
 
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        clusterUpdateSettingTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CLUSTER_UPDATE_SETTINGS_KEY, true);
+        clusterUpdateSettingTaskKey = clusterService.registerClusterManagerTask(CLUSTER_UPDATE_SETTINGS, true);
 
     }
 

server/src/main/java/org/opensearch/action/admin/cluster/storedscripts/TransportDeleteStoredScriptAction.java

@@ -39,7 +39,6 @@
 import org.opensearch.cluster.block.ClusterBlockException;
 import org.opensearch.cluster.block.ClusterBlockLevel;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
@@ -51,6 +50,8 @@
 
 import java.io.IOException;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.DELETE_SCRIPT;
+
 /**
  * Transport action for deleting stored script
  *
@@ -81,7 +82,7 @@ public TransportDeleteStoredScriptAction(
         );
         this.scriptService = scriptService;
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        deleteScriptTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.DELETE_SCRIPT_KEY, true);
+        deleteScriptTaskKey = clusterService.registerClusterManagerTask(DELETE_SCRIPT, true);
     }
 
     @Override

server/src/main/java/org/opensearch/action/admin/cluster/storedscripts/TransportPutStoredScriptAction.java

@@ -39,7 +39,6 @@
 import org.opensearch.cluster.block.ClusterBlockException;
 import org.opensearch.cluster.block.ClusterBlockLevel;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
@@ -51,6 +50,8 @@
 
 import java.io.IOException;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.PUT_SCRIPT;
+
 /**
  * Transport action for putting stored script
  *
@@ -81,7 +82,7 @@ public TransportPutStoredScriptAction(
         );
         this.scriptService = scriptService;
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        putScriptTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.PUT_SCRIPT_KEY, true);
+        putScriptTaskKey = clusterService.registerClusterManagerTask(PUT_SCRIPT, true);
     }
 
     @Override

server/src/main/java/org/opensearch/action/admin/indices/create/AutoCreateAction.java

@@ -49,7 +49,6 @@
 import org.opensearch.cluster.metadata.MetadataCreateDataStreamService.CreateDataStreamClusterStateUpdateRequest;
 import org.opensearch.cluster.metadata.MetadataCreateIndexService;
 import org.opensearch.cluster.metadata.MetadataIndexTemplateService;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -62,6 +61,8 @@
 import java.io.IOException;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.AUTO_CREATE;
+
 /**
  * Api that auto creates an index or data stream that originate from requests that write into an index that doesn't yet exist.
  *
@@ -104,7 +105,7 @@ public TransportAction(
             this.metadataCreateDataStreamService = metadataCreateDataStreamService;
 
             // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-            autoCreateTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.AUTO_CREATE_KEY, true);
+            autoCreateTaskKey = clusterService.registerClusterManagerTask(AUTO_CREATE, true);
         }
 
         @Override

server/src/main/java/org/opensearch/action/admin/indices/dangling/delete/TransportDeleteDanglingIndexAction.java

@@ -51,7 +51,6 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.metadata.Metadata;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.inject.Inject;
@@ -67,6 +66,8 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.DELETE_DANGLING_INDEX;
+
 /**
  * Implements the deletion of a dangling index. When handling a {@link DeleteDanglingIndexAction},
  * this class first checks that such a dangling index exists. It then submits a cluster state update
@@ -105,7 +106,7 @@ public TransportDeleteDanglingIndexAction(
         this.settings = settings;
         this.nodeClient = nodeClient;
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        deleteDanglingIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.DELETE_DANGLING_INDEX_KEY, true);
+        deleteDanglingIndexTaskKey = clusterService.registerClusterManagerTask(DELETE_DANGLING_INDEX, true);
     }
 
     @Override

server/src/main/java/org/opensearch/action/admin/indices/datastream/DeleteDataStreamAction.java

@@ -49,7 +49,6 @@
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.metadata.Metadata;
 import org.opensearch.cluster.metadata.MetadataDeleteIndexService;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -75,6 +74,7 @@
 import java.util.Set;
 
 import static org.opensearch.action.ValidateActions.addValidationError;
+import static org.opensearch.cluster.service.ClusterManagerTask.REMOVE_DATA_STREAM;
 
 /**
  * Transport action for deleting a datastream
@@ -186,7 +186,7 @@ public TransportAction(
             super(NAME, transportService, clusterService, threadPool, actionFilters, Request::new, indexNameExpressionResolver);
             this.deleteIndexService = deleteIndexService;
             // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-            removeDataStreamTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.REMOVE_DATA_STREAM_KEY, true);
+            removeDataStreamTaskKey = clusterService.registerClusterManagerTask(REMOVE_DATA_STREAM, true);
         }
 
         @Override

server/src/main/java/org/opensearch/action/admin/indices/rollover/TransportRolloverAction.java

@@ -47,7 +47,6 @@
 import org.opensearch.cluster.metadata.IndexMetadata;
 import org.opensearch.cluster.metadata.IndexNameExpressionResolver;
 import org.opensearch.cluster.metadata.Metadata;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Nullable;
@@ -71,6 +70,8 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.ROLLOVER_INDEX;
+
 /**
  * Main class to swap the index pointed to by an alias, given some conditions
  *
@@ -106,7 +107,7 @@ public TransportRolloverAction(
         this.client = client;
         this.activeShardsObserver = new ActiveShardsObserver(clusterService, threadPool);
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        rolloverIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.ROLLOVER_INDEX_KEY, true);
+        rolloverIndexTaskKey = clusterService.registerClusterManagerTask(ROLLOVER_INDEX, true);
     }
 
     @Override

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateDataStreamService.java

@@ -43,7 +43,6 @@
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.ack.ClusterStateUpdateRequest;
 import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -63,6 +62,8 @@
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.CREATE_DATA_STREAM;
+
 /**
  * Creates a data stream of metadata
  *
@@ -86,7 +87,7 @@ public MetadataCreateDataStreamService(
         this.activeShardsObserver = new ActiveShardsObserver(clusterService, threadPool);
         this.metadataCreateIndexService = metadataCreateIndexService;
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        createDataStreamTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CREATE_DATA_STREAM_KEY, true);
+        createDataStreamTaskKey = clusterService.registerClusterManagerTask(CREATE_DATA_STREAM, true);
     }
 
     public void createDataStream(CreateDataStreamClusterStateUpdateRequest request, ActionListener<AcknowledgedResponse> finalListener) {

server/src/main/java/org/opensearch/cluster/metadata/MetadataCreateIndexService.java

@@ -61,7 +61,6 @@
 import org.opensearch.cluster.routing.ShardRoutingState;
 import org.opensearch.cluster.routing.allocation.AllocationService;
 import org.opensearch.cluster.routing.allocation.AwarenessReplicaBalance;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Nullable;
@@ -156,6 +155,7 @@
 import static org.opensearch.cluster.metadata.Metadata.DEFAULT_REPLICA_COUNT_SETTING;
 import static org.opensearch.cluster.metadata.MetadataIndexTemplateService.findContextTemplateName;
 import static org.opensearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider.INDEX_TOTAL_PRIMARY_SHARDS_PER_NODE_SETTING;
+import static org.opensearch.cluster.service.ClusterManagerTask.CREATE_INDEX;
 import static org.opensearch.index.IndexModule.INDEX_STORE_TYPE_SETTING;
 import static org.opensearch.index.IndexSettings.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE_SETTING;
 import static org.opensearch.indices.IndicesService.CLUSTER_REPLICATION_TYPE_SETTING;
@@ -225,7 +225,7 @@ public MetadataCreateIndexService(
         this.awarenessReplicaBalance = awarenessReplicaBalance;
 
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        createIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.CREATE_INDEX_KEY, true);
+        createIndexTaskKey = clusterService.registerClusterManagerTask(CREATE_INDEX, true);
         Supplier<Version> minNodeVersionSupplier = () -> clusterService.state().nodes().getMinNodeVersion();
         remoteStoreCustomMetadataResolver = isRemoteDataAttributePresent(settings)
             ? new RemoteStoreCustomMetadataResolver(remoteStoreSettings, minNodeVersionSupplier, repositoriesServiceSupplier, settings)

server/src/main/java/org/opensearch/cluster/metadata/MetadataDeleteIndexService.java

@@ -42,7 +42,6 @@
 import org.opensearch.cluster.block.ClusterBlocks;
 import org.opensearch.cluster.routing.RoutingTable;
 import org.opensearch.cluster.routing.allocation.AllocationService;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -62,6 +61,8 @@
 import java.util.Map;
 import java.util.Set;
 
+import static org.opensearch.cluster.service.ClusterManagerTask.DELETE_INDEX;
+
 /**
  * Deletes indices.
  *
@@ -84,7 +85,7 @@ public MetadataDeleteIndexService(Settings settings, ClusterService clusterServi
         this.allocationService = allocationService;
 
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        deleteIndexTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.DELETE_INDEX_KEY, true);
+        deleteIndexTaskKey = clusterService.registerClusterManagerTask(DELETE_INDEX, true);
 
     }
 

server/src/main/java/org/opensearch/cluster/metadata/MetadataIndexAliasesService.java

@@ -38,7 +38,6 @@
 import org.opensearch.cluster.ClusterState;
 import org.opensearch.cluster.ack.ClusterStateUpdateResponse;
 import org.opensearch.cluster.metadata.AliasAction.NewAliasValidator;
-import org.opensearch.cluster.service.ClusterManagerTaskKeys;
 import org.opensearch.cluster.service.ClusterManagerTaskThrottler;
 import org.opensearch.cluster.service.ClusterService;
 import org.opensearch.common.Priority;
@@ -62,6 +61,7 @@
 import java.util.function.Function;
 
 import static java.util.Collections.emptyList;
+import static org.opensearch.cluster.service.ClusterManagerTask.INDEX_ALIASES;
 import static org.opensearch.indices.cluster.IndicesClusterStateService.AllocatedIndices.IndexRemovalReason.NO_LONGER_ASSIGNED;
 
 /**
@@ -97,7 +97,7 @@ public MetadataIndexAliasesService(
         this.xContentRegistry = xContentRegistry;
 
         // Task is onboarded for throttling, it will get retried from associated TransportClusterManagerNodeAction.
-        indexAliasTaskKey = clusterService.registerClusterManagerTask(ClusterManagerTaskKeys.INDEX_ALIASES_KEY, true);
+        indexAliasTaskKey = clusterService.registerClusterManagerTask(INDEX_ALIASES, true);
 
     }