1 import ceph_module
# noqa
3 from typing
import cast
, Tuple
, Any
, Dict
, Generic
, Optional
, Callable
, List
, \
4 Mapping
, NamedTuple
, Sequence
, Union
, TYPE_CHECKING
7 if sys
.version_info
>= (3, 8):
8 from typing
import Literal
10 from typing_extensions
import Literal
18 from collections
import defaultdict
19 from enum
import IntEnum
24 from ceph_argparse
import CephArgtype
25 from mgr_util
import profile_method
27 if sys
.version_info
>= (3, 8):
28 from typing
import get_args
, get_origin
34 return getattr(tp
, '__args__', ())
37 return getattr(tp
, '__origin__', None)
40 ERROR_MSG_EMPTY_INPUT_FILE
= 'Empty content: please add a password/secret to the file.'
41 ERROR_MSG_NO_INPUT_FILE
= 'Please specify the file containing the password/secret with "-i" option.'
42 # Full list of strings in "osd_types.cc:pg_state_string()"
81 class CommandResult(object):
83 Use with MgrModule.send_command
86 def __init__(self
, tag
: Optional
[str] = None):
87 self
.ev
= threading
.Event()
92 # This is just a convenience for notifications from
93 # C++ land, to avoid passing addresses around in messages.
94 self
.tag
= tag
if tag
else ""
96 def complete(self
, r
: int, outb
: str, outs
: str) -> None:
102 def wait(self
) -> Tuple
[int, str, str]:
104 return self
.r
, self
.outb
, self
.outs
107 class HandleCommandResult(NamedTuple
):
109 Tuple containing the result of `handle_command()`
111 Only write to stderr if there is an error, or in extraordinary circumstances
113 Avoid having `ceph foo bar` commands say "did foo bar" on success unless there
114 is critical information to include there.
116 Everything programmatically consumable should be put on stdout
118 retval
: int = 0 # return code. E.g. 0 or -errno.EINVAL
119 stdout
: str = "" # data of this result.
120 stderr
: str = "" # Typically used for error messages.
123 class MonCommandFailed(RuntimeError): pass
126 class OSDMap(ceph_module
.BasePyOSDMap
):
127 def get_epoch(self
) -> int:
128 return self
._get
_epoch
()
130 def get_crush_version(self
) -> int:
131 return self
._get
_crush
_version
()
133 def dump(self
) -> Dict
[str, Any
]:
136 def get_pools(self
) -> Dict
[int, Dict
[str, Any
]]:
137 # FIXME: efficient implementation
139 return dict([(p
['pool'], p
) for p
in d
['pools']])
141 def get_pools_by_name(self
) -> Dict
[str, Dict
[str, Any
]]:
142 # FIXME: efficient implementation
144 return dict([(p
['pool_name'], p
) for p
in d
['pools']])
146 def new_incremental(self
) -> 'OSDMapIncremental':
147 return self
._new
_incremental
()
149 def apply_incremental(self
, inc
: 'OSDMapIncremental') -> 'OSDMap':
150 return self
._apply
_incremental
(inc
)
152 def get_crush(self
) -> 'CRUSHMap':
153 return self
._get
_crush
()
155 def get_pools_by_take(self
, take
: int) -> List
[int]:
156 return self
._get
_pools
_by
_take
(take
).get('pools', [])
158 def calc_pg_upmaps(self
, inc
: 'OSDMapIncremental',
160 max_iterations
: int = 10,
161 pools
: Optional
[List
[str]] = None) -> int:
164 return self
._calc
_pg
_upmaps
(
166 max_deviation
, max_iterations
, pools
)
168 def map_pool_pgs_up(self
, poolid
: int) -> List
[int]:
169 return self
._map
_pool
_pgs
_up
(poolid
)
171 def pg_to_up_acting_osds(self
, pool_id
: int, ps
: int) -> Dict
[str, Any
]:
172 return self
._pg
_to
_up
_acting
_osds
(pool_id
, ps
)
174 def pool_raw_used_rate(self
, pool_id
: int) -> float:
175 return self
._pool
_raw
_used
_rate
(pool_id
)
177 def get_ec_profile(self
, name
: str) -> Optional
[List
[Dict
[str, str]]]:
178 # FIXME: efficient implementation
180 return d
['erasure_code_profiles'].get(name
, None)
182 def get_require_osd_release(self
) -> str:
184 return d
['require_osd_release']
187 class OSDMapIncremental(ceph_module
.BasePyOSDMapIncremental
):
188 def get_epoch(self
) -> int:
189 return self
._get
_epoch
()
191 def dump(self
) -> Dict
[str, Any
]:
194 def set_osd_reweights(self
, weightmap
: Dict
[int, float]) -> None:
196 weightmap is a dict, int to float. e.g. { 0: .9, 1: 1.0, 3: .997 }
198 return self
._set
_osd
_reweights
(weightmap
)
200 def set_crush_compat_weight_set_weights(self
, weightmap
: Dict
[str, float]) -> None:
202 weightmap is a dict, int to float. devices only. e.g.,
203 { 0: 3.4, 1: 3.3, 2: 3.334 }
205 return self
._set
_crush
_compat
_weight
_set
_weights
(weightmap
)
208 class CRUSHMap(ceph_module
.BasePyCRUSH
):
209 ITEM_NONE
= 0x7fffffff
210 DEFAULT_CHOOSE_ARGS
= '-1'
212 def dump(self
) -> Dict
[str, Any
]:
215 def get_item_weight(self
, item
: int) -> Optional
[int]:
216 return self
._get
_item
_weight
(item
)
218 def get_item_name(self
, item
: int) -> Optional
[str]:
219 return self
._get
_item
_name
(item
)
221 def find_takes(self
) -> List
[int]:
222 return self
._find
_takes
().get('takes', [])
224 def get_take_weight_osd_map(self
, root
: int) -> Dict
[int, float]:
225 uglymap
= self
._get
_take
_weight
_osd
_map
(root
)
226 return {int(k
): v
for k
, v
in uglymap
.get('weights', {}).items()}
229 def have_default_choose_args(dump
: Dict
[str, Any
]) -> bool:
230 return CRUSHMap
.DEFAULT_CHOOSE_ARGS
in dump
.get('choose_args', {})
233 def get_default_choose_args(dump
: Dict
[str, Any
]) -> List
[Dict
[str, Any
]]:
234 choose_args
= dump
.get('choose_args')
235 assert isinstance(choose_args
, dict)
236 return choose_args
.get(CRUSHMap
.DEFAULT_CHOOSE_ARGS
, [])
238 def get_rule(self
, rule_name
: str) -> Optional
[Dict
[str, Any
]]:
239 # TODO efficient implementation
240 for rule
in self
.dump()['rules']:
241 if rule_name
== rule
['rule_name']:
246 def get_rule_by_id(self
, rule_id
: int) -> Optional
[Dict
[str, Any
]]:
247 for rule
in self
.dump()['rules']:
248 if rule
['rule_id'] == rule_id
:
253 def get_rule_root(self
, rule_name
: str) -> Optional
[int]:
254 rule
= self
.get_rule(rule_name
)
259 first_take
= next(s
for s
in rule
['steps'] if s
.get('op') == 'take')
260 except StopIteration:
261 logging
.warning("CRUSH rule '{0}' has no 'take' step".format(
265 return first_take
['item']
267 def get_osds_under(self
, root_id
: int) -> List
[int]:
268 # TODO don't abuse dump like this
270 buckets
= dict([(b
['id'], b
) for b
in d
['buckets']])
274 def accumulate(b
: Dict
[str, Any
]) -> None:
275 for item
in b
['items']:
277 osd_list
.append(item
['id'])
280 accumulate(buckets
[item
['id']])
284 accumulate(buckets
[root_id
])
288 def device_class_counts(self
) -> Dict
[str, int]:
289 result
= defaultdict(int) # type: Dict[str, int]
290 # TODO don't abuse dump like this
292 for device
in d
['devices']:
293 cls
= device
.get('class', None)
299 HandlerFuncType
= Callable
[..., Tuple
[int, str, str]]
302 class CLICommand(object):
303 COMMANDS
= {} # type: Dict[str, CLICommand]
312 self
.func
= None # type: Optional[Callable]
313 self
.arg_spec
= {} # type: Dict[str, Any]
314 self
.first_default
= -1
316 KNOWN_ARGS
= '_', 'self', 'mgr', 'inbuf', 'return'
319 def load_func_metadata(f
: HandlerFuncType
) -> Tuple
[str, Dict
[str, Any
], int, str]:
320 desc
= inspect
.getdoc(f
) or ''
321 full_argspec
= inspect
.getfullargspec(f
)
322 arg_spec
= full_argspec
.annotations
323 first_default
= len(arg_spec
)
324 if full_argspec
.defaults
:
325 first_default
-= len(full_argspec
.defaults
)
327 for index
, arg
in enumerate(full_argspec
.args
):
328 if arg
in CLICommand
.KNOWN_ARGS
:
330 assert arg
in arg_spec
, \
331 f
"'{arg}' is not annotated for {f}: {full_argspec}"
332 has_default
= index
>= first_default
333 args
.append(CephArgtype
.to_argdesc(arg_spec
[arg
],
336 return desc
, arg_spec
, first_default
, ' '.join(args
)
338 def store_func_metadata(self
, f
: HandlerFuncType
) -> None:
339 self
.desc
, self
.arg_spec
, self
.first_default
, self
.args
= \
340 self
.load_func_metadata(f
)
342 def __call__(self
, func
: HandlerFuncType
) -> HandlerFuncType
:
343 self
.store_func_metadata(func
)
345 self
.COMMANDS
[self
.prefix
] = self
348 def _get_arg_value(self
, kwargs_switch
: bool, key
: str, val
: Any
) -> Tuple
[bool, str, Any
]:
349 def start_kwargs() -> bool:
350 if isinstance(val
, str) and '=' in val
:
351 k
, v
= val
.split('=', 1)
352 if k
in self
.arg_spec
:
356 if not kwargs_switch
:
357 kwargs_switch
= start_kwargs()
360 k
, v
= val
.split('=', 1)
363 return kwargs_switch
, k
.replace('-', '_'), v
365 def _collect_args_by_argspec(self
, cmd_dict
: Dict
[str, Any
]) -> Dict
[str, Any
]:
367 kwargs_switch
= False
368 for index
, (name
, tp
) in enumerate(self
.arg_spec
.items()):
369 if name
in CLICommand
.KNOWN_ARGS
:
371 assert self
.first_default
>= 0
372 raw_v
= cmd_dict
.get(name
)
373 if index
>= self
.first_default
:
376 kwargs_switch
, k
, v
= self
._get
_arg
_value
(kwargs_switch
,
378 kwargs
[k
] = CephArgtype
.cast_to(tp
, v
)
383 cmd_dict
: Dict
[str, Any
],
384 inbuf
: Optional
[str] = None) -> HandleCommandResult
:
385 kwargs
= self
._collect
_args
_by
_argspec
(cmd_dict
)
387 kwargs
['inbuf'] = inbuf
389 return self
.func(mgr
, **kwargs
)
391 def dump_cmd(self
) -> Dict
[str, Union
[str, bool]]:
393 'cmd': '{} {}'.format(self
.prefix
, self
.args
),
400 def dump_cmd_list(cls
) -> List
[Dict
[str, Union
[str, bool]]]:
401 return [cmd
.dump_cmd() for cmd
in cls
.COMMANDS
.values()]
404 def CLIReadCommand(prefix
: str, poll
: bool = False) -> CLICommand
:
405 return CLICommand(prefix
, "r", poll
)
408 def CLIWriteCommand(prefix
: str, poll
: bool = False) -> CLICommand
:
409 return CLICommand(prefix
, "w", poll
)
412 def CLICheckNonemptyFileInput(func
: HandlerFuncType
) -> HandlerFuncType
:
413 @functools.wraps(func
)
414 def check(*args
: Any
, **kwargs
: Any
) -> Tuple
[int, str, str]:
415 if 'inbuf' not in kwargs
:
416 return -errno
.EINVAL
, '', ERROR_MSG_NO_INPUT_FILE
417 if isinstance(kwargs
['inbuf'], str):
418 # Delete new line separator at EOF (it may have been added by a text editor).
419 kwargs
['inbuf'] = kwargs
['inbuf'].rstrip('\r\n').rstrip('\n')
420 if not kwargs
['inbuf']:
421 return -errno
.EINVAL
, '', ERROR_MSG_EMPTY_INPUT_FILE
422 return func(*args
, **kwargs
)
423 check
.__signature
__ = inspect
.signature(func
) # type: ignore[attr-defined]
427 def _get_localized_key(prefix
: str, key
: str) -> str:
428 return '{}/{}'.format(prefix
, key
)
432 MODULE_OPTIONS types and Option Class
435 OptionTypeLabel
= Literal
[
436 'uint', 'int', 'str', 'float', 'bool', 'addr', 'addrvec', 'uuid', 'size', 'secs']
439 # common/options.h: value_t
440 OptionValue
= Optional
[Union
[bool, int, float, str]]
445 Helper class to declare options for MODULE_OPTIONS list.
446 TODO: Replace with typing.TypedDict when in python_version >= 3.8
452 default
: OptionValue
= None,
453 type: 'OptionTypeLabel' = 'str',
454 desc
: Optional
[str] = None,
455 long_desc
: Optional
[str] = None,
456 min: OptionValue
= None,
457 max: OptionValue
= None,
458 enum_allowed
: Optional
[List
[str]] = None,
459 tags
: Optional
[List
[str]] = None,
460 see_also
: Optional
[List
[str]] = None,
461 runtime
: bool = False,
463 super(Option
, self
).__init
__(
464 (k
, v
) for k
, v
in vars().items()
465 if k
!= 'self' and v
is not None)
470 Helper class to declare options for COMMANDS list.
472 It also allows to specify prefix and args separately, as well as storing a
476 >>> Command(prefix="example",
477 ... args="name=arg,type=CephInt",
480 {'poll': False, 'cmd': 'example name=arg,type=CephInt', 'perm': 'w', 'desc': 'Blah'}
486 handler
: HandlerFuncType
,
490 super().__init
__(perm
=perm
,
493 self
.handler
= handler
496 def returns_command_result(instance
: Any
,
497 f
: HandlerFuncType
) -> Callable
[..., HandleCommandResult
]:
499 def wrapper(mgr
: Any
, *args
: Any
, **kwargs
: Any
) -> HandleCommandResult
:
500 retval
, stdout
, stderr
= f(instance
or mgr
, *args
, **kwargs
)
501 return HandleCommandResult(retval
, stdout
, stderr
)
502 wrapper
.__signature
__ = inspect
.signature(f
) # type: ignore[attr-defined]
505 def register(self
, instance
: bool = False) -> HandlerFuncType
:
507 Register a CLICommand handler. It allows an instance to register bound
508 methods. In that case, the mgr instance is not passed, and it's expected
509 to be available in the class instance.
510 It also uses HandleCommandResult helper to return a wrapped a tuple of 3
513 cmd
= CLICommand(prefix
=self
.prefix
, perm
=self
['perm'])
514 return cmd(self
.returns_command_result(instance
, self
.handler
))
517 class CPlusPlusHandler(logging
.Handler
):
518 def __init__(self
, module_inst
: Any
):
519 super(CPlusPlusHandler
, self
).__init
__()
520 self
._module
= module_inst
521 self
.setFormatter(logging
.Formatter("[{} %(levelname)-4s %(name)s] %(message)s"
522 .format(module_inst
.module_name
)))
524 def emit(self
, record
: logging
.LogRecord
) -> None:
525 if record
.levelno
>= self
.level
:
526 self
._module
._ceph
_log
(self
.format(record
))
529 class ClusterLogHandler(logging
.Handler
):
530 def __init__(self
, module_inst
: Any
):
532 self
._module
= module_inst
533 self
.setFormatter(logging
.Formatter("%(message)s"))
535 def emit(self
, record
: logging
.LogRecord
) -> None:
537 logging
.DEBUG
: MgrModule
.ClusterLogPrio
.DEBUG
,
538 logging
.INFO
: MgrModule
.ClusterLogPrio
.INFO
,
539 logging
.WARNING
: MgrModule
.ClusterLogPrio
.WARN
,
540 logging
.ERROR
: MgrModule
.ClusterLogPrio
.ERROR
,
541 logging
.CRITICAL
: MgrModule
.ClusterLogPrio
.ERROR
,
543 level
= levelmap
[record
.levelno
]
544 if record
.levelno
>= self
.level
:
545 self
._module
.cluster_log(self
._module
.module_name
,
550 class FileHandler(logging
.FileHandler
):
551 def __init__(self
, module_inst
: Any
):
552 path
= module_inst
.get_ceph_option("log_file")
553 idx
= path
.rfind(".log")
555 self
.path
= "{}.{}.log".format(path
[:idx
], module_inst
.module_name
)
557 self
.path
= "{}.{}".format(path
, module_inst
.module_name
)
558 super(FileHandler
, self
).__init
__(self
.path
, delay
=True)
559 self
.setFormatter(logging
.Formatter("%(asctime)s [%(threadName)s] [%(levelname)-4s] [%(name)s] %(message)s"))
562 class MgrModuleLoggingMixin(object):
563 def _configure_logging(self
,
568 log_to_cluster
: bool) -> None:
569 self
._mgr
_level
: Optional
[str] = None
570 self
._module
_level
: Optional
[str] = None
571 self
._root
_logger
= logging
.getLogger()
573 self
._unconfigure
_logging
()
575 # the ceph log handler is initialized only once
576 self
._mgr
_log
_handler
= CPlusPlusHandler(self
)
577 self
._cluster
_log
_handler
= ClusterLogHandler(self
)
578 self
._file
_log
_handler
= FileHandler(self
)
580 self
.log_to_file
= log_to_file
581 self
.log_to_cluster
= log_to_cluster
583 self
._root
_logger
.addHandler(self
._mgr
_log
_handler
)
585 self
._root
_logger
.addHandler(self
._file
_log
_handler
)
587 self
._root
_logger
.addHandler(self
._cluster
_log
_handler
)
589 self
._root
_logger
.setLevel(logging
.NOTSET
)
590 self
._set
_log
_level
(mgr_level
, module_level
, cluster_level
)
592 def _unconfigure_logging(self
) -> None:
593 # remove existing handlers:
595 h
for h
in self
._root
_logger
.handlers
596 if (isinstance(h
, CPlusPlusHandler
) or
597 isinstance(h
, FileHandler
) or
598 isinstance(h
, ClusterLogHandler
))]
599 for h
in rm_handlers
:
600 self
._root
_logger
.removeHandler(h
)
601 self
.log_to_file
= False
602 self
.log_to_cluster
= False
604 def _set_log_level(self
,
607 cluster_level
: str) -> None:
608 self
._cluster
_log
_handler
.setLevel(cluster_level
.upper())
610 module_level
= module_level
.upper() if module_level
else ''
611 if not self
._module
_level
:
612 # using debug_mgr level
613 if not module_level
and self
._mgr
_level
== mgr_level
:
614 # no change in module level neither in debug_mgr
617 if self
._module
_level
== module_level
:
618 # no change in module level
621 if not self
._module
_level
and not module_level
:
622 level
= self
._ceph
_log
_level
_to
_python
(mgr_level
)
623 self
.getLogger().debug("setting log level based on debug_mgr: %s (%s)",
625 elif self
._module
_level
and not module_level
:
626 level
= self
._ceph
_log
_level
_to
_python
(mgr_level
)
627 self
.getLogger().warning("unsetting module log level, falling back to "
628 "debug_mgr level: %s (%s)", level
, mgr_level
)
631 self
.getLogger().debug("setting log level: %s", level
)
633 self
._module
_level
= module_level
634 self
._mgr
_level
= mgr_level
636 self
._mgr
_log
_handler
.setLevel(level
)
637 self
._file
_log
_handler
.setLevel(level
)
639 def _enable_file_log(self
) -> None:
641 self
.getLogger().warning("enabling logging to file")
642 self
.log_to_file
= True
643 self
._root
_logger
.addHandler(self
._file
_log
_handler
)
645 def _disable_file_log(self
) -> None:
647 self
.getLogger().warning("disabling logging to file")
648 self
.log_to_file
= False
649 self
._root
_logger
.removeHandler(self
._file
_log
_handler
)
651 def _enable_cluster_log(self
) -> None:
653 self
.getLogger().warning("enabling logging to cluster")
654 self
.log_to_cluster
= True
655 self
._root
_logger
.addHandler(self
._cluster
_log
_handler
)
657 def _disable_cluster_log(self
) -> None:
658 # disable cluster log
659 self
.getLogger().warning("disabling logging to cluster")
660 self
.log_to_cluster
= False
661 self
._root
_logger
.removeHandler(self
._cluster
_log
_handler
)
663 def _ceph_log_level_to_python(self
, log_level
: str) -> str:
666 ceph_log_level
= int(log_level
.split("/", 1)[0])
673 if ceph_log_level
<= 0:
674 log_level
= "CRITICAL"
675 elif ceph_log_level
<= 1:
676 log_level
= "WARNING"
677 elif ceph_log_level
<= 4:
681 def getLogger(self
, name
: Optional
[str] = None) -> logging
.Logger
:
682 return logging
.getLogger(name
)
685 class MgrStandbyModule(ceph_module
.BaseMgrStandbyModule
, MgrModuleLoggingMixin
):
687 Standby modules only implement a serve and shutdown method, they
688 are not permitted to implement commands and they do not receive
691 They only have access to the mgrmap (for accessing service URI info
692 from their active peer), and to configuration settings (read only).
695 MODULE_OPTIONS
: List
[Option
] = []
696 MODULE_OPTION_DEFAULTS
= {} # type: Dict[str, Any]
698 def __init__(self
, module_name
: str, capsule
: Any
):
699 super(MgrStandbyModule
, self
).__init
__(capsule
)
700 self
.module_name
= module_name
702 # see also MgrModule.__init__()
703 for o
in self
.MODULE_OPTIONS
:
706 self
.MODULE_OPTION_DEFAULTS
[o
['name']] = o
['default']
708 self
.MODULE_OPTION_DEFAULTS
[o
['name']] = str(o
['default'])
710 # mock does not return a str
711 mgr_level
= cast(str, self
.get_ceph_option("debug_mgr"))
712 log_level
= cast(str, self
.get_module_option("log_level"))
713 cluster_level
= cast(str, self
.get_module_option('log_to_cluster_level'))
714 self
._configure
_logging
(mgr_level
, log_level
, cluster_level
,
717 # for backwards compatibility
718 self
._logger
= self
.getLogger()
720 def __del__(self
) -> None:
722 self
._unconfigure
_logging
()
724 def _cleanup(self
) -> None:
728 def _register_options(cls
, module_name
: str) -> None:
729 cls
.MODULE_OPTIONS
.append(
730 Option(name
='log_level', type='str', default
="", runtime
=True,
731 enum_allowed
=['info', 'debug', 'critical', 'error',
733 cls
.MODULE_OPTIONS
.append(
734 Option(name
='log_to_file', type='bool', default
=False, runtime
=True))
735 if not [x
for x
in cls
.MODULE_OPTIONS
if x
['name'] == 'log_to_cluster']:
736 cls
.MODULE_OPTIONS
.append(
737 Option(name
='log_to_cluster', type='bool', default
=False,
739 cls
.MODULE_OPTIONS
.append(
740 Option(name
='log_to_cluster_level', type='str', default
='info',
742 enum_allowed
=['info', 'debug', 'critical', 'error',
746 def log(self
) -> logging
.Logger
:
749 def serve(self
) -> None:
751 The serve method is mandatory for standby modules.
754 raise NotImplementedError()
756 def get_mgr_id(self
) -> str:
757 return self
._ceph
_get
_mgr
_id
()
759 def get_module_option(self
, key
: str, default
: OptionValue
= None) -> OptionValue
:
761 Retrieve the value of a persistent configuration setting
763 :param default: the default value of the config if it is not found
765 r
= self
._ceph
_get
_module
_option
(key
)
767 return self
.MODULE_OPTION_DEFAULTS
.get(key
, default
)
771 def get_ceph_option(self
, key
: str) -> OptionValue
:
772 return self
._ceph
_get
_option
(key
)
774 def get_store(self
, key
: str) -> Optional
[str]:
776 Retrieve the value of a persistent KV store entry
779 :return: Byte string or None
781 return self
._ceph
_get
_store
(key
)
783 def get_localized_store(self
, key
: str, default
: Optional
[str] = None) -> Optional
[str]:
784 r
= self
._ceph
_get
_store
(_get_localized_key(self
.get_mgr_id(), key
))
786 r
= self
._ceph
_get
_store
(key
)
791 def get_active_uri(self
) -> str:
792 return self
._ceph
_get
_active
_uri
()
794 def get_localized_module_option(self
, key
: str, default
: OptionValue
= None) -> OptionValue
:
795 r
= self
._ceph
_get
_module
_option
(key
, self
.get_mgr_id())
797 return self
.MODULE_OPTION_DEFAULTS
.get(key
, default
)
802 HealthChecksT
= Mapping
[str, Mapping
[str, Union
[int, str, Sequence
[str]]]]
803 # {"type": service_type, "id": service_id}
804 ServiceInfoT
= Dict
[str, str]
805 # {"hostname": hostname,
806 # "ceph_version": version,
807 # "services": [service_info, ..]}
808 ServerInfoT
= Dict
[str, Union
[str, List
[ServiceInfoT
]]]
809 PerfCounterT
= Dict
[str, Any
]
812 class MgrModule(ceph_module
.BaseMgrModule
, MgrModuleLoggingMixin
):
813 COMMANDS
= [] # type: List[Any]
814 MODULE_OPTIONS
: List
[Option
] = []
815 MODULE_OPTION_DEFAULTS
= {} # type: Dict[str, Any]
817 # Priority definitions for perf counters
821 PRIO_UNINTERESTING
= 2
824 # counter value types
829 PERFCOUNTER_LONGRUNAVG
= 4
830 PERFCOUNTER_COUNTER
= 8
831 PERFCOUNTER_HISTOGRAM
= 0x10
832 PERFCOUNTER_TYPE_MASK
= ~
3
838 # Cluster log priorities
839 class ClusterLogPrio(IntEnum
):
846 def __init__(self
, module_name
: str, py_modules_ptr
: object, this_ptr
: object):
847 self
.module_name
= module_name
848 super(MgrModule
, self
).__init
__(py_modules_ptr
, this_ptr
)
850 for o
in self
.MODULE_OPTIONS
:
853 # we'll assume the declared type matches the
854 # supplied default value's type.
855 self
.MODULE_OPTION_DEFAULTS
[o
['name']] = o
['default']
857 # module not declaring it's type, so normalize the
858 # default value to be a string for consistent behavior
859 # with default and user-supplied option values.
860 self
.MODULE_OPTION_DEFAULTS
[o
['name']] = str(o
['default'])
862 mgr_level
= cast(str, self
.get_ceph_option("debug_mgr"))
863 log_level
= cast(str, self
.get_module_option("log_level"))
864 cluster_level
= cast(str, self
.get_module_option('log_to_cluster_level'))
865 log_to_file
= self
.get_module_option("log_to_file")
866 assert isinstance(log_to_file
, bool)
867 log_to_cluster
= self
.get_module_option("log_to_cluster")
868 assert isinstance(log_to_cluster
, bool)
869 self
._configure
_logging
(mgr_level
, log_level
, cluster_level
,
870 log_to_file
, log_to_cluster
)
872 # for backwards compatibility
873 self
._logger
= self
.getLogger()
875 self
._version
= self
._ceph
_get
_version
()
877 self
._perf
_schema
_cache
= None
879 # Keep a librados instance for those that need it.
880 self
._rados
: Optional
[rados
.Rados
] = None
882 def __del__(self
) -> None:
883 self
._unconfigure
_logging
()
886 def _register_options(cls
, module_name
: str) -> None:
887 cls
.MODULE_OPTIONS
.append(
888 Option(name
='log_level', type='str', default
="", runtime
=True,
889 enum_allowed
=['info', 'debug', 'critical', 'error',
891 cls
.MODULE_OPTIONS
.append(
892 Option(name
='log_to_file', type='bool', default
=False, runtime
=True))
893 if not [x
for x
in cls
.MODULE_OPTIONS
if x
['name'] == 'log_to_cluster']:
894 cls
.MODULE_OPTIONS
.append(
895 Option(name
='log_to_cluster', type='bool', default
=False,
897 cls
.MODULE_OPTIONS
.append(
898 Option(name
='log_to_cluster_level', type='str', default
='info',
900 enum_allowed
=['info', 'debug', 'critical', 'error',
904 def _register_commands(cls
, module_name
: str) -> None:
905 cls
.COMMANDS
.extend(CLICommand
.dump_cmd_list())
908 def log(self
) -> logging
.Logger
:
911 def cluster_log(self
, channel
: str, priority
: ClusterLogPrio
, message
: str) -> None:
913 :param channel: The log channel. This can be 'cluster', 'audit', ...
914 :param priority: The log message priority.
915 :param message: The message to log.
917 self
._ceph
_cluster
_log
(channel
, priority
.value
, message
)
920 def version(self
) -> str:
924 def release_name(self
) -> str:
926 Get the release name of the Ceph version, e.g. 'nautilus' or 'octopus'.
927 :return: Returns the release name of the Ceph version in lower case.
930 return self
._ceph
_get
_release
_name
()
932 def lookup_release_name(self
, major
: int) -> str:
933 return self
._ceph
_lookup
_release
_name
(major
)
935 def get_context(self
) -> object:
937 :return: a Python capsule containing a C++ CephContext pointer
939 return self
._ceph
_get
_context
()
941 def notify(self
, notify_type
: str, notify_id
: str) -> None:
943 Called by the ceph-mgr service to notify the Python plugin
944 that new state is available.
946 :param notify_type: string indicating what kind of notification,
947 such as osd_map, mon_map, fs_map, mon_status,
948 health, pg_summary, command, service_map
949 :param notify_id: string (may be empty) that optionally specifies
950 which entity is being notified about. With
951 "command" notifications this is set to the tag
952 ``from send_command``.
956 def _config_notify(self
) -> None:
957 # check logging options for changes
958 mgr_level
= cast(str, self
.get_ceph_option("debug_mgr"))
959 module_level
= cast(str, self
.get_module_option("log_level"))
960 cluster_level
= cast(str, self
.get_module_option("log_to_cluster_level"))
961 assert isinstance(cluster_level
, str)
962 log_to_file
= self
.get_module_option("log_to_file", False)
963 assert isinstance(log_to_file
, bool)
964 log_to_cluster
= self
.get_module_option("log_to_cluster", False)
965 assert isinstance(log_to_cluster
, bool)
966 self
._set
_log
_level
(mgr_level
, module_level
, cluster_level
)
968 if log_to_file
!= self
.log_to_file
:
970 self
._enable
_file
_log
()
972 self
._disable
_file
_log
()
973 if log_to_cluster
!= self
.log_to_cluster
:
975 self
._enable
_cluster
_log
()
977 self
._disable
_cluster
_log
()
979 # call module subclass implementations
982 def config_notify(self
) -> None:
984 Called by the ceph-mgr service to notify the Python plugin
985 that the configuration may have changed. Modules will want to
986 refresh any configuration values stored in config variables.
990 def serve(self
) -> None:
992 Called by the ceph-mgr service to start any server that
993 is provided by this Python plugin. The implementation
994 of this function should block until ``shutdown`` is called.
996 You *must* implement ``shutdown`` if you implement ``serve``
1000 def shutdown(self
) -> None:
1002 Called by the ceph-mgr service to request that this
1003 module drop out of its serve() function. You do not
1004 need to implement this if you do not implement serve()
1009 addrs
= self
._rados
.get_addrs()
1010 self
._rados
.shutdown()
1011 self
._ceph
_unregister
_client
(addrs
)
1013 def get(self
, data_name
: str):
1015 Called by the plugin to fetch named cluster-wide objects from ceph-mgr.
1017 :param str data_name: Valid things to fetch are osd_crush_map_text,
1018 osd_map, osd_map_tree, osd_map_crush, config, mon_map, fs_map,
1019 osd_metadata, pg_summary, io_rate, pg_dump, df, osd_stats,
1020 health, mon_status, devices, device <devid>, pg_stats,
1021 pool_stats, pg_ready, osd_ping_times.
1024 All these structures have their own JSON representations: experiment
1025 or look at the C++ ``dump()`` methods to learn about them.
1027 return self
._ceph
_get
(data_name
)
1029 def _stattype_to_str(self
, stattype
: int) -> str:
1031 typeonly
= stattype
& self
.PERFCOUNTER_TYPE_MASK
1034 if typeonly
== self
.PERFCOUNTER_LONGRUNAVG
:
1035 # this lie matches the DaemonState decoding: only val, no counts
1037 if typeonly
== self
.PERFCOUNTER_COUNTER
:
1039 if typeonly
== self
.PERFCOUNTER_HISTOGRAM
:
1044 def _perfpath_to_path_labels(self
, daemon
: str,
1045 path
: str) -> Tuple
[str, Tuple
[str, ...], Tuple
[str, ...]]:
1046 label_names
= ("ceph_daemon",) # type: Tuple[str, ...]
1047 labels
= (daemon
,) # type: Tuple[str, ...]
1049 if daemon
.startswith('rbd-mirror.'):
1051 r
'^rbd_mirror_image_([^/]+)/(?:(?:([^/]+)/)?)(.*)\.(replay(?:_bytes|_latency)?)$',
1055 path
= 'rbd_mirror_image_' + match
.group(4)
1056 pool
= match
.group(1)
1057 namespace
= match
.group(2) or ''
1058 image
= match
.group(3)
1059 label_names
+= ('pool', 'namespace', 'image')
1060 labels
+= (pool
, namespace
, image
)
1062 return path
, label_names
, labels
,
1064 def _perfvalue_to_value(self
, stattype
: int, value
: Union
[int, float]) -> Union
[float, int]:
1065 if stattype
& self
.PERFCOUNTER_TIME
:
1066 # Convert from ns to seconds
1067 return value
/ 1000000000.0
1071 def _unit_to_str(self
, unit
: int) -> str:
1072 if unit
== self
.NONE
:
1074 elif unit
== self
.BYTES
:
1077 raise ValueError(f
'bad unit "{unit}"')
1080 def to_pretty_iec(n
: int) -> str:
1081 for bits
, suffix
in [(60, 'Ei'), (50, 'Pi'), (40, 'Ti'), (30, 'Gi'),
1082 (20, 'Mi'), (10, 'Ki')]:
1084 return str(n
>> bits
) + ' ' + suffix
1088 def get_pretty_row(elems
: Sequence
[str], width
: int) -> str:
1090 Takes an array of elements and returns a string with those elements
1091 formatted as a table row. Useful for polling modules.
1093 :param elems: the elements to be printed
1094 :param width: the width of the terminal
1097 column_width
= int(width
/ n
)
1101 ret
+= '{0:>{w}} |'.format(elem
, w
=column_width
- 2)
1105 def get_pretty_header(self
, elems
: Sequence
[str], width
: int) -> str:
1107 Like ``get_pretty_row`` but adds dashes, to be used as a table title.
1109 :param elems: the elements to be printed
1110 :param width: the width of the terminal
1113 column_width
= int(width
/ n
)
1117 for i
in range(0, n
):
1118 ret
+= '-' * (column_width
- 1) + '+'
1122 ret
+= self
.get_pretty_row(elems
, width
)
1127 for i
in range(0, n
):
1128 ret
+= '-' * (column_width
- 1) + '+'
1133 def get_server(self
, hostname
) -> ServerInfoT
:
1135 Called by the plugin to fetch metadata about a particular hostname from
1138 This is information that ceph-mgr has gleaned from the daemon metadata
1139 reported by daemons running on a particular server.
1141 :param hostname: a hostname
1143 return cast(ServerInfoT
, self
._ceph
_get
_server
(hostname
))
1145 def get_perf_schema(self
,
1147 svc_name
: str) -> Dict
[str,
1148 Dict
[str, Dict
[str, Union
[str, int]]]]:
1150 Called by the plugin to fetch perf counter schema info.
1151 svc_name can be nullptr, as can svc_type, in which case
1154 :param str svc_type:
1155 :param str svc_name:
1156 :return: list of dicts describing the counters requested
1158 return self
._ceph
_get
_perf
_schema
(svc_type
, svc_name
)
1160 def get_counter(self
,
1163 path
: str) -> Dict
[str, List
[Tuple
[float, int]]]:
1165 Called by the plugin to fetch the latest performance counter data for a
1166 particular counter on a particular service.
1168 :param str svc_type:
1169 :param str svc_name:
1170 :param str path: a period-separated concatenation of the subsystem and the
1171 counter name, for example "mds.inodes".
1172 :return: A dict of counter names to their values. each value is a list of
1173 of two-tuples of (timestamp, value). This may be empty if no data is
1176 return self
._ceph
_get
_counter
(svc_type
, svc_name
, path
)
1178 def get_latest_counter(self
,
1181 path
: str) -> Dict
[str, Union
[Tuple
[float, int],
1182 Tuple
[float, int, int]]]:
1184 Called by the plugin to fetch only the newest performance counter data
1185 point for a particular counter on a particular service.
1187 :param str svc_type:
1188 :param str svc_name:
1189 :param str path: a period-separated concatenation of the subsystem and the
1190 counter name, for example "mds.inodes".
1191 :return: A list of two-tuples of (timestamp, value) or three-tuple of
1192 (timestamp, value, count) is returned. This may be empty if no
1195 return self
._ceph
_get
_latest
_counter
(svc_type
, svc_name
, path
)
1197 def list_servers(self
) -> List
[ServerInfoT
]:
1199 Like ``get_server``, but gives information about all servers (i.e. all
1200 unique hostnames that have been mentioned in daemon metadata)
1202 :return: a list of information about all servers
1205 return cast(List
[ServerInfoT
], self
._ceph
_get
_server
(None))
1207 def get_metadata(self
,
1210 default
: Optional
[Dict
[str, str]] = None) -> Optional
[Dict
[str, str]]:
1212 Fetch the daemon metadata for a particular service.
1214 ceph-mgr fetches metadata asynchronously, so are windows of time during
1215 addition/removal of services where the metadata is not available to
1216 modules. ``None`` is returned if no metadata is available.
1218 :param str svc_type: service type (e.g., 'mds', 'osd', 'mon')
1219 :param str svc_id: service id. convert OSD integer IDs to strings when
1221 :rtype: dict, or None if no metadata found
1223 metadata
= self
._ceph
_get
_metadata
(svc_type
, svc_id
)
1224 if metadata
is None:
1228 def get_daemon_status(self
, svc_type
: str, svc_id
: str) -> Dict
[str, str]:
1230 Fetch the latest status for a particular service daemon.
1232 This method may return ``None`` if no status information is
1233 available, for example because the daemon hasn't fully started yet.
1235 :param svc_type: string (e.g., 'rgw')
1236 :param svc_id: string
1237 :return: dict, or None if the service is not found
1239 return self
._ceph
_get
_daemon
_status
(svc_type
, svc_id
)
1241 def check_mon_command(self
, cmd_dict
: dict, inbuf
: Optional
[str] = None) -> HandleCommandResult
:
1243 Wrapper around :func:`~mgr_module.MgrModule.mon_command`, but raises,
1247 r
= HandleCommandResult(*self
.mon_command(cmd_dict
, inbuf
))
1249 raise MonCommandFailed(f
'{cmd_dict["prefix"]} failed: {r.stderr} retval: {r.retval}')
1252 def mon_command(self
, cmd_dict
: dict, inbuf
: Optional
[str] = None) -> Tuple
[int, str, str]:
1254 Helper for modules that do simple, synchronous mon command
1257 See send_command for general case.
1259 :return: status int, out std, err str
1263 result
= CommandResult()
1264 self
.send_command(result
, "mon", "", json
.dumps(cmd_dict
), "", inbuf
)
1268 self
.log
.debug("mon_command: '{0}' -> {1} in {2:.3f}s".format(
1269 cmd_dict
['prefix'], r
[0], t2
- t1
1276 result
: CommandResult
,
1281 inbuf
: Optional
[str] = None) -> None:
1283 Called by the plugin to send a command to the mon
1286 :param CommandResult result: an instance of the ``CommandResult``
1287 class, defined in the same module as MgrModule. This acts as a
1288 completion and stores the output of the command. Use
1289 ``CommandResult.wait()`` if you want to block on completion.
1290 :param str svc_type:
1292 :param str command: a JSON-serialized command. This uses the same
1293 format as the ceph command line, which is a dictionary of command
1294 arguments, with the extra ``prefix`` key containing the command
1295 name itself. Consult MonCommands.h for available commands and
1296 their expected arguments.
1297 :param str tag: used for nonblocking operation: when a command
1298 completes, the ``notify()`` callback on the MgrModule instance is
1299 triggered, with notify_type set to "command", and notify_id set to
1300 the tag of the command.
1301 :param str inbuf: input buffer for sending additional data.
1303 self
._ceph
_send
_command
(result
, svc_type
, svc_id
, command
, tag
, inbuf
)
1305 def set_health_checks(self
, checks
: HealthChecksT
) -> None:
1307 Set the module's current map of health checks. Argument is a
1308 dict of check names to info, in this form:
1314 'severity': 'warning', # or 'error'
1315 'summary': 'summary string',
1316 'count': 4, # quantify badness
1317 'detail': [ 'list', 'of', 'detail', 'strings' ],
1320 'severity': 'error',
1321 'summary': 'bars are bad',
1322 'detail': [ 'too hard' ],
1326 :param list: dict of health check dicts
1328 self
._ceph
_set
_health
_checks
(checks
)
1330 def _handle_command(self
,
1332 cmd
: Dict
[str, Any
]) -> Union
[HandleCommandResult
,
1333 Tuple
[int, str, str]]:
1334 if cmd
['prefix'] not in CLICommand
.COMMANDS
:
1335 return self
.handle_command(inbuf
, cmd
)
1337 return CLICommand
.COMMANDS
[cmd
['prefix']].call(self
, cmd
, inbuf
)
1339 def handle_command(self
,
1341 cmd
: Dict
[str, Any
]) -> Union
[HandleCommandResult
,
1342 Tuple
[int, str, str]]:
1344 Called by ceph-mgr to request the plugin to handle one
1345 of the commands that it declared in self.COMMANDS
1347 Return a status code, an output buffer, and an
1348 output string. The output buffer is for data results,
1349 the output string is for informative text.
1351 :param inbuf: content of any "-i <file>" supplied to ceph cli
1353 :param cmd: from Ceph's cmdmap_t
1356 :return: HandleCommandResult or a 3-tuple of (int, str, str)
1359 # Should never get called if they didn't declare
1361 raise NotImplementedError()
1363 def get_mgr_id(self
) -> str:
1365 Retrieve the name of the manager daemon where this plugin
1366 is currently being executed (i.e. the active manager).
1370 return self
._ceph
_get
_mgr
_id
()
1372 def get_ceph_option(self
, key
: str) -> OptionValue
:
1373 return self
._ceph
_get
_option
(key
)
1375 def get_foreign_ceph_option(self
, entity
: str, key
: str) -> OptionValue
:
1376 return self
._ceph
_get
_foreign
_option
(entity
, key
)
1378 def _validate_module_option(self
, key
: str) -> None:
1380 Helper: don't allow get/set config callers to
1381 access config options that they didn't declare
1384 if key
not in [o
['name'] for o
in self
.MODULE_OPTIONS
]:
1385 raise RuntimeError("Config option '{0}' is not in {1}.MODULE_OPTIONS".
1386 format(key
, self
.__class
__.__name
__))
1388 def _get_module_option(self
,
1390 default
: OptionValue
,
1391 localized_prefix
: str = "") -> OptionValue
:
1392 r
= self
._ceph
_get
_module
_option
(self
.module_name
, key
,
1395 return self
.MODULE_OPTION_DEFAULTS
.get(key
, default
)
1399 def get_module_option(self
, key
: str, default
: OptionValue
= None) -> OptionValue
:
1401 Retrieve the value of a persistent configuration setting
1403 self
._validate
_module
_option
(key
)
1404 return self
._get
_module
_option
(key
, default
)
1406 def get_module_option_ex(self
, module
: str,
1408 default
: OptionValue
= None) -> OptionValue
:
1410 Retrieve the value of a persistent configuration setting
1411 for the specified module.
1413 :param module: The name of the module, e.g. 'dashboard'
1415 :param key: The configuration key, e.g. 'server_addr'.
1416 :param default: The default value to use when the
1417 returned value is ``None``. Defaults to ``None``.
1419 if module
== self
.module_name
:
1420 self
._validate
_module
_option
(key
)
1421 r
= self
._ceph
_get
_module
_option
(module
, key
)
1422 return default
if r
is None else r
1424 def get_store_prefix(self
, key_prefix
: str) -> Dict
[str, str]:
1426 Retrieve a dict of KV store keys to values, where the keys
1427 have the given prefix
1429 :param str key_prefix:
1432 return self
._ceph
_get
_store
_prefix
(key_prefix
)
1434 def _set_localized(self
,
1437 setter
: Callable
[[str, Optional
[str]], None]) -> None:
1438 return setter(_get_localized_key(self
.get_mgr_id(), key
), val
)
1440 def get_localized_module_option(self
, key
: str, default
: OptionValue
= None) -> OptionValue
:
1442 Retrieve localized configuration for this ceph-mgr instance
1444 self
._validate
_module
_option
(key
)
1445 return self
._get
_module
_option
(key
, default
, self
.get_mgr_id())
1447 def _set_module_option(self
, key
: str, val
: Any
) -> None:
1448 return self
._ceph
_set
_module
_option
(self
.module_name
, key
,
1449 None if val
is None else str(val
))
1451 def set_module_option(self
, key
: str, val
: Any
) -> None:
1453 Set the value of a persistent configuration setting
1456 :type val: str | None
1458 self
._validate
_module
_option
(key
)
1459 return self
._set
_module
_option
(key
, val
)
1461 def set_module_option_ex(self
, module
: str, key
: str, val
: OptionValue
) -> None:
1463 Set the value of a persistent configuration setting
1464 for the specified module.
1470 if module
== self
.module_name
:
1471 self
._validate
_module
_option
(key
)
1472 return self
._ceph
_set
_module
_option
(module
, key
, str(val
))
1474 def set_localized_module_option(self
, key
: str, val
: Optional
[str]) -> None:
1476 Set localized configuration for this ceph-mgr instance
1481 self
._validate
_module
_option
(key
)
1482 return self
._set
_localized
(key
, val
, self
._set
_module
_option
)
1484 def set_store(self
, key
: str, val
: Optional
[str]) -> None:
1486 Set a value in this module's persistent key value store.
1487 If val is None, remove key from store
1489 self
._ceph
_set
_store
(key
, val
)
1491 def get_store(self
, key
: str, default
: Optional
[str] = None) -> Optional
[str]:
1493 Get a value from this module's persistent key value store
1495 r
= self
._ceph
_get
_store
(key
)
1501 def get_localized_store(self
, key
: str, default
: Optional
[str] = None) -> Optional
[str]:
1502 r
= self
._ceph
_get
_store
(_get_localized_key(self
.get_mgr_id(), key
))
1504 r
= self
._ceph
_get
_store
(key
)
1509 def set_localized_store(self
, key
: str, val
: Optional
[str]) -> None:
1510 return self
._set
_localized
(key
, val
, self
.set_store
)
1512 def self_test(self
) -> None:
1514 Run a self-test on the module. Override this function and implement
1515 a best as possible self-test for (automated) testing of the module
1517 Indicate any failures by raising an exception. This does not have
1518 to be pretty, it's mainly for picking up regressions during
1519 development, rather than use in the field.
1521 :return: None, or an advisory string for developer interest, such
1522 as a json dump of some state.
1526 def get_osdmap(self
) -> OSDMap
:
1528 Get a handle to an OSDMap. If epoch==0, get a handle for the latest
1532 return cast(OSDMap
, self
._ceph
_get
_osdmap
())
1534 def get_latest(self
, daemon_type
: str, daemon_name
: str, counter
: str) -> int:
1535 data
= self
.get_latest_counter(
1536 daemon_type
, daemon_name
, counter
)[counter
]
1542 def get_latest_avg(self
, daemon_type
: str, daemon_name
: str, counter
: str) -> Tuple
[int, int]:
1543 data
= self
.get_latest_counter(
1544 daemon_type
, daemon_name
, counter
)[counter
]
1546 # https://github.com/python/mypy/issues/1178
1547 _
, value
, count
= cast(Tuple
[float, int, int], data
)
1553 def get_all_perf_counters(self
, prio_limit
: int = PRIO_USEFUL
,
1554 services
: Sequence
[str] = ("mds", "mon", "osd",
1555 "rbd-mirror", "rgw",
1556 "tcmu-runner")) -> Dict
[str, dict]:
1558 Return the perf counters currently known to this ceph-mgr
1559 instance, filtered by priority equal to or greater than `prio_limit`.
1561 The result is a map of string to dict, associating services
1562 (like "osd.123") with their counters. The counter
1563 dict for each service maps counter paths to a counter
1564 info structure, which is the information from
1565 the schema, plus an additional "value" member with the latest
1569 result
= defaultdict(dict) # type: Dict[str, dict]
1571 for server
in self
.list_servers():
1572 for service
in cast(List
[ServiceInfoT
], server
['services']):
1573 if service
['type'] not in services
:
1576 schemas
= self
.get_perf_schema(service
['type'], service
['id'])
1578 self
.log
.warning("No perf counter schema for {0}.{1}".format(
1579 service
['type'], service
['id']
1583 # Value is returned in a potentially-multi-service format,
1584 # get just the service we're asking about
1585 svc_full_name
= "{0}.{1}".format(
1586 service
['type'], service
['id'])
1587 schema
= schemas
[svc_full_name
]
1589 # Populate latest values
1590 for counter_path
, counter_schema
in schema
.items():
1591 # self.log.debug("{0}: {1}".format(
1592 # counter_path, json.dumps(counter_schema)
1594 priority
= counter_schema
['priority']
1595 assert isinstance(priority
, int)
1596 if priority
< prio_limit
:
1599 tp
= counter_schema
['type']
1600 assert isinstance(tp
, int)
1601 counter_info
= dict(counter_schema
)
1602 # Also populate count for the long running avgs
1603 if tp
& self
.PERFCOUNTER_LONGRUNAVG
:
1604 v
, c
= self
.get_latest_avg(
1609 counter_info
['value'], counter_info
['count'] = v
, c
1610 result
[svc_full_name
][counter_path
] = counter_info
1612 counter_info
['value'] = self
.get_latest(
1618 result
[svc_full_name
][counter_path
] = counter_info
1620 self
.log
.debug("returning {0} counter".format(len(result
)))
1624 def set_uri(self
, uri
: str) -> None:
1626 If the module exposes a service, then call this to publish the
1627 address once it is available.
1631 return self
._ceph
_set
_uri
(uri
)
1633 def set_device_wear_level(self
, devid
: str, wear_level
: float) -> None:
1634 return self
._ceph
_set
_device
_wear
_level
(devid
, wear_level
)
1636 def have_mon_connection(self
) -> bool:
1638 Check whether this ceph-mgr daemon has an open connection
1639 to a monitor. If it doesn't, then it's likely that the
1640 information we have about the cluster is out of date,
1641 and/or the monitor cluster is down.
1644 return self
._ceph
_have
_mon
_connection
()
1646 def update_progress_event(self
,
1650 add_to_ceph_s
: bool) -> None:
1651 return self
._ceph
_update
_progress
_event
(evid
, desc
, progress
, add_to_ceph_s
)
1653 def complete_progress_event(self
, evid
: str) -> None:
1654 return self
._ceph
_complete
_progress
_event
(evid
)
1656 def clear_all_progress_events(self
) -> None:
1657 return self
._ceph
_clear
_all
_progress
_events
()
1660 def rados(self
) -> rados
.Rados
:
1662 A librados instance to be shared by any classes within
1663 this mgr module that want one.
1668 ctx_capsule
= self
.get_context()
1669 self
._rados
= rados
.Rados(context
=ctx_capsule
)
1670 self
._rados
.connect()
1671 self
._ceph
_register
_client
(self
._rados
.get_addrs())
1675 def can_run() -> Tuple
[bool, str]:
1677 Implement this function to report whether the module's dependencies
1678 are met. For example, if the module needs to import a particular
1679 dependency to work, then use a try/except around the import at
1680 file scope, and then report here if the import failed.
1682 This will be called in a blocking way from the C++ code, so do not
1683 do any I/O that could block in this function.
1685 :return a 2-tuple consisting of a boolean and explanatory string
1690 def remote(self
, module_name
: str, method_name
: str, *args
: Any
, **kwargs
: Any
) -> Any
:
1692 Invoke a method on another module. All arguments, and the return
1693 value from the other module must be serializable.
1695 Limitation: Do not import any modules within the called method.
1696 Otherwise you will get an error in Python 2::
1698 RuntimeError('cannot unmarshal code objects in restricted execution mode',)
1702 :param module_name: Name of other module. If module isn't loaded,
1703 an ImportError exception is raised.
1704 :param method_name: Method name. If it does not exist, a NameError
1705 exception is raised.
1706 :param args: Argument tuple
1707 :param kwargs: Keyword argument dict
1708 :raises RuntimeError: **Any** error raised within the method is converted to a RuntimeError
1709 :raises ImportError: No such module
1711 return self
._ceph
_dispatch
_remote
(module_name
, method_name
,
1714 def add_osd_perf_query(self
, query
: Dict
[str, Any
]) -> Optional
[int]:
1716 Register an OSD perf query. Argument is a
1717 dict of the query parameters, in this form:
1723 {'type': subkey_type, 'regex': regex_pattern},
1726 'performance_counter_descriptors': [
1727 list, of, descriptor, types
1729 'limit': {'order_by': performance_counter_type, 'max_count': n},
1733 'client_id', 'client_address', 'pool_id', 'namespace', 'osd_id',
1734 'pg_id', 'object_name', 'snap_id'
1735 Valid performance counter types:
1736 'ops', 'write_ops', 'read_ops', 'bytes', 'write_bytes', 'read_bytes',
1737 'latency', 'write_latency', 'read_latency'
1739 :param object query: query
1740 :rtype: int (query id)
1742 return self
._ceph
_add
_osd
_perf
_query
(query
)
1744 def remove_osd_perf_query(self
, query_id
: int) -> None:
1746 Unregister an OSD perf query.
1748 :param int query_id: query ID
1750 return self
._ceph
_remove
_osd
_perf
_query
(query_id
)
1752 def get_osd_perf_counters(self
, query_id
: int) -> Optional
[Dict
[str, List
[PerfCounterT
]]]:
1754 Get stats collected for an OSD perf query.
1756 :param int query_id: query ID
1758 return self
._ceph
_get
_osd
_perf
_counters
(query_id
)
1760 def add_mds_perf_query(self
, query
: Dict
[str, Any
]) -> Optional
[int]:
1762 Register an MDS perf query. Argument is a
1763 dict of the query parameters, in this form:
1769 {'type': subkey_type, 'regex': regex_pattern},
1772 'performance_counter_descriptors': [
1773 list, of, descriptor, types
1777 NOTE: 'limit' and 'order_by' are not supported (yet).
1780 'mds_rank', 'client_id'
1781 Valid performance counter types:
1784 :param object query: query
1785 :rtype: int (query id)
1787 return self
._ceph
_add
_mds
_perf
_query
(query
)
1789 def remove_mds_perf_query(self
, query_id
: int) -> None:
1791 Unregister an MDS perf query.
1793 :param int query_id: query ID
1795 return self
._ceph
_remove
_mds
_perf
_query
(query_id
)
1797 def get_mds_perf_counters(self
, query_id
: int) -> Optional
[Dict
[str, List
[PerfCounterT
]]]:
1799 Get stats collected for an MDS perf query.
1801 :param int query_id: query ID
1803 return self
._ceph
_get
_mds
_perf
_counters
(query_id
)
1805 def is_authorized(self
, arguments
: Dict
[str, str]) -> bool:
1807 Verifies that the current session caps permit executing the py service
1808 or current module with the provided arguments. This provides a generic
1809 way to allow modules to restrict by more fine-grained controls (e.g.
1812 :param arguments: dict of key/value arguments to test
1814 return self
._ceph
_is
_authorized
(arguments
)