allocation: add an isHotspotting flag metric, and node ids to duration (elastic#140491)

Previous work to add metrics to the write load constraint monitor did not
include two things needed for an in-depth dashboard: a node id/name attribute on
hotspot durations, and a set of currently hotspotting nodes. (Durations are
issued after the hotspot is complete.) This adds a new metric for the current
set of hotspotting nodes with name/id annotations, and adds name/id annotations
to hotspot durations. The NodeIdName record type has been added to contain
annotation data when a hotspot is first noted, as it cannot be resolved	later
through a DiscoveryNode when ClusterInfo/ClusterState may be out of sync or the
node may have dropped.

Fixes: ES-13944

Author: schase-es

server/src/main/java/org/elasticsearch/cluster/routing/allocation/WriteLoadConstraintMonitor.java

@@ -16,19 +16,23 @@
 import org.elasticsearch.cluster.ClusterInfoService;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.NodeUsageStatsForThreadPools;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodeRole;
-import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.cluster.routing.RerouteService;
 import org.elasticsearch.common.Priority;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.telemetry.metric.DoubleHistogram;
+import org.elasticsearch.telemetry.metric.LongGauge;
 import org.elasticsearch.telemetry.metric.LongWithAttributes;
 import org.elasticsearch.telemetry.metric.MeterRegistry;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -45,6 +49,7 @@
 public class WriteLoadConstraintMonitor {
     public static final String HOTSPOT_NODES_COUNT_METRIC_NAME = "es.allocator.allocations.node.write_load_hotspot.current";
     public static final String HOTSPOT_DURATION_METRIC_NAME = "es.allocator.allocations.node.write_load_hotspot.duration.histogram";
+    public static final String HOTSPOT_NODES_FLAG_METRIC_NAME = "es.allocator.allocations.node.write_load_hotspot.node_flag.current";
 
     private static final Logger logger = LogManager.getLogger(WriteLoadConstraintMonitor.class);
     private static final int MAX_NODE_IDS_IN_MESSAGE = 3;
@@ -53,11 +58,12 @@ public class WriteLoadConstraintMonitor {
     private final LongSupplier currentTimeMillisSupplier;
     private final RerouteService rerouteService;
     private volatile long lastRerouteTimeMillis = 0;
-    private final Map<String, Long> hotspotNodeStartTimes = new HashMap<>();
+    private volatile Map<NodeIdName, Long> hotspotNodeStartTimes = Map.of();
     private long hotspotNodeStartTimesLastTerm = -1L;
 
     private final AtomicLong hotspotNodesCount = new AtomicLong(-1L); // metrics source of hotspotting node count
     private final DoubleHistogram hotspotDurationHistogram;
+    private final LongGauge hotspotNodeFlagGauge; // nodes hotspotting have 1 written, with the name/id
 
     protected WriteLoadConstraintMonitor(
         WriteLoadConstraintSettings writeLoadConstraintSettings,
@@ -88,6 +94,12 @@ public WriteLoadConstraintMonitor(
             this::getHotspotNodesCount
         );
         hotspotDurationHistogram = meterRegistry.registerDoubleHistogram(HOTSPOT_DURATION_METRIC_NAME, "hotspot duration", "s");
+        hotspotNodeFlagGauge = meterRegistry.registerLongsGauge(
+            HOTSPOT_NODES_FLAG_METRIC_NAME,
+            "hotspot node flag",
+            "flag",
+            this::getHotspottingNodeFlags
+        );
     }
 
     /**
@@ -110,7 +122,7 @@ public void onNewInfo(ClusterInfo clusterInfo) {
         logger.trace("processing new cluster info");
 
         final int numberOfNodes = clusterInfo.getNodeUsageStatsForThreadPools().size();
-        final Set<String> writeNodeIdsExceedingQueueLatencyThreshold = Sets.newHashSetWithExpectedSize(numberOfNodes);
+        final Set<NodeIdName> writeNodesExceedingQueueLatencyThreshold = Sets.newHashSetWithExpectedSize(numberOfNodes);
         final Map<String, NodeUsageStatsForThreadPools> nodeUsageStats = clusterInfo.getNodeUsageStatsForThreadPools();
         var haveWriteNodesBelowQueueLatencyThreshold = false;
         var totalIngestNodes = 0;
@@ -132,23 +144,23 @@ public void onNewInfo(ClusterInfo clusterInfo) {
                 .get(ThreadPool.Names.WRITE);
             assert writeThreadPoolStats != null : "Write thread pool is not publishing usage stats for node [" + nodeId + "]";
             if (writeThreadPoolStats.maxThreadPoolQueueLatencyMillis() >= writeLoadConstraintSettings.getQueueLatencyThreshold().millis()) {
-                writeNodeIdsExceedingQueueLatencyThreshold.add(nodeId);
+                writeNodesExceedingQueueLatencyThreshold.add(NodeIdName.nodeIdName(node));
             } else {
                 haveWriteNodesBelowQueueLatencyThreshold = true;
             }
         }
 
         final long currentTimeMillis = currentTimeMillisSupplier.getAsLong();
-        Set<String> lastHotspotNodes = recordHotspotDurations(state, writeNodeIdsExceedingQueueLatencyThreshold, currentTimeMillis);
+        Set<NodeIdName> lastHotspotNodes = recordHotspotDurations(state, writeNodesExceedingQueueLatencyThreshold, currentTimeMillis);
 
-        if (writeNodeIdsExceedingQueueLatencyThreshold.isEmpty()) {
+        if (writeNodesExceedingQueueLatencyThreshold.isEmpty()) {
             logger.trace("No hot-spotting write nodes detected");
             return;
         }
         if (haveWriteNodesBelowQueueLatencyThreshold == false) {
             logger.debug("""
                 Nodes [{}] are above the queue latency threshold, but there are no write nodes below the threshold. \
-                Cannot rebalance shards.""", nodeSummary(writeNodeIdsExceedingQueueLatencyThreshold, state));
+                Cannot rebalance shards.""", nodeSummary(writeNodesExceedingQueueLatencyThreshold));
             return;
         }
 
@@ -158,20 +170,20 @@ public void onNewInfo(ClusterInfo clusterInfo) {
 
         // We know that there is at least one hot-spotting node if we've reached this code. Now check whether there are any new hot-spots
         // or hot-spots that are persisting and need further balancing work.
-        Set<String> newHotspotNodes = Sets.difference(writeNodeIdsExceedingQueueLatencyThreshold, lastHotspotNodes);
+        Set<NodeIdName> newHotspotNodes = Sets.difference(writeNodesExceedingQueueLatencyThreshold, lastHotspotNodes);
         if (haveCalledRerouteRecently == false || newHotspotNodes.isEmpty() == false) {
             if (logger.isDebugEnabled()) {
                 logger.debug(
                     """
                         Nodes [{}] are hot-spotting, of {} total ingest nodes. Reroute for hot-spotting {}. \
                         Previously hot-spotting nodes are [{}]. The write thread pool queue latency threshold is [{}]. Triggering reroute.
                         """,
-                    nodeSummary(writeNodeIdsExceedingQueueLatencyThreshold, state),
+                    nodeSummary(writeNodesExceedingQueueLatencyThreshold),
                     totalIngestNodes,
                     lastRerouteTimeMillis == 0
                         ? "has never previously been called"
                         : "was last called [" + TimeValue.timeValueMillis(timeSinceLastRerouteMillis) + "] ago",
-                    nodeSummary(lastHotspotNodes, state),
+                    nodeSummary(lastHotspotNodes),
                     writeLoadConstraintSettings.getQueueLatencyThreshold()
                 );
             }
@@ -195,30 +207,41 @@ public void onNewInfo(ClusterInfo clusterInfo) {
         }
     }
 
-    private void recordHotspotStartTimes(Set<String> nodeIds, long startTimestamp) {
-        for (String nodeId : nodeIds) {
-            hotspotNodeStartTimes.put(nodeId, startTimestamp);
+    private void recordHotspotStartTimes(Set<NodeIdName> newHotspotNodes, long startTimestamp) {
+        if (newHotspotNodes.size() > 0) {
+            Map<NodeIdName, Long> hotspotNodeStartTimesUpdate = new HashMap<>(hotspotNodeStartTimes);
+            for (NodeIdName nodeIdName : newHotspotNodes) {
+                hotspotNodeStartTimesUpdate.put(nodeIdName, startTimestamp);
+            }
+            hotspotNodeStartTimes = Collections.unmodifiableMap(hotspotNodeStartTimesUpdate);
         }
         hotspotNodesCount.set(hotspotNodeStartTimes.size());
     }
 
-    private Set<String> recordHotspotDurations(ClusterState state, Set<String> currentHotspotNodes, long hotspotEndTime) {
+    private Set<NodeIdName> recordHotspotDurations(ClusterState state, Set<NodeIdName> currentHotspotNodes, long hotspotEndTimeMs) {
         // reset hotspotNodeStartTimes if the term has changed
         if (state.term() != hotspotNodeStartTimesLastTerm || state.nodes().isLocalNodeElectedMaster() == false) {
             hotspotNodeStartTimesLastTerm = state.term();
-            hotspotNodeStartTimes.clear();
+            hotspotNodeStartTimes = Map.of();
         }
 
-        Set<String> lastHotspotNodes = hotspotNodeStartTimes.keySet();
-        Set<String> staleHotspotNodes = Sets.difference(lastHotspotNodes, currentHotspotNodes);
-
-        for (String nodeId : staleHotspotNodes) {
-            assert hotspotNodeStartTimes.containsKey(nodeId) : "Map should contain key from its own subset";
-            long hotspotStartTime = hotspotNodeStartTimes.remove(nodeId);
-            long hotspotDuration = hotspotEndTime - hotspotStartTime;
-            assert hotspotDuration >= 0 : "hotspot duration should always be non-negative";
-            hotspotDurationHistogram.record(hotspotDuration / 1000.0);
+        Set<NodeIdName> lastHotspotNodes = hotspotNodeStartTimes.keySet();
+        Set<NodeIdName> staleHotspotNodes = Sets.difference(lastHotspotNodes, currentHotspotNodes);
+        if (staleHotspotNodes.size() > 0) {
+            Map<NodeIdName, Long> hotspotNodeStartTimesUpdate = new HashMap<>(hotspotNodeStartTimes);
+            for (NodeIdName nodeIdName : staleHotspotNodes) {
+                assert hotspotNodeStartTimesUpdate.containsKey(nodeIdName) : "Map should contain key from its own subset";
+                long hotspotStartTimeMs = hotspotNodeStartTimesUpdate.remove(nodeIdName);
+                long hotspotDurationMs = hotspotEndTimeMs - hotspotStartTimeMs;
+                assert hotspotDurationMs >= 0 : "hotspot duration should always be non-negative";
+                hotspotDurationHistogram.record(
+                    hotspotDurationMs / 1000.0,
+                    Map.of("es_node_id", nodeIdName.nodeId(), "es_node_name", nodeIdName.nodeName())
+                );
+            }
+            hotspotNodeStartTimes = Collections.unmodifiableMap(hotspotNodeStartTimesUpdate);
         }
+
         hotspotNodesCount.set(hotspotNodeStartTimes.size());
 
         return lastHotspotNodes;
@@ -233,21 +256,32 @@ private List<LongWithAttributes> getHotspotNodesCount() {
         }
     }
 
-    private static String nodeSummary(Set<String> nodeIds, ClusterState state) {
-        final var nodes = state.nodes();
-        if (nodeIds.isEmpty() == false && nodeIds.size() <= MAX_NODE_IDS_IN_MESSAGE) {
-            return nodeIds.stream().map(nodeId -> nodeShortDescription(nodeId, nodes)).collect(Collectors.joining(", "));
-        } else {
-            return nodeIds.size() + " nodes";
+    private Collection<LongWithAttributes> getHotspottingNodeFlags() {
+        final Map<NodeIdName, Long> hotspotNodeStartTimesView = hotspotNodeStartTimes;
+        List<LongWithAttributes> hotspottingNodeFlags = new ArrayList<>(hotspotNodeStartTimesView.size());
+        for (NodeIdName nodeIdName : hotspotNodeStartTimesView.keySet()) {
+            hotspottingNodeFlags.add(
+                new LongWithAttributes(1L, Map.of("es_node_id", nodeIdName.nodeId(), "es_node_name", nodeIdName.nodeName()))
+            );
         }
+        return hotspottingNodeFlags;
     }
 
-    /**
-     * @return "{nodeId}/{nodeName}" if available, or just "{nodeId}" otherwise
-     */
-    private static String nodeShortDescription(String nodeId, DiscoveryNodes nodes) {
-        final var discoveryNode = nodes.get(nodeId);
-        // It's possible a node might have left the cluster since the ClusterInfo was published
-        return discoveryNode != null ? discoveryNode.getShortNodeDescription() : nodeId;
+    public record NodeIdName(String nodeId, String nodeName) {
+        public static NodeIdName nodeIdName(DiscoveryNode node) {
+            return new NodeIdName(node.getId(), node.getName());
+        }
+
+        public String shortDescription() {
+            return nodeId + "/" + nodeName;
+        }
+    }
+
+    private static String nodeSummary(Set<NodeIdName> nodeIdNames) {
+        if (nodeIdNames.isEmpty() == false && nodeIdNames.size() <= MAX_NODE_IDS_IN_MESSAGE) {
+            return nodeIdNames.stream().map(nodeIdName -> nodeIdName.shortDescription()).collect(Collectors.joining(", "));
+        } else {
+            return nodeIdNames.size() + " nodes";
+        }
     }
 }