Merge pull request ceph#50569 from rishabh-d-dave/CephManager-in-CephFSTestCase

qa/cephfs: add helper methods in CephFSTestCase

Reviewed-by: Venky Shankar <[email protected]>

Author: rishabh-d-dave

qa/tasks/cephfs/caps_helper.py

@@ -156,12 +156,11 @@ def run_mon_cap_tests(self, fs, client_id):
 
         fs is any fs object so that ceph commands can be exceuted.
         """
-        get_cluster_cmd_op = fs.mon_manager.raw_cluster_cmd
-        keyring = get_cluster_cmd_op(args=f'auth get client.{client_id}')
+        keyring = fs.get_ceph_cmd_stdout(args=f'auth get client.{client_id}')
         moncap = get_mon_cap_from_keyring(keyring)
         keyring_path = fs.admin_remote.mktemp(data=keyring)
 
-        fsls = get_cluster_cmd_op(
+        fsls = fs.get_ceph_cmd_stdout(
             args=f'fs ls --id {client_id} -k {keyring_path}')
         log.info(f'output of fs ls cmd run by client.{client_id} -\n{fsls}')
 

qa/tasks/cephfs/cephfs_test_case.py

@@ -2,8 +2,7 @@
 import logging
 import os
 import re
-
-from shlex import split as shlex_split
+from io import StringIO
 
 from tasks.ceph_test_case import CephTestCase
 
@@ -49,7 +48,32 @@ def restore(self, mntobj):
         mntobj.hostfs_mntpt = self.hostfs_mntpt
 
 
-class CephFSTestCase(CephTestCase):
+class RunCephCmd:
+
+    def run_ceph_cmd(self, *args, **kwargs):
+        if kwargs.get('args') is None and args:
+            if len(args) == 1:
+                args = args[0]
+            kwargs['args'] = args
+        return self.mon_manager.run_cluster_cmd(**kwargs)
+
+    def get_ceph_cmd_result(self, *args, **kwargs):
+        if kwargs.get('args') is None and args:
+            if len(args) == 1:
+                args = args[0]
+            kwargs['args'] = args
+        return self.run_ceph_cmd(**kwargs).exitstatus
+
+    def get_ceph_cmd_stdout(self, *args, **kwargs):
+        if kwargs.get('args') is None and args:
+            if len(args) == 1:
+                args = args[0]
+            kwargs['args'] = args
+        kwargs['stdout'] = kwargs.pop('stdout', StringIO())
+        return self.run_ceph_cmd(**kwargs).stdout.getvalue()
+
+
+class CephFSTestCase(CephTestCase, RunCephCmd):
     """
     Test case for Ceph FS, requires caller to populate Filesystem and Mounts,
     into the fs, mount_a, mount_b class attributes (setting mount_b is optional)
@@ -90,25 +114,38 @@ def _remove_blocklist(self):
         # In case anything is in the OSD blocklist list, clear it out.  This is to avoid
         # the OSD map changing in the background (due to blocklist expiry) while tests run.
         try:
-            self.mds_cluster.mon_manager.run_cluster_cmd(args="osd blocklist clear")
+            self.run_ceph_cmd("osd blocklist clear")
         except CommandFailedError:
             # Fallback for older Ceph cluster
             try:
-                blocklist = json.loads(self.mds_cluster.mon_manager.raw_cluster_cmd("osd",
-                                      "dump", "--format=json-pretty"))['blocklist']
+                blocklist = json.loads(self.get_ceph_cmd_stdout("osd",
+                    "dump", "--format=json-pretty"))['blocklist']
                 log.info(f"Removing {len(blocklist)} blocklist entries")
                 for addr, blocklisted_at in blocklist.items():
-                    self.mds_cluster.mon_manager.raw_cluster_cmd("osd", "blocklist", "rm", addr)
+                    self.run_ceph_cmd("osd", "blocklist", "rm", addr)
             except KeyError:
                 # Fallback for more older Ceph clusters, who will use 'blacklist' instead.
-                blacklist = json.loads(self.mds_cluster.mon_manager.raw_cluster_cmd("osd",
-                                      "dump", "--format=json-pretty"))['blacklist']
+                blacklist = json.loads(self.get_ceph_cmd_stdout("osd",
+                    "dump", "--format=json-pretty"))['blacklist']
                 log.info(f"Removing {len(blacklist)} blacklist entries")
                 for addr, blocklisted_at in blacklist.items():
