ceph/src/pybind/mgr/cephadm/utils.py

   1 import logging
   2 import json
   3 import socket
   4 from enum import Enum
   5 from functools import wraps
   6 from typing import Optional, Callable, TypeVar, List, NewType, TYPE_CHECKING, Any, NamedTuple
   7 from orchestrator import OrchestratorError
   8
   9 if TYPE_CHECKING:
  10     from cephadm import CephadmOrchestrator
  11
  12 T = TypeVar('T')
  13 logger = logging.getLogger(__name__)
  14
  15 ConfEntity = NewType('ConfEntity', str)
  16
  17
  18 class CephadmNoImage(Enum):
  19     token = 1
  20
  21
  22 # ceph daemon types that use the ceph container image.
  23 # NOTE: order important here as these are used for upgrade order
  24 CEPH_TYPES = ['mgr', 'mon', 'crash', 'osd', 'mds', 'rgw', 'rbd-mirror', 'cephfs-mirror']
  25 GATEWAY_TYPES = ['iscsi', 'nfs']
  26 MONITORING_STACK_TYPES = ['node-exporter', 'prometheus',
  27                           'alertmanager', 'grafana', 'loki', 'promtail']
  28 RESCHEDULE_FROM_OFFLINE_HOSTS_TYPES = ['nfs']
  29
  30 CEPH_UPGRADE_ORDER = CEPH_TYPES + GATEWAY_TYPES + MONITORING_STACK_TYPES
  31
  32 # these daemon types use the ceph container image
  33 CEPH_IMAGE_TYPES = CEPH_TYPES + ['iscsi', 'nfs']
  34
  35 # Used for _run_cephadm used for check-host etc that don't require an --image parameter
  36 cephadmNoImage = CephadmNoImage.token
  37
  38
  39 class ContainerInspectInfo(NamedTuple):
  40     image_id: str
  41     ceph_version: Optional[str]
  42     repo_digests: Optional[List[str]]
  43
  44
  45 def name_to_config_section(name: str) -> ConfEntity:
  46     """
  47     Map from daemon names to ceph entity names (as seen in config)
  48     """
  49     daemon_type = name.split('.', 1)[0]
  50     if daemon_type in ['rgw', 'rbd-mirror', 'nfs', 'crash', 'iscsi']:
  51         return ConfEntity('client.' + name)
  52     elif daemon_type in ['mon', 'osd', 'mds', 'mgr', 'client']:
  53         return ConfEntity(name)
  54     else:
  55         return ConfEntity('mon')
  56
  57
  58 def forall_hosts(f: Callable[..., T]) -> Callable[..., List[T]]:
  59     @wraps(f)
  60     def forall_hosts_wrapper(*args: Any) -> List[T]:
  61         from cephadm.module import CephadmOrchestrator
  62
  63         # Some weired logic to make calling functions with multiple arguments work.
  64         if len(args) == 1:
  65             vals = args[0]
  66             self = None
  67         elif len(args) == 2:
  68             self, vals = args
  69         else:
  70             assert 'either f([...]) or self.f([...])'
  71
  72         def do_work(arg: Any) -> T:
  73             if not isinstance(arg, tuple):
  74                 arg = (arg, )
  75             try:
  76                 if self:
  77                     return f(self, *arg)
  78                 return f(*arg)
  79             except Exception:
  80                 logger.exception(f'executing {f.__name__}({args}) failed.')
  81                 raise
  82
  83         assert CephadmOrchestrator.instance is not None
  84         return CephadmOrchestrator.instance._worker_pool.map(do_work, vals)
  85
  86     return forall_hosts_wrapper
  87
  88
  89 def get_cluster_health(mgr: 'CephadmOrchestrator') -> str:
  90     # check cluster health
  91     ret, out, err = mgr.check_mon_command({
  92         'prefix': 'health',
  93         'format': 'json',
  94     })
  95     try:
  96         j = json.loads(out)
  97     except ValueError:
  98         msg = 'Failed to parse health status: Cannot decode JSON'
  99         logger.exception('%s: \'%s\'' % (msg, out))
 100         raise OrchestratorError('failed to parse health status')
 101
 102     return j['status']
 103
 104
 105 def is_repo_digest(image_name: str) -> bool:
 106     """
 107     repo digest are something like "ceph/ceph@sha256:blablabla"
 108     """
 109     return '@' in image_name
 110
 111
 112 def resolve_ip(hostname: str) -> str:
 113     try:
 114         r = socket.getaddrinfo(hostname, None, flags=socket.AI_CANONNAME,
 115                                type=socket.SOCK_STREAM)
 116         # pick first v4 IP, if present
 117         for a in r:
 118             if a[0] == socket.AF_INET:
 119                 return a[4][0]
 120         return r[0][4][0]
 121     except socket.gaierror as e:
 122         raise OrchestratorError(f"Cannot resolve ip for host {hostname}: {e}")
 123
 124
 125 def ceph_release_to_major(release: str) -> int:
 126     return ord(release[0]) - ord('a') + 1
 127
 128
 129 def file_mode_to_str(mode: int) -> str:
 130     r = ''
 131     for shift in range(0, 9, 3):
 132         r = (
 133             f'{"r" if (mode >> shift) & 4 else "-"}'
 134             f'{"w" if (mode >> shift) & 2 else "-"}'
 135             f'{"x" if (mode >> shift) & 1 else "-"}'
 136         ) + r
 137     return r