summaryrefslogtreecommitdiffstats
path: root/src/ceph-volume/ceph_volume/util/lsmdisk.py
diff options
context:
space:
mode:
Diffstat (limited to 'src/ceph-volume/ceph_volume/util/lsmdisk.py')
-rw-r--r--src/ceph-volume/ceph_volume/util/lsmdisk.py196
1 files changed, 196 insertions, 0 deletions
diff --git a/src/ceph-volume/ceph_volume/util/lsmdisk.py b/src/ceph-volume/ceph_volume/util/lsmdisk.py
new file mode 100644
index 000000000..607fd4f0a
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/util/lsmdisk.py
@@ -0,0 +1,196 @@
+"""
+This module handles the interaction with libstoragemgmt for local disk
+devices. Interaction may fail with LSM for a number of issues, but the
+intent here is to make this a soft fail, since LSM related data is not
+a critical component of ceph-volume.
+"""
+import logging
+
+try:
+ from lsm import LocalDisk, LsmError
+ from lsm import Disk as lsm_Disk
+except ImportError:
+ lsm_available = False
+ transport_map = {}
+ health_map = {}
+ lsm_Disk = None
+else:
+ lsm_available = True
+ transport_map = {
+ lsm_Disk.LINK_TYPE_UNKNOWN: "Unavailable",
+ lsm_Disk.LINK_TYPE_FC: "Fibre Channel",
+ lsm_Disk.LINK_TYPE_SSA: "IBM SSA",
+ lsm_Disk.LINK_TYPE_SBP: "Serial Bus",
+ lsm_Disk.LINK_TYPE_SRP: "SCSI RDMA",
+ lsm_Disk.LINK_TYPE_ISCSI: "iSCSI",
+ lsm_Disk.LINK_TYPE_SAS: "SAS",
+ lsm_Disk.LINK_TYPE_ADT: "ADT (Tape)",
+ lsm_Disk.LINK_TYPE_ATA: "ATA/SATA",
+ lsm_Disk.LINK_TYPE_USB: "USB",
+ lsm_Disk.LINK_TYPE_SOP: "SCSI over PCI-E",
+ lsm_Disk.LINK_TYPE_PCIE: "PCI-E",
+ }
+ health_map = {
+ lsm_Disk.HEALTH_STATUS_UNKNOWN: "Unknown",
+ lsm_Disk.HEALTH_STATUS_FAIL: "Fail",
+ lsm_Disk.HEALTH_STATUS_WARN: "Warn",
+ lsm_Disk.HEALTH_STATUS_GOOD: "Good",
+ }
+
+logger = logging.getLogger(__name__)
+
+
+class LSMDisk:
+ def __init__(self, dev_path):
+ self.dev_path = dev_path
+ self.error_list = set()
+
+ if lsm_available:
+ self.lsm_available = True
+ self.disk = LocalDisk()
+ else:
+ self.lsm_available = False
+ self.error_list.add("libstoragemgmt (lsm module) is unavailable")
+ logger.info("LSM information is unavailable: libstoragemgmt is not installed")
+ self.disk = None
+
+ self.led_bits = None
+
+ @property
+ def errors(self):
+ """show any errors that the LSM interaction has encountered (str)"""
+ return ", ".join(self.error_list)
+
+ def _query_lsm(self, func, path):
+ """Common method used to call the LSM functions, returning the function's result or None"""
+
+ # if disk is None, lsm is unavailable so all calls should return None
+ if self.disk is None:
+ return None
+
+ method = getattr(self.disk, func)
+ try:
+ output = method(path)
+ except LsmError as err:
+ logger.error("LSM Error: {}".format(err._msg))
+ self.error_list.add(err._msg)
+ return None
+ else:
+ return output
+
+ @property
+ def led_status(self):
+ """Fetch LED status, store in the LSMDisk object and return current status (int)"""
+ if self.led_bits is None:
+ self.led_bits = self._query_lsm('led_status_get', self.dev_path) or 1
+ return self.led_bits
+ else:
+ return self.led_bits
+
+ @property
+ def led_ident_state(self):
+ """Query a disks IDENT LED state to discover when it is On, Off or Unknown (str)"""
+ if self.led_status == 1:
+ return "Unsupported"
+ if self.led_status & lsm_Disk.LED_STATUS_IDENT_ON == lsm_Disk.LED_STATUS_IDENT_ON:
+ return "On"
+ elif self.led_status & lsm_Disk.LED_STATUS_IDENT_OFF == lsm_Disk.LED_STATUS_IDENT_OFF:
+ return "Off"
+ elif self.led_status & lsm_Disk.LED_STATUS_IDENT_UNKNOWN == lsm_Disk.LED_STATUS_IDENT_UNKNOWN:
+ return "Unknown"
+
+ return "Unsupported"
+
+ @property
+ def led_fault_state(self):
+ """Query a disks FAULT LED state to discover when it is On, Off or Unknown (str)"""
+ if self.led_status == 1:
+ return "Unsupported"
+ if self.led_status & lsm_Disk.LED_STATUS_FAULT_ON == lsm_Disk.LED_STATUS_FAULT_ON:
+ return "On"
+ elif self.led_status & lsm_Disk.LED_STATUS_FAULT_OFF == lsm_Disk.LED_STATUS_FAULT_OFF:
+ return "Off"
+ elif self.led_status & lsm_Disk.LED_STATUS_FAULT_UNKNOWN == lsm_Disk.LED_STATUS_FAULT_UNKNOWN:
+ return "Unknown"
+
+ return "Unsupported"
+
+ @property
+ def led_ident_support(self):
+ """Query the LED state to determine IDENT support: Unknown, Supported, Unsupported (str)"""
+ if self.led_status == 1:
+ return "Unknown"
+
+ ident_states = (
+ lsm_Disk.LED_STATUS_IDENT_ON +
+ lsm_Disk.LED_STATUS_IDENT_OFF +
+ lsm_Disk.LED_STATUS_IDENT_UNKNOWN
+ )
+
+ if (self.led_status & ident_states) == 0:
+ return "Unsupported"
+
+ return "Supported"
+
+ @property
+ def led_fault_support(self):
+ """Query the LED state to determine FAULT support: Unknown, Supported, Unsupported (str)"""
+ if self.led_status == 1:
+ return "Unknown"
+
+ fail_states = (
+ lsm_Disk.LED_STATUS_FAULT_ON +
+ lsm_Disk.LED_STATUS_FAULT_OFF +
+ lsm_Disk.LED_STATUS_FAULT_UNKNOWN
+ )
+
+ if self.led_status & fail_states == 0:
+ return "Unsupported"
+
+ return "Supported"
+
+ @property
+ def health(self):
+ """Determine the health of the disk from LSM : Unknown, Fail, Warn or Good (str)"""
+ _health_int = self._query_lsm('health_status_get', self.dev_path)
+ return health_map.get(_health_int, "Unknown")
+
+ @property
+ def transport(self):
+ """Translate a disks link type to a human readable format (str)"""
+ _link_type = self._query_lsm('link_type_get', self.dev_path)
+ return transport_map.get(_link_type, "Unknown")
+
+
+ @property
+ def media_type(self):
+ """Use the rpm value to determine the type of disk media: Flash or HDD (str)"""
+ _rpm = self._query_lsm('rpm_get', self.dev_path)
+ if _rpm is not None:
+ if _rpm == 0:
+ return "Flash"
+ elif _rpm > 1:
+ return "HDD"
+
+ return "Unknown"
+
+ def json_report(self):
+ """Return the LSM related metadata for the current local disk (dict)"""
+ if self.lsm_available:
+ return {
+ "serialNum": self._query_lsm('serial_num_get', self.dev_path) or "Unknown",
+ "transport": self.transport,
+ "mediaType": self.media_type,
+ "rpm": self._query_lsm('rpm_get', self.dev_path) or "Unknown",
+ "linkSpeed": self._query_lsm('link_speed_get', self.dev_path) or "Unknown",
+ "health": self.health,
+ "ledSupport": {
+ "IDENTsupport": self.led_ident_support,
+ "IDENTstatus": self.led_ident_state,
+ "FAILsupport": self.led_fault_support,
+ "FAILstatus": self.led_fault_state,
+ },
+ "errors": list(self.error_list)
+ }
+ else:
+ return {}