-                    self.mds_cluster.mon_manager.raw_cluster_cmd("osd", "blacklist", "rm", addr)
+                    self.run_ceph_cmd("osd", "blacklist", "rm", addr)
+
+    def _init_mon_manager(self):
+        # if vstart_runner.py has invoked this code
+        if 'Local' in str(type(self.ceph_cluster)):
+            from tasks.vstart_runner import LocalCephManager
+            self.mon_manager = LocalCephManager(ctx=self.ctx)
+        # else teuthology has invoked this code
+        else:
+            from tasks.ceph_manager import CephManager
+            self.mon_manager = CephManager(self.ceph_cluster.admin_remote,
+                ctx=self.ctx, logger=log.getChild('ceph_manager'))
 
     def setUp(self):
         super(CephFSTestCase, self).setUp()
+        self._init_mon_manager()
+        self.admin_remote = self.ceph_cluster.admin_remote
 
         self.config_set('mon', 'mon_allow_pool_delete', True)
 
@@ -154,7 +191,7 @@ def setUp(self):
         for entry in self.auth_list():
             ent_type, ent_id = entry['entity'].split(".")
             if ent_type == "client" and ent_id not in client_mount_ids and not (ent_id == "admin" or ent_id[:6] == 'mirror'):
-                self.mds_cluster.mon_manager.raw_cluster_cmd("auth", "del", entry['entity'])
+                self.run_ceph_cmd("auth", "del", entry['entity'])
 
         if self.REQUIRE_FILESYSTEM:
             self.fs = self.mds_cluster.newfs(create=True)
@@ -165,11 +202,11 @@ def setUp(self):
                        'osd', f'allow rw tag cephfs data={self.fs.name}',
                        'mds', 'allow']
 
-                if self.run_cluster_cmd_result(cmd) == 0:
+                if self.get_ceph_cmd_result(*cmd) == 0:
                     break
 
                 cmd[1] = 'add'
-                if self.run_cluster_cmd_result(cmd) != 0:
+                if self.get_ceph_cmd_result(*cmd) != 0:
                     raise RuntimeError(f'Failed to create new client {cmd[2]}')
 
             # wait for ranks to become active
@@ -182,9 +219,8 @@ def setUp(self):
         if self.REQUIRE_BACKUP_FILESYSTEM:
             if not self.REQUIRE_FILESYSTEM:
                 self.skipTest("backup filesystem requires a primary filesystem as well")
-            self.fs.mon_manager.raw_cluster_cmd('fs', 'flag', 'set',
-                                                'enable_multiple', 'true',
-                                                '--yes-i-really-mean-it')
+            self.run_ceph_cmd('fs', 'flag', 'set', 'enable_multiple', 'true',
+                              '--yes-i-really-mean-it')
             self.backup_fs = self.mds_cluster.newfs(name="backup_fs")
             self.backup_fs.wait_for_daemons()
 
@@ -220,9 +256,8 @@ def auth_list(self):
         """
         Convenience wrapper on "ceph auth ls"
         """
-        return json.loads(self.mds_cluster.mon_manager.raw_cluster_cmd(
-            "auth", "ls", "--format=json-pretty"
-        ))['auth_dump']
+        return json.loads(self.get_ceph_cmd_stdout("auth", "ls",
+            "--format=json-pretty"))['auth_dump']
 
     def assert_session_count(self, expected, ls_data=None, mds_id=None):
         if ls_data is None:
@@ -405,16 +440,6 @@ def _wait_random_subtrees(self, count, status=None, rank=None, path=None):
         except contextutil.MaxWhileTries as e:
             raise RuntimeError("rank {0} failed to reach desired subtree state".format(rank)) from e
 
-    def run_cluster_cmd(self, cmd):
-        if isinstance(cmd, str):
-            cmd = shlex_split(cmd)
-        return self.fs.mon_manager.raw_cluster_cmd(*cmd)
-
-    def run_cluster_cmd_result(self, cmd):
-        if isinstance(cmd, str):
-            cmd = shlex_split(cmd)
-        return self.fs.mon_manager.raw_cluster_cmd_result(*cmd)
-
     def create_client(self, client_id, moncap=None, osdcap=None, mdscap=None):
         if not (moncap or osdcap or mdscap):
             if self.fs:
@@ -432,5 +457,5 @@ def create_client(self, client_id, moncap=None, osdcap=None, mdscap=None):
         if mdscap:
             cmd += ['mds', mdscap]
 
-        self.run_cluster_cmd(cmd)
-        return self.run_cluster_cmd(f'auth get {self.client_name}')
+        self.run_ceph_cmd(*cmd)
+        return self.run_ceph_cmd(f'auth get {self.client_name}')