Bug fixes

Signed-off-by: kh3ra <adityakh3ra@gmail.com>

Author: kh3ra

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

@@ -288,6 +288,10 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.INDEX_DERIVED_SOURCE_SETTING,
                 IndexSettings.INDEX_DERIVED_SOURCE_TRANSLOG_ENABLED_SETTING,
 
+                // Settings for merged segment warmer backpressure
+                IndexSettings.INDEX_MERGED_SEGMENT_WARMER_PRESSURE_ENABLED,
+                IndexSettings.INDEX_MERGED_SEGMENT_WARMER_MAX_CONCURRENT_WARMS_FACTOR,
+
                 // validate that built-in similarities don't get redefined
                 Setting.groupSetting("index.similarity.", (s) -> {
                     Map<String, Settings> groups = s.getAsGroups();

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

@@ -800,16 +800,16 @@ public static IndexMergePolicy fromString(String text) {
         "index.merged_segment_warmer.pressure.enabled",
         true,
         Setting.Property.Dynamic,
-        Setting.Property.NodeScope
+        Setting.Property.IndexScope
     );
 
     public static final Setting<Double> INDEX_MERGED_SEGMENT_WARMER_MAX_CONCURRENT_WARMS_FACTOR = Setting.doubleSetting(
         "index.merged_segment_warmer.max_concurrent_warms_factor",
-        0.5,
+        0.5d,
         0,
         1,
         Setting.Property.Dynamic,
-        Setting.Property.NodeScope
+        Setting.Property.IndexScope
     );
 
     public static final Setting<Boolean> INDEX_DERIVED_SOURCE_SETTING = Setting.boolSetting(
@@ -1121,6 +1121,8 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         derivedSourceEnabled = scopedSettings.get(INDEX_DERIVED_SOURCE_SETTING);
         derivedSourceEnabledForTranslog = scopedSettings.get(INDEX_DERIVED_SOURCE_TRANSLOG_ENABLED_SETTING);
         scopedSettings.addSettingsUpdateConsumer(INDEX_DERIVED_SOURCE_TRANSLOG_ENABLED_SETTING, this::setDerivedSourceEnabledForTranslog);
+        this.mergedSegmentWarmerPressureEnabled = scopedSettings.get(INDEX_MERGED_SEGMENT_WARMER_PRESSURE_ENABLED);
+        this.maxConcurrentMergedSegmentWarmsFactor = scopedSettings.get(INDEX_MERGED_SEGMENT_WARMER_MAX_CONCURRENT_WARMS_FACTOR);
         /* There was unintentional breaking change got introduced with [OpenSearch-6424](https://github.com/opensearch-project/OpenSearch/pull/6424) (version 2.7).
          * For indices created prior version (prior to 2.7) which has IndexSort type, they used to type cast the SortField.Type
          * to higher bytes size like integer to long. This behavior was changed from OpenSearch 2.7 version not to

server/src/main/java/org/opensearch/index/merge/MergedSegmentWarmerPressureService.java

@@ -35,8 +35,8 @@ public class MergedSegmentWarmerPressureService {
 
     public MergedSegmentWarmerPressureService(IndexShard indexShard) {
         this.pressureSettings = new PressureSettings(indexShard);
-        this.throttlePredicates = List.of(new ConcurrencyLimiterPredicate(indexShard, pressureSettings));
         this.logger = Loggers.getLogger(MergedSegmentWarmerPressureService.class, indexShard.shardId());
+        this.throttlePredicates = List.of(new ConcurrencyLimiterPredicate(indexShard, pressureSettings));
     }
 
     public boolean isEnabled() {
@@ -117,7 +117,7 @@ String name() {
         String rejectionMessage(MergedSegmentWarmerStats stats) {
             long maxAllowed = calculateMaxAllowedConcurrentWarms(indexShard.getMaxMergesAllowed());
             return super.rejectionMessage(stats) + String.format(
-                    Locale.ROOT,
+                Locale.ROOT,
                 "\nCurrent ongoing warms: %d, max allowed: %d",
                 stats.getOngoingWarms(),
                 maxAllowed
@@ -128,7 +128,6 @@ String rejectionMessage(MergedSegmentWarmerStats stats) {
         public boolean test(MergedSegmentWarmerStats statsSnapshot) {
             long onGoingWarms = statsSnapshot.getOngoingWarms();
             long maxAllowedWarms = calculateMaxAllowedConcurrentWarms(indexShard.getMaxMergesAllowed());
-
             return maxAllowedWarms > onGoingWarms;
         }
     }

server/src/main/java/org/opensearch/rest/action/cat/RestIndicesAction.java

@@ -664,6 +664,12 @@ protected Table getTableWithHeader(final RestRequest request, final PageToken pa
             "default:false;text-align:right;desc:total wallclock time spent sending merged segments by a primary shard"
         );
 
+        table.addCell(
+            "merged_segment_warmer.total_rejected_warms",
+            "alias:mswtrw,mergedSegmentWarmerTotalRejectedWarms;default:false;text-align:right;desc:UPDATE"
+        );
+        table.addCell("pri.merged_segment_warmer.total_rejected_warms", "default:false;text-align:right;desc:UPDATE");
+
         table.addCell("refresh.total", "sibling:pri;alias:rto,refreshTotal;default:false;text-align:right;desc:total refreshes");
         table.addCell("pri.refresh.total", "default:false;text-align:right;desc:total refreshes");
 

server/src/main/java/org/opensearch/rest/action/cat/RestNodesAction.java

@@ -307,6 +307,10 @@ protected Table getTableWithHeader(final RestRequest request) {
             "merges.merged_segment_warmer.total_send_time",
             "alias:mswtst,mergedSegmentWarmerTotalSendTime;default:false;text-align:right;desc:total wallclock time spent sending merged segments by a primary shard"
         );
+        table.addCell(
+            "merged_segment_warmer.total_rejected_warms",
+            "alias:mswtrw,mergedSegmentWarmerTotalRejectedWarms;default:false;text-align:right;desc:UPDATE"
+        );
 
         table.addCell("refresh.total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total refreshes");
         table.addCell("refresh.time", "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in refreshes");

server/src/main/java/org/opensearch/rest/action/cat/RestShardsAction.java

@@ -247,6 +247,10 @@ protected Table getTableWithHeader(final RestRequest request, final PageToken pa
             "merges.merged_segment_warmer.total_send_time",
             "alias:mswtst,mergedSegmentWarmerTotalSendTime;default:false;text-align:right;desc:total wallclock time spent sending merged segments by a primary shard"
         );
+        table.addCell(
+            "merged_segment_warmer.total_rejected_warms",
+            "alias:mswtrw,mergedSegmentWarmerTotalRejectedWarms;default:false;text-align:right;desc:UPDATE"
+        );
 
         table.addCell("refresh.total", "alias:rto,refreshTotal;default:false;text-align:right;desc:total refreshes");
         table.addCell("refresh.time", "alias:rti,refreshTime;default:false;text-align:right;desc:time spent in refreshes");