Merge pull request ceph#60395 from guits/replace-osd-enhancement

ceph-volume: support zapping by osd-id for RAW OSDs

Author: guits

src/ceph-volume/ceph_volume/devices/lvm/zap.py

@@ -10,7 +10,8 @@
 from ceph_volume.util import system, encryption, disk, arg_validators, str_to_int, merge_dict
 from ceph_volume.util.device import Device
 from ceph_volume.systemd import systemctl
-from typing import Any, Dict, List
+from ceph_volume.devices.raw.list import direct_report
+from typing import Any, Dict, List, Set
 
 logger = logging.getLogger(__name__)
 mlogger = terminal.MultiLogger(__name__)
@@ -95,83 +96,126 @@ def zap_data(path):
         'conv=fsync'
     ])
 
-def find_associated_devices(osd_id: str = '', osd_fsid: str = '') -> List[api.Volume]:
-    """
-    From an ``osd_id`` and/or an ``osd_fsid``, filter out all the LVs in the
-    system that match those tag values, further detect if any partitions are
-    part of the OSD, and then return the set of LVs and partitions (if any).
-    """
-    lv_tags = {}
-    lv_tags = {key: value for key, value in {
-        'ceph.osd_id': osd_id,
-        'ceph.osd_fsid': osd_fsid
-    }.items() if value}
 
-    lvs = api.get_lvs(tags=lv_tags)
+class Zap:
+    help = 'Removes all data and filesystems from a logical volume or partition.'
 
-    if not lvs:
-        raise RuntimeError('Unable to find any LV for zapping OSD: '
-                            f'{osd_id or osd_fsid}')
-    devices_to_zap = ensure_associated_lvs(lvs, lv_tags)
+    def __init__(self, argv: List[str]) -> None:
+        self.argv = argv
+        self.osd_ids_to_zap: List[str] = []
 
-    return [Device(path) for path in set(devices_to_zap) if path]
+    def ensure_associated_raw(self, raw_report: Dict[str, Any]) -> List[str]:
+        osd_id: str = self.args.osd_id
+        osd_uuid: str = self.args.osd_fsid
+        raw_devices: Set[str] = set()
 
-def ensure_associated_lvs(lvs: List[api.Volume],
-                          lv_tags: Dict[str, Any] = {}) -> List[str]:
-    """
-    Go through each LV and ensure if backing devices (journal, wal, block)
-    are LVs or partitions, so that they can be accurately reported.
-    """
-    # look for many LVs for each backing type, because it is possible to
-    # receive a filtering for osd.1, and have multiple failed deployments
-    # leaving many journals with osd.1 - usually, only a single LV will be
-    # returned
-
-    db_lvs = api.get_lvs(tags=merge_dict(lv_tags, {'ceph.type': 'db'}))
-    wal_lvs = api.get_lvs(tags=merge_dict(lv_tags, {'ceph.type': 'wal'}))
-    backing_devices = [(db_lvs, 'db'),
-                       (wal_lvs, 'wal')]
-
-    verified_devices = []
-
-    for lv in lvs:
-        # go through each lv and append it, otherwise query `blkid` to find
-        # a physical device. Do this for each type (journal,db,wal) regardless
-        # if they have been processed in the previous LV, so that bad devices
-        # with the same ID can be caught
-        for ceph_lvs, _type in backing_devices:
-            if ceph_lvs:
-                verified_devices.extend([l.lv_path for l in ceph_lvs])
-                continue
-
-            # must be a disk partition, by querying blkid by the uuid we are
-            # ensuring that the device path is always correct
-            try:
-                device_uuid = lv.tags['ceph.%s_uuid' % _type]
-            except KeyError:
-                # Bluestore will not have ceph.journal_uuid, and Filestore
-                # will not not have ceph.db_uuid
-                continue
+        if len([details.get('osd_id') for _, details in raw_report.items() if details.get('osd_id') == osd_id]) > 1:
+            if not osd_uuid:
+                raise RuntimeError(f'Multiple OSDs found with id {osd_id}, pass --osd-fsid')
 
-            osd_device = disk.get_device_from_partuuid(device_uuid)
-            if not osd_device:
-                # if the osd_device is not found by the partuuid, then it is
-                # not possible to ensure this device exists anymore, so skip it
-                continue
-            verified_devices.append(osd_device)
+        if not osd_uuid:
+            for _, details in raw_report.items():
+                if details.get('osd_id') == int(osd_id):
+                    osd_uuid = details.get('osd_uuid')
+                    break
 
-        verified_devices.append(lv.lv_path)
+        for osd_uuid, details in raw_report.items():
+            device: str = details.get('device')
+            if details.get('osd_uuid') == osd_uuid:
+                raw_devices.add(device)
 
-    # reduce the list from all the duplicates that were added
-    return list(set(verified_devices))
+        return list(raw_devices)
+        
 
