Merge PR ceph#63214 into main

* refs/pull/63214/head:
	release note: add a note that "subvolume info" cmd output can also...
	doc/cephfs: update docs since "subvolume info" cmd output can also...
	qa/cephfs: add test to check clone source info's present in...
	mgr/vol: show clone source info in "subvolume info" cmd output
	mgr/vol: keep clone source info even after cloning is finished

Reviewed-by: Venky Shankar <[email protected]>
Reviewed-by: Kotresh Hiremath Ravishankar <[email protected]>
Reviewed-by: Neeraj Pratap Singh <[email protected]>

Author: vshankar

PendingReleaseNotes

@@ -147,6 +147,14 @@
   allowed.  `rbd trash mv` command now behaves the same way as `rbd rm` in this
   scenario.
 
+* CephFS: If the subvolume name passed to the command `ceph fs subvolume info`
+  is a clone, the output now will also contain a "source" field that tells user
+  the name of source snapshot along with the name of volume, subvolume group and
+  subvolume in which the source snapshot is located. For clones created with
+  Tentacle or earlier release, the value of this field will be 'N/A'. Regular
+  subvolumes don't have a source subvolume and therefore the output for them
+  won't contain a "source" field regardless of the release.
+
 * RGW: Replication policies now validate permissions using `s3:ReplicateObject`,
   `s3:ReplicateDelete`, and `s3:ReplicateTags` for destination buckets. For source
   buckets, both `s3:GetObjectVersionForReplication` and `s3:GetObject(Version)`

doc/cephfs/fs-volumes.rst

@@ -480,6 +480,10 @@ The output format is JSON and contains the following fields.
 * ``features``: features supported by the subvolume
 * ``state``: current state of the subvolume
 * ``earmark``: earmark of the subvolume
+* ``source``: exists only if subvolume is a clone. It contains name of the
+  source snapshot and names of the volume, subvolume group and subvolume in
+  which the source snapshot is located. If the clone was created with Tentacle
+  or earlier release, value of this field is 'N/A'.
 
 If a subvolume has been removed but its snapshots have been retained, the
 output contains only the following fields.

qa/tasks/cephfs/test_volumes.py

@@ -2887,12 +2887,34 @@ def test_subvolume_info(self):
 
         self.assertEqual(subvol_info["earmark"], earmark)
 
+        self.assertNotIn('source', subvol_info)
+
         # remove subvolumes
         self._fs_cmd("subvolume", "rm", self.volname, subvolume)
 
         # verify trash dir is clean
         self._wait_for_trash_empty()
 
+    def test_subvol_src_info_with_custom_group(self):
+        '''
+        Test that source info is NOT printed by "subvolume info" command for a
+        subvolume that is not created by cloning even when it is located in a
+        custom group.
+        '''
+        subvol_name = self._gen_subvol_name()
+        group_name = self._gen_subvol_grp_name()
+
+        self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} '
+                          f'{group_name}')
+        self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} '
+                          f'{group_name}')
+
+        subvol_info = self.get_ceph_cmd_stdout(
+            f'fs subvolume info {self.volname} {subvol_name} {group_name}')
+        subvol_info = json.loads(subvol_info)
+
+        self.assertNotIn('source', subvol_info)
+
     def test_subvolume_ls(self):
         # tests the 'fs subvolume ls' command
 
@@ -6810,6 +6832,7 @@ def test_clone_subvolume_info(self):
         # remove snapshot
         self._fs_cmd("subvolume", "snapshot", "rm", self.volname, subvolume, snapshot)
 
+        # actual testing begins now...
         subvol_info = json.loads(self._get_subvolume_info(self.volname, clone))
         if len(subvol_info) == 0:
             raise RuntimeError("Expected the 'fs subvolume info' command to list metadata of subvolume")
@@ -6819,13 +6842,48 @@ def test_clone_subvolume_info(self):
         if subvol_info["type"] != "clone":
             raise RuntimeError("type should be set to clone")
 
+        self.assertEqual(subvol_info['source']['volume'], self.volname)
+        self.assertEqual(subvol_info['source']['subvolume'], subvolume)
+        self.assertEqual(subvol_info['source']['snapshot'], snapshot)
+        self.assertEqual(subvol_info['source']['group'], '_nogroup')
+
         # remove subvolumes
         self._fs_cmd("subvolume", "rm", self.volname, subvolume)
         self._fs_cmd("subvolume", "rm", self.volname, clone)
 
         # verify trash dir is clean
         self._wait_for_trash_empty()
 
