Balancer changes to use Decision#NOT_PREFERRED (#134160)

Closes ES-12716

Author: DiannaHohensee

server/src/internalClusterTest/java/org/elasticsearch/cluster/routing/allocation/decider/WriteLoadConstraintDeciderIT.java

@@ -179,6 +179,7 @@ public Decision getCanRemainDecision() {
      * the result of this method is meaningless, as no rebalance decision was taken.  If {@link #isDecisionTaken()}
      * returns {@code false}, then invoking this method will throw an {@code IllegalStateException}.
      */
+    // @VisibleForTesting
     public boolean canRebalanceCluster() {
         checkDecisionState();
         return clusterRebalanceDecision != null && clusterRebalanceDecision.type() == Type.YES;
@@ -192,6 +193,7 @@ public boolean canRebalanceCluster() {
      * If {@link #isDecisionTaken()} returns {@code false}, then invoking this method will throw an
      * {@code IllegalStateException}.
      */
+    // @VisibleForTesting
     @Nullable
     public Decision getClusterRebalanceDecision() {
         checkDecisionState();

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

@@ -861,7 +861,7 @@ public MoveDecision decideMove(final ProjectIndex index, final ShardRouting shar
             assert sourceNode != null && sourceNode.containsShard(index, shardRouting);
             RoutingNode routingNode = sourceNode.getRoutingNode();
             Decision canRemain = allocation.deciders().canRemain(shardRouting, routingNode, allocation);
-            if (canRemain.type() != Decision.Type.NO) {
+            if (canRemain.type() != Decision.Type.NO && canRemain.type() != Decision.Type.NOT_PREFERRED) {
                 return MoveDecision.remain(canRemain);
             }
 
@@ -901,7 +901,11 @@ private MoveDecision decideMove(
                     if (explain) {
                         nodeResults.add(new NodeAllocationResult(currentNode.getRoutingNode().node(), allocationDecision, ++weightRanking));
                     }
-                    // TODO maybe we can respect throttling here too?
+                    // TODO (ES-12633): test that nothing moves when the source is not-preferred and the target is not-preferred.
+                    if (allocationDecision.type() == Type.NOT_PREFERRED && remainDecision.type() == Type.NOT_PREFERRED) {
+                        // Relocating a shard from one NOT_PREFERRED node to another would not improve the situation.
+                        continue;
+                    }
                     if (allocationDecision.type().higherThan(bestDecision)) {
                         bestDecision = allocationDecision.type();
                         if (bestDecision == Type.YES) {
@@ -911,6 +915,10 @@ private MoveDecision decideMove(
                                 // no need to continue iterating
                                 break;
                             }
+                        } else if (bestDecision == Type.NOT_PREFERRED) {
+                            assert remainDecision.type() != Type.NOT_PREFERRED;
+                            // If we don't ever find a YES decision, we'll settle for NOT_PREFERRED as preferable to NO.
+                            targetNode = target;
                         }
                     }
                 }
@@ -1221,7 +1229,7 @@ private boolean tryRelocateShard(ModelNode minNode, ModelNode maxNode, ProjectIn
                         continue;
                     }
                     final Decision allocationDecision = deciders.canAllocate(shard, minNode.getRoutingNode(), allocation);
-                    if (allocationDecision.type() == Type.NO) {
+                    if (allocationDecision.type() == Type.NO || allocationDecision.type() == Type.NOT_PREFERRED) {
                         continue;
                     }
 
@@ -1407,7 +1415,7 @@ public boolean containsShard(ShardRouting shard) {
     public static final class NodeSorter extends IntroSorter {
 
         final ModelNode[] modelNodes;
-        /* the nodes weights with respect to the current weight function / index */
+        /** The nodes weights with respect to the current weight function / index */
         final float[] weights;
         private final WeightFunction function;
         private ProjectIndex index;

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java

@@ -503,9 +503,10 @@ private void moveShards() {
 
                 final var routingNode = routingNodes.node(shardRouting.currentNodeId());
                 final var canRemainDecision = allocation.deciders().canRemain(shardRouting, routingNode, allocation);
-                if (canRemainDecision.type() != Decision.Type.NO) {
-                    // it's desired elsewhere but technically it can remain on its current node. Defer its movement until later on to give
-                    // priority to shards that _must_ move.
+                if (canRemainDecision.type() != Decision.Type.NO && canRemainDecision.type() != Decision.Type.NOT_PREFERRED) {
+                    // If movement is throttled, a future reconciliation round will see a resolution. For now, leave it alone.
+                    // Reconciliation treats canRemain NOT_PREFERRED answers as YES because the DesiredBalance computation already decided
+                    // how to handle the situation.
                     continue;
                 }
 
@@ -650,6 +651,7 @@ private DiscoveryNode findRelocationTarget(
             Set<String> desiredNodeIds,
             BiFunction<ShardRouting, RoutingNode, Decision> canAllocateDecider
         ) {
+            DiscoveryNode chosenNode = null;
             for (final var nodeId : desiredNodeIds) {
                 // TODO consider ignored nodes here too?
                 if (nodeId.equals(shardRouting.currentNodeId())) {
@@ -661,12 +663,24 @@ private DiscoveryNode findRelocationTarget(
                 }
                 final var decision = canAllocateDecider.apply(shardRouting, node);
                 logger.trace("relocate {} to {}: {}", shardRouting, nodeId, decision);
+
+                // Assign shards to the YES nodes first. This way we might delay moving shards to NOT_PREFERRED nodes until after shards are
+                // first moved away. The DesiredBalance could be moving shards away from a hot node as well as moving shards to it, and it's
+                // better to offload shards first.
                 if (decision.type() == Decision.Type.YES) {
-                    return node.node();
+                    chosenNode = node.node();
+                    // As soon as we get any YES, we return it.
+                    break;
+                } else if (decision.type() == Decision.Type.NOT_PREFERRED && chosenNode == null) {
+                    // If the best answer is not-preferred, then the shard will still be assigned. It is okay to assign to a not-preferred
+                    // node because the desired balance computation had a reason to override it: when there aren't any better nodes to
+                    // choose and the shard cannot remain where it is, we accept not-preferred. NOT_PREFERRED is essentially a YES for
+                    // reconciliation.
+                    chosenNode = node.node();
                 }
             }
 
-            return null;
+            return chosenNode;
         }
 
         private Decision decideCanAllocate(ShardRouting shardRouting, RoutingNode target) {

server/src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/DesiredBalanceReconciler.java

@@ -73,10 +73,11 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
                 shardRouting.shardId()
             );
             logger.debug(explain);
-            return Decision.single(Decision.Type.NO, NAME, explain);
+            return Decision.single(Decision.Type.NOT_PREFERRED, NAME, explain);
         }
 
-        if (calculateShardMovementChange(nodeWriteThreadPoolStats, shardWriteLoad) >= nodeWriteThreadPoolLoadThreshold) {
+        var newWriteThreadPoolUtilization = calculateShardMovementChange(nodeWriteThreadPoolStats, shardWriteLoad);
+        if (newWriteThreadPoolUtilization >= nodeWriteThreadPoolLoadThreshold) {
             // The node's write thread pool usage would be raised above the high utilization threshold with assignment of the new shard.
             // This could lead to a hot spot on this node and is undesirable.
             String explain = Strings.format(
@@ -92,10 +93,22 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, Routing
                 nodeWriteThreadPoolStats.totalThreadPoolThreads()
             );
             logger.debug(explain);
-            return Decision.single(Decision.Type.NO, NAME, explain);
+            return Decision.single(Decision.Type.NOT_PREFERRED, NAME, explain);
         }
 
-        return Decision.YES;
+        String explanation = Strings.format(
+            "Shard [%s] in index [%s] can be assigned to node [%s]. The node's utilization would become [%s]",
+            shardRouting.shardId(),
+            shardRouting.index(),
+            node.nodeId(),
+            newWriteThreadPoolUtilization
+        );
+
+        if (logger.isTraceEnabled()) {
+            logger.trace(explanation);
+        }
+
+        return allocation.decision(Decision.YES, NAME, explanation);
     }
 
     @Override

server/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/WriteLoadConstraintDecider.java

@@ -104,7 +104,7 @@ public void testWriteLoadDeciderCanAllocate() {
         );
         assertEquals(
             "Assigning a new shard to a node that is above the threshold should fail",
-            Decision.Type.NO,
+            Decision.Type.NOT_PREFERRED,
             writeLoadDecider.canAllocate(
                 testHarness.shardRouting2,
                 testHarness.exceedingThresholdRoutingNode,
@@ -128,7 +128,7 @@ public void testWriteLoadDeciderCanAllocate() {
         );
         assertEquals(
             "Assigning a new shard that would cause the node to exceed capacity should fail",
-            Decision.Type.NO,
+            Decision.Type.NOT_PREFERRED,
             writeLoadDecider.canAllocate(testHarness.shardRouting1, testHarness.nearThresholdRoutingNode, testHarness.routingAllocation)
                 .type()
         );