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

   1 import errno
   2 try:
   3     from typing import Optional, List, Any
   4 except ImportError:
   5     pass  # just for type checking
   6
   7
   8 class SpecValidationError(Exception):
   9     """
  10     Defining an exception here is a bit problematic, cause you cannot properly catch it,
  11     if it was raised in a different mgr module.
  12     """
  13     def __init__(self,
  14                  msg: str,
  15                  errno: int = -errno.EINVAL):
  16         super(SpecValidationError, self).__init__(msg)
  17         self.errno = errno
  18
  19
  20 class HostSpec(object):
  21     """
  22     Information about hosts. Like e.g. ``kubectl get nodes``
  23     """
  24     def __init__(self,
  25                  hostname,  # type: str
  26                  addr=None,  # type: Optional[str]
  27                  labels=None,  # type: Optional[List[str]]
  28                  status=None,  # type: Optional[str]
  29                  ):
  30         self.service_type = 'host'
  31
  32         #: the bare hostname on the host. Not the FQDN.
  33         self.hostname = hostname  # type: str
  34
  35         #: DNS name or IP address to reach it
  36         self.addr = addr or hostname  # type: str
  37
  38         #: label(s), if any
  39         self.labels = labels or []  # type: List[str]
  40
  41         #: human readable status
  42         self.status = status or ''  # type: str
  43
  44     def to_json(self) -> dict:
  45         return {
  46             'hostname': self.hostname,
  47             'addr': self.addr,
  48             'labels': list(set((self.labels))),
  49             'status': self.status,
  50         }
  51
  52     @classmethod
  53     def from_json(cls, host_spec: dict) -> 'HostSpec':
  54         host_spec = cls.normalize_json(host_spec)
  55         _cls = cls(host_spec['hostname'],
  56                    host_spec['addr'] if 'addr' in host_spec else None,
  57                    list(set(host_spec['labels'])) if 'labels' in host_spec else None,
  58                    host_spec['status'] if 'status' in host_spec else None)
  59         return _cls
  60
  61     @staticmethod
  62     def normalize_json(host_spec: dict) -> dict:
  63         labels = host_spec.get('labels')
  64         if labels is None:
  65             return host_spec
  66         if isinstance(labels, list):
  67             return host_spec
  68         if not isinstance(labels, str):
  69             raise SpecValidationError(f'Labels ({labels}) must be a string or list of strings')
  70         host_spec['labels'] = [labels]
  71         return host_spec
  72
  73     def __repr__(self) -> str:
  74         args = [self.hostname]  # type: List[Any]
  75         if self.addr is not None:
  76             args.append(self.addr)
  77         if self.labels:
  78             args.append(self.labels)
  79         if self.status:
  80             args.append(self.status)
  81
  82         return "HostSpec({})".format(', '.join(map(repr, args)))
  83
  84     def __str__(self) -> str:
  85         if self.hostname != self.addr:
  86             return f'{self.hostname} ({self.addr})'
  87         return self.hostname
  88
  89     def __eq__(self, other: Any) -> bool:
  90         # Let's omit `status` for the moment, as it is still the very same host.
  91         return self.hostname == other.hostname and \
  92                self.addr == other.addr and \
  93                self.labels == other.labels