+    def find_associated_devices(self) -> List[api.Volume]:
+        """From an ``osd_id`` and/or an ``osd_fsid``, filter out all the Logical Volumes (LVs) in the
+        system that match those tag values, further detect if any partitions are
+        part of the OSD, and then return the set of LVs and partitions (if any).
 
-class Zap:
-    help = 'Removes all data and filesystems from a logical volume or partition.'
+        The function first queries the LVM-based OSDs using the provided `osd_id` or `osd_fsid`.
+        If no matches are found, it then searches the system for RAW-based OSDs.
 
-    def __init__(self, argv: List[str]) -> None:
-        self.argv = argv
-        self.osd_ids_to_zap: List[str] = []
+        Raises:
+            SystemExit: If no OSDs are found, the function raises a `SystemExit` with an appropriate message.
+
+        Returns:
+            List[api.Volume]: A list of `api.Volume` objects corresponding to the OSD's Logical Volumes (LVs)
+            or partitions that are associated with the given `osd_id` or `osd_fsid`.
+
+        Notes:
+            - If neither `osd_id` nor `osd_fsid` are provided, the function will not be able to find OSDs.
+            - The search proceeds from LVM-based OSDs to RAW-based OSDs if no Logical Volumes are found.
+        """
+        lv_tags = {}
+        lv_tags = {key: value for key, value in {
+            'ceph.osd_id': self.args.osd_id,
+            'ceph.osd_fsid': self.args.osd_fsid
+        }.items() if value}
+        devices_to_zap: List[str] = []
+        lvs = api.get_lvs(tags=lv_tags)
+
+        if lvs:
+            devices_to_zap = self.ensure_associated_lvs(lvs, lv_tags)
+        else:
+            mlogger.debug(f'No OSD identified by "{self.args.osd_id or self.args.osd_fsid}" was found among LVM-based OSDs.')
+            mlogger.debug('Proceeding to check RAW-based OSDs.')
+            raw_osds: Dict[str, Any] = direct_report()
+            if raw_osds:
+                devices_to_zap = self.ensure_associated_raw(raw_osds)
+        if not devices_to_zap:
+            raise SystemExit('No OSD were found.')
+
+        return [Device(path) for path in set(devices_to_zap) if path]
+
+    def ensure_associated_lvs(self,
+                              lvs: List[api.Volume],
+                              lv_tags: Dict[str, Any] = {}) -> List[str]:
+        """
+        Go through each LV and ensure if backing devices (journal, wal, block)
+        are LVs or partitions, so that they can be accurately reported.
+        """
+        # look for many LVs for each backing type, because it is possible to
+        # receive a filtering for osd.1, and have multiple failed deployments
+        # leaving many journals with osd.1 - usually, only a single LV will be
+        # returned
+
+        db_lvs = api.get_lvs(tags=merge_dict(lv_tags, {'ceph.type': 'db'}))
+        wal_lvs = api.get_lvs(tags=merge_dict(lv_tags, {'ceph.type': 'wal'}))
+        backing_devices = [(db_lvs, 'db'),
+                        (wal_lvs, 'wal')]
+
+        verified_devices = []
+
+        for lv in lvs:
+            # go through each lv and append it, otherwise query `blkid` to find
+            # a physical device. Do this for each type (journal,db,wal) regardless
+            # if they have been processed in the previous LV, so that bad devices
+            # with the same ID can be caught
+            for ceph_lvs, _type in backing_devices:
+                if ceph_lvs:
+                    verified_devices.extend([l.lv_path for l in ceph_lvs])
+                    continue
+
+                # must be a disk partition, by querying blkid by the uuid we are
+                # ensuring that the device path is always correct
+                try:
+                    device_uuid = lv.tags['ceph.%s_uuid' % _type]
+                except KeyError:
+                    # Bluestore will not have ceph.journal_uuid, and Filestore
+                    # will not not have ceph.db_uuid
+                    continue
+
+                osd_device = disk.get_device_from_partuuid(device_uuid)
+                if not osd_device:
+                    # if the osd_device is not found by the partuuid, then it is
+                    # not possible to ensure this device exists anymore, so skip it
+                    continue
+                verified_devices.append(osd_device)
+
+            verified_devices.append(lv.lv_path)
+
+        # reduce the list from all the duplicates that were added
+        return list(set(verified_devices))
 
     def unmount_lv(self, lv: api.Volume) -> None:
         if lv.tags.get('ceph.cluster_name') and lv.tags.get('ceph.osd_id'):
@@ -355,7 +399,6 @@ def zap(self) -> None:
             SystemExit: When the device is a mapper and not a mpath device.
         """
         devices = self.args.devices
-
         for device in devices:
             mlogger.info("Zapping: %s", device.path)
             if device.is_mapper and not device.is_mpath:
@@ -388,7 +431,7 @@ def zap_osd(self) -> None:
                 mlogger.error("OSD ID %s is running, stop it with:" % self.args.osd_id)
                 mlogger.error("systemctl stop ceph-osd@%s" % self.args.osd_id)
                 raise SystemExit("Unable to zap devices associated with OSD ID: %s" % self.args.osd_id)
-        self.args.devices = find_associated_devices(self.args.osd_id, self.args.osd_fsid)
+        self.args.devices = self.find_associated_devices()
         self.zap()
 
     def dmcrypt_close(self, dmcrypt_uuid: str) -> None: