QA: Switch rolling upgrade to 3 nodes

Switches the rolling upgrade tests from upgrading two nodes to upgrading
three nodes which is much more realistic and much better able to find
unexpected bugs. It upgrades the nodes one at a time and runs tests
between each upgrade. As such this now has four test runs:

1. Old
2. One third upgraded
3. Two thirds upgraded
4. Upgraded

It sets system properties so the tests can figure out which stage they
are in. It reuses the same yml tests for the "one third" and "two
thirds" cases because they are *almost* the same case.

This rewrites the yml-based indexing tests to be Java based because the
yml-based tests can't handle different expected values for the counts.
And the indexing tests need that when they are run twice.

Closes #25336

Author: nik9000

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*'
     setting 'node.attr.gen', 'old'
@@ -54,45 +74,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*'
-    setting 'node.attr.gen', 'new'
+  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*'
-    setting 'node.attr.gen', 'new'
-  }
+  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));
+    }
+}