Backport rolling upgrade multi cluster module

x-pack/plugin/ccr/src/main/java/org/elasticsearch/xpack/ccr/action/TransportResumeFollowAction.java

@@ -243,7 +243,8 @@ static void validate(
         Settings leaderSettings = filter(leaderIndex.getSettings());
         Settings followerSettings = filter(followIndex.getSettings());
         if (leaderSettings.equals(followerSettings) == false) {
-            throw new IllegalArgumentException("the leader and follower index settings must be identical");
+            throw new IllegalArgumentException("the leader index setting[" + leaderSettings + "] and follower index settings [" +
+                followerSettings + "] must be identical");
         }
 
         // Validates if the current follower mapping is mergable with the leader mapping.
@@ -456,6 +457,11 @@ static Settings filter(Settings originalSettings) {
         settings.remove(IndexMetaData.SETTING_INDEX_PROVIDED_NAME);
         settings.remove(IndexMetaData.SETTING_CREATION_DATE);
 
+        // Follower index may be upgraded, while the leader index hasn't been upgraded, so it is expected
+        // that these settings are different:
+        settings.remove(IndexMetaData.SETTING_VERSION_UPGRADED);
+        settings.remove(IndexMetaData.SETTING_VERSION_UPGRADED_STRING);
+
         Iterator<String> iterator = settings.keys().iterator();
         while (iterator.hasNext()) {
             String key = iterator.next();

x-pack/plugin/ccr/src/test/java/org/elasticsearch/xpack/ccr/action/TransportResumeFollowActionTests.java

@@ -150,7 +150,10 @@ public void testValidation() throws IOException {
                 .put("index.analysis.analyzer.my_analyzer.type", "custom")
                 .put("index.analysis.analyzer.my_analyzer.tokenizer", "standard").build(), customMetaData);
             Exception e = expectThrows(IllegalArgumentException.class, () -> validate(request, leaderIMD, followIMD, UUIDs, null));
-            assertThat(e.getMessage(), equalTo("the leader and follower index settings must be identical"));
+            assertThat(e.getMessage(), equalTo("the leader index setting[{\"index.analysis.analyzer.my_analyzer.tokenizer\"" +
+                ":\"whitespace\",\"index.analysis.analyzer.my_analyzer.type\":\"custom\",\"index.number_of_shards\":\"5\"}] " +
+                "and follower index settings [{\"index.analysis.analyzer.my_analyzer.tokenizer\":\"standard\"," +
+                "\"index.analysis.analyzer.my_analyzer.type\":\"custom\",\"index.number_of_shards\":\"5\"}] must be identical"));
         }
         {
             // should fail because the following index does not have the following_index settings
@@ -242,6 +245,21 @@ public void testDynamicIndexSettingsAreClassified() {
         }
     }
 
+    public void testFilter() {
+        Settings.Builder settings = Settings.builder();
+        settings.put(CcrSettings.CCR_FOLLOWING_INDEX_SETTING.getKey(), "");
+        settings.put(IndexSettings.INDEX_SOFT_DELETES_SETTING.getKey(), "");
+        settings.put(IndexMetaData.SETTING_INDEX_VERSION_CREATED.getKey(), "");
+        settings.put(IndexMetaData.SETTING_INDEX_UUID, "");
+        settings.put(IndexMetaData.SETTING_INDEX_PROVIDED_NAME, "");
+        settings.put(IndexMetaData.SETTING_CREATION_DATE, "");
+        settings.put(IndexMetaData.SETTING_VERSION_UPGRADED, "");
+        settings.put(IndexMetaData.SETTING_VERSION_UPGRADED_STRING, "");
+
+        Settings result = TransportResumeFollowAction.filter(settings.build());
+        assertThat(result.size(), equalTo(0));
+    }
+
     private static IndexMetaData createIMD(String index,
                                            int numberOfShards,
                                            Settings settings,

x-pack/qa/rolling-upgrade-multi-cluster/build.gradle

@@ -0,0 +1,262 @@
+import org.elasticsearch.gradle.Version
+import org.elasticsearch.gradle.test.RestIntegTestTask
+
+apply plugin: 'elasticsearch.standalone-test'
+
+dependencies {
+    // "org.elasticsearch.plugin:x-pack-core:${version}" doesn't work with idea because the testArtifacts are also here
+    testCompile project(path: xpackModule('core'), configuration: 'default')
+    testCompile project(path: xpackModule('core'), configuration: 'testArtifacts') // to be moved in a later commit
+}
+
+// This is a top level task which we will add dependencies to below.
+// It is a single task that can be used to backcompat tests against all versions.
+task bwcTest {
+    description = 'Runs backwards compatibility tests.'
+    group = 'verification'
+}
+
+for (Version version : bwcVersions.wireCompatible) {
+    String taskPrefix = "v${version}"
+
+    // ============================================================================================
+    // Create leader cluster
+    // ============================================================================================
+
+    RestIntegTestTask leaderClusterTest = tasks.create(name: "${taskPrefix}#leader#clusterTest", type: RestIntegTestTask) {
+        mustRunAfter(precommit)
+    }
+
+    configure(extensions.findByName("${taskPrefix}#leader#clusterTestCluster")) {
+        bwcVersion = version
+        numBwcNodes = 3
+        numNodes = 3
+        clusterName = 'leader'
+        setting 'xpack.security.enabled', 'false'
+        setting 'xpack.monitoring.enabled', 'false'
+        setting 'xpack.ml.enabled', 'false'
+        setting 'xpack.watcher.enabled', 'false'
+        setting 'xpack.license.self_generated.type', 'trial'
+    }
+
+    Task leaderClusterTestRunner = tasks.getByName("${taskPrefix}#leader#clusterTestRunner")
+    leaderClusterTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'none'
+        systemProperty 'tests.rest.cluster_name', 'leader'
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(0).transportUri()}"
+    }
+
+    // ============================================================================================
+    // Create follower cluster
+    // ============================================================================================
+
+    RestIntegTestTask followerClusterTest = tasks.create(name: "${taskPrefix}#follower#clusterTest", type: RestIntegTestTask) {
+        mustRunAfter(precommit)
+    }
+
+    configure(extensions.findByName("${taskPrefix}#follower#clusterTestCluster")) {
+        dependsOn leaderClusterTestRunner
+        bwcVersion = version
+        numBwcNodes = 3
+        numNodes = 3
+        clusterName = 'follower'
+        setting 'xpack.security.enabled', 'false'
+        setting 'xpack.monitoring.enabled', 'false'
+        setting 'xpack.ml.enabled', 'false'
+        setting 'xpack.watcher.enabled', 'false'
+        setting 'xpack.license.self_generated.type', 'trial'
+    }
+
+    Task followerClusterTestRunner = tasks.getByName("${taskPrefix}#follower#clusterTestRunner")
+    followerClusterTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'none'
+        systemProperty 'tests.rest.cluster_name', 'follower'
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(0).transportUri()}"
+
+        systemProperty 'tests.follower_host', "${-> followerClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerClusterTest.nodes.get(0).transportUri()}"
+    }
+
+    // ============================================================================================
+    // Upgrade follower cluster
+    // ============================================================================================
+
+    Closure configureUpgradeCluster = {String prefix, String cluster, String name, Task lastRunner, int stopNode,
+                                       RestIntegTestTask clusterTest, Closure getOtherUnicastHostAddresses ->
+        configure(extensions.findByName("${prefix}#${cluster}#${name}")) {
+            dependsOn lastRunner, "${prefix}#${cluster}#clusterTestCluster#node${stopNode}.stop"
+            clusterName = cluster
+            otherUnicastHostAddresses = { getOtherUnicastHostAddresses() }
+            minimumMasterNodes = { 2 }
+            autoSetInitialMasterNodes = false
+            /* Override the data directory so the new node always gets the node we
+            * just stopped's data directory. */
+            dataDir = { nodeNumber -> clusterTest.nodes[stopNode].dataDir }
+            setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+            setting 'xpack.security.enabled', 'false'
+            setting 'xpack.monitoring.enabled', 'false'
+            setting 'xpack.ml.enabled', 'false'
+            setting 'xpack.watcher.enabled', 'false'
+            setting 'xpack.license.self_generated.type', 'trial'
+            setting 'node.name', "upgraded-node-${cluster}-${stopNode}"
+            setting 'node.attr.upgraded', 'true'
+        }
+    }
+
+    Task followerOneThirdUpgradedTest = tasks.create(name: "${taskPrefix}#follower#oneThirdUpgradedTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'follower', 'oneThirdUpgradedTestCluster', followerClusterTestRunner, 0, followerClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [followerClusterTest.nodes.get(1).transportUri(), followerClusterTest.nodes.get(2).transportUri()] })
+
+    Task followerOneThirdUpgradedTestRunner = tasks.getByName("${taskPrefix}#follower#oneThirdUpgradedTestRunner")
+    followerOneThirdUpgradedTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'one_third'
+        systemProperty 'tests.rest.cluster_name', 'follower'
+
+        systemProperty 'tests.follower_host', "${-> followerClusterTest.nodes.get(1).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerClusterTest.nodes.get(1).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(0).transportUri()}"
+
+        finalizedBy "${taskPrefix}#follower#clusterTestCluster#node1.stop"
+    }
+
+    Task followerTwoThirdsUpgradedTest = tasks.create(name: "${taskPrefix}#follower#twoThirdsUpgradedTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'follower', 'twoThirdsUpgradedTestCluster', followerOneThirdUpgradedTestRunner, 1, followerClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [followerClusterTest.nodes.get(2).transportUri(), followerOneThirdUpgradedTest.nodes.get(0).transportUri()] })
+
+    Task followerTwoThirdsUpgradedTestRunner = tasks.getByName("${taskPrefix}#follower#twoThirdsUpgradedTestRunner")
+    followerTwoThirdsUpgradedTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'two_third'
+        systemProperty 'tests.rest.cluster_name', 'follower'
+
+        systemProperty 'tests.follower_host', "${-> followerClusterTest.nodes.get(2).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerClusterTest.nodes.get(2).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(0).transportUri()}"
+
+        finalizedBy "${taskPrefix}#follower#clusterTestCluster#node2.stop"
+    }
+
+    Task followerUpgradedClusterTest = tasks.create(name: "${taskPrefix}#follower#upgradedClusterTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'follower', 'upgradedClusterTestCluster', followerTwoThirdsUpgradedTestRunner, 2, followerClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [followerOneThirdUpgradedTest.nodes.get(0).transportUri(), followerTwoThirdsUpgradedTest.nodes.get(0).transportUri()] })
+
+    Task followerUpgradedClusterTestRunner = tasks.getByName("${taskPrefix}#follower#upgradedClusterTestRunner")
+    followerUpgradedClusterTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'all'
+        systemProperty 'tests.rest.cluster_name', 'follower'
+
+        systemProperty 'tests.follower_host', "${-> followerOneThirdUpgradedTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerOneThirdUpgradedTest.nodes.get(0).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(0).transportUri()}"
+
+        // This is needed, otherwise leader node 0 will stop after the leaderClusterTestRunner task has run.
+        // Here it is ok to stop, because in the next task, the leader node 0 gets upgraded.
+        finalizedBy "v${version}#leader#clusterTestCluster#node0.stop"
+    }
+
+    // ============================================================================================
+    // Upgrade leader cluster
+    // ============================================================================================
+
+    Task leaderOneThirdUpgradedTest = tasks.create(name: "${taskPrefix}#leader#oneThirdUpgradedTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'leader', 'oneThirdUpgradedTestCluster', followerUpgradedClusterTestRunner, 0, leaderClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [leaderClusterTest.nodes.get(1).transportUri(), leaderClusterTest.nodes.get(2).transportUri()] })
+
+    Task leaderOneThirdUpgradedTestRunner = tasks.getByName("${taskPrefix}#leader#oneThirdUpgradedTestRunner")
+    leaderOneThirdUpgradedTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'one_third'
+        systemProperty 'tests.rest.cluster_name', 'leader'
+
+        systemProperty 'tests.follower_host', "${-> followerUpgradedClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerUpgradedClusterTest.nodes.get(0).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderClusterTest.nodes.get(2).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderClusterTest.nodes.get(2).transportUri()}"
+
+        finalizedBy "${taskPrefix}#leader#clusterTestCluster#node1.stop"
+    }
+
+    Task leaderTwoThirdsUpgradedTest = tasks.create(name: "${taskPrefix}#leader#twoThirdsUpgradedTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'leader', 'twoThirdsUpgradedTestCluster', leaderOneThirdUpgradedTestRunner, 1, leaderClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [leaderClusterTest.nodes.get(2).transportUri(), leaderOneThirdUpgradedTest.nodes.get(0).transportUri()] })
+
+    Task leaderTwoThirdsUpgradedTestRunner = tasks.getByName("${taskPrefix}#leader#twoThirdsUpgradedTestRunner")
+    leaderTwoThirdsUpgradedTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'two_third'
+        systemProperty 'tests.rest.cluster_name', 'leader'
+
+        systemProperty 'tests.follower_host', "${-> followerUpgradedClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerUpgradedClusterTest.nodes.get(0).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderOneThirdUpgradedTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderOneThirdUpgradedTest.nodes.get(0).transportUri()}"
+
+        finalizedBy "${taskPrefix}#leader#clusterTestCluster#node2.stop"
+    }
+
+    Task leaderUpgradedClusterTest = tasks.create(name: "${taskPrefix}#leader#upgradedClusterTest", type: RestIntegTestTask)
+
+    configureUpgradeCluster(taskPrefix, 'leader', "upgradedClusterTestCluster", leaderTwoThirdsUpgradedTestRunner, 2, leaderClusterTest,
+            // Use all running nodes as seed nodes so there is no race between pinging and the tests
+            { [leaderOneThirdUpgradedTest.nodes.get(0).transportUri(), leaderTwoThirdsUpgradedTest.nodes.get(0).transportUri()] })
+
+    Task leaderUpgradedClusterTestRunner = tasks.getByName("${taskPrefix}#leader#upgradedClusterTestRunner")
+    leaderUpgradedClusterTestRunner.configure {
+        systemProperty 'tests.rest.upgrade_state', 'all'
+        systemProperty 'tests.rest.cluster_name', 'leader'
+
+        systemProperty 'tests.follower_host', "${-> followerUpgradedClusterTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.follower_remote_cluster_seed', "${-> followerUpgradedClusterTest.nodes.get(0).transportUri()}"
+
+        systemProperty 'tests.leader_host', "${-> leaderTwoThirdsUpgradedTest.nodes.get(0).httpUri()}"
+        systemProperty 'tests.leader_remote_cluster_seed', "${-> leaderTwoThirdsUpgradedTest.nodes.get(0).transportUri()}"
+
+        /*
+         * Force stopping all the upgraded nodes after the test runner
+         * so they are alive during the test.
+         */
+        finalizedBy "${taskPrefix}#follower#oneThirdUpgradedTestCluster#stop"
+        finalizedBy "${taskPrefix}#follower#twoThirdsUpgradedTestCluster#stop"
+        finalizedBy "${taskPrefix}#follower#upgradedClusterTestCluster#stop"
+        finalizedBy "${taskPrefix}#leader#oneThirdUpgradedTestCluster#stop"
+        finalizedBy "${taskPrefix}#leader#twoThirdsUpgradedTestCluster#stop"
+    }
+
+    if (project.bwc_tests_enabled) {
+        Task versionBwcTest = tasks.create(name: "${taskPrefix}#bwcTest") {
+            dependsOn = [leaderUpgradedClusterTest]
+        }
+        bwcTest.dependsOn(versionBwcTest)
+    }
+}
+
+unitTest.enabled = false // no unit tests for rolling upgrades, only the rest integration test
+
+// basic integ tests includes testing bwc against the most recent version
+task integTest {
+    if (project.bwc_tests_enabled) {
+        for (final def version : bwcVersions.unreleasedWireCompatible) {
+            dependsOn "v${version}#bwcTest"
+        }
+    }
+}
+check.dependsOn(integTest)