kwargs[k] = CephArgtype.cast_to(tp, v)
return kwargs
- def call(self, mgr: Any, cmd_dict: Dict[str, Any], inbuf: Optional[str] = None) -> HandleCommandResult:
+ def call(self,
+ mgr: Any,
+ cmd_dict: Dict[str, Any],
+ inbuf: Optional[str] = None) -> HandleCommandResult:
kwargs = self._collect_args_by_argspec(cmd_dict)
if inbuf:
kwargs['inbuf'] = inbuf
return r
+HealthChecksT = Mapping[str, Mapping[str, Union[int, str, Sequence[str]]]]
+
+
class MgrModule(ceph_module.BaseMgrModule, MgrModuleLoggingMixin):
COMMANDS = [] # type: List[Any]
MODULE_OPTIONS: List[Option] = []
"""
self._ceph_send_command(result, svc_type, svc_id, command, tag, inbuf)
- def set_health_checks(self,
- checks: Mapping[str, Mapping[str, Union[int, str, Sequence[str]]]]) -> None:
+ def set_health_checks(self, checks: HealthChecksT) -> None:
"""
Set the module's current map of health checks. Argument is a
dict of check names to info, in this form:
return self._ceph_have_mon_connection()
- def update_progress_event(self, evid: str, desc: str, progress: float, add_to_ceph_s: bool) -> None:
+ def update_progress_event(self,
+ evid: str,
+ desc: str,
+ progress: float,
+ add_to_ceph_s: bool) -> None:
return self._ceph_update_progress_event(evid, desc, progress, add_to_ceph_s)
def complete_progress_event(self, evid: str) -> None: