IGNITE-26717 [ducktests] add separate module for RU

Author: maksaska

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

@@ -0,0 +1,111 @@
+# 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 provides test coverage for cluster rebalancing during rolling upgrade workflows.
+"""
+from ducktape.mark import defaults
+
+from ignitetest.services.ignite import IgniteService
+from ignitetest.services.utils.control_utility import ControlUtility
+from ignitetest.services.utils.ignite_configuration.discovery import from_ignite_cluster
+from ignitetest.tests.rebalance.persistent_test import await_and_check_rebalance
+from ignitetest.tests.rebalance.util import NUM_NODES, BaseRebalanceTest, start_ignite, await_rebalance_start, \
+    get_result
+from ignitetest.tests.util import preload_data
+from ignitetest.utils import cluster, ignite_versions
+from ignitetest.utils.version import LATEST, DEV_BRANCH, IgniteVersion
+
+
+class RollingUpgradeRebalanceTest(BaseRebalanceTest):
+    """
+    Test validates the rebalance workflow and includes the following steps:
+        * Start the cluster.
+        * Populate data using IgniteClientApp.
+        * Trigger a rebalance by joining a node running the target version, and wait for the rebalance to complete.
+    """
+    @cluster(num_nodes=NUM_NODES)
+    @ignite_versions(str(LATEST))
+    @defaults(upgrade_version=[str(DEV_BRANCH)], force=[False], backups=[1], cache_count=[1], entry_count=[15_000],
+              entry_size=[50_000], preloaders=[1], thread_pool_size=[None], batch_size=[None],
+              batches_prefetch_count=[None], throttle=[None])
+    def test_in_memory(self, ignite_version, upgrade_version, force, backups, cache_count, entry_count, entry_size,
+                       preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle):
+        """
+        Tests rebalance in-memory.
+        """
+        return self.test_rebalance(ignite_version, upgrade_version, force, backups, cache_count, entry_count,
+                                   entry_size, preloaders, thread_pool_size, batch_size, batches_prefetch_count,
+                                   throttle, False)
+
+    @cluster(num_nodes=NUM_NODES)
+    @ignite_versions(str(LATEST))
+    @defaults(upgrade_version=[str(DEV_BRANCH)], force=[False], backups=[1], cache_count=[1], entry_count=[5_000],
+              entry_size=[50_000], preloaders=[1], thread_pool_size=[None], batch_size=[None],
+              batches_prefetch_count=[None], throttle=[None])
+    def test_pds(self, ignite_version, upgrade_version, force, backups, cache_count, entry_count, entry_size,
+                 preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle):
+        """
+        Tests rebalance with persistence.
+        """
+        return self.test_rebalance(ignite_version, upgrade_version, force, backups, cache_count, entry_count,
+                                   entry_size, preloaders, thread_pool_size, batch_size, batches_prefetch_count,
+                                   throttle, True)
+
+    def test_rebalance(self, ignite_version, upgrade_version, force, backups, cache_count, entry_count, entry_size,
+                       preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle, with_persistence):
+        """
+        Tests rebalance on node join with version upgrade.
+        """
+        reb_params = self.get_reb_params(thread_pool_size=thread_pool_size, batch_size=batch_size,
+                                         batches_prefetch_count=batches_prefetch_count, throttle=throttle,
+                                         persistent=with_persistence)
+
+        data_gen_params = self.get_data_gen_params(backups=backups, cache_count=cache_count, entry_count=entry_count,
+                                                   entry_size=entry_size, preloaders=preloaders)
+
+        ignites = start_ignite(self.test_context, ignite_version, reb_params, data_gen_params)
+
+        control_utility = ControlUtility(ignites)
+
+        if with_persistence:
+            control_utility.activate()
+
+        preload_time = preload_data(
+            self.test_context,
+            ignites.config._replace(client_mode=True, discovery_spi=from_ignite_cluster(ignites)),
+            data_gen_params=data_gen_params)
+
+        new_node = IgniteService(self.test_context, ignites.config._replace(discovery_spi=from_ignite_cluster(ignites)),
+                                 num_nodes=1, modules=reb_params.modules)
+
+        control_utility.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring, force)
+
+        new_node.config._replace(version=upgrade_version)
+
+        new_node.start()
+
+        if with_persistence:
+            control_utility.add_to_baseline(new_node.nodes)
+
+            await_and_check_rebalance(new_node)
+
+            control_utility.deactivate()
+        else:
+            await_rebalance_start(new_node)
+
+            new_node.await_rebalance()
+
+        return get_result(new_node.nodes, preload_time, cache_count, entry_count, entry_size)

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

@@ -16,8 +16,6 @@
 """
 Module contains rolling upgrade tests
 """
-from time import monotonic
-
 from ducktape.mark import defaults, matrix
 
 from ignitetest.services.ignite import IgniteService
@@ -34,64 +32,51 @@ class RollingUpgradeTest(IgniteTest):
     Tests validates rolling upgrade
     """
     @cluster(num_nodes=NUM_NODES)
