gian/ceph-mirror
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2026-08-02 07:03:18 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

cephadm: clean up dbus_systemd.py and systemd.py per code review

Browse Source 
- dbus_systemd.py: top-level lazy dbus import (try/except ImportError),
  SystemdManager.get() classmethod as singleton factory with ImportError
  guard when python3-dbus absent, retry_connection decorator reconnects
  only on org.freedesktop.DBus.Error.Disconnected (not all DBusExceptions),
  use get_dbus_name() for error classification, removed redundant
  per-method docstrings and section headers
- systemd.py: drop ABC and subprocess backend entirely per reviewer
  request, import SystemdManager directly at module top, use dict
  _ACTIVE_STATE_MAP for state translation instead of if/elif chain,
  rename local var to sysd_mgr, per-step error handling in
  terminate_service and enable_service, preserve original log message
  wording for maintenance-exit path

Agent-Logs-Url: https://github.com/ceph/ceph/sessions/a975d2eb-63ac-4521-9e32-ea96c8db792b

Co-authored-by: epuertat <37327689+epuertat@users.noreply.github.com>
This commit is contained in:
anthropic-code-agent[bot] 2026-07-29 10:18:43 +00:00 committed by GitHub
parent bbb60b95d3
commit a795ec8bd5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 115 additions and 306 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

168
src/cephadm/cephadmlib/dbus_systemd.py
View File

@ -1,174 +1,120 @@
# dbus_systemd.py - D-Bus interface to the systemd manager
#
# Provides a class-based wrapper around the systemd1 D-Bus API so that
# cephadm can query and control host systemd units without invoking the
# 'systemctl' binary as a subprocess. Using D-Bus instead of subprocess
# removes the need for --pid=host and avoids mounting the systemd namespace
# inside the container.
#
# The interface follows the systemd D-Bus API specification:
# https://www.freedesktop.org/software/systemd/man/latest/org.freedesktop.systemd1.html
#
# Requires: python3-dbus (dbus-python)
"""D-Bus interface to the org.freedesktop.systemd1 manager."""
import functools
import logging
from typing import Any, Callable, Optional, TypeVar
from typing import Callable, Optional, TypeVar
logger = logging.getLogger()
# D-Bus constants for the systemd1 manager object.
# python3-dbus (dbus-python) is an optional runtime dependency.
# All code that uses the `dbus` name is guarded via SystemdManager.get(),
# which raises ImportError early when the package is absent.
try:
import dbus
import dbus.exceptions
_DBUS_AVAILABLE = True
except ImportError:
_DBUS_AVAILABLE = False
_SYSTEMD_BUS_NAME = 'org.freedesktop.systemd1'
_SYSTEMD_OBJECT_PATH = '/org/freedesktop/systemd1'
_MANAGER_IFACE = 'org.freedesktop.systemd1.Manager'
_UNIT_IFACE = 'org.freedesktop.systemd1.Unit'
_PROPS_IFACE = 'org.freedesktop.DBus.Properties'
_F = TypeVar('_F', bound=Callable[..., Any])
_DISCONNECTED_ERROR = 'org.freedesktop.DBus.Error.Disconnected'
_R = TypeVar('_R')
def _dbus_method(fn: _F) -> _F:
"""Decorator that retries a SystemdManager method once on a stale connection.
If the decorated method raises ``dbus.DBusException`` on the first call,
the shared connection is reset and the method is retried. A second
failure propagates to the caller unchanged.
"""
def retry_connection(fn: Callable[..., _R]) -> Callable[..., _R]:
"""Reconnect and retry once if the D-Bus transport is disconnected."""
@functools.wraps(fn)
def wrapper(self: 'SystemdManager', *args: Any, **kwargs: Any) -> Any:
import dbus # noqa: PLC0415
def wrapper(self: 'SystemdManager', *args: object, **kwargs: object) -> _R:
try:
return fn(self, *args, **kwargs)
except dbus.exceptions.DBusException:
self._reconnect()
except Exception as exc:
# Only reconnect for transport-level disconnections, not for
# application-level errors like NoSuchUnit.
dbus_name = getattr(exc, 'get_dbus_name', lambda: '')()
if dbus_name != _DISCONNECTED_ERROR:
raise
self._connect()
return fn(self, *args, **kwargs)
return wrapper # type: ignore[return-value]
class SystemdManager:
"""Singleton gateway to the org.freedesktop.systemd1 D-Bus service.
"""Singleton gateway to the org.freedesktop.systemd1 D-Bus service."""
Use the module-level :func:`get_manager` factory to obtain the shared
instance; do not instantiate this class directly.
All public methods map directly to the corresponding systemd Manager or
Unit D-Bus methods. Each call retries once on a stale connection via the
``@_dbus_method`` decorator.
"""
_instance: Optional['SystemdManager'] = None
def __init__(self) -> None:
self._connect()
@classmethod
def get(cls) -> 'SystemdManager':
if not _DBUS_AVAILABLE:
raise ImportError(
'python3-dbus is required for D-Bus systemd integration'
)
if cls._instance is None:
cls._instance = cls()
return cls._instance
def _connect(self) -> None:
import dbus # noqa: PLC0415
self._bus = dbus.SystemBus()
self._bus = dbus.SystemBus() # type: ignore[name-defined]
systemd_obj = self._bus.get_object(_SYSTEMD_BUS_NAME, _SYSTEMD_OBJECT_PATH)
self._manager = dbus.Interface(systemd_obj, dbus_interface=_MANAGER_IFACE)
self._manager = dbus.Interface(systemd_obj, dbus_interface=_MANAGER_IFACE) # type: ignore[name-defined]
def _reconnect(self) -> None:
logger.debug('D-Bus connection stale; reconnecting to systemd1')
self._connect()
# ------------------------------------------------------------------
# Manager-level operations
# ------------------------------------------------------------------
@_dbus_method
@retry_connection
def daemon_reload(self) -> None:
"""Equivalent to: systemctl daemon-reload"""
self._manager.Reload()
# ------------------------------------------------------------------
# Unit lifecycle operations
# ------------------------------------------------------------------
@_dbus_method
@retry_connection
def start_unit(self, unit_name: str, mode: str = 'replace') -> None:
"""Equivalent to: systemctl start <unit_name>"""
self._manager.StartUnit(unit_name, mode)
@_dbus_method
@retry_connection
def stop_unit(self, unit_name: str, mode: str = 'replace') -> None:
"""Equivalent to: systemctl stop <unit_name>"""
self._manager.StopUnit(unit_name, mode)
@_dbus_method
@retry_connection
def restart_unit(self, unit_name: str, mode: str = 'replace') -> None:
"""Equivalent to: systemctl restart <unit_name>"""
self._manager.RestartUnit(unit_name, mode)
@_dbus_method
@retry_connection
def reset_failed_unit(self, unit_name: str) -> None:
"""Equivalent to: systemctl reset-failed <unit_name>"""
self._manager.ResetFailedUnit(unit_name)
# ------------------------------------------------------------------
# Unit file enablement
# ------------------------------------------------------------------
@_dbus_method
@retry_connection
def enable_unit_file(self, unit_name: str) -> None:
"""Equivalent to: systemctl enable <unit_name>"""
self._manager.EnableUnitFiles([unit_name], False, True)
self._manager.Reload()
@_dbus_method
@retry_connection
def disable_unit_file(self, unit_name: str) -> None:
"""Equivalent to: systemctl disable <unit_name>"""
self._manager.DisableUnitFiles([unit_name], False)
self._manager.Reload()
# ------------------------------------------------------------------
# Unit state queries
# ------------------------------------------------------------------
@_dbus_method
@retry_connection
def is_active(self, unit_name: str) -> str:
"""Return the ActiveState for *unit_name*.
Uses ``LoadUnit`` so that stopped-but-installed units return
'inactive' rather than raising a DBusException.
"""
import dbus # noqa: PLC0415
"""Return the ActiveState for *unit_name* using LoadUnit so stopped units return 'inactive'."""
unit_path = self._manager.LoadUnit(unit_name)
unit_obj = self._bus.get_object(_SYSTEMD_BUS_NAME, unit_path)
props = dbus.Interface(unit_obj, dbus_interface=_PROPS_IFACE)
props = dbus.Interface(unit_obj, dbus_interface=_PROPS_IFACE) # type: ignore[name-defined]
return str(props.Get(_UNIT_IFACE, 'ActiveState'))
@_dbus_method
@retry_connection
def is_enabled(self, unit_name: str) -> Optional[str]:
"""Return the UnitFileState for *unit_name*, or None if not found.
``GetUnitFileState`` raises ``org.freedesktop.systemd1.NoSuchUnit``
when the unit file does not exist. That error is treated as "not
installed" and returns None. Any other DBusException (stale
connection, etc.) propagates so the ``@_dbus_method`` decorator can
retry.
"""
import dbus # noqa: PLC0415
"""Return the UnitFileState for *unit_name*, or None if the unit file does not exist."""
try:
return str(self._manager.GetUnitFileState(unit_name))
except dbus.exceptions.DBusException as exc:
if 'NoSuchUnit' in str(exc) or 'org.freedesktop.DBus.Error.FileNotFound' in str(exc):
except Exception as exc:
name = getattr(exc, 'get_dbus_name', lambda: '')()
if name in (
'org.freedesktop.systemd1.NoSuchUnit',
'org.freedesktop.DBus.Error.FileNotFound',
):
return None
raise
# ---------------------------------------------------------------------------
# Module-level singleton factory
# ---------------------------------------------------------------------------
_manager_instance: Optional[SystemdManager] = None
def get_manager() -> SystemdManager:
"""Return the process-wide :class:`SystemdManager` singleton.
The instance is created on the first call and reused thereafter.
Raises ``ImportError`` if python3-dbus is not installed, or
``dbus.DBusException`` if the system bus is unreachable.
"""
global _manager_instance
if _manager_instance is None:
_manager_instance = SystemdManager()
return _manager_instance

