Use all available hosts in REST tests and allow for real master election (#21161)

Today we only use a single node to send requests to when we run REST tests.
In some cases we have more than one node (ie. in the BWC case) where we should
send requests to all nodes in a round-robin fashion. This change passes all
available node endpoints to the rest test.

Additionally, this change adds the setting of `discovery.zen.minimum_master_nodes`
to the cluster formation forcing the nodes to wait for all other nodes until the cluster
is formed. This allows for a more realistic master election and allows all master eligable
nodes to become master while before always the first node in the cluster became the master.

This also adds logging to each test run to log the master nodes version and the minimum node
version in the cluster to help debugging BWC test failures.

buildSrc/src/main/groovy/org/elasticsearch/gradle/test/ClusterFormationTasks.groovy

@@ -255,6 +255,12 @@ class ClusterFormationTasks {
                 'node.attr.testattr'           : 'test',
                 'repositories.url.allowed_urls': 'http://snapshot.test*'
         ]
+        // we set min master nodes to the total number of nodes in the cluster and
+        // basically skip initial state recovery to allow the cluster to form using a realistic master election
+        // this means all nodes must be up, join the seed node and do a master election. This will also allow new and
+        // old nodes in the BWC case to become the master
+        esConfig['discovery.zen.minimum_master_nodes'] = node.config.numNodes
+        esConfig['discovery.initial_state_timeout'] = '0s' // don't wait for state.. just start up quickly
         esConfig['node.max_local_storage_nodes'] = node.config.numNodes
         esConfig['http.port'] = node.config.httpPort
         esConfig['transport.tcp.port'] =  node.config.transportPort

buildSrc/src/main/groovy/org/elasticsearch/gradle/test/RestIntegTestTask.groovy

@@ -55,7 +55,9 @@ public class RestIntegTestTask extends RandomizedTestingTask {
         parallelism = '1'
         include('**/*IT.class')
         systemProperty('tests.rest.load_packaged', 'false')
-        systemProperty('tests.rest.cluster', "${-> nodes[0].httpUri()}")
+        // we pass all nodes to the rest cluster to allow the clients to round-robin between them
+        // this is more realistic than just talking to a single node
+        systemProperty('tests.rest.cluster', "${-> nodes.collect{it.httpUri()}.join(",")}")
         systemProperty('tests.config.dir', "${-> nodes[0].confDir}")
         // TODO: our "client" qa tests currently use the rest-test plugin. instead they should have their own plugin
         // that sets up the test cluster and passes this transport uri instead of http uri. Until then, we pass

test/framework/src/main/java/org/elasticsearch/test/rest/yaml/ClientYamlTestClient.java

@@ -31,6 +31,7 @@ import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.test.rest.yaml.restspec.ClientYamlSuiteRestApi;
@@ -66,34 +67,39 @@ public class ClientYamlTestClient {
         assert hosts.size() > 0;
         this.restSpec = restSpec;
         this.restClient = restClient;
-        this.esVersion = readAndCheckVersion(hosts);
+        Tuple<Version, Version> versionTuple = readMasterAndMinNodeVersion();
+        this.esVersion = versionTuple.v1();
+        Version masterVersion = versionTuple.v2();
+        // this will be logged in each test such that if something fails we get it in the logs for each test
+        logger.info("initializing client, minimum es version: [{}] master version: [{}] hosts: {}", esVersion, masterVersion, hosts);
     }
 
-    private Version readAndCheckVersion(List<HttpHost> hosts) throws IOException {
+    private Tuple<Version, Version> readMasterAndMinNodeVersion() throws IOException {
-        ClientYamlSuiteRestApi restApi = restApi("info");
+        // we simply go to the _cat/nodes API and parse all versions in the cluster
-        assert restApi.getPaths().size() == 1;
+        Response response = restClient.performRequest("GET", "/_cat/nodes", Collections.singletonMap("h", "version,master"));
-        assert restApi.getMethods().size() == 1;
+        ClientYamlTestResponse restTestResponse = new ClientYamlTestResponse(response);
-
+        String nodesCatResponse = restTestResponse.getBodyAsString();
-        String version = null;
+        String[] split = nodesCatResponse.split("\n");
-        for (HttpHost ignored : hosts) {
+        Version version = null;
-            //we don't really use the urls here, we rely on the client doing round-robin to touch all the nodes in the cluster
+        Version masterVersion = null;
-            String method = restApi.getMethods().get(0);
+        for (String perNode : split) {
-            String endpoint = restApi.getPaths().get(0);
+            final String[] versionAndMaster = perNode.split(" ");
-            Response response = restClient.performRequest(method, endpoint);
+            assert versionAndMaster.length == 2 : "invalid line: " + perNode + " length: " + versionAndMaster.length;
-            ClientYamlTestResponse restTestResponse = new ClientYamlTestResponse(response);
+            final Version currentVersion = Version.fromString(versionAndMaster[0]);
-            Object latestVersion = restTestResponse.evaluate("version.number");
+            final boolean master = versionAndMaster[1].trim().equals("*");
-            if (latestVersion == null) {
+            if (master) {
-                throw new RuntimeException("elasticsearch version not found in the response");
+                assert masterVersion == null;
+                masterVersion = currentVersion;
             }
             if (version == null) {
-                version = latestVersion.toString();
+                version = currentVersion;
             } else {
-                if (!latestVersion.equals(version)) {
+                if (version.onOrAfter(currentVersion)) {
-                    throw new IllegalArgumentException("provided nodes addresses run different elasticsearch versions");
+                    version = currentVersion;
                 }
             }
         }
-        return Version.fromString(version);
+        return new Tuple<>(version, masterVersion);
     }
 
     public Version getEsVersion() {