Async fetch of shard started and store during allocation

Today, when a primary shard is not allocated we go to all the nodes to find where it is allocated (listing its started state). When we allocate a replica shard, we head to all the nodes and list its store to allocate the replica on a node that holds the closest matching index files to the primary.

Those two operations today execute synchronously within the GatewayAllocator, which means they execute in a sync manner within the cluster update thread. For large clusters, or environments with very slow disk, those operations will stall the cluster update thread, making it seem like its stuck.

Worse, if the FS is really slow, we timeout after 30s the operation (to not stall the cluster update thread completely). This means that we will have another run for the primary shard if we didn't find one, or we won't find the best node to place a shard since it might have timed out (listing stores need to list all files and read the checksum at the end of each file).

On top of that, this sync operation happen one shard at a time, so its effectively compounding the problem in a serial manner the more shards we have and the slower FS is...

This change moves to perform both listing the shard started states and the shard stores to an async manner. During the allocation by the GatewayAllocator, if data needs to be fetched from a node, it is done in an async fashion, with the response triggering a reroute to make sure the results will be taken into account. Also, if there are on going operations happening, the relevant shard data will not be taken into account until all the ongoing listing operations are done executing.

The execution of listing shard states and stores has been moved to their own respective thread pools (scaling, so will go down to 0 when not needed anymore, unbounded queue, since we don't want to timeout, just let it execute based on how fast the local FS is). This is needed sine we are going to blast nodes with a lot of requests and we need to make sure there is no thread explosion.

This change also improves the handling of shard failures coming from a specific node. Today, those nodes were ignored from allocation only for the single reroute round. Now, since fetching is async, we need to keep those failures around at least until a single successful fetch without the node is done, to make sure not to repeat allocating to the failed node all the time.

Note, if before the indication of slow allocation was high pending tasks since the allocator was waiting for responses, not the pending tasks will be much smaller. In order to still indicate that the cluster is in the middle of fetching shard data, 2 attributes were added to the cluster health API, indicating the number of ongoing fetches of both started shards and shard store.

closes elastic#9502
closes elastic#11101

Author: kimchy

src/main/java/org/elasticsearch/action/admin/cluster/health/ClusterHealthResponse.java

@@ -58,6 +58,7 @@ public class ClusterHealthResponse extends ActionResponse implements Iterable<Cl
     int initializingShards = 0;
     int unassignedShards = 0;
     int numberOfPendingTasks = 0;
+    int numberOfInFlightFetch = 0;
     boolean timedOut = false;
     ClusterHealthStatus status = ClusterHealthStatus.RED;
     private List<String> validationFailures;
@@ -68,12 +69,14 @@ public class ClusterHealthResponse extends ActionResponse implements Iterable<Cl
 
     /** needed for plugins BWC */
     public ClusterHealthResponse(String clusterName, String[] concreteIndices, ClusterState clusterState) {
-        this(clusterName, concreteIndices, clusterState, -1);
+        this(clusterName, concreteIndices, clusterState, -1, -1);
     }
 
-    public ClusterHealthResponse(String clusterName, String[] concreteIndices, ClusterState clusterState, int numberOfPendingTasks) {
+    public ClusterHealthResponse(String clusterName, String[] concreteIndices, ClusterState clusterState, int numberOfPendingTasks,
+                                 int numberOfInFlightFetch) {
         this.clusterName = clusterName;
         this.numberOfPendingTasks = numberOfPendingTasks;
+        this.numberOfInFlightFetch = numberOfInFlightFetch;
         RoutingTableValidation validation = clusterState.routingTable().validate(clusterState.metaData());
         validationFailures = validation.failures();
         numberOfNodes = clusterState.nodes().size();
@@ -167,6 +170,10 @@ public int getNumberOfPendingTasks() {
         return this.numberOfPendingTasks;
     }
 
+    public int getNumberOfInFlightFetch() {
+        return this.numberOfInFlightFetch;
+    }
+
     /**
      * <tt>true</tt> if the waitForXXX has timeout out and did not match.
      */
@@ -225,6 +232,10 @@ public void readFrom(StreamInput in) throws IOException {
                 validationFailures.add(in.readString());
             }
         }
+
+        if (in.getVersion().onOrAfter(Version.V_1_6_0)) {
+            numberOfInFlightFetch = in.readInt();
+        }
     }
 
     @Override
@@ -252,6 +263,10 @@ public void writeTo(StreamOutput out) throws IOException {
         for (String failure : validationFailures) {
             out.writeString(failure);
         }
+
+        if (out.getVersion().onOrAfter(Version.V_1_6_0)) {
+            out.writeInt(numberOfInFlightFetch);
+        }
     }
 
 
@@ -275,6 +290,7 @@ static final class Fields {
         static final XContentBuilderString NUMBER_OF_NODES = new XContentBuilderString("number_of_nodes");
         static final XContentBuilderString NUMBER_OF_DATA_NODES = new XContentBuilderString("number_of_data_nodes");
         static final XContentBuilderString NUMBER_OF_PENDING_TASKS = new XContentBuilderString("number_of_pending_tasks");
+        static final XContentBuilderString NUMBER_OF_IN_FLIGHT_FETCH = new XContentBuilderString("number_of_in_flight_fetch");
         static final XContentBuilderString ACTIVE_PRIMARY_SHARDS = new XContentBuilderString("active_primary_shards");
         static final XContentBuilderString ACTIVE_SHARDS = new XContentBuilderString("active_shards");
         static final XContentBuilderString RELOCATING_SHARDS = new XContentBuilderString("relocating_shards");
@@ -297,6 +313,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         builder.field(Fields.INITIALIZING_SHARDS, getInitializingShards());
         builder.field(Fields.UNASSIGNED_SHARDS, getUnassignedShards());
         builder.field(Fields.NUMBER_OF_PENDING_TASKS, getNumberOfPendingTasks());
+        builder.field(Fields.NUMBER_OF_IN_FLIGHT_FETCH, getNumberOfInFlightFetch());
 
         String level = params.param("level", "cluster");
         boolean outputIndices = "indices".equals(level) || "shards".equals(level);

src/main/java/org/elasticsearch/action/admin/cluster/health/TransportClusterHealthAction.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.gateway.local.LocalGatewayAllocator;
 import org.elasticsearch.indices.IndexMissingException;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -41,12 +42,14 @@
 public class TransportClusterHealthAction extends TransportMasterNodeReadOperationAction<ClusterHealthRequest, ClusterHealthResponse> {
 
     private final ClusterName clusterName;
+    private final LocalGatewayAllocator gatewayAllocator;
 
     @Inject
     public TransportClusterHealthAction(Settings settings, TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
-                                        ClusterName clusterName, ActionFilters actionFilters) {
+                                        ClusterName clusterName, ActionFilters actionFilters, LocalGatewayAllocator gatewayAllocator) {
         super(settings, ClusterHealthAction.NAME, transportService, clusterService, threadPool, actionFilters);
         this.clusterName = clusterName;
+        this.gatewayAllocator = gatewayAllocator;
     }
 
     @Override
@@ -164,12 +167,12 @@ public void onTimeout(TimeValue timeout) {
     }
 
     private boolean validateRequest(final ClusterHealthRequest request, ClusterState clusterState, final int waitFor) {
-        ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.numberOfPendingTasks());
+        ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.numberOfPendingTasks(), gatewayAllocator.getNumberOfInFlightFetch());
         return prepareResponse(request, response, clusterState, waitFor);
     }
 
     private ClusterHealthResponse getResponse(final ClusterHealthRequest request, ClusterState clusterState, final int waitFor, boolean timedOut) {
-        ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.numberOfPendingTasks());
+        ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.numberOfPendingTasks(), gatewayAllocator.getNumberOfInFlightFetch());
         boolean valid = prepareResponse(request, response, clusterState, waitFor);
         assert valid || timedOut;
         // we check for a timeout here since this method might be called from the wait_for_events
