test(nemesis-refresh): add integration test for refresh monkey flow

Add test_refresh_monkey_flow that exercises the full nodetool refresh
path using a Docker-based Scylla node: writes data with cassandra-stress,
uploads sstables, runs nodetool refresh, and verifies the data is loaded
via CQL query.

Includes related changes to nemesis, sstable load utilities, and test
fixtures needed to support the refresh test flow.

Author: fruch

sdcm/nemesis/__init__.py

@@ -1674,37 +1674,43 @@ def disrupt_major_compaction(self):
         self._major_compaction()
 
     def disrupt_load_and_stream(self):
-        # Checking the columns number of keyspace1.standard1
-        self.log.debug("Prepare keyspace1.standard1 if it does not exist")
-        self._prepare_test_table(ks="keyspace1")
-        column_num = SstableLoadUtils.calculate_columns_count_in_table(self.target_node)
+        # Checking the columns number of keyspace_refresh.standard1
+        self.log.debug("Prepare keyspace_refresh.standard1 if it does not exist")
+        self._prepare_test_table(ks="keyspace_refresh")
+        column_num = SstableLoadUtils.calculate_columns_count_in_table(
+            self.target_node, keyspace_name="keyspace_refresh"
+        )
 
         # Run load-and-stream test on regular standard1 table of cassandra-stress.
         if column_num < 5:
             raise UnsupportedNemesis("Schema doesn't match the snapshot, not uploading")
 
         test_data = SstableLoadUtils.get_load_test_data_inventory(column_num, big_sstable=False, load_and_stream=True)
 
-        result = self.target_node.run_nodetool(sub_cmd="cfstats", args="keyspace1.standard1")
+        result = self.target_node.run_nodetool(sub_cmd="cfstats", args="keyspace_refresh.standard1")
 
         if result is not None and result.exit_status == 0:
             map_files_to_node = SstableLoadUtils.distribute_test_files_to_cluster_nodes(
                 nodes=self.cluster.data_nodes, test_data=test_data
             )
             for sstables_info, load_on_node in map_files_to_node:
                 self.actions_log.info(f"Uploading sstables to {load_on_node.name}")
-                SstableLoadUtils.upload_sstables(load_on_node, test_data=sstables_info, table_name="standard1")
+                SstableLoadUtils.upload_sstables(
+                    load_on_node, test_data=sstables_info, keyspace_name="keyspace_refresh", table_name="standard1"
+                )
                 # NOTE: on K8S logs may appear with a delay, so add a bigger timeout for it.
                 #       See https://github.com/scylladb/scylla-cluster-tests/issues/6314
                 kwargs = {"start_timeout": 1800, "end_timeout": 1800} if self._is_it_on_kubernetes() else {}
                 with self.action_log_scope(f"Loading and streaming sstables on {load_on_node.name} node"):
                     SstableLoadUtils.run_load_and_stream(load_on_node, **kwargs)
 
     def disrupt_nodetool_refresh(self, big_sstable: bool = False):
-        # Checking the columns number of keyspace1.standard1
-        self.log.debug("Prepare keyspace1.standard1 if it does not exist")
-        self._prepare_test_table(ks="keyspace1")
-        column_num = SstableLoadUtils.calculate_columns_count_in_table(self.target_node)
+        # Checking the columns number of keyspace_refresh.standard1
+        self.log.debug("Prepare keyspace_refresh.standard1 if it does not exist")
+        self._prepare_test_table(ks="keyspace_refresh")
+        column_num = SstableLoadUtils.calculate_columns_count_in_table(
+            self.target_node, keyspace_name="keyspace_refresh", table_name="standard1"
+        )
 
         # Note: when issue #6617 is fixed, we can try to load snapshot (cols=5) to a table (1 < cols < 5),
         #       expect that refresh will fail (no serious db error).
@@ -1715,14 +1721,14 @@ def disrupt_nodetool_refresh(self, big_sstable: bool = False):
             column_num, big_sstable=big_sstable, load_and_stream=False
         )
 
-        result = self.target_node.run_nodetool(sub_cmd="cfstats", args="keyspace1.standard1")
+        result = self.target_node.run_nodetool(sub_cmd="cfstats", args="keyspace_refresh.standard1")
 
         if result is not None and result.exit_status == 0:
             key = "0x32373131364f334f3830"
             # Check one special key before refresh, we will verify refresh by query in the end
             # Note: we can't DELETE the key before refresh, otherwise the old sstable won't be loaded
-            #       TRUNCATE can be used the clean the table, but we can't do it for keyspace1.standard1
-            query_verify = f"SELECT * FROM keyspace1.standard1 WHERE key={key}"
+            #       TRUNCATE can be used the clean the table, but we can't do it for keyspace_refresh.standard1
+            query_verify = f"SELECT * FROM keyspace_refresh.standard1 WHERE key={key}"
             result = self.target_node.run_cqlsh(query_verify)
             if "(0 rows)" in result.stdout:
                 self.log.debug("Key %s does not exist before refresh", key)
