From b88a0eb9abe011ba323ae5801af0cbf060fa2245 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Micha=C5=82=20Chojnowski?= <michal.chojnowski@scylladb.com>
Date: Wed, 24 Jan 2024 00:47:47 +0100
Subject: [PATCH] test: pylib: add ScyllaCluster.wipe_sstables

Add a method which wipes sstables files for a particular table on a particular
stopped node.
---
 test/pylib/manager_client.py |  5 +++++
 test/pylib/scylla_cluster.py | 20 ++++++++++++++++++++
 2 files changed, 25 insertions(+)

diff --git a/test/pylib/manager_client.py b/test/pylib/manager_client.py
index 194565bfff..bc36521d8a 100644
--- a/test/pylib/manager_client.py
+++ b/test/pylib/manager_client.py
@@ -197,6 +197,11 @@ class ManagerClient():
         logger.debug("ManagerClient unpausing %s", server_id)
         await self.client.put_json(f"/cluster/server/{server_id}/unpause")
 
+    async def server_wipe_sstables(self, server_id: ServerNum, keyspace: str, table: str) -> None:
+        """Delete all files for the given table from the data directory"""
+        logger.debug("ManagerClient wiping sstables on %s, keyspace=%s, table=%s", server_id, keyspace, table)
+        await self.client.put_json(f"/cluster/server/{server_id}/wipe_sstables", {"keyspace": keyspace, "table": table})
+
     def _create_server_add_data(self, replace_cfg: Optional[ReplaceConfig] = None,
                                 cmdline: Optional[List[str]] = None,
                                 config: Optional[dict[str, Any]] = None,
diff --git a/test/pylib/scylla_cluster.py b/test/pylib/scylla_cluster.py
index 7144df3178..9256c0ac1d 100644
--- a/test/pylib/scylla_cluster.py
+++ b/test/pylib/scylla_cluster.py
@@ -31,6 +31,7 @@ import aiohttp
 import aiohttp.web
 import yaml
 import signal
+import glob
 
 from cassandra import InvalidRequest                    # type: ignore
 from cassandra import OperationTimedOut                 # type: ignore
@@ -267,6 +268,12 @@ class ScyllaServer:
         self.config["rpc_address"] = rpc_address
         self._write_config_file()
 
+    def wipe_sstables(self, keyspace: str, table: str):
+        root_dir = self.workdir/"data"
+        for f in glob.iglob(f"./{keyspace}/{table}-????????????????????????????????/**/*", root_dir=root_dir, recursive=True):
+            if ((root_dir/f).is_file()):
+                (root_dir/f).unlink()
+
     async def install_and_start(self, api: ScyllaRESTAPIClient, expected_error: Optional[str] = None) -> None:
         """Setup and start this server"""
         await self.install()
@@ -1001,6 +1008,14 @@ class ScyllaCluster:
         server.change_rpc_address(rpc_address)
         return rpc_address
 
+    def wipe_sstables(self, server_id: ServerNum, keyspace: str, table: str):
+        """Delete all sstable files for the given <node, keyspace, table>."""
+        assert server_id in self.servers, f"Server {server_id} unknown"
+        server = self.servers[server_id]
+        assert not server.is_running, f"Server {server_id} is running: stop it first and then delete its files"
+        self.is_dirty = True
+        server.wipe_sstables(keyspace, table)
+
 class ScyllaClusterManager:
     """Manages a Scylla cluster for running test cases
        Provides an async API for tests to request changes in the Cluster.
@@ -1130,6 +1145,7 @@ class ScyllaClusterManager:
         add_get('/cluster/server/{server_id}/get_log_filename', self._server_get_log_filename)
         add_get('/cluster/server/{server_id}/workdir', self._server_get_workdir)
         add_get('/cluster/server/{server_id}/exe', self._server_get_exe)
+        add_put('/cluster/server/{server_id}/wipe_sstables', self._cluster_server_wipe_sstables)
 
     async def _manager_up(self, _request) -> bool:
         return self.is_running
@@ -1390,6 +1406,10 @@ class ScyllaClusterManager:
     async def _server_get_exe(self, request: aiohttp.web.Request) -> str:
         return str(await self._server_get_attribute(request, "exe"))
 
+    async def _cluster_server_wipe_sstables(self, request: aiohttp.web.Request):
+        data = await request.json()
+        server_id = ServerNum(int(request.match_info["server_id"]))
+        return self.cluster.wipe_sstables(server_id, data["keyspace"], data["table"])
 
 @asynccontextmanager
 async def get_cluster_manager(test_uname: str, clusters: Pool[ScyllaCluster], test_path: str) \