Add cluster-wide shard limit

Adds a safety limit on the number of shards in a cluster, based on
the number of nodes in the cluster. The limit is checked on operations
that add (or activate) shards, such as index creation, snapshot
restoration, and opening closed indices, and can be changed via the
cluster settings API.

Closes elastic#20705

Author: gwbrown

server/src/main/java/org/elasticsearch/cluster/metadata/MetaData.java

@@ -127,6 +127,9 @@ public interface Custom extends NamedDiffable<Custom>, ToXContentFragment, Clust
 
     }
 
+    public static final Setting<Integer> SETTING_CLUSTER_MAX_SHARDS_PER_NODE =
+        Setting.intSetting("cluster.shards.max_per_node", 1000, Property.Dynamic, Property.NodeScope);
+
     public static final Setting<Boolean> SETTING_READ_ONLY_SETTING =
         Setting.boolSetting("cluster.blocks.read_only", false, Property.Dynamic, Property.NodeScope);
 
@@ -162,6 +165,7 @@ public interface Custom extends NamedDiffable<Custom>, ToXContentFragment, Clust
     private final ImmutableOpenMap<String, Custom> customs;
 
     private final transient int totalNumberOfShards; // Transient ? not serializable anyway?
+    private final int totalOpenIndexShards;
     private final int numberOfShards;
 
     private final String[] allIndices;
@@ -183,12 +187,17 @@ public interface Custom extends NamedDiffable<Custom>, ToXContentFragment, Clust
         this.customs = customs;
         this.templates = templates;
         int totalNumberOfShards = 0;
+        int totalOpenIndexShards = 0;
         int numberOfShards = 0;
         for (ObjectCursor<IndexMetaData> cursor : indices.values()) {
             totalNumberOfShards += cursor.value.getTotalNumberOfShards();
             numberOfShards += cursor.value.getNumberOfShards();
+            if (IndexMetaData.State.OPEN.equals(cursor.value.getState())) {
+                totalOpenIndexShards += cursor.value.getTotalNumberOfShards();
+            }
         }
         this.totalNumberOfShards = totalNumberOfShards;
+        this.totalOpenIndexShards = totalOpenIndexShards;
         this.numberOfShards = numberOfShards;
 
         this.allIndices = allIndices;
@@ -661,10 +670,29 @@ public <T extends Custom> T custom(String type) {
     }
 
 
+    /**
+     * Gets the total number of shards from all indices, including replicas and
+     * closed indices.
+     * @return The total number shards from all indices.
+     */
     public int getTotalNumberOfShards() {
         return this.totalNumberOfShards;
     }
 
+    /**
+     * Gets the total number of active shards from all indices. Includes
+     * replicas, but does not include shards that are part of closed indices.
+     * @return The total number of active shards from all indices.
+     */
+    public int getTotalOpenIndexShards() {
+        return this.totalOpenIndexShards;
+    }
+
+    /**
+     * Gets the number of primary shards from all indices, not including
+     * replicas.
+     * @return The number of primary shards from all indices.
+     */
     public int getNumberOfShards() {
         return this.numberOfShards;
     }

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataCreateIndexService.java

@@ -84,6 +84,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
@@ -597,18 +598,38 @@ public void onFailure(String source, Exception e) {
 
     private void validate(CreateIndexClusterStateUpdateRequest request, ClusterState state) {
         validateIndexName(request.index(), state);
-        validateIndexSettings(request.index(), request.settings());
+        validateIndexSettings(request.index(), request.settings(), state);
     }
 
-    public void validateIndexSettings(String indexName, Settings settings) throws IndexCreationException {
+    public void validateIndexSettings(String indexName, Settings settings, ClusterState clusterState) throws IndexCreationException {
         List<String> validationErrors = getIndexSettingsValidationErrors(settings);
+
+        Optional<String> shardAllocation = getShardLimitError(indexName, settings, clusterState);
+        shardAllocation.ifPresent(validationErrors::add);
+
         if (validationErrors.isEmpty() == false) {
             ValidationException validationException = new ValidationException();
             validationException.addValidationErrors(validationErrors);
             throw new IndexCreationException(indexName, validationException);
         }
     }
 
+    private Optional<String> getShardLimitError(String indexName, Settings settings, ClusterState clusterState) {
+        int currentOpenShards = clusterState.getMetaData().getTotalOpenIndexShards();
+        int maxShardsPerNode = clusterService.getClusterSettings().get(MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE);
+        int nodeCount = clusterState.getNodes().getDataNodes().size();
+        long maxShardsInCluster = maxShardsPerNode * nodeCount;
+        int shardsToCreate = IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.get(settings)
+            * (1 + IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.get(settings));
+
+        if ((currentOpenShards + shardsToCreate) > maxShardsInCluster) {
+            return Optional.of("creating [" + indexName + "] "
+                + "would create " + shardsToCreate + " total shards, but this cluster currently has "
+                + currentOpenShards + "/" + maxShardsInCluster + " maximum shards open");
+        }
+        return Optional.empty();
+    }
+
     List<String> getIndexSettingsValidationErrors(Settings settings) {
         String customPath = IndexMetaData.INDEX_DATA_PATH_SETTING.get(settings);
         List<String> validationErrors = new ArrayList<>();

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataIndexStateService.java

@@ -22,6 +22,7 @@
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.admin.indices.close.CloseIndexClusterStateUpdateRequest;
 import org.elasticsearch.action.admin.indices.open.OpenIndexClusterStateUpdateRequest;
 import org.elasticsearch.action.support.ActiveShardsObserver;
@@ -175,6 +176,8 @@ public ClusterState execute(ClusterState currentState) {
                     }
                 }
 
+                validateShardLimit(currentState, request);
+
                 if (indicesToOpen.isEmpty()) {
                     return currentState;
                 }
@@ -217,4 +220,32 @@ public ClusterState execute(ClusterState currentState) {
         });
     }
 
+    private void validateShardLimit(ClusterState currentState, OpenIndexClusterStateUpdateRequest request) {
+        int currentOpenShards = currentState.getMetaData().getTotalOpenIndexShards();
+        int maxShardsInCluster = getMaxAllowedShardCount(currentState);
+        int shardsToOpen = Arrays.stream(request.indices())
+            .mapToInt(index -> getTotalShardCount(currentState, index))
+            .sum();
+        if ((currentOpenShards + shardsToOpen) > maxShardsInCluster) {
+            ActionRequestValidationException exception = new ActionRequestValidationException();
+            String[] indexNames = Arrays.stream(request.indices()).map(Index::getName).toArray(String[]::new);
+            exception.addValidationError("opening " + Arrays.toString(indexNames)
+                + " would open " + shardsToOpen + " total shards, but this cluster currently has "
+                + currentOpenShards + "/" + maxShardsInCluster + " maximum shards open");
+            throw exception;
+        }
+    }
+
+
+    private int getMaxAllowedShardCount(ClusterState state) {
+        int maxShardsPerNode = clusterService.getClusterSettings().get(MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE);
+        int nodeCount = state.getNodes().getDataNodes().size();
+        return maxShardsPerNode * nodeCount;
+    }
+
+    private int getTotalShardCount(ClusterState state, Index index) {
+        IndexMetaData indexMetaData = state.metaData().index(index);
+        return indexMetaData.getNumberOfShards() * (1 + indexMetaData.getNumberOfReplicas());
+    }
+
 }

server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -189,6 +189,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                     MappingUpdatedAction.INDICES_MAPPING_DYNAMIC_TIMEOUT_SETTING,
                     MetaData.SETTING_READ_ONLY_SETTING,
                     MetaData.SETTING_READ_ONLY_ALLOW_DELETE_SETTING,
+                    MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE,
                     RecoverySettings.INDICES_RECOVERY_MAX_BYTES_PER_SEC_SETTING,
                     RecoverySettings.INDICES_RECOVERY_RETRY_DELAY_STATE_SYNC_SETTING,
                     RecoverySettings.INDICES_RECOVERY_RETRY_DELAY_NETWORK_SETTING,

server/src/main/java/org/elasticsearch/snapshots/RestoreService.java

@@ -268,7 +268,7 @@ public ClusterState execute(ClusterState currentState) {
                                 // Index doesn't exist - create it and start recovery
                                 // Make sure that the index we are about to create has a validate name
                                 MetaDataCreateIndexService.validateIndexName(renamedIndexName, currentState);
-                                createIndexService.validateIndexSettings(renamedIndexName, snapshotIndexMetaData.getSettings());
+                                createIndexService.validateIndexSettings(renamedIndexName, snapshotIndexMetaData.getSettings(), currentState);
                                 IndexMetaData.Builder indexMdBuilder = IndexMetaData.builder(snapshotIndexMetaData).state(IndexMetaData.State.OPEN).index(renamedIndexName);
                                 indexMdBuilder.settings(Settings.builder().put(snapshotIndexMetaData.getSettings()).put(IndexMetaData.SETTING_INDEX_UUID, UUIDs.randomBase64UUID()));
                                 if (!request.includeAliases() && !snapshotIndexMetaData.getAliases().isEmpty()) {