[Rest] reroute API response didn't filter metadata

By default the reroute API should return the new cluster state, excluding the metadata. It was however it was wrongly using an old parameter (filter_metadata) and thus failed to do so. This commits restores but wiring it to the correct `metric` parameter. We also add an enum representing the possible metrics, to avoid similar future mistakes.

Closes #7520
Closes #7523

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

@@ -16,9 +16,10 @@
           "type" : "boolean",
           "description" : "Return an explanation of why the commands can or cannot be executed"
         },
-        "filter_metadata": {
-          "type" : "boolean",
-          "description" : "Don't return cluster state metadata (default: false)"
+        "metric": {
+          "type": "list",
+          "options": ["_all", "blocks", "metadata", "nodes", "routing_table", "master_node", "version"],
+          "description": "Limit the information returned to the specified metrics. Defaults to all but metadata"
         },
         "master_timeout": {
           "type" : "time",

rest-api-spec/test/cluster.reroute/20_response_filtering.yaml

@@ -0,0 +1,14 @@
+---
+"Do not return metadata by default":
+  - do:
+      cluster.reroute: {}
+  - is_false: state.metadata
+---
+"return metadata if requested":
+  - do:
+      cluster.reroute:
+        metric: metadata
+
+  - is_true: state.metadata
+  - is_false: state.nodes
+

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

@@ -51,10 +51,10 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import java.io.IOException;
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
-import java.util.Set;
 
 /**
  *
@@ -262,20 +262,71 @@ public String toString() {
         }
     }
 
+    public enum Metric {
+        VERSION("version"),
+        MASTER_NODE("master_node"),
+        BLOCKS("blocks"),
+        NODES("nodes"),
+        METADATA("metadata"),
+        ROUTING_TABLE("routing_table"),
+        CUSTOMS("customs");
+
+        private static Map<String, Metric> valueToEnum;
+
+        static {
+            valueToEnum = new HashMap<>();
+            for (Metric metric : Metric.values()) {
+                valueToEnum.put(metric.value, metric);
+            }
+        }
+
+        private final String value;
+
+        private Metric(String value) {
+            this.value = value;
+        }
+
+        public static EnumSet<Metric> parseString(String param, boolean ignoreUnknown) {
+            String[] metrics = Strings.splitStringByCommaToArray(param);
+            EnumSet<Metric> result = EnumSet.noneOf(Metric.class);
+            for (String metric : metrics) {
+                if ("_all".equals(metric)) {
+                    result = EnumSet.allOf(Metric.class);
+                    break;
+                }
+                Metric m = valueToEnum.get(metric);
+                if (m == null) {
+                    if (!ignoreUnknown) {
+                        throw new ElasticsearchIllegalArgumentException("Unknown metric [" + metric + "]");
+                    }
+                } else {
+                    result.add(m);
+                }
+            }
+            return result;
+        }
+
+        @Override
+        public String toString() {
+            return value;
+        }
+    }
+
+
+
     @Override
     public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
-        Set<String> metrics = Strings.splitStringByCommaToSet(params.param("metric", "_all"));
-        boolean isAllMetricsOnly = metrics.size() == 1 && metrics.contains("_all");
+        EnumSet<Metric> metrics = Metric.parseString(params.param("metric", "_all"), true);
 
-        if (isAllMetricsOnly || metrics.contains("version")) {
+        if (metrics.contains(Metric.VERSION)) {
             builder.field("version", version);
         }
 
-        if (isAllMetricsOnly || metrics.contains("master_node")) {
+        if (metrics.contains(Metric.MASTER_NODE)) {
             builder.field("master_node", nodes().masterNodeId());
         }
 
-        if (isAllMetricsOnly || metrics.contains("blocks")) {
+        if (metrics.contains(Metric.BLOCKS)) {
             builder.startObject("blocks");
 
             if (!blocks().global().isEmpty()) {
@@ -302,7 +353,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
 
         // nodes
-        if (isAllMetricsOnly || metrics.contains("nodes")) {
+        if (metrics.contains(Metric.NODES)) {
             builder.startObject("nodes");
             for (DiscoveryNode node : nodes()) {
                 builder.startObject(node.id(), XContentBuilder.FieldCaseConversion.NONE);
@@ -321,7 +372,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
 
         // meta data
-        if (isAllMetricsOnly || metrics.contains("metadata")) {
+        if (metrics.contains(Metric.METADATA)) {
             builder.startObject("metadata");
 
             builder.startObject("templates");
@@ -407,7 +458,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
 
         // routing table
-        if (isAllMetricsOnly || metrics.contains("routing_table")) {
+        if (metrics.contains(Metric.ROUTING_TABLE)) {
             builder.startObject("routing_table");
             builder.startObject("indices");
             for (IndexRoutingTable indexRoutingTable : routingTable()) {
@@ -428,7 +479,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         }
 
         // routing nodes
-        if (isAllMetricsOnly || metrics.contains("routing_table")) {
+        if (metrics.contains(Metric.ROUTING_TABLE)) {
             builder.startObject("routing_nodes");
             builder.startArray("unassigned");
             for (ShardRouting shardRouting : readOnlyRoutingNodes().unassigned()) {
@@ -448,8 +499,7 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
 
             builder.endObject();
         }
-
-        if (isAllMetricsOnly || metrics.contains("customs")) {
+        if (metrics.contains(Metric.CUSTOMS)) {
             for (ObjectObjectCursor<String, Custom> cursor : customs) {
                 builder.startObject(cursor.key);
                 lookupFactorySafe(cursor.key).toXContent(cursor.value, builder, params);

src/main/java/org/elasticsearch/rest/action/admin/cluster/reroute/RestClusterRerouteAction.java

@@ -23,6 +23,8 @@
 import org.elasticsearch.action.admin.cluster.reroute.ClusterRerouteResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.ClusterState;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsFilter;
@@ -32,13 +34,16 @@
 import org.elasticsearch.rest.action.support.AcknowledgedRestListener;
 
 import java.io.IOException;
+import java.util.EnumSet;
 
 /**
  */
 public class RestClusterRerouteAction extends BaseRestHandler {
 
     private final SettingsFilter settingsFilter;
 
+    private static String DEFAULT_METRICS = Strings.arrayToCommaDelimitedString(EnumSet.complementOf(EnumSet.of(ClusterState.Metric.METADATA)).toArray());
+
     @Inject
     public RestClusterRerouteAction(Settings settings, RestController controller,
                                     SettingsFilter settingsFilter, RestClientFactory restClientFactory) {
@@ -63,9 +68,9 @@ public void handleRequest(final RestRequest request, final RestChannel channel,
             @Override
             protected void addCustomFields(XContentBuilder builder, ClusterRerouteResponse response) throws IOException {
                 builder.startObject("state");
-                // by default, filter metadata
-                if (request.param("filter_metadata") == null) {
-                    request.params().put("filter_metadata", "true");
+                // by default, return everything but metadata
+                if (request.param("metric") == null) {
+                    request.params().put("metric", DEFAULT_METRICS);
                 }
                 response.getState().settingsFilter(settingsFilter).toXContent(builder, request);
                 builder.endObject();

src/main/java/org/elasticsearch/rest/action/admin/cluster/state/RestClusterStateAction.java

@@ -23,6 +23,7 @@
 import org.elasticsearch.action.admin.cluster.state.ClusterStateResponse;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.client.Requests;
+import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
@@ -32,7 +33,7 @@
 import org.elasticsearch.rest.*;
 import org.elasticsearch.rest.action.support.RestBuilderListener;
 
-import java.util.Set;
+import java.util.EnumSet;
 
 
 /**
@@ -66,13 +67,13 @@ public void handleRequest(final RestRequest request, final RestChannel channel,
             clusterStateRequest.indices(indices);
         }
 
-        Set<String> metrics = Strings.splitStringByCommaToSet(request.param("metric", "_all"));
-        boolean isAllMetricsOnly = metrics.size() == 1 && metrics.contains("_all");
-        if (!isAllMetricsOnly) {
-            clusterStateRequest.nodes(metrics.contains("nodes") || metrics.contains("master_node"));
-            clusterStateRequest.routingTable(metrics.contains("routing_table"));
-            clusterStateRequest.metaData(metrics.contains("metadata"));
-            clusterStateRequest.blocks(metrics.contains("blocks"));
+        if (request.hasParam("metric")) {
+            EnumSet<ClusterState.Metric> metrics = ClusterState.Metric.parseString(request.param("metric"), true);
+            // do not ask for what we do not need.
+            clusterStateRequest.nodes(metrics.contains(ClusterState.Metric.NODES) || metrics.contains(ClusterState.Metric.MASTER_NODE));
+            clusterStateRequest.routingTable(metrics.contains(ClusterState.Metric.ROUTING_TABLE));
+            clusterStateRequest.metaData(metrics.contains(ClusterState.Metric.METADATA));
+            clusterStateRequest.blocks(metrics.contains(ClusterState.Metric.BLOCKS));
         }
 
         client.admin().cluster().state(clusterStateRequest, new RestBuilderListener<ClusterStateResponse>(channel) {