QA: Switch rolling upgrade to 3 nodes

qa/rolling-upgrade/build.gradle

@@ -30,6 +30,26 @@ task bwcTest {
 }
 
 for (Version version : bwcVersions.wireCompatible) {
+  /*
+   * The goal here is to:
+   * <ul>
+   *  <li>start three nodes on the old version
+   *  <li>run tests with systemProperty 'tests.rest.suite', 'old_cluster'
+   *  <li>shut down one node
+   *  <li>start a node with the new version
+   *  <li>run tests with systemProperty 'tests.rest.suite', 'mixed_cluster'
+   *  <li>shut down one node on the old version
+   *  <li>start a node with the new version
+   *  <li>run tests with systemProperty 'tests.rest.suite', 'mixed_cluster' again
+   *  <li>shut down the last node with the old version
+   *  <li>start a node with the new version
+   *  <li>run tests with systemProperty 'tests.rest.suite', 'upgraded_cluster'
+   *  <li>shut down the entire cluster
+   * </ul>
+   *
+   * Be careful: gradle dry run spits out tasks in the wrong order but,
+   * strangely, running the tasks works properly.
+   */
   String baseName = "v${version}"
 
   Task oldClusterTest = tasks.create(name: "${baseName}#oldClusterTest", type: RestIntegTestTask) {
@@ -39,8 +59,8 @@ for (Version version : bwcVersions.wireCompatible) {
   Object extension = extensions.findByName("${baseName}#oldClusterTestCluster")
   configure(extensions.findByName("${baseName}#oldClusterTestCluster")) {
     bwcVersion = version
-    numBwcNodes = 2
-    numNodes = 2
+    numBwcNodes = 3
+    numNodes = 3
     clusterName = 'rolling-upgrade'
     setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
     if (version.onOrAfter('5.3.0')) {
@@ -53,43 +73,57 @@ for (Version version : bwcVersions.wireCompatible) {
     systemProperty 'tests.rest.suite', 'old_cluster'
   }
 
-  Task mixedClusterTest = tasks.create(name: "${baseName}#mixedClusterTest", type: RestIntegTestTask)
+  Closure configureUpgradeCluster = {String name, Task lastRunner, int stopNode, Closure unicastSeed ->
+    configure(extensions.findByName("${baseName}#${name}")) {
+      dependsOn lastRunner, "${baseName}#oldClusterTestCluster#node${stopNode}.stop"
+      clusterName = 'rolling-upgrade'
+      unicastTransportUri = { seedNode, node, ant -> unicastSeed() }
+      minimumMasterNodes = { 3 }
+      /* Override the data directory so the new node always gets the node we
+      * just stopped's data directory. */
+      dataDir = { nodeNumber -> oldClusterTest.nodes[stopNode].dataDir }
+      setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+    }
+  }
 
-  configure(extensions.findByName("${baseName}#mixedClusterTestCluster")) {
-    dependsOn oldClusterTestRunner, "${baseName}#oldClusterTestCluster#node1.stop"
-    clusterName = 'rolling-upgrade'
-    unicastTransportUri = { seedNode, node, ant -> oldClusterTest.nodes.get(0).transportUri() }
-    minimumMasterNodes = { 2 }
-    /* Override the data directory so the new node always gets the node we
-     * just stopped's data directory. */
-    dataDir = { nodeNumber -> oldClusterTest.nodes[1].dataDir }
-    setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+  Task oneThirdUpgradedTest = tasks.create(name: "${baseName}#oneThirdUpgradedTest", type: RestIntegTestTask)
+
+  configureUpgradeCluster("oneThirdUpgradedTestCluster", oldClusterTestRunner,
+      0, { oldClusterTest.nodes.get(1).transportUri() })
+
+  Task oneThirdUpgradedTestRunner = tasks.getByName("${baseName}#oneThirdUpgradedTestRunner")
+  oneThirdUpgradedTestRunner.configure {
+    systemProperty 'tests.rest.suite', 'mixed_cluster'
+    systemProperty 'tests.first_round', 'true'
+    finalizedBy "${baseName}#oldClusterTestCluster#node1.stop"
   }
 
-  Task mixedClusterTestRunner = tasks.getByName("${baseName}#mixedClusterTestRunner")
-  mixedClusterTestRunner.configure {
+  Task twoThirdsUpgradedTest = tasks.create(name: "${baseName}#twoThirdsUpgradedTest", type: RestIntegTestTask)
+
+  configureUpgradeCluster("twoThirdsUpgradedTestCluster", oneThirdUpgradedTestRunner,
+      1, { oneThirdUpgradedTest.nodes.get(0).transportUri() })
+
+  Task twoThirdsUpgradedTestRunner = tasks.getByName("${baseName}#twoThirdsUpgradedTestRunner")
+  twoThirdsUpgradedTestRunner.configure {
     systemProperty 'tests.rest.suite', 'mixed_cluster'
-    finalizedBy "${baseName}#oldClusterTestCluster#node0.stop"
+    systemProperty 'tests.first_round', 'false'
+    finalizedBy "${baseName}#oldClusterTestCluster#node2.stop"
   }
 
   Task upgradedClusterTest = tasks.create(name: "${baseName}#upgradedClusterTest", type: RestIntegTestTask)
 
-  configure(extensions.findByName("${baseName}#upgradedClusterTestCluster")) {
-    dependsOn mixedClusterTestRunner, "${baseName}#oldClusterTestCluster#node0.stop"
-    clusterName = 'rolling-upgrade'
-    unicastTransportUri = { seedNode, node, ant -> mixedClusterTest.nodes.get(0).transportUri() }
-    minimumMasterNodes = { 2 }
-    /* Override the data directory so the new node always gets the node we
-     * just stopped's data directory. */
-    dataDir = { nodeNumber -> oldClusterTest.nodes[0].dataDir}
-    setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
-  }
+  configureUpgradeCluster("upgradedClusterTestCluster", twoThirdsUpgradedTestRunner,
+      2, { twoThirdsUpgradedTest.nodes.get(0).transportUri() })
 
   Task upgradedClusterTestRunner = tasks.getByName("${baseName}#upgradedClusterTestRunner")
   upgradedClusterTestRunner.configure {
     systemProperty 'tests.rest.suite', 'upgraded_cluster'
-    // only need to kill the mixed cluster tests node here because we explicitly told it to not stop nodes upon completion
-    finalizedBy "${baseName}#mixedClusterTestCluster#stop"
+    /*
+     * Force stopping all the upgraded nodes after the test runner
+     * so they are alive during the test.
+     */
+    finalizedBy "${baseName}#oneThirdUpgradedTestCluster#stop"
+    finalizedBy "${baseName}#twoThirdsUpgradedTestCluster#stop"
   }
 
   Task versionBwcTest = tasks.create(name: "${baseName}#bwcTest") {

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/AbstractRollingTestCase.java

@@ -0,0 +1,91 @@
+/*
+ * 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.upgrades;
+
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.elasticsearch.Version;
+import org.elasticsearch.action.support.PlainActionFuture;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.cluster.metadata.IndexMetaData;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.util.concurrent.AbstractRunnable;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.test.rest.yaml.ObjectPath;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Future;
+import java.util.function.Predicate;
+
+import static com.carrotsearch.randomizedtesting.RandomizedTest.randomAsciiOfLength;
+import static java.util.Collections.emptyMap;
+import static org.elasticsearch.cluster.routing.UnassignedInfo.INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.decider.EnableAllocationDecider.INDEX_ROUTING_ALLOCATION_ENABLE_SETTING;
+import static org.elasticsearch.cluster.routing.allocation.decider.MaxRetryAllocationDecider.SETTING_ALLOCATION_MAX_RETRY;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.notNullValue;
+
+public abstract class AbstractRollingTestCase extends ESRestTestCase {
+    protected enum ClusterType {
+        OLD,
+        MIXED,
+        UPGRADED;
+
+        public static ClusterType parse(String value) {
+            switch (value) {
+                case "old_cluster":
+                    return OLD;
+                case "mixed_cluster":
+                    return MIXED;
+                case "upgraded_cluster":
+                    return UPGRADED;
+                    default:
+                        throw new AssertionError("unknown cluster type: " + value);
+            }
+        }
+    }
+
+    protected static final ClusterType CLUSTER_TYPE = ClusterType.parse(System.getProperty("tests.rest.suite"));
+
+    @Override
+    protected final boolean preserveIndicesUponCompletion() {
+        return true;
+    }
+
+    @Override
+    protected final boolean preserveReposUponCompletion() {
+        return true;
+    }
+
+    @Override
+    protected final Settings restClientSettings() {
+        return Settings.builder().put(super.restClientSettings())
+            // increase the timeout here to 90 seconds to handle long waits for a green
+            // cluster health. the waits for green need to be longer than a minute to
+            // account for delayed shards
+            .put(ESRestTestCase.CLIENT_RETRY_TIMEOUT, "90s")
+            .put(ESRestTestCase.CLIENT_SOCKET_TIMEOUT, "90s")
+            .build();
+    }
+}

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/IndexingIT.java

@@ -0,0 +1,135 @@
+/*
+ * 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.upgrades;
+
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.common.Booleans;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Basic test that indexed documents survive the rolling restart. See
+ * {@link RecoveryIT} for much more in depth testing of the mechanism
+ * by which they survive.
+ */
+public class IndexingIT extends AbstractRollingTestCase {
+    public void testIndexing() throws IOException {
+        switch (CLUSTER_TYPE) {
+        case OLD:
+            break;
+        case MIXED:
+            Request waitForYellow = new Request("GET", "/_cluster/health");
+            waitForYellow.addParameter("wait_for_nodes", "3");
+            waitForYellow.addParameter("wait_for_status", "yellow");
+            client().performRequest(waitForYellow);
+            break;
+        case UPGRADED:
+            Request waitForGreen = new Request("GET", "/_cluster/health/test_index,index_with_replicas,empty_index");
+            waitForGreen.addParameter("wait_for_nodes", "3");
+            waitForGreen.addParameter("wait_for_status", "green");
+            // wait for long enough that we give delayed unassigned shards to stop being delayed
+            waitForGreen.addParameter("timeout", "70s");
+            waitForGreen.addParameter("level", "shards");
+            client().performRequest(waitForGreen);
+            break;
+        default:
+            throw new UnsupportedOperationException("Unknown cluster type [" + CLUSTER_TYPE + "]");
+        }
+
+        if (CLUSTER_TYPE == ClusterType.OLD) {
+            Request createTestIndex = new Request("PUT", "/test_index");
+            createTestIndex.setJsonEntity("{\"settings\": {\"index.number_of_replicas\": 0}}");
+            client().performRequest(createTestIndex);
+
+            String recoverQuickly = "{\"settings\": {\"index.unassigned.node_left.delayed_timeout\": \"100ms\"}}";
+            Request createIndexWithReplicas = new Request("PUT", "/index_with_replicas");
+            createIndexWithReplicas.setJsonEntity(recoverQuickly);
+            client().performRequest(createIndexWithReplicas);
+
+            Request createEmptyIndex = new Request("PUT", "/empty_index");
+            // Ask for recovery to be quick
+            createEmptyIndex.setJsonEntity(recoverQuickly);
+            client().performRequest(createEmptyIndex);
+
+            bulk("test_index", "_OLD", 5);
+            bulk("index_with_replicas", "_OLD", 5);
+        }
+
+        int expectedCount;
+        switch (CLUSTER_TYPE) {
+        case OLD:
+            expectedCount = 5;
+            break;
+        case MIXED:
+            if (Booleans.parseBoolean(System.getProperty("tests.first_round"))) {
+                expectedCount = 5;
+            } else {
+                expectedCount = 10;
+            }
+            break;
+        case UPGRADED:
+            expectedCount = 15;
+            break;
+        default:
+            throw new UnsupportedOperationException("Unknown cluster type [" + CLUSTER_TYPE + "]");
+        }
+
+        assertCount("test_index", expectedCount);
+        assertCount("index_with_replicas", 5);
+        assertCount("empty_index", 0);
+
+        if (CLUSTER_TYPE != ClusterType.OLD) {
+            bulk("test_index", "_" + CLUSTER_TYPE, 5);
+            Request toBeDeleted = new Request("PUT", "/test_index/doc/to_be_deleted");
+            toBeDeleted.addParameter("refresh", "true");
+            toBeDeleted.setJsonEntity("{\"f1\": \"delete-me\"}");
+            client().performRequest(toBeDeleted);
+            assertCount("test_index", expectedCount + 6);
+
+            Request delete = new Request("DELETE", "/test_index/doc/to_be_deleted");
+            delete.addParameter("refresh", "true");
+            client().performRequest(delete);
+
+            assertCount("test_index", expectedCount + 5);
+        }
+    }
+
+    private void bulk(String index, String valueSuffix, int count) throws IOException {
+        StringBuilder b = new StringBuilder();
+        for (int i = 0; i < count; i++) {
+            b.append("{\"index\": {\"_index\": \"").append(index).append("\", \"_type\": \"doc\"}}\n");
+            b.append("{\"f1\": \"v").append(i).append(valueSuffix).append("\", \"f2\": ").append(i).append("}\n");
+        }
+        Request bulk = new Request("POST", "/_bulk");
+        bulk.addParameter("refresh", "true");
+        bulk.setJsonEntity(b.toString());
+        client().performRequest(bulk);
+    }
+
+    private void assertCount(String index, int count) throws IOException {
+        Request searchTestIndexRequest = new Request("POST", "/" + index + "/_search");
+        searchTestIndexRequest.addParameter("filter_path", "hits.total");
+        Response searchTestIndexResponse = client().performRequest(searchTestIndexRequest);
+        assertEquals("{\"hits\":{\"total\":" + count + "}}",
+                EntityUtils.toString(searchTestIndexResponse.getEntity(), StandardCharsets.UTF_8));
+    }
+}