IGNITE-26717 [ducktests] Add ducktest for Rolling Upgrade (#12451)

Authored-by: Maksim Davydov <[email protected]>

Author: maksaska

modules/ducktests/src/main/java/org/apache/ignite/internal/ducktest/tests/persistence_upgrade_test/DataLoaderAndCheckerApplication.java

@@ -34,17 +34,21 @@ public class DataLoaderAndCheckerApplication extends IgniteAwareApplication {
     /** {@inheritDoc} */
     @Override public void run(JsonNode jNode) throws IgniteInterruptedCheckedException {
         boolean check = jNode.get("check").asBoolean();
+        int backups = jNode.path("backups").asInt(0);
+        int entryCnt = jNode.path("entryCount").asInt(10_000);
 
         markInitialized();
         waitForActivation();
 
         CacheConfiguration<Integer, CustomObject> cacheCfg = new CacheConfiguration<>("cache");
 
+        cacheCfg.setBackups(backups);
+
         IgniteCache<Integer, CustomObject> cache = ignite.getOrCreateCache(cacheCfg);
 
         log.info(check ? "Checking..." : " Preparing...");
 
-        for (int i = 0; i < 10_000; i++) {
+        for (int i = 0; i < entryCnt; i++) {
             CustomObject obj = new CustomObject(i);
 
             if (!check)

modules/ducktests/tests/ignitetest/services/utils/control_utility.py

@@ -231,6 +231,33 @@ def snapshot_check(self, snapshot_name: str):
 
         return res
 
+    def enable_rolling_upgrade(self, target_version: str, force: bool = False):
+        """
+        Enable Rolling Upgrade with the target Ignite version.
+        :param target_version: Target Ignite version.
+        :param force: If {@code true}, skips target version compatibility checks and forcibly enables rolling upgrade.
+                    This flag does not override an already active upgrade configuration.
+        """
+
+        if force:
+            result = self.__run(f"--rolling-upgrade enable {target_version} --force --enable-experimental --yes")
+        else:
+            result = self.__run(f"--rolling-upgrade enable {target_version} --enable-experimental --yes")
+
+        assert "Rolling upgrade enabled" in result, f"Unexpected response: {result}"
+
+        return result
+
+    def disable_rolling_upgrade(self):
+        """
+        Disable Rolling Upgrade.
+        """
+        result = self.__run(f"--rolling-upgrade disable --enable-experimental --yes")
+
+        assert "Rolling upgrade disabled" in result, f"Unexpected response: {result}"
+
+        return result
+
     def start_performance_statistics(self):
         """
         Start performance statistics collecting in the cluster.

modules/ducktests/tests/ignitetest/services/utils/ignite_aware.py

@@ -142,21 +142,31 @@ def await_stopped(self):
         self.logger.info("Waiting for IgniteAware(s) to stop ...")
 
         for node in self.nodes:
-            stopped = self.wait_node(node, timeout_sec=self.shutdown_timeout_sec)
-            assert stopped, "Node %s's worker thread did not stop in %d seconds" % \
-                            (str(node.account), self.shutdown_timeout_sec)
+            self.await_stopped_node(node)
 
-        for node in self.nodes:
-            wait_until(lambda: not self.alive(node), timeout_sec=self.shutdown_timeout_sec,
-                       err_msg="Node %s's remote processes failed to stop in %d seconds" %
-                               (str(node.account), self.shutdown_timeout_sec))
+    def await_stopped_node(self, node):
+        """
+        Awaits node stop finished.
+        """
+        self.logger.info(f"Waiting for {self.who_am_i(node)}) to stop ...")
+
+        stopped = self.wait_node(node, timeout_sec=self.shutdown_timeout_sec)
+        assert stopped, "Node %s's worker thread did not stop in %d seconds" % \
+                        (str(node.account), self.shutdown_timeout_sec)
+
+        wait_until(lambda: not self.alive(node), timeout_sec=self.shutdown_timeout_sec,
+                   err_msg="Node %s's remote processes failed to stop in %d seconds" %
+                           (str(node.account), self.shutdown_timeout_sec))
 
     def stop_node(self, node, force_stop=False, **kwargs):
         pids = self.pids(node, self.main_java_class)
 
         for pid in pids:
             node.account.signal(pid, signal.SIGKILL if force_stop else signal.SIGTERM, allow_fail=False)
 
+        if not force_stop:
+            self.await_stopped_node(node)
+
     def clean(self, **kwargs):
         self.__restore_iptables()
 

modules/ducktests/tests/ignitetest/services/utils/ignite_configuration/discovery.py

@@ -19,6 +19,9 @@
 
 from abc import ABCMeta, abstractmethod
 
+from typing import List
+
+from ignitetest.services.ignite import IgniteService
 from ignitetest.services.utils.ignite_aware import IgniteAwareService
 from ignitetest.services.zk.zookeeper import ZookeeperService
 
@@ -27,6 +30,7 @@ class DiscoverySpi(metaclass=ABCMeta):
     """
     Abstract class for DiscoverySpi.
     """
+
     @property
     @abstractmethod
     def type(self):
@@ -45,6 +49,7 @@ class ZookeeperDiscoverySpi(DiscoverySpi):
     """
     ZookeeperDiscoverySpi.
     """
+
     def __init__(self, zoo_service, root_path):
         self.connection_string = zoo_service.connection_string()
         self.port = zoo_service.settings.client_port
@@ -63,6 +68,7 @@ class TcpDiscoveryIpFinder(metaclass=ABCMeta):
     """
     Abstract class for TcpDiscoveryIpFinder.
     """
+
     @property
     @abstractmethod
     def type(self):
@@ -81,6 +87,7 @@ class TcpDiscoveryVmIpFinder(TcpDiscoveryIpFinder):
     """
     IpFinder with static ips, obtained from cluster nodes.
     """
+
     def __init__(self, nodes=None):
         self.addresses = TcpDiscoveryVmIpFinder.__get_addresses(nodes) if nodes else None
 
@@ -102,6 +109,7 @@ class TcpDiscoverySpi(DiscoverySpi):
     """
     TcpDiscoverySpi.
     """
+
     def __init__(self, ip_finder=TcpDiscoveryVmIpFinder(), port=47500, port_range=100, local_address=None):
         self.ip_finder = ip_finder
         self.port = port
@@ -134,6 +142,23 @@ def from_ignite_cluster(cluster, subset=None):
     return TcpDiscoverySpi(ip_finder=TcpDiscoveryVmIpFinder(nodes))
 
 
+def from_ignite_services(ignite_service_list: List[IgniteService]):
+    """
+    Constructs a `TcpDiscoverySpi` instance from the provided Ignite services.
+    :param ignite_service_list: A list of `IgniteService` objects representing the cluster.
+    :return: A configured `TcpDiscoverySpi` containing the static IP addresses of the services.
+    """
+    assert isinstance(ignite_service_list, list)
+    assert all(isinstance(ignite_service, IgniteService) for ignite_service in ignite_service_list)
+
+    nodes = []
+
+    for ignite_service in ignite_service_list:
+        nodes.extend(ignite_service.nodes)
+
+    return TcpDiscoverySpi(ip_finder=TcpDiscoveryVmIpFinder(nodes))
+
+
 def from_zookeeper_cluster(cluster, root_path="/apacheIgnite"):
     """
     Form ZookeeperDiscoverySpi from zookeeper service cluster.

modules/ducktests/tests/ignitetest/tests/rolling_upgrade/__init__.py

@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF 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.
+
+"""
+This package contains rolling upgrade tests.
+"""

modules/ducktests/tests/ignitetest/tests/rolling_upgrade/add_remove_node_upgrade_test.py

@@ -0,0 +1,80 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF 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.
+
+"""
+Module contains rolling upgrade tests with new nodes introduced into the topology and older nodes gracefully removed.
+"""
+from ducktape.mark import defaults, matrix
+
+from ignitetest.services.ignite import IgniteService
+from ignitetest.services.utils.control_utility import ControlUtility
+from ignitetest.services.utils.ignite_configuration.discovery import from_ignite_services
+from ignitetest.tests.rebalance.persistent_test import await_and_check_rebalance
+from ignitetest.tests.rolling_upgrade.util import BaseRollingUpgradeTest, PRELOADERS_COUNT, NUM_NODES
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.version import LATEST, DEV_BRANCH, IgniteVersion
+
+
+class AddRemoveNodeUpgradeTest(BaseRollingUpgradeTest):
+
+    @cluster(num_nodes=2 * NUM_NODES + PRELOADERS_COUNT)
+    @ignite_versions(str(LATEST))
+    @matrix(with_persistence=[True, False])
+    @defaults(upgrade_version=[str(DEV_BRANCH)], force=[False], backups=[1], entry_count=[15_000])
+    def test_add_remove_rolling_upgrade(self, ignite_version, upgrade_version, force, with_persistence,
+                                        backups, entry_count):
+        node_count = (self.test_context.expected_num_nodes - PRELOADERS_COUNT) // 2
+
+        self.check_rolling_upgrade(ignite_version, upgrade_version, force, with_persistence,
+                                   backups, entry_count, self._upgrade_ignite_cluster, node_count)
+
+    def _upgrade_ignite_cluster(self, ignites, upgrade_version, force, with_persistence):
+        control_sh = ControlUtility(ignites)
+
+        control_sh.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring, force)
+
+        self.logger.info(f"Starting rolling upgrade.")
+
+        upgraded_nodes = []
+
+        for ignite in ignites.nodes:
+            new_node_cfg = ignites.config._replace(
+                version=IgniteVersion(upgrade_version),
+                discovery_spi=from_ignite_services([ignites] + upgraded_nodes)
+            )
+
+            new_node = IgniteService(self.test_context, new_node_cfg, num_nodes=1)
+
+            new_node.start()
+
+            control_sh = ControlUtility(new_node)
+
+            if with_persistence:
+                control_sh.add_to_baseline(new_node.nodes)
+
+            await_and_check_rebalance(new_node)
+
+            upgraded_nodes.append(new_node)
+
+            ignites.stop_node(ignite)
+
+            if with_persistence:
+                control_sh.remove_from_baseline([ignite])
+
+        self.logger.info(f"Cluster upgrade is complete.")
+
+        control_sh.disable_rolling_upgrade()
+
+        return upgraded_nodes

modules/ducktests/tests/ignitetest/tests/rolling_upgrade/in_place_node_upgrade_test.py

@@ -0,0 +1,64 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF 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.
+
+"""
+Module contains in-place rolling upgrade tests
+"""
+from ducktape.mark import matrix, defaults
+
+from ignitetest.services.utils.control_utility import ControlUtility
+from ignitetest.tests.rolling_upgrade.util import BaseRollingUpgradeTest, NUM_NODES
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.version import IgniteVersion, LATEST, DEV_BRANCH
+
+
+class InPlaceNodeUpgradeTest(BaseRollingUpgradeTest):
+    @cluster(num_nodes=NUM_NODES)
+    @ignite_versions(str(LATEST))
+    @matrix(with_persistence=[True, False], upgrade_coordinator_first=[True, False])
+    @defaults(upgrade_version=[str(DEV_BRANCH)], force=[False], backups=[1], entry_count=[15_000])
+    def test_in_place_rolling_upgrade(self, ignite_version, upgrade_version, force, with_persistence,
+                                      backups, entry_count, upgrade_coordinator_first):
+        self.upgrade_coordinator_first = upgrade_coordinator_first
+
+        self.check_rolling_upgrade(ignite_version, upgrade_version, force, with_persistence,
+                                   backups, entry_count, self._upgrade_ignite_cluster)
+
+    def _upgrade_ignite_cluster(self, ignites, upgrade_version, force, with_persistence):
+        control_sh = ControlUtility(ignites)
+
+        control_sh.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring, force)
+
+        self.logger.info(
+            f"Starting in-place rolling upgrade "
+            f"{'with coordinator going first' if self.upgrade_coordinator_first else 'from the last node in the ring'}"
+        )
+
+        ignites.config = ignites.config._replace(version=IgniteVersion(upgrade_version))
+
+        for ignite in ignites.nodes if self.upgrade_coordinator_first else reversed(ignites.nodes):
+            self.logger.debug(f"Upgrading {ignites.who_am_i(ignite)}")
+
+            ignites.stop_node(ignite)
+
+            ignites.start_node(ignite)
+
+            ignites.await_started([ignite])
+
+        self.logger.info(f"Cluster upgrade is complete.")
+
+        control_sh.disable_rolling_upgrade()
+
+        return [ignites]