REST Client: NodeSelector for node attributes (#31296)

Add a `NodeSelector` so that users can filter the nodes that receive
requests based on node attributes.

I believe we'll need this to backport #30523 and we want it anyway.

I also added a bash script to help with rebuilding the sniffer parsing
test documents.

client/rest/src/main/java/org/elasticsearch/client/HasAttributeNodeSelector.java

@@ -0,0 +1,56 @@
+/*
+ * Licensed to Elasticsearch 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.client;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * A {@link NodeSelector} that selects nodes that have a particular value
+ * for an attribute.
+ */
+public final class HasAttributeNodeSelector implements NodeSelector {
+    private final String key;
+    private final String value;
+
+    public HasAttributeNodeSelector(String key, String value) {
+        this.key = key;
+        this.value = value;
+    }
+
+    @Override
+    public void select(Iterable<Node> nodes) {
+        Iterator<Node> itr = nodes.iterator();
+        while (itr.hasNext()) {
+            Map<String, List<String>> allAttributes = itr.next().getAttributes();
+            if (allAttributes == null) continue;
+            List<String> values = allAttributes.get(key);
+            if (values == null || false == values.contains(value)) {
+                itr.remove();
+            }
+        }
+    }
+
+    @Override
+    public String toString() {
+        return key + "=" + value;
+    }
+}

client/rest/src/main/java/org/elasticsearch/client/Node.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.client;
 
+import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 
@@ -52,13 +54,18 @@ public class Node {
      * if we don't know what roles the node has.
      */
     private final Roles roles;
+    /**
+     * Attributes declared on the node.
+     */
+    private final Map<String, List<String>> attributes;
 
     /**
      * Create a {@linkplain Node} with metadata. All parameters except
      * {@code host} are nullable and implementations of {@link NodeSelector}
      * need to decide what to do in their absence.
      */
-    public Node(HttpHost host, Set<HttpHost> boundHosts, String name, String version, Roles roles) {
+    public Node(HttpHost host, Set<HttpHost> boundHosts, String name, String version,
+            Roles roles, Map<String, List<String>> attributes) {
         if (host == null) {
             throw new IllegalArgumentException("host cannot be null");
         }
@@ -67,13 +74,14 @@ public class Node {
         this.name = name;
         this.version = version;
         this.roles = roles;
+        this.attributes = attributes;
     }
 
     /**
      * Create a {@linkplain Node} without any metadata.
      */
     public Node(HttpHost host) {
-        this(host, null, null, null, null);
+        this(host, null, null, null, null, null);
     }
 
     /**
@@ -115,6 +123,13 @@ public class Node {
         return roles;
     }
 
+    /**
+     * Attributes declared on the node.
+     */
+    public Map<String, List<String>> getAttributes() {
+        return attributes;
+    }
+
     @Override
     public String toString() {
         StringBuilder b = new StringBuilder();
@@ -131,6 +146,9 @@ public class Node {
         if (roles != null) {
             b.append(", roles=").append(roles);
         }
+        if (attributes != null) {
+            b.append(", attributes=").append(attributes);
+        }
         return b.append(']').toString();
     }
 
@@ -144,12 +162,13 @@ public class Node {
             && Objects.equals(boundHosts, other.boundHosts)
             && Objects.equals(name, other.name)
             && Objects.equals(version, other.version)
-            && Objects.equals(roles, other.roles);
+            && Objects.equals(roles, other.roles)
+            && Objects.equals(attributes, other.attributes);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(host, boundHosts, name, version, roles);
+        return Objects.hash(host, boundHosts, name, version, roles, attributes);
     }
 
     /**

client/rest/src/test/java/org/elasticsearch/client/HasAttributeNodeSelectorTests.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to Elasticsearch 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.client;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.client.Node.Roles;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+import static org.junit.Assert.assertEquals;
+
+public class HasAttributeNodeSelectorTests extends RestClientTestCase {
+    public void testHasAttribute() {
+        Node hasAttributeValue = dummyNode(singletonMap("attr", singletonList("val")));
+        Node hasAttributeButNotValue = dummyNode(singletonMap("attr", singletonList("notval")));
+        Node hasAttributeValueInList = dummyNode(singletonMap("attr", Arrays.asList("val", "notval")));
+        Node notHasAttribute = dummyNode(singletonMap("notattr", singletonList("val")));
+        List<Node> nodes = new ArrayList<>();
+        nodes.add(hasAttributeValue);
+        nodes.add(hasAttributeButNotValue);
+        nodes.add(hasAttributeValueInList);
+        nodes.add(notHasAttribute);
+        List<Node> expected = new ArrayList<>();
+        expected.add(hasAttributeValue);
+        expected.add(hasAttributeValueInList);
+        new HasAttributeNodeSelector("attr", "val").select(nodes);
+        assertEquals(expected, nodes);
+    }
+
+    private static Node dummyNode(Map<String, List<String>> attributes) {
+        return new Node(new HttpHost("dummy"), Collections.<HttpHost>emptySet(),
+                randomAsciiAlphanumOfLength(5), randomAsciiAlphanumOfLength(5),
+                new Roles(randomBoolean(), randomBoolean(), randomBoolean()),
+                attributes);
+    }
+}

client/rest/src/test/java/org/elasticsearch/client/NodeSelectorTests.java

@@ -63,9 +63,10 @@ public class NodeSelectorTests extends RestClientTestCase {
         assertEquals(expected, nodes);
     }
 
-    private Node dummyNode(boolean master, boolean data, boolean ingest) {
+    private static Node dummyNode(boolean master, boolean data, boolean ingest) {
         return new Node(new HttpHost("dummy"), Collections.<HttpHost>emptySet(),
                 randomAsciiAlphanumOfLength(5), randomAsciiAlphanumOfLength(5),
-                new Roles(master, data, ingest));
+                new Roles(master, data, ingest),
+                Collections.<String, List<String>>emptyMap());
     }
 }

client/rest/src/test/java/org/elasticsearch/client/NodeTests.java

@@ -23,49 +23,67 @@ import org.apache.http.HttpHost;
 import org.elasticsearch.client.Node.Roles;
 
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
 
 import static java.util.Collections.singleton;
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class NodeTests extends RestClientTestCase {
     public void testToString() {
+        Map<String, List<String>> attributes = new HashMap<>();
+        attributes.put("foo", singletonList("bar"));
+        attributes.put("baz", Arrays.asList("bort", "zoom"));
         assertEquals("[host=http://1]", new Node(new HttpHost("1")).toString());
+        assertEquals("[host=http://1, attributes={foo=[bar], baz=[bort, zoom]}]",
+                new Node(new HttpHost("1"), null, null, null, null, attributes).toString());
         assertEquals("[host=http://1, roles=mdi]", new Node(new HttpHost("1"),
-                null, null, null, new Roles(true, true, true)).toString());
+                null, null, null, new Roles(true, true, true), null).toString());
         assertEquals("[host=http://1, version=ver]", new Node(new HttpHost("1"),
-                null, null, "ver", null).toString());
+                null, null, "ver", null, null).toString());
         assertEquals("[host=http://1, name=nam]", new Node(new HttpHost("1"),
-                null, "nam", null, null).toString());
+                null, "nam", null, null, null).toString());
         assertEquals("[host=http://1, bound=[http://1, http://2]]", new Node(new HttpHost("1"),
-                new HashSet<>(Arrays.asList(new HttpHost("1"), new HttpHost("2"))), null, null, null).toString());
-        assertEquals("[host=http://1, bound=[http://1, http://2], name=nam, version=ver, roles=m]",
+                new HashSet<>(Arrays.asList(new HttpHost("1"), new HttpHost("2"))), null, null, null, null).toString());
+        assertEquals(
+                "[host=http://1, bound=[http://1, http://2], name=nam, version=ver, roles=m, attributes={foo=[bar], baz=[bort, zoom]}]",
                 new Node(new HttpHost("1"), new HashSet<>(Arrays.asList(new HttpHost("1"), new HttpHost("2"))),
-                    "nam", "ver", new Roles(true, false, false)).toString());
+                    "nam", "ver", new Roles(true, false, false), attributes).toString());
 
     }
 
     public void testEqualsAndHashCode() {
         HttpHost host = new HttpHost(randomAsciiAlphanumOfLength(5));
         Node node = new Node(host,
-            randomBoolean() ? null : singleton(host),
-            randomBoolean() ? null : randomAsciiAlphanumOfLength(5),
-            randomBoolean() ? null : randomAsciiAlphanumOfLength(5),
-            randomBoolean() ? null : new Roles(true, true, true));
+                randomBoolean() ? null : singleton(host),
+                randomBoolean() ? null : randomAsciiAlphanumOfLength(5),
+                randomBoolean() ? null : randomAsciiAlphanumOfLength(5),
+                randomBoolean() ? null : new Roles(true, true, true),
+                randomBoolean() ? null : singletonMap("foo", singletonList("bar")));
         assertFalse(node.equals(null));
         assertTrue(node.equals(node));
         assertEquals(node.hashCode(), node.hashCode());
-        Node copy = new Node(host, node.getBoundHosts(), node.getName(), node.getVersion(), node.getRoles());
+        Node copy = new Node(host, node.getBoundHosts(), node.getName(), node.getVersion(),
+                node.getRoles(), node.getAttributes());
         assertTrue(node.equals(copy));
         assertEquals(node.hashCode(), copy.hashCode());
         assertFalse(node.equals(new Node(new HttpHost(host.toHostString() + "changed"), node.getBoundHosts(),
-                node.getName(), node.getVersion(), node.getRoles())));
+                node.getName(), node.getVersion(), node.getRoles(), node.getAttributes())));
         assertFalse(node.equals(new Node(host, new HashSet<>(Arrays.asList(host, new HttpHost(host.toHostString() + "changed"))),
-                node.getName(), node.getVersion(), node.getRoles())));
-        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName() + "changed", node.getVersion(), node.getRoles())));
-        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName(), node.getVersion() + "changed", node.getRoles())));
-        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName(), node.getVersion(), new Roles(false, false, false))));
+                node.getName(), node.getVersion(), node.getRoles(), node.getAttributes())));
+        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName() + "changed",
+                node.getVersion(), node.getRoles(), node.getAttributes())));
+        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName(),
+                node.getVersion() + "changed", node.getRoles(), node.getAttributes())));
+        assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName(),
+                node.getVersion(), new Roles(false, false, false), node.getAttributes())));
+                assertFalse(node.equals(new Node(host, node.getBoundHosts(), node.getName(),
+                node.getVersion(), node.getRoles(), singletonMap("bort", singletonList("bing")))));
     }
 }

client/rest/src/test/java/org/elasticsearch/client/RestClientMultipleHostsTests.java

@@ -342,7 +342,7 @@ public class RestClientMultipleHostsTests extends RestClientTestCase {
         List<Node> newNodes = new ArrayList<>(nodes.size());
         for (int i = 0; i < nodes.size(); i++) {
             Roles roles = i == 0 ? new Roles(false, true, true) : new Roles(true, false, false);
-            newNodes.add(new Node(nodes.get(i).getHost(), null, null, null, roles));
+            newNodes.add(new Node(nodes.get(i).getHost(), null, null, null, roles, null));
         }
         restClient.setNodes(newNodes);
         int rounds = between(1, 10);

client/rest/src/test/java/org/elasticsearch/client/RestClientTests.java

@@ -341,9 +341,9 @@ public class RestClientTests extends RestClientTestCase {
     }
 
     public void testSelectHosts() throws IOException {
-        Node n1 = new Node(new HttpHost("1"), null, null, "1", null);
-        Node n2 = new Node(new HttpHost("2"), null, null, "2", null);
-        Node n3 = new Node(new HttpHost("3"), null, null, "3", null);
+        Node n1 = new Node(new HttpHost("1"), null, null, "1", null, null);
+        Node n2 = new Node(new HttpHost("2"), null, null, "2", null, null);
+        Node n3 = new Node(new HttpHost("3"), null, null, "3", null, null);
 
         NodeSelector not1 = new NodeSelector() {
             @Override

client/rest/src/test/java/org/elasticsearch/client/documentation/RestClientDocumentation.java

@@ -36,6 +36,7 @@ import org.apache.http.nio.entity.NStringEntity;
 import org.apache.http.ssl.SSLContextBuilder;
 import org.apache.http.ssl.SSLContexts;
 import org.apache.http.util.EntityUtils;
+import org.elasticsearch.client.HasAttributeNodeSelector;
 import org.elasticsearch.client.HttpAsyncResponseConsumerFactory.HeapBufferedResponseConsumerFactory;
 import org.elasticsearch.client.Node;
 import org.elasticsearch.client.NodeSelector;
@@ -190,11 +191,20 @@ public class RestClientDocumentation {
             //tag::rest-client-options-set-singleton
             request.setOptions(COMMON_OPTIONS);
             //end::rest-client-options-set-singleton
-            //tag::rest-client-options-customize
-            RequestOptions.Builder options = COMMON_OPTIONS.toBuilder();
-            options.addHeader("cats", "knock things off of other things");
-            request.setOptions(options);
-            //end::rest-client-options-customize
+            {
+                //tag::rest-client-options-customize-header
+                RequestOptions.Builder options = COMMON_OPTIONS.toBuilder();
+                options.addHeader("cats", "knock things off of other things");
+                request.setOptions(options);
+                //end::rest-client-options-customize-header
+            }
+            {
+                //tag::rest-client-options-customize-attribute
+                RequestOptions.Builder options = COMMON_OPTIONS.toBuilder();
+                options.setNodeSelector(new HasAttributeNodeSelector("rack", "c12")); // <1>
+                request.setOptions(options);
+                //end::rest-client-options-customize-attribute
+            }
         }
         {
             HttpEntity[] documents = new HttpEntity[10];

client/sniffer/src/main/java/org/elasticsearch/client/sniff/ElasticsearchNodesSniffer.java

@@ -36,12 +36,18 @@ import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
+import static java.util.Collections.singletonList;
+import static java.util.Collections.unmodifiableList;
+import static java.util.Collections.unmodifiableMap;
+
 /**
  * Class responsible for sniffing the http hosts from elasticsearch through the nodes info api and returning them back.
  * Compatible with elasticsearch 2.x+.
@@ -138,16 +144,19 @@ public final class ElasticsearchNodesSniffer implements NodesSniffer {
         Set<HttpHost> boundHosts = new HashSet<>();
         String name = null;
         String version = null;
-        String fieldName = null;
-        // Used to read roles from 5.0+
+        /*
+         * Multi-valued attributes come with key = `real_key.index` and we
+         * unflip them after reading them because we can't rely on the order
+         * that they arive.
+         */
+        final Map<String, String> protoAttributes = new HashMap<String, String>();
+
         boolean sawRoles = false;
         boolean master = false;
         boolean data = false;
         boolean ingest = false;
-        // Used to read roles from 2.x
-        Boolean masterAttribute = null;
-        Boolean dataAttribute = null;
-        boolean clientAttribute = false;
+
+        String fieldName = null;
         while (parser.nextToken() != JsonToken.END_OBJECT) {
             if (parser.getCurrentToken() == JsonToken.FIELD_NAME) {
                 fieldName = parser.getCurrentName();
@@ -170,13 +179,12 @@ public final class ElasticsearchNodesSniffer implements NodesSniffer {
                     }
                 } else if ("attributes".equals(fieldName)) {
                     while (parser.nextToken() != JsonToken.END_OBJECT) {
-                        if (parser.getCurrentToken() == JsonToken.VALUE_STRING && "master".equals(parser.getCurrentName())) {
-                            masterAttribute = toBoolean(parser.getValueAsString());
-                        } else if (parser.getCurrentToken() == JsonToken.VALUE_STRING && "data".equals(parser.getCurrentName())) {
-                            dataAttribute = toBoolean(parser.getValueAsString());
-                        } else if (parser.getCurrentToken() == JsonToken.VALUE_STRING && "client".equals(parser.getCurrentName())) {
-                            clientAttribute = toBoolean(parser.getValueAsString());
-                        } else if (parser.getCurrentToken() == JsonToken.START_OBJECT) {
+                        if (parser.getCurrentToken() == JsonToken.VALUE_STRING) {
+                            String oldValue = protoAttributes.put(parser.getCurrentName(), parser.getValueAsString());
+                            if (oldValue != null) {
+                                throw new IOException("repeated attribute key [" + parser.getCurrentName() + "]");
+                            }
+                        } else {
                             parser.skipChildren();
                         }
                     }
@@ -216,21 +224,74 @@ public final class ElasticsearchNodesSniffer implements NodesSniffer {
         if (publishedHost == null) {
             logger.debug("skipping node [" + nodeId + "] with http disabled");
             return null;
-        } else {
-            logger.trace("adding node [" + nodeId + "]");
-            if (version.startsWith("2.")) {
-                /*
-                 * 2.x doesn't send roles, instead we try to read them from
-                 * attributes.
-                 */
-                master = masterAttribute == null ? false == clientAttribute : masterAttribute;
-                data = dataAttribute == null ? false == clientAttribute : dataAttribute;
-            } else {
-                assert sawRoles : "didn't see roles for [" + nodeId + "]";
+        }
+
+        Map<String, List<String>> realAttributes = new HashMap<>(protoAttributes.size());
+        List<String> keys = new ArrayList<>(protoAttributes.keySet());
+        for (String key : keys) {
+            if (key.endsWith(".0")) {
+                String realKey = key.substring(0, key.length() - 2);
+                List<String> values = new ArrayList<>();
+                int i = 0;
+                while (true) {
+                    String value = protoAttributes.remove(realKey + "." + i);
+                    if (value == null) {
+                        break;
+                    }
+                    values.add(value);
+                    i++;
+                }
+                realAttributes.put(realKey, unmodifiableList(values));
             }
-            assert boundHosts.contains(publishedHost) :
-                    "[" + nodeId + "] doesn't make sense! publishedHost should be in boundHosts";
-            return new Node(publishedHost, boundHosts, name, version, new Roles(master, data, ingest));
+        }
+        for (Map.Entry<String, String> entry : protoAttributes.entrySet()) {
+            realAttributes.put(entry.getKey(), singletonList(entry.getValue()));
+        }
+
+        if (version.startsWith("2.")) {
+            /*
+             * 2.x doesn't send roles, instead we try to read them from
+             * attributes.
+             */
+            boolean clientAttribute = v2RoleAttributeValue(realAttributes, "client", false);
+            Boolean masterAttribute = v2RoleAttributeValue(realAttributes, "master", null);
+            Boolean dataAttribute = v2RoleAttributeValue(realAttributes, "data", null);
+            master = masterAttribute == null ? false == clientAttribute : masterAttribute;
+            data = dataAttribute == null ? false == clientAttribute : dataAttribute;
+        } else {
+            assert sawRoles : "didn't see roles for [" + nodeId + "]";
+        }
+        assert boundHosts.contains(publishedHost) :
+                "[" + nodeId + "] doesn't make sense! publishedHost should be in boundHosts";
+        logger.trace("adding node [" + nodeId + "]");
+        return new Node(publishedHost, boundHosts, name, version, new Roles(master, data, ingest),
+                unmodifiableMap(realAttributes));
+    }
+
+    /**
+     * Returns {@code defaultValue} if the attribute didn't come back,
+     * {@code true} or {@code false} if it did come back as
+     * either of those, or throws an IOException if the attribute
+     * came back in a strange way.
+     */
+    private static Boolean v2RoleAttributeValue(Map<String, List<String>> attributes,
+            String name, Boolean defaultValue) throws IOException {
+        List<String> valueList = attributes.remove(name);
+        if (valueList == null) {
+            return defaultValue;
+        }
+        if (valueList.size() != 1) {
+            throw new IOException("expected only a single attribute value for [" + name + "] but got "
+                    + valueList);
+        }
+        switch (valueList.get(0)) {
+        case "true":
+            return true;
+        case "false":
+            return false;
+        default:
+            throw new IOException("expected [" + name + "] to be either [true] or [false] but was ["
+                    + valueList.get(0) + "]");
         }
     }
 
@@ -248,15 +309,4 @@ public final class ElasticsearchNodesSniffer implements NodesSniffer {
             return name;
         }
     }
-
-    private static boolean toBoolean(String string) {
-        switch (string) {
-        case "true":
-            return true;
-        case "false":
-            return false;
-        default:
-            throw new IllegalArgumentException("[" + string + "] is not a valid boolean");
-        }
-    }
 }

client/sniffer/src/test/java/org/elasticsearch/client/sniff/ElasticsearchNodesSnifferParseTests.java

@@ -30,14 +30,18 @@ import org.elasticsearch.client.sniff.ElasticsearchNodesSniffer.Scheme;
 
 import java.io.IOException;
 import java.io.InputStream;
+import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 
 import com.fasterxml.jackson.core.JsonFactory;
 
-import static org.hamcrest.Matchers.hasItem;
+import static java.util.Collections.singletonList;
 import static org.hamcrest.Matchers.hasSize;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 
 /**
@@ -53,10 +57,14 @@ public class ElasticsearchNodesSnifferParseTests extends RestClientTestCase {
         try {
             HttpEntity entity = new InputStreamEntity(in, ContentType.APPLICATION_JSON);
             List<Node> nodes = ElasticsearchNodesSniffer.readHosts(entity, Scheme.HTTP, new JsonFactory());
-            // Use these assertions because the error messages are nicer than hasItems.
+            /*
+             * Use these assertions because the error messages are nicer
+             * than hasItems and we know the results are in order because
+             * that is how we generated the file.
+             */
             assertThat(nodes, hasSize(expected.length));
-            for (Node expectedNode : expected) {
-                assertThat(nodes, hasItem(expectedNode));
+            for (int i = 0; i < expected.length; i++) {
+                assertEquals(expected[i], nodes.get(i));
             }
         } finally {
             in.close();
@@ -66,13 +74,13 @@ public class ElasticsearchNodesSnifferParseTests extends RestClientTestCase {
     public void test2x() throws IOException {
         checkFile("2.0.0_nodes_http.json",
                 node(9200, "m1", "2.0.0", true, false, false),
-                node(9202, "m2", "2.0.0", true, true, false),
-                node(9201, "m3", "2.0.0", true, false, false),
-                node(9205, "d1", "2.0.0", false, true, false),
+                node(9201, "m2", "2.0.0", true, true, false),
+                node(9202, "m3", "2.0.0", true, false, false),
+                node(9203, "d1", "2.0.0", false, true, false),
                 node(9204, "d2", "2.0.0", false, true, false),
-                node(9203, "d3", "2.0.0", false, true, false),
-                node(9207, "c1", "2.0.0", false, false, false),
-                node(9206, "c2", "2.0.0", false, false, false));
+                node(9205, "d3", "2.0.0", false, true, false),
+                node(9206, "c1", "2.0.0", false, false, false),
+                node(9207, "c2", "2.0.0", false, false, false));
     }
 
     public void test5x() throws IOException {
@@ -104,6 +112,10 @@ public class ElasticsearchNodesSnifferParseTests extends RestClientTestCase {
         Set<HttpHost> boundHosts = new HashSet<>(2);
         boundHosts.add(host);
         boundHosts.add(new HttpHost("[::1]", port));
-        return new Node(host, boundHosts, name, version, new Roles(master, data, ingest));
+        Map<String, List<String>> attributes = new HashMap<>();
+        attributes.put("dummy", singletonList("everyone_has_me"));
+        attributes.put("number", singletonList(name.substring(1)));
+        attributes.put("array", Arrays.asList(name.substring(0, 1), name.substring(1)));
+        return new Node(host, boundHosts, name, version, new Roles(master, data, ingest), attributes);
     }
 }

client/sniffer/src/test/java/org/elasticsearch/client/sniff/ElasticsearchNodesSnifferTests.java

@@ -200,9 +200,21 @@ public class ElasticsearchNodesSnifferTests extends RestClientTestCase {
                 }
             }
 
+            int numAttributes = between(0, 5);
+            Map<String, List<String>> attributes = new HashMap<>(numAttributes);
+            for (int j = 0; j < numAttributes; j++) {
+                int numValues = frequently() ? 1 : between(2, 5);
+                List<String> values = new ArrayList<>();
+                for (int v = 0; v < numValues; v++) {
+                    values.add(j + "value" + v);
+                }
+                attributes.put("attr" + j, values);
+            }
+
             Node node = new Node(publishHost, boundHosts, randomAsciiAlphanumOfLength(5),
                     randomAsciiAlphanumOfLength(5),
-                    new Node.Roles(randomBoolean(), randomBoolean(), randomBoolean()));
+                    new Node.Roles(randomBoolean(), randomBoolean(), randomBoolean()),
+                    attributes);
 
             generator.writeObjectFieldStart(nodeId);
             if (getRandom().nextBoolean()) {
@@ -256,18 +268,17 @@ public class ElasticsearchNodesSnifferTests extends RestClientTestCase {
             generator.writeFieldName("name");
             generator.writeString(node.getName());
 
-            int numAttributes = RandomNumbers.randomIntBetween(getRandom(), 0, 3);
-            Map<String, String> attributes = new HashMap<>(numAttributes);
-            for (int j = 0; j < numAttributes; j++) {
-                attributes.put("attr" + j, "value" + j);
-            }
             if (numAttributes > 0) {
                 generator.writeObjectFieldStart("attributes");
-            }
-            for (Map.Entry<String, String> entry : attributes.entrySet()) {
-                generator.writeStringField(entry.getKey(), entry.getValue());
-            }
-            if (numAttributes > 0) {
+                for (Map.Entry<String, List<String>> entry : attributes.entrySet()) {
+                    if (entry.getValue().size() == 1) {
+                        generator.writeStringField(entry.getKey(), entry.getValue().get(0));
+                    } else {
+                        for (int v = 0; v < entry.getValue().size(); v++) {
+                            generator.writeStringField(entry.getKey() + "." + v, entry.getValue().get(v));
+                        }
+                    }
+                }
                 generator.writeEndObject();
             }
             generator.writeEndObject();

client/sniffer/src/test/resources/2.0.0_nodes_http.json

@@ -1,140 +1,200 @@
 {
-  "cluster_name" : "elasticsearch",
-  "nodes" : {
-    "qYUZ_8bTRwODPxukDlFw6Q" : {
-      "name" : "d2",
-      "transport_address" : "127.0.0.1:9304",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9204",
-      "attributes" : {
-        "master" : "false"
+  "cluster_name": "elasticsearch",
+  "nodes": {
+    "qr-SOrELSaGW8SlU8nflBw": {
+      "name": "m1",
+      "transport_address": "127.0.0.1:9300",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9200",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "m",
+        "data": "false",
+        "array.1": "1",
+        "master": "true"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9204", "[::1]:9204" ],
-        "publish_address" : "127.0.0.1:9204",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9200",
+          "[::1]:9200"
+        ],
+        "publish_address": "127.0.0.1:9200",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "Yej5UVNgR2KgBjUFHOQpCw" : {
-      "name" : "c1",
-      "transport_address" : "127.0.0.1:9307",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9207",
-      "attributes" : {
-        "data" : "false",
-        "master" : "false"
+    "osfiXxUOQzCVIs-eepgSCA": {
+      "name": "m2",
+      "transport_address": "127.0.0.1:9301",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9201",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "m",
+        "array.1": "2",
+        "master": "true"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9207", "[::1]:9207" ],
-        "publish_address" : "127.0.0.1:9207",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9201",
+          "[::1]:9201"
+        ],
+        "publish_address": "127.0.0.1:9201",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "mHttJwhwReangKEx9EGuAg" : {
-      "name" : "m3",
-      "transport_address" : "127.0.0.1:9301",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9201",
-      "attributes" : {
-        "data" : "false",
-        "master" : "true"
+    "lazeJFiIQ8eHHV4GeIdMPg": {
+      "name": "m3",
+      "transport_address": "127.0.0.1:9302",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9202",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "m",
+        "data": "false",
+        "array.1": "3",
+        "master": "true"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9201", "[::1]:9201" ],
-        "publish_address" : "127.0.0.1:9201",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9202",
+          "[::1]:9202"
+        ],
+        "publish_address": "127.0.0.1:9202",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "6Erdptt_QRGLxMiLi9mTkg" : {
-      "name" : "c2",
-      "transport_address" : "127.0.0.1:9306",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9206",
-      "attributes" : {
-        "data" : "false",
-        "client" : "true"
+    "t9WxK-fNRsqV5G0Mm09KpQ": {
+      "name": "d1",
+      "transport_address": "127.0.0.1:9303",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9203",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "d",
+        "array.1": "1",
+        "master": "false"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9206", "[::1]:9206" ],
-        "publish_address" : "127.0.0.1:9206",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9203",
+          "[::1]:9203"
+        ],
+        "publish_address": "127.0.0.1:9203",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "mLRCZBypTiys6e8KY5DMnA" : {
-      "name" : "m1",
-      "transport_address" : "127.0.0.1:9300",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9200",
-      "attributes" : {
-        "data" : "false"
+    "wgoDzluvTViwUjEsmVesKw": {
+      "name": "d2",
+      "transport_address": "127.0.0.1:9304",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9204",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "d",
+        "array.1": "2",
+        "master": "false"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9200", "[::1]:9200" ],
-        "publish_address" : "127.0.0.1:9200",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9204",
+          "[::1]:9204"
+        ],
+        "publish_address": "127.0.0.1:9204",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "pVqOhytXQwetsZVzCBppYw" : {
-      "name" : "m2",
-      "transport_address" : "127.0.0.1:9302",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9202",
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9202", "[::1]:9202" ],
-        "publish_address" : "127.0.0.1:9202",
-        "max_content_length_in_bytes" : 104857600
+    "6j_t3pPhSm-oRTyypTzu5g": {
+      "name": "d3",
+      "transport_address": "127.0.0.1:9305",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9205",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "d",
+        "array.1": "3",
+        "master": "false"
+      },
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9205",
+          "[::1]:9205"
+        ],
+        "publish_address": "127.0.0.1:9205",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "ARyzVfpJSw2a9TOIUpbsBA" : {
-      "name" : "d1",
-      "transport_address" : "127.0.0.1:9305",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9205",
-      "attributes" : {
-        "master" : "false"
+    "PaEkm0z7Ssiuyfkh3aASag": {
+      "name": "c1",
+      "transport_address": "127.0.0.1:9306",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9206",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "c",
+        "data": "false",
+        "array.1": "1",
+        "master": "false"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9205", "[::1]:9205" ],
-        "publish_address" : "127.0.0.1:9205",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9206",
+          "[::1]:9206"
+        ],
+        "publish_address": "127.0.0.1:9206",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "2Hpid-g5Sc2BKCevhN6VQw" : {
-      "name" : "d3",
-      "transport_address" : "127.0.0.1:9303",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "2.0.0",
-      "build" : "de54438",
-      "http_address" : "127.0.0.1:9203",
-      "attributes" : {
-        "master" : "false"
+    "LAFKr2K_QmupqnM_atJqkQ": {
+      "name": "c2",
+      "transport_address": "127.0.0.1:9307",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "2.0.0",
+      "build": "de54438",
+      "http_address": "127.0.0.1:9207",
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "c",
+        "data": "false",
+        "array.1": "2",
+        "master": "false"
       },
-      "http" : {
-        "bound_address" : [ "127.0.0.1:9203", "[::1]:9203" ],
-        "publish_address" : "127.0.0.1:9203",
-        "max_content_length_in_bytes" : 104857600
+      "http": {
+        "bound_address": [
+          "127.0.0.1:9207",
+          "[::1]:9207"
+        ],
+        "publish_address": "127.0.0.1:9207",
+        "max_content_length_in_bytes": 104857600
       }
     }
   }

client/sniffer/src/test/resources/5.0.0_nodes_http.json

@@ -1,168 +1,216 @@
 {
-  "_nodes" : {
-    "total" : 8,
-    "successful" : 8,
-    "failed" : 0
+  "_nodes": {
+    "total": 8,
+    "successful": 8,
+    "failed": 0
   },
-  "cluster_name" : "test",
-  "nodes" : {
-    "DXz_rhcdSF2xJ96qyjaLVw" : {
-      "name" : "m1",
-      "transport_address" : "127.0.0.1:9300",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+  "cluster_name": "elasticsearch",
+  "nodes": {
+    "0S4r3NurTYSFSb8R9SxwWA": {
+      "name": "m1",
+      "transport_address": "127.0.0.1:9300",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "master",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "m",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
           "[::1]:9200",
           "127.0.0.1:9200"
         ],
-        "publish_address" : "127.0.0.1:9200",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9200",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "53Mi6jYdRgeR1cdyuoNfQQ" : {
-      "name" : "m2",
-      "transport_address" : "127.0.0.1:9301",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+    "k_CBrMXARkS57Qb5-3Mw5g": {
+      "name": "m2",
+      "transport_address": "127.0.0.1:9301",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "master",
         "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "m",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
           "[::1]:9201",
           "127.0.0.1:9201"
         ],
-        "publish_address" : "127.0.0.1:9201",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9201",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "XBIghcHiRlWP9c4vY6rETw" : {
-      "name" : "c2",
-      "transport_address" : "127.0.0.1:9307",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+    "6eynRPQ1RleJTeGDuTR9mw": {
+      "name": "m3",
+      "transport_address": "127.0.0.1:9302",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
+        "master",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9207",
-          "127.0.0.1:9207"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "m",
+        "array.1": "3"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9202",
+          "127.0.0.1:9202"
         ],
-        "publish_address" : "127.0.0.1:9207",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9202",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "cFM30FlyS8K1njH_bovwwQ" : {
-      "name" : "d1",
-      "transport_address" : "127.0.0.1:9303",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+    "cbGC-ay1QNWaESvEh5513w": {
+      "name": "d1",
+      "transport_address": "127.0.0.1:9303",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "d",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
           "[::1]:9203",
           "127.0.0.1:9203"
         ],
-        "publish_address" : "127.0.0.1:9203",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9203",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "eoVUVRGNRDyyOapqIcrsIA" : {
-      "name" : "d2",
-      "transport_address" : "127.0.0.1:9304",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+    "LexndPpXR2ytYsU5fTElnQ": {
+      "name": "d2",
+      "transport_address": "127.0.0.1:9304",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "d",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
           "[::1]:9204",
           "127.0.0.1:9204"
         ],
-        "publish_address" : "127.0.0.1:9204",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9204",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "xPN76uDcTP-DyXaRzPg2NQ" : {
-      "name" : "c1",
-      "transport_address" : "127.0.0.1:9306",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
+    "SbNG1DKYSBu20zfOz2gDZQ": {
+      "name": "d3",
+      "transport_address": "127.0.0.1:9305",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
+        "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9206",
-          "127.0.0.1:9206"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "d",
+        "array.1": "3"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9205",
+          "127.0.0.1:9205"
         ],
-        "publish_address" : "127.0.0.1:9206",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9205",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "RY0oW2d7TISEqazk-U4Kcw" : {
-      "name" : "d3",
-      "transport_address" : "127.0.0.1:9305",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
-        "data",
+    "fM4H-m2WTDWmsGsL7jIJew": {
+      "name": "c1",
+      "transport_address": "127.0.0.1:9306",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9205",
-          "127.0.0.1:9205"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "c",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9206",
+          "127.0.0.1:9206"
         ],
-        "publish_address" : "127.0.0.1:9205",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9206",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "tU0rXEZmQ9GsWfn2TQ4kow" : {
-      "name" : "m3",
-      "transport_address" : "127.0.0.1:9302",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "5.0.0",
-      "build_hash" : "253032b",
-      "roles" : [
-        "master",
+    "pFoh7d0BTbqqI3HKd9na5A": {
+      "name": "c2",
+      "transport_address": "127.0.0.1:9307",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "5.0.0",
+      "build_hash": "253032b",
+      "roles": [
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9202",
-          "127.0.0.1:9202"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "c",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9207",
+          "127.0.0.1:9207"
         ],
-        "publish_address" : "127.0.0.1:9202",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9207",
+        "max_content_length_in_bytes": 104857600
       }
     }
   }

client/sniffer/src/test/resources/6.0.0_nodes_http.json

@@ -1,168 +1,216 @@
 {
-  "_nodes" : {
-    "total" : 8,
-    "successful" : 8,
-    "failed" : 0
+  "_nodes": {
+    "total": 8,
+    "successful": 8,
+    "failed": 0
   },
-  "cluster_name" : "test",
-  "nodes" : {
-    "FX9npqGQSL2mOGF8Zkf3hw" : {
-      "name" : "m2",
-      "transport_address" : "127.0.0.1:9301",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
+  "cluster_name": "elasticsearch",
+  "nodes": {
+    "ikXK_skVTfWkhONhldnbkw": {
+      "name": "m1",
+      "transport_address": "127.0.0.1:9300",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
         "master",
-        "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9201",
-          "127.0.0.1:9201"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "m",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9200",
+          "127.0.0.1:9200"
         ],
-        "publish_address" : "127.0.0.1:9201",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9200",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "jmUqzYLGTbWCg127kve3Tg" : {
-      "name" : "d1",
-      "transport_address" : "127.0.0.1:9303",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
+    "TMHa34w4RqeuYoHCfJGXZg": {
+      "name": "m2",
+      "transport_address": "127.0.0.1:9301",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "master",
         "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9203",
-          "127.0.0.1:9203"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "m",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9201",
+          "127.0.0.1:9201"
         ],
-        "publish_address" : "127.0.0.1:9203",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9201",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "soBU6bzvTOqdLxPstSbJ2g" : {
-      "name" : "d3",
-      "transport_address" : "127.0.0.1:9305",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
-        "data",
+    "lzaMRJTVT166sgVZdQ5thA": {
+      "name": "m3",
+      "transport_address": "127.0.0.1:9302",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "master",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9205",
-          "127.0.0.1:9205"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "m",
+        "array.1": "3"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9202",
+          "127.0.0.1:9202"
         ],
-        "publish_address" : "127.0.0.1:9205",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9202",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "mtYDAhURTP6twdmNAkMnOg" : {
-      "name" : "m3",
-      "transport_address" : "127.0.0.1:9302",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
-        "master",
+    "tGP5sUecSd6BLTWk1NWF8Q": {
+      "name": "d1",
+      "transport_address": "127.0.0.1:9303",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9202",
-          "127.0.0.1:9202"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "d",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9203",
+          "127.0.0.1:9203"
         ],
-        "publish_address" : "127.0.0.1:9202",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9203",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "URxHiUQPROOt1G22Ev6lXw" : {
-      "name" : "c2",
-      "transport_address" : "127.0.0.1:9307",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
+    "c1UgW5ROTkSa2YnM_T56tw": {
+      "name": "d2",
+      "transport_address": "127.0.0.1:9304",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9207",
-          "127.0.0.1:9207"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "d",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9204",
+          "127.0.0.1:9204"
         ],
-        "publish_address" : "127.0.0.1:9207",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9204",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "_06S_kWoRqqFR8Z8CS3JRw" : {
-      "name" : "c1",
-      "transport_address" : "127.0.0.1:9306",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
+    "QM9yjqjmS72MstpNYV_trg": {
+      "name": "d3",
+      "transport_address": "127.0.0.1:9305",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
+        "data",
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9206",
-          "127.0.0.1:9206"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "3",
+        "array.0": "d",
+        "array.1": "3"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9205",
+          "127.0.0.1:9205"
         ],
-        "publish_address" : "127.0.0.1:9206",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9205",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "QZE5Bd6DQJmnfVs2dglOvA" : {
-      "name" : "d2",
-      "transport_address" : "127.0.0.1:9304",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
-        "data",
+    "wLtzAssoQYeX_4TstgCj0Q": {
+      "name": "c1",
+      "transport_address": "127.0.0.1:9306",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9204",
-          "127.0.0.1:9204"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "1",
+        "array.0": "c",
+        "array.1": "1"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9206",
+          "127.0.0.1:9206"
         ],
-        "publish_address" : "127.0.0.1:9204",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9206",
+        "max_content_length_in_bytes": 104857600
       }
     },
-    "_3mTXg6dSweZn5ReB2fQqw" : {
-      "name" : "m1",
-      "transport_address" : "127.0.0.1:9300",
-      "host" : "127.0.0.1",
-      "ip" : "127.0.0.1",
-      "version" : "6.0.0",
-      "build_hash" : "8f0685b",
-      "roles" : [
-        "master",
+    "ONOzpst8TH-ZebG7fxGwaA": {
+      "name": "c2",
+      "transport_address": "127.0.0.1:9307",
+      "host": "127.0.0.1",
+      "ip": "127.0.0.1",
+      "version": "6.0.0",
+      "build_hash": "8f0685b",
+      "roles": [
         "ingest"
       ],
-      "http" : {
-        "bound_address" : [
-          "[::1]:9200",
-          "127.0.0.1:9200"
+      "attributes": {
+        "dummy": "everyone_has_me",
+        "number": "2",
+        "array.0": "c",
+        "array.1": "2"
+      },
+      "http": {
+        "bound_address": [
+          "[::1]:9207",
+          "127.0.0.1:9207"
         ],
-        "publish_address" : "127.0.0.1:9200",
-        "max_content_length_in_bytes" : 104857600
+        "publish_address": "127.0.0.1:9207",
+        "max_content_length_in_bytes": 104857600
       }
     }
   }

client/sniffer/src/test/resources/create_test_nodes_info.bash

@@ -0,0 +1,107 @@
+#!/bin/bash
+
+# Recreates the v_nodes_http.json files in this directory. This is
+# meant to be an "every once in a while" thing that we do only when
+# we want to add a new version of Elasticsearch or configure the
+# nodes differently. That is why we don't do this in gradle. It also
+# allows us to play fast and loose with error handling. If something
+# goes wrong you have to manually clean up which is good because it
+# leaves around the kinds of things that we need to debug the failure.
+
+# I built this file so the next time I have to regenerate these
+# v_nodes_http.json files I won't have to reconfigure Elasticsearch
+# from scratch. While I was at it I took the time to make sure that
+# when we do rebuild the files they don't jump around too much. That
+# way the diffs are smaller.
+
+set -e
+
+script_path="$( cd "$(dirname "$0")" ; pwd -P )"
+work=$(mktemp -d)
+pushd ${work} >> /dev/null
+echo Working in ${work}
+
+wget https://download.elasticsearch.org/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.0.0/elasticsearch-2.0.0.tar.gz
+wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-5.0.0.tar.gz
+wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.0.0.tar.gz
+sha1sum -c - << __SHAs
+e369d8579bd3a2e8b5344278d5043f19f14cac88 elasticsearch-2.0.0.tar.gz
+d25f6547bccec9f0b5ea7583815f96a6f50849e0 elasticsearch-5.0.0.tar.gz
+__SHAs
+sha512sum -c - << __SHAs
+25bb622d2fc557d8b8eded634a9b333766f7b58e701359e1bcfafee390776eb323cb7ea7a5e02e8803e25d8b1d3aabec0ec1b0cf492d0bab5689686fe440181c elasticsearch-6.0.0.tar.gz
+__SHAs
+
+
+function do_version() {
+    local version=$1
+    local nodes='m1 m2 m3 d1 d2 d3 c1 c2'
+    rm -rf ${version}
+    mkdir -p ${version}
+    pushd ${version} >> /dev/null
+
+    tar xf ../elasticsearch-${version}.tar.gz
+    local http_port=9200
+    for node in ${nodes}; do
+        mkdir ${node}
+        cp -r elasticsearch-${version}/* ${node}
+        local master=$([[ "$node" =~ ^m.* ]] && echo true || echo false)
+        local data=$([[ "$node" =~ ^d.* ]] && echo true || echo false)
+        # m2 is always master and data for these test just so we have a node like that
+        data=$([[ "$node" == 'm2' ]] && echo true || echo ${data})
+        local attr=$([ ${version} == '2.0.0' ] && echo '' || echo '.attr')
+        local transport_port=$((http_port+100))
+
+        cat >> ${node}/config/elasticsearch.yml << __ES_YML
+node.name:          ${node}
+node.master:        ${master}
+node.data:          ${data}
+node${attr}.dummy:  everyone_has_me
+node${attr}.number: ${node:1}
+node${attr}.array:  [${node:0:1}, ${node:1}]
+http.port:          ${http_port}
+transport.tcp.port: ${transport_port}
+discovery.zen.minimum_master_nodes: 3
+discovery.zen.ping.unicast.hosts: ['localhost:9300','localhost:9301','localhost:9302']
+__ES_YML
+
+        if [ ${version} != '2.0.0' ]; then
+            perl -pi -e 's/-Xm([sx]).+/-Xm${1}512m/g' ${node}/config/jvm.options
+        fi
+
+        echo "starting ${version}/${node}..."
+        ${node}/bin/elasticsearch -d -p ${node}/pidfile
+
+        ((http_port++))
+    done
+
+    echo "waiting for cluster to form"
+    # got to wait for all the nodes
+    until curl -s localhost:9200; do
+        sleep .25
+    done
+
+    echo "waiting for all nodes to join"
+    until [ $(echo ${nodes} | wc -w) -eq $(curl -s localhost:9200/_cat/nodes | wc -l) ]; do
+        sleep .25
+    done
+
+    # jq sorts the nodes by their http host so the file doesn't jump around when we regenerate it
+    curl -s localhost:9200/_nodes/http?pretty \
+        | jq '[to_entries[] | ( select(.key == "nodes").value|to_entries|sort_by(.value.http.publish_address)|from_entries|{"key": "nodes", "value": .} ) // .] | from_entries' \
+        > ${script_path}/${version}_nodes_http.json
+
+    for node in ${nodes}; do
+        echo "stopping ${version}/${node}..."
+        kill $(cat ${node}/pidfile)
+    done
+
+    popd >> /dev/null
+}
+
+JAVA_HOME=$JAVA8_HOME do_version 2.0.0
+JAVA_HOME=$JAVA8_HOME do_version 5.0.0
+JAVA_HOME=$JAVA8_HOME do_version 6.0.0
+
+popd >> /dev/null
+rm -rf ${work}

client/sniffer/src/test/resources/readme.txt

@@ -2,3 +2,5 @@
 few nodes in different configurations locally at various versions. They are
 for testing `ElasticsearchNodesSniffer` against different versions of
 Elasticsearch.
+
+See create_test_nodes_info.bash for how to create these.

docs/java-rest/low-level/usage.asciidoc

@@ -312,9 +312,17 @@ adds an extra header:
 
 ["source","java",subs="attributes,callouts,macros"]
 --------------------------------------------------
-include-tagged::{doc-tests}/RestClientDocumentation.java[rest-client-options-customize]
+include-tagged::{doc-tests}/RestClientDocumentation.java[rest-client-options-customize-header]
 --------------------------------------------------
 
+Or you can send requests to nodes with a particular attribute:
+
+["source","java",subs="attributes,callouts,macros"]
+--------------------------------------------------
+include-tagged::{doc-tests}/RestClientDocumentation.java[rest-client-options-customize-attribute]
+--------------------------------------------------
+<1> Replace the node selector with one that selects nodes on a particular rack.
+
 
 ==== Multiple parallel asynchronous actions
 

rest-api-spec/src/main/resources/rest-api-spec/test/README.asciidoc

@@ -198,9 +198,7 @@ header. The warnings must match exactly. Using it looks like this:
 ....
 
 If the arguments to `do` include `node_selector` then the request is only
-sent to nodes that match the `node_selector`. Currently only the `version`
-selector is supported and it has the same logic as the `version` field in
-`skip`. It looks like this:
+sent to nodes that match the `node_selector`. It looks like this:
 
 ....
 "test id":
@@ -216,6 +214,19 @@ selector is supported and it has the same logic as the `version` field in
           body:   { foo: bar }
 ....
 
+If you list multiple selectors then the request will only go to nodes that
+match all of those selectors. The following selectors are supported:
+* `version`: Only nodes who's version is within the range will receive the
+request. The syntax for the pattern is the same as when `version` is within
+`skip`.
+* `attribute`: Only nodes that have an attribute matching the name and value
+of the provided attribute match. Looks like:
+....
+      node_selector:
+          attribute:
+              name: value
+....
+
 === `set`
 
 For some tests, it is necessary to extract a value from the previous `response`, in

test/framework/src/main/java/org/elasticsearch/test/rest/yaml/section/DoSection.java

@@ -21,6 +21,7 @@ package org.elasticsearch.test.rest.yaml.section;
 
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.Version;
+import org.elasticsearch.client.HasAttributeNodeSelector;
 import org.elasticsearch.client.Node;
 import org.elasticsearch.client.NodeSelector;
 import org.elasticsearch.common.ParsingException;
@@ -31,6 +32,7 @@ import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.xcontent.DeprecationHandler;
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentLocation;
+import org.elasticsearch.common.xcontent.XContentParseException;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.test.rest.yaml.ClientYamlTestExecutionContext;
@@ -131,11 +133,10 @@ public class DoSection implements ExecutableSection {
                     while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                         if (token == XContentParser.Token.FIELD_NAME) {
                             selectorName = parser.currentName();
-                        } else if (token.isValue()) {
-                            NodeSelector newSelector = buildNodeSelector(
-                                parser.getTokenLocation(), selectorName, parser.text());
-                                nodeSelector = nodeSelector == NodeSelector.ANY ?
-                                    newSelector : new ComposeNodeSelector(nodeSelector, newSelector);
+                        } else {
+                            NodeSelector newSelector = buildNodeSelector(selectorName, parser);
+                            nodeSelector = nodeSelector == NodeSelector.ANY ?
+                                newSelector : new ComposeNodeSelector(nodeSelector, newSelector);
                         }
                     }
                 } else if (currentFieldName != null) { // must be part of API call then
@@ -368,34 +369,64 @@ public class DoSection implements ExecutableSection {
                 not(equalTo(409)))));
     }
 
-    private static NodeSelector buildNodeSelector(XContentLocation location, String name, String value) {
+    private static NodeSelector buildNodeSelector(String name, XContentParser parser) throws IOException {
         switch (name) {
+        case "attribute":
+            return parseAttributeValuesSelector(parser);
         case "version":
-            Version[] range = SkipSection.parseVersionRange(value);
-            return new NodeSelector() {
-                @Override
-                public void select(Iterable<Node> nodes) {
-                    for (Iterator<Node> itr = nodes.iterator(); itr.hasNext();) {
-                        Node node = itr.next();
-                        if (node.getVersion() == null) {
-                            throw new IllegalStateException("expected [version] metadata to be set but got "
-                                    + node);
-                        }
-                        Version version = Version.fromString(node.getVersion());
-                        if (false == (version.onOrAfter(range[0]) && version.onOrBefore(range[1]))) {
-                            itr.remove();
-                        }
+            return parseVersionSelector(parser);
+        default:
+            throw new XContentParseException(parser.getTokenLocation(), "unknown node_selector [" + name + "]");
+        }
+    }
+
+    private static NodeSelector parseAttributeValuesSelector(XContentParser parser) throws IOException {
+        if (parser.currentToken() != XContentParser.Token.START_OBJECT) {
+            throw new XContentParseException(parser.getTokenLocation(), "expected START_OBJECT");
+        }
+        String key = null;
+        XContentParser.Token token;
+        NodeSelector result = NodeSelector.ANY;
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                key = parser.currentName();
+            } else if (token.isValue()) {
+                NodeSelector newSelector = new HasAttributeNodeSelector(key, parser.text());
+                result = result == NodeSelector.ANY ?
+                    newSelector : new ComposeNodeSelector(result, newSelector);
+            } else {
+                throw new XContentParseException(parser.getTokenLocation(), "expected [" + key + "] to be a value");
+            }
+        }
+        return result;
+    }
+
+    private static NodeSelector parseVersionSelector(XContentParser parser) throws IOException {
+        if (false == parser.currentToken().isValue()) {
+            throw new XContentParseException(parser.getTokenLocation(), "expected [version] to be a value");
+        }
+        Version[] range = SkipSection.parseVersionRange(parser.text());
+        return new NodeSelector() {
+            @Override
+            public void select(Iterable<Node> nodes) {
+                for (Iterator<Node> itr = nodes.iterator(); itr.hasNext();) {
+                    Node node = itr.next();
+                    if (node.getVersion() == null) {
+                        throw new IllegalStateException("expected [version] metadata to be set but got "
+                                + node);
+                    }
+                    Version version = Version.fromString(node.getVersion());
+                    if (false == (version.onOrAfter(range[0]) && version.onOrBefore(range[1]))) {
+                        itr.remove();
                     }
                 }
+            }
 
-                @Override
-                public String toString() {
-                    return "version between [" + range[0] + "] and [" + range[1] + "]";
-                }
-            };
-        default:
-            throw new IllegalArgumentException("unknown node_selector [" + name + "]");
-        }
+            @Override
+            public String toString() {
+                return "version between [" + range[0] + "] and [" + range[1] + "]";
+            }
+        };
     }
 
     /**

test/framework/src/test/java/org/elasticsearch/test/rest/yaml/section/DoSectionTests.java

@@ -35,6 +35,7 @@ import org.hamcrest.MatcherAssert;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -511,7 +512,7 @@ public class DoSectionTests extends AbstractClientYamlTestFragmentParserTestCase
                 "just one entry this time")));
     }
 
-    public void testNodeSelector() throws IOException {
+    public void testNodeSelectorByVersion() throws IOException {
         parser = createParser(YamlXContent.yamlXContent,
                 "node_selector:\n" +
                 "    version: 5.2.0-6.0.0\n" +
@@ -541,8 +542,90 @@ public class DoSectionTests extends AbstractClientYamlTestFragmentParserTestCase
                 emptyList(), emptyMap(), doSection.getApiCallSection().getNodeSelector());
     }
 
-    private Node nodeWithVersion(String version) {
-        return new Node(new HttpHost("dummy"), null, null, version, null);
+    private static Node nodeWithVersion(String version) {
+        return new Node(new HttpHost("dummy"), null, null, version, null, null);
+    }
+
+    public void testNodeSelectorByAttribute() throws IOException {
+        parser = createParser(YamlXContent.yamlXContent,
+                "node_selector:\n" +
+                "    attribute:\n" +
+                "        attr: val\n" +
+                "indices.get_field_mapping:\n" +
+                "    index: test_index"
+        );
+
+        DoSection doSection = DoSection.parse(parser);
+        assertNotSame(NodeSelector.ANY, doSection.getApiCallSection().getNodeSelector());
+        Node hasAttr = nodeWithAttributes(singletonMap("attr", singletonList("val")));
+        Node hasAttrWrongValue = nodeWithAttributes(singletonMap("attr", singletonList("notval")));
+        Node notHasAttr = nodeWithAttributes(singletonMap("notattr", singletonList("val")));
+        {
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(hasAttr);
+            nodes.add(hasAttrWrongValue);
+            nodes.add(notHasAttr);
+            doSection.getApiCallSection().getNodeSelector().select(nodes);
+            assertEquals(Arrays.asList(hasAttr), nodes);
+        }
+
+        parser = createParser(YamlXContent.yamlXContent,
+                "node_selector:\n" +
+                "    attribute:\n" +
+                "        attr: val\n" +
+                "        attr2: val2\n" +
+                "indices.get_field_mapping:\n" +
+                "    index: test_index"
+        );
+
+        DoSection doSectionWithTwoAttributes = DoSection.parse(parser);
+        assertNotSame(NodeSelector.ANY, doSection.getApiCallSection().getNodeSelector());
+        Node hasAttr2 = nodeWithAttributes(singletonMap("attr2", singletonList("val2")));
+        Map<String, List<String>> bothAttributes = new HashMap<>();
+        bothAttributes.put("attr", singletonList("val"));
+        bothAttributes.put("attr2", singletonList("val2"));
+        Node hasBoth = nodeWithAttributes(bothAttributes);
+        {
+            List<Node> nodes = new ArrayList<>();
+            nodes.add(hasAttr);
+            nodes.add(hasAttrWrongValue);
+            nodes.add(notHasAttr);
+            nodes.add(hasAttr2);
+            nodes.add(hasBoth);
+            doSectionWithTwoAttributes.getApiCallSection().getNodeSelector().select(nodes);
+            assertEquals(Arrays.asList(hasBoth), nodes);
+        }
+    }
+
+    private static Node nodeWithAttributes(Map<String, List<String>> attributes) {
+        return new Node(new HttpHost("dummy"), null, null, null, null, attributes);
+    }
+
+    public void testNodeSelectorByTwoThings() throws IOException {
+        parser = createParser(YamlXContent.yamlXContent,
+                "node_selector:\n" +
+                "    version: 5.2.0-6.0.0\n" +
+                "    attribute:\n" +
+                "        attr: val\n" +
+                "indices.get_field_mapping:\n" +
+                "    index: test_index"
+        );
+
+        DoSection doSection = DoSection.parse(parser);
+        assertNotSame(NodeSelector.ANY, doSection.getApiCallSection().getNodeSelector());
+        Node both = nodeWithVersionAndAttributes("5.2.1", singletonMap("attr", singletonList("val")));
+        Node badVersion = nodeWithVersionAndAttributes("5.1.1", singletonMap("attr", singletonList("val")));
+        Node badAttr = nodeWithVersionAndAttributes("5.2.1", singletonMap("notattr", singletonList("val")));
+        List<Node> nodes = new ArrayList<>();
+        nodes.add(both);
+        nodes.add(badVersion);
+        nodes.add(badAttr);
+        doSection.getApiCallSection().getNodeSelector().select(nodes);
+        assertEquals(Arrays.asList(both), nodes);
+    }
+
+    private static Node nodeWithVersionAndAttributes(String version, Map<String, List<String>> attributes) {
+        return new Node(new HttpHost("dummy"), null, null, version, null, attributes);
     }
 
     private void assertJsonEquals(Map<String, Object> actual, String expected) throws IOException {