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

   1 # -*- coding: utf-8 -*-
   2
   3 import json
   4 import logging
   5
   6 import rados
   7 from mgr_module import CommandResult
   8 from mgr_util import get_most_recent_rate, get_time_series_rates, name_to_config_section
   9
  10 from .. import mgr
  11
  12 try:
  13     from typing import Any, Dict, Optional, Union
  14 except ImportError:
  15     pass  # For typing only
  16
  17 logger = logging.getLogger('ceph_service')
  18
  19
  20 class SendCommandError(rados.Error):
  21     def __init__(self, err, prefix, argdict, errno):
  22         self.prefix = prefix
  23         self.argdict = argdict
  24         super(SendCommandError, self).__init__(err, errno)
  25
  26
  27 # pylint: disable=too-many-public-methods
  28 class CephService(object):
  29
  30     OSD_FLAG_NO_SCRUB = 'noscrub'
  31     OSD_FLAG_NO_DEEP_SCRUB = 'nodeep-scrub'
  32
  33     PG_STATUS_SCRUBBING = 'scrubbing'
  34     PG_STATUS_DEEP_SCRUBBING = 'deep'
  35
  36     SCRUB_STATUS_DISABLED = 'Disabled'
  37     SCRUB_STATUS_ACTIVE = 'Active'
  38     SCRUB_STATUS_INACTIVE = 'Inactive'
  39
  40     @classmethod
  41     def get_service_map(cls, service_name):
  42         service_map = {}  # type: Dict[str, dict]
  43         for server in mgr.list_servers():
  44             for service in server['services']:
  45                 if service['type'] == service_name:
  46                     if server['hostname'] not in service_map:
  47                         service_map[server['hostname']] = {
  48                             'server': server,
  49                             'services': []
  50                         }
  51                     inst_id = service['id']
  52                     metadata = mgr.get_metadata(service_name, inst_id)
  53                     status = mgr.get_daemon_status(service_name, inst_id)
  54                     service_map[server['hostname']]['services'].append({
  55                         'id': inst_id,
  56                         'type': service_name,
  57                         'hostname': server['hostname'],
  58                         'metadata': metadata,
  59                         'status': status
  60                     })
  61         return service_map
  62
  63     @classmethod
  64     def get_service_list(cls, service_name):
  65         service_map = cls.get_service_map(service_name)
  66         return [svc for _, svcs in service_map.items() for svc in svcs['services']]
  67
  68     @classmethod
  69     def get_service_data_by_metadata_id(cls,
  70                                         service_type: str,
  71                                         metadata_id: str) -> Optional[Dict[str, Any]]:
  72         for server in mgr.list_servers():
  73             for service in server['services']:
  74                 if service['type'] == service_type:
  75                     metadata = mgr.get_metadata(service_type, service['id'])
  76                     if metadata_id == metadata['id']:
  77                         return {
  78                             'id': metadata['id'],
  79                             'service_map_id': str(service['id']),
  80                             'type': service_type,
  81                             'hostname': server['hostname'],
  82                             'metadata': metadata
  83                         }
  84         return None
  85
  86     @classmethod
  87     def get_service(cls, service_type: str, metadata_id: str) -> Optional[Dict[str, Any]]:
  88         svc_data = cls.get_service_data_by_metadata_id(service_type, metadata_id)
  89         if svc_data:
  90             svc_data['status'] = mgr.get_daemon_status(svc_data['type'], svc_data['service_map_id'])
  91         return svc_data
  92
  93     @classmethod
  94     def get_service_perf_counters(cls, service_type: str, service_id: str) -> Dict[str, Any]:
  95         schema_dict = mgr.get_perf_schema(service_type, service_id)
  96         schema = schema_dict["{}.{}".format(service_type, service_id)]
  97         counters = []
  98         for key, value in sorted(schema.items()):
  99             counter = {'name': str(key), 'description': value['description']}
 100             # pylint: disable=W0212
 101             if mgr._stattype_to_str(value['type']) == 'counter':
 102                 counter['value'] = cls.get_rate(
 103                     service_type, service_id, key)
 104                 counter['unit'] = mgr._unit_to_str(value['units'])
 105             else:
 106                 counter['value'] = mgr.get_latest(
 107                     service_type, service_id, key)
 108                 counter['unit'] = ''
 109             counters.append(counter)
 110
 111         return {
 112             'service': {
 113                 'type': service_type,
 114                 'id': str(service_id)
 115             },
 116             'counters': counters
 117         }
 118
 119     @classmethod
 120     def get_pool_list(cls, application=None):
 121         osd_map = mgr.get('osd_map')
 122         if not application:
 123             return osd_map['pools']
 124         return [pool for pool in osd_map['pools']
 125                 if application in pool.get('application_metadata', {})]
 126
 127     @classmethod
 128     def get_pool_list_with_stats(cls, application=None):
 129         # pylint: disable=too-many-locals
 130         pools = cls.get_pool_list(application)
 131
 132         pools_w_stats = []
 133
 134         pg_summary = mgr.get("pg_summary")
 135         pool_stats = mgr.get_updated_pool_stats()
 136
 137         for pool in pools:
 138             pool['pg_status'] = pg_summary['by_pool'][pool['pool'].__str__()]
 139             stats = pool_stats[pool['pool']]
 140             s = {}
 141
 142             for stat_name, stat_series in stats.items():
 143                 rates = get_time_series_rates(stat_series)
 144                 s[stat_name] = {
 145                     'latest': stat_series[0][1],
 146                     'rate': get_most_recent_rate(rates),
 147                     'rates': rates
 148                 }
 149             pool['stats'] = s
 150             pools_w_stats.append(pool)
 151         return pools_w_stats
 152
 153     @classmethod
 154     def get_erasure_code_profiles(cls):
 155         def _serialize_ecp(name, ecp):
 156             def serialize_numbers(key):
 157                 value = ecp.get(key)
 158                 if value is not None:
 159                     ecp[key] = int(value)
 160
 161             ecp['name'] = name
 162             serialize_numbers('k')
 163             serialize_numbers('m')
 164             return ecp
 165
 166         ret = []
 167         for name, ecp in mgr.get('osd_map').get('erasure_code_profiles', {}).items():
 168             ret.append(_serialize_ecp(name, ecp))
 169         return ret
 170
 171     @classmethod
 172     def get_pool_name_from_id(cls, pool_id):
 173         # type: (int) -> Union[str, None]
 174         return mgr.rados.pool_reverse_lookup(pool_id)
 175
 176     @classmethod
 177     def get_pool_by_attribute(cls, attribute, value):
 178         # type: (str, Any) -> Union[dict, None]
 179         pool_list = cls.get_pool_list()
 180         for pool in pool_list:
 181             if attribute in pool and pool[attribute] == value:
 182                 return pool
 183         return None
 184
 185     @classmethod
 186     def get_encryption_config(cls, daemon_name):
 187         kms_vault_configured = False
 188         s3_vault_configured = False
 189         kms_backend: str = ''
 190         sse_s3_backend: str = ''
 191         vault_stats = []
 192         full_daemon_name = 'rgw.' + daemon_name
 193
 194         kms_backend = CephService.send_command('mon', 'config get',
 195                                                who=name_to_config_section(full_daemon_name),
 196                                                key='rgw_crypt_s3_kms_backend')
 197         sse_s3_backend = CephService.send_command('mon', 'config get',
 198                                                   who=name_to_config_section(full_daemon_name),
 199                                                   key='rgw_crypt_sse_s3_backend')
 200
 201         if kms_backend.strip() == 'vault':
 202             kms_vault_auth: str = CephService.send_command('mon', 'config get',
 203                                                            who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 204                                                            key='rgw_crypt_vault_auth')
 205             kms_vault_engine: str = CephService.send_command('mon', 'config get',
 206                                                              who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 207                                                              key='rgw_crypt_vault_secret_engine')
 208             kms_vault_address: str = CephService.send_command('mon', 'config get',
 209                                                               who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 210                                                               key='rgw_crypt_vault_addr')
 211             kms_vault_token: str = CephService.send_command('mon', 'config get',
 212                                                             who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 213                                                             key='rgw_crypt_vault_token_file')  # noqa E501 #pylint: disable=line-too-long
 214             if (kms_vault_auth.strip() != "" and kms_vault_engine.strip() != "" and kms_vault_address.strip() != ""):  # noqa E501 #pylint: disable=line-too-long
 215                 if(kms_vault_auth == 'token' and kms_vault_token.strip() == ""):
 216                     kms_vault_configured = False
 217                 else:
 218                     kms_vault_configured = True
 219
 220         if sse_s3_backend.strip() == 'vault':
 221             s3_vault_auth: str = CephService.send_command('mon', 'config get',
 222                                                           who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 223                                                           key='rgw_crypt_sse_s3_vault_auth')
 224             s3_vault_engine: str = CephService.send_command('mon',
 225                                                             'config get',
 226                                                             who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 227                                                             key='rgw_crypt_sse_s3_vault_secret_engine')  # noqa E501 #pylint: disable=line-too-long
 228             s3_vault_address: str = CephService.send_command('mon', 'config get',
 229                                                              who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 230                                                              key='rgw_crypt_sse_s3_vault_addr')
 231             s3_vault_token: str = CephService.send_command('mon', 'config get',
 232                                                            who=name_to_config_section(full_daemon_name),  # noqa E501 #pylint: disable=line-too-long
 233                                                            key='rgw_crypt_sse_s3_vault_token_file')  # noqa E501 #pylint: disable=line-too-long
 234
 235             if (s3_vault_auth.strip() != "" and s3_vault_engine.strip() != "" and s3_vault_address.strip() != ""):  # noqa E501 #pylint: disable=line-too-long
 236                 if(s3_vault_auth == 'token' and s3_vault_token.strip() == ""):
 237                     s3_vault_configured = False
 238                 else:
 239                     s3_vault_configured = True
 240
 241         vault_stats.append(kms_vault_configured)
 242         vault_stats.append(s3_vault_configured)
 243         return vault_stats
 244
 245     @classmethod
 246     def set_encryption_config(cls, encryption_type, kms_provider, auth_method,
 247                               secret_engine, secret_path, namespace, address,
 248                               token, daemon_name, ssl_cert, client_cert, client_key):
 249         full_daemon_name = 'rgw.' + daemon_name
 250         if encryption_type == 'aws:kms':
 251
 252             KMS_CONFIG = [
 253                 ['rgw_crypt_s3_kms_backend', kms_provider],
 254                 ['rgw_crypt_vault_auth', auth_method],
 255                 ['rgw_crypt_vault_prefix', secret_path],
 256                 ['rgw_crypt_vault_namespace', namespace],
 257                 ['rgw_crypt_vault_secret_engine', secret_engine],
 258                 ['rgw_crypt_vault_addr', address],
 259                 ['rgw_crypt_vault_token_file', token],
 260                 ['rgw_crypt_vault_ssl_cacert', ssl_cert],
 261                 ['rgw_crypt_vault_ssl_clientcert', client_cert],
 262                 ['rgw_crypt_vault_ssl_clientkey', client_key]
 263             ]
 264
 265             for (key, value) in KMS_CONFIG:
 266                 if value == 'null':
 267                     continue
 268                 CephService.send_command('mon', 'config set',
 269                                          who=name_to_config_section(full_daemon_name),
 270                                          name=key, value=value)
 271
 272         if encryption_type == 'AES256':
 273
 274             SSE_S3_CONFIG = [
 275                 ['rgw_crypt_sse_s3_backend', kms_provider],
 276                 ['rgw_crypt_sse_s3_vault_auth', auth_method],
 277                 ['rgw_crypt_sse_s3_vault_prefix', secret_path],
 278                 ['rgw_crypt_sse_s3_vault_namespace', namespace],
 279                 ['rgw_crypt_sse_s3_vault_secret_engine', secret_engine],
 280                 ['rgw_crypt_sse_s3_vault_addr', address],
 281                 ['rgw_crypt_sse_s3_vault_token_file', token],
 282                 ['rgw_crypt_sse_s3_vault_ssl_cacert', ssl_cert],
 283                 ['rgw_crypt_sse_s3_vault_ssl_clientcert', client_cert],
 284                 ['rgw_crypt_sse_s3_vault_ssl_clientkey', client_key]
 285             ]
 286
 287             for (key, value) in SSE_S3_CONFIG:
 288                 if value == 'null':
 289                     continue
 290                 CephService.send_command('mon', 'config set',
 291                                          who=name_to_config_section(full_daemon_name),
 292                                          name=key, value=value)
 293
 294         return {}
 295
 296     @classmethod
 297     def get_pool_pg_status(cls, pool_name):
 298         # type: (str) -> dict
 299         pool = cls.get_pool_by_attribute('pool_name', pool_name)
 300         if pool is None:
 301             return {}
 302         return mgr.get("pg_summary")['by_pool'][pool['pool'].__str__()]
 303
 304     @staticmethod
 305     def send_command(srv_type, prefix, srv_spec='', to_json=True, inbuf='', **kwargs):
 306         # type: (str, str, Optional[str], bool, str, Any) -> Any
 307         """
 308         :type prefix: str
 309         :param srv_type: mon |
 310         :param kwargs: will be added to argdict
 311         :param srv_spec: typically empty. or something like "<fs_id>:0"
 312         :param to_json: if true return as json format
 313
 314         :raises PermissionError: See rados.make_ex
 315         :raises ObjectNotFound: See rados.make_ex
 316         :raises IOError: See rados.make_ex
 317         :raises NoSpace: See rados.make_ex
 318         :raises ObjectExists: See rados.make_ex
 319         :raises ObjectBusy: See rados.make_ex
 320         :raises NoData: See rados.make_ex
 321         :raises InterruptedOrTimeoutError: See rados.make_ex
 322         :raises TimedOut: See rados.make_ex
 323         :raises ValueError: return code != 0
 324         """
 325         argdict = {
 326             "prefix": prefix,
 327         }
 328         if to_json:
 329             argdict["format"] = "json"
 330         argdict.update({k: v for k, v in kwargs.items() if v is not None})
 331         result = CommandResult("")
 332         mgr.send_command(result, srv_type, srv_spec, json.dumps(argdict), "", inbuf=inbuf)
 333         r, outb, outs = result.wait()
 334         if r != 0:
 335             logger.error("send_command '%s' failed. (r=%s, outs=\"%s\", kwargs=%s)", prefix, r,
 336                          outs, kwargs)
 337
 338             raise SendCommandError(outs, prefix, argdict, r)
 339
 340         try:
 341             return json.loads(outb or outs)
 342         except Exception:  # pylint: disable=broad-except
 343             return outb
 344
 345     @staticmethod
 346     def _get_smart_data_by_device(device):
 347         # type: (dict) -> Dict[str, dict]
 348         # Check whether the device is associated with daemons.
 349         if 'daemons' in device and device['daemons']:
 350             dev_smart_data: Dict[str, Any] = {}
 351
 352             # Get a list of all OSD daemons on all hosts that are 'up'
 353             # because SMART data can not be retrieved from daemons that
 354             # are 'down' or 'destroyed'.
 355             osd_tree = CephService.send_command('mon', 'osd tree')
 356             osd_daemons_up = [
 357                 node['name'] for node in osd_tree.get('nodes', {})
 358                 if node.get('status') == 'up'
 359             ]
 360
 361             # All daemons on the same host can deliver SMART data,
 362             # thus it is not relevant for us which daemon we are using.
 363             # NOTE: the list may contain daemons that are 'down' or 'destroyed'.
 364             for daemon in device['daemons']:
 365                 svc_type, svc_id = daemon.split('.', 1)
 366                 if 'osd' in svc_type:
 367                     if daemon not in osd_daemons_up:
 368                         continue
 369                     try:
 370                         dev_smart_data = CephService.send_command(
 371                             svc_type, 'smart', svc_id, devid=device['devid'])
 372                     except SendCommandError as error:
 373                         logger.warning(str(error))
 374                         # Try to retrieve SMART data from another daemon.
 375                         continue
 376                 elif 'mon' in svc_type:
 377                     try:
 378                         dev_smart_data = CephService.send_command(
 379                             svc_type, 'device query-daemon-health-metrics', who=daemon)
 380                     except SendCommandError as error:
 381                         logger.warning(str(error))
 382                         # Try to retrieve SMART data from another daemon.
 383                         continue
 384                 else:
 385                     dev_smart_data = {}
 386
 387                 CephService.log_dev_data_error(dev_smart_data)
 388
 389                 break
 390
 391             return dev_smart_data
 392         logger.warning('[SMART] No daemons associated with device ID "%s"',
 393                        device['devid'])
 394         return {}
 395
 396     @staticmethod
 397     def log_dev_data_error(dev_smart_data):
 398         for dev_id, dev_data in dev_smart_data.items():
 399             if 'error' in dev_data:
 400                 logger.warning(
 401                     '[SMART] Error retrieving smartctl data for device ID "%s": %s',
 402                     dev_id, dev_data)
 403
 404     @staticmethod
 405     def get_devices_by_host(hostname):
 406         # type: (str) -> dict
 407         return CephService.send_command('mon',
 408                                         'device ls-by-host',
 409                                         host=hostname)
 410
 411     @staticmethod
 412     def get_devices_by_daemon(daemon_type, daemon_id):
 413         # type: (str, str) -> dict
 414         return CephService.send_command('mon',
 415                                         'device ls-by-daemon',
 416                                         who='{}.{}'.format(
 417                                             daemon_type, daemon_id))
 418
 419     @staticmethod
 420     def get_smart_data_by_host(hostname):
 421         # type: (str) -> dict
 422         """
 423         Get the SMART data of all devices on the given host, regardless
 424         of the daemon (osd, mon, ...).
 425         :param hostname: The name of the host.
 426         :return: A dictionary containing the SMART data of every device
 427           on the given host. The device name is used as the key in the
 428           dictionary.
 429         """
 430         devices = CephService.get_devices_by_host(hostname)
 431         smart_data = {}  # type: dict
 432         if devices:
 433             for device in devices:
 434                 if device['devid'] not in smart_data:
 435                     smart_data.update(
 436                         CephService._get_smart_data_by_device(device))
 437         else:
 438             logger.debug('[SMART] could not retrieve device list from host %s', hostname)
 439         return smart_data
 440
 441     @staticmethod
 442     def get_smart_data_by_daemon(daemon_type, daemon_id):
 443         # type: (str, str) -> Dict[str, dict]
 444         """
 445         Get the SMART data of the devices associated with the given daemon.
 446         :param daemon_type: The daemon type, e.g. 'osd' or 'mon'.
 447         :param daemon_id: The daemon identifier.
 448         :return: A dictionary containing the SMART data of every device
 449           associated with the given daemon. The device name is used as the
 450           key in the dictionary.
 451         """
 452         devices = CephService.get_devices_by_daemon(daemon_type, daemon_id)
 453         smart_data = {}  # type: Dict[str, dict]
 454         if devices:
 455             for device in devices:
 456                 if device['devid'] not in smart_data:
 457                     smart_data.update(
 458                         CephService._get_smart_data_by_device(device))
 459         else:
 460             msg = '[SMART] could not retrieve device list from daemon with type %s and ' +\
 461                 'with ID %s'
 462             logger.debug(msg, daemon_type, daemon_id)
 463         return smart_data
 464
 465     @classmethod
 466     def get_rates(cls, svc_type, svc_name, path):
 467         """
 468         :return: the derivative of mgr.get_counter()
 469         :rtype: list[tuple[int, float]]"""
 470         data = mgr.get_counter(svc_type, svc_name, path)[path]
 471         return get_time_series_rates(data)
 472
 473     @classmethod
 474     def get_rate(cls, svc_type, svc_name, path):
 475         """returns most recent rate"""
 476         return get_most_recent_rate(cls.get_rates(svc_type, svc_name, path))
 477
 478     @classmethod
 479     def get_client_perf(cls):
 480         pools_stats = mgr.get('osd_pool_stats')['pool_stats']
 481
 482         io_stats = {
 483             'read_bytes_sec': 0,
 484             'read_op_per_sec': 0,
 485             'write_bytes_sec': 0,
 486             'write_op_per_sec': 0,
 487         }
 488         recovery_stats = {'recovering_bytes_per_sec': 0}
 489
 490         for pool_stats in pools_stats:
 491             client_io = pool_stats['client_io_rate']
 492             for stat in list(io_stats.keys()):
 493                 if stat in client_io:
 494                     io_stats[stat] += client_io[stat]
 495
 496             client_recovery = pool_stats['recovery_rate']
 497             for stat in list(recovery_stats.keys()):
 498                 if stat in client_recovery:
 499                     recovery_stats[stat] += client_recovery[stat]
 500
 501         client_perf = io_stats.copy()
 502         client_perf.update(recovery_stats)
 503
 504         return client_perf
 505
 506     @classmethod
 507     def get_scrub_status(cls):
 508         enabled_flags = mgr.get('osd_map')['flags_set']
 509         if cls.OSD_FLAG_NO_SCRUB in enabled_flags or cls.OSD_FLAG_NO_DEEP_SCRUB in enabled_flags:
 510             return cls.SCRUB_STATUS_DISABLED
 511
 512         grouped_pg_statuses = mgr.get('pg_summary')['all']
 513         for grouped_pg_status in grouped_pg_statuses.keys():
 514             if len(grouped_pg_status.split(cls.PG_STATUS_SCRUBBING)) > 1 \
 515                     or len(grouped_pg_status.split(cls.PG_STATUS_DEEP_SCRUBBING)) > 1:
 516                 return cls.SCRUB_STATUS_ACTIVE
 517
 518         return cls.SCRUB_STATUS_INACTIVE
 519
 520     @classmethod
 521     def get_pg_info(cls):
 522         pg_summary = mgr.get('pg_summary')
 523         object_stats = {stat: pg_summary['pg_stats_sum']['stat_sum'][stat] for stat in [
 524             'num_objects', 'num_object_copies', 'num_objects_degraded',
 525             'num_objects_misplaced', 'num_objects_unfound']}
 526
 527         pgs_per_osd = 0.0
 528         total_osds = len(pg_summary['by_osd'])
 529         if total_osds > 0:
 530             total_pgs = 0.0
 531             for _, osd_pg_statuses in pg_summary['by_osd'].items():
 532                 for _, pg_amount in osd_pg_statuses.items():
 533                     total_pgs += pg_amount
 534
 535             pgs_per_osd = total_pgs / total_osds
 536
 537         return {
 538             'object_stats': object_stats,
 539             'statuses': pg_summary['all'],
 540             'pgs_per_osd': pgs_per_osd,
 541         }