Simplify nodes stats API

First, this breaks backwards compatibility!

* Removed /_cluster/nodes/stats endpoint
* Excpect the stats types not as parameters, but as part of the URL
* Returning all indices stats by default, returning all nodes stats by default
* Supporting groups & types in nodes stats now as well
* Updated documentation & tests accordingly
* Allow level parameter for "shards" and "indices" (cluster does not make sense here)

Closes #4057

docs/reference/cluster/nodes-stats.asciidoc

@@ -9,10 +9,6 @@ the cluster nodes statistics.
 
 [source,js]
 --------------------------------------------------
-curl -XGET 'http://localhost:9200/_cluster/nodes/stats'
-curl -XGET 'http://localhost:9200/_cluster/nodes/nodeId1,nodeId2/stats'
-
-# simplified
 curl -XGET 'http://localhost:9200/_nodes/stats'
 curl -XGET 'http://localhost:9200/_nodes/nodeId1,nodeId2/stats'
 --------------------------------------------------
@@ -22,9 +18,9 @@ second command selectively retrieves nodes stats of only `nodeId1` and
 `nodeId2`. All the nodes selective options are explained
 <<cluster-nodes,here>>.
 
-By default, `indices` stats are returned. With options for `indices`,
-`os`, `process`, `jvm`, `network`, `transport`, `http`, `fs`, `breaker`, and
-`thread_pool`. For example:
+By default, all stats are returned. You can limit this by combining any
+of `indices`, `os`, `process`, `jvm`, `network`, `transport`, `http`,
+`fs`, `breaker` and `thread_pool`. For example:
 
 [horizontal]
 `indices`:: 
@@ -70,13 +66,10 @@ By default, `indices` stats are returned. With options for `indices`,
 [source,js]
 --------------------------------------------------
 # return indices and os
-curl -XGET 'http://localhost:9200/_nodes/stats?os=true'
+curl -XGET 'http://localhost:9200/_nodes/stats/os'
 # return just os and process
-curl -XGET 'http://localhost:9200/_nodes/stats?clear=true&os=true&process=true'
+curl -XGET 'http://localhost:9200/_nodes/stats/os,process'
 # specific type endpoint
-curl -XGET 'http://localhost:9200/_nodes/process/stats'
-curl -XGET 'http://localhost:9200/_nodes/10.0.0.1/process/stats'
-# or, if you like the other way
 curl -XGET 'http://localhost:9200/_nodes/stats/process'
 curl -XGET 'http://localhost:9200/_nodes/10.0.0.1/stats/process'
 --------------------------------------------------
@@ -93,12 +86,12 @@ level or on index level.
 [source,js]
 --------------------------------------------------
 # Node Stats
-curl localhost:9200/_nodes/stats/indices/fielddata/field1,field2?pretty
+curl localhost:9200/_nodes/stats/indices/field1,field2?pretty
 
 # Indices Stat
 curl localhost:9200/_stats/fielddata/field1,field2?pretty
 
 # You can use wildcards for field names
 curl localhost:9200/_stats/fielddata/field*?pretty
-curl localhost:9200/_nodes/stats/indices/fielddata/field*?pretty
+curl localhost:9200/_nodes/stats/indices/field*?pretty
 --------------------------------------------------

src/main/java/org/elasticsearch/action/admin/indices/stats/IndexShardStats.java

@@ -20,19 +20,25 @@
 package org.elasticsearch.action.admin.indices.stats;
 
 import com.google.common.collect.Iterators;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Streamable;
 import org.elasticsearch.index.shard.ShardId;
 
+import java.io.IOException;
 import java.util.Iterator;
 
 /**
  */
