Add simple queue latency simulation by shard movemment (#133197)

ywangd · web-flow · commit a31485f6e8f1 · 2025-08-22T10:12:10.000+02:00
Any momving away shard reduces queue latency to zero, effectively
"fixes' hot-spotting. Rinse and repeat when the next ClusterInfo
arrives.

Resolves: ES-12622
diff --git a/server/src/main/java/org/elasticsearch/cluster/routing/ShardMovementWriteLoadSimulator.java b/server/src/main/java/org/elasticsearch/cluster/routing/ShardMovementWriteLoadSimulator.java
@@ -19,7 +19,9 @@
 import org.elasticsearch.threadpool.ThreadPool;
 
 import java.util.Collections;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * Simulates the impact to each node's write-load in response to the movement of individual
@@ -30,11 +32,14 @@ public class ShardMovementWriteLoadSimulator {
     private final Map<String, NodeUsageStatsForThreadPools> originalNodeUsageStatsForThreadPools;
     private final ObjectDoubleMap<String> simulatedNodeWriteLoadDeltas;
     private final Map<ShardId, Double> writeLoadsPerShard;
+    // The set to track whether a node has seen a shard move away from it
+    private final Set<String> nodesWithMovedAwayShard;
 
     public ShardMovementWriteLoadSimulator(RoutingAllocation routingAllocation) {
         this.originalNodeUsageStatsForThreadPools = routingAllocation.clusterInfo().getNodeUsageStatsForThreadPools();
         this.writeLoadsPerShard = routingAllocation.clusterInfo().getShardWriteLoads();
         this.simulatedNodeWriteLoadDeltas = new ObjectDoubleHashMap<>();
+        this.nodesWithMovedAwayShard = new HashSet<>();
     }
 
     public void simulateShardStarted(ShardRouting shardRouting) {
@@ -46,6 +51,7 @@ public void simulateShardStarted(ShardRouting shardRouting) {
                 // This is a shard being relocated
                 simulatedNodeWriteLoadDeltas.addTo(shardRouting.relocatingNodeId(), -1 * writeLoadForShard);
                 simulatedNodeWriteLoadDeltas.addTo(shardRouting.currentNodeId(), writeLoadForShard);
+                nodesWithMovedAwayShard.add(shardRouting.relocatingNodeId());
             } else {
                 // This is a new shard starting, it's unlikely we'll have a write-load value for a new
                 // shard, but we may be able to estimate if the new shard is created as part of a datastream
@@ -69,7 +75,11 @@ public Map<String, NodeUsageStatsForThreadPools> simulatedNodeUsageStatsForThrea
                     Maps.copyMapWithAddedOrReplacedEntry(
                         entry.getValue().threadPoolUsageStatsMap(),
                         ThreadPool.Names.WRITE,
-                        replaceWritePoolStats(entry.getValue(), simulatedNodeWriteLoadDeltas.get(entry.getKey()))
+                        replaceWritePoolStats(
+                            entry.getValue(),
+                            simulatedNodeWriteLoadDeltas.get(entry.getKey()),
+                            nodesWithMovedAwayShard.contains(entry.getKey())
+                        )
                     )
                 );
                 adjustedNodeUsageStatsForThreadPools.put(entry.getKey(), adjustedValue);
@@ -82,7 +92,8 @@ public Map<String, NodeUsageStatsForThreadPools> simulatedNodeUsageStatsForThrea
 
     private static NodeUsageStatsForThreadPools.ThreadPoolUsageStats replaceWritePoolStats(
         NodeUsageStatsForThreadPools value,
-        double writeLoadDelta
+        double writeLoadDelta,
+        boolean hasSeenMovedAwayShard
     ) {
         final NodeUsageStatsForThreadPools.ThreadPoolUsageStats writeThreadPoolStats = value.threadPoolUsageStatsMap()
             .get(ThreadPool.Names.WRITE);
@@ -93,7 +104,7 @@ private static NodeUsageStatsForThreadPools.ThreadPoolUsageStats replaceWritePoo
                 (float) writeLoadDelta,
                 writeThreadPoolStats.totalThreadPoolThreads()
             ),
-            writeThreadPoolStats.maxThreadPoolQueueLatencyMillis()
+            adjustThreadPoolQueueLatencyWithShardMovements(writeThreadPoolStats.maxThreadPoolQueueLatencyMillis(), hasSeenMovedAwayShard)
         );
     }
 
@@ -116,4 +127,18 @@ public static float updateNodeUtilizationWithShardMovements(
         float newNodeUtilization = nodeUtilization + (shardWriteLoadDelta / numberOfWriteThreads);
         return (float) Math.max(newNodeUtilization, 0.0);
     }
+
+    /**
+     * Adjust the max thread pool queue latency by accounting for whether shard has moved away from the node.
+     * @param maxThreadPoolQueueLatencyMillis The current max thread pool queue latency.
+     * @param hasSeenMovedAwayShard Whether the node has seen a shard move away from it.
+     * @return The new adjusted max thread pool queue latency.
+     */
+    public static long adjustThreadPoolQueueLatencyWithShardMovements(long maxThreadPoolQueueLatencyMillis, boolean hasSeenMovedAwayShard) {
+        // Intentionally keep it simple by reducing queue latency to zero if we move any shard off the node.
+        // This means the node is considered as no longer hot-spotting (with respect to queue latency) once a shard moves away.
+        // The next ClusterInfo will come in up-to 30 seconds later, and we will see the actual impact of the
+        // shard movement and repeat the process. We keep the queue latency unchanged if shard moves onto the node.
+        return hasSeenMovedAwayShard ? 0L : maxThreadPoolQueueLatencyMillis;
+    }
 }
