Implemented index and cluster force merge rate limiting with index-level precedence

Author: harycash

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -111,6 +111,7 @@
 import org.opensearch.http.HttpTransportSettings;
 import org.opensearch.index.IndexModule;
 import org.opensearch.index.IndexSettings;
+import org.opensearch.index.MergeSchedulerConfig;
 import org.opensearch.index.IndexingPressure;
 import org.opensearch.index.SegmentReplicationPressureService;
 import org.opensearch.index.ShardIndexingPressureMemoryManager;
@@ -514,6 +515,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 IndexSettings.QUERY_STRING_ANALYZE_WILDCARD,
                 IndexSettings.QUERY_STRING_ALLOW_LEADING_WILDCARD,
                 IndexSettings.TIME_SERIES_INDEX_MERGE_POLICY,
+                MergeSchedulerConfig.CLUSTER_MAX_FORCED_MERGE_MB_PER_SEC_SETTING,
                 ScriptService.SCRIPT_GENERAL_CACHE_SIZE_SETTING,
                 ScriptService.SCRIPT_GENERAL_CACHE_EXPIRE_SETTING,
                 ScriptService.SCRIPT_GENERAL_MAX_COMPILATIONS_RATE_SETTING,

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -90,6 +90,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 MergeSchedulerConfig.AUTO_THROTTLE_SETTING,
                 MergeSchedulerConfig.MAX_MERGE_COUNT_SETTING,
                 MergeSchedulerConfig.MAX_THREAD_COUNT_SETTING,
+                MergeSchedulerConfig.MAX_FORCED_MERGE_MB_PER_SEC_SETTING,
                 IndexMetadata.SETTING_INDEX_VERSION_CREATED,
                 IndexMetadata.SETTING_INDEX_CREATION_DATE,
                 IndexMetadata.INDEX_UUID_SETTING,

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -1178,6 +1178,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             mergeSchedulerConfig::setMaxThreadAndMergeCount
         );
         scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.AUTO_THROTTLE_SETTING, mergeSchedulerConfig::setAutoThrottle);