+    def test_clone_src_info_with_custom_group(self):
+        '''
+        Test that clone's source subvolume's group is printed properly when
+        "subvolume info" command is run for clone.
+        '''
+        subvol_name = self._gen_subvol_name()
+        group_name = self._gen_subvol_grp_name()
+        snap_name = self._gen_subvol_snap_name()
+        clone_name = self._gen_subvol_clone_name()
+
+        self.run_ceph_cmd(f'fs subvolumegroup create {self.volname} '
+                          f'{group_name} --mode=777')
+        self.run_ceph_cmd(f'fs subvolume create {self.volname} {subvol_name} '
+                          f'{group_name} --mode=777')
+        self._do_subvolume_io(subvol_name, group_name, number_of_files=1)
+        self.run_ceph_cmd(f'fs subvolume snapshot create {self.volname} '
+                          f'{subvol_name} {snap_name} {group_name}')
+        self.run_ceph_cmd(f'fs subvolume snapshot clone {self.volname} '
+                          f'{subvol_name} {snap_name} {clone_name} '
+                          f'--group-name {group_name}')
+        self._wait_for_clone_to_complete(clone_name)
+
+        subvol_info = self.get_ceph_cmd_stdout(
+            f'fs subvolume info {self.volname} {clone_name}')
+        subvol_info = json.loads(subvol_info)
+        self.assertEqual(subvol_info['source']['volume'], self.volname)
+        self.assertEqual(subvol_info['source']['subvolume'], subvol_name)
+        self.assertEqual(subvol_info['source']['snapshot'], snap_name)
+        self.assertEqual(subvol_info['source']['group'], group_name)
+
     def test_subvolume_snapshot_info_without_snapshot_clone(self):
         """
         Verify subvolume snapshot info output without cloning snapshot.

src/pybind/mgr/volumes/fs/async_cloner.py

@@ -273,7 +273,6 @@ def handle_clone_complete(fs_client, volspec, volname, index, groupname, subvoln
         with open_clone_subvol_pair_in_vol(fs_client, volspec, volname,
                 groupname, subvolname) as (subvol0, subvol1, subvol2):
             subvol1.detach_snapshot(subvol2, index)
-            subvol0.remove_clone_source(flush=True)
     except (MetadataMgrException, VolumeException) as e:
         log.error("failed to detach clone from snapshot: {0}".format(e))
     return (None, True)

src/pybind/mgr/volumes/fs/operations/versions/subvolume_base.py

@@ -442,6 +442,34 @@ def create_base_dir(self, mode):
         except cephfs.Error as e:
             raise VolumeException(-e.args[0], e.args[1])
 
+    def _get_clone_source(self):
+        try:
+            clone_source = {
+                'volume'   : self.metadata_mgr.get_option("source", "volume"),
+                'subvolume': self.metadata_mgr.get_option("source", "subvolume"),
+                'snapshot' : self.metadata_mgr.get_option("source", "snapshot"),
+            }
+
+            try:
+                clone_source["group"] = self.metadata_mgr.get_option("source", "group")
+            except MetadataMgrException as me:
+                if me.errno == -errno.ENOENT:
+                    pass
+                else:
+                    raise
+        except MetadataMgrException as e:
+                if e.errno == -errno.ENOENT:
+                    clone_source = {}
+                else:
+                    raise VolumeException(-errno.EINVAL,
+                                          "error fetching subvolume metadata")
+        return clone_source
+
+    def get_clone_source(self):
+        src = self._get_clone_source()
+        return (src['volume'], src.get('group', None), src['subvolume'],
+                src['snapshot'])
+
     def info(self):
         subvolpath = (self.metadata_mgr.get_global_option(
                       MetadataManager.GLOBAL_META_KEY_PATH))
@@ -494,7 +522,8 @@ def info(self):
         except cephfs.NoData:
             casesensitive = True
 
-        return {'path': subvolpath,
+        subvol_info = {
+                'path': subvolpath,
                 'type': etype.value,
                 'uid': int(st["uid"]),
                 'gid': int(st["gid"]),
@@ -517,6 +546,26 @@ def info(self):
                 'casesensitive': casesensitive,
         }
 
+        subvol_src_info = self._get_clone_source()
+        if subvol_src_info:
+            if subvol_src_info.get('group', None) == None:
+                # group name won't be saved in .meta file in case it's
+                # default group
+                subvol_src_info['group'] = '_nogroup'
+            subvol_info['source'] = subvol_src_info
+        else:
+            # it could be that the clone was created in previous release of Ceph
+            # where its source info used to be deleted after cloning finishes.
+            # print "N/A" for such cases.
+            if self.subvol_type == SubvolumeTypes.TYPE_CLONE:
+                subvol_info['source'] = 'N/A'
+            else:
+                # only clones can have a source subvol, therefore don't even
+                # print "N/A" for source info if subvolume is not a clone.
+                pass
+
+        return subvol_info
+
     def set_user_metadata(self, keyname, value):
         try:
             self.metadata_mgr.add_section(MetadataManager.USER_METADATA_SECTION)

src/pybind/mgr/volumes/fs/operations/versions/subvolume_v1.py

@@ -138,11 +138,6 @@ def add_clone_source(self, volname, subvolume, snapname, flush=False):
         if flush:
             self.metadata_mgr.flush()
 
-    def remove_clone_source(self, flush=False):
-        self.metadata_mgr.remove_section("source")
-        if flush:
-            self.metadata_mgr.flush()
-
     def add_clone_failure(self, errno, error_msg):
         try:
             self.metadata_mgr.add_section(MetadataManager.CLONE_FAILURE_SECTION)
@@ -657,30 +652,6 @@ def evict(self, volname, auth_id, timeout=30):
                 log.error(msg)
                 raise EvictionError(msg)
 
-    def _get_clone_source(self):
-        try:
-            clone_source = {
-                'volume'   : self.metadata_mgr.get_option("source", "volume"),
-                'subvolume': self.metadata_mgr.get_option("source", "subvolume"),
-                'snapshot' : self.metadata_mgr.get_option("source", "snapshot"),
-            }
-
-            try:
-                clone_source["group"] = self.metadata_mgr.get_option("source", "group")
-            except MetadataMgrException as me:
-                if me.errno == -errno.ENOENT:
-                    pass
-                else:
-                    raise
-        except MetadataMgrException:
-            raise VolumeException(-errno.EINVAL, "error fetching subvolume metadata")
-        return clone_source
-
-    def get_clone_source(self):
-        src = self._get_clone_source()
-        return (src['volume'], src.get('group', None), src['subvolume'],
-                src['snapshot'])
-
     def _get_clone_failure(self):
         clone_failure = {
             'errno'     : self.metadata_mgr.get_option(MetadataManager.CLONE_FAILURE_SECTION, MetadataManager.CLONE_FAILURE_META_KEY_ERRNO),