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