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

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