Node Stats: Add fs level stats (size + iostats), closes #1622.

src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodeStats.java

@@ -26,6 +26,7 @@
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.http.HttpStats;
 import org.elasticsearch.indices.NodeIndicesStats;
+import org.elasticsearch.monitor.fs.FsStats;
 import org.elasticsearch.monitor.jvm.JvmStats;
 import org.elasticsearch.monitor.network.NetworkStats;
 import org.elasticsearch.monitor.os.OsStats;
@@ -61,6 +62,9 @@ public class NodeStats extends NodeOperationResponse {
     @Nullable
     private NetworkStats network;
 
+    @Nullable
+    private FsStats fs;
+
     @Nullable
     private TransportStats transport;
 
@@ -72,7 +76,7 @@ public class NodeStats extends NodeOperationResponse {
 
     public NodeStats(DiscoveryNode node, @Nullable String hostname, @Nullable NodeIndicesStats indices,
                      @Nullable OsStats os, @Nullable ProcessStats process, @Nullable JvmStats jvm, @Nullable ThreadPoolStats threadPool, @Nullable NetworkStats network,
-                     @Nullable TransportStats transport, @Nullable HttpStats http) {
+                     @Nullable FsStats fs, @Nullable TransportStats transport, @Nullable HttpStats http) {
         super(node);
         this.hostname = hostname;
         this.indices = indices;
@@ -81,6 +85,7 @@ public NodeStats(DiscoveryNode node, @Nullable String hostname, @Nullable NodeIn
         this.jvm = jvm;
         this.threadPool = threadPool;
         this.network = network;
+        this.fs = fs;
         this.transport = transport;
         this.http = http;
     }
@@ -191,6 +196,22 @@ public NetworkStats getNetwork() {
         return network();
     }
 
+    /**
+     * File system level stats.
+     */
+    @Nullable
+    FsStats fs() {
+        return fs;
+    }
+
+    /**
+     * File system level stats.
+     */
+    @Nullable
+    FsStats getFs() {
+        return fs();
+    }
+
     @Nullable
     public TransportStats transport() {
         return this.transport;
@@ -241,6 +262,9 @@ public void readFrom(StreamInput in) throws IOException {
         if (in.readBoolean()) {
             network = NetworkStats.readNetworkStats(in);
         }
+        if (in.readBoolean()) {
+            fs = FsStats.readFsStats(in);
+        }
         if (in.readBoolean()) {
             transport = TransportStats.readTransportStats(in);
         }
@@ -294,6 +318,12 @@ public void writeTo(StreamOutput out) throws IOException {
             out.writeBoolean(true);
             network.writeTo(out);
         }
+        if (fs == null) {
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            fs.writeTo(out);
+        }
         if (transport == null) {
             out.writeBoolean(false);
         } else {

src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequest.java

@@ -36,6 +36,7 @@ public class NodesStatsRequest extends NodesOperationRequest {
     private boolean jvm;
     private boolean threadPool;
     private boolean network;
+    private boolean fs;
     private boolean transport;
     private boolean http;
 
@@ -60,6 +61,7 @@ public NodesStatsRequest clear() {
         this.jvm = false;
         this.threadPool = false;
         this.network = false;
+        this.fs = false;
         this.transport = false;
         this.http = false;
         return this;
@@ -155,6 +157,21 @@ public NodesStatsRequest network(boolean network) {
         return this;
     }
 
+    /**
+     * Should the node file system stats be returned.
+     */
+    public boolean fs() {
+        return this.fs;
+    }
+
+    /**
+     * Should the node file system stats be returned.
+     */
+    public NodesStatsRequest fs(boolean fs) {
+        this.fs = fs;
+        return this;
+    }
+
     /**
      * Should the node Transport be returned.
      */

src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsRequestBuilder.java

@@ -93,6 +93,14 @@ public NodesStatsRequestBuilder setNetwork(boolean network) {
         return this;
     }
 
+    /**
+     * Should the node file system stats be returned.
+     */
+    public NodesStatsRequestBuilder setFs(boolean fs) {
+        request.fs(fs);
+        return this;
+    }
+
     /**
      * Should the node Transport stats be returned.
      */

src/main/java/org/elasticsearch/action/admin/cluster/node/stats/NodesStatsResponse.java

@@ -101,6 +101,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
             if (nodeStats.network() != null) {
                 nodeStats.network().toXContent(builder, params);
             }
+            if (nodeStats.fs() != null) {
+                nodeStats.fs().toXContent(builder, params);
+            }
             if (nodeStats.transport() != null) {
                 nodeStats.transport().toXContent(builder, params);
             }

src/main/java/org/elasticsearch/action/admin/cluster/node/stats/TransportNodesStatsAction.java

@@ -97,7 +97,7 @@ protected NodeStats newNodeResponse() {
     @Override
     protected NodeStats nodeOperation(NodeStatsRequest nodeStatsRequest) throws ElasticSearchException {
         NodesStatsRequest request = nodeStatsRequest.request;
-        return nodeService.stats(request.indices(), request.os(), request.process(), request.jvm(), request.threadPool(), request.network(), request.transport(), request.http());
+        return nodeService.stats(request.indices(), request.os(), request.process(), request.jvm(), request.threadPool(), request.network(), request.fs(), request.transport(), request.http());
     }
 
     @Override

src/main/java/org/elasticsearch/common/io/stream/AdapterStreamInput.java

@@ -84,6 +84,11 @@ public long readVLong() throws IOException {
         return in.readVLong();
     }
 
+    @Override
+    public String readOptionalUTF() throws IOException {
+        return in.readOptionalUTF();
+    }
+
     @Override
     public String readUTF() throws IOException {
         return in.readUTF();

src/main/java/org/elasticsearch/common/io/stream/AdapterStreamOutput.java

@@ -107,6 +107,11 @@ public void writeVLong(long i) throws IOException {
         out.writeVLong(i);
     }
 
+    @Override
+    public void writeOptionalUTF(@Nullable String str) throws IOException {
+        super.writeOptionalUTF(str);
+    }
+
     @Override
     public void writeUTF(String str) throws IOException {
         out.writeUTF(str);

src/main/java/org/elasticsearch/common/io/stream/StreamInput.java

@@ -147,6 +147,14 @@ public long readVLong() throws IOException {
         return i | ((b & 0x7FL) << 56);
     }
 
+    @Nullable
+    public String readOptionalUTF() throws IOException {
+        if (readBoolean()) {
+            return readUTF();
+        }
+        return null;
+    }
+
     public String readUTF() throws IOException {
         int charCount = readVInt();
         char[] chars = CachedStreamInput.getCharArray(charCount);

src/main/java/org/elasticsearch/common/io/stream/StreamOutput.java

@@ -129,6 +129,15 @@ public void writeVLong(long i) throws IOException {
         writeByte((byte) i);
     }
 
+    public void writeOptionalUTF(@Nullable String str) throws IOException {
+        if (str == null) {
+            writeBoolean(false);
+        } else {
+            writeBoolean(true);
+            writeUTF(str);
+        }
+    }
+
     /**
      * Writes a string.
      */

src/main/java/org/elasticsearch/monitor/MonitorModule.java

@@ -31,6 +31,10 @@
 import org.elasticsearch.monitor.dump.heap.HeapDumpContributor;
 import org.elasticsearch.monitor.dump.summary.SummaryDumpContributor;
 import org.elasticsearch.monitor.dump.thread.ThreadDumpContributor;
+import org.elasticsearch.monitor.fs.FsProbe;
+import org.elasticsearch.monitor.fs.FsService;
+import org.elasticsearch.monitor.fs.JmxFsProbe;
+import org.elasticsearch.monitor.fs.SigarFsProbe;
 import org.elasticsearch.monitor.jvm.JvmMonitorService;
 import org.elasticsearch.monitor.jvm.JvmService;
 import org.elasticsearch.monitor.network.JmxNetworkProbe;
@@ -80,6 +84,7 @@ protected void configure() {
                 bind(ProcessProbe.class).to(SigarProcessProbe.class).asEagerSingleton();
                 bind(OsProbe.class).to(SigarOsProbe.class).asEagerSingleton();
                 bind(NetworkProbe.class).to(SigarNetworkProbe.class).asEagerSingleton();
+                bind(FsProbe.class).to(SigarFsProbe.class).asEagerSingleton();
                 sigarLoaded = true;
             }
         } catch (Throwable e) {
@@ -91,12 +96,14 @@ protected void configure() {
             bind(ProcessProbe.class).to(JmxProcessProbe.class).asEagerSingleton();
             bind(OsProbe.class).to(JmxOsProbe.class).asEagerSingleton();
             bind(NetworkProbe.class).to(JmxNetworkProbe.class).asEagerSingleton();
+            bind(FsProbe.class).to(JmxFsProbe.class).asEagerSingleton();
         }
         // bind other services
         bind(ProcessService.class).asEagerSingleton();
         bind(OsService.class).asEagerSingleton();
         bind(NetworkService.class).asEagerSingleton();
         bind(JvmService.class).asEagerSingleton();
+        bind(FsService.class).asEagerSingleton();
 
         bind(JvmMonitorService.class).asEagerSingleton();
 

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

@@ -23,6 +23,7 @@
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.monitor.fs.FsService;
 import org.elasticsearch.monitor.jvm.JvmMonitorService;
 import org.elasticsearch.monitor.jvm.JvmService;
 import org.elasticsearch.monitor.network.NetworkService;
@@ -44,15 +45,19 @@ public class MonitorService extends AbstractLifecycleComponent<MonitorService> {
 
     private final NetworkService networkService;
 
+    private final FsService fsService;
+
     @Inject
     public MonitorService(Settings settings, JvmMonitorService jvmMonitorService,
-                          OsService osService, ProcessService processService, JvmService jvmService, NetworkService networkService) {
+                          OsService osService, ProcessService processService, JvmService jvmService, NetworkService networkService,
+                          FsService fsService) {
         super(settings);
         this.jvmMonitorService = jvmMonitorService;
         this.osService = osService;
         this.processService = processService;
         this.jvmService = jvmService;
         this.networkService = networkService;
+        this.fsService = fsService;
     }
 
     public OsService osService() {
@@ -71,6 +76,10 @@ public NetworkService networkService() {
         return this.networkService;
     }
 
+    public FsService fsService() {
+        return this.fsService;
+    }
+
     @Override
     protected void doStart() throws ElasticSearchException {
         jvmMonitorService.start();

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

@@ -0,0 +1,27 @@
+/*
+ * 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.monitor.fs;
+
+/**
+ */
+public interface FsProbe {
+
+    FsStats stats();
+}

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

@@ -0,0 +1,55 @@
+/*
+ * 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.monitor.fs;
+
+import org.elasticsearch.common.component.AbstractComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+
+/**
+ */
+public class FsService extends AbstractComponent {
+
+    private final FsProbe probe;
+
+    private final TimeValue refreshInterval;
+
+    private FsStats cachedStats;
+
+    @Inject
+    public FsService(Settings settings, FsProbe probe) {
+        super(settings);
+        this.probe = probe;
+        this.cachedStats = probe.stats();
+
+        this.refreshInterval = componentSettings.getAsTime("refresh_interval", TimeValue.timeValueSeconds(1));
+
+        logger.debug("Using probe [{}] with refresh_interval [{}]", probe, refreshInterval);
+    }
+
+    public synchronized FsStats stats() {
+        if ((System.currentTimeMillis() - cachedStats.timestamp()) > refreshInterval.millis()) {
+            cachedStats = probe.stats();
+        }
+        return cachedStats;
+    }
+
+}