Expose disk usage estimates in nodes stats

This exposes the least and most used disk usage estimates within the "fs" nodes
stats output:

```json
GET /_nodes/stats/fs?pretty&human
{
  "nodes" : {
    "34fPVU0uQ_-wWitDzDXX_g" : {
      "fs" : {
        "timestamp" : 1481238723550,
        "total" : {
          "total" : "396.1gb",
          "total_in_bytes" : 425343254528,
          "free" : "140.6gb",
          "free_in_bytes" : 151068725248,
          "available" : "120.5gb",
          "available_in_bytes" : 129438912512
        },
        "least_usage_estimate" : {
          "path" : "/home/hinmanm/es/elasticsearch/distribution/build/cluster/run node0/elasticsearch-6.0.0-alpha1-SNAPSHOT/data/nodes/0",
          "total" : "396.1gb",
          "total_in_bytes" : 425343254528,
          "available" : "120.5gb",
          "available_in_bytes" : 129438633984,
          "used_disk_percent" : 69.56842912023208
        },
        "most_usage_estimate" : {
          "path" : "/home/hinmanm/es/elasticsearch/distribution/build/cluster/run node0/elasticsearch-6.0.0-alpha1-SNAPSHOT/data/nodes/0",
          "total" : "396.1gb",
          "total_in_bytes" : 425343254528,
          "available" : "120.5gb",
          "available_in_bytes" : 129438633984,
          "used_disk_percent" : 69.56842912023208
        },
        "data" : [{...}],
        "io_stats" : {...}
      }
    }
  }
}
```

Resolves #8686
Resolves #22081

core/src/main/java/org/elasticsearch/monitor/MonitorService.java

@@ -28,6 +28,7 @@ import org.elasticsearch.monitor.jvm.JvmService;
 import org.elasticsearch.monitor.os.OsService;
 import org.elasticsearch.monitor.process.ProcessService;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.cluster.ClusterInfoService;
 
 import java.io.IOException;
 
