feat(cli): status queries systemd for every decnet-* unit

'decnet status' used to psutil-scan for cmdlines matching hand-coded
service launch args. That worked on dev boxes running workers via
'python -m decnet.cli ...' but missed the systemd reality on real
hosts: units may be installed but not started, failed, or in
auto-restart — all invisible to a cmdline grep.

New behaviour: status calls `systemctl list-units --type=service --all
--output=json 'decnet-*.service'` and renders the unit/load/active/
sub/description matrix. One view works for masters, agents, and
mixed hosts — iterates over whatever 'decnet-*' units were installed
by 'decnet init' / the enroll-bundle. Agent/master mode filtering is
no longer needed in the CLI; the host literally does not have
master-only units installed if it enrolled as an agent.

The psutil path survives as a fallback for boxes without systemd
(dev laptops, CI containers, minimal init systems) so the command
stays useful there. Clearly labelled 'psutil fallback' in the table
title so operators know which view they're looking at.
This commit is contained in:
2026-04-24 00:23:00 -04:00
parent d61e143b71
commit cb692d570a
2 changed files with 92 additions and 2 deletions

View File

@@ -134,6 +134,46 @@ def _service_registry(log_file: str) -> list[tuple[str, callable, list[str]]]:
]
def _systemd_units(pattern: str = "decnet-*.service") -> list[dict] | None:
"""Return state of every systemd unit matching *pattern*, or ``None``
when systemctl is unavailable (non-systemd host, container lab,
PATH-stripped env, user-manager unreachable).
Output shape mirrors ``systemctl list-units --output=json``: each
dict has ``unit``, ``load``, ``active``, ``sub``, ``description``.
Empty list = systemd works but no matching units are loaded (fresh
host that never ran ``decnet init``).
"""
import json # local import — avoids paying it on every CLI startup
import shutil
if not shutil.which("systemctl"):
return None
try:
proc = subprocess.run( # nosec B603 B607 — fixed argv, no shell
[
"systemctl", "list-units",
"--type=service", "--all",
"--no-legend", "--no-pager",
"--output=json",
pattern,
],
capture_output=True,
text=True,
timeout=5,
check=False,
)
except (OSError, subprocess.SubprocessError):
return None
if proc.returncode != 0:
return None
try:
data = json.loads(proc.stdout or "[]")
except json.JSONDecodeError:
return None
return data if isinstance(data, list) else None
def _kill_all_services() -> None:
"""Find and kill all running DECNET microservice processes."""
registry = _service_registry(str(DECNET_INGEST_LOG_FILE))