-    @ignite_versions(str(DEV_BRANCH))
-    @defaults(init_version=[str(LATEST)], upgrade_version=[str(DEV_BRANCH)])
-    @matrix(upgrade_coordinator=[True, False])
-    def test_rolling_upgrade(self, ignite_version, init_version, upgrade_version, upgrade_coordinator=False,
-                             force_upgrade=False):
-        self.logger.info(f"Initiating Rolling Upgrade test from {init_version} to {upgrade_version} "
-                         f"starting from coordinator [{upgrade_coordinator}]")
-
-        results = {}
+    @ignite_versions(str(LATEST))
+    @defaults(upgrade_version=[str(DEV_BRANCH)], force=[False])
+    @matrix(upgrade_coordinator_first=[True, False])
+    def test_rolling_upgrade(self, ignite_version, upgrade_version, upgrade_coordinator_first, force):
+        self.logger.info(f"Initiating Rolling Upgrade test from {ignite_version} to {upgrade_version} "
+                         f"starting from coordinator [{upgrade_coordinator_first}]")
 
-        ignites = self.start_ignite_cluster(init_version, results)
+        ignites = self.start_ignite_cluster(ignite_version)
 
         control_sh = ControlUtility(ignites)
 
-        if not force_upgrade:
-            control_sh.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring)
+        control_sh.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring, force)
 
-        self.upgrade_ignite_cluster(ignites, upgrade_version, upgrade_coordinator, results)
+        self.upgrade_ignite_cluster(ignites, upgrade_version, upgrade_coordinator_first)
 
-        if not force_upgrade:
-            control_sh.disable_rolling_upgrade()
+        control_sh.disable_rolling_upgrade()
 
         ignites.stop()
 
-        return results
-
-    def start_ignite_cluster(self, ignite_version: str, results):
+    def start_ignite_cluster(self, ignite_version: str):
         self.logger.info("Cluster start-up.")
 
         ignite_cfg = IgniteConfiguration(
             version=IgniteVersion(ignite_version),
             metric_exporters={"org.apache.ignite.spi.metric.jmx.JmxMetricExporterSpi"})
 
-        start = monotonic()
-
         ignites = IgniteService(self.test_context, ignite_cfg, num_nodes=self.test_context.expected_num_nodes)
 
         ignites.start()
 
-        results['Ignite cluster start time (s)'] = round(monotonic() - start, 1)
-
         ignites.await_event(f"Topology snapshot \\[ver={self.test_context.expected_num_nodes}",
                             ignites.startup_timeout_sec, from_the_beginning=True)
 
-        self.logger.info(f"Initial cluster is up [nodes={self.test_context.expected_num_nodes}].")
+        self.logger.info(f"Initial cluster is up [nodes={len(ignites.nodes)}].")
 
         return ignites
 
-    def upgrade_ignite_cluster(self, ignites: IgniteService, upgrade_version: str, upgrade_coordinator: bool, results):
+    def upgrade_ignite_cluster(self, ignites: IgniteService, upgrade_version: str, upgrade_coordinator_first: bool):
         self.logger.info(f"Starting rolling upgrade.")
 
         ignites.config = IgniteConfiguration(version=IgniteVersion(upgrade_version))
 
         ignite_upgraded = 0
 
-        start = monotonic()
-
-        for ignite in ignites.nodes if upgrade_coordinator else reversed(ignites.nodes):
+        for ignite in ignites.nodes if upgrade_coordinator_first else reversed(ignites.nodes):
             self.logger.debug(f"Updating {ignites.who_am_i(ignite)}")
 
             ignites.stop_node(ignite)
