From aa01e9d0b10caeeb236edd57c4bea8ae2cabdf50 Mon Sep 17 00:00:00 2001 From: John Mulligan Date: Tue, 12 Sep 2023 14:11:37 -0400 Subject: [PATCH] cephadm: move a pair of systemd unit status funcs to systemd.py Signed-off-by: John Mulligan --- src/cephadm/cephadm.py | 54 +---------------------- src/cephadm/cephadmlib/systemd.py | 64 ++++++++++++++++++++++++++++ src/cephadm/tests/test_util_funcs.py | 22 +++++----- 3 files changed, 77 insertions(+), 63 deletions(-) create mode 100644 src/cephadm/cephadmlib/systemd.py diff --git a/src/cephadm/cephadm.py b/src/cephadm/cephadm.py index adcaa8ebe4b..de3b3fd0899 100755 --- a/src/cephadm/cephadm.py +++ b/src/cephadm/cephadm.py @@ -138,6 +138,7 @@ from cephadmlib.locking import FileLock from cephadmlib.daemon_identity import DaemonIdentity, DaemonSubIdentity from cephadmlib.packagers import create_packager, Packager from cephadmlib.logging import cephadm_init_logging, Highlight, LogDestination +from cephadmlib.systemd import check_unit, check_units FuncT = TypeVar('FuncT', bound=Callable) @@ -1905,59 +1906,6 @@ def get_unit_name_by_daemon_name(ctx: CephadmContext, fsid: str, name: str) -> s raise Error('Failed to get unit name for {}'.format(daemon)) -def check_unit(ctx, unit_name): - # type: (CephadmContext, str) -> Tuple[bool, str, bool] - # NOTE: we ignore the exit code here because systemctl outputs - # various exit codes based on the state of the service, but the - # string result is more explicit (and sufficient). - enabled = False - installed = False - try: - out, err, code = call(ctx, ['systemctl', 'is-enabled', unit_name], - verbosity=CallVerbosity.QUIET) - if code == 0: - enabled = True - installed = True - elif 'disabled' in out: - installed = True - except Exception as e: - logger.warning('unable to run systemctl: %s' % e) - enabled = False - installed = False - - state = 'unknown' - try: - out, err, code = call(ctx, ['systemctl', 'is-active', unit_name], - verbosity=CallVerbosity.QUIET) - out = out.strip() - if out in ['active']: - state = 'running' - elif out in ['inactive']: - state = 'stopped' - elif out in ['failed', 'auto-restart']: - state = 'error' - else: - state = 'unknown' - except Exception as e: - logger.warning('unable to run systemctl: %s' % e) - state = 'unknown' - return (enabled, state, installed) - - -def check_units(ctx, units, enabler=None): - # type: (CephadmContext, List[str], Optional[Packager]) -> bool - for u in units: - (enabled, state, installed) = check_unit(ctx, u) - if enabled and state == 'running': - logger.info('Unit %s is enabled and running' % u) - return True - if enabler is not None: - if installed: - logger.info('Enabling unit %s' % u) - enabler.enable_service(u) - return False - - def is_container_running(ctx: CephadmContext, c: 'CephContainer') -> bool: if ctx.name.split('.', 1)[0] in ['agent', 'cephadm-exporter']: # these are non-containerized daemon types diff --git a/src/cephadm/cephadmlib/systemd.py b/src/cephadm/cephadmlib/systemd.py new file mode 100644 index 00000000000..e48f1ddb0f8 --- /dev/null +++ b/src/cephadm/cephadmlib/systemd.py @@ -0,0 +1,64 @@ +# systemd.py - general systemd related types and funcs + +import logging + +from typing import Tuple, List, Optional + +from .context import CephadmContext +from .call_wrappers import call, CallVerbosity +from .packagers import Packager + +logger = logging.getLogger() + + +def check_unit(ctx, unit_name): + # type: (CephadmContext, str) -> Tuple[bool, str, bool] + # NOTE: we ignore the exit code here because systemctl outputs + # various exit codes based on the state of the service, but the + # string result is more explicit (and sufficient). + enabled = False + installed = False + try: + out, err, code = call(ctx, ['systemctl', 'is-enabled', unit_name], + verbosity=CallVerbosity.QUIET) + if code == 0: + enabled = True + installed = True + elif 'disabled' in out: + installed = True + except Exception as e: + logger.warning('unable to run systemctl: %s' % e) + enabled = False + installed = False + + state = 'unknown' + try: + out, err, code = call(ctx, ['systemctl', 'is-active', unit_name], + verbosity=CallVerbosity.QUIET) + out = out.strip() + if out in ['active']: + state = 'running' + elif out in ['inactive']: + state = 'stopped' + elif out in ['failed', 'auto-restart']: + state = 'error' + else: + state = 'unknown' + except Exception as e: + logger.warning('unable to run systemctl: %s' % e) + state = 'unknown' + return (enabled, state, installed) + + +def check_units(ctx, units, enabler=None): + # type: (CephadmContext, List[str], Optional[Packager]) -> bool + for u in units: + (enabled, state, installed) = check_unit(ctx, u) + if enabled and state == 'running': + logger.info('Unit %s is enabled and running' % u) + return True + if enabler is not None: + if installed: + logger.info('Enabling unit %s' % u) + enabler.enable_service(u) + return False diff --git a/src/cephadm/tests/test_util_funcs.py b/src/cephadm/tests/test_util_funcs.py index 6f71dd2d429..ffcf3909c4e 100644 --- a/src/cephadm/tests/test_util_funcs.py +++ b/src/cephadm/tests/test_util_funcs.py @@ -411,11 +411,12 @@ def _mk_fake_call(enabled, active): ) def test_check_unit(enabled_out, active_out, expected): with with_cephadm_ctx([]) as ctx: - _cephadm.call.side_effect = _mk_fake_call( - enabled=enabled_out, - active=active_out, - ) - enabled, state, installed = _cephadm.check_unit(ctx, "foobar") + with mock.patch('cephadmlib.systemd.call') as _call: + _call.side_effect = _mk_fake_call( + enabled=enabled_out, + active=active_out, + ) + enabled, state, installed = _cephadm.check_unit(ctx, "foobar") assert (enabled, state, installed) == expected @@ -489,11 +490,12 @@ def test_check_time_sync(call_fn, enabler, expected): is enabled. It is also the only consumer of check_units. """ with with_cephadm_ctx([]) as ctx: - _cephadm.call.side_effect = call_fn - result = _cephadm.check_time_sync(ctx, enabler=enabler) - assert result == expected - if enabler is not None: - enabler.check_expected() + with mock.patch('cephadmlib.systemd.call') as _call: + _call.side_effect = call_fn + result = _cephadm.check_time_sync(ctx, enabler=enabler) + assert result == expected + if enabler is not None: + enabler.check_expected() @pytest.mark.parametrize(