Merge PR ceph#58966 into main

* refs/pull/58966/head:
	qa: test 'ceph fs snapshot mirror ls' command
	doc: update docs for 'fs snapshot mirror ls' command
	cephfs_mirror: ceph fs snapshot mirror ls command

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

Author: vshankar

doc/cephfs/cephfs-mirroring.rst

@@ -189,6 +189,12 @@ To configure a directory for mirroring, run a command of the following form:
 
    ceph fs snapshot mirror add <fs_name> <path>
 
+To list the configured directories, run a command of the following form:
+
+.. prompt:: bash $
+
+   ceph fs snapshot mirror ls <fs_name>
+
 To stop mirroring directory snapshots, run a command of the following form:
 
 .. prompt:: bash $

qa/tasks/cephfs/test_mirroring.py

@@ -432,6 +432,34 @@ def test_directory_commands(self):
         self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
         self.mount_a.run_shell(["rmdir", "d1"])
 
+    def test_directory_command_ls(self):
+        dir1 = 'dls1'
+        dir2 = 'dls2'
+        self.mount_a.run_shell(["mkdir", dir1])
+        self.mount_a.run_shell(["mkdir", dir2])
+        self.enable_mirroring(self.primary_fs_name, self.primary_fs_id)
+        try:
+            self.add_directory(self.primary_fs_name, self.primary_fs_id, f'/{dir1}')
+            self.add_directory(self.primary_fs_name, self.primary_fs_id, f'/{dir2}')
+            time.sleep(10)
+            dirs_list = json.loads(self.get_ceph_cmd_stdout("fs", "snapshot", "mirror", "ls", self.primary_fs_name))
+            # verify via asok
+            res = self.mirror_daemon_command(f'mirror status for fs: {self.primary_fs_name}',
+                                             'fs', 'mirror', 'status', f'{self.primary_fs_name}@{self.primary_fs_id}')
+            dir_count = res['snap_dirs']['dir_count']
+            self.assertTrue(len(dirs_list) == dir_count and f'/{dir1}' in dirs_list and f'/{dir2}' in dirs_list)
+        except CommandFailedError:
+            raise RuntimeError('Error listing directories')
+        except AssertionError:
+            raise RuntimeError('Wrong number of directories listed')
+        finally:
+            self.remove_directory(self.primary_fs_name, self.primary_fs_id, f'/{dir1}')
+            self.remove_directory(self.primary_fs_name, self.primary_fs_id, f'/{dir2}')
+
+        self.disable_mirroring(self.primary_fs_name, self.primary_fs_id)
+        self.mount_a.run_shell(["rmdir", dir1])
+        self.mount_a.run_shell(["rmdir",  dir2])
+
     def test_add_relative_directory_path(self):
         self.enable_mirroring(self.primary_fs_name, self.primary_fs_id)
         try:

src/pybind/mgr/mirroring/fs/snapshot_mirror.py

@@ -722,6 +722,20 @@ def remove_dir(self, filesystem, dir_path):
         except Exception as e:
             return e.args[0], '', 'failed to remove directory'
 
+    def list_dirs(self, filesystem):
+        try:
+            with self.lock:
+                if not self.filesystem_exist(filesystem):
+                    raise MirrorException(-errno.ENOENT, f'filesystem {filesystem} does not exist')
+                fspolicy = self.pool_policy.get(filesystem, None)
+                if not fspolicy:
+                    raise MirrorException(-errno.EINVAL, f'filesystem {filesystem} is not mirrored')
+                return 0, json.dumps(list(fspolicy.policy.dir_states.keys()), indent=4, sort_keys=True), ''
+        except MirrorException as me:
+            return me.args[0], '', me.args[1]
+        except Exception as e:
+            return e.args[0], '', 'failed to list directories'
+
     def status(self,filesystem, dir_path):
         try:
             with self.lock:

src/pybind/mgr/mirroring/module.py

@@ -84,6 +84,12 @@ def snapshot_mirror_remove_dir(self,
         """Remove a snapshot mirrored directory"""
         return self.fs_snapshot_mirror.remove_dir(fs_name, path)
 
+    @CLIWriteCommand('fs snapshot mirror ls')
+    def snapshot_mirror_ls(self,
+                           fs_name: str):
+        """List the snapshot mirrored directories"""
+        return self.fs_snapshot_mirror.list_dirs(fs_name)
+
     @CLIReadCommand('fs snapshot mirror dirmap')
     def snapshot_mirror_dirmap(self,
                                fs_name: str,