Uses ClusterSettings instead of Node Settings in HealthMetadataService (#96843)

When building the first version of `HealthMetadata` object, we were
using the `Settings` object, which has the Node's settings, what does
not seem to be propagated to the Node, hence we always used the default
values of the settings. This made that every time a new master was
selected, the initial `HealthMetadata` was built with the default values
instead of the settings configured by the customer.

Author: Pablo Alcantar Morales

docs/changelog/96843.yaml

@@ -0,0 +1,6 @@
+pr: 96843
+summary: Uses `ClusterSettings` instead of Node `Settings` in `HealthMetadataService`
+area: Health
+type: bug
+issues:
+ - 96219

server/src/internalClusterTest/java/org/elasticsearch/health/HealthMetadataServiceIT.java

@@ -10,7 +10,6 @@
 
 import org.elasticsearch.action.admin.cluster.settings.ClusterUpdateSettingsRequest;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
-import org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.health.metadata.HealthMetadata;
@@ -22,12 +21,15 @@
 import java.util.Map;
 import java.util.Set;
 
+import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_MAX_HEADROOM_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_WATERMARK_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_LOW_DISK_MAX_HEADROOM_SETTING;
 import static org.elasticsearch.cluster.routing.allocation.DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_LOW_DISK_WATERMARK_SETTING;
+import static org.elasticsearch.common.settings.Settings.EMPTY;
 import static org.elasticsearch.indices.ShardLimitValidator.SETTING_CLUSTER_MAX_SHARDS_PER_NODE;
 import static org.elasticsearch.indices.ShardLimitValidator.SETTING_CLUSTER_MAX_SHARDS_PER_NODE_FROZEN;
 import static org.elasticsearch.test.NodeRoles.onlyRoles;
@@ -52,11 +54,10 @@ public void testEachMasterPublishesTheirThresholds() throws Exception {
             for (int i = 0; i < numberOfNodes; i++) {
                 ByteSizeValue randomBytes = ByteSizeValue.ofBytes(randomLongBetween(6, 19));
                 String customWatermark = percentageMode ? randomIntBetween(86, 94) + "%" : randomBytes.toString();
-                ByteSizeValue customMaxHeadroom = percentageMode ? randomBytes : ByteSizeValue.MINUS_ONE;
                 var customShardLimits = new HealthMetadata.ShardLimits(randomIntBetween(1, 1000), randomIntBetween(1001, 2000));
-                String nodeName = startNode(internalCluster, customWatermark, customMaxHeadroom.toString(), customShardLimits);
+                String nodeName = startNode(internalCluster, customWatermark, randomBytes.toString(), customShardLimits);
                 watermarkByNode.put(nodeName, customWatermark);
-                maxHeadroomByNode.put(nodeName, customMaxHeadroom);
+                maxHeadroomByNode.put(nodeName, randomBytes);
                 shardLimitsPerNode.put(nodeName, customShardLimits);
             }
             ensureStableCluster(numberOfNodes);
@@ -66,7 +67,16 @@ public void testEachMasterPublishesTheirThresholds() throws Exception {
                 var healthMetadata = HealthMetadata.getFromClusterState(internalCluster.clusterService().state());
                 var diskMetadata = healthMetadata.getDiskMetadata();
                 assertThat(diskMetadata.describeHighWatermark(), equalTo(watermarkByNode.get(electedMaster)));
-                assertThat(diskMetadata.highMaxHeadroom(), equalTo(maxHeadroomByNode.get(electedMaster)));
+                // The value of the setting `cluster.routing.allocation.disk.watermark.high.max_headroom` depends upon the existence of
+                // `cluster.routing.allocation.disk.watermark.high`. Check {@link CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING}
+                assertThat(
+                    diskMetadata.highMaxHeadroom(),
+                    equalTo(
+                        percentageMode
+                            ? maxHeadroomByNode.get(electedMaster)
+                            : CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getDefault(EMPTY)
+                    )
+                );
 
                 var shardLimitsMetadata = healthMetadata.getShardLimitsMetadata();
                 assertEquals(shardLimitsMetadata, shardLimitsPerNode.get(electedMaster));
@@ -80,7 +90,18 @@ public void testEachMasterPublishesTheirThresholds() throws Exception {
                 var healthMetadata = HealthMetadata.getFromClusterState(internalCluster.clusterService().state());
                 var diskMetadata = healthMetadata.getDiskMetadata();
                 assertThat(diskMetadata.describeHighWatermark(), equalTo(watermarkByNode.get(electedMaster)));
-                assertThat(diskMetadata.highMaxHeadroom(), equalTo(maxHeadroomByNode.get(electedMaster)));
+
+                // The value of the setting `cluster.routing.allocation.disk.watermark.high.max_headroom` depends upon the existence of
+                // `cluster.routing.allocation.disk.watermark.high`.
+                // Check {@link DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING}
+                assertThat(
+                    diskMetadata.highMaxHeadroom(),
+                    equalTo(
+                        percentageMode
+                            ? maxHeadroomByNode.get(electedMaster)
+                            : CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getDefault(EMPTY)
+                    )
+                );
 
                 var shardLimitsMetadata = healthMetadata.getShardLimitsMetadata();
                 assertEquals(shardLimitsMetadata, shardLimitsPerNode.get(electedMaster));
@@ -93,7 +114,7 @@ public void testWatermarkSettingUpdate() throws Exception {
             int numberOfNodes = 3;
             ByteSizeValue randomBytes = ByteSizeValue.ofBytes(randomLongBetween(6, 19));
             String initialWatermark = percentageMode ? randomIntBetween(86, 94) + "%" : randomBytes.toString();
-            ByteSizeValue initialMaxHeadroom = percentageMode ? randomBytes : ByteSizeValue.MINUS_ONE;
+            ByteSizeValue initialMaxHeadroom = randomBytes;
             HealthMetadata.ShardLimits initialShardLimits = new HealthMetadata.ShardLimits(
                 randomIntBetween(1, 1000),
                 randomIntBetween(1001, 2000)
@@ -107,7 +128,7 @@ public void testWatermarkSettingUpdate() throws Exception {
             ByteSizeValue updatedLowMaxHeadroom = percentageMode ? randomBytes : ByteSizeValue.MINUS_ONE;
             randomBytes = ByteSizeValue.ofBytes(randomLongBetween(50, 100));
             String updatedHighWatermark = percentageMode ? randomIntBetween(60, 90) + "%" : randomBytes.toString();
-            ByteSizeValue updatedHighMaxHeadroom = percentageMode ? randomBytes : ByteSizeValue.MINUS_ONE;
+            ByteSizeValue updatedHighMaxHeadroom = randomBytes;
             randomBytes = ByteSizeValue.ofBytes(randomLongBetween(5, 10));
             String updatedFloodStageWatermark = percentageMode ? randomIntBetween(91, 95) + "%" : randomBytes.toString();
             ByteSizeValue updatedFloodStageMaxHeadroom = percentageMode ? randomBytes : ByteSizeValue.MINUS_ONE;
@@ -121,7 +142,16 @@ public void testWatermarkSettingUpdate() throws Exception {
                 var healthMetadata = HealthMetadata.getFromClusterState(internalCluster.clusterService().state());
                 var diskMetadata = healthMetadata.getDiskMetadata();
                 assertThat(diskMetadata.describeHighWatermark(), equalTo(initialWatermark));
-                assertThat(diskMetadata.highMaxHeadroom(), equalTo(initialMaxHeadroom));
+
+                // The value of the setting `cluster.routing.allocation.disk.watermark.high.max_headroom` depends upon the existence of
+                // `cluster.routing.allocation.disk.watermark.high`.
+                // Check {@link DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING}
+                assertThat(
+                    diskMetadata.highMaxHeadroom(),
+                    equalTo(
+                        percentageMode ? initialMaxHeadroom : CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getDefault(EMPTY)
+                    )
+                );
 
                 var shardLimitsMetadata = healthMetadata.getShardLimitsMetadata();
                 assertEquals(shardLimitsMetadata, initialShardLimits);
@@ -147,9 +177,31 @@ public void testWatermarkSettingUpdate() throws Exception {
                 var healthMetadata = HealthMetadata.getFromClusterState(internalCluster.clusterService().state());
                 var diskMetadata = healthMetadata.getDiskMetadata();
                 assertThat(diskMetadata.describeHighWatermark(), equalTo(updatedHighWatermark));
-                assertThat(diskMetadata.highMaxHeadroom(), equalTo(updatedHighMaxHeadroom));
+
+                // The value of the setting `cluster.routing.allocation.disk.watermark.high.max_headroom` depends upon the existence of
+                // `cluster.routing.allocation.disk.watermark.high`.
+                // Check {@link DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING}
+                assertThat(
+                    diskMetadata.highMaxHeadroom(),
+                    equalTo(
+                        percentageMode
+                            ? updatedHighMaxHeadroom
+                            : CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getDefault(EMPTY)
+                    )
+                );
                 assertThat(diskMetadata.describeFloodStageWatermark(), equalTo(updatedFloodStageWatermark));
-                assertThat(diskMetadata.floodStageMaxHeadroom(), equalTo(updatedFloodStageMaxHeadroom));
+
+                // The value of the setting `cluster.routing.allocation.disk.watermark.flood_stage.max_headroom` depends upon the existence
+                // of `cluster.routing.allocation.disk.watermark.flood_stage`.
+                // Check{@link DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING}
+                assertThat(
+                    diskMetadata.floodStageMaxHeadroom(),
+                    equalTo(
+                        percentageMode
+                            ? updatedFloodStageMaxHeadroom
+                            : CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING.getDefault(EMPTY)
+                    )
+                );
 
                 var shardLimitsMetadata = healthMetadata.getShardLimitsMetadata();
                 assertEquals(shardLimitsMetadata, updatedShardLimits);
@@ -175,21 +227,17 @@ private String startNode(
 
     private Settings createWatermarkSettings(String highWatermark, String highMaxHeadroom) {
         // We define both thresholds to avoid inconsistencies over the type of the thresholds
-        Settings.Builder settings = Settings.builder()
+        var settings = Settings.builder()
             .put(CLUSTER_ROUTING_ALLOCATION_LOW_DISK_WATERMARK_SETTING.getKey(), percentageMode ? "85%" : "20b")
             .put(CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING.getKey(), highWatermark)
             .put(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING.getKey(), percentageMode ? "95%" : "1b")
-            .put(
-                DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_WATERMARK_SETTING.getKey(),
-                percentageMode ? "95%" : "5b"
-            );
+            .put(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_WATERMARK_SETTING.getKey(), percentageMode ? "95%" : "5b");
+
         if (percentageMode) {
-            settings = settings.put(CLUSTER_ROUTING_ALLOCATION_LOW_DISK_MAX_HEADROOM_SETTING.getKey(), "20b")
+            settings.put(CLUSTER_ROUTING_ALLOCATION_LOW_DISK_MAX_HEADROOM_SETTING.getKey(), "20b")
                 .put(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING.getKey(), "1b")
-                .put(DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_MAX_HEADROOM_SETTING.getKey(), "5b");
-            if (highMaxHeadroom.equals("-1") == false) {
-                settings = settings.put(CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getKey(), highMaxHeadroom);
-            }
+                .put(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_MAX_HEADROOM_SETTING.getKey(), "5b")
+                .put(CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.getKey(), highMaxHeadroom);
         }
         return settings.build();
     }

server/src/main/java/org/elasticsearch/health/metadata/HealthMetadataService.java

@@ -24,7 +24,6 @@
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.settings.ClusterSettings;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Tuple;
 
@@ -50,7 +49,6 @@ public class HealthMetadataService {
 
     private final ClusterService clusterService;
     private final ClusterStateListener clusterStateListener;
-    private final Settings settings;
     private final MasterServiceTaskQueue<UpsertHealthMetadataTask> taskQueue;
     private volatile boolean enabled;
 
@@ -61,20 +59,19 @@ public class HealthMetadataService {
     // us from checking the cluster state before the cluster state is initialized
     private volatile boolean isMaster = false;
 
-    private HealthMetadataService(ClusterService clusterService, Settings settings) {
+    private HealthMetadataService(ClusterService clusterService) {
         this.clusterService = clusterService;
-        this.settings = settings;
         this.clusterStateListener = this::updateOnClusterStateChange;
-        this.enabled = ENABLED_SETTING.get(settings);
+        this.enabled = clusterService.getClusterSettings().get(ENABLED_SETTING);
         this.taskQueue = clusterService.createTaskQueue(
             "health metadata service",
             Priority.NORMAL,
             new UpsertHealthMetadataTask.Executor()
         );
     }
 
-    public static HealthMetadataService create(ClusterService clusterService, Settings settings) {
-        HealthMetadataService healthMetadataService = new HealthMetadataService(clusterService, settings);
+    public static HealthMetadataService create(ClusterService clusterService) {
+        HealthMetadataService healthMetadataService = new HealthMetadataService(clusterService);
         healthMetadataService.registerListeners();
         return healthMetadataService;
     }
@@ -84,7 +81,7 @@ private void registerListeners() {
             this.clusterService.addListener(clusterStateListener);
         }
 
-        ClusterSettings clusterSettings = clusterService.getClusterSettings();
+        var clusterSettings = clusterService.getClusterSettings();
         Stream.of(
             CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING,
             CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING,
@@ -159,7 +156,7 @@ private void updateOnSettingsUpdated(UpsertHealthMetadataTask task) {
     }
 
     private void resetHealthMetadata(String source) {
-        taskQueue.submitTask(source, new InsertHealthMetadata(settings), null);
+        taskQueue.submitTask(source, new InsertHealthMetadata(clusterService.getClusterSettings()), null);
     }
 
     public static List<NamedWriteableRegistry.Entry> getNamedWriteables() {
@@ -290,26 +287,26 @@ private HealthMetadata.ShardLimits shardLimitsMetadataFrom(HealthMetadata.ShardL
      */
     static class InsertHealthMetadata extends UpsertHealthMetadataTask {
 
-        private final Settings settings;
+        private final ClusterSettings settings;
 
-        InsertHealthMetadata(Settings settings) {
+        InsertHealthMetadata(ClusterSettings settings) {
             this.settings = settings;
         }
 
         @Override
         HealthMetadata doExecute(HealthMetadata initialHealthMetadata) {
             return new HealthMetadata(
                 new HealthMetadata.Disk(
-                    CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING.get(settings),
-                    CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING.get(settings),
-                    CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING.get(settings),
-                    CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING.get(settings),
-                    CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_WATERMARK_SETTING.get(settings),
-                    CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_MAX_HEADROOM_SETTING.get(settings)
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING),
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_MAX_HEADROOM_SETTING),
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING),
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_MAX_HEADROOM_SETTING),
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_WATERMARK_SETTING),
+                    settings.get(CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_FROZEN_MAX_HEADROOM_SETTING)
                 ),
                 new HealthMetadata.ShardLimits(
-                    SETTING_CLUSTER_MAX_SHARDS_PER_NODE.get(settings),
-                    SETTING_CLUSTER_MAX_SHARDS_PER_NODE_FROZEN.get(settings)
+                    settings.get(SETTING_CLUSTER_MAX_SHARDS_PER_NODE),
+                    settings.get(SETTING_CLUSTER_MAX_SHARDS_PER_NODE_FROZEN)
                 )
             );
         }

server/src/main/java/org/elasticsearch/node/Node.java

@@ -1033,7 +1033,7 @@ protected Node(
                 threadPool,
                 systemIndices
             );
-            HealthMetadataService healthMetadataService = HealthMetadataService.create(clusterService, settings);
+            HealthMetadataService healthMetadataService = HealthMetadataService.create(clusterService);
             LocalHealthMonitor localHealthMonitor = LocalHealthMonitor.create(settings, clusterService, nodeService, threadPool, client);
             HealthInfoCache nodeHealthOverview = HealthInfoCache.create(clusterService);
             HealthApiStats healthApiStats = new HealthApiStats();