mgr/volumes: handling dangling symlinks gracefully

If we come across situations where we encounter
dangling symlinks for clones due to any reason,
it needs to handled gracefully by deleting it,
presence of which may not allow deletion of the
snapshot.
Fixes: https://tracker.ceph.com/issues/58090
Signed-off-by: Neeraj Pratap Singh <neesingh@redhat.com>

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

@@ -847,6 +847,30 @@ def snapshot_info(self, snapname):
                                       "snapshot '{0}' does not exist".format(snapname))
             raise VolumeException(-e.args[0], e.args[1])
 
+    def check_dangling_clone_index(self, snapname):
+        try:
+            if self.has_pending_clones(snapname):
+                pending_track_id_list = self.metadata_mgr.list_all_keys_with_specified_values_from_section('clone snaps', snapname)
+        except MetadataMgrException as me:
+            if me.errno != -errno.ENOENT:
+                raise VolumeException(-me.args[0], me.args[1])
+
+        try:
+            with open_clone_index(self.fs, self.vol_spec) as index:
+                index_path = index.path.decode('utf-8')
+        except IndexException as e:
+            log.warning("failed to open clone index '{0}' for snapshot '{1}'".format(e, snapname))
+            raise VolumeException(-errno.EINVAL, "failed to open clone index")
+
+        for track_id in pending_track_id_list:
+            try:
+                path = os.path.join(index_path, track_id)
+                if not os.path.exists(self.fs.readlink(path, 4096)):
+                    self.fs.rmdir(path)
+            except cephfs.Error as e:
+                if e.errno != errno.ENOENT:
+                    raise VolumeException(-e.args[0], e.args[1])
+
     def list_snapshots(self):
         try:
             dirpath = self.snapshot_base_path()

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

@@ -586,6 +586,7 @@ def subvolume_snapshot_info(self, **kwargs):
             with open_volume(self, volname) as fs_handle:
                 with open_group(fs_handle, self.volspec, groupname) as group:
                     with open_subvol(self.mgr, fs_handle, self.volspec, group, subvolname, SubvolumeOpType.SNAP_INFO) as subvolume:
+                        subvolume.check_dangling_clone_index(snapname)
                         snap_info_dict = subvolume.snapshot_info(snapname)
                         ret = 0, json.dumps(snap_info_dict, indent=4, sort_keys=True), ""
         except VolumeException as ve: