[TEST] run REST tests against multiple nodes (round-robin)

Multiple nodes are now started when running REST tests against the `TestCluster` (default randomized settings are now used instead of the hardcoded `1`)

Added also randomized round-robin based on all available nodes, and ability to provide multiple addresses when running tests against an external cluster to have the same behaviour

TESTING.asciidoc

@@ -184,7 +184,8 @@ The following are the options supported by the REST tests runner:
 
 * `tests.rest[true|false|host:port]`: determines whether the REST tests need
 to be run and if so whether to rely on an external cluster (providing host
-and port) or fire a test cluster (default)
+and port) or fire a test cluster (default). It's possible to provide a
+comma separated list of addresses to send requests in a round-robin fashion.
 * `tests.rest.suite`: comma separated paths of the test suites to be run
 (by default loaded from /rest-api-spec/test). It is possible to run only a subset
 of the tests providing a sub-folder or even a single yaml file (the default

src/test/java/org/elasticsearch/test/ElasticsearchIntegrationTest.java

@@ -850,6 +850,16 @@ public abstract class ElasticsearchIntegrationTest extends ElasticsearchTestCase
         return annotation == null ? -1 : annotation.numNodes();
     }
 
+    private int getMinNumNodes() {
+        ClusterScope annotation = getAnnotation(this.getClass());
+        return annotation == null ? TestCluster.DEFAULT_MIN_NUM_NODES : annotation.minNumNodes();
+    }
+
+    private int getMaxNumNodes() {
+        ClusterScope annotation = getAnnotation(this.getClass());
+        return annotation == null ? TestCluster.DEFAULT_MAX_NUM_NODES : annotation.maxNumNodes();
+    }
+
     /**
      * This method is used to obtain settings for the <tt>Nth</tt> node in the cluster.
      * Nodes in this cluster are associated with an ordinal number such that nodes can
@@ -880,7 +890,15 @@ public abstract class ElasticsearchIntegrationTest extends ElasticsearchTestCase
             };
         }
 
-        return new TestCluster(currentClusterSeed, numNodes, clusterName(scope.name(), ElasticsearchTestCase.CHILD_VM_ID, currentClusterSeed), nodeSettingsSource);
+        int minNumNodes, maxNumNodes;
+        if (numNodes >= 0) {
+            minNumNodes = maxNumNodes = numNodes;
+        } else {
+            minNumNodes = getMinNumNodes();
+            maxNumNodes = getMaxNumNodes();
+        }
+
+        return new TestCluster(currentClusterSeed, minNumNodes, maxNumNodes, clusterName(scope.name(), ElasticsearchTestCase.CHILD_VM_ID, currentClusterSeed), nodeSettingsSource);
     }
 
     /**
@@ -898,10 +916,23 @@ public abstract class ElasticsearchIntegrationTest extends ElasticsearchTestCase
 
         /**
          * Returns the number of nodes in the cluster. Default is <tt>-1</tt> which means
-         * a random number of nodes but at least <code>2</code></tt> is used./
+         * a random number of nodes is used, where the minimum and maximum number of nodes
+         * are either the specified ones or the default ones if not specified.
          */
         int numNodes() default -1;
 
+        /**
+         * Returns the minimum number of nodes in the cluster. Default is {@link TestCluster#DEFAULT_MIN_NUM_NODES}.
+         * Ignored when {@link ClusterScope#numNodes()} is set.
+         */
+        int minNumNodes() default TestCluster.DEFAULT_MIN_NUM_NODES;
+
+        /**
+         * Returns the maximum number of nodes in the cluster.  Default is {@link TestCluster#DEFAULT_MAX_NUM_NODES}.
+         * Ignored when {@link ClusterScope#numNodes()} is set.
+         */
+        int maxNumNodes() default TestCluster.DEFAULT_MAX_NUM_NODES;
+
         /**
          * Returns the transport client ratio. By default this returns <code>-1</code> which means a random
          * ratio in the interval <code>[0..1]</code> is used.

src/test/java/org/elasticsearch/test/TestCluster.java

@@ -104,7 +104,7 @@ public final class TestCluster implements Iterable<Client> {
     public static final String TESTS_ENABLE_MOCK_MODULES = "tests.enable_mock_modules";
 
     /**
-     *  A node level setting that holds a per node random seed that is consistent across node restarts
+     * A node level setting that holds a per node random seed that is consistent across node restarts
      */
     public static final String SETTING_CLUSTER_NODE_SEED = "test.cluster.node.seed";
 
