ceph/src/pybind/mgr/dashboard/services/orchestrator.py

   1 # -*- coding: utf-8 -*-
   2
   3 import logging
   4 from functools import wraps
   5 from typing import Any, Dict, List, Optional, Tuple
   6
   7 from ceph.deployment.service_spec import ServiceSpec
   8 from orchestrator import DaemonDescription, DeviceLightLoc, HostSpec, \
   9     InventoryFilter, OrchestratorClientMixin, OrchestratorError, OrchResult, \
  10     ServiceDescription, raise_if_exception
  11
  12 from .. import mgr
  13 from ._paginate import ListPaginator
  14
  15 logger = logging.getLogger('orchestrator')
  16
  17
  18 # pylint: disable=abstract-method
  19 class OrchestratorAPI(OrchestratorClientMixin):
  20     def __init__(self):
  21         super(OrchestratorAPI, self).__init__()
  22         self.set_mgr(mgr)  # type: ignore
  23
  24     def status(self):
  25         try:
  26             status, message, _module_details = super().available()
  27             logger.info("is orchestrator available: %s, %s", status, message)
  28             return dict(available=status, message=message)
  29         except (RuntimeError, OrchestratorError, ImportError) as e:
  30             return dict(
  31                 available=False,
  32                 message='Orchestrator is unavailable: {}'.format(str(e)))
  33
  34
  35 def wait_api_result(method):
  36     @wraps(method)
  37     def inner(self, *args, **kwargs):
  38         completion = method(self, *args, **kwargs)
  39         raise_if_exception(completion)
  40         return completion.result
  41     return inner
  42
  43
  44 class ResourceManager(object):
  45     def __init__(self, api):
  46         self.api = api
  47
  48
  49 class HostManger(ResourceManager):
  50     @wait_api_result
  51     def list(self) -> List[HostSpec]:
  52         return self.api.get_hosts()
  53
  54     @wait_api_result
  55     def enter_maintenance(self, hostname: str, force: bool = False):
  56         return self.api.enter_host_maintenance(hostname, force)
  57
  58     @wait_api_result
  59     def exit_maintenance(self, hostname: str):
  60         return self.api.exit_host_maintenance(hostname)
  61
  62     def get(self, hostname: str) -> Optional[HostSpec]:
  63         hosts = [host for host in self.list() if host.hostname == hostname]
  64         return hosts[0] if hosts else None
  65
  66     @wait_api_result
  67     def add(self, hostname: str, addr: str, labels: List[str]):
  68         return self.api.add_host(HostSpec(hostname, addr=addr, labels=labels))
  69
  70     @wait_api_result
  71     def get_facts(self, hostname: Optional[str] = None) -> List[Dict[str, Any]]:
  72         return self.api.get_facts(hostname)
  73
  74     @wait_api_result
  75     def remove(self, hostname: str):
  76         return self.api.remove_host(hostname)
  77
  78     @wait_api_result
  79     def add_label(self, host: str, label: str) -> OrchResult[str]:
  80         return self.api.add_host_label(host, label)
  81
  82     @wait_api_result
  83     def remove_label(self, host: str, label: str) -> OrchResult[str]:
  84         return self.api.remove_host_label(host, label)
  85
  86     @wait_api_result
  87     def drain(self, hostname: str):
  88         return self.api.drain_host(hostname)
  89
  90
  91 class InventoryManager(ResourceManager):
  92     @wait_api_result
  93     def list(self, hosts=None, refresh=False):
  94         host_filter = InventoryFilter(hosts=hosts) if hosts else None
  95         return self.api.get_inventory(host_filter=host_filter, refresh=refresh)
  96
  97
  98 class ServiceManager(ResourceManager):
  99     def list(self,
 100              service_type: Optional[str] = None,
 101              service_name: Optional[str] = None,
 102              offset: int = 0, limit: int = -1,
 103              sort: str = '+service_name', search: str = '') -> Tuple[List[Dict[Any, Any]], int]:
 104         services = self.api.describe_service(service_type, service_name)
 105         services = [service.to_dict() for service in services.result]
 106         paginator = ListPaginator(offset, limit, sort, search,
 107                                   input_list=services,
 108                                   searchable_params=['service_name', 'status.running',
 109                                                      'status.last_refreshed', 'status.size'],
 110                                   sortable_params=['service_name', 'status.running',
 111                                                    'status.last_refreshed', 'status.size'],
 112                                   default_sort='+service_name')
 113         return list(paginator.list()), paginator.get_count()
 114
 115     @wait_api_result
 116     def get(self, service_name: str) -> ServiceDescription:
 117         return self.api.describe_service(None, service_name)
 118
 119     @wait_api_result
 120     def list_daemons(self,
 121                      service_name: Optional[str] = None,
 122                      daemon_type: Optional[str] = None,
 123                      hostname: Optional[str] = None) -> List[DaemonDescription]:
 124         return self.api.list_daemons(service_name=service_name,
 125                                      daemon_type=daemon_type,
 126                                      host=hostname)
 127
 128     def reload(self, service_type, service_ids):
 129         if not isinstance(service_ids, list):
 130             service_ids = [service_ids]
 131
 132         completion_list = [
 133             self.api.service_action('reload', service_type, service_name,
 134                                     service_id)
 135             for service_name, service_id in service_ids
 136         ]
 137         self.api.orchestrator_wait(completion_list)
 138         for c in completion_list:
 139             raise_if_exception(c)
 140
 141     @wait_api_result
 142     def apply(self,
 143               service_spec: Dict,
 144               no_overwrite: Optional[bool] = False) -> OrchResult[List[str]]:
 145         spec = ServiceSpec.from_json(service_spec)
 146         return self.api.apply([spec], no_overwrite)
 147
 148     @wait_api_result
 149     def remove(self, service_name: str) -> List[str]:
 150         return self.api.remove_service(service_name)
 151
 152
 153 class OsdManager(ResourceManager):
 154     @wait_api_result
 155     def create(self, drive_group_specs):
 156         return self.api.apply_drivegroups(drive_group_specs)
 157
 158     @wait_api_result
 159     def remove(self, osd_ids, replace=False, force=False):
 160         return self.api.remove_osds(osd_ids, replace, force)
 161
 162     @wait_api_result
 163     def removing_status(self):
 164         return self.api.remove_osds_status()
 165
 166
 167 class DaemonManager(ResourceManager):
 168     @wait_api_result
 169     def action(self, daemon_name='', action='', image=None):
 170         return self.api.daemon_action(daemon_name=daemon_name, action=action, image=image)
 171
 172
 173 class OrchClient(object):
 174
 175     _instance = None
 176
 177     @classmethod
 178     def instance(cls):
 179         # type: () -> OrchClient
 180         if cls._instance is None:
 181             cls._instance = cls()
 182         return cls._instance
 183
 184     def __init__(self):
 185         self.api = OrchestratorAPI()
 186
 187         self.hosts = HostManger(self.api)
 188         self.inventory = InventoryManager(self.api)
 189         self.services = ServiceManager(self.api)
 190         self.osds = OsdManager(self.api)
 191         self.daemons = DaemonManager(self.api)
 192
 193     def available(self, features: Optional[List[str]] = None) -> bool:
 194         available = self.status()['available']
 195         if available and features is not None:
 196             return not self.get_missing_features(features)
 197         return available
 198
 199     def status(self) -> Dict[str, Any]:
 200         status = self.api.status()
 201         status['features'] = {}
 202         if status['available']:
 203             status['features'] = self.api.get_feature_set()
 204         return status
 205
 206     def get_missing_features(self, features: List[str]) -> List[str]:
 207         supported_features = {k for k, v in self.api.get_feature_set().items() if v['available']}
 208         return list(set(features) - supported_features)
 209
 210     @wait_api_result
 211     def blink_device_light(self, hostname, device, ident_fault, on):
 212         # type: (str, str, str, bool) -> OrchResult[List[str]]
 213         return self.api.blink_device_light(
 214             ident_fault, on, [DeviceLightLoc(hostname, device, device)])
 215
 216
 217 class OrchFeature(object):
 218     HOST_LIST = 'get_hosts'
 219     HOST_ADD = 'add_host'
 220     HOST_REMOVE = 'remove_host'
 221     HOST_LABEL_ADD = 'add_host_label'
 222     HOST_LABEL_REMOVE = 'remove_host_label'
 223     HOST_MAINTENANCE_ENTER = 'enter_host_maintenance'
 224     HOST_MAINTENANCE_EXIT = 'exit_host_maintenance'
 225     HOST_DRAIN = 'drain_host'
 226
 227     SERVICE_LIST = 'describe_service'
 228     SERVICE_CREATE = 'apply'
 229     SERVICE_EDIT = 'apply'
 230     SERVICE_DELETE = 'remove_service'
 231     SERVICE_RELOAD = 'service_action'
 232     DAEMON_LIST = 'list_daemons'
 233
 234     OSD_GET_REMOVE_STATUS = 'remove_osds_status'
 235
 236     OSD_CREATE = 'apply_drivegroups'
 237     OSD_DELETE = 'remove_osds'
 238
 239     DEVICE_LIST = 'get_inventory'
 240     DEVICE_BLINK_LIGHT = 'blink_device_light'
 241
 242     DAEMON_ACTION = 'daemon_action'