+        scopedSettings.addSettingsUpdateConsumer(MergeSchedulerConfig.MAX_FORCED_MERGE_MB_PER_SEC_SETTING, mergeSchedulerConfig::setMaxForcedMergeMBPerSec);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_DURABILITY_SETTING, this::setTranslogDurability);
         scopedSettings.addSettingsUpdateConsumer(INDEX_TRANSLOG_SYNC_INTERVAL_SETTING, this::setTranslogSyncInterval);
         scopedSettings.addSettingsUpdateConsumer(

server/src/main/java/org/opensearch/index/MergeSchedulerConfig.java

@@ -63,6 +63,28 @@
  *     unluckily suddenly requires a large merge will see that merge aggressively
  *     throttled, while an application doing heavy indexing will see the throttle
  *     move higher to allow merges to keep up with ongoing indexing.
+ *
+ * <li><code>index.merge.scheduler.max_forced_merge_mb_per_sec</code>:
+ * <p>
+ *     Controls the rate limiting for forced merges in MB per second at the index level.
+ *     The default value of Double.POSITIVE_INFINITY means no rate limiting is applied.
+ *     Setting a finite positive value will limit the throughput of forced merge operations
+ *     to the specified rate. This setting takes precedence over the cluster-level setting.
+ *
+ * <li><code>cluster.merge.scheduler.max_forced_merge_mb_per_sec</code>:
+ * <p>
+ *     Controls the rate limiting for forced merges in MB per second at the cluster level.
+ *     The default value of Double.POSITIVE_INFINITY means no rate limiting is applied.
+ *     This setting is used as a fallback when the index-level setting is not specified.
+ *     Index-level settings take precedence over this cluster-level setting.
+ * </ul>
+ *
+ * <p><b>Setting Precedence:</b>
+ * <ul>
+ * <li>If only index-level setting is specified: uses index value
+ * <li>If only cluster-level setting is specified: uses cluster value  
+ * <li>If both settings are specified: uses index value (index takes precedence)
+ * <li>If neither setting is specified: uses default value (Double.POSITIVE_INFINITY)
  * </ul>
  *
  * @opensearch.api
@@ -90,16 +112,41 @@ public final class MergeSchedulerConfig {
         Property.Dynamic,
         Property.IndexScope
     );
+    public static final Setting<Double> MAX_FORCED_MERGE_MB_PER_SEC_SETTING = Setting.doubleSetting(
+        "index.merge.scheduler.max_forced_merge_mb_per_sec",
+        Double.POSITIVE_INFINITY,
+        0.0d,
+        Double.POSITIVE_INFINITY,
+        Property.Dynamic,
+        Property.IndexScope
+    );
+    
+    public static final Setting<Double> CLUSTER_MAX_FORCED_MERGE_MB_PER_SEC_SETTING = Setting.doubleSetting(
+        "cluster.merge.scheduler.max_forced_merge_mb_per_sec",
+        Double.POSITIVE_INFINITY,
+        0.0d,
+        Double.POSITIVE_INFINITY,
+        Property.Dynamic,
+        Property.NodeScope
+    );
 
     private volatile boolean autoThrottle;
     private volatile int maxThreadCount;
     private volatile int maxMergeCount;
+    private volatile double maxForcedMergeMBPerSec;
 
     MergeSchedulerConfig(IndexSettings indexSettings) {
         int maxThread = indexSettings.getValue(MAX_THREAD_COUNT_SETTING);
         int maxMerge = indexSettings.getValue(MAX_MERGE_COUNT_SETTING);
         setMaxThreadAndMergeCount(maxThread, maxMerge);
         this.autoThrottle = indexSettings.getValue(AUTO_THROTTLE_SETTING);
+        
+        // Index setting takes precedence over cluster setting
+        if (MAX_FORCED_MERGE_MB_PER_SEC_SETTING.exists(indexSettings.getSettings())) {
+            this.maxForcedMergeMBPerSec = indexSettings.getValue(MAX_FORCED_MERGE_MB_PER_SEC_SETTING);
+        } else {
+            this.maxForcedMergeMBPerSec = CLUSTER_MAX_FORCED_MERGE_MB_PER_SEC_SETTING.get(indexSettings.getNodeSettings());
+        }
     }
 
     /**
@@ -151,4 +198,20 @@ void setMaxThreadAndMergeCount(int maxThreadCount, int maxMergeCount) {
     public int getMaxMergeCount() {
         return maxMergeCount;
     }
+
+    /**
+     * Returns the maximum forced merge rate in MB per second.
+     * A value of Double.POSITIVE_INFINITY indicates no rate limiting.
+     */
+    public double getMaxForcedMergeMBPerSec() {
+        return maxForcedMergeMBPerSec;
+    }
+
+    /**
+     * Sets the maximum forced merge rate in MB per second.
+     * A value of Double.POSITIVE_INFINITY disables rate limiting.
+     */
+    void setMaxForcedMergeMBPerSec(double maxForcedMergeMBPerSec) {
+        this.maxForcedMergeMBPerSec = maxForcedMergeMBPerSec;
+    }
 }

server/src/main/java/org/opensearch/index/engine/OpenSearchConcurrentMergeScheduler.java

@@ -224,6 +224,23 @@ void refreshConfig() {
         } else if (config.isAutoThrottle() == false && isEnabled) {
             disableAutoIOThrottle();
         }
+        
+        // Apply forced merge rate limiting
+        applyMergeRateLimit();
+    }
+
+    /**
+     * Applies the merge rate limit based on the current configuration.
+     * If the setting value is Double.POSITIVE_INFINITY, rate limiting is disabled.
+     * Otherwise, auto-throttling is enabled as the available rate limiting mechanism.
+     */
+    private void applyMergeRateLimit() {
+        double maxForcedMergeMBPerSec = config.getMaxForcedMergeMBPerSec();
+        if (maxForcedMergeMBPerSec != super.getForceMergeMBPerSec()) {
+            logger.info("[{}][{}] updating forced merge rate limit from [{}] to [{}] MB/sec", 
+                shardId.getIndexName(), shardId.id(), super.getForceMergeMBPerSec(), maxForcedMergeMBPerSec);
+            super.setForceMergeMBPerSec(maxForcedMergeMBPerSec);
+        }
     }
 
 }