@@ -2169,7 +2175,7 @@ def _prepare_test_table(self, ks="keyspace1"):
         """Populate ``ks.standard1`` with 400 K rows via cassandra-stress."""
         stress_cmd = (
             "cassandra-stress write n=400000 cl=QUORUM -mode native cql3 "
-            f"-schema 'replication(strategy=NetworkTopologyStrategy,"
+            f"-schema 'keyspace={ks} replication(strategy=NetworkTopologyStrategy,"
             f"replication_factor={self.tester.reliable_replication_factor})' -log interval=5"
         )
         cs_thread = self.tester.run_stress_thread(

sdcm/remote/local_cmd_runner.py

@@ -56,6 +56,7 @@ def run(
         watchers: Optional[List[StreamWatcher]] = None,
         change_context: bool = False,
         timestamp_logs: bool = False,
+        user: Optional[str] = None,
     ) -> Result:
         watchers = self._setup_watchers(
             verbose=verbose, log_file=log_file, additional_watchers=watchers, timestamp_logs=timestamp_logs

sdcm/utils/sstable/load_utils.py

@@ -32,7 +32,7 @@ class SstableLoadUtils:
 
     @staticmethod
     def calculate_columns_count_in_table(
-        target_node, keyspace_name: str = "keyspace1", table_name: str = "standard1"
+        target_node, keyspace_name: str = "keyspace_refresh", table_name: str = "standard1"
     ) -> int:
         query_cmd = f"SELECT * FROM {keyspace_name}.{table_name} LIMIT 1"
         result = target_node.run_cqlsh(query_cmd)
@@ -67,7 +67,7 @@ def distribute_test_files_to_cluster_nodes(cls, nodes, test_data: List[TestDataI
     def upload_sstables(
         node,
         test_data: TestDataInventory,
-        keyspace_name: str = "keyspace1",
+        keyspace_name: str = "keyspace_refresh",
         table_name=None,
         create_schema: bool = False,
         is_cloud_cluster=False,
@@ -129,7 +129,12 @@ def upload_sstables(
 
     @classmethod
     def run_load_and_stream(
-        cls, node, keyspace_name: str = "keyspace1", table_name: str = "standard1", start_timeout=60, end_timeout=600
+        cls,
+        node,
+        keyspace_name: str = "keyspace_refresh",
+        table_name: str = "standard1",
+        start_timeout=60,
+        end_timeout=600,
     ):
         """runs load and stream using API request and waits for it to finish"""
         with wait_for_log_lines(
@@ -155,7 +160,7 @@ def run_refresh(node, test_data: namedtuple) -> Iterable[str]:
         # Find the compaction output that reported about the resharding
 
         system_log_follower = node.follow_system_log(patterns=[r"Resharded.*"])
-        node.run_nodetool(sub_cmd="refresh", args="-- keyspace1 standard1")
+        node.run_nodetool(sub_cmd="refresh", args="-- keyspace_refresh standard1")
         return system_log_follower
 
     @staticmethod
@@ -169,24 +174,24 @@ def validate_resharding_after_refresh(node, system_log_follower):
         # Validate that files after resharding were saved in the "upload" folder.
         # Example of compaction output:
 
-        #   scylla[6653]:  [shard 0] compaction - [Reshard keyspace1.standard1 3cad4140-f8c3-11ea-acb1-000000000002]
+        #   scylla[6653]:  [shard 0] compaction - [Reshard keyspace_refresh.standard1 3cad4140-f8c3-11ea-acb1-000000000002]
         #   Resharded 1 sstables to [
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-9-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-10-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-11-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-12-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-13-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-22-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-15-big-Data.db:level=0,
-        #   /var/lib/scylla/data/keyspace1/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-16-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-9-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-10-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-11-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-12-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-13-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-22-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-15-big-Data.db:level=0,
+        #   /var/lib/scylla/data/keyspace_refresh/standard1-9fbed8d0f8c211ea9bb1000000000000/upload/md-16-big-Data.db:level=0,
         #   ]. 91MB to 92MB (~100% of original) in 5009ms = 18MB/s. ~370176 total partitions merged to 370150
 
         Starting with Scylla 4.7 messages have changed to the following:
-        [shard 1] sstables_loader - Loading new SSTables for keyspace=keyspace1, table=standard1, ...
-        [shard 1] database - Resharding 223kB for keyspace1.standard1
-        [shard 1] database - Resharded 223kB for keyspace1.standard1 in 0.14 seconds, 1MB/s
-        [shard 1] database - Loaded 16 SSTables into /var/lib/scylla/data/keyspace1/standard1-eb0401905d8311ecb391aa52ebf0b3e1
-        [shard 1] sstables_loader - Done loading new SSTables for keyspace=keyspace1, table=standard1, ...
+        [shard 1] sstables_loader - Loading new SSTables for keyspace=keyspace_refresh, table=standard1, ...
+        [shard 1] database - Resharding 223kB for keyspace_refresh.standard1
+        [shard 1] database - Resharded 223kB for keyspace_refresh.standard1 in 0.14 seconds, 1MB/s
+        [shard 1] database - Loaded 16 SSTables into /var/lib/scylla/data/keyspace_refresh/standard1-eb0401905d8311ecb391aa52ebf0b3e1
+        [shard 1] sstables_loader - Done loading new SSTables for keyspace=keyspace_refresh, table=standard1, ...
 
         So, there is no per-file paths anymore for resharding log messages, only root dir path.
         """
@@ -230,7 +235,7 @@ def get_load_test_data_inventory(
     def create_keyspace(
         cls,
         node,
-        keyspace_name: str = "keyspace1",
+        keyspace_name: str = "keyspace_refresh",
         strategy: str = "NetworkTopologyStrategy",
         replication_factor: int = 1,
     ):
@@ -245,7 +250,9 @@ def create_table_for_load(cls, node, schema_file_and_path: str, session):
         session.execute(schema.replace("\n", ""))
 
     @classmethod
-    def validate_data_count_after_upload(cls, node, keyspace_name: str = "keyspace1", table_name: str = "standard1"):
+    def validate_data_count_after_upload(
+        cls, node, keyspace_name: str = "keyspace_refresh", table_name: str = "standard2"
+    ):
         result = node.run_cqlsh(f"consistency QUORUM;SELECT COUNT(*) FROM {keyspace_name}.{table_name}")
 
         next_line_is_result = False

unit_tests/integration/test_nemesis_refresh.py

@@ -0,0 +1,73 @@
+import logging
+
+import pytest
+
+from sdcm.utils.sstable.load_utils import SstableLoadUtils
+
+from sdcm.stress_thread import CassandraStressThread
+from unit_tests.lib.dummy_remote import LocalLoaderSetDummy
+
+pytestmark = [
+    pytest.mark.usefixtures("events"),
+    pytest.mark.integration,
+    pytest.mark.xdist_group("docker_heavy"),
+]
+
+
+@pytest.mark.integration
+def test_refresh_monkey_flow(docker_scylla, params, events, request):
+    """test the flow of refrash monkey locall with a docker base scylla"""
+
+    loader_set = LocalLoaderSetDummy(params=params)
+
+    ks = "keyspace_refresh"
+    # Checking the columns number of keyspace_refresh.standard1
+    stress_cmd = (
+        "cassandra-stress write n=40000 cl=ONE -mode native cql3 "
+        f"-schema 'keyspace={ks} replication(strategy=NetworkTopologyStrategy,"
+        f"replication_factor=1)' -log interval=5"
+    )
+    cs_thread = CassandraStressThread(loader_set, stress_cmd, node_list=[docker_scylla], timeout=120, params=params)
+
+    def cleanup_thread():
+        cs_thread.kill()
+
+    request.addfinalizer(cleanup_thread)
+
+    cs_thread.run()
+
+    output, _ = cs_thread.parse_results()
+    print(output)
+    column_num = SstableLoadUtils.calculate_columns_count_in_table(
+        docker_scylla, keyspace_name="keyspace_refresh", table_name="standard1"
+    )
+
+    assert column_num
+    test_data = SstableLoadUtils.get_load_test_data_inventory(column_num, big_sstable=False, load_and_stream=False)
+
+    result = docker_scylla.run_nodetool(sub_cmd="cfstats", args="keyspace_refresh.standard1")
+
+    if result is not None and result.exit_status == 0:
+        key = "0x32373131364f334f3830"
+        # Check one special key before refresh, we will verify refresh by query in the end
+        # Note: we can't DELETE the key before refresh, otherwise the old sstable won't be loaded
+        #       TRUNCATE can be used the clean the table, but we can't do it for keyspace_refresh.standard1
+        query_verify = f"SELECT * FROM keyspace_refresh.standard1 WHERE key={key}"
+        result = docker_scylla.run_cqlsh(query_verify)
+        if "(0 rows)" in result.stdout:
+            logging.debug("Key %s does not exist before refresh", key)
+        else:
+            logging.debug("Key %s already exists before refresh", key)
+
+        # Executing rolling refresh one by one
+        for node in [docker_scylla]:
+            SstableLoadUtils.upload_sstables(
+                node,
+                test_data=test_data[0],
+                table_name="standard1",
+                is_cloud_cluster=False,
+            )
+            SstableLoadUtils.run_refresh(node, test_data=test_data[0])
+        # Verify that the special key is loaded by SELECT query
+        result = docker_scylla.run_cqlsh(query_verify)
+        assert "(1 rows)" in result.stdout, f"The key {key} is not loaded by `nodetool refresh`"

unit_tests/lib/fake_cluster.py

@@ -37,7 +37,7 @@ def __init__(self, system_log):
     def run(self, *args, **kwargs):
         lines = [
             "[shard 11] sstables_loader - load_and_stream: started ops_uuid=a2661989-6836-418f-aa67-2c5466499848, process [0-1] out",
-            "[shard  2] sstables_loader - Done loading new SSTables for keyspace=keyspace1, table=standard1, load_and_stream=true, "
+            "[shard  2] sstables_loader - Done loading new SSTables for keyspace=keyspace_refresh, table=standard1, load_and_stream=true, "
             "primary_replica_only=false, status=succeeded",
         ]
         for line in lines: