Add AllocationDecider that takes free disk space into account

This commit adds two main pieces, the first is a ClusterInfoService
that provides a service running on the master nodes that fetches the
total/free bytes for each data node in the cluster as well as the
sizes of all shards in the cluster. This information is gathered by
default every 30 seconds, and can be changed dynamically by setting
the `cluster.info.update.interval` setting. This ClusterInfoService
can hopefully be used in the future to weight nodes for allocation
based on their disk usage, if desired.

The second main piece is the DiskThresholdDecider, which can disallow
a shard from being allocated to a node, or from remaining on the node
depending on configuration parameters. There are three main
configuration parameters for the DiskThresholdDecider:

`cluster.routing.allocation.disk.threshold_enabled` controls whether
the decider is enabled. It defaults to false (disabled). Note that the
decider is also disabled for clusters with only a single data node.

`cluster.routing.allocation.disk.watermark.low` controls the low
watermark for disk usage. It defaults to 0.70, meaning ES will not
allocate new shards to nodes once they have more than 70% disk
used. It can also be set to an absolute byte value (like 500mb) to
prevent ES from allocating shards if less than the configured amount
of space is available.

`cluster.routing.allocation.disk.watermark.high` controls the high
watermark. It defaults to 0.85, meaning ES will attempt to relocate
shards to another node if the node disk usage rises above 85%. It can
also be set to an absolute byte value (similar to the low watermark)
to relocate shards once less than the configured amount of space is
available on the node.

Closes #3480

docs/reference/index-modules/allocation.asciidoc

@@ -93,3 +93,41 @@ The `index.routing.allocation.total_shards_per_node` setting allows to
 control how many total shards for an index will be allocated per node.
 It can be dynamically set on a live index using the update index
 settings API.
+
+[float]
+=== Disk-based Shard Allocation
+In 0.90.4 and later, Elasticsearch con be configured to prevent shard
+allocation on nodes depending on disk usage for the node. This
+functionality is disabled by default, and can be changed either in the
+configuration file, or dynamically using:
+
+[source,js]
+--------------------------------------------------
+curl -XPUT localhost:9200/_cluster/settings -d '{
+    "transient" : {
+        "cluster.routing.allocation.disk.threshold_enabled" : true
+    }
+}'
+--------------------------------------------------
+
+Once enabled, Elasticsearch uses two watermarks to decide whether
+shards should be allocated or can remain on the node.
+
+`cluster.routing.allocation.disk.watermark.low` controls the low
+watermark for disk usage. It defaults to 0.70, meaning ES will not
+allocate new shards to nodes once they have more than 70% disk
+used. It can also be set to an absolute byte value (like 500mb) to
+prevent ES from allocating shards if less than the configured amount
+of space is available.
+
+`cluster.routing.allocation.disk.watermark.high` controls the high
+watermark. It defaults to 0.85, meaning ES will attempt to relocate
+shards to another node if the node disk usage rises above 85%. It can
+also be set to an absolute byte value (similar to the low watermark)
+to relocate shards once less than the configured amount of space is
+available on the node.
+
+Both watermark settings can be changed dynamically using the cluster
+settings API. By default, Elasticsearch will retrieve information
+about the disk usage of the nodes every 30 seconds. This can also be
+changed by setting the `cluster.info.update.interval` setting.

src/main/java/org/elasticsearch/cluster/ClusterInfo.java

@@ -0,0 +1,49 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.cluster;
+
+import com.google.common.collect.ImmutableMap;
+
+import java.util.Map;
+
+/**
+ * ClusterInfo is an object representing a map of nodes to {@link DiskUsage}
+ * and a map of shard ids to shard sizes, see
+ * <code>InternalClusterInfoService.shardIdentifierFromRouting(String)</code>
+ * for the key used in the shardSizes map
+ */
+public class ClusterInfo {
+
+    private final ImmutableMap<String, DiskUsage> usages;
+    private final ImmutableMap<String, Long> shardSizes;
+
+    public ClusterInfo(ImmutableMap<String, DiskUsage> usages, ImmutableMap<String, Long> shardSizes) {
+        this.usages = usages;
+        this.shardSizes = shardSizes;
+    }
+
+    public Map<String, DiskUsage> getNodeDiskUsages() {
+        return this.usages;
+    }
+
+    public Map<String, Long> getShardSizes() {
+        return this.shardSizes;
+    }
+}

src/main/java/org/elasticsearch/cluster/ClusterInfoService.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.cluster;
+
+public interface ClusterInfoService {
+
+    public static ClusterInfoService EMPTY = EmptyClusterInfoService.getInstance();
+
+    public ClusterInfo getClusterInfo();
+
+}

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

@@ -78,5 +78,7 @@ protected void configure() {
         bind(MappingUpdatedAction.class).asEagerSingleton();
         bind(NodeAliasesUpdatedAction.class).asEagerSingleton();
         bind(NodeIndicesStateUpdatedAction.class).asEagerSingleton();
+
+        bind(ClusterInfoService.class).to(InternalClusterInfoService.class).asEagerSingleton();
     }
 }

src/main/java/org/elasticsearch/cluster/DiskUsage.java

@@ -0,0 +1,60 @@
+/*
+ * Licensed to ElasticSearch and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. ElasticSearch licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.cluster;
+
+/**
+ * Encapsulation class used to represent the amount of disk used on a node.
+ */
+public class DiskUsage {
+    final String nodeId;
+    final long totalBytes;
+    final long freeBytes;
+
+    public DiskUsage(String nodeId, long totalBytes, long freeBytes) {
+        if ((totalBytes < freeBytes) || (totalBytes < 0)) {
+            throw new IllegalStateException("Free bytes [" + freeBytes +
+                    "] cannot be less than 0 or greater than total bytes [" + totalBytes + "]");
+        }
+        this.nodeId = nodeId;
+        this.totalBytes = totalBytes;
+        this.freeBytes = freeBytes;
+    }
+
+    public double getFreeDiskAsPercentage() {
+        double freePct = 100.0 * ((double)freeBytes / totalBytes);
+        return freePct;
+    }
+
+    public long getFreeBytes() {
+        return freeBytes;
+    }
+
+    public long getTotalBytes() {
+        return totalBytes;
+    }
+
+    public long getUsedBytes() {
+        return getTotalBytes() - getFreeBytes();
+    }
+
+    public String toString() {
+        return "[" + nodeId + "] free: " + getFreeBytes() + "[" + getFreeDiskAsPercentage() + "]";
+    }
+}

src/main/java/org/elasticsearch/cluster/EmptyClusterInfoService.java

@@ -0,0 +1,30 @@
+package org.elasticsearch.cluster;
+
+import com.google.common.collect.ImmutableMap;
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.settings.ImmutableSettings;
+
+/**
+ * ClusterInfoService that provides empty maps for disk usage and shard sizes
+ */
+public class EmptyClusterInfoService extends AbstractComponent implements ClusterInfoService {
+
+    private final static class Holder {
+        private final static EmptyClusterInfoService instance = new EmptyClusterInfoService();
+    }
+    private final ClusterInfo emptyClusterInfo;
+
+    private EmptyClusterInfoService() {
+        super(ImmutableSettings.EMPTY);
+        emptyClusterInfo = new ClusterInfo(ImmutableMap.<String, DiskUsage>of(), ImmutableMap.<String, Long>of());
+    }
+
+    public static EmptyClusterInfoService getInstance() {
+        return Holder.instance;
+    }
+
+    @Override
+    public ClusterInfo getClusterInfo() {
+        return emptyClusterInfo;
+    }
+}