Primary shard allocator observes limits in forcing allocation (#19811)

Primary shard allocation observes limits in forcing allocation

Previously, during primary shards allocation of shards
with prior allocation IDs, if all nodes returned a
NO decision for allocation (e.g. the settings blocked
allocation on that node), we would chose one of those
nodes and force the primary shard to be allocated to it.

However, this meant that primary shard allocation
would not adhere to the decision of the MaxRetryAllocationDecider,
which would lead to attempting to allocate a shard
which has failed N number of times already (presumably
due to some configuration issue).

This commit solves this issue by introducing the
notion of force allocating a primary shard to a node
and each decider implementation must implement whether
this is allowed or not. In the case of MaxRetryAllocationDecider,
it just forwards the request to canAllocate.

Closes #19446

Author: Ali Beyad

core/src/main/java/org/elasticsearch/cluster/routing/allocation/RoutingAllocation.java

@@ -252,6 +252,17 @@ public void addIgnoreShardForNode(ShardId shardId, String nodeId) {
         nodes.add(nodeId);
     }
 
+    /**
+     * Returns whether the given node id should be ignored from consideration when {@link AllocationDeciders}
+     * is deciding whether to allocate the specified shard id to that node.  The node will be ignored if
+     * the specified shard failed on that node, triggering the current round of allocation.  Since the shard
+     * just failed on that node, we don't want to try to reassign it there, if the node is still a part
+     * of the cluster.
+     *
+     * @param shardId the shard id to be allocated
+     * @param nodeId the node id to check against
+     * @return true if the node id should be ignored in allocation decisions, false otherwise
+     */
     public boolean shouldIgnoreShardForNode(ShardId shardId, String nodeId) {
         if (ignoredShardToNodes == null) {
             return false;

core/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/AllocationDecider.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.ShardRouting;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
+import org.elasticsearch.cluster.routing.allocation.decider.Decision.Type;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Settings;
 
@@ -98,4 +99,30 @@ public Decision canAllocate(RoutingNode node, RoutingAllocation allocation) {
     public Decision canRebalance(RoutingAllocation allocation) {
         return Decision.ALWAYS;
     }
+
+    /**
+     * Returns a {@link Decision} whether the given primary shard can be
+     * forcibly allocated on the given node. This method should only be called
+     * for unassigned primary shards where the node has a shard copy on disk.
+     *
+     * Note: all implementations that override this behavior should take into account
+     * the results of {@link #canAllocate(ShardRouting, RoutingNode, RoutingAllocation)}
+     * before making a decision on force allocation, because force allocation should only
+     * be considered if all deciders return {@link Decision#NO}.
+     */
+    public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
+        assert shardRouting.primary() : "must not call canForceAllocatePrimary on a non-primary shard " + shardRouting;
+        assert shardRouting.unassigned() : "must not call canForceAllocatePrimary on an assigned shard " + shardRouting;
+        Decision decision = canAllocate(shardRouting, node, allocation);
+        if (decision.type() == Type.NO) {
+            // On a NO decision, by default, we allow force allocating the primary.
+            return allocation.decision(Decision.YES,
+                                       decision.label(),
+                                       "primary shard [{}] allowed to force allocate on node [{}]",
+                                       shardRouting.shardId(), node.nodeId());
+        } else {
+            // On a THROTTLE/YES decision, we use the same decision instead of forcing allocation
+            return decision;
+        }
+    }
 }

core/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/AllocationDeciders.java

@@ -196,4 +196,32 @@ public Decision canRebalance(RoutingAllocation allocation) {
         }
         return ret;
     }
+
+    @Override
+    public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
+        assert shardRouting.primary() : "must not call canForceAllocatePrimary on a non-primary shard routing " + shardRouting;
+
+        if (allocation.shouldIgnoreShardForNode(shardRouting.shardId(), node.nodeId())) {
+            return Decision.NO;
+        }
+        Decision.Multi ret = new Decision.Multi();
+        for (AllocationDecider decider : allocations) {
+            Decision decision = decider.canForceAllocatePrimary(shardRouting, node, allocation);
+            // short track if a NO is returned.
+            if (decision == Decision.NO) {
+                if (logger.isTraceEnabled()) {
+                    logger.trace("Shard [{}] can not be forcefully allocated to node [{}] due to [{}].",
+                        shardRouting.shardId(), node.nodeId(), decider.getClass().getSimpleName());
+                }
+                if (!allocation.debugDecision()) {
+                    return decision;
+                } else {
+                    ret.add(decision);
+                }
+            } else if (decision != Decision.ALWAYS) {
+                ret.add(decision);
+            }
+        }
+        return ret;
+    }
 }

core/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/Decision.java