@@ -253,7 +256,7 @@ private boolean prepareResponse(final ClusterHealthRequest request, final Cluste
     }
 
 
-    private ClusterHealthResponse clusterHealth(ClusterHealthRequest request, ClusterState clusterState, int numberOfPendingTasks) {
+    private ClusterHealthResponse clusterHealth(ClusterHealthRequest request, ClusterState clusterState, int numberOfPendingTasks, int numberOfInFlightFetch) {
         if (logger.isTraceEnabled()) {
             logger.trace("Calculating health based on state version [{}]", clusterState.version());
         }
@@ -262,11 +265,11 @@ private ClusterHealthResponse clusterHealth(ClusterHealthRequest request, Cluste
             concreteIndices = clusterState.metaData().concreteIndices(request.indicesOptions(), request.indices());
         } catch (IndexMissingException e) {
             // one of the specified indices is not there - treat it as RED.
-            ClusterHealthResponse response = new ClusterHealthResponse(clusterName.value(), Strings.EMPTY_ARRAY, clusterState, numberOfPendingTasks);
+            ClusterHealthResponse response = new ClusterHealthResponse(clusterName.value(), Strings.EMPTY_ARRAY, clusterState, numberOfPendingTasks, numberOfInFlightFetch);
             response.status = ClusterHealthStatus.RED;
             return response;
         }
 
-        return new ClusterHealthResponse(clusterName.value(), concreteIndices, clusterState, numberOfPendingTasks);
+        return new ClusterHealthResponse(clusterName.value(), concreteIndices, clusterState, numberOfPendingTasks, numberOfInFlightFetch);
     }
 }