diff --git a/server/src/test/java/org/elasticsearch/cluster/routing/ShardMovementWriteLoadSimulatorTests.java b/server/src/test/java/org/elasticsearch/cluster/routing/ShardMovementWriteLoadSimulatorTests.java
@@ -48,9 +48,9 @@ public void testNoShardMovement() {
         final var originalNode0ThreadPoolStats = randomThreadPoolUsageStats();
         final var originalNode1ThreadPoolStats = randomThreadPoolUsageStats();
         final var allocation = createRoutingAllocationWithRandomisedWriteLoads(
+            Set.of(),
             originalNode0ThreadPoolStats,
-            originalNode1ThreadPoolStats,
-            Set.of()
+            originalNode1ThreadPoolStats
         );
 
         final var shardMovementWriteLoadSimulator = new ShardMovementWriteLoadSimulator(allocation);
@@ -66,13 +66,15 @@ public void testNoShardMovement() {
         );
     }
 
-    public void testMovementOfAShardWillMoveThreadPoolUtilisation() {
+    public void testMovementOfAShardWillMoveThreadPoolStats() {
         final var originalNode0ThreadPoolStats = randomThreadPoolUsageStats();
         final var originalNode1ThreadPoolStats = randomThreadPoolUsageStats();
+        final var originalNode2ThreadPoolStats = randomThreadPoolUsageStats();
         final var allocation = createRoutingAllocationWithRandomisedWriteLoads(
+            Set.of(),
             originalNode0ThreadPoolStats,
             originalNode1ThreadPoolStats,
-            Set.of()
+            originalNode2ThreadPoolStats
         );
         final var shardMovementWriteLoadSimulator = new ShardMovementWriteLoadSimulator(allocation);
 
@@ -84,7 +86,7 @@ public void testMovementOfAShardWillMoveThreadPoolUtilisation() {
         final ShardRouting movedAndStartedShard = allocation.routingNodes().startShard(moveShardTuple.v2(), NOOP, expectedShardSize);
 
         final var calculatedNodeUsageStats = shardMovementWriteLoadSimulator.simulatedNodeUsageStatsForThreadPools();
-        assertThat(calculatedNodeUsageStats, Matchers.aMapWithSize(2));
+        assertThat(calculatedNodeUsageStats, Matchers.aMapWithSize(3));
 
         final var shardWriteLoad = allocation.clusterInfo().getShardWriteLoads().get(randomShard.shardId());
         final var expectedUtilisationReductionAtSource = shardWriteLoad / originalNode0ThreadPoolStats.totalThreadPoolThreads();
@@ -109,6 +111,19 @@ public void testMovementOfAShardWillMoveThreadPoolUtilisation() {
             closeTo(expectedUtilisationIncreaseAtDestination, 0.001f)
         );
 
+        // Queue latency reduced for node_0 since it has a shard moved out
+        assertThat(getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_0"), equalTo(0L));
+        // Queue latency stays unchanged for node_1 since it only has a shard moved in
+        assertThat(
+            getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_1"),
+            equalTo(originalNode1ThreadPoolStats.maxThreadPoolQueueLatencyMillis())
+        );
+        // Queue latency stays unchanged for node_2 since it has no shard movement
+        assertThat(
+            getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_2"),
+            equalTo(originalNode2ThreadPoolStats.maxThreadPoolQueueLatencyMillis())
+        );
+
         // Then move it back
         final var moveBackTuple = allocation.routingNodes()
             .relocateShard(movedAndStartedShard, "node_0", expectedShardSize, "testing", NOOP);
@@ -123,15 +138,25 @@ public void testMovementOfAShardWillMoveThreadPoolUtilisation() {
             getAverageWritePoolUtilization(shardMovementWriteLoadSimulator, "node_1"),
             equalTo(originalNode1ThreadPoolStats.averageThreadPoolUtilization())
         );
+
+        // We intentionally keep things simple so that if a shard has moved away from a node, its queue latency is reduced to zero
+        // regardless of whether other shards have subsequently moved onto or out of the same node.
+        assertThat(getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_0"), equalTo(0L));
+        assertThat(getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_1"), equalTo(0L));
+        // Queue latency stays unchanged for node_2 since it has no shard movement
+        assertThat(
+            getMaxThreadPoolQueueLatency(shardMovementWriteLoadSimulator, "node_2"),
+            equalTo(originalNode2ThreadPoolStats.maxThreadPoolQueueLatencyMillis())
+        );
     }
 
     public void testMovementBetweenNodesWithNoThreadPoolAndWriteLoadStats() {
         final var originalNode0ThreadPoolStats = randomBoolean() ? randomThreadPoolUsageStats() : null;
         final var originalNode1ThreadPoolStats = randomBoolean() ? randomThreadPoolUsageStats() : null;
         final var allocation = createRoutingAllocationWithRandomisedWriteLoads(
+            new HashSet<>(randomSubsetOf(Arrays.asList(INDICES))),
             originalNode0ThreadPoolStats,
-            originalNode1ThreadPoolStats,
-            new HashSet<>(randomSubsetOf(Arrays.asList(INDICES)))
+            originalNode1ThreadPoolStats
         );
         final var shardMovementWriteLoadSimulator = new ShardMovementWriteLoadSimulator(allocation);
 
@@ -147,12 +172,29 @@ public void testMovementBetweenNodesWithNoThreadPoolAndWriteLoadStats() {
         assertThat(simulated.containsKey("node_1"), equalTo(originalNode1ThreadPoolStats != null));
     }
 
+    public void testUpdateThreadPoolQueueLatencyWithShardMovements() {
+        final long originalLatency = randomNonNegativeLong();
+
+        assertThat(
+            ShardMovementWriteLoadSimulator.adjustThreadPoolQueueLatencyWithShardMovements(originalLatency, false),
+            equalTo(originalLatency)
+        );
+
+        assertThat(ShardMovementWriteLoadSimulator.adjustThreadPoolQueueLatencyWithShardMovements(originalLatency, true), equalTo(0L));
+    }
+
     private float getAverageWritePoolUtilization(ShardMovementWriteLoadSimulator shardMovementWriteLoadSimulator, String nodeId) {
         final var generatedNodeUsageStates = shardMovementWriteLoadSimulator.simulatedNodeUsageStatsForThreadPools();
         final var node0WritePoolStats = generatedNodeUsageStates.get(nodeId).threadPoolUsageStatsMap().get("write");
         return node0WritePoolStats.averageThreadPoolUtilization();
     }
 
+    private long getMaxThreadPoolQueueLatency(ShardMovementWriteLoadSimulator shardMovementWriteLoadSimulator, String nodeId) {
+        final var generatedNodeUsageStates = shardMovementWriteLoadSimulator.simulatedNodeUsageStatsForThreadPools();
+        final var writePoolStats = generatedNodeUsageStates.get(nodeId).threadPoolUsageStatsMap().get("write");
+        return writePoolStats.maxThreadPoolQueueLatencyMillis();
+    }
+
     private NodeUsageStatsForThreadPools.ThreadPoolUsageStats randomThreadPoolUsageStats() {
         return new NodeUsageStatsForThreadPools.ThreadPoolUsageStats(
             randomIntBetween(4, 16),
@@ -162,16 +204,16 @@ private NodeUsageStatsForThreadPools.ThreadPoolUsageStats randomThreadPoolUsageS
     }
 
     private RoutingAllocation createRoutingAllocationWithRandomisedWriteLoads(
-        NodeUsageStatsForThreadPools.ThreadPoolUsageStats node0ThreadPoolStats,
-        NodeUsageStatsForThreadPools.ThreadPoolUsageStats node1ThreadPoolStats,
-        Set<String> indicesWithNoWriteLoad
+        Set<String> indicesWithNoWriteLoad,
+        NodeUsageStatsForThreadPools.ThreadPoolUsageStats... arrayOfNodeThreadPoolStats
     ) {
         final Map<String, NodeUsageStatsForThreadPools> nodeUsageStats = new HashMap<>();
-        if (node0ThreadPoolStats != null) {
-            nodeUsageStats.put("node_0", new NodeUsageStatsForThreadPools("node_0", Map.of("write", node0ThreadPoolStats)));
-        }
-        if (node1ThreadPoolStats != null) {
-            nodeUsageStats.put("node_1", new NodeUsageStatsForThreadPools("node_1", Map.of("write", node1ThreadPoolStats)));
+        for (int i = 0; i < arrayOfNodeThreadPoolStats.length; i++) {
+            final var nodeThreadPoolStats = arrayOfNodeThreadPoolStats[i];
+            if (nodeThreadPoolStats != null) {
+                final var nodeId = "node_" + i;
+                nodeUsageStats.put(nodeId, new NodeUsageStatsForThreadPools(nodeId, Map.of("write", nodeThreadPoolStats)));
+            }
         }
 
         final ClusterState clusterState = createClusterState();
