diff options
Diffstat (limited to 'src/pybind/mgr/mgr_module.py')
| -rw-r--r-- | src/pybind/mgr/mgr_module.py | 2381 |
1 files changed, 2381 insertions, 0 deletions
diff --git a/src/pybind/mgr/mgr_module.py b/src/pybind/mgr/mgr_module.py new file mode 100644 index 000000000..5a7b9bfc6 --- /dev/null +++ b/src/pybind/mgr/mgr_module.py @@ -0,0 +1,2381 @@ +import ceph_module # noqa + +from typing import cast, Tuple, Any, Dict, Generic, Optional, Callable, List, \ + Mapping, NamedTuple, Sequence, Union, Set, TYPE_CHECKING +if TYPE_CHECKING: + import sys + if sys.version_info >= (3, 8): + from typing import Literal + else: + from typing_extensions import Literal + +import inspect +import logging +import errno +import functools +import json +import subprocess +import threading +from collections import defaultdict +from enum import IntEnum, Enum +import rados +import re +import socket +import sqlite3 +import sys +import time +from ceph_argparse import CephArgtype +from mgr_util import profile_method + +if sys.version_info >= (3, 8): + from typing import get_args, get_origin +else: + def get_args(tp: Any) -> Any: + if tp is Generic: + return tp + else: + return getattr(tp, '__args__', ()) + + def get_origin(tp: Any) -> Any: + return getattr(tp, '__origin__', None) + + +ERROR_MSG_EMPTY_INPUT_FILE = 'Empty input file' +ERROR_MSG_NO_INPUT_FILE = 'Input file not specified' +# Full list of strings in "osd_types.cc:pg_state_string()" +PG_STATES = [ + "active", + "clean", + "down", + "recovery_unfound", + "backfill_unfound", + "scrubbing", + "degraded", + "inconsistent", + "peering", + "repair", + "recovering", + "forced_recovery", + "backfill_wait", + "incomplete", + "stale", + "remapped", + "deep", + "backfilling", + "forced_backfill", + "backfill_toofull", + "recovery_wait", + "recovery_toofull", + "undersized", + "activating", + "peered", + "snaptrim", + "snaptrim_wait", + "snaptrim_error", + "creating", + "unknown", + "premerge", + "failed_repair", + "laggy", + "wait", +] + +NFS_GANESHA_SUPPORTED_FSALS = ['CEPH', 'RGW'] +NFS_POOL_NAME = '.nfs' + + +class NotifyType(str, Enum): + mon_map = 'mon_map' + pg_summary = 'pg_summary' + health = 'health' + clog = 'clog' + osd_map = 'osd_map' + fs_map = 'fs_map' + command = 'command' + + # these are disabled because there are no users. + # see Mgr.cc: + # service_map = 'service_map' + # mon_status = 'mon_status' + # see DaemonServer.cc: + # perf_schema_update = 'perf_schema_update' + + +class CommandResult(object): + """ + Use with MgrModule.send_command + """ + + def __init__(self, tag: Optional[str] = None): + self.ev = threading.Event() + self.outs = "" + self.outb = "" + self.r = 0 + + # This is just a convenience for notifications from + # C++ land, to avoid passing addresses around in messages. + self.tag = tag if tag else "" + + def complete(self, r: int, outb: str, outs: str) -> None: + self.r = r + self.outb = outb + self.outs = outs + self.ev.set() + + def wait(self) -> Tuple[int, str, str]: + self.ev.wait() + return self.r, self.outb, self.outs + + +class HandleCommandResult(NamedTuple): + """ + Tuple containing the result of `handle_command()` + + Only write to stderr if there is an error, or in extraordinary circumstances + + Avoid having `ceph foo bar` commands say "did foo bar" on success unless there + is critical information to include there. + + Everything programmatically consumable should be put on stdout + """ + retval: int = 0 # return code. E.g. 0 or -errno.EINVAL + stdout: str = "" # data of this result. + stderr: str = "" # Typically used for error messages. + + +class MonCommandFailed(RuntimeError): pass +class MgrDBNotReady(RuntimeError): pass + + +class OSDMap(ceph_module.BasePyOSDMap): + def get_epoch(self) -> int: + return self._get_epoch() + + def get_crush_version(self) -> int: + return self._get_crush_version() + + def dump(self) -> Dict[str, Any]: + return self._dump() + + def get_pools(self) -> Dict[int, Dict[str, Any]]: + # FIXME: efficient implementation + d = self._dump() + return dict([(p['pool'], p) for p in d['pools']]) + + def get_pools_by_name(self) -> Dict[str, Dict[str, Any]]: + # FIXME: efficient implementation + d = self._dump() + return dict([(p['pool_name'], p) for p in d['pools']]) + + def new_incremental(self) -> 'OSDMapIncremental': + return self._new_incremental() + + def apply_incremental(self, inc: 'OSDMapIncremental') -> 'OSDMap': + return self._apply_incremental(inc) + + def get_crush(self) -> 'CRUSHMap': + return self._get_crush() + + def get_pools_by_take(self, take: int) -> List[int]: + return self._get_pools_by_take(take).get('pools', []) + + def calc_pg_upmaps(self, inc: 'OSDMapIncremental', + max_deviation: int, + max_iterations: int = 10, + pools: Optional[List[str]] = None) -> int: + if pools is None: + pools = [] + return self._calc_pg_upmaps( + inc, + max_deviation, max_iterations, pools) + + def map_pool_pgs_up(self, poolid: int) -> List[int]: + return self._map_pool_pgs_up(poolid) + + def pg_to_up_acting_osds(self, pool_id: int, ps: int) -> Dict[str, Any]: + return self._pg_to_up_acting_osds(pool_id, ps) + + def pool_raw_used_rate(self, pool_id: int) -> float: + return self._pool_raw_used_rate(pool_id) + + @classmethod + def build_simple(cls, epoch: int = 1, uuid: Optional[str] = None, num_osd: int = -1) -> 'ceph_module.BasePyOSDMap': + return cls._build_simple(epoch, uuid, num_osd) + + def get_ec_profile(self, name: str) -> Optional[List[Dict[str, str]]]: + # FIXME: efficient implementation + d = self._dump() + return d['erasure_code_profiles'].get(name, None) + + def get_require_osd_release(self) -> str: + d = self._dump() + return d['require_osd_release'] + + +class OSDMapIncremental(ceph_module.BasePyOSDMapIncremental): + def get_epoch(self) -> int: + return self._get_epoch() + + def dump(self) -> Dict[str, Any]: + return self._dump() + + def set_osd_reweights(self, weightmap: Dict[int, float]) -> None: + """ + weightmap is a dict, int to float. e.g. { 0: .9, 1: 1.0, 3: .997 } + """ + return self._set_osd_reweights(weightmap) + + def set_crush_compat_weight_set_weights(self, weightmap: Dict[str, float]) -> None: + """ + weightmap is a dict, int to float. devices only. e.g., + { 0: 3.4, 1: 3.3, 2: 3.334 } + """ + return self._set_crush_compat_weight_set_weights(weightmap) + + +class CRUSHMap(ceph_module.BasePyCRUSH): + ITEM_NONE = 0x7fffffff + DEFAULT_CHOOSE_ARGS = '-1' + + def dump(self) -> Dict[str, Any]: + return self._dump() + + def get_item_weight(self, item: int) -> Optional[int]: + return self._get_item_weight(item) + + def get_item_name(self, item: int) -> Optional[str]: + return self._get_item_name(item) + + def find_takes(self) -> List[int]: + return self._find_takes().get('takes', []) + + def find_roots(self) -> List[int]: + return self._find_roots().get('roots', []) + + def get_take_weight_osd_map(self, root: int) -> Dict[int, float]: + uglymap = self._get_take_weight_osd_map(root) + return {int(k): v for k, v in uglymap.get('weights', {}).items()} + + @staticmethod + def have_default_choose_args(dump: Dict[str, Any]) -> bool: + return CRUSHMap.DEFAULT_CHOOSE_ARGS in dump.get('choose_args', {}) + + @staticmethod + def get_default_choose_args(dump: Dict[str, Any]) -> List[Dict[str, Any]]: + choose_args = dump.get('choose_args') + assert isinstance(choose_args, dict) + return choose_args.get(CRUSHMap.DEFAULT_CHOOSE_ARGS, []) + + def get_rule(self, rule_name: str) -> Optional[Dict[str, Any]]: + # TODO efficient implementation + for rule in self.dump()['rules']: + if rule_name == rule['rule_name']: + return rule + + return None + + def get_rule_by_id(self, rule_id: int) -> Optional[Dict[str, Any]]: + for rule in self.dump()['rules']: + if rule['rule_id'] == rule_id: + return rule + + return None + + def get_rule_root(self, rule_name: str) -> Optional[int]: + rule = self.get_rule(rule_name) + if rule is None: + return None + + try: + first_take = next(s for s in rule['steps'] if s.get('op') == 'take') + except StopIteration: + logging.warning("CRUSH rule '{0}' has no 'take' step".format( + rule_name)) + return None + else: + return first_take['item'] + + def get_osds_under(self, root_id: int) -> List[int]: + # TODO don't abuse dump like this + d = self.dump() + buckets = dict([(b['id'], b) for b in d['buckets']]) + + osd_list = [] + + def accumulate(b: Dict[str, Any]) -> None: + for item in b['items']: + if item['id'] >= 0: + osd_list.append(item['id']) + else: + try: + accumulate(buckets[item['id']]) + except KeyError: + pass + + accumulate(buckets[root_id]) + + return osd_list + + def device_class_counts(self) -> Dict[str, int]: + result = defaultdict(int) # type: Dict[str, int] + # TODO don't abuse dump like this + d = self.dump() + for device in d['devices']: + cls = device.get('class', None) + result[cls] += 1 + + return dict(result) + + +HandlerFuncType = Callable[..., Tuple[int, str, str]] + +def _extract_target_func( + f: HandlerFuncType +) -> Tuple[HandlerFuncType, Dict[str, Any]]: + """In order to interoperate with other decorated functions, + we need to find the original function which will provide + the main set of arguments. While we descend through the + stack of wrapped functions, gather additional arguments + the decorators may want to provide. + """ + # use getattr to keep mypy happy + wrapped = getattr(f, "__wrapped__", None) + if not wrapped: + return f, {} + extra_args: Dict[str, Any] = {} + while wrapped is not None: + extra_args.update(getattr(f, "extra_args", {})) + f = wrapped + wrapped = getattr(f, "__wrapped__", None) + return f, extra_args + + +class CLICommand(object): + COMMANDS = {} # type: Dict[str, CLICommand] + + def __init__(self, + prefix: str, + perm: str = 'rw', + poll: bool = False): + self.prefix = prefix + self.perm = perm + self.poll = poll + self.func = None # type: Optional[Callable] + self.arg_spec = {} # type: Dict[str, Any] + self.first_default = -1 + + KNOWN_ARGS = '_', 'self', 'mgr', 'inbuf', 'return' + + @classmethod + def _load_func_metadata(cls: Any, f: HandlerFuncType) -> Tuple[str, Dict[str, Any], int, str]: + f, extra_args = _extract_target_func(f) + desc = (inspect.getdoc(f) or '').replace('\n', ' ') + full_argspec = inspect.getfullargspec(f) + arg_spec = full_argspec.annotations + first_default = len(arg_spec) + if full_argspec.defaults: + first_default -= len(full_argspec.defaults) + args = [] + positional = True + for index, arg in enumerate(full_argspec.args): + if arg in cls.KNOWN_ARGS: + # record that this function takes an inbuf if it is present + # in the full_argspec and not already in the arg_spec + if arg == 'inbuf' and 'inbuf' not in arg_spec: + arg_spec['inbuf'] = 'str' + continue + if arg == '_end_positional_': + positional = False + continue + if ( + arg == 'format' + or arg_spec[arg] is Optional[bool] + or arg_spec[arg] is bool + ): + # implicit switch to non-positional on any + # Optional[bool] or the --format option + positional = False + assert arg in arg_spec, \ + f"'{arg}' is not annotated for {f}: {full_argspec}" + has_default = index >= first_default + args.append(CephArgtype.to_argdesc(arg_spec[arg], + dict(name=arg), + has_default, + positional)) + for argname, argtype in extra_args.items(): + # avoid shadowing args from the function + if argname in arg_spec: + continue + arg_spec[argname] = argtype + args.append(CephArgtype.to_argdesc( + argtype, dict(name=argname), has_default=True, positional=False + )) + return desc, arg_spec, first_default, ' '.join(args) + + def store_func_metadata(self, f: HandlerFuncType) -> None: + self.desc, self.arg_spec, self.first_default, self.args = \ + self._load_func_metadata(f) + + def __call__(self, func: HandlerFuncType) -> HandlerFuncType: + self.store_func_metadata(func) + self.func = func + self.COMMANDS[self.prefix] = self + return self.func + + def _get_arg_value(self, kwargs_switch: bool, key: str, val: Any) -> Tuple[bool, str, Any]: + def start_kwargs() -> bool: + if isinstance(val, str) and '=' in val: + k, v = val.split('=', 1) + if k in self.arg_spec: + return True + return False + + if not kwargs_switch: + kwargs_switch = start_kwargs() + + if kwargs_switch: + k, v = val.split('=', 1) + else: + k, v = key, val + return kwargs_switch, k.replace('-', '_'), v + + def _collect_args_by_argspec(self, cmd_dict: Dict[str, Any]) -> Tuple[Dict[str, Any], Set[str]]: + kwargs = {} + special_args = set() + kwargs_switch = False + for index, (name, tp) in enumerate(self.arg_spec.items()): + if name in CLICommand.KNOWN_ARGS: + special_args.add(name) + continue + assert self.first_default >= 0 + raw_v = cmd_dict.get(name) + if index >= self.first_default: + if raw_v is None: + continue + kwargs_switch, k, v = self._get_arg_value(kwargs_switch, + name, raw_v) + kwargs[k] = CephArgtype.cast_to(tp, v) + return kwargs, special_args + + def call(self, + mgr: Any, + cmd_dict: Dict[str, Any], + inbuf: Optional[str] = None) -> HandleCommandResult: + kwargs, specials = self._collect_args_by_argspec(cmd_dict) + if inbuf: + if 'inbuf' not in specials: + return HandleCommandResult( + -errno.EINVAL, + '', + 'Invalid command: Input file data (-i) not supported', + ) + kwargs['inbuf'] = inbuf + assert self.func + return self.func(mgr, **kwargs) + + def dump_cmd(self) -> Dict[str, Union[str, bool]]: + return { + 'cmd': '{} {}'.format(self.prefix, self.args), + 'desc': self.desc, + 'perm': self.perm, + 'poll': self.poll, + } + + @classmethod + def dump_cmd_list(cls) -> List[Dict[str, Union[str, bool]]]: + return [cmd.dump_cmd() for cmd in cls.COMMANDS.values()] + + +def CLIReadCommand(prefix: str, poll: bool = False) -> CLICommand: + return CLICommand(prefix, "r", poll) + + +def CLIWriteCommand(prefix: str, poll: bool = False) -> CLICommand: + return CLICommand(prefix, "w", poll) + + +def CLICheckNonemptyFileInput(desc: str) -> Callable[[HandlerFuncType], HandlerFuncType]: + def CheckFileInput(func: HandlerFuncType) -> HandlerFuncType: + @functools.wraps(func) + def check(*args: Any, **kwargs: Any) -> Tuple[int, str, str]: + if 'inbuf' not in kwargs: + return -errno.EINVAL, '', f'{ERROR_MSG_NO_INPUT_FILE}: Please specify the file '\ + f'containing {desc} with "-i" option' + if isinstance(kwargs['inbuf'], str): + # Delete new line separator at EOF (it may have been added by a text editor). + kwargs['inbuf'] = kwargs['inbuf'].rstrip('\r\n').rstrip('\n') + if not kwargs['inbuf'] or not kwargs['inbuf'].strip(): + return -errno.EINVAL, '', f'{ERROR_MSG_EMPTY_INPUT_FILE}: Please add {desc} to '\ + 'the file' + return func(*args, **kwargs) + check.__signature__ = inspect.signature(func) # type: ignore[attr-defined] + return check + return CheckFileInput + +# If the mgr loses its lock on the database because e.g. the pgs were +# transiently down, then close it and allow it to be reopened. +MAX_DBCLEANUP_RETRIES = 3 +def MgrModuleRecoverDB(func: Callable) -> Callable: + @functools.wraps(func) + def check(self: MgrModule, *args: Any, **kwargs: Any) -> Any: + retries = 0 + while True: + try: + return func(self, *args, **kwargs) + except sqlite3.DatabaseError as e: + self.log.error(f"Caught fatal database error: {e}") + retries = retries+1 + if retries > MAX_DBCLEANUP_RETRIES: + raise + self.log.debug(f"attempting reopen of database") + self.close_db() + self.open_db(); + # allow retry of func(...) + check.__signature__ = inspect.signature(func) # type: ignore[attr-defined] + return check + +def CLIRequiresDB(func: HandlerFuncType) -> HandlerFuncType: + @functools.wraps(func) + def check(self: MgrModule, *args: Any, **kwargs: Any) -> Tuple[int, str, str]: + if not self.db_ready(): + return -errno.EAGAIN, "", "mgr db not yet available" + return func(self, *args, **kwargs) + check.__signature__ = inspect.signature(func) # type: ignore[attr-defined] + return check + +def _get_localized_key(prefix: str, key: str) -> str: + return '{}/{}'.format(prefix, key) + + +""" +MODULE_OPTIONS types and Option Class +""" +if TYPE_CHECKING: + OptionTypeLabel = Literal[ + 'uint', 'int', 'str', 'float', 'bool', 'addr', 'addrvec', 'uuid', 'size', 'secs'] + + +# common/options.h: value_t +OptionValue = Optional[Union[bool, int, float, str]] + + +class Option(Dict): + """ + Helper class to declare options for MODULE_OPTIONS list. + TODO: Replace with typing.TypedDict when in python_version >= 3.8 + """ + + def __init__( + self, + name: str, + default: OptionValue = None, + type: 'OptionTypeLabel' = 'str', + desc: Optional[str] = None, + long_desc: Optional[str] = None, + min: OptionValue = None, + max: OptionValue = None, + enum_allowed: Optional[List[str]] = None, + tags: Optional[List[str]] = None, + see_also: Optional[List[str]] = None, + runtime: bool = False, + ): + super(Option, self).__init__( + (k, v) for k, v in vars().items() + if k != 'self' and v is not None) + + +class Command(dict): + """ + Helper class to declare options for COMMANDS list. + + It also allows to specify prefix and args separately, as well as storing a + handler callable. + + Usage: + >>> def handler(): return 0, "", "" + >>> Command(prefix="example", + ... handler=handler, + ... perm='w') + {'perm': 'w', 'poll': False} + """ + + def __init__( + self, + prefix: str, + handler: HandlerFuncType, + perm: str = "rw", + poll: bool = False, + ): + super().__init__(perm=perm, + poll=poll) + self.prefix = prefix + self.handler = handler + + @staticmethod + def returns_command_result(instance: Any, + f: HandlerFuncType) -> Callable[..., HandleCommandResult]: + @functools.wraps(f) + def wrapper(mgr: Any, *args: Any, **kwargs: Any) -> HandleCommandResult: + retval, stdout, stderr = f(instance or mgr, *args, **kwargs) + return HandleCommandResult(retval, stdout, stderr) + wrapper.__signature__ = inspect.signature(f) # type: ignore[attr-defined] + return wrapper + + def register(self, instance: bool = False) -> HandlerFuncType: + """ + Register a CLICommand handler. It allows an instance to register bound + methods. In that case, the mgr instance is not passed, and it's expected + to be available in the class instance. + It also uses HandleCommandResult helper to return a wrapped a tuple of 3 + items. + """ + cmd = CLICommand(prefix=self.prefix, perm=self['perm']) + return cmd(self.returns_command_result(instance, self.handler)) + + +class CPlusPlusHandler(logging.Handler): + def __init__(self, module_inst: Any): + super(CPlusPlusHandler, self).__init__() + self._module = module_inst + self.setFormatter(logging.Formatter("[{} %(levelname)-4s %(name)s] %(message)s" + .format(module_inst.module_name))) + + def emit(self, record: logging.LogRecord) -> None: + if record.levelno >= self.level: + self._module._ceph_log(self.format(record)) + + +class ClusterLogHandler(logging.Handler): + def __init__(self, module_inst: Any): + super().__init__() + self._module = module_inst + self.setFormatter(logging.Formatter("%(message)s")) + + def emit(self, record: logging.LogRecord) -> None: + levelmap = { + logging.DEBUG: MgrModule.ClusterLogPrio.DEBUG, + logging.INFO: MgrModule.ClusterLogPrio.INFO, + logging.WARNING: MgrModule.ClusterLogPrio.WARN, + logging.ERROR: MgrModule.ClusterLogPrio.ERROR, + logging.CRITICAL: MgrModule.ClusterLogPrio.ERROR, + } + level = levelmap[record.levelno] + if record.levelno >= self.level: + self._module.cluster_log(self._module.module_name, + level, + self.format(record)) + + +class FileHandler(logging.FileHandler): + def __init__(self, module_inst: Any): + path = module_inst.get_ceph_option("log_file") + idx = path.rfind(".log") + if idx != -1: + self.path = "{}.{}.log".format(path[:idx], module_inst.module_name) + else: + self.path = "{}.{}".format(path, module_inst.module_name) + super(FileHandler, self).__init__(self.path, delay=True) + self.setFormatter(logging.Formatter("%(asctime)s [%(threadName)s] [%(levelname)-4s] [%(name)s] %(message)s")) + + +class MgrModuleLoggingMixin(object): + def _configure_logging(self, + mgr_level: str, + module_level: str, + cluster_level: str, + log_to_file: bool, + log_to_cluster: bool) -> None: + self._mgr_level: Optional[str] = None + self._module_level: Optional[str] = None + self._root_logger = logging.getLogger() + + self._unconfigure_logging() + + # the ceph log handler is initialized only once + self._mgr_log_handler = CPlusPlusHandler(self) + self._cluster_log_handler = ClusterLogHandler(self) + self._file_log_handler = FileHandler(self) + + self.log_to_file = log_to_file + self.log_to_cluster = log_to_cluster + + self._root_logger.addHandler(self._mgr_log_handler) + if log_to_file: + self._root_logger.addHandler(self._file_log_handler) + if log_to_cluster: + self._root_logger.addHandler(self._cluster_log_handler) + + self._root_logger.setLevel(logging.NOTSET) + self._set_log_level(mgr_level, module_level, cluster_level) + + def _unconfigure_logging(self) -> None: + # remove existing handlers: + rm_handlers = [ + h for h in self._root_logger.handlers + if (isinstance(h, CPlusPlusHandler) or + isinstance(h, FileHandler) or + isinstance(h, ClusterLogHandler))] + for h in rm_handlers: + self._root_logger.removeHandler(h) + self.log_to_file = False + self.log_to_cluster = False + + def _set_log_level(self, + mgr_level: str, + module_level: str, + cluster_level: str) -> None: + self._cluster_log_handler.setLevel(cluster_level.upper()) + + module_level = module_level.upper() if module_level else '' + if not self._module_level: + # using debug_mgr level + if not module_level and self._mgr_level == mgr_level: + # no change in module level neither in debug_mgr + return + else: + if self._module_level == module_level: + # no change in module level + return + + if not self._module_level and not module_level: + level = self._ceph_log_level_to_python(mgr_level) + self.getLogger().debug("setting log level based on debug_mgr: %s (%s)", + level, mgr_level) + elif self._module_level and not module_level: + level = self._ceph_log_level_to_python(mgr_level) + self.getLogger().warning("unsetting module log level, falling back to " + "debug_mgr level: %s (%s)", level, mgr_level) + elif module_level: + level = module_level + self.getLogger().debug("setting log level: %s", level) + + self._module_level = module_level + self._mgr_level = mgr_level + + self._mgr_log_handler.setLevel(level) + self._file_log_handler.setLevel(level) + + def _enable_file_log(self) -> None: + # enable file log + self.getLogger().warning("enabling logging to file") + self.log_to_file = True + self._root_logger.addHandler(self._file_log_handler) + + def _disable_file_log(self) -> None: + # disable file log + self.getLogger().warning("disabling logging to file") + self.log_to_file = False + self._root_logger.removeHandler(self._file_log_handler) + + def _enable_cluster_log(self) -> None: + # enable cluster log + self.getLogger().warning("enabling logging to cluster") + self.log_to_cluster = True + self._root_logger.addHandler(self._cluster_log_handler) + + def _disable_cluster_log(self) -> None: + # disable cluster log + self.getLogger().warning("disabling logging to cluster") + self.log_to_cluster = False + self._root_logger.removeHandler(self._cluster_log_handler) + + def _ceph_log_level_to_python(self, log_level: str) -> str: + if log_level: + try: + ceph_log_level = int(log_level.split("/", 1)[0]) + except ValueError: + ceph_log_level = 0 + else: + ceph_log_level = 0 + + log_level = "DEBUG" + if ceph_log_level <= 0: + log_level = "CRITICAL" + elif ceph_log_level <= 1: + log_level = "WARNING" + elif ceph_log_level <= 4: + log_level = "INFO" + return log_level + + def getLogger(self, name: Optional[str] = None) -> logging.Logger: + return logging.getLogger(name) + + +class MgrStandbyModule(ceph_module.BaseMgrStandbyModule, MgrModuleLoggingMixin): + """ + Standby modules only implement a serve and shutdown method, they + are not permitted to implement commands and they do not receive + any notifications. + + They only have access to the mgrmap (for accessing service URI info + from their active peer), and to configuration settings (read only). + """ + + MODULE_OPTIONS: List[Option] = [] + MODULE_OPTION_DEFAULTS = {} # type: Dict[str, Any] + + def __init__(self, module_name: str, capsule: Any): + super(MgrStandbyModule, self).__init__(capsule) + self.module_name = module_name + + # see also MgrModule.__init__() + for o in self.MODULE_OPTIONS: + if 'default' in o: + if 'type' in o: + self.MODULE_OPTION_DEFAULTS[o['name']] = o['default'] + else: + self.MODULE_OPTION_DEFAULTS[o['name']] = str(o['default']) + + # mock does not return a str + mgr_level = cast(str, self.get_ceph_option("debug_mgr")) + log_level = cast(str, self.get_module_option("log_level")) + cluster_level = cast(str, self.get_module_option('log_to_cluster_level')) + self._configure_logging(mgr_level, log_level, cluster_level, + False, False) + + # for backwards compatibility + self._logger = self.getLogger() + + def __del__(self) -> None: + self._cleanup() + self._unconfigure_logging() + + def _cleanup(self) -> None: + pass + + @classmethod + def _register_options(cls, module_name: str) -> None: + cls.MODULE_OPTIONS.append( + Option(name='log_level', type='str', default="", runtime=True, + enum_allowed=['info', 'debug', 'critical', 'error', + 'warning', ''])) + cls.MODULE_OPTIONS.append( + Option(name='log_to_file', type='bool', default=False, runtime=True)) + if not [x for x in cls.MODULE_OPTIONS if x['name'] == 'log_to_cluster']: + cls.MODULE_OPTIONS.append( + Option(name='log_to_cluster', type='bool', default=False, + runtime=True)) + cls.MODULE_OPTIONS.append( + Option(name='log_to_cluster_level', type='str', default='info', + runtime=True, + enum_allowed=['info', 'debug', 'critical', 'error', + 'warning', ''])) + + @property + def log(self) -> logging.Logger: + return self._logger + + def serve(self) -> None: + """ + The serve method is mandatory for standby modules. + :return: + """ + raise NotImplementedError() + + def get_mgr_id(self) -> str: + return self._ceph_get_mgr_id() + + def get_module_option(self, key: str, default: OptionValue = None) -> OptionValue: + """ + Retrieve the value of a persistent configuration setting + + :param default: the default value of the config if it is not found + """ + r = self._ceph_get_module_option(key) + if r is None: + return self.MODULE_OPTION_DEFAULTS.get(key, default) + else: + return r + + def get_ceph_option(self, key: str) -> OptionValue: + return self._ceph_get_option(key) + + def get_store(self, key: str) -> Optional[str]: + """ + Retrieve the value of a persistent KV store entry + + :param key: String + :return: Byte string or None + """ + return self._ceph_get_store(key) + + def get_localized_store(self, key: str, default: Optional[str] = None) -> Optional[str]: + r = self._ceph_get_store(_get_localized_key(self.get_mgr_id(), key)) + if r is None: + r = self._ceph_get_store(key) + if r is None: + r = default + return r + + def get_active_uri(self) -> str: + return self._ceph_get_active_uri() + + def get(self, data_name: str) -> Dict[str, Any]: + return self._ceph_get(data_name) + + def get_mgr_ip(self) -> str: + ips = self.get("mgr_ips").get('ips', []) + if not ips: + return socket.gethostname() + return ips[0] + + def get_hostname(self) -> str: + return socket.gethostname() + + def get_localized_module_option(self, key: str, default: OptionValue = None) -> OptionValue: + r = self._ceph_get_module_option(key, self.get_mgr_id()) + if r is None: + return self.MODULE_OPTION_DEFAULTS.get(key, default) + else: + return r + + +HealthChecksT = Mapping[str, Mapping[str, Union[int, str, Sequence[str]]]] +# {"type": service_type, "id": service_id} +ServiceInfoT = Dict[str, str] +# {"hostname": hostname, +# "ceph_version": version, +# "services": [service_info, ..]} +ServerInfoT = Dict[str, Union[str, List[ServiceInfoT]]] +PerfCounterT = Dict[str, Any] + + +class API: + def DecoratorFactory(attr: str, default: Any): # type: ignore + class DecoratorClass: + _ATTR_TOKEN = f'__ATTR_{attr.upper()}__' + + def __init__(self, value: Any=default) -> None: + self.value = value + + def __call__(self, func: Callable) -> Any: + setattr(func, self._ATTR_TOKEN, self.value) + return func + + @classmethod + def get(cls, func: Callable) -> Any: + return getattr(func, cls._ATTR_TOKEN, default) + + return DecoratorClass + + perm = DecoratorFactory('perm', default='r') + expose = DecoratorFactory('expose', default=False)(True) + + +class MgrModule(ceph_module.BaseMgrModule, MgrModuleLoggingMixin): + MGR_POOL_NAME = ".mgr" + + COMMANDS = [] # type: List[Any] + MODULE_OPTIONS: List[Option] = [] + MODULE_OPTION_DEFAULTS = {} # type: Dict[str, Any] + + # Database Schema + SCHEMA = None # type: Optional[str] + SCHEMA_VERSIONED = None # type: Optional[List[str]] + + # Priority definitions for perf counters + PRIO_CRITICAL = 10 + PRIO_INTERESTING = 8 + PRIO_USEFUL = 5 + PRIO_UNINTERESTING = 2 + PRIO_DEBUGONLY = 0 + + # counter value types + PERFCOUNTER_TIME = 1 + PERFCOUNTER_U64 = 2 + + # counter types + PERFCOUNTER_LONGRUNAVG = 4 + PERFCOUNTER_COUNTER = 8 + PERFCOUNTER_HISTOGRAM = 0x10 + PERFCOUNTER_TYPE_MASK = ~3 + + # units supported + BYTES = 0 + NONE = 1 + + # Cluster log priorities + class ClusterLogPrio(IntEnum): + DEBUG = 0 + INFO = 1 + SEC = 2 + WARN = 3 + ERROR = 4 + + def __init__(self, module_name: str, py_modules_ptr: object, this_ptr: object): + self.module_name = module_name + super(MgrModule, self).__init__(py_modules_ptr, this_ptr) + + for o in self.MODULE_OPTIONS: + if 'default' in o: + if 'type' in o: + # we'll assume the declared type matches the + # supplied default value's type. + self.MODULE_OPTION_DEFAULTS[o['name']] = o['default'] + else: + # module not declaring it's type, so normalize the + # default value to be a string for consistent behavior + # with default and user-supplied option values. + self.MODULE_OPTION_DEFAULTS[o['name']] = str(o['default']) + + mgr_level = cast(str, self.get_ceph_option("debug_mgr")) + log_level = cast(str, self.get_module_option("log_level")) + cluster_level = cast(str, self.get_module_option('log_to_cluster_level')) + log_to_file = self.get_module_option("log_to_file") + assert isinstance(log_to_file, bool) + log_to_cluster = self.get_module_option("log_to_cluster") + assert isinstance(log_to_cluster, bool) + self._configure_logging(mgr_level, log_level, cluster_level, + log_to_file, log_to_cluster) + + # for backwards compatibility + self._logger = self.getLogger() + + self._db = None # type: Optional[sqlite3.Connection] + + self._version = self._ceph_get_version() + + self._perf_schema_cache = None + + # Keep a librados instance for those that need it. + self._rados: Optional[rados.Rados] = None + + # this does not change over the lifetime of an active mgr + self._mgr_ips: Optional[str] = None + + self._db_lock = threading.Lock() + + def __del__(self) -> None: + self._unconfigure_logging() + + @classmethod + def _register_options(cls, module_name: str) -> None: + cls.MODULE_OPTIONS.append( + Option(name='log_level', type='str', default="", runtime=True, + enum_allowed=['info', 'debug', 'critical', 'error', + 'warning', ''])) + cls.MODULE_OPTIONS.append( + Option(name='log_to_file', type='bool', default=False, runtime=True)) + if not [x for x in cls.MODULE_OPTIONS if x['name'] == 'log_to_cluster']: + cls.MODULE_OPTIONS.append( + Option(name='log_to_cluster', type='bool', default=False, + runtime=True)) + cls.MODULE_OPTIONS.append( + Option(name='log_to_cluster_level', type='str', default='info', + runtime=True, + enum_allowed=['info', 'debug', 'critical', 'error', + 'warning', ''])) + + @classmethod + def _register_commands(cls, module_name: str) -> None: + cls.COMMANDS.extend(CLICommand.dump_cmd_list()) + + @property + def log(self) -> logging.Logger: + return self._logger + + def cluster_log(self, channel: str, priority: ClusterLogPrio, message: str) -> None: + """ + :param channel: The log channel. This can be 'cluster', 'audit', ... + :param priority: The log message priority. + :param message: The message to log. + """ + self._ceph_cluster_log(channel, priority.value, message) + + @property + def version(self) -> str: + return self._version + + @API.expose + def pool_exists(self, name: str) -> bool: + pools = [p['pool_name'] for p in self.get('osd_map')['pools']] + return name in pools + + @API.expose + def have_enough_osds(self) -> bool: + # wait until we have enough OSDs to allow the pool to be healthy + ready = 0 + for osd in self.get("osd_map")["osds"]: + if osd["up"] and osd["in"]: + ready += 1 + + need = cast(int, self.get_ceph_option("osd_pool_default_size")) + return ready >= need + + @API.perm('w') + @API.expose + def rename_pool(self, srcpool: str, destpool: str) -> None: + c = { + 'prefix': 'osd pool rename', + 'format': 'json', + 'srcpool': srcpool, + 'destpool': destpool, + 'yes_i_really_mean_it': True + } + self.check_mon_command(c) + + @API.perm('w') + @API.expose + def create_pool(self, pool: str) -> None: + c = { + 'prefix': 'osd pool create', + 'format': 'json', + 'pool': pool, + 'pg_num': 1, + 'pg_num_min': 1, + 'pg_num_max': 32, + 'yes_i_really_mean_it': True + } + self.check_mon_command(c) + + @API.perm('w') + @API.expose + def appify_pool(self, pool: str, app: str) -> None: + c = { + 'prefix': 'osd pool application enable', + 'format': 'json', + 'pool': pool, + 'app': app, + 'yes_i_really_mean_it': True + } + self.check_mon_command(c) + + @API.perm('w') + @API.expose + def create_mgr_pool(self) -> None: + self.log.info("creating mgr pool") + + ov = self.get_module_option_ex('devicehealth', 'pool_name', 'device_health_metrics') + devhealth = cast(str, ov) + if devhealth is not None and self.pool_exists(devhealth): + self.log.debug("reusing devicehealth pool") + self.rename_pool(devhealth, self.MGR_POOL_NAME) + self.appify_pool(self.MGR_POOL_NAME, 'mgr') + else: + self.log.debug("creating new mgr pool") + self.create_pool(self.MGR_POOL_NAME) + self.appify_pool(self.MGR_POOL_NAME, 'mgr') + + def create_skeleton_schema(self, db: sqlite3.Connection) -> None: + SQL = """ + CREATE TABLE IF NOT EXISTS MgrModuleKV ( + key TEXT PRIMARY KEY, + value NOT NULL + ) WITHOUT ROWID; + INSERT OR IGNORE INTO MgrModuleKV (key, value) VALUES ('__version', 0); + """ + + db.executescript(SQL) + + def update_schema_version(self, db: sqlite3.Connection, version: int) -> None: + SQL = "UPDATE OR ROLLBACK MgrModuleKV SET value = ? WHERE key = '__version';" + + db.execute(SQL, (version,)) + + def set_kv(self, key: str, value: Any) -> None: + SQL = "INSERT OR REPLACE INTO MgrModuleKV (key, value) VALUES (?, ?);" + + assert key[:2] != "__" + + self.log.debug(f"set_kv('{key}', '{value}')") + + with self._db_lock, self.db: + self.db.execute(SQL, (key, value)) + + @API.expose + def get_kv(self, key: str) -> Any: + SQL = "SELECT value FROM MgrModuleKV WHERE key = ?;" + + assert key[:2] != "__" + + self.log.debug(f"get_kv('{key}')") + + with self._db_lock, self.db: + cur = self.db.execute(SQL, (key,)) + row = cur.fetchone() + if row is None: + return None + else: + v = row['value'] + self.log.debug(f" = {v}") + return v + + def maybe_upgrade(self, db: sqlite3.Connection, version: int) -> None: + if version <= 0: + self.log.info(f"creating main.db for {self.module_name}") + assert self.SCHEMA is not None + db.executescript(self.SCHEMA) + self.update_schema_version(db, 1) + else: + assert self.SCHEMA_VERSIONED is not None + latest = len(self.SCHEMA_VERSIONED) + if latest < version: + raise RuntimeError(f"main.db version is newer ({version}) than module ({latest})") + for i in range(version, latest): + self.log.info(f"upgrading main.db for {self.module_name} from {i-1}:{i}") + SQL = self.SCHEMA_VERSIONED[i] + db.executescript(SQL) + if version < latest: + self.update_schema_version(db, latest) + + def load_schema(self, db: sqlite3.Connection) -> None: + SQL = """ + SELECT value FROM MgrModuleKV WHERE key = '__version'; + """ + + with db: + self.create_skeleton_schema(db) + cur = db.execute(SQL) + row = cur.fetchone() + self.maybe_upgrade(db, int(row['value'])) + assert cur.fetchone() is None + cur.close() + + def configure_db(self, db: sqlite3.Connection) -> None: + db.execute('PRAGMA FOREIGN_KEYS = 1') + db.execute('PRAGMA JOURNAL_MODE = PERSIST') + db.execute('PRAGMA PAGE_SIZE = 65536') + db.execute('PRAGMA CACHE_SIZE = 64') + db.execute('PRAGMA TEMP_STORE = memory') + db.row_factory = sqlite3.Row + self.load_schema(db) + + def close_db(self) -> None: + with self._db_lock: + if self._db is not None: + self._db.close() + self._db = None + + def open_db(self) -> Optional[sqlite3.Connection]: + if not self.pool_exists(self.MGR_POOL_NAME): + if not self.have_enough_osds(): + return None + self.create_mgr_pool() + uri = f"file:///{self.MGR_POOL_NAME}:{self.module_name}/main.db?vfs=ceph"; + self.log.debug(f"using uri {uri}") + db = sqlite3.connect(uri, check_same_thread=False, uri=True) + # if libcephsqlite reconnects, update the addrv for blocklist + with db: + cur = db.execute('SELECT json_extract(ceph_status(), "$.addr");') + (addrv,) = cur.fetchone() + assert addrv is not None + self.log.debug(f"new libcephsqlite addrv = {addrv}") + self._ceph_register_client("libcephsqlite", addrv, True) + self.configure_db(db) + return db + + @API.expose + def db_ready(self) -> bool: + with self._db_lock: + try: + return self.db is not None + except MgrDBNotReady: + return False + + @property + def db(self) -> sqlite3.Connection: + assert self._db_lock.locked() + if self._db is not None: + return self._db + db_allowed = self.get_ceph_option("mgr_pool") + if not db_allowed: + raise MgrDBNotReady(); + self._db = self.open_db() + if self._db is None: + raise MgrDBNotReady(); + return self._db + + @property + def release_name(self) -> str: + """ + Get the release name of the Ceph version, e.g. 'nautilus' or 'octopus'. + :return: Returns the release name of the Ceph version in lower case. + :rtype: str + """ + return self._ceph_get_release_name() + + @API.expose + def lookup_release_name(self, major: int) -> str: + return self._ceph_lookup_release_name(major) + + def get_context(self) -> object: + """ + :return: a Python capsule containing a C++ CephContext pointer + """ + return self._ceph_get_context() + + def notify(self, notify_type: NotifyType, notify_id: str) -> None: + """ + Called by the ceph-mgr service to notify the Python plugin + that new state is available. This method is *only* called for + notify_types that are listed in the NOTIFY_TYPES string list + member of the module class. + + :param notify_type: string indicating what kind of notification, + such as osd_map, mon_map, fs_map, mon_status, + health, pg_summary, command, service_map + :param notify_id: string (may be empty) that optionally specifies + which entity is being notified about. With + "command" notifications this is set to the tag + ``from send_command``. + """ + pass + + def _config_notify(self) -> None: + # check logging options for changes + mgr_level = cast(str, self.get_ceph_option("debug_mgr")) + module_level = cast(str, self.get_module_option("log_level")) + cluster_level = cast(str, self.get_module_option("log_to_cluster_level")) + assert isinstance(cluster_level, str) + log_to_file = self.get_module_option("log_to_file", False) + assert isinstance(log_to_file, bool) + log_to_cluster = self.get_module_option("log_to_cluster", False) + assert isinstance(log_to_cluster, bool) + self._set_log_level(mgr_level, module_level, cluster_level) + + if log_to_file != self.log_to_file: + if log_to_file: + self._enable_file_log() + else: + self._disable_file_log() + if log_to_cluster != self.log_to_cluster: + if log_to_cluster: + self._enable_cluster_log() + else: + self._disable_cluster_log() + + # call module subclass implementations + self.config_notify() + + def config_notify(self) -> None: + """ + Called by the ceph-mgr service to notify the Python plugin + that the configuration may have changed. Modules will want to + refresh any configuration values stored in config variables. + """ + pass + + def serve(self) -> None: + """ + Called by the ceph-mgr service to start any server that + is provided by this Python plugin. The implementation + of this function should block until ``shutdown`` is called. + + You *must* implement ``shutdown`` if you implement ``serve`` + """ + pass + + def shutdown(self) -> None: + """ + Called by the ceph-mgr service to request that this + module drop out of its serve() function. You do not + need to implement this if you do not implement serve() + + :return: None + """ + if self._rados: + addrs = self._rados.get_addrs() + self._rados.shutdown() + self._ceph_unregister_client(None, addrs) + self._rados = None + + @API.expose + def get(self, data_name: str) -> Any: + """ + Called by the plugin to fetch named cluster-wide objects from ceph-mgr. + + :param str data_name: Valid things to fetch are osdmap_crush_map_text, + osd_map, osd_map_tree, osd_map_crush, config, mon_map, fs_map, + osd_metadata, pg_summary, io_rate, pg_dump, df, osd_stats, + health, mon_status, devices, device <devid>, pg_stats, + pool_stats, pg_ready, osd_ping_times, mgr_map, mgr_ips, + modified_config_options, service_map, mds_metadata, + have_local_config_map, osd_pool_stats, pg_status. + + Note: + All these structures have their own JSON representations: experiment + or look at the C++ ``dump()`` methods to learn about them. + """ + obj = self._ceph_get(data_name) + if isinstance(obj, bytes): + obj = json.loads(obj) + + return obj + + def _stattype_to_str(self, stattype: int) -> str: + + typeonly = stattype & self.PERFCOUNTER_TYPE_MASK + if typeonly == 0: + return 'gauge' + if typeonly == self.PERFCOUNTER_LONGRUNAVG: + # this lie matches the DaemonState decoding: only val, no counts + return 'counter' + if typeonly == self.PERFCOUNTER_COUNTER: + return 'counter' + if typeonly == self.PERFCOUNTER_HISTOGRAM: + return 'histogram' + + return '' + + def _perfpath_to_path_labels(self, daemon: str, + path: str) -> Tuple[str, Tuple[str, ...], Tuple[str, ...]]: + if daemon.startswith('rgw.'): + label_name = 'instance_id' + daemon = daemon[len('rgw.'):] + else: + label_name = 'ceph_daemon' + + label_names = (label_name,) # type: Tuple[str, ...] + labels = (daemon,) # type: Tuple[str, ...] + + if daemon.startswith('rbd-mirror.'): + match = re.match( + r'^rbd_mirror_image_([^/]+)/(?:(?:([^/]+)/)?)(.*)\.(replay(?:_bytes|_latency)?)$', + path + ) + if match: + path = 'rbd_mirror_image_' + match.group(4) + pool = match.group(1) + namespace = match.group(2) or '' + image = match.group(3) + label_names += ('pool', 'namespace', 'image') + labels += (pool, namespace, image) + + return path, label_names, labels, + + def _perfvalue_to_value(self, stattype: int, value: Union[int, float]) -> Union[float, int]: + if stattype & self.PERFCOUNTER_TIME: + # Convert from ns to seconds + return value / 1000000000.0 + else: + return value + + def _unit_to_str(self, unit: int) -> str: + if unit == self.NONE: + return "/s" + elif unit == self.BYTES: + return "B/s" + else: + raise ValueError(f'bad unit "{unit}"') + + @staticmethod + def to_pretty_iec(n: int) -> str: + for bits, suffix in [(60, 'Ei'), (50, 'Pi'), (40, 'Ti'), (30, 'Gi'), + (20, 'Mi'), (10, 'Ki')]: + if n > 10 << bits: + return str(n >> bits) + ' ' + suffix + return str(n) + ' ' + + @staticmethod + def get_pretty_row(elems: Sequence[str], width: int) -> str: + """ + Takes an array of elements and returns a string with those elements + formatted as a table row. Useful for polling modules. + + :param elems: the elements to be printed + :param width: the width of the terminal + """ + n = len(elems) + column_width = int(width / n) + + ret = '|' + for elem in elems: + ret += '{0:>{w}} |'.format(elem, w=column_width - 2) + + return ret + + def get_pretty_header(self, elems: Sequence[str], width: int) -> str: + """ + Like ``get_pretty_row`` but adds dashes, to be used as a table title. + + :param elems: the elements to be printed + :param width: the width of the terminal + """ + n = len(elems) + column_width = int(width / n) + + # dash line + ret = '+' + for i in range(0, n): + ret += '-' * (column_width - 1) + '+' + ret += '\n' + + # title + ret += self.get_pretty_row(elems, width) + ret += '\n' + + # dash line + ret += '+' + for i in range(0, n): + ret += '-' * (column_width - 1) + '+' + ret += '\n' + + return ret + + @API.expose + def get_server(self, hostname: str) -> ServerInfoT: + """ + Called by the plugin to fetch metadata about a particular hostname from + ceph-mgr. + + This is information that ceph-mgr has gleaned from the daemon metadata + reported by daemons running on a particular server. + + :param hostname: a hostname + """ + return cast(ServerInfoT, self._ceph_get_server(hostname)) + + @API.expose + def get_perf_schema(self, + svc_type: str, + svc_name: str) -> Dict[str, + Dict[str, Dict[str, Union[str, int]]]]: + """ + Called by the plugin to fetch perf counter schema info. + svc_name can be nullptr, as can svc_type, in which case + they are wildcards + + :param str svc_type: + :param str svc_name: + :return: list of dicts describing the counters requested + """ + return self._ceph_get_perf_schema(svc_type, svc_name) + + def get_rocksdb_version(self) -> str: + """ + Called by the plugin to fetch the latest RocksDB version number. + + :return: str representing the major, minor, and patch RocksDB version numbers + """ + return self._ceph_get_rocksdb_version() + + @API.expose + def get_counter(self, + svc_type: str, + svc_name: str, + path: str) -> Dict[str, List[Tuple[float, int]]]: + """ + Called by the plugin to fetch the latest performance counter data for a + particular counter on a particular service. + + :param str svc_type: + :param str svc_name: + :param str path: a period-separated concatenation of the subsystem and the + counter name, for example "mds.inodes". + :return: A dict of counter names to their values. each value is a list of + of two-tuples of (timestamp, value). This may be empty if no data is + available. + """ + return self._ceph_get_counter(svc_type, svc_name, path) + + @API.expose + def get_latest_counter(self, + svc_type: str, + svc_name: str, + path: str) -> Dict[str, Union[Tuple[float, int], + Tuple[float, int, int]]]: + """ + Called by the plugin to fetch only the newest performance counter data + point for a particular counter on a particular service. + + :param str svc_type: + :param str svc_name: + :param str path: a period-separated concatenation of the subsystem and the + counter name, for example "mds.inodes". + :return: A list of two-tuples of (timestamp, value) or three-tuple of + (timestamp, value, count) is returned. This may be empty if no + data is available. + """ + return self._ceph_get_latest_counter(svc_type, svc_name, path) + + @API.expose + def list_servers(self) -> List[ServerInfoT]: + """ + Like ``get_server``, but gives information about all servers (i.e. all + unique hostnames that have been mentioned in daemon metadata) + + :return: a list of information about all servers + :rtype: list + """ + return cast(List[ServerInfoT], self._ceph_get_server(None)) + + def get_metadata(self, + svc_type: str, + svc_id: str, + default: Optional[Dict[str, str]] = None) -> Optional[Dict[str, str]]: + """ + Fetch the daemon metadata for a particular service. + + ceph-mgr fetches metadata asynchronously, so are windows of time during + addition/removal of services where the metadata is not available to + modules. ``None`` is returned if no metadata is available. + + :param str svc_type: service type (e.g., 'mds', 'osd', 'mon') + :param str svc_id: service id. convert OSD integer IDs to strings when + calling this + :rtype: dict, or None if no metadata found + """ + metadata = self._ceph_get_metadata(svc_type, svc_id) + if not metadata: + return default + return metadata + + @API.expose + def get_daemon_status(self, svc_type: str, svc_id: str) -> Dict[str, str]: + """ + Fetch the latest status for a particular service daemon. + + This method may return ``None`` if no status information is + available, for example because the daemon hasn't fully started yet. + + :param svc_type: string (e.g., 'rgw') + :param svc_id: string + :return: dict, or None if the service is not found + """ + return self._ceph_get_daemon_status(svc_type, svc_id) + + def check_mon_command(self, cmd_dict: dict, inbuf: Optional[str] = None) -> HandleCommandResult: + """ + Wrapper around :func:`~mgr_module.MgrModule.mon_command`, but raises, + if ``retval != 0``. + """ + + r = HandleCommandResult(*self.mon_command(cmd_dict, inbuf)) + if r.retval: + raise MonCommandFailed(f'{cmd_dict["prefix"]} failed: {r.stderr} retval: {r.retval}') + return r + + def mon_command(self, cmd_dict: dict, inbuf: Optional[str] = None) -> Tuple[int, str, str]: + """ + Helper for modules that do simple, synchronous mon command + execution. + + See send_command for general case. + + :return: status int, out std, err str + """ + + t1 = time.time() + result = CommandResult() + self.send_command(result, "mon", "", json.dumps(cmd_dict), "", inbuf) + r = result.wait() + t2 = time.time() + + self.log.debug("mon_command: '{0}' -> {1} in {2:.3f}s".format( + cmd_dict['prefix'], r[0], t2 - t1 + )) + + return r + + def osd_command(self, cmd_dict: dict, inbuf: Optional[str] = None) -> Tuple[int, str, str]: + """ + Helper for osd command execution. + + See send_command for general case. Also, see osd/OSD.cc for available commands. + + :param dict cmd_dict: expects a prefix and an osd id, i.e.: + cmd_dict = { + 'prefix': 'perf histogram dump', + 'id': '0' + } + :return: status int, out std, err str + """ + t1 = time.time() + result = CommandResult() + self.send_command(result, "osd", cmd_dict['id'], json.dumps(cmd_dict), "", inbuf) + r = result.wait() + t2 = time.time() + + self.log.debug("osd_command: '{0}' -> {1} in {2:.3f}s".format( + cmd_dict['prefix'], r[0], t2 - t1 + )) + + return r + + def tell_command(self, daemon_type: str, daemon_id: str, cmd_dict: dict, inbuf: Optional[str] = None) -> Tuple[int, str, str]: + """ + Helper for `ceph tell` command execution. + + See send_command for general case. + + :param dict cmd_dict: expects a prefix i.e.: + cmd_dict = { + 'prefix': 'heap', + 'heapcmd': 'stats', + } + :return: status int, out std, err str + """ + t1 = time.time() + result = CommandResult() + self.send_command(result, daemon_type, daemon_id, json.dumps(cmd_dict), "", inbuf) + r = result.wait() + t2 = time.time() + + self.log.debug("tell_command on {0}.{1}: '{2}' -> {3} in {4:.5f}s".format( + daemon_type, daemon_id, cmd_dict['prefix'], r[0], t2 - t1 + )) + + return r + + def send_command( + self, + result: CommandResult, + svc_type: str, + svc_id: str, + command: str, + tag: str, + inbuf: Optional[str] = None) -> None: + """ + Called by the plugin to send a command to the mon + cluster. + + :param CommandResult result: an instance of the ``CommandResult`` + class, defined in the same module as MgrModule. This acts as a + completion and stores the output of the command. Use + ``CommandResult.wait()`` if you want to block on completion. + :param str svc_type: + :param str svc_id: + :param str command: a JSON-serialized command. This uses the same + format as the ceph command line, which is a dictionary of command + arguments, with the extra ``prefix`` key containing the command + name itself. Consult MonCommands.h for available commands and + their expected arguments. + :param str tag: used for nonblocking operation: when a command + completes, the ``notify()`` callback on the MgrModule instance is + triggered, with notify_type set to "command", and notify_id set to + the tag of the command. + :param str inbuf: input buffer for sending additional data. + """ + self._ceph_send_command(result, svc_type, svc_id, command, tag, inbuf) + + def tool_exec( + self, + args: List[str], + timeout: int = 10, + stdin: Optional[bytes] = None + ) -> Tuple[int, str, str]: + try: + tool = args.pop(0) + cmd = [ + tool, + '-k', str(self.get_ceph_option('keyring')), + '-n', f'mgr.{self.get_mgr_id()}', + ] + args + self.log.debug('exec: ' + ' '.join(cmd)) + p = subprocess.run( + cmd, + input=stdin, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=timeout, + ) + except subprocess.TimeoutExpired as ex: + self.log.error(ex) + return -errno.ETIMEDOUT, '', str(ex) + if p.returncode: + self.log.error(f'Non-zero return from {cmd}: {p.stderr.decode()}') + return p.returncode, p.stdout.decode(), p.stderr.decode() + + def set_health_checks(self, checks: HealthChecksT) -> None: + """ + Set the module's current map of health checks. Argument is a + dict of check names to info, in this form: + + :: + + { + 'CHECK_FOO': { + 'severity': 'warning', # or 'error' + 'summary': 'summary string', + 'count': 4, # quantify badness + 'detail': [ 'list', 'of', 'detail', 'strings' ], + }, + 'CHECK_BAR': { + 'severity': 'error', + 'summary': 'bars are bad', + 'detail': [ 'too hard' ], + }, + } + + :param list: dict of health check dicts + """ + self._ceph_set_health_checks(checks) + + def _handle_command(self, + inbuf: str, + cmd: Dict[str, Any]) -> Union[HandleCommandResult, + Tuple[int, str, str]]: + if cmd['prefix'] not in CLICommand.COMMANDS: + return self.handle_command(inbuf, cmd) + + return CLICommand.COMMANDS[cmd['prefix']].call(self, cmd, inbuf) + + def handle_command(self, + inbuf: str, + cmd: Dict[str, Any]) -> Union[HandleCommandResult, + Tuple[int, str, str]]: + """ + Called by ceph-mgr to request the plugin to handle one + of the commands that it declared in self.COMMANDS + + Return a status code, an output buffer, and an + output string. The output buffer is for data results, + the output string is for informative text. + + :param inbuf: content of any "-i <file>" supplied to ceph cli + :type inbuf: str + :param cmd: from Ceph's cmdmap_t + :type cmd: dict + + :return: HandleCommandResult or a 3-tuple of (int, str, str) + """ + + # Should never get called if they didn't declare + # any ``COMMANDS`` + raise NotImplementedError() + + def get_mgr_id(self) -> str: + """ + Retrieve the name of the manager daemon where this plugin + is currently being executed (i.e. the active manager). + + :return: str + """ + return self._ceph_get_mgr_id() + + @API.expose + def get_ceph_conf_path(self) -> str: + return self._ceph_get_ceph_conf_path() + + @API.expose + def get_mgr_ip(self) -> str: + if not self._mgr_ips: + ips = self.get("mgr_ips").get('ips', []) + if not ips: + return socket.gethostname() + self._mgr_ips = ips[0] + assert self._mgr_ips is not None + return self._mgr_ips + + @API.expose + def get_hostname(self) -> str: + return socket.gethostname() + + @API.expose + def get_ceph_option(self, key: str) -> OptionValue: + return self._ceph_get_option(key) + + @API.expose + def get_foreign_ceph_option(self, entity: str, key: str) -> OptionValue: + return self._ceph_get_foreign_option(entity, key) + + def _validate_module_option(self, key: str) -> None: + """ + Helper: don't allow get/set config callers to + access config options that they didn't declare + in their schema. + """ + if key not in [o['name'] for o in self.MODULE_OPTIONS]: + raise RuntimeError("Config option '{0}' is not in {1}.MODULE_OPTIONS". + format(key, self.__class__.__name__)) + + def _get_module_option(self, + key: str, + default: OptionValue, + localized_prefix: str = "") -> OptionValue: + r = self._ceph_get_module_option(self.module_name, key, + localized_prefix) + if r is None: + return self.MODULE_OPTION_DEFAULTS.get(key, default) + else: + return r + + def get_module_option(self, key: str, default: OptionValue = None) -> OptionValue: + """ + Retrieve the value of a persistent configuration setting + """ + self._validate_module_option(key) + return self._get_module_option(key, default) + + def get_module_option_ex(self, module: str, + key: str, + default: OptionValue = None) -> OptionValue: + """ + Retrieve the value of a persistent configuration setting + for the specified module. + + :param module: The name of the module, e.g. 'dashboard' + or 'telemetry'. + :param key: The configuration key, e.g. 'server_addr'. + :param default: The default value to use when the + returned value is ``None``. Defaults to ``None``. + """ + if module == self.module_name: + self._validate_module_option(key) + r = self._ceph_get_module_option(module, key) + return default if r is None else r + + @API.expose + def get_store_prefix(self, key_prefix: str) -> Dict[str, str]: + """ + Retrieve a dict of KV store keys to values, where the keys + have the given prefix + + :param str key_prefix: + :return: str + """ + return self._ceph_get_store_prefix(key_prefix) + + def _set_localized(self, + key: str, + val: Optional[str], + setter: Callable[[str, Optional[str]], None]) -> None: + return setter(_get_localized_key(self.get_mgr_id(), key), val) + + def get_localized_module_option(self, key: str, default: OptionValue = None) -> OptionValue: + """ + Retrieve localized configuration for this ceph-mgr instance + """ + self._validate_module_option(key) + return self._get_module_option(key, default, self.get_mgr_id()) + + def _set_module_option(self, key: str, val: Any) -> None: + return self._ceph_set_module_option(self.module_name, key, + None if val is None else str(val)) + + def set_module_option(self, key: str, val: Any) -> None: + """ + Set the value of a persistent configuration setting + + :param str key: + :type val: str | None + :raises ValueError: if `val` cannot be parsed or it is out of the specified range + """ + self._validate_module_option(key) + return self._set_module_option(key, val) + + def set_module_option_ex(self, module: str, key: str, val: OptionValue) -> None: + """ + Set the value of a persistent configuration setting + for the specified module. + + :param str module: + :param str key: + :param str val: + """ + if module == self.module_name: + self._validate_module_option(key) + return self._ceph_set_module_option(module, key, str(val)) + + @API.perm('w') + @API.expose + def set_localized_module_option(self, key: str, val: Optional[str]) -> None: + """ + Set localized configuration for this ceph-mgr instance + :param str key: + :param str val: + :return: str + """ + self._validate_module_option(key) + return self._set_localized(key, val, self._set_module_option) + + @API.perm('w') + @API.expose + def set_store(self, key: str, val: Optional[str]) -> None: + """ + Set a value in this module's persistent key value store. + If val is None, remove key from store + """ + self._ceph_set_store(key, val) + + @API.expose + def get_store(self, key: str, default: Optional[str] = None) -> Optional[str]: + """ + Get a value from this module's persistent key value store + """ + r = self._ceph_get_store(key) + if r is None: + return default + else: + return r + + @API.expose + def get_localized_store(self, key: str, default: Optional[str] = None) -> Optional[str]: + r = self._ceph_get_store(_get_localized_key(self.get_mgr_id(), key)) + if r is None: + r = self._ceph_get_store(key) + if r is None: + r = default + return r + + @API.perm('w') + @API.expose + def set_localized_store(self, key: str, val: Optional[str]) -> None: + return self._set_localized(key, val, self.set_store) + + def self_test(self) -> Optional[str]: + """ + Run a self-test on the module. Override this function and implement + a best as possible self-test for (automated) testing of the module + + Indicate any failures by raising an exception. This does not have + to be pretty, it's mainly for picking up regressions during + development, rather than use in the field. + + :return: None, or an advisory string for developer interest, such + as a json dump of some state. + """ + pass + + def get_osdmap(self) -> OSDMap: + """ + Get a handle to an OSDMap. If epoch==0, get a handle for the latest + OSDMap. + :return: OSDMap + """ + return cast(OSDMap, self._ceph_get_osdmap()) + + @API.expose + def get_latest(self, daemon_type: str, daemon_name: str, counter: str) -> int: + data = self.get_latest_counter( + daemon_type, daemon_name, counter)[counter] + if data: + return data[1] + else: + return 0 + + @API.expose + def get_latest_avg(self, daemon_type: str, daemon_name: str, counter: str) -> Tuple[int, int]: + data = self.get_latest_counter( + daemon_type, daemon_name, counter)[counter] + if data: + # https://github.com/python/mypy/issues/1178 + _, value, count = cast(Tuple[float, int, int], data) + return value, count + else: + return 0, 0 + + @API.expose + @profile_method() + def get_unlabeled_perf_counters(self, prio_limit: int = PRIO_USEFUL, + services: Sequence[str] = ("mds", "mon", "osd", + "rbd-mirror", "rgw", + "tcmu-runner")) -> Dict[str, dict]: + """ + Return the perf counters currently known to this ceph-mgr + instance, filtered by priority equal to or greater than `prio_limit`. + + The result is a map of string to dict, associating services + (like "osd.123") with their counters. The counter + dict for each service maps counter paths to a counter + info structure, which is the information from + the schema, plus an additional "value" member with the latest + value. + """ + + result = defaultdict(dict) # type: Dict[str, dict] + + for server in self.list_servers(): + for service in cast(List[ServiceInfoT], server['services']): + if service['type'] not in services: + continue + + schemas = self.get_perf_schema(service['type'], service['id']) + if not schemas: + self.log.warning("No perf counter schema for {0}.{1}".format( + service['type'], service['id'] + )) + continue + + # Value is returned in a potentially-multi-service format, + # get just the service we're asking about + svc_full_name = "{0}.{1}".format( + service['type'], service['id']) + schema = schemas[svc_full_name] + + # Populate latest values + for counter_path, counter_schema in schema.items(): + # self.log.debug("{0}: {1}".format( + # counter_path, json.dumps(counter_schema) + # )) + priority = counter_schema['priority'] + assert isinstance(priority, int) + if priority < prio_limit: + continue + + tp = counter_schema['type'] + assert isinstance(tp, int) + counter_info = dict(counter_schema) + # Also populate count for the long running avgs + if tp & self.PERFCOUNTER_LONGRUNAVG: + v, c = self.get_latest_avg( + service['type'], + service['id'], + counter_path + ) + counter_info['value'], counter_info['count'] = v, c + result[svc_full_name][counter_path] = counter_info + else: + counter_info['value'] = self.get_latest( + service['type'], + service['id'], + counter_path + ) + + result[svc_full_name][counter_path] = counter_info + + self.log.debug("returning {0} counter".format(len(result))) + + return result + + @API.expose + def set_uri(self, uri: str) -> None: + """ + If the module exposes a service, then call this to publish the + address once it is available. + + :return: a string + """ + return self._ceph_set_uri(uri) + + @API.perm('w') + @API.expose + def set_device_wear_level(self, devid: str, wear_level: float) -> None: + return self._ceph_set_device_wear_level(devid, wear_level) + + @API.expose + def have_mon_connection(self) -> bool: + """ + Check whether this ceph-mgr daemon has an open connection + to a monitor. If it doesn't, then it's likely that the + information we have about the cluster is out of date, + and/or the monitor cluster is down. + """ + + return self._ceph_have_mon_connection() + + def update_progress_event(self, + evid: str, + desc: str, + progress: float, + add_to_ceph_s: bool) -> None: + return self._ceph_update_progress_event(evid, desc, progress, add_to_ceph_s) + + @API.perm('w') + @API.expose + def complete_progress_event(self, evid: str) -> None: + return self._ceph_complete_progress_event(evid) + + @API.perm('w') + @API.expose + def clear_all_progress_events(self) -> None: + return self._ceph_clear_all_progress_events() + + @property + def rados(self) -> rados.Rados: + """ + A librados instance to be shared by any classes within + this mgr module that want one. + """ + if self._rados: + return self._rados + + ctx_capsule = self.get_context() + self._rados = rados.Rados(context=ctx_capsule) + self._rados.connect() + self._ceph_register_client(None, self._rados.get_addrs(), False) + return self._rados + + @staticmethod + def can_run() -> Tuple[bool, str]: + """ + Implement this function to report whether the module's dependencies + are met. For example, if the module needs to import a particular + dependency to work, then use a try/except around the import at + file scope, and then report here if the import failed. + + This will be called in a blocking way from the C++ code, so do not + do any I/O that could block in this function. + + :return a 2-tuple consisting of a boolean and explanatory string + """ + + return True, "" + + @API.expose + def remote(self, module_name: str, method_name: str, *args: Any, **kwargs: Any) -> Any: + """ + Invoke a method on another module. All arguments, and the return + value from the other module must be serializable. + + Limitation: Do not import any modules within the called method. + Otherwise you will get an error in Python 2:: + + RuntimeError('cannot unmarshal code objects in restricted execution mode',) + + + + :param module_name: Name of other module. If module isn't loaded, + an ImportError exception is raised. + :param method_name: Method name. If it does not exist, a NameError + exception is raised. + :param args: Argument tuple + :param kwargs: Keyword argument dict + :raises RuntimeError: **Any** error raised within the method is converted to a RuntimeError + :raises ImportError: No such module + """ + return self._ceph_dispatch_remote(module_name, method_name, + args, kwargs) + + def add_osd_perf_query(self, query: Dict[str, Any]) -> Optional[int]: + """ + Register an OSD perf query. Argument is a + dict of the query parameters, in this form: + + :: + + { + 'key_descriptor': [ + {'type': subkey_type, 'regex': regex_pattern}, + ... + ], + 'performance_counter_descriptors': [ + list, of, descriptor, types + ], + 'limit': {'order_by': performance_counter_type, 'max_count': n}, + } + + Valid subkey types: + 'client_id', 'client_address', 'pool_id', 'namespace', 'osd_id', + 'pg_id', 'object_name', 'snap_id' + Valid performance counter types: + 'ops', 'write_ops', 'read_ops', 'bytes', 'write_bytes', 'read_bytes', + 'latency', 'write_latency', 'read_latency' + + :param object query: query + :rtype: int (query id) + """ + return self._ceph_add_osd_perf_query(query) + + @API.perm('w') + @API.expose + def remove_osd_perf_query(self, query_id: int) -> None: + """ + Unregister an OSD perf query. + + :param int query_id: query ID + """ + return self._ceph_remove_osd_perf_query(query_id) + + @API.expose + def get_osd_perf_counters(self, query_id: int) -> Optional[Dict[str, List[PerfCounterT]]]: + """ + Get stats collected for an OSD perf query. + + :param int query_id: query ID + """ + return self._ceph_get_osd_perf_counters(query_id) + + def add_mds_perf_query(self, query: Dict[str, Any]) -> Optional[int]: + """ + Register an MDS perf query. Argument is a + dict of the query parameters, in this form: + + :: + + { + 'key_descriptor': [ + {'type': subkey_type, 'regex': regex_pattern}, + ... + ], + 'performance_counter_descriptors': [ + list, of, descriptor, types + ], + } + + NOTE: 'limit' and 'order_by' are not supported (yet). + + Valid subkey types: + 'mds_rank', 'client_id' + Valid performance counter types: + 'cap_hit_metric' + + :param object query: query + :rtype: int (query id) + """ + return self._ceph_add_mds_perf_query(query) + + @API.perm('w') + @API.expose + def remove_mds_perf_query(self, query_id: int) -> None: + """ + Unregister an MDS perf query. + + :param int query_id: query ID + """ + return self._ceph_remove_mds_perf_query(query_id) + + @API.expose + + def reregister_mds_perf_queries(self) -> None: + """ + Re-register MDS perf queries. + """ + return self._ceph_reregister_mds_perf_queries() + + def get_mds_perf_counters(self, query_id: int) -> Optional[Dict[str, List[PerfCounterT]]]: + """ + Get stats collected for an MDS perf query. + + :param int query_id: query ID + """ + return self._ceph_get_mds_perf_counters(query_id) + + def get_daemon_health_metrics(self) -> Dict[str, List[Dict[str, Any]]]: + """ + Get the list of health metrics per daemon. This includes SLOW_OPS health metrics + in MON and OSD daemons, and PENDING_CREATING_PGS health metrics for OSDs. + """ + return self._ceph_get_daemon_health_metrics() + + def is_authorized(self, arguments: Dict[str, str]) -> bool: + """ + Verifies that the current session caps permit executing the py service + or current module with the provided arguments. This provides a generic + way to allow modules to restrict by more fine-grained controls (e.g. + pools). + + :param arguments: dict of key/value arguments to test + """ + return self._ceph_is_authorized(arguments) + + @API.expose + def send_rgwadmin_command(self, args: List[str], + stdout_as_json: bool = True) -> Tuple[int, Union[str, dict], str]: + try: + cmd = [ + 'radosgw-admin', + '-c', str(self.get_ceph_conf_path()), + '-k', str(self.get_ceph_option('keyring')), + '-n', f'mgr.{self.get_mgr_id()}', + ] + args + self.log.debug('Executing %s', str(cmd)) + result = subprocess.run( # pylint: disable=subprocess-run-check + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=10, + ) + stdout = result.stdout.decode('utf-8') + stderr = result.stderr.decode('utf-8') + if stdout and stdout_as_json: + stdout = json.loads(stdout) + if result.returncode: + self.log.debug('Error %s executing %s: %s', result.returncode, str(cmd), stderr) + return result.returncode, stdout, stderr + except subprocess.CalledProcessError as ex: + self.log.exception('Error executing radosgw-admin %s: %s', str(ex.cmd), str(ex.output)) + raise + except subprocess.TimeoutExpired as ex: + self.log.error('Timeout (10s) executing radosgw-admin %s', str(ex.cmd)) + raise |