--- /dev/null
+
+import ceph_module # noqa
+#import ceph_osdmap #noqa
+#import ceph_osdmap_incremental #noqa
+#import ceph_crushmap #noqa
+
+import json
+import logging
+import threading
+from collections import defaultdict
+
+
+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, name):
+ logger = logging.getLogger(name)
+
+
+ # Don't filter any logs at the python level, leave it to C++
+ logger.setLevel(logging.DEBUG)
+
+ # 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.addHandler(CPlusPlusHandler(module_inst))
+
+ return logger
+
+
+def unconfigure_logger(module_inst, name):
+ logger = logging.getLogger(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):
+ 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
+
+ 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 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 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=[]):
+ 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)
+
+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):
+ 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 uglymap.get('weights', {}).iteritems() }
+
+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 acecssing service URI info
+ from their active peer), and to configuration settings (read only).
+ """
+
+ def __init__(self, module_name, capsule):
+ super(MgrStandbyModule, self).__init__(capsule)
+ self.module_name = module_name
+ self._logger = configure_logger(self, module_name)
+
+ def __del__(self):
+ unconfigure_logger(self, self.module_name)
+
+ @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_config(self, key):
+ return self._ceph_get_config(key)
+
+ def get_active_uri(self):
+ return self._ceph_get_active_uri()
+
+ def get_localized_config(self, key, default=None):
+ r = self.get_config(self.get_mgr_id() + '/' + key)
+ if r is None:
+ r = self.get_config(key)
+
+ if r is None:
+ r = default
+ return r
+
+class MgrModule(ceph_module.BaseMgrModule):
+ COMMANDS = []
+
+ # 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 = ~2
+
+ 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, self.module_name)
+ 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
+
+ def __del__(self):
+ unconfigure_logger(self, self.module_name)
+
+ def update_perf_schema(self, daemon_type, daemon_name):
+ """
+ For plugins that use get_all_perf_counters, call this when
+ receiving a notification of type 'perf_schema_update', to
+ prompt MgrModule to update its cache of counter schemas.
+
+ :param daemon_type:
+ :param daemon_name:
+ :return:
+ """
+
+ @property
+ def log(self):
+ return self._logger
+
+ @property
+ def version(self):
+ return self._version
+
+ 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.
+ """
+ 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
+ """
+ pass
+
+ def get(self, data_name):
+ """
+ Called by the plugin to load some cluster state from ceph-mgr
+ """
+ return self._ceph_get(data_name)
+
+ def get_server(self, hostname):
+ """
+ Called by the plugin to load information about a particular
+ node from ceph-mgr.
+
+ :param hostname: a hostame
+ """
+ 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 svc_type:
+ :param 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 data for a particular perf counter
+ on a particular service.
+
+ :param svc_type:
+ :param svc_name:
+ :param path:
+ :return: A list of two-element lists containing time and value
+ """
+ return self._ceph_get_counter(svc_type, svc_name, path)
+
+ def list_servers(self):
+ """
+ Like ``get_server``, but instead of returning information
+ about just one node, return all the nodes in an array.
+ """
+ return self._ceph_get_server(None)
+
+ def get_metadata(self, svc_type, svc_id):
+ """
+ Fetch the metadata for a particular service.
+
+ :param svc_type: string (e.g., 'mds', 'osd', 'mon')
+ :param svc_id: string
+ :return: dict
+ """
+ return self._ceph_get_metadata(svc_type, svc_id)
+
+ def get_daemon_status(self, svc_type, svc_id):
+ """
+ Fetch the latest status for a particular service daemon.
+
+ :param svc_type: string (e.g., 'rgw')
+ :param svc_id: string
+ :return: dict
+ """
+ return self._ceph_get_daemon_status(svc_type, svc_id)
+
+ def send_command(self, *args, **kwargs):
+ """
+ Called by the plugin to send a command to the mon
+ cluster.
+ """
+ self._ceph_send_command(*args, **kwargs)
+
+ def set_health_checks(self, checks):
+ """
+ Set module's health 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, 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 cmd: dict, from Ceph's cmdmap_t
+
+ :return: 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 mgr id.
+
+ :return: str
+ """
+ return self._ceph_get_mgr_id()
+
+ def get_config(self, key, default=None):
+ """
+ Retrieve the value of a persistent configuration setting
+
+ :param key: str
+ :return: str
+ """
+ r = self._ceph_get_config(key)
+ if r is None:
+ return default
+ else:
+ return r
+
+ def get_config_prefix(self, key_prefix):
+ """
+ Retrieve a dict of config values with the given prefix
+
+ :param key_prefix: str
+ :return: str
+ """
+ return self._ceph_get_config_prefix(key_prefix)
+
+ def get_localized_config(self, key, default=None):
+ """
+ Retrieve localized configuration for this ceph-mgr instance
+ :param key: str
+ :param default: str
+ :return: str
+ """
+ r = self.get_config(self.get_mgr_id() + '/' + key)
+ if r is None:
+ r = self.get_config(key)
+
+ if r is None:
+ r = default
+ return r
+
+ def set_config(self, key, val):
+ """
+ Set the value of a persistent configuration setting
+
+ :param key: str
+ :param val: str
+ """
+ self._ceph_set_config(key, val)
+
+ def set_localized_config(self, key, val):
+ """
+ Set localized configuration for this ceph-mgr instance
+ :param key: str
+ :param default: str
+ :return: str
+ """
+ return self._ceph_set_config(self.get_mgr_id() + '/' + key, val)
+
+ def set_config_json(self, key, val):
+ """
+ Helper for setting json-serialized-config
+
+ :param key: str
+ :param val: json-serializable object
+ """
+ self._ceph_set_config(key, json.dumps(val))
+
+ def get_config_json(self, key):
+ """
+ Helper for getting json-serialized config
+
+ :param key: str
+ :return: object
+ """
+ raw = self.get_config(key)
+ if raw is None:
+ return None
+ else:
+ return json.loads(raw)
+
+ 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
+ :return: bool
+ """
+ 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_all_perf_counters(self, prio_limit=PRIO_USEFUL):
+ """
+ Return the perf counters currently known to this ceph-mgr
+ instance, filtered by priority equal to or greater than `prio_limit`.
+
+ The result us 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)
+
+ # TODO: improve C++->Python interface to return just
+ # the latest if that's all we want.
+ def get_latest(daemon_type, daemon_name, counter):
+ data = self.get_counter(daemon_type, daemon_name, counter)[counter]
+ if data:
+ return data[-1][1]
+ else:
+ return 0
+
+ for server in self.list_servers():
+ for service in server['services']:
+ if service['type'] not in ("mds", "osd", "mon"):
+ 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 = counter_schema
+ counter_info['value'] = 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)
Code Review / stor4nfv.git / blobdiff