src/main/java/org/elasticsearch/action/support/nodes/NodesOperationResponse.java

@@ -21,7 +21,9 @@
 
 import com.google.common.collect.Maps;
 import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.FailedNodeException;
 import org.elasticsearch.cluster.ClusterName;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 
@@ -46,6 +48,14 @@ protected NodesOperationResponse(ClusterName clusterName, NodeResponse[] nodes)
         this.nodes = nodes;
     }
 
+    /**
+     * The failed nodes, if set to be captured.
+     */
+    @Nullable
+    public FailedNodeException[] failures() {
+        return null;
+    }
+
     public ClusterName getClusterName() {
         return this.clusterName;
     }

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

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.cluster.routing.allocation;
 
+import com.google.common.collect.ImmutableSet;
 import org.elasticsearch.cluster.ClusterInfo;
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
@@ -217,6 +218,17 @@ public boolean shouldIgnoreShardForNode(ShardId shardId, String nodeId) {
         return nodes != null && nodes.contains(nodeId);
     }
 
+    public Set<String> getIgnoreNodes(ShardId shardId) {
+        if (ignoredShardToNodes == null) {
+            return ImmutableSet.of();
+        }
+        Set<String> ignore = ignoredShardToNodes.get(shardId);
+        if (ignore == null) {
+            return ImmutableSet.of();
+        }
+        return ImmutableSet.copyOf(ignore);
+    }
+
     /**
      * Create a routing decision, including the reason if the debug flag is
      * turned on

src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/ShardsAllocatorModule.java

@@ -21,6 +21,7 @@
 
 import org.elasticsearch.common.inject.AbstractModule;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.gateway.local.LocalGatewayAllocator;
 import org.elasticsearch.gateway.none.NoneGatewayAllocator;
 
 /**
@@ -57,6 +58,7 @@ protected void configure() {
         if (shardsAllocator == null) {
             shardsAllocator = loadShardsAllocator(settings);
         }
+        bind(LocalGatewayAllocator.class).asEagerSingleton();
         bind(GatewayAllocator.class).to(gatewayAllocator).asEagerSingleton();
         bind(ShardsAllocator.class).to(shardsAllocator).asEagerSingleton();
     }