]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/mgr_module.py
a5e8889bb7cfcd582b2653a1486022d368d2d9a9
2 import ceph_module
# noqa
3 #import ceph_osdmap #noqa
4 #import ceph_osdmap_incremental #noqa
5 #import ceph_crushmap #noqa
10 from collections
import defaultdict
13 class CPlusPlusHandler(logging
.Handler
):
14 def __init__(self
, module_inst
):
15 super(CPlusPlusHandler
, self
).__init
__()
16 self
._module
= module_inst
18 def emit(self
, record
):
19 if record
.levelno
<= logging
.DEBUG
:
21 elif record
.levelno
<= logging
.INFO
:
23 elif record
.levelno
<= logging
.WARNING
:
28 self
._module
._ceph
_log
(ceph_level
, self
.format(record
))
31 def configure_logger(module_inst
, name
):
32 logger
= logging
.getLogger(name
)
35 # Don't filter any logs at the python level, leave it to C++
36 logger
.setLevel(logging
.DEBUG
)
38 # FIXME: we should learn the log level from C++ land, and then
39 # avoid calling the C++ level log when we know a message is of
40 # an insufficient level to be ultimately output
41 logger
.addHandler(CPlusPlusHandler(module_inst
))
46 def unconfigure_logger(module_inst
, name
):
47 logger
= logging
.getLogger(name
)
48 rm_handlers
= [h
for h
in logger
.handlers
if isinstance(h
, CPlusPlusHandler
)]
50 logger
.removeHandler(h
)
52 class CommandResult(object):
54 Use with MgrModule.send_command
56 def __init__(self
, tag
):
57 self
.ev
= threading
.Event()
62 # This is just a convenience for notifications from
63 # C++ land, to avoid passing addresses around in messages.
66 def complete(self
, r
, outb
, outs
):
74 return self
.r
, self
.outb
, self
.outs
77 class OSDMap(ceph_module
.BasePyOSDMap
):
79 return self
._get
_epoch
()
81 def get_crush_version(self
):
82 return self
._get
_crush
_version
()
87 def new_incremental(self
):
88 return self
._new
_incremental
()
90 def apply_incremental(self
, inc
):
91 return self
._apply
_incremental
(inc
)
94 return self
._get
_crush
()
96 def get_pools_by_take(self
, take
):
97 return self
._get
_pools
_by
_take
(take
).get('pools', [])
99 def calc_pg_upmaps(self
, inc
,
100 max_deviation
=.01, max_iterations
=10, pools
=[]):
101 return self
._calc
_pg
_upmaps
(
103 max_deviation
, max_iterations
, pools
)
105 def map_pool_pgs_up(self
, poolid
):
106 return self
._map
_pool
_pgs
_up
(poolid
)
108 class OSDMapIncremental(ceph_module
.BasePyOSDMapIncremental
):
110 return self
._get
_epoch
()
115 def set_osd_reweights(self
, weightmap
):
117 weightmap is a dict, int to float. e.g. { 0: .9, 1: 1.0, 3: .997 }
119 return self
._set
_osd
_reweights
(weightmap
)
121 def set_crush_compat_weight_set_weights(self
, weightmap
):
123 weightmap is a dict, int to float. devices only. e.g.,
124 { 0: 3.4, 1: 3.3, 2: 3.334 }
126 return self
._set
_crush
_compat
_weight
_set
_weights
(weightmap
)
128 class CRUSHMap(ceph_module
.BasePyCRUSH
):
129 ITEM_NONE
= 0x7fffffff
134 def get_item_weight(self
, item
):
135 return self
._get
_item
_weight
(item
)
137 def get_item_name(self
, item
):
138 return self
._get
_item
_name
(item
)
140 def find_takes(self
):
141 return self
._find
_takes
().get('takes', [])
143 def get_take_weight_osd_map(self
, root
):
144 uglymap
= self
._get
_take
_weight
_osd
_map
(root
)
145 return { int(k
): v
for k
, v
in uglymap
.get('weights', {}).iteritems() }
147 class MgrStandbyModule(ceph_module
.BaseMgrStandbyModule
):
149 Standby modules only implement a serve and shutdown method, they
150 are not permitted to implement commands and they do not receive
153 They only have access to the mgrmap (for accessing service URI info
154 from their active peer), and to configuration settings (read only).
157 def __init__(self
, module_name
, capsule
):
158 super(MgrStandbyModule
, self
).__init
__(capsule
)
159 self
.module_name
= module_name
160 self
._logger
= configure_logger(self
, module_name
)
163 unconfigure_logger(self
, self
.module_name
)
171 The serve method is mandatory for standby modules.
174 raise NotImplementedError()
176 def get_mgr_id(self
):
177 return self
._ceph
_get
_mgr
_id
()
179 def get_config(self
, key
, default
=None):
181 Retrieve the value of a persistent configuration setting
184 :param default: the default value of the config if it is not found
187 r
= self
._ceph
_get
_config
(key
)
194 def get_active_uri(self
):
195 return self
._ceph
_get
_active
_uri
()
197 def get_localized_config(self
, key
, default
=None):
198 r
= self
.get_config(self
.get_mgr_id() + '/' + key
)
200 r
= self
.get_config(key
)
206 class MgrModule(ceph_module
.BaseMgrModule
):
209 # Priority definitions for perf counters
213 PRIO_UNINTERESTING
= 2
216 # counter value types
221 PERFCOUNTER_LONGRUNAVG
= 4
222 PERFCOUNTER_COUNTER
= 8
223 PERFCOUNTER_HISTOGRAM
= 0x10
224 PERFCOUNTER_TYPE_MASK
= ~
2
226 def __init__(self
, module_name
, py_modules_ptr
, this_ptr
):
227 self
.module_name
= module_name
229 # If we're taking over from a standby module, let's make sure
230 # its logger was unconfigured before we hook ours up
231 unconfigure_logger(self
, self
.module_name
)
232 self
._logger
= configure_logger(self
, module_name
)
234 super(MgrModule
, self
).__init
__(py_modules_ptr
, this_ptr
)
236 self
._version
= self
._ceph
_get
_version
()
238 self
._perf
_schema
_cache
= None
241 unconfigure_logger(self
, self
.module_name
)
243 def update_perf_schema(self
, daemon_type
, daemon_name
):
245 For plugins that use get_all_perf_counters, call this when
246 receiving a notification of type 'perf_schema_update', to
247 prompt MgrModule to update its cache of counter schemas.
262 def get_context(self
):
264 :return: a Python capsule containing a C++ CephContext pointer
266 return self
._ceph
_get
_context
()
268 def notify(self
, notify_type
, notify_id
):
270 Called by the ceph-mgr service to notify the Python plugin
271 that new state is available.
277 Called by the ceph-mgr service to start any server that
278 is provided by this Python plugin. The implementation
279 of this function should block until ``shutdown`` is called.
281 You *must* implement ``shutdown`` if you implement ``serve``
287 Called by the ceph-mgr service to request that this
288 module drop out of its serve() function. You do not
289 need to implement this if you do not implement serve()
295 def get(self
, data_name
):
297 Called by the plugin to load some cluster state from ceph-mgr
299 return self
._ceph
_get
(data_name
)
301 def get_server(self
, hostname
):
303 Called by the plugin to load information about a particular
306 :param hostname: a hostame
308 return self
._ceph
_get
_server
(hostname
)
310 def get_perf_schema(self
, svc_type
, svc_name
):
312 Called by the plugin to fetch perf counter schema info.
313 svc_name can be nullptr, as can svc_type, in which case
318 :return: list of dicts describing the counters requested
320 return self
._ceph
_get
_perf
_schema
(svc_type
, svc_name
)
322 def get_counter(self
, svc_type
, svc_name
, path
):
324 Called by the plugin to fetch data for a particular perf counter
325 on a particular service.
330 :return: A list of two-element lists containing time and value
332 return self
._ceph
_get
_counter
(svc_type
, svc_name
, path
)
334 def list_servers(self
):
336 Like ``get_server``, but instead of returning information
337 about just one node, return all the nodes in an array.
339 return self
._ceph
_get
_server
(None)
341 def get_metadata(self
, svc_type
, svc_id
):
343 Fetch the metadata for a particular service.
345 :param svc_type: string (e.g., 'mds', 'osd', 'mon')
346 :param svc_id: string
349 return self
._ceph
_get
_metadata
(svc_type
, svc_id
)
351 def get_daemon_status(self
, svc_type
, svc_id
):
353 Fetch the latest status for a particular service daemon.
355 :param svc_type: string (e.g., 'rgw')
356 :param svc_id: string
359 return self
._ceph
_get
_daemon
_status
(svc_type
, svc_id
)
361 def send_command(self
, *args
, **kwargs
):
363 Called by the plugin to send a command to the mon
366 self
._ceph
_send
_command
(*args
, **kwargs
)
368 def set_health_checks(self
, checks
):
370 Set module's health checks
372 Set the module's current map of health checks. Argument is a
373 dict of check names to info, in this form:
377 'severity': 'warning', # or 'error'
378 'summary': 'summary string',
379 'detail': [ 'list', 'of', 'detail', 'strings' ],
383 'summary': 'bars are bad',
384 'detail': [ 'too hard' ],
388 :param list: dict of health check dicts
390 self
._ceph
_set
_health
_checks
(checks
)
392 def handle_command(self
, cmd
):
394 Called by ceph-mgr to request the plugin to handle one
395 of the commands that it declared in self.COMMANDS
397 Return a status code, an output buffer, and an
398 output string. The output buffer is for data results,
399 the output string is for informative text.
401 :param cmd: dict, from Ceph's cmdmap_t
403 :return: 3-tuple of (int, str, str)
406 # Should never get called if they didn't declare
408 raise NotImplementedError()
410 def get_mgr_id(self
):
416 return self
._ceph
_get
_mgr
_id
()
418 def get_config(self
, key
, default
=None):
420 Retrieve the value of a persistent configuration setting
425 r
= self
._ceph
_get
_config
(key
)
431 def get_config_prefix(self
, key_prefix
):
433 Retrieve a dict of config values with the given prefix
435 :param key_prefix: str
438 return self
._ceph
_get
_config
_prefix
(key_prefix
)
440 def get_localized_config(self
, key
, default
=None):
442 Retrieve localized configuration for this ceph-mgr instance
447 r
= self
.get_config(self
.get_mgr_id() + '/' + key
)
449 r
= self
.get_config(key
)
455 def set_config(self
, key
, val
):
457 Set the value of a persistent configuration setting
462 self
._ceph
_set
_config
(key
, val
)
464 def set_localized_config(self
, key
, val
):
466 Set localized configuration for this ceph-mgr instance
471 return self
._ceph
_set
_config
(self
.get_mgr_id() + '/' + key
, val
)
473 def set_config_json(self
, key
, val
):
475 Helper for setting json-serialized-config
478 :param val: json-serializable object
480 self
._ceph
_set
_config
(key
, json
.dumps(val
))
482 def get_config_json(self
, key
):
484 Helper for getting json-serialized config
489 raw
= self
.get_config(key
)
493 return json
.loads(raw
)
497 Run a self-test on the module. Override this function and implement
498 a best as possible self-test for (automated) testing of the module
503 def get_osdmap(self
):
505 Get a handle to an OSDMap. If epoch==0, get a handle for the latest
509 return self
._ceph
_get
_osdmap
()
511 def get_all_perf_counters(self
, prio_limit
=PRIO_USEFUL
):
513 Return the perf counters currently known to this ceph-mgr
514 instance, filtered by priority equal to or greater than `prio_limit`.
516 The result us a map of string to dict, associating services
517 (like "osd.123") with their counters. The counter
518 dict for each service maps counter paths to a counter
519 info structure, which is the information from
520 the schema, plus an additional "value" member with the latest
524 result
= defaultdict(dict)
526 # TODO: improve C++->Python interface to return just
527 # the latest if that's all we want.
528 def get_latest(daemon_type
, daemon_name
, counter
):
529 data
= self
.get_counter(daemon_type
, daemon_name
, counter
)[counter
]
535 for server
in self
.list_servers():
536 for service
in server
['services']:
537 if service
['type'] not in ("mds", "osd", "mon"):
540 schema
= self
.get_perf_schema(service
['type'], service
['id'])
542 self
.log
.warn("No perf counter schema for {0}.{1}".format(
543 service
['type'], service
['id']
547 # Value is returned in a potentially-multi-service format,
548 # get just the service we're asking about
549 svc_full_name
= "{0}.{1}".format(service
['type'], service
['id'])
550 schema
= schema
[svc_full_name
]
552 # Populate latest values
553 for counter_path
, counter_schema
in schema
.items():
554 # self.log.debug("{0}: {1}".format(
555 # counter_path, json.dumps(counter_schema)
557 if counter_schema
['priority'] < prio_limit
:
560 counter_info
= counter_schema
561 counter_info
['value'] = get_latest(service
['type'], service
['id'], counter_path
)
562 result
[svc_full_name
][counter_path
] = counter_info
564 self
.log
.debug("returning {0} counter".format(len(result
)))
568 def set_uri(self
, uri
):
570 If the module exposes a service, then call this to publish the
571 address once it is available.
575 return self
._ceph
_set
_uri
(uri
)