ceph/src/ceph-volume/ceph_volume/util/lsmdisk.py

   1 """
   2 This module handles the interaction with libstoragemgmt for local disk
   3 devices. Interaction may fail with LSM for a number of issues, but the
   4 intent here is to make this a soft fail, since LSM related data is not
   5 a critical component of ceph-volume.
   6 """
   7 import logging
   8
   9 try:
  10     from lsm import LocalDisk, LsmError
  11     from lsm import Disk as lsm_Disk
  12 except ImportError:
  13     lsm_available = False
  14     transport_map = {}
  15     health_map = {}
  16     lsm_Disk = None
  17 else:
  18     lsm_available = True
  19     transport_map = {
  20             lsm_Disk.LINK_TYPE_UNKNOWN: "Unavailable",
  21             lsm_Disk.LINK_TYPE_FC: "Fibre Channel",
  22             lsm_Disk.LINK_TYPE_SSA: "IBM SSA",
  23             lsm_Disk.LINK_TYPE_SBP: "Serial Bus",
  24             lsm_Disk.LINK_TYPE_SRP: "SCSI RDMA",
  25             lsm_Disk.LINK_TYPE_ISCSI: "iSCSI",
  26             lsm_Disk.LINK_TYPE_SAS: "SAS",
  27             lsm_Disk.LINK_TYPE_ADT: "ADT (Tape)",
  28             lsm_Disk.LINK_TYPE_ATA: "ATA/SATA",
  29             lsm_Disk.LINK_TYPE_USB: "USB",
  30             lsm_Disk.LINK_TYPE_SOP: "SCSI over PCI-E",
  31             lsm_Disk.LINK_TYPE_PCIE: "PCI-E",
  32     }
  33     health_map = {
  34         lsm_Disk.HEALTH_STATUS_UNKNOWN: "Unknown",
  35         lsm_Disk.HEALTH_STATUS_FAIL: "Fail",
  36         lsm_Disk.HEALTH_STATUS_WARN: "Warn",
  37         lsm_Disk.HEALTH_STATUS_GOOD: "Good",
  38     }
  39
  40 logger = logging.getLogger(__name__)
  41
  42
  43 class LSMDisk:
  44     def __init__(self, dev_path):
  45         self.dev_path = dev_path
  46         self.error_list = set()
  47
  48         if lsm_available:
  49             self.lsm_available = True
  50             self.disk = LocalDisk()
  51         else:
  52             self.lsm_available = False
  53             self.error_list.add("libstoragemgmt (lsm module) is unavailable")
  54             logger.info("LSM information is unavailable: libstoragemgmt is not installed")
  55             self.disk = None
  56
  57         self.led_bits = None
  58
  59     @property
  60     def errors(self):
  61         """show any errors that the LSM interaction has encountered (str)"""
  62         return ", ".join(self.error_list)
  63
  64     def _query_lsm(self, func, path):
  65         """Common method used to call the LSM functions, returning the function's result or None"""
  66
  67         # if disk is None, lsm is unavailable so all calls should return None
  68         if self.disk is None:
  69             return None
  70
  71         method = getattr(self.disk, func)
  72         try:
  73             output = method(path)
  74         except LsmError as err:
  75             logger.error("LSM Error: {}".format(err._msg))
  76             self.error_list.add(err._msg)
  77             return None
  78         else:
  79             return output
  80
  81     @property
  82     def led_status(self):
  83         """Fetch LED status, store in the LSMDisk object and return current status (int)"""
  84         if self.led_bits is None:
  85             self.led_bits = self._query_lsm('led_status_get', self.dev_path) or 1
  86             return self.led_bits
  87         else:
  88             return self.led_bits
  89
  90     @property
  91     def led_ident_state(self):
  92         """Query a disks IDENT LED state to discover when it is On, Off or Unknown (str)"""
  93         if self.led_status == 1:
  94             return "Unsupported"
  95         if self.led_status & lsm_Disk.LED_STATUS_IDENT_ON == lsm_Disk.LED_STATUS_IDENT_ON:
  96             return "On"
  97         elif self.led_status & lsm_Disk.LED_STATUS_IDENT_OFF == lsm_Disk.LED_STATUS_IDENT_OFF:
  98             return "Off"
  99         elif self.led_status & lsm_Disk.LED_STATUS_IDENT_UNKNOWN == lsm_Disk.LED_STATUS_IDENT_UNKNOWN:
 100             return "Unknown"
 101
 102         return "Unsupported"
 103
 104     @property
 105     def led_fault_state(self):
 106         """Query a disks FAULT LED state to discover when it is On, Off or Unknown (str)"""
 107         if self.led_status == 1:
 108             return "Unsupported"
 109         if self.led_status & lsm_Disk.LED_STATUS_FAULT_ON == lsm_Disk.LED_STATUS_FAULT_ON:
 110             return "On"
 111         elif self.led_status & lsm_Disk.LED_STATUS_FAULT_OFF == lsm_Disk.LED_STATUS_FAULT_OFF:
 112             return "Off"
 113         elif self.led_status & lsm_Disk.LED_STATUS_FAULT_UNKNOWN == lsm_Disk.LED_STATUS_FAULT_UNKNOWN:
 114             return "Unknown"
 115
 116         return "Unsupported"
 117
 118     @property
 119     def led_ident_support(self):
 120         """Query the LED state to determine IDENT support: Unknown, Supported, Unsupported (str)"""
 121         if self.led_status == 1:
 122             return "Unknown"
 123
 124         ident_states = (
 125             lsm_Disk.LED_STATUS_IDENT_ON +
 126             lsm_Disk.LED_STATUS_IDENT_OFF +
 127             lsm_Disk.LED_STATUS_IDENT_UNKNOWN
 128         )
 129
 130         if (self.led_status & ident_states) == 0:
 131             return "Unsupported"
 132
 133         return "Supported"
 134
 135     @property
 136     def led_fault_support(self):
 137         """Query the LED state to determine FAULT support: Unknown, Supported, Unsupported (str)"""
 138         if self.led_status == 1:
 139             return "Unknown"
 140
 141         fail_states = (
 142             lsm_Disk.LED_STATUS_FAULT_ON +
 143             lsm_Disk.LED_STATUS_FAULT_OFF +
 144             lsm_Disk.LED_STATUS_FAULT_UNKNOWN
 145         )
 146
 147         if self.led_status & fail_states == 0:
 148             return "Unsupported"
 149
 150         return "Supported"
 151
 152     @property
 153     def health(self):
 154         """Determine the health of the disk from LSM : Unknown, Fail, Warn or Good (str)"""
 155         _health_int = self._query_lsm('health_status_get', self.dev_path)
 156         return health_map.get(_health_int, "Unknown")
 157
 158     @property
 159     def transport(self):
 160         """Translate a disks link type to a human readable format (str)"""
 161         _link_type = self._query_lsm('link_type_get', self.dev_path)
 162         return transport_map.get(_link_type, "Unknown")
 163
 164
 165     @property
 166     def media_type(self):
 167         """Use the rpm value to determine the type of disk media: Flash or HDD (str)"""
 168         _rpm = self._query_lsm('rpm_get', self.dev_path)
 169         if _rpm is not None:
 170             if _rpm == 0:
 171                 return "Flash"
 172             elif _rpm > 1:
 173                 return "HDD"
 174
 175         return "Unknown"
 176
 177     def json_report(self):
 178         """Return the LSM related metadata for the current local disk (dict)"""
 179         if self.lsm_available:
 180             return {
 181                 "serialNum": self._query_lsm('serial_num_get', self.dev_path) or "Unknown",
 182                 "transport": self.transport,
 183                 "mediaType": self.media_type,
 184                 "rpm": self._query_lsm('rpm_get', self.dev_path) or "Unknown",
 185                 "linkSpeed": self._query_lsm('link_speed_get', self.dev_path) or "Unknown",
 186                 "health": self.health,
 187                 "ledSupport": {
 188                     "IDENTsupport": self.led_ident_support,
 189                     "IDENTstatus": self.led_ident_state,
 190                     "FAILsupport": self.led_fault_support,
 191                     "FAILstatus": self.led_fault_state,
 192                 },
 193                 "errors": list(self.error_list)
 194             }
 195         else:
 196             return {}