@@ -53,7 +53,7 @@ public abstract class Decision implements ToXContent {
      * @param explanationParams additional parameters for the decision
      * @return new {@link Decision} instance
      */
-    public static Decision single(Type type, String label, String explanation, Object... explanationParams) {
+    public static Decision single(Type type, @Nullable String label, @Nullable String explanation, @Nullable Object... explanationParams) {
         return new Single(type, label, explanation, explanationParams);
     }
 
@@ -146,6 +146,9 @@ public static void writeTo(Type type, StreamOutput out) throws IOException {
      */
     public abstract Type type();
 
+    /**
+     * Get the description label for this decision.
+     */
     @Nullable
     public abstract String label();
 

core/src/main/java/org/elasticsearch/cluster/routing/allocation/decider/MaxRetryAllocationDecider.java

@@ -80,4 +80,12 @@ public Decision canAllocate(ShardRouting shardRouting, RoutingAllocation allocat
     public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
         return canAllocate(shardRouting, allocation);
     }
+
+    @Override
+    public Decision canForceAllocatePrimary(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
+        assert shardRouting.primary() : "must not call canForceAllocatePrimary on a non-primary shard " + shardRouting;
+        // check if we have passed the maximum retry threshold through canAllocate,
+        // if so, we don't want to force the primary allocation here
+        return canAllocate(shardRouting, node, allocation);
+    }
 }

core/src/main/java/org/elasticsearch/gateway/PrimaryShardAllocator.java

@@ -27,10 +27,12 @@
 import org.elasticsearch.cluster.routing.RoutingNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
+import org.elasticsearch.cluster.routing.UnassignedInfo;
 import org.elasticsearch.cluster.routing.UnassignedInfo.AllocationStatus;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.decider.Decision;
 import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.logging.ESLogger;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
@@ -47,8 +49,18 @@
 import java.util.stream.Collectors;
 
 /**
- * The primary shard allocator allocates primary shard that were not created as
- * a result of an API to a node that held them last to be recovered.
+ * The primary shard allocator allocates unassigned primary shards to nodes that hold
+ * valid copies of the unassigned primaries.  It does this by iterating over all unassigned
+ * primary shards in the routing table and fetching shard metadata from each node in the cluster
+ * that holds a copy of the shard.  The shard metadata from each node is compared against the
+ * set of valid allocation IDs and for all valid shard copies (if any), the primary shard allocator
+ * executes the allocation deciders to chose a copy to assign the primary shard to.
+ *
+ * Note that the PrimaryShardAllocator does *not* allocate primaries on index creation
+ * (see {@link org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator}),
+ * nor does it allocate primaries when a primary shard failed and there is a valid replica
+ * copy that can immediately be promoted to primary, as this takes place in
+ * {@link RoutingNodes#failShard(ESLogger, ShardRouting, UnassignedInfo, IndexMetaData)}.
  */
 public abstract class PrimaryShardAllocator extends AbstractComponent {
 
@@ -154,17 +166,35 @@ public boolean allocateUnassigned(RoutingAllocation allocation) {
                 continue;
             }
 
-            final NodesToAllocate nodesToAllocate = buildNodesToAllocate(shard, allocation, nodeShardsResult.orderedAllocationCandidates);
+            final NodesToAllocate nodesToAllocate = buildNodesToAllocate(
+                allocation, nodeShardsResult.orderedAllocationCandidates, shard, false
+            );
             if (nodesToAllocate.yesNodeShards.isEmpty() == false) {
                 NodeGatewayStartedShards nodeShardState = nodesToAllocate.yesNodeShards.get(0);
                 logger.debug("[{}][{}]: allocating [{}] to [{}] on primary allocation", shard.index(), shard.id(), shard, nodeShardState.getNode());
                 changed = true;
                 unassignedIterator.initialize(nodeShardState.getNode().getId(), nodeShardState.allocationId(), ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
             } else if (nodesToAllocate.throttleNodeShards.isEmpty() == true && nodesToAllocate.noNodeShards.isEmpty() == false) {
-                NodeGatewayStartedShards nodeShardState = nodesToAllocate.noNodeShards.get(0);
-                logger.debug("[{}][{}]: forcing allocating [{}] to [{}] on primary allocation", shard.index(), shard.id(), shard, nodeShardState.getNode());
-                changed = true;
-                unassignedIterator.initialize(nodeShardState.getNode().getId(), nodeShardState.allocationId(), ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
+                // The deciders returned a NO decision for all nodes with shard copies, so we check if primary shard
+                // can be force-allocated to one of the nodes.
+                final NodesToAllocate nodesToForceAllocate = buildNodesToAllocate(
+                    allocation, nodeShardsResult.orderedAllocationCandidates, shard, true
+                );
+                if (nodesToForceAllocate.yesNodeShards.isEmpty() == false) {
+                    NodeGatewayStartedShards nodeShardState = nodesToForceAllocate.yesNodeShards.get(0);
+                    logger.debug("[{}][{}]: allocating [{}] to [{}] on forced primary allocation",
+                                 shard.index(), shard.id(), shard, nodeShardState.getNode());
+                    changed = true;
+                    unassignedIterator.initialize(nodeShardState.getNode().getId(), nodeShardState.allocationId(),
+                                                  ShardRouting.UNAVAILABLE_EXPECTED_SHARD_SIZE);
+                } else if (nodesToForceAllocate.throttleNodeShards.isEmpty() == false) {
+                    logger.debug("[{}][{}]: throttling allocation [{}] to [{}] on forced primary allocation",
+                                 shard.index(), shard.id(), shard, nodesToForceAllocate.throttleNodeShards);
+                    changed |= unassignedIterator.removeAndIgnore(AllocationStatus.DECIDERS_THROTTLED);
+                } else {
+                    logger.debug("[{}][{}]: forced primary allocation denied [{}]", shard.index(), shard.id(), shard);
+                    changed |= unassignedIterator.removeAndIgnore(AllocationStatus.DECIDERS_NO);
+                }
             } else {
                 // we are throttling this, but we have enough to allocate to this node, ignore it for now
                 logger.debug("[{}][{}]: throttling allocation [{}] to [{}] on primary allocation", shard.index(), shard.id(), shard, nodesToAllocate.throttleNodeShards);
@@ -268,7 +298,10 @@ private boolean isEnoughVersionBasedAllocationsFound(IndexMetaData indexMetaData
     /**
      * Split the list of node shard states into groups yes/no/throttle based on allocation deciders
      */
-    private NodesToAllocate buildNodesToAllocate(ShardRouting shard, RoutingAllocation allocation, List<NodeGatewayStartedShards> nodeShardStates) {
+    private NodesToAllocate buildNodesToAllocate(RoutingAllocation allocation,
+                                                 List<NodeGatewayStartedShards> nodeShardStates,
+                                                 ShardRouting shardRouting,
+                                                 boolean forceAllocate) {
         List<NodeGatewayStartedShards> yesNodeShards = new ArrayList<>();
         List<NodeGatewayStartedShards> throttledNodeShards = new ArrayList<>();
         List<NodeGatewayStartedShards> noNodeShards = new ArrayList<>();
@@ -278,7 +311,8 @@ private NodesToAllocate buildNodesToAllocate(ShardRouting shard, RoutingAllocati
                 continue;
             }
 
-            Decision decision = allocation.deciders().canAllocate(shard, node, allocation);
+            Decision decision = forceAllocate ? allocation.deciders().canForceAllocatePrimary(shardRouting, node, allocation) :
+                                                allocation.deciders().canAllocate(shardRouting, node, allocation);
             if (decision.type() == Decision.Type.THROTTLE) {
                 throttledNodeShards.add(nodeShardState);
             } else if (decision.type() == Decision.Type.NO) {

core/src/test/java/org/elasticsearch/cluster/routing/PrimaryAllocationIT.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.action.admin.indices.shards.IndicesShardStoresResponse;
 import org.elasticsearch.action.support.ActiveShardCount;
 import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
 import org.elasticsearch.cluster.routing.allocation.command.AllocateEmptyPrimaryAllocationCommand;
 import org.elasticsearch.cluster.routing.allocation.command.AllocateStalePrimaryAllocationCommand;
 import org.elasticsearch.common.collect.ImmutableOpenIntMap;
@@ -219,4 +220,32 @@ public void testNotWaitForQuorumCopies() throws Exception {
         ensureYellow("test");
         assertHitCount(client().prepareSearch().setSize(0).setQuery(matchAllQuery()).get(), 1L);
     }
+
+    /**
+     * This test ensures that for an unassigned primary shard that has a valid shard copy on at least one node,
+     * we will force allocate the primary shard to one of those nodes, even if the allocation deciders all return
+     * a NO decision to allocate.
+     */
+    public void testForceAllocatePrimaryOnNoDecision() throws Exception {
+        logger.info("--> starting 1 node");
+        final String node = internalCluster().startNodeAsync().get();
+        logger.info("--> creating index with 1 primary and 0 replicas");
+        final String indexName = "test-idx";
+        assertAcked(client().admin().indices()
+                        .prepareCreate(indexName)
+                        .setSettings(Settings.builder().put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                                         .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0))
+                        .get());
+        logger.info("--> update the settings to prevent allocation to the data node");
+        assertTrue(client().admin().indices().prepareUpdateSettings(indexName)
+                       .setSettings(Settings.builder().put(IndexMetaData.INDEX_ROUTING_EXCLUDE_GROUP_SETTING.getKey() + "_name", node))
+                       .get()
+                       .isAcknowledged());
+        logger.info("--> full cluster restart");
+        internalCluster().fullRestart();
+        logger.info("--> checking that the primary shard is force allocated to the data node despite being blocked by the exclude filter");
+        ensureGreen(indexName);
+        assertEquals(1, client().admin().cluster().prepareState().get().getState()
+                            .routingTable().index(indexName).shardsWithState(ShardRoutingState.STARTED).size());
+    }
 }