change NodeLoad data structure

Author: DiannaHohensee

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

@@ -24,8 +24,8 @@
 import org.elasticsearch.cluster.EstimatedHeapUsage;
 import org.elasticsearch.cluster.EstimatedHeapUsageCollector;
 import org.elasticsearch.cluster.InternalClusterInfoService;
-import org.elasticsearch.cluster.NodeWriteLoad;
-import org.elasticsearch.cluster.WriteLoadCollector;
+import org.elasticsearch.cluster.NodeExecutionLoad;
+import org.elasticsearch.cluster.NodeUsageLoadCollector;
 import org.elasticsearch.cluster.metadata.IndexMetadata;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeUtils;
@@ -78,6 +78,7 @@
 import org.elasticsearch.test.ESSingleNodeTestCase;
 import org.elasticsearch.test.IndexSettingsModule;
 import org.elasticsearch.test.InternalSettingsPlugin;
+import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xcontent.XContentType;
 import org.junit.Assert;
 
@@ -90,6 +91,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -131,7 +133,7 @@ public class IndexShardIT extends ESSingleNodeTestCase {
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
-        return pluginList(InternalSettingsPlugin.class, BogusEstimatedHeapUsagePlugin.class, BogusWriteLoadCollectorPlugin.class);
+        return pluginList(InternalSettingsPlugin.class, BogusEstimatedHeapUsagePlugin.class, BogusNodeUsageLoadCollectorPlugin.class);
     }
 
     public void testLockTryingToDelete() throws Exception {
@@ -305,11 +307,11 @@ public void testHeapUsageEstimateIsPresent() {
     public void testNodeWriteLoadsArePresent() {
         InternalClusterInfoService clusterInfoService = (InternalClusterInfoService) getInstanceFromNode(ClusterInfoService.class);
         ClusterInfoServiceUtils.refresh(clusterInfoService);
-        Map<String, NodeWriteLoad> nodeWriteLoads = clusterInfoService.getClusterInfo().getNodeWriteLoads();
-        assertNotNull(nodeWriteLoads);
+        Map<String, NodeExecutionLoad> nodeThreadPoolStats = clusterInfoService.getClusterInfo().getNodeExecutionStats();
+        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(nodeWriteLoads.isEmpty());
+        assertTrue(nodeThreadPoolStats.isEmpty());
 
         // Enable collection for node write loads.
         updateClusterSettings(
@@ -323,18 +325,21 @@ public void testNodeWriteLoadsArePresent() {
         try {
             // Force a ClusterInfo refresh to run collection of the node write loads.
             ClusterInfoServiceUtils.refresh(clusterInfoService);
-            nodeWriteLoads = clusterInfoService.getClusterInfo().getNodeWriteLoads();
+            nodeThreadPoolStats = clusterInfoService.getClusterInfo().getNodeExecutionStats();
 
-            /** Verify that each node has a write load reported. The test {@link BogusWriteLoadCollector} generates random load values */
+            /** Verify that each node has a write load reported. The test {@link BogusNodeUsageLoadCollector} generates random load values */
             ClusterState state = getInstanceFromNode(ClusterService.class).state();
-            assertEquals(state.nodes().size(), nodeWriteLoads.size());
+            assertEquals(state.nodes().size(), nodeThreadPoolStats.size());
             for (DiscoveryNode node : state.nodes()) {
-                assertTrue(nodeWriteLoads.containsKey(node.getId()));
-                NodeWriteLoad nodeWriteLoad = nodeWriteLoads.get(node.getId());
-                assertThat(nodeWriteLoad.nodeId(), equalTo(node.getId()));
-                assertThat(nodeWriteLoad.totalWriteThreadPoolThreads(), greaterThanOrEqualTo(0));
-                assertThat(nodeWriteLoad.averageWriteThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
-                assertThat(nodeWriteLoad.averageWriteThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
+                assertTrue(nodeThreadPoolStats.containsKey(node.getId()));
+                NodeExecutionLoad nodeExecutionLoad = nodeThreadPoolStats.get(node.getId());
+                assertThat(nodeExecutionLoad.nodeId(), equalTo(node.getId()));
+                NodeExecutionLoad.ThreadPoolUsageStats writeThreadPoolStats = nodeExecutionLoad.threadPoolUsageStatsMap()
+                    .get(ThreadPool.Names.WRITE);
+                assertNotNull(writeThreadPoolStats);
+                assertThat(writeThreadPoolStats.totalThreadPoolThreads(), greaterThanOrEqualTo(0));
+                assertThat(writeThreadPoolStats.averageThreadPoolUtilization(), greaterThanOrEqualTo(0.0f));
+                assertThat(writeThreadPoolStats.averageThreadPoolQueueLatencyMillis(), greaterThanOrEqualTo(0L));
             }
         } finally {
             updateClusterSettings(
@@ -925,42 +930,48 @@ public ClusterService getClusterService() {
     }
 
     /**
-     * A simple {@link WriteLoadCollector} implementation that creates and returns random {@link NodeWriteLoad} for each node in the
+     * A simple {@link NodeUsageLoadCollector} implementation that creates and returns random {@link NodeExecutionLoad} for each node in the
      * cluster.
      * <p>
      * Note: there's an 'org.elasticsearch.cluster.WriteLoadCollector' file that declares this implementation so that the plugin system can
      * pick it up and use it for the test set-up.
      */
-    public static class BogusWriteLoadCollector implements WriteLoadCollector {
+    public static class BogusNodeUsageLoadCollector implements NodeUsageLoadCollector {
 
-        private final BogusWriteLoadCollectorPlugin plugin;
+        private final BogusNodeUsageLoadCollectorPlugin plugin;
 
-        public BogusWriteLoadCollector(BogusWriteLoadCollectorPlugin plugin) {
+        public BogusNodeUsageLoadCollector(BogusNodeUsageLoadCollectorPlugin plugin) {
             this.plugin = plugin;
         }
 
         @Override
-        public void collectWriteLoads(ActionListener<Map<String, NodeWriteLoad>> listener) {
+        public void collectUsageStats(ActionListener<Map<String, NodeExecutionLoad>> listener) {
             ActionListener.completeWith(
                 listener,
                 () -> plugin.getClusterService()
                     .state()
                     .nodes()
                     .stream()
-                    .collect(
-                        Collectors.toUnmodifiableMap(
-                            DiscoveryNode::getId,
-                            node -> new NodeWriteLoad(node.getId(), randomNonNegativeInt(), randomNonNegativeInt(), randomNonNegativeLong())
-                        )
-                    )
+                    .collect(Collectors.toUnmodifiableMap(DiscoveryNode::getId, node -> makeRandomNodeLoad(node.getId())))
+            );
+        }
+
+        private NodeExecutionLoad makeRandomNodeLoad(String nodeId) {
+            NodeExecutionLoad.ThreadPoolUsageStats writeThreadPoolStats = new NodeExecutionLoad.ThreadPoolUsageStats(
+                randomNonNegativeInt(),
+                randomFloat(),
+                randomNonNegativeLong()
             );
+            Map<String, NodeExecutionLoad.ThreadPoolUsageStats> statsForThreadPools = new HashMap<>();
+            statsForThreadPools.put(ThreadPool.Names.WRITE, writeThreadPoolStats);
+            return new NodeExecutionLoad(nodeId, statsForThreadPools);
         }
     }
 
     /**
      * Make a plugin to gain access to the {@link ClusterService} instance.
      */
-    public static class BogusWriteLoadCollectorPlugin extends Plugin implements ClusterPlugin {
+    public static class BogusNodeUsageLoadCollectorPlugin extends Plugin implements ClusterPlugin {
 
         private final SetOnce<ClusterService> clusterService = new SetOnce<>();
 

server/src/internalClusterTest/resources/META-INF/services/org.elasticsearch.cluster.WriteLoadCollector renamed to server/src/internalClusterTest/resources/META-INF/services/org.elasticsearch.cluster.NodeUsageLoadCollector

@@ -7,4 +7,4 @@
 # License v3.0 only", or the "Server Side Public License, v 1".
 #
 
-org.elasticsearch.index.shard.IndexShardIT$BogusWriteLoadCollector
+org.elasticsearch.index.shard.IndexShardIT$BogusNodeUsageLoadCollector

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

@@ -58,7 +58,7 @@ public class ClusterInfo implements ChunkedToXContent, Writeable {
     final Map<NodeAndShard, String> dataPath;
     final Map<NodeAndPath, ReservedSpace> reservedSpace;
     final Map<String, EstimatedHeapUsage> estimatedHeapUsages;
-    final Map<String, NodeWriteLoad> nodeWriteLoads;
+    final Map<String, NodeExecutionLoad> nodeExecutionStats;
 
     protected ClusterInfo() {
         this(Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of(), Map.of());
@@ -74,7 +74,7 @@ protected ClusterInfo() {
      * @param dataPath the shard routing to datapath mapping
      * @param reservedSpace reserved space per shard broken down by node and data path
      * @param estimatedHeapUsages estimated heap usage broken down by node
-     * @param nodeWriteLoads estimated node-level write load broken down by node
+     * @param nodeExecutionStats estimated node-level execution load broken down by node
      * @see #shardIdentifierFromRouting
      */
     public ClusterInfo(
@@ -85,7 +85,7 @@ public ClusterInfo(
         Map<NodeAndShard, String> dataPath,
         Map<NodeAndPath, ReservedSpace> reservedSpace,
         Map<String, EstimatedHeapUsage> estimatedHeapUsages,
-        Map<String, NodeWriteLoad> nodeWriteLoads
+        Map<String, NodeExecutionLoad> nodeExecutionStats
     ) {
         this.leastAvailableSpaceUsage = Map.copyOf(leastAvailableSpaceUsage);
         this.mostAvailableSpaceUsage = Map.copyOf(mostAvailableSpaceUsage);
@@ -94,7 +94,7 @@ public ClusterInfo(
         this.dataPath = Map.copyOf(dataPath);
         this.reservedSpace = Map.copyOf(reservedSpace);
         this.estimatedHeapUsages = Map.copyOf(estimatedHeapUsages);
-        this.nodeWriteLoads = Map.copyOf(nodeWriteLoads);
+        this.nodeExecutionStats = Map.copyOf(nodeExecutionStats);
     }
 
     public ClusterInfo(StreamInput in) throws IOException {
@@ -112,9 +112,9 @@ public ClusterInfo(StreamInput in) throws IOException {
             this.estimatedHeapUsages = Map.of();
         }
         if (in.getTransportVersion().onOrAfter(TransportVersions.NODE_WRITE_LOAD_IN_CLUSTER_INFO)) {
-            this.nodeWriteLoads = in.readImmutableMap(NodeWriteLoad::new);
+            this.nodeExecutionStats = in.readImmutableMap(NodeExecutionLoad::new);
         } else {
-            this.nodeWriteLoads = Map.of();
+            this.nodeExecutionStats = Map.of();
         }
     }
 
@@ -134,7 +134,7 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeMap(this.estimatedHeapUsages, StreamOutput::writeWriteable);
         }
         if (out.getTransportVersion().onOrAfter(TransportVersions.NODE_WRITE_LOAD_IN_CLUSTER_INFO)) {
-            out.writeMap(this.nodeWriteLoads, StreamOutput::writeWriteable);
+            out.writeMap(this.nodeExecutionStats, StreamOutput::writeWriteable);
         }
     }
 
@@ -216,7 +216,7 @@ public Iterator<? extends ToXContent> toXContentChunked(ToXContent.Params params
                 return builder.endObject(); // NodeAndPath
             }),
             endArray() // end "reserved_sizes"
-            // NOTE: We don't serialize estimatedHeapUsages/nodeWriteLoads at this stage, to avoid
+            // NOTE: We don't serialize estimatedHeapUsages/nodeExecutionStats at this stage, to avoid
             // committing to API payloads until the features are settled
         );
     }
@@ -235,8 +235,8 @@ public Map<String, EstimatedHeapUsage> getEstimatedHeapUsages() {
     /**
      * Returns a map containing the node-level write load estimate for each node by node ID.
      */
-    public Map<String, NodeWriteLoad> getNodeWriteLoads() {
-        return nodeWriteLoads;
+    public Map<String, NodeExecutionLoad> getNodeExecutionStats() {
+        return nodeExecutionStats;
     }
 
     /**
@@ -331,7 +331,7 @@ public boolean equals(Object o) {
             && shardDataSetSizes.equals(that.shardDataSetSizes)
             && dataPath.equals(that.dataPath)
             && reservedSpace.equals(that.reservedSpace)
-            && nodeWriteLoads.equals(that.nodeWriteLoads);
+            && nodeExecutionStats.equals(that.nodeExecutionStats);
     }
 
     @Override
@@ -343,7 +343,7 @@ public int hashCode() {
             shardDataSetSizes,
             dataPath,
             reservedSpace,
-            nodeWriteLoads
+            nodeExecutionStats
         );
     }
 

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

@@ -34,7 +34,7 @@ public class ClusterInfoSimulator {
     private final Map<ShardId, Long> shardDataSetSizes;
     private final Map<NodeAndShard, String> dataPath;
     private final Map<String, EstimatedHeapUsage> estimatedHeapUsages;
-    private final Map<String, NodeWriteLoad> nodeWriteLoads;
+    private final Map<String, NodeExecutionLoad> nodeExecutionStats;
 
     public ClusterInfoSimulator(RoutingAllocation allocation) {
         this.allocation = allocation;
@@ -44,7 +44,7 @@ public ClusterInfoSimulator(RoutingAllocation allocation) {
         this.shardDataSetSizes = Map.copyOf(allocation.clusterInfo().shardDataSetSizes);
         this.dataPath = Map.copyOf(allocation.clusterInfo().dataPath);
         this.estimatedHeapUsages = allocation.clusterInfo().getEstimatedHeapUsages();
-        this.nodeWriteLoads = allocation.clusterInfo().getNodeWriteLoads();
+        this.nodeExecutionStats = allocation.clusterInfo().getNodeExecutionStats();
     }
 
     /**
@@ -159,7 +159,7 @@ public ClusterInfo getClusterInfo() {
             dataPath,
             Map.of(),
             estimatedHeapUsages,
-            nodeWriteLoads
+            nodeExecutionStats
         );
     }
 }

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

@@ -102,7 +102,7 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     private volatile Map<String, DiskUsage> mostAvailableSpaceUsages;
     private volatile Map<String, ByteSizeValue> maxHeapPerNode;
     private volatile Map<String, Long> estimatedHeapUsagePerNode;
-    private volatile Map<String, NodeWriteLoad> nodeWriteLoadPerNode;
+    private volatile Map<String, NodeExecutionLoad> nodeExecutionLoadPerNode;
     private volatile IndicesStatsSummary indicesStatsSummary;
 
     private final ThreadPool threadPool;
@@ -112,7 +112,7 @@ public class InternalClusterInfoService implements ClusterInfoService, ClusterSt
     private final Object mutex = new Object();
     private final List<ActionListener<ClusterInfo>> nextRefreshListeners = new ArrayList<>();
     private final EstimatedHeapUsageCollector estimatedHeapUsageCollector;
-    private final WriteLoadCollector writeLoadCollector;
+    private final NodeUsageLoadCollector nodeUsageLoadCollector;
 
     private AsyncRefresh currentRefresh;
     private RefreshScheduler refreshScheduler;
@@ -124,18 +124,18 @@ public InternalClusterInfoService(
         ThreadPool threadPool,
         Client client,
         EstimatedHeapUsageCollector estimatedHeapUsageCollector,
-        WriteLoadCollector writeLoadCollector
+        NodeUsageLoadCollector nodeUsageLoadCollector
     ) {
         this.leastAvailableSpaceUsages = Map.of();
         this.mostAvailableSpaceUsages = Map.of();
         this.maxHeapPerNode = Map.of();
         this.estimatedHeapUsagePerNode = Map.of();
-        this.nodeWriteLoadPerNode = Map.of();
+        this.nodeExecutionLoadPerNode = Map.of();
         this.indicesStatsSummary = IndicesStatsSummary.EMPTY;
         this.threadPool = threadPool;
         this.client = client;
         this.estimatedHeapUsageCollector = estimatedHeapUsageCollector;
-        this.writeLoadCollector = writeLoadCollector;
+        this.nodeUsageLoadCollector = nodeUsageLoadCollector;
         this.updateFrequency = INTERNAL_CLUSTER_INFO_UPDATE_INTERVAL_SETTING.get(settings);
         this.fetchTimeout = INTERNAL_CLUSTER_INFO_TIMEOUT_SETTING.get(settings);
         this.diskThresholdEnabled = DiskThresholdSettings.CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING.get(settings);
@@ -264,21 +264,21 @@ private void maybeFetchNodesWriteLoadStats(WriteLoadDeciderStatus writeLoadConst
                 }
             } else {
                 logger.trace("skipping collecting shard/node write load estimates from cluster, feature currently disabled");
-                nodeWriteLoadPerNode = Map.of();
+                nodeExecutionLoadPerNode = Map.of();
             }
         }
 
         private void fetchNodesWriteLoadStats() {
-            writeLoadCollector.collectWriteLoads(ActionListener.releaseAfter(new ActionListener<>() {
+            nodeUsageLoadCollector.collectUsageStats(ActionListener.releaseAfter(new ActionListener<>() {
                 @Override
-                public void onResponse(Map<String, NodeWriteLoad> writeLoads) {
-                    nodeWriteLoadPerNode = writeLoads;
+                public void onResponse(Map<String, NodeExecutionLoad> writeLoads) {
+                    nodeExecutionLoadPerNode = writeLoads;
                 }
 
                 @Override
                 public void onFailure(Exception e) {
                     logger.warn("failed to fetch write load estimates for nodes", e);
-                    nodeWriteLoadPerNode = Map.of();
+                    nodeExecutionLoadPerNode = Map.of();
                 }
             }, fetchRefs.acquire()));
         }
@@ -527,8 +527,8 @@ public ClusterInfo getClusterInfo() {
                 estimatedHeapUsages.put(nodeId, new EstimatedHeapUsage(nodeId, maxHeapSize.getBytes(), estimatedHeapUsage));
             }
         });
-        final Map<String, NodeWriteLoad> nodeWriteLoads = new HashMap<>();
-        nodeWriteLoadPerNode.forEach((nodeId, nodeWriteLoad) -> { nodeWriteLoads.put(nodeId, nodeWriteLoad); });
+        final Map<String, NodeExecutionLoad> nodeWriteLoads = new HashMap<>();
+        nodeExecutionLoadPerNode.forEach((nodeId, nodeWriteLoad) -> { nodeWriteLoads.put(nodeId, nodeWriteLoad); });
         return new ClusterInfo(
             leastAvailableSpaceUsages,
             mostAvailableSpaceUsages,