Always fetch node usage stats for write load decider (#136212)

The NodeUsageStats Action will not be unconditionally called
every 30 seconds, regardless of the write load decider being
enabled. It is a lightweight action that fetches stats that
are collected on the node regardless. This supports better
seeing what happens when the write load decider is enabled,
providing a before- (not just after) enablement picture.

ES-12631

Author: DiannaHohensee

server/src/internalClusterTest/java/org/elasticsearch/index/shard/IndexShardIT.java

@@ -305,46 +305,27 @@ public void testHeapUsageEstimateIsPresent() {
     }
 
     public void testNodeWriteLoadsArePresent() {
-        // Disable write load decider to begin with
-        setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);
-
         InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);
+
+        // Force a ClusterInfo refresh to run collection of the node thread pool usage stats.
         ClusterInfoServiceUtils.refresh(clusterInfoService);
         Map<String, NodeUsageStatsForThreadPools> nodeThreadPoolStats = clusterInfoService.getClusterInfo()
             .getNodeUsageStatsForThreadPools();
         assertNotNull(nodeThreadPoolStats);
-        /** Not collecting stats yet because allocation write load stats collection is disabled by default.
-         *  see {@link WriteLoadConstraintSettings.WRITE_LOAD_DECIDER_ENABLED_SETTING} */
-        assertTrue(nodeThreadPoolStats.isEmpty());
-
-        // Enable collection for node write loads.
-        setWriteLoadDeciderEnablement(
-            randomBoolean()
-                ? WriteLoadConstraintSettings.WriteLoadDeciderStatus.ENABLED
-                : WriteLoadConstraintSettings.WriteLoadDeciderStatus.LOW_THRESHOLD_ONLY
-        );
-        try {
-            // Force a ClusterInfo refresh to run collection of the node thread pool usage stats.
-            ClusterInfoServiceUtils.refresh(clusterInfoService);
-            nodeThreadPoolStats = clusterInfoService.getClusterInfo().getNodeUsageStatsForThreadPools();
 
-            /** Verify that each node has usage stats reported. */
-            ClusterState state = getInstanceFromNode(ClusterService.class).state();
-            assertEquals(state.nodes().size(), nodeThreadPoolStats.size());
-            for (DiscoveryNode node : state.nodes()) {
-                assertTrue(nodeThreadPoolStats.containsKey(node.getId()));
-                NodeUsageStatsForThreadPools nodeUsageStatsForThreadPools = nodeThreadPoolStats.get(node.getId());
-                assertThat(nodeUsageStatsForThreadPools.nodeId(), equalTo(node.getId()));
-                NodeUsageStatsForThreadPools.ThreadPoolUsageStats writeThreadPoolStats = nodeUsageStatsForThreadPools
-                    .threadPoolUsageStatsMap()
-                    .get(ThreadPool.Names.WRITE);
-                assertNotNull(writeThreadPoolStats);
-                assertThat(writeThreadPoolStats.totalThreadPoolThreads(), greaterThanOrEqualTo(0));
-                assertThat(writeThreadPoolStats.averageThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
-                assertThat(writeThreadPoolStats.maxThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
-            }
-        } finally {
-            clearWriteLoadDeciderEnablementSetting();
+        /** Verify that each node has usage stats reported. */
+        ClusterState state = getInstanceFromNode(ClusterService.class).state();
+        assertEquals(state.nodes().size(), nodeThreadPoolStats.size());
+        for (DiscoveryNode node : state.nodes()) {
+            assertTrue(nodeThreadPoolStats.containsKey(node.getId()));
+            NodeUsageStatsForThreadPools nodeUsageStatsForThreadPools = nodeThreadPoolStats.get(node.getId());
+            assertThat(nodeUsageStatsForThreadPools.nodeId(), equalTo(node.getId()));
+            NodeUsageStatsForThreadPools.ThreadPoolUsageStats writeThreadPoolStats = nodeUsageStatsForThreadPools.threadPoolUsageStatsMap()
+                .get(ThreadPool.Names.WRITE);
+            assertNotNull(writeThreadPoolStats);
+            assertThat(writeThreadPoolStats.totalThreadPoolThreads(), greaterThanOrEqualTo(0));
+            assertThat(writeThreadPoolStats.averageThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
+            assertThat(writeThreadPoolStats.maxThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
         }
     }
 
@@ -361,10 +342,10 @@ public void testShardWriteLoadsArePresent() {
 
         final InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);
 
-        // Explicitly disable write load decider
-        setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);
-
         try {
+            // Explicitly disable write load decider
+            setWriteLoadDeciderEnablement(WriteLoadConstraintSettings.WriteLoadDeciderStatus.DISABLED);
+
             // Stats should not be collected when the decider is disabled
             {
                 ClusterInfoServiceUtils.refresh(clusterInfoService);

server/src/main/java/org/elasticsearch/cluster/InternalClusterInfoService.java

@@ -218,7 +218,7 @@ void execute() {
                 maybeFetchIndicesStats(diskThresholdEnabled || writeLoadConstraintEnabled.atLeastLowThresholdEnabled());
                 maybeFetchNodeStats(diskThresholdEnabled || estimatedHeapThresholdEnabled);
                 maybeFetchNodesEstimatedHeapUsage(estimatedHeapThresholdEnabled);
-                maybeFetchNodesUsageStatsForThreadPools(writeLoadConstraintEnabled);
+                fetchNodesUsageStatsForThreadPools();
             }
         }
 
@@ -257,34 +257,25 @@ private void maybeFetchNodesEstimatedHeapUsage(boolean shouldFetch) {
             }
         }
 
-        private void maybeFetchNodesUsageStatsForThreadPools(WriteLoadDeciderStatus writeLoadConstraintEnabled) {
-            if (writeLoadConstraintEnabled.atLeastLowThresholdEnabled()) {
-                try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
-                    fetchNodesUsageStatsForThreadPools();
-                }
-            } else {
-                logger.trace("skipping collecting shard/node write load estimates from cluster, feature currently disabled");
-                nodeThreadPoolUsageStatsPerNode = Map.of();
-            }
-        }
-
         private void fetchNodesUsageStatsForThreadPools() {
-            nodeUsageStatsForThreadPoolsCollector.collectUsageStats(
-                client,
-                clusterStateSupplier.get(),
-                ActionListener.releaseAfter(new ActionListener<>() {
-                    @Override
-                    public void onResponse(Map<String, NodeUsageStatsForThreadPools> threadPoolStats) {
-                        nodeThreadPoolUsageStatsPerNode = threadPoolStats;
-                    }
+            try (var ignored = threadPool.getThreadContext().clearTraceContext()) {
+                nodeUsageStatsForThreadPoolsCollector.collectUsageStats(
+                    client,
+                    clusterStateSupplier.get(),
+                    ActionListener.releaseAfter(new ActionListener<>() {
+                        @Override
+                        public void onResponse(Map<String, NodeUsageStatsForThreadPools> threadPoolStats) {
+                            nodeThreadPoolUsageStatsPerNode = threadPoolStats;
+                        }
 
-                    @Override
-                    public void onFailure(Exception e) {
-                        logger.warn("failed to fetch thread pool usage estimates for nodes", e);
-                        nodeThreadPoolUsageStatsPerNode = Map.of();
-                    }
-                }, fetchRefs.acquire())
-            );
+                        @Override
+                        public void onFailure(Exception e) {
+                            logger.warn("failed to fetch thread pool usage estimates for nodes", e);
+                            nodeThreadPoolUsageStatsPerNode = Map.of();
+                        }
+                    }, fetchRefs.acquire())
+                );
+            }
         }
 
         private void fetchNodesEstimatedHeapUsage() {