Merge PR ceph#38835 into master

* refs/pull/38835/head:
	qa: add test for fs rm idempotency
	mon/FSCommands: restore idempotent behavior of fs rm
	qa: add ceph cmd helper
	qa: allow kwargs for raw_cluster_cmd

Reviewed-by: Rishabh Dave <[email protected]>

qa/tasks/ceph_manager.py

@@ -13,6 +13,7 @@
 import threading
 import traceback
 import os
+import shlex
 
 from io import BytesIO, StringIO
 from subprocess import DEVNULL
@@ -1326,6 +1327,17 @@ def tmp(x):
             except CommandFailedError:
                 self.log('Failed to get pg_num from pool %s, ignoring' % pool)
 
+    def ceph(self, cmd, **kwargs):
+        """
+        Simple Ceph admin command wrapper around run_cluster_cmd.
+        """
+
+        kwargs.pop('args', None)
+        args = shlex.split(cmd)
+        stdout = kwargs.pop('stdout', StringIO())
+        stderr = kwargs.pop('stderr', StringIO())
+        return self.run_cluster_cmd(args=args, stdout=stdout, stderr=stderr, **kwargs)
+
     def run_cluster_cmd(self, **kwargs):
         """
         Run a Ceph command and return the object representing the process
@@ -1346,12 +1358,13 @@ def run_cluster_cmd(self, **kwargs):
         kwargs['args'] = prefix + list(kwargs['args'])
         return self.controller.run(**kwargs)
 
-    def raw_cluster_cmd(self, *args):
+    def raw_cluster_cmd(self, *args, **kwargs):
         """
         Start ceph on a raw cluster.  Return count
         """
-        return self.run_cluster_cmd(**{'args': args,
-                                       'stdout': StringIO()}).stdout.getvalue()
+        stdout = kwargs.pop('stdout', StringIO())
+        p = self.run_cluster_cmd(args=args, stdout=stdout, **kwargs)
+        return p.stdout.getvalue()
 
     def raw_cluster_cmd_result(self, *args, **kwargs):
         """

qa/tasks/cephfs/filesystem.py

@@ -680,27 +680,19 @@ def run_client_payload(self, cmd):
         m.run_shell_payload(cmd)
         m.umount_wait(require_clean=True)
 
-    def destroy(self, reset_obj_attrs=True):
-        log.info(f'Destroying file system {self.name} and related pools')
+    def _remove_pool(self, name, **kwargs):
+        c = f'osd pool rm {name} {name} --yes-i-really-really-mean-it'
+        return self.mon_manager.ceph(c, **kwargs)
 
-        if self.dead():
-            log.debug('already dead...')
-            return
+    def rm(self, **kwargs):
+        c = f'fs rm {self.name} --yes-i-really-mean-it'
+        return self.mon_manager.ceph(c, **kwargs)
 
-        data_pools = self.get_data_pool_names(refresh=True)
-
-        # make sure no MDSs are attached to given FS.
-        self.mon_manager.raw_cluster_cmd('fs', 'fail', self.name)
-        self.mon_manager.raw_cluster_cmd(
-            'fs', 'rm', self.name, '--yes-i-really-mean-it')
-
-        self.mon_manager.raw_cluster_cmd('osd', 'pool', 'rm',
-            self.get_metadata_pool_name(), self.get_metadata_pool_name(),
-            '--yes-i-really-really-mean-it')
+    def remove_pools(self, data_pools):
+        self._remove_pool(self.get_metadata_pool_name())
         for poolname in data_pools:
             try:
-                self.mon_manager.raw_cluster_cmd('osd', 'pool', 'rm', poolname,
-                    poolname, '--yes-i-really-really-mean-it')
+                self._remove_pool(poolname)
             except CommandFailedError as e:
                 # EBUSY, this data pool is used by two metadata pools, let the
                 # 2nd pass delete it
@@ -709,6 +701,21 @@ def destroy(self, reset_obj_attrs=True):
                 else:
                     raise
 
+    def destroy(self, reset_obj_attrs=True):
+        log.info(f'Destroying file system {self.name} and related pools')
+
+        if self.dead():
+            log.debug('already dead...')
+            return
+
+        data_pools = self.get_data_pool_names(refresh=True)
+
+        # make sure no MDSs are attached to given FS.
+        self.fail()
+        self.rm()
+
+        self.remove_pools(data_pools)
+
         if reset_obj_attrs:
             self.id = None
             self.name = None

qa/tasks/cephfs/test_admin.py

@@ -5,7 +5,7 @@
 from teuthology.orchestra.run import CommandFailedError, Raw
 
 from tasks.cephfs.cephfs_test_case import CephFSTestCase
-from tasks.cephfs.filesystem import FileLayout
+from tasks.cephfs.filesystem import FileLayout, FSMissing
 from tasks.cephfs.fuse_mount import FuseMount
 from tasks.cephfs.caps_helper import CapsHelper
 
@@ -576,3 +576,29 @@ def setup_test_env(self, perm, paths=()):
         mounts = (self.mount_a, )
 
         return filepaths, filedata, mounts, keyring
+
+class TestAdminCommandIdempotency(CephFSTestCase):
+    """
+    Tests for administration command idempotency.
+    """
+
+    CLIENTS_REQUIRED = 0
+    MDSS_REQUIRED = 1
+
+    def test_rm_idempotency(self):
+        """
+        That a removing a fs twice is idempotent.
+        """
+
+        data_pools = self.fs.get_data_pool_names(refresh=True)
+        self.fs.fail()
+        self.fs.rm()
+        try:
+            self.fs.get_mds_map()
+        except FSMissing:
+            pass
+        else:
+            self.fail("get_mds_map should raise")
+        p = self.fs.rm()
+        self.assertIn("does not exist", p.stderr.getvalue())
+        self.fs.remove_pools(data_pools)

src/mon/FSCommands.cc

@@ -1355,8 +1355,11 @@ int FileSystemCommandHandler::is_op_allowed(
 
     auto fs = fsmap_copy.get_filesystem(fs_name);
     if (fs == nullptr) {
-      ss << "Filesystem not found: '" << fs_name << "'";
-      return -ENOENT;
+      /* let "fs rm" handle idempotent case where file system does not exist */
+      if (!(get_prefix() == "fs rm" && fsmap.get_filesystem(fs_name) == nullptr)) {
+        ss << "Filesystem not found: '" << fs_name << "'";
+        return -ENOENT;
+      }
     }
 
     if (!op->get_session()->fs_name_capable(fs_name, MON_CAP_W)) {