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

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