@@ -39,13 +40,14 @@ public class MonitorService extends AbstractLifecycleComponent {
     private final JvmService jvmService;
     private final FsService fsService;
 
-    public MonitorService(Settings settings, NodeEnvironment nodeEnvironment, ThreadPool threadPool) throws IOException {
+    public MonitorService(Settings settings, NodeEnvironment nodeEnvironment, ThreadPool threadPool,
+                          ClusterInfoService clusterInfoService) throws IOException {
         super(settings);
         this.jvmGcMonitorService = new JvmGcMonitorService(settings, threadPool);
         this.osService = new OsService(settings);
         this.processService = new ProcessService(settings);
         this.jvmService = new JvmService(settings);
-        this.fsService = new FsService(settings, nodeEnvironment);
+        this.fsService = new FsService(settings, nodeEnvironment, clusterInfoService);
     }
 
     public OsService osService() {

core/src/main/java/org/elasticsearch/monitor/fs/FsInfo.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.monitor.fs;
 
+import org.elasticsearch.cluster.DiskUsage;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -438,12 +439,20 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContent {
     private final Path[] paths;
     private final IoStats ioStats;
     private final Path total;
+    private final DiskUsage leastDiskEstimate;
+    private final DiskUsage mostDiskEstimate;
 
     public FsInfo(long timestamp, IoStats ioStats, Path[] paths) {
+        this(timestamp, ioStats, paths, null, null);
+    }
+
+    public FsInfo(long timestamp, IoStats ioStats, Path[] paths, @Nullable DiskUsage leastUsage, @Nullable DiskUsage mostUsage) {
         this.timestamp = timestamp;
         this.ioStats = ioStats;
         this.paths = paths;
         this.total = total();
+        this.leastDiskEstimate = leastUsage;
+        this.mostDiskEstimate = mostUsage;
     }
 
     /**
@@ -457,6 +466,8 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContent {
             paths[i] = new Path(in);
         }
         this.total = total();
+        this.leastDiskEstimate = in.readOptionalWriteable(DiskUsage::new);
+        this.mostDiskEstimate = in.readOptionalWriteable(DiskUsage::new);
     }
 
     @Override
@@ -467,12 +478,24 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContent {
         for (Path path : paths) {
             path.writeTo(out);
         }
+        out.writeOptionalWriteable(this.leastDiskEstimate);
+        out.writeOptionalWriteable(this.mostDiskEstimate);
     }
 
     public Path getTotal() {
         return total;
     }
 
+    @Nullable
+    public DiskUsage getLeastDiskEstimate() {
+        return this.leastDiskEstimate;
+    }
+
+    @Nullable
+    public DiskUsage getMostDiskEstimate() {
+        return this.mostDiskEstimate;
+    }
+
     private Path total() {
         Path res = new Path();
         Set<String> seenDevices = new HashSet<>(paths.length);
@@ -506,6 +529,27 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContent {
         builder.field(Fields.TIMESTAMP, timestamp);
         builder.field(Fields.TOTAL);
         total().toXContent(builder, params);
+        if (leastDiskEstimate != null) {
+            builder.startObject(Fields.LEAST_ESTIMATE);
+            {
+                builder.field(Fields.PATH, leastDiskEstimate.getPath());
+                builder.byteSizeField(Fields.TOTAL_IN_BYTES, Fields.TOTAL, leastDiskEstimate.getTotalBytes());
+                builder.byteSizeField(Fields.AVAILABLE_IN_BYTES, Fields.AVAILABLE, leastDiskEstimate.getFreeBytes());
+                builder.field(Fields.USAGE_PERCENTAGE, leastDiskEstimate.getUsedDiskAsPercentage());
+            }
+            builder.endObject();
+        }
+
+        if (mostDiskEstimate != null) {
+            builder.startObject(Fields.MOST_ESTIMATE);
+            {
+                builder.field(Fields.PATH, mostDiskEstimate.getPath());
+                builder.byteSizeField(Fields.TOTAL_IN_BYTES, Fields.TOTAL, mostDiskEstimate.getTotalBytes());
+                builder.byteSizeField(Fields.AVAILABLE_IN_BYTES, Fields.AVAILABLE, mostDiskEstimate.getFreeBytes());
+                builder.field(Fields.USAGE_PERCENTAGE, mostDiskEstimate.getUsedDiskAsPercentage());
+            }
+            builder.endObject();
+        }
         builder.startArray(Fields.DATA);
         for (Path path : paths) {
             path.toXContent(builder, params);
@@ -525,6 +569,13 @@ public class FsInfo implements Iterable<FsInfo.Path>, Writeable, ToXContent {
         static final String TIMESTAMP = "timestamp";
         static final String DATA = "data";
         static final String TOTAL = "total";
+        static final String TOTAL_IN_BYTES = "total_in_bytes";
         static final String IO_STATS = "io_stats";
+        static final String LEAST_ESTIMATE = "least_usage_estimate";
+        static final String MOST_ESTIMATE = "most_usage_estimate";
+        static final String USAGE_PERCENTAGE = "used_disk_percent";
+        static final String AVAILABLE = "available";
+        static final String AVAILABLE_IN_BYTES = "available_in_bytes";
+        static final String PATH = "path";
     }
 }

core/src/main/java/org/elasticsearch/monitor/fs/FsProbe.java

@@ -22,6 +22,9 @@ package org.elasticsearch.monitor.fs;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.logging.log4j.util.Supplier;
 import org.apache.lucene.util.Constants;
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.cluster.DiskUsage;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.SuppressForbidden;
 import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.component.AbstractComponent;
@@ -48,7 +51,7 @@ public class FsProbe extends AbstractComponent {
         this.nodeEnv = nodeEnv;
     }
 
-    public FsInfo stats(FsInfo previous) throws IOException {
+    public FsInfo stats(FsInfo previous, @Nullable ClusterInfo clusterInfo) throws IOException {
         if (!nodeEnv.hasNodeFile()) {
             return new FsInfo(System.currentTimeMillis(), null, new FsInfo.Path[0]);
         }
@@ -67,7 +70,13 @@ public class FsProbe extends AbstractComponent {
             }
             ioStats = ioStats(devicesNumbers, previous);
         }
-        return new FsInfo(System.currentTimeMillis(), ioStats, paths);
+        DiskUsage leastDiskEstimate = null;
+        DiskUsage mostDiskEstimate = null;
+        if (clusterInfo != null) {
+            leastDiskEstimate = clusterInfo.getNodeLeastAvailableDiskUsages().get(nodeEnv.nodeId());
+            mostDiskEstimate = clusterInfo.getNodeMostAvailableDiskUsages().get(nodeEnv.nodeId());
+        }
+        return new FsInfo(System.currentTimeMillis(), ioStats, paths, leastDiskEstimate, mostDiskEstimate);
     }
 
     final FsInfo.IoStats ioStats(final Set<Tuple<Integer, Integer>> devicesNumbers, final FsInfo previous) {

core/src/main/java/org/elasticsearch/monitor/fs/FsService.java

@@ -20,6 +20,8 @@
 package org.elasticsearch.monitor.fs;
 
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.cluster.ClusterInfo;
+import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractComponent;
 import org.elasticsearch.common.settings.Setting;
 import org.elasticsearch.common.settings.Setting.Property;
@@ -27,6 +29,7 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.SingleObjectCache;
 import org.elasticsearch.env.NodeEnvironment;
+import org.elasticsearch.cluster.ClusterInfoService;
 
 import java.io.IOException;
 
@@ -35,6 +38,7 @@ public class FsService extends AbstractComponent {
     private final FsProbe probe;
     private final TimeValue refreshInterval;
     private final SingleObjectCache<FsInfo> cache;
+    private final ClusterInfoService clusterInfoService;
 
     public static final Setting<TimeValue> REFRESH_INTERVAL_SETTING =
         Setting.timeSetting(
@@ -43,21 +47,22 @@ public class FsService extends AbstractComponent {
             TimeValue.timeValueSeconds(1),
             Property.NodeScope);
 
-    public FsService(final Settings settings, final NodeEnvironment nodeEnvironment) {
+    public FsService(final Settings settings, final NodeEnvironment nodeEnvironment, ClusterInfoService clusterInfoService) {
         super(settings);
         this.probe = new FsProbe(settings, nodeEnvironment);
+        this.clusterInfoService = clusterInfoService;
         refreshInterval = REFRESH_INTERVAL_SETTING.get(settings);
         logger.debug("using refresh_interval [{}]", refreshInterval);
-        cache = new FsInfoCache(refreshInterval, stats(probe, null, logger));
+        cache = new FsInfoCache(refreshInterval, stats(probe, null, logger, null));
     }
 
     public FsInfo stats() {
         return cache.getOrRefresh();
     }
 
-    private static FsInfo stats(FsProbe probe, FsInfo initialValue, Logger logger) {
+    private static FsInfo stats(FsProbe probe, FsInfo initialValue, Logger logger, @Nullable ClusterInfo clusterInfo) {
         try {
-            return probe.stats(initialValue);
+            return probe.stats(initialValue, clusterInfo);
         } catch (IOException e) {
             logger.debug("unexpected exception reading filesystem info", e);
             return null;
@@ -75,7 +80,7 @@ public class FsService extends AbstractComponent {
 
         @Override
         protected FsInfo refresh() {
-            return stats(probe, initialValue, logger);
+            return stats(probe, initialValue, logger, clusterInfoService.getClusterInfo());
         }
 
     }

core/src/main/java/org/elasticsearch/node/Node.java

@@ -342,7 +342,7 @@ public class Node implements Closeable {
             for (Module pluginModule : pluginsService.createGuiceModules()) {
                 modules.add(pluginModule);
             }
-            final MonitorService monitorService = new MonitorService(settings, nodeEnvironment, threadPool);
+            final MonitorService monitorService = new MonitorService(settings, nodeEnvironment, threadPool, clusterInfoService);
             modules.add(new NodeModule(this, monitorService));
             ClusterModule clusterModule = new ClusterModule(settings, clusterService,
                 pluginsService.filterPlugins(ClusterPlugin.class));

core/src/test/java/org/elasticsearch/cluster/ClusterInfoServiceIT.java

@@ -172,7 +172,6 @@ public class ClusterInfoServiceIT extends ESIntegTestCase {
             IndexShard indexShard = indexService.getShardOrNull(shard.id());
             assertEquals(indexShard.shardPath().getRootDataPath().toString(), dataPath);
         }
-
     }
 
     public void testClusterInfoServiceInformationClearOnError() throws InterruptedException, ExecutionException {

core/src/test/java/org/elasticsearch/monitor/fs/FsProbeTests.java

@@ -45,7 +45,7 @@ public class FsProbeTests extends ESTestCase {
         try (NodeEnvironment env = newNodeEnvironment()) {
             FsProbe probe = new FsProbe(Settings.EMPTY, env);
 
-            FsInfo stats = probe.stats(null);
+            FsInfo stats = probe.stats(null, null);
             assertNotNull(stats);
             assertThat(stats.getTimestamp(), greaterThan(0L));