Enabled default throttling for all tasks submitted to cluster manager

Signed-off-by: Manik Garg <[email protected]>

Author: gargmanik13

CHANGELOG.md

@@ -9,10 +9,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - [Rule Based Auto-tagging] Add rule schema for auto tagging ([#17238](https://github.com/opensearch-project/OpenSearch/pull/17238))
 - Renaming the node role search to warm ([#17573](https://github.com/opensearch-project/OpenSearch/pull/17573))
 - Introduce a new search node role to hold search only shards ([#17620](https://github.com/opensearch-project/OpenSearch/pull/17620))
-- Fix systemd integTest on deb regarding path ownership check ([#17641](https://github.com/opensearch-project/OpenSearch/pull/17641)) 
+- Fix systemd integTest on deb regarding path ownership check ([#17641](https://github.com/opensearch-project/OpenSearch/pull/17641))
 - Add dfs transformation function in XContentMapValues ([#17612](https://github.com/opensearch-project/OpenSearch/pull/17612))
 - [Security Manager Replacement] Add support of Java policies ([#17663](https://github.com/opensearch-project/OpenSearch/pull/17663))
 - Added Kinesis support as a plugin for the pull-based ingestion ([#17615](https://github.com/opensearch-project/OpenSearch/pull/17615)
+- Enabled default throttling for all tasks submitted to cluster manager ([#17711](https://github.com/opensearch-project/OpenSearch/pull/17711))
 
 ### Changed
 - Migrate BC libs to their FIPS counterparts ([#14912](https://github.com/opensearch-project/OpenSearch/pull/14912))

server/src/main/java/org/opensearch/cluster/service/ClusterManagerTaskThrottler.java

@@ -21,6 +21,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
@@ -31,9 +32,11 @@
  * This class does throttling on task submission to cluster manager node, it uses throttling key defined in various executors
  * as key for throttling. Throttling will be performed over task executor's class level, different task types have different executors class.
  * <p>
- * Set specific setting to for setting the threshold of throttling of particular task type.
+ * Set specific setting for setting the threshold of throttling of a particular task type.
  * e.g : Set "cluster_manager.throttling.thresholds.put_mapping" to set throttling limit of "put mapping" tasks,
- * Set it to default value(-1) to disable the throttling for this task type.
+ * <p>
+ * Set it to (-1) to disable the throttling for this task type.
+ * If no setting is set, default threshold of (50) will be used for throttling when a task is registered with throttling enabled.
  */
 public class ClusterManagerTaskThrottler implements TaskBatcherListener {
     private static final Logger logger = LogManager.getLogger(ClusterManagerTaskThrottler.class);
@@ -67,6 +70,8 @@ public class ClusterManagerTaskThrottler implements TaskBatcherListener {
     protected Map<String, ThrottlingKey> THROTTLING_TASK_KEYS = new ConcurrentHashMap<>();
 
     private final int MIN_THRESHOLD_VALUE = -1; // Disabled throttling
+    // Note: Add different thresholds based on task type if required in the future.
+    private final int DEFAULT_THRESHOLD_VALUE = 50;
     private final ClusterManagerTaskThrottlerListener clusterManagerTaskThrottlerListener;
 
     final ConcurrentMap<String, Long> tasksCount;
@@ -122,13 +127,19 @@ public static TimeValue getMaxDelayForRetry() {
      * Added retry mechanism in TransportClusterManagerNodeAction, so it would be retried for customer generated tasks.
      * <p>
      * If tasks are not getting retried then we can register with false flag, so user won't be able to configure threshold limits for it.
+     * <p>
+     * If throttling is enabled, default threshold of (50) will be used if not specified in settings.
      */
     protected ThrottlingKey registerClusterManagerTask(String taskKey, boolean throttlingEnabled) {
         ThrottlingKey throttlingKey = new ThrottlingKey(taskKey, throttlingEnabled);
         if (THROTTLING_TASK_KEYS.containsKey(taskKey)) {
             throw new IllegalArgumentException("There is already a Throttling key registered with same name: " + taskKey);
         }
         THROTTLING_TASK_KEYS.put(taskKey, throttlingKey);
+
+        if (throttlingEnabled && Objects.isNull(getThrottlingLimit(taskKey))) {
+            tasksThreshold.put(taskKey, (long) DEFAULT_THRESHOLD_VALUE);
+        }
         return throttlingKey;
     }
 
@@ -176,7 +187,7 @@ void validateSetting(final Settings settings) {
             if (!THROTTLING_TASK_KEYS.get(key).isThrottlingEnabled()) {
                 throw new IllegalArgumentException("Throttling is not enabled for given task type: " + key);
             }
-            int threshold = groups.get(key).getAsInt("value", MIN_THRESHOLD_VALUE);
+            int threshold = groups.get(key).getAsInt("value", DEFAULT_THRESHOLD_VALUE);
             if (threshold < MIN_THRESHOLD_VALUE) {
                 throw new IllegalArgumentException("Provide positive integer for limit or -1 for disabling throttling");
             }
@@ -192,7 +203,7 @@ void updateSetting(final Settings newSettings) {
         settingKeys.addAll(tasksThreshold.keySet());
         for (String key : settingKeys) {
             Settings setting = groups.get(key);
-            updateLimit(key, setting == null ? MIN_THRESHOLD_VALUE : setting.getAsInt("value", MIN_THRESHOLD_VALUE));
+            updateLimit(key, setting == null ? DEFAULT_THRESHOLD_VALUE : setting.getAsInt("value", DEFAULT_THRESHOLD_VALUE));
         }
     }
 

server/src/test/java/org/opensearch/cluster/service/ClusterManagerTaskThrottlerTests.java

@@ -34,6 +34,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import static org.opensearch.cluster.service.ClusterManagerTaskThrottler.THRESHOLD_SETTINGS;
 import static org.opensearch.test.ClusterServiceUtils.setState;
 
 /**
@@ -75,8 +76,9 @@ public void testDefaults() {
         }, new ClusterManagerThrottlingStats());
         throttler.registerClusterManagerTask("put-mapping", true);
         throttler.registerClusterManagerTask("create-index", true);
+
         for (String key : throttler.THROTTLING_TASK_KEYS.keySet()) {
-            assertNull(throttler.getThrottlingLimit(key));
+            assertEquals(50, throttler.getThrottlingLimit(key).intValue());
         }
     }
 
@@ -94,11 +96,11 @@ public void testValidateSettingsForDifferentVersion() {
         }, new ClusterManagerThrottlingStats());
         throttler.registerClusterManagerTask("put-mapping", true);
 
-        // set some limit for update snapshot tasks
+        // set some limit for put-mapping tasks
         int newLimit = randomIntBetween(1, 10);
 
         Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", newLimit).build();
-        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(newSettings));
+        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(THRESHOLD_SETTINGS.get(newSettings)));
 
         // validate for empty setting, it shouldn't throw exception
         Settings emptySettings = Settings.builder().build();
@@ -124,11 +126,11 @@ public void testValidateSettingsForTaskWithoutRetryOnDataNode() {
         }, new ClusterManagerThrottlingStats());
         throttler.registerClusterManagerTask("put-mapping", false);
 
-        // set some limit for update snapshot tasks
+        // set some limit for put-mapping tasks
         int newLimit = randomIntBetween(1, 10);
 
         Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", newLimit).build();
-        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(newSettings));
+        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(THRESHOLD_SETTINGS.get(newSettings)));
     }
 
     public void testUpdateSettingsForNullValue() {
@@ -154,7 +156,7 @@ public void testUpdateSettingsForNullValue() {
         // set limit to null
         Settings nullSettings = Settings.builder().build();
         clusterSettings.applySettings(nullSettings);
-        assertNull(throttler.getThrottlingLimit("put-mapping"));
+        assertEquals(50, throttler.getThrottlingLimit("put-mapping").intValue());
     }
 
     public void testSettingsOnBootstrap() {
@@ -222,10 +224,10 @@ public void testValidateSettingsForUnknownTask() {
             return clusterService.getClusterManagerService().getMinNodeVersion();
         }, new ClusterManagerThrottlingStats());
 
-        // set some limit for update snapshot tasks
+        // set some limit for random tasks
         int newLimit = randomIntBetween(1, 10);
         Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.random-task.value", newLimit).build();
-        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(newSettings));
+        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(THRESHOLD_SETTINGS.get(newSettings)));
     }
 
     public void testUpdateThrottlingLimitForBasicSanity() {
@@ -242,17 +244,17 @@ public void testUpdateThrottlingLimitForBasicSanity() {
         }, new ClusterManagerThrottlingStats());
         throttler.registerClusterManagerTask("put-mapping", true);
 
-        // set some limit for update snapshot tasks
+        // set some limit for put-mapping tasks
         long newLimit = randomLongBetween(1, 10);
 
         Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", newLimit).build();
         clusterSettings.applySettings(newSettings);
         assertEquals(newLimit, throttler.getThrottlingLimit("put-mapping").intValue());
 
-        // set update snapshot task limit to default
-        newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", -1).build();
+        // set put-mapping task limit to 20
+        newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", 20).build();
         clusterSettings.applySettings(newSettings);
-        assertNull(throttler.getThrottlingLimit("put-mapping"));
+        assertEquals(20, throttler.getThrottlingLimit("put-mapping").intValue());
     }
 
     public void testValidateSettingForLimit() {
@@ -269,8 +271,8 @@ public void testValidateSettingForLimit() {
         }, new ClusterManagerThrottlingStats());
         throttler.registerClusterManagerTask("put-mapping", true);
 
-        Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.values", -5).build();
-        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(newSettings));
+        Settings newSettings = Settings.builder().put("cluster_manager.throttling.thresholds.put-mapping.value", -5).build();
+        assertThrows(IllegalArgumentException.class, () -> throttler.validateSetting(THRESHOLD_SETTINGS.get(newSettings)));
     }
 
     public void testUpdateLimit() {
@@ -348,7 +350,7 @@ public void testThrottlingForInitialStaticSettingAndVersionCheck() {
         }, throttlingStats);
         ClusterManagerTaskThrottler.ThrottlingKey throttlingKey = throttler.registerClusterManagerTask("put-mapping", true);
 
-        // verifying adding more tasks then threshold passes
+        // verifying adding more tasks than threshold passes
         throttler.onBeginSubmit(getMockUpdateTaskList("put-mapping", throttlingKey, put_mapping_threshold_value + 5));
         assertEquals(0L, throttlingStats.getThrottlingCount("put-mapping"));