add to nodes stats the total store size of the indices shards allocated to that node

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

@@ -23,6 +23,7 @@ import org.elasticsearch.action.support.nodes.NodeOperationResponse;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.indices.IndicesStats;
 import org.elasticsearch.monitor.jvm.JvmStats;
 import org.elasticsearch.monitor.network.NetworkStats;
 import org.elasticsearch.monitor.os.OsStats;
@@ -39,6 +40,8 @@ import java.io.IOException;
  */
 public class NodeStats extends NodeOperationResponse {
 
+    private IndicesStats indices;
+
     private OsStats os;
 
     private ProcessStats process;
@@ -54,10 +57,11 @@ public class NodeStats extends NodeOperationResponse {
     NodeStats() {
     }
 
-    public NodeStats(DiscoveryNode node,
+    public NodeStats(DiscoveryNode node, IndicesStats indices,
                      OsStats os, ProcessStats process, JvmStats jvm, NetworkStats network,
                      ThreadPoolStats threadPool, TransportStats transport) {
         super(node);
+        this.indices = indices;
         this.os = os;
         this.process = process;
         this.jvm = jvm;
@@ -66,6 +70,20 @@ public class NodeStats extends NodeOperationResponse {
         this.transport = transport;
     }
 
+    /**
+     * Indices level stats.
+     */
+    public IndicesStats indices() {
+        return this.indices;
+    }
+
+    /**
+     * Indices level stats.
+     */
+    public IndicesStats getIndices() {
+        return indices();
+    }
+
     /**
      * Operating System level statistics.
      */
@@ -152,6 +170,9 @@ public class NodeStats extends NodeOperationResponse {
 
     @Override public void readFrom(StreamInput in) throws IOException {
         super.readFrom(in);
+        if (in.readBoolean()) {
+            indices = IndicesStats.readIndicesStats(in);
+        }
         if (in.readBoolean()) {
             os = OsStats.readOsStats(in);
         }
@@ -174,6 +195,12 @@ public class NodeStats extends NodeOperationResponse {
 
     @Override public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
+        if (indices == null) {
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            indices.writeTo(out);
+        }
         if (os == null) {
             out.writeBoolean(false);
         } else {

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

@@ -31,7 +31,7 @@ import java.io.IOException;
  */
 public class NodesStatsResponse extends NodesOperationResponse<NodeStats> {
 
-    public NodesStatsResponse() {
+    NodesStatsResponse() {
     }
 
     public NodesStatsResponse(ClusterName clusterName, NodeStats[] nodes) {

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

@@ -28,6 +28,7 @@ import org.elasticsearch.cluster.ClusterService;
 import org.elasticsearch.common.collect.Lists;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.monitor.MonitorService;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
@@ -42,11 +43,14 @@ public class TransportNodesStatsAction extends TransportNodesOperationAction<Nod
 
     private final MonitorService monitorService;
 
+    private final IndicesService indicesService;
+
     @Inject public TransportNodesStatsAction(Settings settings, ClusterName clusterName, ThreadPool threadPool,
                                              ClusterService clusterService, TransportService transportService,
-                                             MonitorService monitorService) {
+                                             MonitorService monitorService, IndicesService indicesService) {
         super(settings, clusterName, threadPool, clusterService, transportService);
         this.monitorService = monitorService;
+        this.indicesService = indicesService;
     }
 
     @Override protected String transportAction() {
@@ -85,7 +89,7 @@ public class TransportNodesStatsAction extends TransportNodesOperationAction<Nod
     }
 
     @Override protected NodeStats nodeOperation(NodeStatsRequest request) throws ElasticSearchException {
-        return new NodeStats(clusterService.state().nodes().localNode(),
+        return new NodeStats(clusterService.state().nodes().localNode(), indicesService.stats(),
                 monitorService.osService().stats(), monitorService.processService().stats(),
                 monitorService.jvmService().stats(), monitorService.networkService().stats(),
                 threadPool.stats(), transportService.stats());

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/cluster/IndexerClusterService.java

@@ -142,7 +142,7 @@ public class IndexerClusterService extends AbstractLifecycleComponent<IndexerClu
     private class UpdateClusterStateListener implements PublishIndexerClusterStateAction.NewClusterStateListener {
         @Override public void onNewClusterState(final IndexerClusterState clusterState) {
             ClusterState state = clusterService.state();
-            if (!state.nodes().localNodeMaster()) {
+            if (state.nodes().localNodeMaster()) {
                 logger.warn("master should not receive new cluster state from [{}]", state.nodes().masterNode());
                 return;
             }

modules/elasticsearch/src/main/java/org/elasticsearch/indices/IndicesService.java

@@ -40,6 +40,8 @@ public interface IndicesService extends Iterable<IndexService>, LifecycleCompone
      */
     public boolean changesAllowed();
 
+    IndicesStats stats();
+
     boolean hasIndex(String index);
 
     IndicesLifecycle indicesLifecycle();

modules/elasticsearch/src/main/java/org/elasticsearch/indices/IndicesStats.java

@@ -0,0 +1,82 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search 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.indices;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Streamable;
+import org.elasticsearch.common.unit.ByteSizeValue;
+import org.elasticsearch.common.xcontent.ToXContent;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+import java.io.IOException;
+import java.io.Serializable;
+
+/**
+ * Global information on indices stats running on a specific node.
+ *
+ * @author kimchy (shay.banon)
+ */
+public class IndicesStats implements Streamable, Serializable, ToXContent {
+
+    private ByteSizeValue storeSize;
+
+    IndicesStats() {
+    }
+
+    public IndicesStats(ByteSizeValue storeSize) {
+        this.storeSize = storeSize;
+    }
+
+    /**
+     * The size of the index storage taken on the node.
+     */
+    public ByteSizeValue storeSize() {
+        return this.storeSize;
+    }
+
+    /**
+     * The size of the index storage taken on the node.
+     */
+    public ByteSizeValue getStoreSize() {
+        return storeSize;
+    }
+
+    public static IndicesStats readIndicesStats(StreamInput in) throws IOException {
+        IndicesStats stats = new IndicesStats();
+        stats.readFrom(in);
+        return stats;
+    }
+
+    @Override public void readFrom(StreamInput in) throws IOException {
+        storeSize = ByteSizeValue.readBytesSizeValue(in);
+    }
+
+    @Override public void writeTo(StreamOutput out) throws IOException {
+        storeSize.writeTo(out);
+    }
+
+    @Override public void toXContent(XContentBuilder builder, Params params) throws IOException {
+        builder.startObject("indices");
+        builder.field("store_size", storeSize.toString());
+        builder.field("store_size_in_bytes", storeSize.bytes());
+        builder.endObject();
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indices/InternalIndicesService.java

@@ -34,6 +34,7 @@ import org.elasticsearch.common.inject.Injector;
 import org.elasticsearch.common.inject.Injectors;
 import org.elasticsearch.common.inject.ModulesBuilder;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.util.concurrent.ThreadSafe;
 import org.elasticsearch.gateway.Gateway;
 import org.elasticsearch.index.*;
@@ -50,6 +51,8 @@ import org.elasticsearch.index.query.IndexQueryParserModule;
 import org.elasticsearch.index.routing.OperationRoutingModule;
 import org.elasticsearch.index.service.IndexService;
 import org.elasticsearch.index.settings.IndexSettingsModule;
+import org.elasticsearch.index.shard.service.IndexShard;
+import org.elasticsearch.index.shard.service.InternalIndexShard;
 import org.elasticsearch.index.similarity.SimilarityModule;
 import org.elasticsearch.index.store.IndexStoreModule;
 import org.elasticsearch.indices.analysis.IndicesAnalysisService;
@@ -58,6 +61,7 @@ import org.elasticsearch.plugins.IndexPluginsModule;
 import org.elasticsearch.plugins.PluginsService;
 import org.elasticsearch.threadpool.ThreadPool;
 
+import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -147,6 +151,20 @@ public class InternalIndicesService extends AbstractLifecycleComponent<IndicesSe
         return this.indicesLifecycle;
     }
 
+    @Override public IndicesStats stats() {
+        long totalSize = 0;
+        for (IndexService indexService : indices.values()) {
+            for (IndexShard indexShard : indexService) {
+                try {
+                    totalSize += ((InternalIndexShard) indexShard).store().estimateSize().bytes();
+                } catch (IOException e) {
+                    // ignore
+                }
+            }
+        }
+        return new IndicesStats(new ByteSizeValue(totalSize));
+    }
+
     /**
      * Returns <tt>true</tt> if changes (adding / removing) indices, shards and so on are allowed.
      */

modules/elasticsearch/src/main/java/org/elasticsearch/monitor/jvm/JvmStats.java

@@ -413,8 +413,8 @@ public class JvmStats implements Streamable, Serializable, ToXContent {
             }
 
             @Override public void writeTo(StreamOutput out) throws IOException {
-                out.writeLong(startTime);
+                out.writeVLong(startTime);
-                out.writeLong(endTime);
+                out.writeVLong(endTime);
                 out.writeVLong(max);
                 out.writeVLong(beforeUsed);
                 out.writeVLong(afterUsed);

modules/elasticsearch/src/main/java/org/elasticsearch/monitor/network/NetworkStats.java

@@ -73,6 +73,16 @@ public class NetworkStats implements Streamable, Serializable, ToXContent {
         }
     }
 
+    @Override public void writeTo(StreamOutput out) throws IOException {
+        out.writeVLong(timestamp);
+        if (tcp == null) {
+            out.writeBoolean(false);
+        } else {
+            out.writeBoolean(true);
+            tcp.writeTo(out);
+        }
+    }
+
     public long timestamp() {
         return timestamp;
     }
@@ -89,16 +99,6 @@ public class NetworkStats implements Streamable, Serializable, ToXContent {
         return tcp();
     }
 
-    @Override public void writeTo(StreamOutput out) throws IOException {
-        out.writeVLong(timestamp);
-        if (tcp == null) {
-            out.writeBoolean(false);
-        } else {
-            out.writeBoolean(true);
-            tcp.writeTo(out);
-        }
-    }
-
     public static class Tcp implements Serializable, Streamable {
 
         long activeOpens;

modules/elasticsearch/src/main/java/org/elasticsearch/rest/action/admin/cluster/node/stats/RestNodesStatsAction.java

@@ -61,6 +61,10 @@ public class RestNodesStatsAction extends BaseRestHandler {
 
                         builder.field("name", nodeStats.node().name());
 
+                        if (nodeStats.indices() != null) {
+                            nodeStats.indices().toXContent(builder, request);
+                        }
+
                         if (nodeStats.os() != null) {
                             nodeStats.os().toXContent(builder, request);
                         }