Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1504 insertions, 0 deletions

diff --git a/src/pybind/mgr/mgr_module.py b/src/pybind/mgr/mgr_module.py
new file mode 100644
index 00000000..79cbc950
--- /dev/null
+++ b/src/pybind/mgr/mgr_module.py
@@ -0,0 +1,1504 @@
+import ceph_module  # noqa
+
+try:
+    from typing import Set, Tuple, Iterator, Any
+except ImportError:
+    # just for type checking
+    pass
+import errno
+import logging
+import json
+import six
+import threading
+from collections import defaultdict, namedtuple
+import rados
+import re
+import time
+
+ERROR_MSG_EMPTY_INPUT_FILE = 'Empty content: please add a password/secret to the file.'
+ERROR_MSG_NO_INPUT_FILE = 'Please specify the file containing the password/secret with "-i" option.'
+# 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",
+]
+
+
+class CPlusPlusHandler(logging.Handler):
+    def __init__(self, module_inst):
+        super(CPlusPlusHandler, self).__init__()
+        self._module = module_inst
+
+    def emit(self, record):
+        if record.levelno <= logging.DEBUG:
+            ceph_level = 20
+        elif record.levelno <= logging.INFO:
+            ceph_level = 4
+        elif record.levelno <= logging.WARNING:
+            ceph_level = 1
+        else:
+            ceph_level = 0
+
+        self._module._ceph_log(ceph_level, self.format(record))
+
+
+def configure_logger(module_inst, module_name):
+    """
+    Create and configure the logger with the specified module.
+
+    A handler will be added to the root logger which will redirect
+    the messages from all loggers (incl. 3rd party libraries) to the
+    Ceph log.
+
+    :param module_inst: The module instance.
+    :type module_inst: instance
+    :param module_name: The module name.
+    :type module_name: str
+    :return: Return the logger with the specified name.
+    """
+    logger = logging.getLogger(module_name)
+    # Don't filter any logs at the python level, leave it to C++.
+    # FIXME: We should learn the log level from C++ land, and then
+    #        avoid calling the C++ level log when we know a message
+    #        is of an insufficient level to be ultimately output.
+    logger.setLevel(logging.DEBUG)  # Don't use NOTSET
+
+    root_logger = logging.getLogger()
+    # Add handler to the root logger, thus this module and all
+    # 3rd party libraries will log their messages to the Ceph log.
+    root_logger.addHandler(CPlusPlusHandler(module_inst))
+    # Set the log level to ``ERROR`` to ensure that we only get
+    # those message from 3rd party libraries (only effective if
+    # they use the default log level ``NOTSET``).
+    # Check https://docs.python.org/3/library/logging.html#logging.Logger.setLevel
+    # for more information about how the effective log level is
+    # determined.
+    root_logger.setLevel(logging.ERROR)
+
+    return logger
+
+
+def unconfigure_logger(module_name=None):
+    """
+    :param module_name: The module name. Defaults to ``None``.
+    :type module_name: str or None
+    """
+    logger = logging.getLogger(module_name)
+    rm_handlers = [
+        h for h in logger.handlers if isinstance(h, CPlusPlusHandler)]
+    for h in rm_handlers:
+        logger.removeHandler(h)
+
+
+class CommandResult(object):
+    """
+    Use with MgrModule.send_command
+    """
+
+    def __init__(self, tag=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, outb, outs):
+        self.r = r
+        self.outb = outb
+        self.outs = outs
+        self.ev.set()
+
+    def wait(self):
+        self.ev.wait()
+        return self.r, self.outb, self.outs
+
+
+class HandleCommandResult(namedtuple('HandleCommandResult', ['retval', 'stdout', 'stderr'])):
+    def __new__(cls, retval=0, stdout="", stderr=""):
+        """
+        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
+
+        :param retval: return code. E.g. 0 or -errno.EINVAL
+        :type retval: int
+        :param stdout: data of this result.
+        :type stdout: str
+        :param stderr: Typically used for error messages.
+        :type stderr: str
+        """
+        return super(HandleCommandResult, cls).__new__(cls, retval, stdout, stderr)
+
+
+class MonCommandFailed(RuntimeError): pass
+
+
+class OSDMap(ceph_module.BasePyOSDMap):
+    def get_epoch(self):
+        return self._get_epoch()
+
+    def get_crush_version(self):
+        return self._get_crush_version()
+
+    def dump(self):
+        return self._dump()
+
+    def get_pools(self):
+        # FIXME: efficient implementation
+        d = self._dump()
+        return dict([(p['pool'], p) for p in d['pools']])
+
+    def get_pools_by_name(self):
+        # FIXME: efficient implementation
+        d = self._dump()
+        return dict([(p['pool_name'], p) for p in d['pools']])
+
+    def new_incremental(self):
+        return self._new_incremental()
+
+    def apply_incremental(self, inc):
+        return self._apply_incremental(inc)
+
+    def get_crush(self):
+        return self._get_crush()
+
+    def get_pools_by_take(self, take):
+        return self._get_pools_by_take(take).get('pools', [])
+
+    def calc_pg_upmaps(self, inc,
+                       max_deviation=.01, max_iterations=10, pools=None):
+        if pools is None:
+            pools = []
+        return self._calc_pg_upmaps(
+            inc,
+            max_deviation, max_iterations, pools)
+
+    def map_pool_pgs_up(self, poolid):
+        return self._map_pool_pgs_up(poolid)
+
+    def pg_to_up_acting_osds(self, pool_id, ps):
+        return self._pg_to_up_acting_osds(pool_id, ps)
+
+    def pool_raw_used_rate(self, pool_id):
+        return self._pool_raw_used_rate(pool_id)
+
+    def get_ec_profile(self, name):
+        # FIXME: efficient implementation
+        d = self._dump()
+        return d['erasure_code_profiles'].get(name, None)
+
+
+class OSDMapIncremental(ceph_module.BasePyOSDMapIncremental):
+    def get_epoch(self):
+        return self._get_epoch()
+
+    def dump(self):
+        return self._dump()
+
+    def set_osd_reweights(self, weightmap):
+        """
+        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):
+        """
+        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):
+        return self._dump()
+
+    def get_item_weight(self, item):
+        return self._get_item_weight(item)
+
+    def get_item_name(self, item):
+        return self._get_item_name(item)
+
+    def find_takes(self):
+        return self._find_takes().get('takes', [])
+
+    def get_take_weight_osd_map(self, root):
+        uglymap = self._get_take_weight_osd_map(root)
+        return {int(k): v for k, v in six.iteritems(uglymap.get('weights', {}))}
+
+    @staticmethod
+    def have_default_choose_args(dump):
+        return CRUSHMap.DEFAULT_CHOOSE_ARGS in dump.get('choose_args', {})
+
+    @staticmethod
+    def get_default_choose_args(dump):
+        return dump.get('choose_args').get(CRUSHMap.DEFAULT_CHOOSE_ARGS, [])
+
+    def get_rule(self, rule_name):
+        # 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):
+        for rule in self.dump()['rules']:
+            if rule['rule_id'] == rule_id:
+                return rule
+
+        return None
+
+    def get_rule_root(self, rule_name):
+        rule = self.get_rule(rule_name)
+        if rule is None:
+            return None
+
+        try:
+            first_take = [s for s in rule['steps'] if s['op'] == 'take'][0]
+        except IndexError:
+            self.log.warn("CRUSH rule '{0}' has no 'take' step".format(
+                rule_name))
+            return None
+        else:
+            return first_take['item']
+
+    def get_osds_under(self, root_id):
+        # 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):
+            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):
+        result = defaultdict(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)
+
+
+class CLICommand(object):
+    COMMANDS = {}
+
+    def __init__(self, prefix, args="", desc="", perm="rw"):
+        self.prefix = prefix
+        self.args = args
+        self.args_dict = {}
+        self.desc = desc
+        self.perm = perm
+        self.func = None
+        self._parse_args()
+
+    def _parse_args(self):
+        if not self.args:
+            return
+        args = self.args.split(" ")
+        for arg in args:
+            arg_desc = arg.strip().split(",")
+            arg_d = {}
+            for kv in arg_desc:
+                k, v = kv.split("=")
+                if k != "name":
+                    arg_d[k] = v
+                else:
+                    self.args_dict[v] = arg_d
+
+    def __call__(self, func):
+        self.func = func
+        self.COMMANDS[self.prefix] = self
+        return self.func
+
+    def call(self, mgr, cmd_dict, inbuf):
+        kwargs = {}
+        for a, d in self.args_dict.items():
+            if 'req' in d and d['req'] == "false" and a not in cmd_dict:
+                continue
+            kwargs[a.replace("-", "_")] = cmd_dict[a]
+        if inbuf:
+            kwargs['inbuf'] = inbuf
+        return self.func(mgr, **kwargs)
+
+    @classmethod
+    def dump_cmd_list(cls):
+        return [{
+            'cmd': '{} {}'.format(cmd.prefix, cmd.args),
+            'desc': cmd.desc,
+            'perm': cmd.perm
+        } for _, cmd in cls.COMMANDS.items()]
+
+
+def CLIReadCommand(prefix, args="", desc=""):
+    return CLICommand(prefix, args, desc, "r")
+
+
+def CLIWriteCommand(prefix, args="", desc=""):
+    return CLICommand(prefix, args, desc, "w")
+
+
+def CLICheckNonemptyFileInput(func):
+    def check(*args, **kwargs):
+        if not 'inbuf' in kwargs:
+            return -errno.EINVAL, '', ERROR_MSG_NO_INPUT_FILE
+        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']:
+            return -errno.EINVAL, '', ERROR_MSG_EMPTY_INPUT_FILE
+        return func(*args, **kwargs)
+    return check
+
+
+def _get_localized_key(prefix, key):
+    return '{}/{}'.format(prefix, key)
+
+
+class Option(dict):
+    """
+    Helper class to declare options for MODULE_OPTIONS list.
+
+    Caveat: it uses argument names matching Python keywords (type, min, max),
+    so any further processing should happen in a separate method.
+
+    TODO: type validation.
+    """
+
+    def __init__(
+            self, name,
+            default=None,
+            type='str',
+            desc=None, longdesc=None,
+            min=None, max=None,
+            enum_allowed=None,
+            see_also=None,
+            tags=None,
+            runtime=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:
+    >>> Command(prefix="example",
+    ...         args="name=arg,type=CephInt",
+    ...         perm='w',
+    ...         desc="Blah")
+    {'poll': False, 'cmd': 'example name=arg,type=CephInt', 'perm': 'w', 'desc': 'Blah'}
+    """
+
+    def __init__(
+            self,
+            prefix,
+            args=None,
+            perm="rw",
+            desc=None,
+            poll=False,
+            handler=None
+    ):
+        super(Command, self).__init__(
+            cmd=prefix + (' ' + args if args else ''),
+            perm=perm,
+            desc=desc,
+            poll=poll)
+        self.prefix = prefix
+        self.args = args
+        self.handler = handler
+
+    def register(self, instance=False):
+        """
+        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.
+        """
+        return CLICommand(
+            prefix=self.prefix,
+            args=self.args,
+            desc=self['desc'],
+            perm=self['perm']
+        )(
+            func=lambda mgr, *args, **kwargs: HandleCommandResult(*self.handler(
+                *((instance or mgr,) + args), **kwargs))
+        )
+
+
+class MgrStandbyModule(ceph_module.BaseMgrStandbyModule):
+    """
+    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 = []
+    MODULE_OPTION_DEFAULTS = {}
+
+    def __init__(self, module_name, capsule):
+        super(MgrStandbyModule, self).__init__(capsule)
+        self.module_name = module_name
+        self._logger = configure_logger(self, 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'])
+
+    def __del__(self):
+        unconfigure_logger()
+
+    @property
+    def log(self):
+        return self._logger
+
+    def serve(self):
+        """
+        The serve method is mandatory for standby modules.
+        :return:
+        """
+        raise NotImplementedError()
+
+    def get_mgr_id(self):
+        return self._ceph_get_mgr_id()
+
+    def get_module_option(self, key, default=None):
+        """
+        Retrieve the value of a persistent configuration setting
+
+        :param str key:
+        :param default: the default value of the config if it is not found
+        :return: str
+        """
+        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):
+        return self._ceph_get_option(key)
+
+    def get_store(self, key):
+        """
+        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_active_uri(self):
+        return self._ceph_get_active_uri()
+
+    def get_localized_module_option(self, key, default=None):
+        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
+
+
+class MgrModule(ceph_module.BaseMgrModule):
+    COMMANDS = []
+    MODULE_OPTIONS = []
+    MODULE_OPTION_DEFAULTS = {}
+
+    # 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
+    CLUSTER_LOG_PRIO_DEBUG = 0
+    CLUSTER_LOG_PRIO_INFO = 1
+    CLUSTER_LOG_PRIO_SEC = 2
+    CLUSTER_LOG_PRIO_WARN = 3
+    CLUSTER_LOG_PRIO_ERROR = 4
+
+    def __init__(self, module_name, py_modules_ptr, this_ptr):
+        self.module_name = module_name
+
+        # If we're taking over from a standby module, let's make sure
+        # its logger was unconfigured before we hook ours up
+        unconfigure_logger()
+        self._logger = configure_logger(self, module_name)
+
+        super(MgrModule, self).__init__(py_modules_ptr, this_ptr)
+
+        self._version = self._ceph_get_version()
+
+        self._perf_schema_cache = None
+
+        # Keep a librados instance for those that need it.
+        self._rados = None
+
+        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'])
+
+    def __del__(self):
+        unconfigure_logger()
+
+    @classmethod
+    def _register_commands(cls):
+        cls.COMMANDS.extend(CLICommand.dump_cmd_list())
+
+    @property
+    def log(self):
+        return self._logger
+
+    def cluster_log(self, channel, priority, message):
+        """
+        :param channel: The log channel. This can be 'cluster', 'audit', ...
+        :type channel: str
+        :param priority: The log message priority. This can be
+                         CLUSTER_LOG_PRIO_DEBUG, CLUSTER_LOG_PRIO_INFO,
+                         CLUSTER_LOG_PRIO_SEC, CLUSTER_LOG_PRIO_WARN or
+                         CLUSTER_LOG_PRIO_ERROR.
+        :type priority: int
+        :param message: The message to log.
+        :type message: str
+        """
+        self._ceph_cluster_log(channel, priority, message)
+
+    @property
+    def version(self):
+        return self._version
+
+    @property
+    def release_name(self):
+        """
+        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()
+
+    def get_context(self):
+        """
+        :return: a Python capsule containing a C++ CephContext pointer
+        """
+        return self._ceph_get_context()
+
+    def notify(self, notify_type, notify_id):
+        """
+        Called by the ceph-mgr service to notify the Python plugin
+        that new state is available.
+
+        :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):
+        """
+        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):
+        """
+        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):
+        """
+        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:
+            self._rados.shutdown()
+
+    def get(self, data_name):
+        """
+        Called by the plugin to fetch named cluster-wide objects from ceph-mgr.
+
+        :param str data_name: Valid things to fetch are osd_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.
+
+        Note:
+            All these structures have their own JSON representations: experiment
+            or look at the C++ ``dump()`` methods to learn about them.
+        """
+        return self._ceph_get(data_name)
+
+    def _stattype_to_str(self, stattype):
+
+        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, path):
+        label_names = ("ceph_daemon",)
+        labels = (daemon,)
+
+        if daemon.startswith('rbd-mirror.'):
+            match = re.match(
+                r'^rbd_mirror_([^/]+)/(?:(?:([^/]+)/)?)(.*)\.(replay(?:_bytes|_latency)?)$',
+                path
+            )
+            if match:
+                path = 'rbd_mirror_' + 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, value):
+        if stattype & self.PERFCOUNTER_TIME:
+            # Convert from ns to seconds
+            return value / 1000000000.0
+        else:
+            return value
+
+    def _unit_to_str(self, unit):
+        if unit == self.NONE:
+            return "/s"
+        elif unit == self.BYTES:
+            return "B/s"
+
+    @staticmethod
+    def to_pretty_iec(n):
+        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, width):
+        """
+        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, width):
+        """
+        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
+
+    def get_server(self, hostname):
+        """
+        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 self._ceph_get_server(hostname)
+
+    def get_perf_schema(self, svc_type, svc_name):
+        """
+        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_counter(self, svc_type, svc_name, path):
+        """
+        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 list of two-tuples of (timestamp, value) is returned.  This may be
+            empty if no data is available.
+        """
+        return self._ceph_get_counter(svc_type, svc_name, path)
+
+    def get_latest_counter(self, svc_type, svc_name, path):
+        """
+        Called by the plugin to fetch only the newest performance counter data
+        pointfor 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) is returned.  This may be
+            empty if no data is available.
+        """
+        return self._ceph_get_latest_counter(svc_type, svc_name, path)
+
+    def list_servers(self):
+        """
+        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 self._ceph_get_server(None)
+
+    def get_metadata(self, svc_type, svc_id, default=None):
+        """
+        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 metadata is None:
+            return default
+        return metadata
+
+    def get_daemon_status(self, svc_type, svc_id):
+        """
+        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, inbuf=None):
+        """
+        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(
+                '{} failed: {} retval: {}'.format(cmd_dict["prefix"], r.stderr, r.retval)
+            )
+        return r
+
+    def mon_command(self, cmd_dict, inbuf=None):
+        """
+        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 send_command(
+            self,
+            result,
+            svc_type,
+            svc_id,
+            command,
+            tag,
+            inbuf=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 set_health_checks(self, checks):
+        """
+        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',
+               '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, cmd):
+        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, cmd):
+        """
+        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):
+        """
+        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()
+
+    def get_ceph_option(self, key):
+        return self._ceph_get_option(key)
+
+    def _validate_module_option(self, key):
+        """
+        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, default, localized_prefix=""):
+        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, default=None):
+        """
+        Retrieve the value of a persistent configuration setting
+
+        :param str key:
+        :param str default:
+        :return: str
+        """
+        self._validate_module_option(key)
+        return self._get_module_option(key, default)
+
+    def get_module_option_ex(self, module, key, default=None):
+        """
+        Retrieve the value of a persistent configuration setting
+        for the specified module.
+
+        :param str module: The name of the module, e.g. 'dashboard'
+            or 'telemetry'.
+        :param str key: The configuration key, e.g. 'server_addr'.
+        :param str,None default: The default value to use when the
+            returned value is ``None``. Defaults to ``None``.
+        :return: str,int,bool,float,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
+
+    def get_store_prefix(self, key_prefix):
+        """
+        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, val, setter):
+        return setter(_get_localized_key(self.get_mgr_id(), key), val)
+
+    def get_localized_module_option(self, key, default=None):
+        """
+        Retrieve localized configuration for this ceph-mgr instance
+        :param str key:
+        :param str default:
+        :return: str
+        """
+        self._validate_module_option(key)
+        return self._get_module_option(key, default, self.get_mgr_id())
+
+    def _set_module_option(self, key, val):
+        return self._ceph_set_module_option(self.module_name, key,
+                                            None if val is None else str(val))
+
+    def set_module_option(self, key, val):
+        """
+        Set the value of a persistent configuration setting
+
+        :param str key:
+        :type val: str | None
+        """
+        self._validate_module_option(key)
+        return self._set_module_option(key, val)
+
+    def set_module_option_ex(self, module, key, val):
+        """
+        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))
+
+    def set_localized_module_option(self, key, val):
+        """
+        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)
+
+    def set_store(self, key, val):
+        """
+        Set a value in this module's persistent key value store.
+        If val is None, remove key from store
+
+        :param str key:
+        :param str val:
+        """
+        self._ceph_set_store(key, val)
+
+    def get_store(self, key, default=None):
+        """
+        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
+
+    def get_localized_store(self, key, default=None):
+        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 set_localized_store(self, key, val):
+        return self._set_localized(key, val, self.set_store)
+
+    def self_test(self):
+        """
+        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):
+        """
+        Get a handle to an OSDMap.  If epoch==0, get a handle for the latest
+        OSDMap.
+        :return: OSDMap
+        """
+        return self._ceph_get_osdmap()
+
+    def get_latest(self, daemon_type, daemon_name, counter):
+        data = self.get_latest_counter(
+            daemon_type, daemon_name, counter)[counter]
+        if data:
+            return data[1]
+        else:
+            return 0
+
+    def get_latest_avg(self, daemon_type, daemon_name, counter):
+        data = self.get_latest_counter(
+            daemon_type, daemon_name, counter)[counter]
+        if data:
+            return data[1], data[2]
+        else:
+            return 0, 0
+
+    def get_all_perf_counters(self, prio_limit=PRIO_USEFUL,
+                              services=("mds", "mon", "osd",
+                                        "rbd-mirror", "rgw", "tcmu-runner")):
+        """
+        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)
+
+        for server in self.list_servers():
+            for service in server['services']:
+                if service['type'] not in services:
+                    continue
+
+                schema = self.get_perf_schema(service['type'], service['id'])
+                if not schema:
+                    self.log.warn("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 = schema[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)
+                    # ))
+                    if counter_schema['priority'] < prio_limit:
+                        continue
+
+                    counter_info = dict(counter_schema)
+
+                    # Also populate count for the long running avgs
+                    if counter_schema['type'] & 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
+
+    def set_uri(self, uri):
+        """
+        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)
+
+    def have_mon_connection(self):
+        """
+        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, desc, progress):
+        return self._ceph_update_progress_event(str(evid), str(desc), float(progress))
+
+    def complete_progress_event(self, evid):
+        return self._ceph_complete_progress_event(str(evid))
+
+    def clear_all_progress_events(self):
+        return self._ceph_clear_all_progress_events()
+
+    @property
+    def rados(self):
+        """
+        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()
+
+        return self._rados
+
+    @staticmethod
+    def can_run():
+        """
+        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, ""
+
+    def remote(self, module_name, method_name, *args, **kwargs):
+        """
+        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):
+        """
+        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)
+
+    def remove_osd_perf_query(self, query_id):
+        """
+        Unregister an OSD perf query.
+
+        :param int query_id: query ID
+        """
+        return self._ceph_remove_osd_perf_query(query_id)
+
+    def get_osd_perf_counters(self, query_id):
+        """
+        Get stats collected for an OSD perf query.
+
+        :param int query_id: query ID
+        """
+        return self._ceph_get_osd_perf_counters(query_id)
+
+    def is_authorized(self, arguments):
+        """
+        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)
+
+
+class PersistentStoreDict(object):
+    def __init__(self, mgr, prefix):
+        # type: (MgrModule, str) -> None
+        self.mgr = mgr
+        self.prefix = prefix + '.'
+
+    def _mk_store_key(self, key):
+        return self.prefix + key
+
+    def __missing__(self, key):
+        # KeyError won't work for the `in` operator.
+        # https://docs.python.org/3/reference/expressions.html#membership-test-details
+        raise IndexError('PersistentStoreDict: "{}" not found'.format(key))
+
+    def clear(self):
+        # Don't make any assumptions about the content of the values.
+        for item in six.iteritems(self.mgr.get_store_prefix(self.prefix)):
+            k, _ = item
+            self.mgr.set_store(k, None)
+
+    def __getitem__(self, item):
+        # type: (str) -> Any
+        key = self._mk_store_key(item)
+        try:
+            val = self.mgr.get_store(key)
+            if val is None:
+                self.__missing__(key)
+            return json.loads(val)
+        except (KeyError, AttributeError, IndexError, ValueError, TypeError):
+            logging.getLogger(__name__).exception('failed to deserialize')
+            self.mgr.set_store(key, None)
+            raise
+
+    def __setitem__(self, item, value):
+        # type: (str, Any) -> None
+        """
+        value=None is not allowed, as it will remove the key.
+        """
+        key = self._mk_store_key(item)
+        self.mgr.set_store(key, json.dumps(value) if value is not None else None)
+
+    def __delitem__(self, item):
+        self[item] = None
+
+    def __len__(self):
+        return len(self.keys())
+
+    def items(self):
+        # type: () -> Iterator[Tuple[str, Any]]
+        prefix_len = len(self.prefix)
+        try:
+            for item in six.iteritems(self.mgr.get_store_prefix(self.prefix)):
+                k, v = item
+                yield k[prefix_len:], json.loads(v)
+        except (KeyError, AttributeError, IndexError, ValueError, TypeError):
+            logging.getLogger(__name__).exception('failed to deserialize')
+            self.clear()
+
+    def keys(self):
+        # type: () -> Set[str]
+        return {item[0] for item in self.items()}
+
+    def __iter__(self):
+        return iter(self.keys())