253
src/cephadm/cephadmlib/systemd.py
View File

@ -1,209 +1,48 @@
# systemd.py - general systemd related types and funcs
import abc
import logging
import functools
from typing import Optional, Tuple, List
from typing import Tuple, List
from .constants import DISABLED_SERVICES
from .context import CephadmContext
from .call_wrappers import call, CallVerbosity
from .dbus_systemd import SystemdManager
from .listing import DaemonEntry, daemons_matching
logger = logging.getLogger()
# ---------------------------------------------------------------------------
# Backend ABC and implementations
# ---------------------------------------------------------------------------
# D-Bus ActiveState -> cephadm state string
_ACTIVE_STATE_MAP = {
'active': 'running',
'inactive': 'stopped',
'failed': 'error',
'maintenance': 'error',
}
class _SystemdBackend(abc.ABC):
"""Abstract interface for systemd operations."""
@abc.abstractmethod
def check_unit(self, ctx: CephadmContext, unit_name: str) -> Tuple[bool, str, bool]:
"""Return (enabled, state, installed) for the given unit."""
@abc.abstractmethod
def enable_service(self, ctx: CephadmContext, service_name: str) -> None:
"""Enable and start the service."""
@abc.abstractmethod
def terminate_service(self, ctx: CephadmContext, service_name: str) -> None:
"""Stop, reset-failed, and disable the service."""
@abc.abstractmethod
def start_unit(self, ctx: CephadmContext, unit_name: str) -> None:
"""Start the unit."""
class _SubprocessBackend(_SystemdBackend):
"""Systemd backend that shells out to the systemctl binary."""
def _systemctl(
self, ctx: CephadmContext, args: List[str]
) -> Tuple[str, str, int]:
return call(ctx, ['systemctl'] + args, verbosity=CallVerbosity.QUIET)
def _systemctl_debug(
self, ctx: CephadmContext, args: List[str]
) -> Tuple[str, str, int]:
return call(ctx, ['systemctl'] + args, verbosity=CallVerbosity.DEBUG)
def check_unit(
self, ctx: CephadmContext, unit_name: str
) -> Tuple[bool, str, bool]:
enabled = False
installed = False
try:
out, err, code = self._systemctl(ctx, ['is-enabled', unit_name])
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)
state = 'unknown'
try:
out, err, code = self._systemctl(ctx, ['is-active', unit_name])
out = out.strip()
if out in ['active']:
state = 'running'
elif out in ['inactive']:
state = 'stopped'
elif out in ['failed', 'auto-restart']:
state = 'error'
except Exception as e:
logger.warning('unable to run systemctl: %s' % e)
return (enabled, state, installed)
def enable_service(
self, ctx: CephadmContext, service_name: str
) -> None:
self._systemctl_debug(ctx, ['enable', '--now', service_name])
def terminate_service(
self, ctx: CephadmContext, service_name: str
) -> None:
self._systemctl_debug(ctx, ['stop', service_name])
self._systemctl_debug(ctx, ['reset-failed', service_name])
self._systemctl_debug(ctx, ['disable', service_name])
def start_unit(self, ctx: CephadmContext, unit_name: str) -> None:
out, err, code = self._systemctl_debug(ctx, ['start', unit_name])
if code:
logger.warning(
'Failed to start %s: %s', unit_name, err
)
else:
logger.info('Started %s', unit_name)
class _DbusBackend(_SystemdBackend):
"""Systemd backend that communicates via D-Bus (python3-dbus)."""
def check_unit(
self, ctx: CephadmContext, unit_name: str
) -> Tuple[bool, str, bool]:
from . import dbus_systemd # noqa: PLC0415
mgr = dbus_systemd.get_manager()
enabled = False
installed = False
try:
unit_file_state = mgr.is_enabled(unit_name)
if unit_file_state is not None:
installed = True
if unit_file_state in ('enabled', 'enabled-runtime', 'static'):
enabled = True
except Exception as e:
logger.warning('D-Bus is-enabled query failed for %s: %s', unit_name, e)
state = 'unknown'
try:
active_state = mgr.is_active(unit_name)
if active_state == 'active':
state = 'running'
elif active_state == 'inactive':
state = 'stopped'
elif active_state in ('failed', 'maintenance'):
state = 'error'
except Exception as e:
logger.warning('D-Bus is-active query failed for %s: %s', unit_name, e)
return (enabled, state, installed)
def enable_service(
self, ctx: CephadmContext, service_name: str
) -> None:
from . import dbus_systemd # noqa: PLC0415
mgr = dbus_systemd.get_manager()
try:
mgr.enable_unit_file(service_name)
except Exception as e:
logger.warning('D-Bus enable failed for %s: %s', service_name, e)
return
try:
mgr.start_unit(service_name)
except Exception as e:
logger.warning('D-Bus start failed for %s: %s', service_name, e)
def terminate_service(
self, ctx: CephadmContext, service_name: str
) -> None:
from . import dbus_systemd # noqa: PLC0415
mgr = dbus_systemd.get_manager()
for op, method in [
('stop', mgr.stop_unit),
('reset-failed', mgr.reset_failed_unit),
('disable', mgr.disable_unit_file),
]:
try:
method(service_name)
except Exception as e:
logger.warning('D-Bus %s failed for %s: %s', op, service_name, e)
def start_unit(self, ctx: CephadmContext, unit_name: str) -> None:
from . import dbus_systemd # noqa: PLC0415
mgr = dbus_systemd.get_manager()
try:
mgr.start_unit(unit_name)
logger.info('Started %s', unit_name)
except Exception as e:
logger.warning('Failed to start %s: %s', unit_name, e)
# ---------------------------------------------------------------------------
# Backend selection (once at import/startup)
# ---------------------------------------------------------------------------
@functools.lru_cache(maxsize=None)
def _get_backend() -> _SystemdBackend:
"""Return the D-Bus backend if available, otherwise subprocess."""
try:
import dbus # noqa: PLC0415
dbus.SystemBus()
return _DbusBackend()
except Exception:
return _SubprocessBackend()
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
# D-Bus UnitFileState values that count as "enabled"
_ENABLED_STATES = {'enabled', 'enabled-runtime', 'static'}
def check_unit(ctx: CephadmContext, unit_name: str) -> Tuple[bool, str, bool]:
"""Return (enabled, state, installed) for the given systemd unit."""
return _get_backend().check_unit(ctx, unit_name)
sysd_mgr = SystemdManager.get()
enabled = False
installed = False
try:
unit_file_state = sysd_mgr.is_enabled(unit_name)
if unit_file_state is not None:
installed = True
if unit_file_state in _ENABLED_STATES:
enabled = True
except Exception as e:
logger.warning('unable to query systemd for %s: %s', unit_name, e)
state = 'unknown'
try:
state = _ACTIVE_STATE_MAP.get(sysd_mgr.is_active(unit_name), 'unknown')
except Exception as e:
logger.warning('unable to query systemd for %s: %s', unit_name, e)
return (enabled, state, installed)
def check_units(ctx: CephadmContext, units: List[str]) -> bool:
@ -219,12 +58,30 @@ def check_units(ctx: CephadmContext, units: List[str]) -> bool:
def terminate_service(ctx: CephadmContext, service_name: str) -> None:
_get_backend().terminate_service(ctx, service_name)
sysd_mgr = SystemdManager.get()
for op, method in [
('stop', sysd_mgr.stop_unit),
('reset-failed', sysd_mgr.reset_failed_unit),
('disable', sysd_mgr.disable_unit_file),
]:
try:
method(service_name)
except Exception as e:
logger.warning('unable to %s %s: %s', op, service_name, e)
def enable_service(ctx: CephadmContext, service_name: str) -> None:
"""Start and enable the service (typically using systemd)."""
_get_backend().enable_service(ctx, service_name)
sysd_mgr = SystemdManager.get()
try:
sysd_mgr.enable_unit_file(service_name)
except Exception as e:
logger.warning('unable to enable %s: %s', service_name, e)
return
try:
sysd_mgr.start_unit(service_name)
except Exception as e:
logger.warning('unable to start %s: %s', service_name, e)
def start_disabled_services_after_maintenance_exit(
@ -233,7 +90,7 @@ def start_disabled_services_after_maintenance_exit(
"""Start nfs/keepalived units after host-maintenance exit."""
if not ctx.fsid:
return
backend = _get_backend()
sysd_mgr = SystemdManager.get()
for daemon_type in DISABLED_SERVICES:
for entry in daemons_matching(
ctx, fsid=ctx.fsid, daemon_type=daemon_type
@ -242,4 +99,10 @@ def start_disabled_services_after_maintenance_exit(
unit = entry.identity.unit_name
else:
unit = entry.status['systemd_unit']
backend.start_unit(ctx, unit)
try:
sysd_mgr.start_unit(unit)
logger.info('Started %s after maintenance exit', unit)
except Exception as e:
logger.warning(
'Failed to start %s after maintenance exit: %s', unit, e
)