@@ -112,6 +112,10 @@ public final class TestCluster implements Iterable<Client> {
 
     private static final boolean ENABLE_MOCK_MODULES = systemPropertyAsBoolean(TESTS_ENABLE_MOCK_MODULES, true);
 
+    static final int DEFAULT_MIN_NUM_NODES = 2;
+
+    static final int DEFAULT_MAX_NUM_NODES = 6;
+
     private static long clusterSeed() {
         String property = System.getProperty(TESTS_CLUSTER_SEED);
         if (!Strings.hasLength(property)) {
@@ -135,9 +139,6 @@ public final class TestCluster implements Iterable<Client> {
 
     private AtomicInteger nextNodeId = new AtomicInteger(0);
 
-    /* We have a fixed number of shared nodes that we keep around across tests */
-    private final int numSharedNodes;
-
     /* Each shared node has a node seed that is used to start up the node and get default settings
      * this is important if a node is randomly shut down in a test since the next test relies on a
      * fully shared cluster to be more reproducible */
@@ -147,18 +148,34 @@ public final class TestCluster implements Iterable<Client> {
 
     private final NodeSettingsSource nodeSettingsSource;
 
-    TestCluster(long clusterSeed, String clusterName) {
-        this(clusterSeed, -1, clusterName, NodeSettingsSource.EMPTY);
+    public TestCluster(long clusterSeed, String clusterName) {
+        this(clusterSeed, DEFAULT_MIN_NUM_NODES, DEFAULT_MAX_NUM_NODES, clusterName, NodeSettingsSource.EMPTY);
     }
 
-    public TestCluster(long clusterSeed, int numNodes, String clusterName) {
-        this(clusterSeed, numNodes, clusterName, NodeSettingsSource.EMPTY);
+    public TestCluster(long clusterSeed, int minNumNodes, int maxNumNodes, String clusterName) {
+        this(clusterSeed, minNumNodes, maxNumNodes, clusterName, NodeSettingsSource.EMPTY);
     }
 
-    TestCluster(long clusterSeed, int numNodes, String clusterName, NodeSettingsSource nodeSettingsSource) {
+    public TestCluster(long clusterSeed, int minNumNodes, int maxNumNodes, String clusterName, NodeSettingsSource nodeSettingsSource) {
         this.clusterName = clusterName;
+
+        if (minNumNodes < 0 || maxNumNodes < 0) {
+            throw new IllegalArgumentException("minimum and maximum number of nodes must be >= 0");
+        }
+
+        if (maxNumNodes < minNumNodes) {
+            throw new IllegalArgumentException("maximum number of nodes must be >= minimum number of nodes");
+        }
+
         Random random = new Random(clusterSeed);
-        numSharedNodes = numNodes == -1 ? 2 + random.nextInt(4) : numNodes; // at least 2 nodes if randomized
+
+        int numSharedNodes;
+        if (minNumNodes == maxNumNodes) {
+            numSharedNodes = minNumNodes;
+        } else {
+            numSharedNodes = minNumNodes + random.nextInt(maxNumNodes - minNumNodes);
+        }
+
         assert numSharedNodes >= 0;
         /*
          *  TODO 
@@ -193,7 +210,7 @@ public final class TestCluster implements Iterable<Client> {
 
     private static boolean isLocalTransportConfigured() {
         if ("local".equals(System.getProperty("es.node.mode", "network"))) {
-           return true;
+            return true;
         }
         return Boolean.parseBoolean(System.getProperty("es.node.local", "false"));
     }
@@ -204,7 +221,7 @@ public final class TestCluster implements Iterable<Client> {
         Settings settings = nodeSettingsSource.settings(nodeOrdinal);
         if (settings != null) {
             if (settings.get(CLUSTER_NAME_KEY) != null) {
-                throw new ElasticsearchIllegalStateException("Tests must not set a '"+CLUSTER_NAME_KEY+"' as a node setting set '" + CLUSTER_NAME_KEY + "': [" + settings.get(CLUSTER_NAME_KEY) + "]");
+                throw new ElasticsearchIllegalStateException("Tests must not set a '" + CLUSTER_NAME_KEY + "' as a node setting set '" + CLUSTER_NAME_KEY + "': [" + settings.get(CLUSTER_NAME_KEY) + "]");
             }
             builder.put(settings);
         }
@@ -220,7 +237,7 @@ public final class TestCluster implements Iterable<Client> {
         Builder builder = ImmutableSettings.settingsBuilder()
         /* use RAM directories in 10% of the runs */
                 //.put("index.store.type", random.nextInt(10) == 0 ? MockRamIndexStoreModule.class.getName() : MockFSIndexStoreModule.class.getName())
-                        // decrease the routing schedule so new nodes will be added quickly - some random value between 30 and 80 ms
+                // decrease the routing schedule so new nodes will be added quickly - some random value between 30 and 80 ms
                 .put("cluster.routing.schedule", (30 + random.nextInt(50)) + "ms")
                         // default to non gateway
                 .put("gateway.type", "none")
@@ -575,8 +592,7 @@ public final class TestCluster implements Iterable<Client> {
             node = (InternalNode) nodeBuilder().settings(node.settings()).settings(newSettings).node();
             resetClient();
         }
-        
-        
+
 
         @Override
         public void close() {
@@ -865,7 +881,7 @@ public final class TestCluster implements Iterable<Client> {
             nodeAndClient.restart(callback);
         }
     }
-   
+
     private void restartAllNodes(boolean rollingRestart, RestartCallback callback) throws Exception {
         ensureOpen();
         List<NodeAndClient> toRemove = new ArrayList<TestCluster.NodeAndClient>();
@@ -938,7 +954,7 @@ public final class TestCluster implements Iterable<Client> {
     public void fullRestart(RestartCallback function) throws Exception {
         restartAllNodes(false, function);
     }
-    
+
 
     private String getMasterName() {
         try {
@@ -1085,7 +1101,7 @@ public final class TestCluster implements Iterable<Client> {
         @Override
         public boolean apply(Settings settings) {
             return nodeNames.contains(settings.get("name"));
-            		
+
         }
     }
 

src/test/java/org/elasticsearch/test/rest/RestTestExecutionContext.java

@@ -29,6 +29,7 @@ import org.elasticsearch.test.rest.spec.RestSpec;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -50,9 +51,8 @@ public class RestTestExecutionContext implements Closeable {
 
     private RestResponse response;
 
-    public RestTestExecutionContext(String host, int port, RestSpec restSpec) throws RestException, IOException {
-
-        this.restClient = new RestClient(host, port, restSpec);
+    public RestTestExecutionContext(InetSocketAddress[] addresses, RestSpec restSpec) throws RestException, IOException {
+        this.restClient = new RestClient(addresses, restSpec);
         this.esVersion = restClient.getEsVersion();
     }
 

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

@@ -33,6 +33,7 @@ import org.elasticsearch.test.rest.spec.RestSpec;
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
 
@@ -47,35 +48,47 @@ public class RestClient implements Closeable {
     private final RestSpec restSpec;
     private final CloseableHttpClient httpClient;
 
-    private final String host;
-    private final int port;
+    private final InetSocketAddress[] addresses;
 
     private final String esVersion;
 
-    public RestClient(String host, int port, RestSpec restSpec) throws IOException, RestException {
+    public RestClient(InetSocketAddress[] addresses, RestSpec restSpec) throws IOException, RestException {
+        assert addresses.length > 0;
         this.restSpec = restSpec;
         this.httpClient = createHttpClient();
-        this.host = host;
-        this.port = port;
-        this.esVersion = readVersion();
-        logger.info("REST client initialized [{}:{}], elasticsearch version: [{}]", host, port, esVersion);
+        this.addresses = addresses;
+        this.esVersion = readAndCheckVersion();
+        logger.info("REST client initialized {}, elasticsearch version: [{}]", addresses, esVersion);
     }
 
-    private String readVersion() throws IOException, RestException {
+    private String readAndCheckVersion() throws IOException, RestException {
         //we make a manual call here without using callApi method, mainly because we are initializing
         //and the randomized context doesn't exist for the current thread (would be used to choose the method otherwise)
         RestApi restApi = restApi("info");
         assert restApi.getPaths().size() == 1;
         assert restApi.getMethods().size() == 1;
-        RestResponse restResponse = new RestResponse(httpRequestBuilder()
-                .path(restApi.getPaths().get(0))
-                .method(restApi.getMethods().get(0)).execute());
-        checkStatusCode(restResponse);
-        Object version = restResponse.evaluate("version.number");
-        if (version == null) {
-            throw new RuntimeException("elasticsearch version not found in the response");
+
+        String version = null;
+        for (InetSocketAddress address : addresses) {
+            RestResponse restResponse = new RestResponse(new HttpRequestBuilder(httpClient)
+                    .host(address.getHostName()).port(address.getPort())
+                    .path(restApi.getPaths().get(0))
+                    .method(restApi.getMethods().get(0)).execute());
+            checkStatusCode(restResponse);
+
+            Object latestVersion = restResponse.evaluate("version.number");
+            if (latestVersion == null) {
+                throw new RuntimeException("elasticsearch version not found in the response");
+            }
+            if (version == null) {
+                version = latestVersion.toString();
+            } else {
+                if (!latestVersion.equals(version)) {
+                    throw new IllegalArgumentException("provided nodes addresses run different elasticsearch versions");
+                }
+            }
         }
-        return version.toString();
+        return version;
     }
 
     public String getEsVersion() {
@@ -208,7 +221,9 @@ public class RestClient implements Closeable {
     }
 
     protected HttpRequestBuilder httpRequestBuilder() {
-        return new HttpRequestBuilder(httpClient).host(host).port(port);
+        //the address used is randomized between the available ones
+        InetSocketAddress address = RandomizedTest.randomFrom(addresses);
+        return new HttpRequestBuilder(httpClient).host(address.getHostName()).port(address.getPort());
     }
 
     protected CloseableHttpClient createHttpClient() {

src/test/java/org/elasticsearch/test/rest/junit/RestTestSuiteRunner.java

@@ -51,6 +51,7 @@ import org.junit.runners.model.Statement;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
@@ -181,35 +182,35 @@ public class RestTestSuiteRunner extends ParentRunner<RestTestCandidate> {
         this.testSectionRandomnessOverride = randomnessOverride;
         logger.info("Master seed: {}", SeedUtils.formatSeed(initialSeed));
 
-        String host;
-        int port;
+        List<InetSocketAddress> addresses = Lists.newArrayList();
         if (runMode == RunMode.TEST_CLUSTER) {
-            this.testCluster = new TestCluster(SHARED_CLUSTER_SEED, 1, clusterName("REST-tests", ElasticsearchTestCase.CHILD_VM_ID, SHARED_CLUSTER_SEED));
+            this.testCluster = new TestCluster(SHARED_CLUSTER_SEED, 1, 3,
+                    clusterName("REST-tests", ElasticsearchTestCase.CHILD_VM_ID, SHARED_CLUSTER_SEED));
             this.testCluster.beforeTest(runnerRandomness.getRandom(), 0.0f);
-            HttpServerTransport httpServerTransport = testCluster.getInstance(HttpServerTransport.class);
-            InetSocketTransportAddress inetSocketTransportAddress = (InetSocketTransportAddress) httpServerTransport.boundAddress().publishAddress();
-            host = inetSocketTransportAddress.address().getHostName();
-            port = inetSocketTransportAddress.address().getPort();
+            for (HttpServerTransport httpServerTransport : testCluster.getInstances(HttpServerTransport.class)) {
+                addresses.add(((InetSocketTransportAddress) httpServerTransport.boundAddress().publishAddress()).address());
+            }
         } else {
             this.testCluster = null;
             String testsMode = System.getProperty(REST_TESTS_MODE);
-            String[] split = testsMode.split(":");
-            if (split.length < 2) {
-                throw new InitializationError("address [" + testsMode + "] not valid");
-            }
-            host = split[0];
-            try {
-                port = Integer.valueOf(split[1]);
-            } catch(NumberFormatException e) {
-                throw new InitializationError("port is not valid, expected number but was [" + split[1] + "]");
+            String[] stringAddresses = testsMode.split(",");
+            for (String stringAddress : stringAddresses) {
+                String[] split = stringAddress.split(":");
+                if (split.length < 2) {
+                    throw new InitializationError("address [" + testsMode + "] not valid");
+                }
+                try {
+                    addresses.add(new InetSocketAddress(split[0], Integer.valueOf(split[1])));
+                } catch(NumberFormatException e) {
+                    throw new InitializationError("port is not valid, expected number but was [" + split[1] + "]");
+                }
             }
         }
 
         try {
             String[] specPaths = resolvePathsProperty(REST_TESTS_SPEC, DEFAULT_SPEC_PATH);
             RestSpec restSpec = RestSpec.parseFrom(DEFAULT_SPEC_PATH, specPaths);
-
-            this.restTestExecutionContext = new RestTestExecutionContext(host, port, restSpec);
+            this.restTestExecutionContext = new RestTestExecutionContext(addresses.toArray(new InetSocketAddress[addresses.size()]), restSpec);
             this.rootDescription = createRootDescription(getRootSuiteTitle());
             this.restTestCandidates = collectTestCandidates(rootDescription);
         } catch (InitializationError e) {

src/test/java/org/elasticsearch/tribe/TribeTests.java

@@ -52,7 +52,7 @@ public class TribeTests extends ElasticsearchIntegrationTest {
     @Before
     public void setupSecondCluster() {
         // create another cluster
-        cluster2 = new TestCluster(randomLong(), 2, cluster().getClusterName() + "-2");
+        cluster2 = new TestCluster(randomLong(), 2, 2, cluster().getClusterName() + "-2");
         cluster2.beforeTest(getRandom(), getPerTestTransportClientRatio());
         cluster2.ensureAtLeastNumNodes(2);