Auto-allocate searchable snapshots (#52527)

This commit allows plugins to supply their own allocator for existing shards,
generalizing the default `GatewayAllocator`. It uses this to implement a very
simple auto-allocation process for searchable snapshots which respects the
allocation deciders but which can otherwise be assigned to any node.

Author: DaveCTurner

benchmarks/src/main/java/org/elasticsearch/benchmark/routing/allocation/Allocators.java

@@ -47,17 +47,21 @@ private static class NoopGatewayAllocator extends GatewayAllocator {
         public static final NoopGatewayAllocator INSTANCE = new NoopGatewayAllocator();
 
         @Override
-        public void applyStartedShards(RoutingAllocation allocation, List<ShardRouting> startedShards) {
+        public void applyStartedShards(List<ShardRouting> startedShards, RoutingAllocation allocation) {
             // noop
         }
 
         @Override
-        public void applyFailedShards(RoutingAllocation allocation, List<FailedShard> failedShards) {
+        public void applyFailedShards(List<FailedShard> failedShards, RoutingAllocation allocation) {
             // noop
         }
 
         @Override
-        public void allocateUnassigned(RoutingAllocation allocation) {
+        public void allocateUnassigned(
+            ShardRouting shardRouting,
+            RoutingAllocation allocation,
+            UnassignedAllocationHandler unassignedAllocationHandler
+        ) {
             // noop
         }
     }

server/src/main/java/org/elasticsearch/action/admin/cluster/allocation/TransportClusterAllocationExplainAction.java

@@ -33,8 +33,7 @@
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.RoutingNodes;
 import org.elasticsearch.cluster.routing.ShardRouting;
-import org.elasticsearch.cluster.routing.allocation.AllocateUnassignedDecision;
-import org.elasticsearch.cluster.routing.allocation.MoveDecision;
+import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation.DebugMode;
 import org.elasticsearch.cluster.routing.allocation.ShardAllocationDecision;
@@ -43,7 +42,6 @@
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.gateway.GatewayAllocator;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -63,20 +61,20 @@ public class TransportClusterAllocationExplainAction
     private final ClusterInfoService clusterInfoService;
     private final AllocationDeciders allocationDeciders;
     private final ShardsAllocator shardAllocator;
-    private final GatewayAllocator gatewayAllocator;
+    private final AllocationService allocationService;
 
     @Inject
     public TransportClusterAllocationExplainAction(TransportService transportService, ClusterService clusterService,
                                                    ThreadPool threadPool, ActionFilters actionFilters,
                                                    IndexNameExpressionResolver indexNameExpressionResolver,
                                                    ClusterInfoService clusterInfoService, AllocationDeciders allocationDeciders,
-                                                   ShardsAllocator shardAllocator, GatewayAllocator gatewayAllocator) {
+                                                   ShardsAllocator shardAllocator, AllocationService allocationService) {
         super(ClusterAllocationExplainAction.NAME, transportService, clusterService, threadPool, actionFilters,
             ClusterAllocationExplainRequest::new, indexNameExpressionResolver);
         this.clusterInfoService = clusterInfoService;
         this.allocationDeciders = allocationDeciders;
         this.shardAllocator = shardAllocator;
-        this.gatewayAllocator = gatewayAllocator;
+        this.allocationService = allocationService;
     }
 
     @Override
@@ -106,27 +104,21 @@ protected void masterOperation(Task task, final ClusterAllocationExplainRequest
         logger.debug("explaining the allocation for [{}], found shard [{}]", request, shardRouting);
 
         ClusterAllocationExplanation cae = explainShard(shardRouting, allocation,
-            request.includeDiskInfo() ? clusterInfo : null, request.includeYesDecisions(), gatewayAllocator, shardAllocator);
+            request.includeDiskInfo() ? clusterInfo : null, request.includeYesDecisions(), allocationService);
         listener.onResponse(new ClusterAllocationExplainResponse(cae));
     }
 
     // public for testing
     public static ClusterAllocationExplanation explainShard(ShardRouting shardRouting, RoutingAllocation allocation,
                                                             ClusterInfo clusterInfo, boolean includeYesDecisions,
-                                                            GatewayAllocator gatewayAllocator, ShardsAllocator shardAllocator) {
+                                                            AllocationService allocationService) {
         allocation.setDebugMode(includeYesDecisions ? DebugMode.ON : DebugMode.EXCLUDE_YES_DECISIONS);
 
         ShardAllocationDecision shardDecision;
         if (shardRouting.initializing() || shardRouting.relocating()) {
             shardDecision = ShardAllocationDecision.NOT_TAKEN;
         } else {
-            AllocateUnassignedDecision allocateDecision = shardRouting.unassigned() ?
-                gatewayAllocator.decideUnassignedShardAllocation(shardRouting, allocation) : AllocateUnassignedDecision.NOT_TAKEN;
-            if (allocateDecision.isDecisionTaken() == false) {
-                shardDecision = shardAllocator.decideShardAllocation(shardRouting, allocation);
-            } else {
-                shardDecision = new ShardAllocationDecision(allocateDecision, MoveDecision.NOT_TAKEN);
-            }
+            shardDecision = allocationService.explainShardAllocation(shardRouting, allocation);
         }
 
         return new ClusterAllocationExplanation(shardRouting,

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

@@ -36,12 +36,12 @@
 import org.elasticsearch.cluster.health.ClusterHealthStatus;
 import org.elasticsearch.cluster.metadata.IndexNameExpressionResolver;
 import org.elasticsearch.cluster.routing.UnassignedInfo;
+import org.elasticsearch.cluster.routing.allocation.AllocationService;
 import org.elasticsearch.cluster.service.ClusterService;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.gateway.GatewayAllocator;
 import org.elasticsearch.index.IndexNotFoundException;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.threadpool.ThreadPool;
@@ -55,15 +55,15 @@ public class TransportClusterHealthAction extends TransportMasterNodeReadAction<
 
     private static final Logger logger = LogManager.getLogger(TransportClusterHealthAction.class);
 
-    private final GatewayAllocator gatewayAllocator;
+    private final AllocationService allocationService;
 
     @Inject
     public TransportClusterHealthAction(TransportService transportService, ClusterService clusterService,
                                         ThreadPool threadPool, ActionFilters actionFilters,
-                                        IndexNameExpressionResolver indexNameExpressionResolver, GatewayAllocator gatewayAllocator) {
+                                        IndexNameExpressionResolver indexNameExpressionResolver, AllocationService allocationService) {
         super(ClusterHealthAction.NAME, false, transportService, clusterService, threadPool, actionFilters,
             ClusterHealthRequest::new, indexNameExpressionResolver);
-        this.gatewayAllocator = gatewayAllocator;
+        this.allocationService = allocationService;
     }
 
     @Override
@@ -229,14 +229,14 @@ private static int getWaitCount(ClusterHealthRequest request) {
 
     private boolean validateRequest(final ClusterHealthRequest request, ClusterState clusterState, final int waitCount) {
         ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.getMasterService().numberOfPendingTasks(),
-                gatewayAllocator.getNumberOfInFlightFetch(), clusterService.getMasterService().getMaxTaskWaitTime());
+            allocationService.getNumberOfInFlightFetches(), clusterService.getMasterService().getMaxTaskWaitTime());
         return prepareResponse(request, response, clusterState, indexNameExpressionResolver) == waitCount;
     }
 
     private ClusterHealthResponse getResponse(final ClusterHealthRequest request, ClusterState clusterState,
                                               final int waitFor, boolean timedOut) {
         ClusterHealthResponse response = clusterHealth(request, clusterState, clusterService.getMasterService().numberOfPendingTasks(),
-                gatewayAllocator.getNumberOfInFlightFetch(), clusterService.getMasterService().getMaxTaskWaitTime());
+            allocationService.getNumberOfInFlightFetches(), clusterService.getMasterService().getMaxTaskWaitTime());
         int readyCounter = prepareResponse(request, response, clusterState, indexNameExpressionResolver);
         boolean valid = (readyCounter == waitFor);
         assert valid || timedOut;

server/src/main/java/org/elasticsearch/cluster/ClusterModule.java

@@ -34,6 +34,7 @@
 import org.elasticsearch.cluster.metadata.RepositoriesMetaData;
 import org.elasticsearch.cluster.routing.DelayedAllocationService;
 import org.elasticsearch.cluster.routing.allocation.AllocationService;
+import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.allocator.ShardsAllocator;
 import org.elasticsearch.cluster.routing.allocation.decider.AllocationDecider;
@@ -97,12 +98,14 @@ public class ClusterModule extends AbstractModule {
     private final IndexNameExpressionResolver indexNameExpressionResolver;
     private final AllocationDeciders allocationDeciders;
     private final AllocationService allocationService;
+    private final List<ClusterPlugin> clusterPlugins;
     // pkg private for tests
     final Collection<AllocationDecider> deciderList;
     final ShardsAllocator shardsAllocator;
 
     public ClusterModule(Settings settings, ClusterService clusterService, List<ClusterPlugin> clusterPlugins,
                          ClusterInfoService clusterInfoService) {
+        this.clusterPlugins = clusterPlugins;
         this.deciderList = createAllocationDeciders(settings, clusterService.getClusterSettings(), clusterPlugins);
         this.allocationDeciders = new AllocationDeciders(deciderList);
         this.shardsAllocator = createShardsAllocator(settings, clusterService.getClusterSettings(), clusterPlugins);
@@ -251,4 +254,22 @@ protected void configure() {
         bind(AllocationDeciders.class).toInstance(allocationDeciders);
         bind(ShardsAllocator.class).toInstance(shardsAllocator);
     }
+
+    public void setExistingShardsAllocators(GatewayAllocator gatewayAllocator) {
+        final Map<String, ExistingShardsAllocator> existingShardsAllocators = new HashMap<>();
+        existingShardsAllocators.put(GatewayAllocator.ALLOCATOR_NAME, gatewayAllocator);
+
+        for (ClusterPlugin clusterPlugin : clusterPlugins) {
+            for (Map.Entry<String, ExistingShardsAllocator> existingShardsAllocatorEntry
+                : clusterPlugin.getExistingShardsAllocators().entrySet()) {
+                final String allocatorName = existingShardsAllocatorEntry.getKey();
+                if (existingShardsAllocators.put(allocatorName, existingShardsAllocatorEntry.getValue()) != null) {
+                    throw new IllegalArgumentException("ExistingShardsAllocator [" + allocatorName + "] from [" +
+                        clusterPlugin.getClass().getName() + "] was already defined");
+                }
+            }
+        }
+        allocationService.setExistingShardsAllocators(existingShardsAllocators);
+    }
+
 }

server/src/main/java/org/elasticsearch/cluster/routing/RoutingNodes.java

@@ -29,6 +29,7 @@
 import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.routing.UnassignedInfo.AllocationStatus;
+import org.elasticsearch.cluster.routing.allocation.ExistingShardsAllocator;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.Randomness;
 import org.elasticsearch.common.collect.Tuple;
@@ -883,7 +884,7 @@ public void ignoreShard(ShardRouting shard, AllocationStatus allocationStatus, R
             ignored.add(shard);
         }
 
-        public class UnassignedIterator implements Iterator<ShardRouting> {
+        public class UnassignedIterator implements Iterator<ShardRouting>, ExistingShardsAllocator.UnassignedAllocationHandler {
 
             private final ListIterator<ShardRouting> iterator;
             private ShardRouting current;
@@ -907,6 +908,7 @@ public ShardRouting next() {
              *
              * @param existingAllocationId allocation id to use. If null, a fresh allocation id is generated.
              */
+            @Override
             public ShardRouting initialize(String nodeId, @Nullable String existingAllocationId, long expectedShardSize,
                                            RoutingChangesObserver routingChangesObserver) {
                 nodes.ensureMutable();
@@ -922,6 +924,7 @@ public ShardRouting initialize(String nodeId, @Nullable String existingAllocatio
              *
              * @param attempt the result of the allocation attempt
              */
+            @Override
             public void removeAndIgnore(AllocationStatus attempt, RoutingChangesObserver changes) {
                 nodes.ensureMutable();
                 innerRemove();
@@ -940,6 +943,7 @@ private void updateShardRouting(ShardRouting shardRouting) {
              * @param  recoverySource the new recovery source to use
              * @return the shard with unassigned info updated
              */
+            @Override
             public ShardRouting updateUnassigned(UnassignedInfo unassignedInfo, RecoverySource recoverySource,
                                                  RoutingChangesObserver changes) {
                 nodes.ensureMutable();