ceph/src/python-common/ceph/deployment/inventory.py

   1 try:
   2     from typing import List, Optional, Dict, Any, Union
   3 except ImportError:
   4     pass  # for type checking
   5
   6 from ceph.utils import datetime_now, datetime_to_str, str_to_datetime
   7 import datetime
   8 import json
   9
  10
  11 class Devices(object):
  12     """
  13     A container for Device instances with reporting
  14     """
  15
  16     def __init__(self, devices):
  17         # type: (List[Device]) -> None
  18         # sort devices by path name so ordering is consistent
  19         self.devices: List[Device] = sorted(devices, key=lambda d: d.path if d.path else '')
  20
  21     def __eq__(self, other: Any) -> bool:
  22         if not isinstance(other, Devices):
  23             return NotImplemented
  24         if len(self.devices) != len(other.devices):
  25             return False
  26         for d1, d2 in zip(other.devices, self.devices):
  27             if d1 != d2:
  28                 return False
  29         return True
  30
  31     def to_json(self):
  32         # type: () -> List[dict]
  33         return [d.to_json() for d in self.devices]
  34
  35     @classmethod
  36     def from_json(cls, input):
  37         # type: (List[Dict[str, Any]]) -> Devices
  38         return cls([Device.from_json(i) for i in input])
  39
  40     def copy(self):
  41         # type: () -> Devices
  42         return Devices(devices=list(self.devices))
  43
  44
  45 class Device(object):
  46     report_fields = [
  47         'ceph_device',
  48         'rejected_reasons',
  49         'available',
  50         'path',
  51         'sys_api',
  52         'created',
  53         'lvs',
  54         'human_readable_type',
  55         'device_id',
  56         'lsm_data',
  57         'crush_device_class'
  58     ]
  59
  60     def __init__(self,
  61                  path,  # type: str
  62                  sys_api=None,  # type: Optional[Dict[str, Any]]
  63                  available=None,  # type: Optional[bool]
  64                  rejected_reasons=None,  # type: Optional[List[str]]
  65                  lvs=None,  # type: Optional[List[Dict[str, str]]]
  66                  device_id=None,  # type: Optional[str]
  67                  lsm_data=None,  # type: Optional[Dict[str, Dict[str, str]]]
  68                  created=None,  # type: Optional[datetime.datetime]
  69                  ceph_device=None,  # type: Optional[bool]
  70                  crush_device_class=None  # type: Optional[str]
  71                  ):
  72
  73         self.path = path
  74         self.sys_api = sys_api if sys_api is not None else {}  # type: Dict[str, Any]
  75         self.available = available
  76         self.rejected_reasons = rejected_reasons if rejected_reasons is not None else []
  77         self.lvs = lvs
  78         self.device_id = device_id
  79         self.lsm_data = lsm_data if lsm_data is not None else {}  # type: Dict[str, Dict[str, str]]
  80         self.created = created if created is not None else datetime_now()
  81         self.ceph_device = ceph_device
  82         self.crush_device_class = crush_device_class
  83
  84     def __eq__(self, other):
  85         # type: (Any) -> bool
  86         if not isinstance(other, Device):
  87             return NotImplemented
  88         diff = [k for k in self.report_fields if k != 'created' and (getattr(self, k)
  89                                                                      != getattr(other, k))]
  90         return not diff
  91
  92     def to_json(self):
  93         # type: () -> dict
  94         return {
  95             k: (getattr(self, k) if k != 'created'
  96                 or not isinstance(getattr(self, k), datetime.datetime)
  97                 else datetime_to_str(getattr(self, k)))
  98             for k in self.report_fields
  99         }
 100
 101     @classmethod
 102     def from_json(cls, input):
 103         # type: (Dict[str, Any]) -> Device
 104         if not isinstance(input, dict):
 105             raise ValueError('Device: Expected dict. Got `{}...`'.format(json.dumps(input)[:10]))
 106         ret = cls(
 107             **{
 108                 key: (input.get(key, None) if key != 'created'
 109                       or not input.get(key, None)
 110                       else str_to_datetime(input.get(key, None)))
 111                 for key in Device.report_fields
 112                 if key != 'human_readable_type'
 113             }
 114         )
 115         if ret.rejected_reasons:
 116             ret.rejected_reasons = sorted(ret.rejected_reasons)
 117         return ret
 118
 119     @property
 120     def human_readable_type(self):
 121         # type: () -> str
 122         if self.sys_api is None or 'rotational' not in self.sys_api:
 123             return "unknown"
 124         return 'hdd' if self.sys_api["rotational"] == "1" else 'ssd'
 125
 126     def __repr__(self) -> str:
 127         device_desc: Dict[str, Union[str, List[str], List[Dict[str, str]]]] = {
 128             'path': self.path if self.path is not None else 'unknown',
 129             'lvs': self.lvs if self.lvs else 'None',
 130             'available': str(self.available),
 131             'ceph_device': str(self.ceph_device),
 132             'crush_device_class': str(self.crush_device_class)
 133         }
 134         if not self.available and self.rejected_reasons:
 135             device_desc['rejection reasons'] = self.rejected_reasons
 136         return "Device({})".format(
 137             ', '.join('{}={}'.format(key, device_desc[key]) for key in device_desc.keys())
 138         )