-public class IndexShardStats implements Iterable<ShardStats> {
+public class IndexShardStats implements Iterable<ShardStats>, Streamable {
 
-    private final ShardId shardId;
+    private ShardId shardId;
 
-    private final ShardStats[] shards;
+    private ShardStats[] shards;
 
-    IndexShardStats(ShardId shardId, ShardStats[] shards) {
+    private IndexShardStats() {}
+
+    public IndexShardStats(ShardId shardId, ShardStats[] shards) {
         this.shardId = shardId;
         this.shards = shards;
     }
@@ -83,4 +89,30 @@ public class IndexShardStats implements Iterable<ShardStats> {
         primary = stats;
         return stats;
     }
+
+    @Override
+    public void readFrom(StreamInput in) throws IOException {
+        shardId = ShardId.readShardId(in);
+        int shardSize = in.readVInt();
+        shards = new ShardStats[shardSize];
+        for (int i = 0; i < shardSize; i++) {
+            shards[i] = ShardStats.readShardStats(in);
+        }
+    }
+
+    @Override
+    public void writeTo(StreamOutput out) throws IOException {
+        shardId.writeTo(out);
+        out.writeVInt(shards.length);
+        for (ShardStats stats : shards) {
+            stats.writeTo(out);
+        }
+    }
+
+    public static IndexShardStats readIndexShardStats(StreamInput in) throws IOException {
+        IndexShardStats indexShardStats = new IndexShardStats();
+        indexShardStats.readFrom(in);
+        return indexShardStats;
+    }
+
 }

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

@@ -19,14 +19,14 @@
 
 package org.elasticsearch.indices;
 
-import com.google.common.collect.ImmutableMap;
-import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.UnmodifiableIterator;
+import com.google.common.collect.*;
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.ElasticSearchIllegalStateException;
 import org.elasticsearch.action.admin.indices.stats.CommonStats;
 import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags;
 import org.elasticsearch.action.admin.indices.stats.CommonStatsFlags.Flag;
+import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
+import org.elasticsearch.action.admin.indices.stats.ShardStats;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.*;
@@ -72,6 +72,7 @@ import org.elasticsearch.plugins.IndexPluginsModule;
 import org.elasticsearch.plugins.PluginsService;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
@@ -175,44 +176,50 @@ public class InternalIndicesService extends AbstractLifecycleComponent<IndicesSe
 
     @Override
     public NodeIndicesStats stats(boolean includePrevious, CommonStatsFlags flags) {
-        CommonStats stats = new CommonStats(flags);
+        CommonStats oldStats = new CommonStats(flags);
 
         if (includePrevious) {
             Flag[] setFlags = flags.getFlags();
             for (Flag flag : setFlags) {
                 switch (flag) {
                     case Get:
-                        stats.get.add(oldShardsStats.getStats);
+                        oldStats.get.add(oldShardsStats.getStats);
                         break;
                     case Indexing:
-                        stats.indexing.add(oldShardsStats.indexingStats);
+                        oldStats.indexing.add(oldShardsStats.indexingStats);
                         break;
                     case Search:
-                        stats.search.add(oldShardsStats.searchStats);
+                        oldStats.search.add(oldShardsStats.searchStats);
                         break;
                     case Merge:
-                        stats.merge.add(oldShardsStats.mergeStats);
+                        oldStats.merge.add(oldShardsStats.mergeStats);
                         break;
                     case Refresh:
-                        stats.refresh.add(oldShardsStats.refreshStats);
+                        oldStats.refresh.add(oldShardsStats.refreshStats);
                         break;
                     case Flush:
-                        stats.flush.add(oldShardsStats.flushStats);
+                        oldStats.flush.add(oldShardsStats.flushStats);
                         break;
                 }
             }
         }
 
+        Map<Index, List<IndexShardStats>> statsByShard = Maps.newHashMap();
         for (IndexService indexService : indices.values()) {
             for (IndexShard indexShard : indexService) {
                 try {
-                    stats.add(new CommonStats(indexShard, flags));
+                    IndexShardStats indexShardStats = new IndexShardStats(indexShard.shardId(), new ShardStats[] { new ShardStats(indexShard, flags) });
+                    if (!statsByShard.containsKey(indexService.index())) {
+                        statsByShard.put(indexService.index(), Lists.<IndexShardStats>newArrayList(indexShardStats));
+                    } else {
+                        statsByShard.get(indexService.index()).add(indexShardStats);
+                    }
                 } catch (IllegalIndexShardStateException e) {
                     // we can safely ignore illegal state on ones that are closing for example
                 }
             }
         }
-        return new NodeIndicesStats(stats);
+        return new NodeIndicesStats(oldStats, statsByShard);
     }
 
     /**

src/main/java/org/elasticsearch/indices/NodeIndicesStats.java

@@ -19,7 +19,11 @@
 
 package org.elasticsearch.indices;
 
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
 import org.elasticsearch.action.admin.indices.stats.CommonStats;
+import org.elasticsearch.action.admin.indices.stats.IndexShardStats;
+import org.elasticsearch.action.admin.indices.stats.ShardStats;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -27,6 +31,7 @@ import org.elasticsearch.common.io.stream.Streamable;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentBuilderString;
+import org.elasticsearch.index.Index;
 import org.elasticsearch.index.cache.filter.FilterCacheStats;
 import org.elasticsearch.index.cache.id.IdCacheStats;
 import org.elasticsearch.index.engine.SegmentsStats;
@@ -39,11 +44,15 @@ import org.elasticsearch.index.percolator.stats.PercolateStats;
 import org.elasticsearch.index.refresh.RefreshStats;
 import org.elasticsearch.index.search.stats.SearchStats;
 import org.elasticsearch.index.shard.DocsStats;
+import org.elasticsearch.index.shard.ShardId;
+import org.elasticsearch.index.shard.service.IndexShard;
 import org.elasticsearch.index.store.StoreStats;
 import org.elasticsearch.search.suggest.completion.CompletionStats;
 
 import java.io.IOException;
 import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
 
 /**
  * Global information on indices stats running on a specific node.
@@ -51,12 +60,24 @@ import java.io.Serializable;
 public class NodeIndicesStats implements Streamable, Serializable, ToXContent {
 
     private CommonStats stats;
+    private Map<Index, List<IndexShardStats>> statsByShard;
 
     NodeIndicesStats() {
     }
 
-    public NodeIndicesStats(CommonStats stats) {
-        this.stats = stats;
+    public NodeIndicesStats(CommonStats oldStats, Map<Index, List<IndexShardStats>> statsByShard) {
+        //this.stats = stats;
+        this.statsByShard = statsByShard;
+
+        // make a total common stats from old ones and current ones
+        this.stats = oldStats;
+        for (List<IndexShardStats> shardStatsList : statsByShard.values()) {
+            for (IndexShardStats indexShardStats : shardStatsList) {
+                for (ShardStats shardStats : indexShardStats.getShards()) {
+                    stats.add(shardStats.getStats());
+                }
+            }
+        }
     }
 
     @Nullable
@@ -138,21 +159,95 @@ public class NodeIndicesStats implements Streamable, Serializable, ToXContent {
     @Override
     public void readFrom(StreamInput in) throws IOException {
         stats = CommonStats.readCommonStats(in);
+        if (in.readBoolean()) {
+            int entries = in.readVInt();
+            statsByShard = Maps.newHashMap();
+            for (int i = 0; i < entries; i++) {
+                Index index = Index.readIndexName(in);
+                int indexShardListSize = in.readVInt();
+                List<IndexShardStats> indexShardStats = Lists.newArrayListWithCapacity(indexShardListSize);
+                for (int j = 0; j < indexShardListSize; j++) {
+                    indexShardStats.add(IndexShardStats.readIndexShardStats(in));
+                }
+                statsByShard.put(index, indexShardStats);
+            }
+        }
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         stats.writeTo(out);
+        out.writeBoolean(statsByShard != null);
+        if (statsByShard != null) {
+            out.writeVInt(statsByShard.size());
+            for (Map.Entry<Index, List<IndexShardStats>> entry : statsByShard.entrySet()) {
+                entry.getKey().writeTo(out);
+                out.writeVInt(entry.getValue().size());
+                for (IndexShardStats indexShardStats : entry.getValue()) {
+                    indexShardStats.writeTo(out);
+                }
+            }
+        }
     }
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+        String level = params.param("level", "node");
+        boolean isLevelValid = "node".equalsIgnoreCase(level) || "indices".equalsIgnoreCase(level) || "shards".equalsIgnoreCase(level);
+        if (!isLevelValid) {
+            return builder;
+        }
+
+        // "node" level
         builder.startObject(Fields.INDICES);
         stats.toXContent(builder, params);
+
+        if ("indices".equals(level)) {
+            Map<Index, CommonStats> indexStats = createStatsByIndex();
+            builder.startObject(Fields.INDICES);
+            for (Map.Entry<Index, CommonStats> entry : indexStats.entrySet()) {
+                builder.startObject(entry.getKey().name());
+                entry.getValue().toXContent(builder, params);
+                builder.endObject();
+            }
+            builder.endObject();
+        } else if ("shards".equals(level)) {
+            builder.startObject("shards");
+            for (Map.Entry<Index, List<IndexShardStats>> entry : statsByShard.entrySet()) {
+                builder.startArray(entry.getKey().name());
+                for (IndexShardStats indexShardStats : entry.getValue()) {
+                    builder.startObject().startObject(String.valueOf(indexShardStats.getShardId().getId()));
+                    for (ShardStats shardStats : indexShardStats.getShards()) {
+                        shardStats.toXContent(builder, params);
+                    }
+                    builder.endObject().endObject();
+                }
+                builder.endArray();
+            }
+            builder.endObject();
+        }
+
         builder.endObject();
         return builder;
     }
 
+    private Map<Index, CommonStats> createStatsByIndex() {
+        Map<Index, CommonStats> statsMap = Maps.newHashMap();
+        for (Map.Entry<Index, List<IndexShardStats>> entry : statsByShard.entrySet()) {
+            if (!statsMap.containsKey(entry.getKey())) {
+                statsMap.put(entry.getKey(), new CommonStats());
+            }
+
+            for (IndexShardStats indexShardStats : entry.getValue()) {
+                for (ShardStats shardStats : indexShardStats.getShards()) {
+                    statsMap.get(entry.getKey()).add(shardStats.getStats());
+                }
+            }
+        }
+
+        return statsMap;
+    }
+
     static final class Fields {
         static final XContentBuilderString INDICES = new XContentBuilderString("indices");
     }

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

@@ -33,6 +33,9 @@ import org.elasticsearch.rest.*;
 import org.elasticsearch.rest.action.support.RestXContentBuilder;
 
 import java.io.IOException;
+import java.util.Set;
+
+import static org.elasticsearch.rest.RestRequest.Method.GET;
 
 
 /**
@@ -43,116 +46,71 @@ public class RestNodesStatsAction extends BaseRestHandler {
     @Inject
     public RestNodesStatsAction(Settings settings, Client client, RestController controller) {
         super(settings, client);
-        controller.registerHandler(RestRequest.Method.GET, "/_cluster/nodes/stats", this);
-        controller.registerHandler(RestRequest.Method.GET, "/_cluster/nodes/{nodeId}/stats", this);
-
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats", this);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats", this);
-
-        RestIndicesHandler indicesHandler = new RestIndicesHandler(new CommonStatsFlags().all());
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/indices", indicesHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/indices", indicesHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/indices/stats", indicesHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/indices/stats", indicesHandler);
-        for (Flag flag : CommonStatsFlags.Flag.values()) {
-            indicesHandler = new RestIndicesHandler(new CommonStatsFlags().clear().set(flag, true));
-            controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/indices/" + flag.getRestName(), indicesHandler);
-            controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/indices/" + flag.getRestName(), indicesHandler);
-            controller.registerHandler(RestRequest.Method.GET, "/_nodes/indices/" + flag.getRestName() + "/stats", indicesHandler);
-            controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/indices/" + flag.getRestName() + "/stats", indicesHandler);
-            if (flag == Flag.FieldData || flag == Flag.Completion) {
-                // add field specific endpoints
-                controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/indices/" + flag.getRestName() + "/{fields}", indicesHandler);
-                controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/indices/" + flag.getRestName() + "/{fields}", indicesHandler);
-                controller.registerHandler(RestRequest.Method.GET, "/_nodes/indices/" + flag.getRestName() + "/{fields}/stats", indicesHandler);
-                controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/indices/" + flag.getRestName() + "/{fields}/stats", indicesHandler);
-            }
-        }
-
-        RestOsHandler osHandler = new RestOsHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/os", osHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/os", osHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/os/stats", osHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/os/stats", osHandler);
-
-        RestProcessHandler processHandler = new RestProcessHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/process", processHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/process", processHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/process/stats", processHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/process/stats", processHandler);
-
-        RestJvmHandler jvmHandler = new RestJvmHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/jvm", jvmHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/jvm", jvmHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/jvm/stats", jvmHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/jvm/stats", jvmHandler);
+        controller.registerHandler(GET, "/_nodes/stats", this);
+        controller.registerHandler(GET, "/_nodes/{nodeId}/stats", this);
 
-        RestThreadPoolHandler threadPoolHandler = new RestThreadPoolHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/thread_pool", threadPoolHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/thread_pool", threadPoolHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/thread_pool/stats", threadPoolHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/thread_pool/stats", threadPoolHandler);
+        controller.registerHandler(GET, "/_nodes/stats/{metric}", this);
+        controller.registerHandler(GET, "/_nodes/{nodeId}/stats/{metric}", this);
 
-        RestNetworkHandler networkHandler = new RestNetworkHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/network", networkHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/network", networkHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/network/stats", networkHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/network/stats", networkHandler);
+        controller.registerHandler(GET, "/_nodes/stats/{metric}/{indexMetric}", this);
+        controller.registerHandler(GET, "/_nodes/stats/{metric}/{indexMetric}/{fields}", this);
 
-        RestFsHandler fsHandler = new RestFsHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/fs", fsHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/fs", fsHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/fs/stats", fsHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/fs/stats", fsHandler);
-
-        RestTransportHandler transportHandler = new RestTransportHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/transport", transportHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/transport", transportHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/transport/stats", transportHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/transport/stats", transportHandler);
-
-        RestHttpHandler httpHandler = new RestHttpHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/http", httpHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/http", httpHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/http/stats", httpHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/http/stats", httpHandler);
-
-        RestBreakerHandler breakerHandler = new RestBreakerHandler();
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/stats/breaker", breakerHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/stats/breaker", breakerHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/breaker/stats", breakerHandler);
-        controller.registerHandler(RestRequest.Method.GET, "/_nodes/{nodeId}/breaker/stats", breakerHandler);
+        controller.registerHandler(GET, "/_nodes/{nodeId}/stats/{metric}/{indexMetric}", this);
+        controller.registerHandler(GET, "/_nodes/{nodeId}/stats/{metric}/{indexMetric}/{fields}", this);
     }
 
     @Override
     public void handleRequest(final RestRequest request, final RestChannel channel) {
         String[] nodesIds = Strings.splitStringByCommaToArray(request.param("nodeId"));
+        Set<String> metrics = Strings.splitStringByCommaToSet(request.param("metric", "_all"));
+
         NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(nodesIds);
-        boolean clear = request.paramAsBoolean("clear", false);
-        if (clear) {
+        nodesStatsRequest.listenerThreaded(false);
+
+        if (metrics.size() == 1 && metrics.contains("_all")) {
+            nodesStatsRequest.all();
+            nodesStatsRequest.indices(CommonStatsFlags.ALL);
+        } else {
             nodesStatsRequest.clear();
+            nodesStatsRequest.os(metrics.contains("os"));
+            nodesStatsRequest.jvm(metrics.contains("jvm"));
+            nodesStatsRequest.threadPool(metrics.contains("thread_pool"));
+            nodesStatsRequest.network(metrics.contains("network"));
+            nodesStatsRequest.fs(metrics.contains("fs"));
+            nodesStatsRequest.transport(metrics.contains("transport"));
+            nodesStatsRequest.http(metrics.contains("http"));
+            nodesStatsRequest.indices(metrics.contains("indices"));
+            nodesStatsRequest.process(metrics.contains("process"));
+            nodesStatsRequest.breaker(metrics.contains("breaker"));
+
+            // check for index specific metrics
+            if (metrics.contains("indices")) {
+                Set<String> indexMetrics = Strings.splitStringByCommaToSet(request.param("indexMetric", "_all"));
+                if (indexMetrics.size() == 1 && indexMetrics.contains("_all")) {
+                    nodesStatsRequest.indices(CommonStatsFlags.ALL);
+                } else {
+                    CommonStatsFlags flags = new CommonStatsFlags();
+                    for (Flag flag : CommonStatsFlags.Flag.values()) {
+                        flags.set(flag, indexMetrics.contains(flag.getRestName()));
+                    }
+                    nodesStatsRequest.indices(flags);
+                }
+            }
         }
-        boolean all = request.paramAsBoolean("all", false);
-        if (all) {
-            nodesStatsRequest.all();
+
+        if (nodesStatsRequest.indices().isSet(Flag.FieldData) && (request.hasParam("fields") || request.hasParam("fielddata_fields"))) {
+            nodesStatsRequest.indices().fieldDataFields(request.paramAsStringArray("fielddata_fields", request.paramAsStringArray("fields", null)));
         }
-        if (request.hasParam("indices")) {
-            nodesStatsRequest.indices(request.paramAsBoolean("indices", false));
+        if (nodesStatsRequest.indices().isSet(Flag.Completion) && (request.hasParam("fields") || request.hasParam("completion_fields"))) {
+            nodesStatsRequest.indices().completionDataFields(request.paramAsStringArray("completion_fields", request.paramAsStringArray("fields", null)));
+        }
+        if (nodesStatsRequest.indices().isSet(Flag.Search) && (request.hasParam("groups"))) {
+            nodesStatsRequest.indices().groups(request.paramAsStringArray("groups", null));
+        }
+        if (nodesStatsRequest.indices().isSet(Flag.Indexing) && (request.hasParam("types"))) {
+            nodesStatsRequest.indices().types(request.paramAsStringArray("types", null));
         }
-        nodesStatsRequest.os(request.paramAsBoolean("os", nodesStatsRequest.os()));
-        nodesStatsRequest.process(request.paramAsBoolean("process", nodesStatsRequest.process()));
-        nodesStatsRequest.jvm(request.paramAsBoolean("jvm", nodesStatsRequest.jvm()));
-        nodesStatsRequest.threadPool(request.paramAsBoolean("thread_pool", nodesStatsRequest.threadPool()));
-        nodesStatsRequest.network(request.paramAsBoolean("network", nodesStatsRequest.network()));
-        nodesStatsRequest.fs(request.paramAsBoolean("fs", nodesStatsRequest.fs()));
-        nodesStatsRequest.transport(request.paramAsBoolean("transport", nodesStatsRequest.transport()));
-        nodesStatsRequest.http(request.paramAsBoolean("http", nodesStatsRequest.http()));
-        nodesStatsRequest.breaker(request.paramAsBoolean("breaker", nodesStatsRequest.breaker()));
-        executeNodeStats(request, channel, nodesStatsRequest);
-    }
 
-    void executeNodeStats(final RestRequest request, final RestChannel channel, final NodesStatsRequest nodesStatsRequest) {
-        nodesStatsRequest.listenerThreaded(false);
         client.admin().cluster().nodesStats(nodesStatsRequest, new ActionListener<NodesStatsResponse>() {
             @Override
             public void onResponse(NodesStatsResponse response) {
@@ -177,107 +135,4 @@ public class RestNodesStatsAction extends BaseRestHandler {
             }
         });
     }
-
-    class RestIndicesHandler implements RestHandler {
-
-        private final CommonStatsFlags flags;
-
-        RestIndicesHandler(CommonStatsFlags flags) {
-            this.flags = flags;
-        }
-
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            CommonStatsFlags flags = this.flags;
-            if (flags.isSet(Flag.FieldData) && (request.hasParam("fields") || request.hasParam("fielddata_fields"))) {
-                flags = flags.clone().fieldDataFields(request.paramAsStringArray("fielddata_fields", request.paramAsStringArray("fields", null)));
-            } else if (flags.isSet(Flag.Completion) && (request.hasParam("fields") || request.hasParam("completion_fields"))) {
-                flags = flags.clone().completionDataFields(request.paramAsStringArray("completion_fields", request.paramAsStringArray("fields", null)));
-            }
-            nodesStatsRequest.clear().indices(flags);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestOsHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().os(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestProcessHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().process(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestJvmHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().jvm(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestThreadPoolHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().threadPool(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestNetworkHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().network(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestFsHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().fs(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestTransportHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().transport(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestHttpHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().http(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
-
-    class RestBreakerHandler implements RestHandler {
-        @Override
-        public void handleRequest(final RestRequest request, final RestChannel channel) {
-            NodesStatsRequest nodesStatsRequest = new NodesStatsRequest(Strings.splitStringByCommaToArray(request.param("nodeId")));
-            nodesStatsRequest.clear().breaker(true);
-            executeNodeStats(request, channel, nodesStatsRequest);
-        }
-    }
 }