Cluster state toXContent serialization only returns needed data

In order to make sure, that only the requested data is returned to the client,
a couple of fixes have been applied in the ClusterState.toXContent() method.
Also some tests were added to the yaml test suite

Closes #4885

rest-api-spec/api/cluster.state.json

@@ -16,7 +16,7 @@
         },
         "metric" : {
           "type" : "list",
-          "options" : ["_all", "blocks", "index_templates", "metadata", "nodes", "routing_table"],
+          "options" : ["_all", "blocks", "metadata", "nodes", "routing_table"],
           "description" : "Limit the information returned to the specified metrics"
         }
       },
@@ -29,6 +29,10 @@
           "type": "time",
           "description": "Specify timeout for connection to master"
         },
+        "index_templates": {
+          "type": "list",
+          "description": "A comma separated list to return specific index templates when returning metadata"
+        },
         "flat_settings": {
           "type": "boolean",
           "description": "Return settings in flat format (default: false)"

rest-api-spec/test/cluster.state/20_filtering.yaml

@@ -0,0 +1,164 @@
+setup:
+  - do:
+      index:
+        index: testidx
+        type:  testtype
+        id:    testing_document
+        body:
+            "text" : "The quick brown fox is brown."
+  - do:
+      indices.refresh: {}
+
+---
+"Filtering the cluster state by blocks should return the blocks field even if the response is empty":
+  - do:
+      cluster.state:
+        metric: [ blocks ]  
+  
+  - is_true: blocks
+  - is_false: nodes
+  - is_false: metadata
+  - is_false: routing_table
+  - is_false: routing_nodes
+  - is_false: allocations
+  - length:   { blocks: 0 }
+
+---
+"Filtering the cluster state by blocks should return the blocks":
+# read only index
+# TODO: can this cause issues leaving it read only when deleting it in teardown
+  - do:
+      indices.put_settings:
+        index: testidx
+        body:
+          index.blocks.read_only: true
+  - do:
+      cluster.state:
+        metric: [ blocks ]  
+
+  - is_true: blocks
+  - is_false: nodes
+  - is_false: metadata
+  - is_false: routing_table
+  - is_false: routing_nodes
+  - is_false: allocations
+  - length:   { blocks: 1 }
+
+---
+"Filtering the cluster state by nodes only should work":
+  - do:
+      cluster.state:
+        metric: [ nodes ] 
+  
+  - is_false: blocks
+  - is_true: nodes
+  - is_false: metadata
+  - is_false: routing_table
+  - is_false: routing_nodes
+  - is_false: allocations
+
+---
+"Filtering the cluster state by metadata only should work":
+  - do:
+      cluster.state:
+        metric: [ metadata ] 
+  
+  - is_false: blocks
+  - is_false: nodes
+  - is_true: metadata
+  - is_false: routing_table
+  - is_false: routing_nodes
+  - is_false: allocations
+
+
+---
+"Filtering the cluster state by routing table only should work":
+  - do:
+      cluster.state:
+        metric: [ routing_table ] 
+  
+  - is_false: blocks
+  - is_false: nodes
+  - is_false: metadata
+  - is_true: routing_table
+  - is_true: routing_nodes
+  - is_true: allocations
+
+
+---
+"Filtering the cluster state for specific index templates should work ":
+  - do:
+      indices.put_template:
+        name: test1
+        body:
+          template: test-*
+          settings:
+            number_of_shards:   1
+  
+  - do:
+      indices.put_template:
+        name: test2
+        body:
+          template: test-*
+          settings:
+            number_of_shards:   2
+
+  - do:
+      indices.put_template:
+        name: foo
+        body:
+          template: foo-*
+          settings:
+            number_of_shards:   3
+  - do:
+      cluster.state:
+        metric: [ metadata ]
+        index_templates: [ test1, test2 ]
+  
+  - is_false: blocks
+  - is_false: nodes
+  - is_true: metadata
+  - is_false: routing_table
+  - is_false: routing_nodes
+  - is_false: allocations
+  - is_true: metadata.templates.test1
+  - is_true: metadata.templates.test2
+  - is_false: metadata.templates.foo
+
+---
+"Filtering the cluster state by indices should work in routing table and metadata":
+  - do:
+      index:
+        index: another
+        type:  type
+        id:    testing_document
+        body:
+            "text" : "The quick brown fox is brown."
+  
+  - do:
+      indices.refresh: {}
+
+  - do:
+      cluster.state:
+        metric: [ routing_table, metadata ]
+        index: [ testidx ]
+  
+  - is_false: metadata.indices.another
+  - is_false: routing_table.indices.another
+  - is_true: metadata.indices.testidx
+  - is_true: routing_table.indices.testidx
+
+---
+"Filtering the cluster state using _all for indices and metrics should work":
+  - do:
+      cluster.state:
+        metric: [ '_all' ] 
+        index: [ '_all' ] 
+  
+  - is_true: blocks
+  - is_true: nodes
+  - is_true: metadata
+  - is_true: routing_table
+  - is_true: routing_nodes
+  - is_true: allocations
+

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

@@ -35,6 +35,7 @@ import org.elasticsearch.cluster.routing.*;
 import org.elasticsearch.cluster.routing.allocation.AllocationExplanation;
 import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.collect.ImmutableOpenMap;
 import org.elasticsearch.common.compress.CompressedString;
 import org.elasticsearch.common.io.stream.BytesStreamInput;
@@ -50,10 +51,7 @@ import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.shard.ShardId;
 
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
+import java.util.*;
 
 /**
  *
@@ -234,12 +232,14 @@ public class ClusterState implements ToXContent {
 
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        if (!params.paramAsBoolean("filter_nodes", false)) {
+        Set<String> metrics = Strings.splitStringByCommaToSet(params.param("metric", "_all"));
+        boolean isAllMetricsOnly = metrics.size() == 1 && metrics.contains("_all");
+
+        if (isAllMetricsOnly || metrics.contains("nodes")) {
             builder.field("master_node", nodes().masterNodeId());
         }
 
-        // blocks
-        if (!params.paramAsBoolean("filter_blocks", false)) {
+        if (isAllMetricsOnly || metrics.contains("blocks")) {
             builder.startObject("blocks");
 
             if (!blocks().global().isEmpty()) {
@@ -266,7 +266,7 @@ public class ClusterState implements ToXContent {
         }
 
         // nodes
-        if (!params.paramAsBoolean("filter_nodes", false)) {
+        if (isAllMetricsOnly || metrics.contains("nodes")) {
             builder.startObject("nodes");
             for (DiscoveryNode node : nodes()) {
                 builder.startObject(node.id(), XContentBuilder.FieldCaseConversion.NONE);
@@ -285,7 +285,7 @@ public class ClusterState implements ToXContent {
         }
 
         // meta data
-        if (!params.paramAsBoolean("filter_metadata", false)) {
+        if (isAllMetricsOnly || metrics.contains("metadata")) {
             builder.startObject("metadata");
 
             builder.startObject("templates");
@@ -371,7 +371,7 @@ public class ClusterState implements ToXContent {
         }
 
         // routing table
-        if (!params.paramAsBoolean("filter_routing_table", false)) {
+        if (isAllMetricsOnly || metrics.contains("routing_table")) {
             builder.startObject("routing_table");
             builder.startObject("indices");
             for (IndexRoutingTable indexRoutingTable : routingTable()) {
@@ -392,7 +392,7 @@ public class ClusterState implements ToXContent {
         }
 
         // routing nodes
-        if (!params.paramAsBoolean("filter_routing_table", false)) {
+        if (isAllMetricsOnly || metrics.contains("routing_table")) {
             builder.startObject("routing_nodes");
             builder.startArray("unassigned");
             for (ShardRouting shardRouting : readOnlyRoutingNodes().unassigned()) {
@@ -413,7 +413,7 @@ public class ClusterState implements ToXContent {
             builder.endObject();
         }
 
-        if (!params.paramAsBoolean("filter_routing_table", false)) {
+        if (isAllMetricsOnly || metrics.contains("routing_table")) {
             builder.startArray("allocations");
             for (Map.Entry<ShardId, List<AllocationExplanation.NodeExplanation>> entry : allocationExplanation().explanations().entrySet()) {
                 builder.startObject();
@@ -435,7 +435,7 @@ public class ClusterState implements ToXContent {
             builder.endArray();
         }
 
-        if (!params.paramAsBoolean("filter_customs", false)) {
+        if (isAllMetricsOnly || metrics.contains("customs")) {
             for (ObjectObjectCursor<String, Custom> cursor : customs) {
                 builder.startObject(cursor.key);
                 lookupFactorySafe(cursor.key).toXContent(cursor.value, builder, params);