@@ -105,6 +90,4 @@ def upgrade_ignite_cluster(self, ignites: IgniteService, upgrade_version: str, u
             ignites.await_event(f"Topology snapshot \\[ver={exp_topology}", ignites.startup_timeout_sec,
                                 from_the_beginning=True)
 
-        results['Ignite cluster upgrade time (s)'] = round(monotonic() - start, 1)
-
         self.logger.info(f"Upgrade is complete.")

modules/ducktests/tests/ignitetest/tests/rolling_upgrade/rolling_upgrade_test.py renamed to modules/ducktests/tests/ignitetest/rolling_upgrade/rolling_upgrade_test.py

@@ -231,12 +231,18 @@ def snapshot_check(self, snapshot_name: str):
 
         return res
 
-    def enable_rolling_upgrade(self, target_version: str):
+    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.
         """
-        result = self.__run(f"--rolling-upgrade enable {target_version} --enable-experimental --yes")
+
+        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}"
 

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

@@ -19,13 +19,12 @@
 from ducktape.mark import defaults
 
 from ignitetest.services.ignite import IgniteService
-from ignitetest.services.utils.control_utility import ControlUtility
 from ignitetest.services.utils.ignite_configuration.discovery import from_ignite_cluster
 from ignitetest.tests.rebalance.util import start_ignite, get_result, TriggerEvent, NUM_NODES, \
     await_rebalance_start, BaseRebalanceTest
 from ignitetest.tests.util import preload_data
 from ignitetest.utils import cluster, ignite_versions
-from ignitetest.utils.version import DEV_BRANCH, LATEST, IgniteVersion
+from ignitetest.utils.version import DEV_BRANCH, LATEST
 
 
 class RebalanceInMemoryTest(BaseRebalanceTest):
@@ -36,44 +35,33 @@ class RebalanceInMemoryTest(BaseRebalanceTest):
     @ignite_versions(str(DEV_BRANCH), str(LATEST))
     @defaults(backups=[1], cache_count=[1], entry_count=[15_000], entry_size=[50_000], preloaders=[1],
               thread_pool_size=[None], batch_size=[None], batches_prefetch_count=[None], throttle=[None])
-    def test_node_join(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
+    def test_node_join(self, ignite_version,
+                       backups, cache_count, entry_count, entry_size, preloaders,
                        thread_pool_size, batch_size, batches_prefetch_count, throttle):
         """
         Tests rebalance on node join.
         """
-        return self.__run(ignite_version, TriggerEvent.NODE_JOIN, backups, cache_count, entry_count, entry_size,
-                          preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle)
+        return self.__run(ignite_version, TriggerEvent.NODE_JOIN,
+                          backups, cache_count, entry_count, entry_size, preloaders,
+                          thread_pool_size, batch_size, batches_prefetch_count, throttle)
 
     @cluster(num_nodes=NUM_NODES)
     @ignite_versions(str(DEV_BRANCH), str(LATEST))
     @defaults(backups=[1], cache_count=[1], entry_count=[15_000], entry_size=[50_000], preloaders=[1],
               thread_pool_size=[None], batch_size=[None], batches_prefetch_count=[None], throttle=[None])
-    def test_node_left(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
+    def test_node_left(self, ignite_version,
+                       backups, cache_count, entry_count, entry_size, preloaders,
                        thread_pool_size, batch_size, batches_prefetch_count, throttle):
         """
         Tests rebalance on node left.
         """
-        return self.__run(ignite_version, TriggerEvent.NODE_LEFT, backups, cache_count, entry_count, entry_size,
-                          preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle)
+        return self.__run(ignite_version, TriggerEvent.NODE_LEFT,
+                          backups, cache_count, entry_count, entry_size, preloaders,
+                          thread_pool_size, batch_size, batches_prefetch_count, throttle)
 
-    @cluster(num_nodes=NUM_NODES)
-    @ignite_versions(str(DEV_BRANCH))
-    @defaults(backups=[1], cache_count=[1], entry_count=[15_000], entry_size=[50_000], preloaders=[1],
-              thread_pool_size=[None], batch_size=[None], batches_prefetch_count=[None], throttle=[None],
-              init_version=str(LATEST), upgrade_version=str(DEV_BRANCH))
-    def test_node_join_with_upgrade(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
-                                    thread_pool_size, batch_size, batches_prefetch_count, throttle, init_version,
-                                    upgrade_version, force_upgrade=False):
-        """
-        Tests rebalance on node left.
-        """
-        return self.__run(init_version, TriggerEvent.NODE_JOIN, backups, cache_count, entry_count, entry_size,
-                          preloaders, thread_pool_size, batch_size, batches_prefetch_count, throttle, upgrade_version,
-                          force_upgrade)
-
-    def __run(self, ignite_version, trigger_event, backups, cache_count, entry_count, entry_size, preloaders,
-              thread_pool_size, batch_size, batches_prefetch_count, throttle, upgrade_version=None,
-              force_upgrade=False):
+    def __run(self, ignite_version, trigger_event,
+              backups, cache_count, entry_count, entry_size, preloaders,
+              thread_pool_size, batch_size, batches_prefetch_count, throttle):
         """
         Test performs rebalance test which consists of following steps:
             * Start cluster.
@@ -113,15 +101,6 @@ def __run(self, ignite_version, trigger_event, backups, cache_count, entry_count
             ignite = IgniteService(self.test_context,
                                    ignites.config._replace(discovery_spi=from_ignite_cluster(ignites)), num_nodes=1,
                                    modules=reb_params.modules)
-
-            if upgrade_version is not None:
-                if not force_upgrade:
-                    control_sh = ControlUtility(ignites)
-
-                    control_sh.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring)
-
-                ignite.config._replace(version=upgrade_version)
-
             ignite.start()
             rebalance_nodes = ignite.nodes
 

modules/ducktests/tests/ignitetest/tests/rebalance/in_memory_test.py

@@ -27,7 +27,7 @@
     get_result, check_type_of_rebalancing, await_rebalance_start, BaseRebalanceTest
 from ignitetest.tests.util import preload_data
 from ignitetest.utils import cluster, ignite_versions
-from ignitetest.utils.version import DEV_BRANCH, LATEST, IgniteVersion
+from ignitetest.utils.version import DEV_BRANCH, LATEST
 
 
 class RebalancePersistentTest(BaseRebalanceTest):
@@ -43,28 +43,6 @@ def test_node_join(self, ignite_version, backups, cache_count, entry_count, entr
         """
         Tests rebalance on node join.
         """
-        return self.test_join(ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
-                              thread_pool_size, batch_size, batches_prefetch_count, throttle)
-
-    @cluster(num_nodes=NUM_NODES)
-    @ignite_versions(str(DEV_BRANCH))
-    @defaults(backups=[1], cache_count=[1], entry_count=[5_000], entry_size=[50_000], preloaders=[1],
-              thread_pool_size=[None], batch_size=[None], batches_prefetch_count=[None], throttle=[None],
-              init_version=str(LATEST), upgrade_version=str(DEV_BRANCH))
-    def test_node_join_with_upgrade(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders,
-                                    thread_pool_size, batch_size, batches_prefetch_count, throttle, init_version,
-                                    upgrade_version, force_upgrade=False):
-        """
-        Tests rebalance on node join with version upgrade.
-        """
-        return self.test_join(init_version, backups, cache_count, entry_count, entry_size, preloaders, thread_pool_size,
-                              batch_size, batches_prefetch_count, throttle, upgrade_version, force_upgrade)
-
-    def test_join(self, ignite_version, backups, cache_count, entry_count, entry_size, preloaders, thread_pool_size,
-                  batch_size, batches_prefetch_count, throttle, upgrade_version=None, force_upgrade=False):
-        """
-        Tests rebalance on node join.
-        """
 
         reb_params = self.get_reb_params(trigger_event=TriggerEvent.NODE_JOIN, thread_pool_size=thread_pool_size,
                                          batch_size=batch_size, batches_prefetch_count=batches_prefetch_count,
@@ -86,13 +64,6 @@ def test_join(self, ignite_version, backups, cache_count, entry_count, entry_siz
 
         new_node = IgniteService(self.test_context, ignites.config._replace(discovery_spi=from_ignite_cluster(ignites)),
                                  num_nodes=1, modules=reb_params.modules)
-
-        if upgrade_version is not None:
-            if not force_upgrade:
-                control_utility.enable_rolling_upgrade(IgniteVersion(upgrade_version).vstring)
-
-            new_node.config._replace(version=upgrade_version)
-
         new_node.start()
 
         control_utility.add_to_baseline(new_node.nodes)