"""Init system abstraction for Fenris monitoring (issue #84). Detects the active init system (systemd or runit) and provides a unified interface for timer/collection control and service-state queries. This keeps the privileged helper and status composition init-system agnostic without introducing a generalized plugin framework. Design: ADR 0008 — keep service-specific operations behind a cohesive responsibility shared by the privileged control path and read-only status composition. Runit service layout: /etc/sv/fenris-collect/run — scheduler (sleep 120; loop { collect; sleep 300 }) /etc/sv/fenris-collect/log/run — logger to /var/log/fenris-collect/ /var/service/fenris-collect — symlink to enable /etc/sv/fenris-collect/down — marker for dormant install Runit guarantees: - Completion-relative 5-minute cadence (sleep 300 after each collect) - Initial 2-minute boot delay (sleep 120 before first collect) - Bounded execution (90s timeout via timeout(1)) - No catch-up (service sleeps fixed interval, no Persistent= flag) - Serialized scheduled runs (runsv does not restart until exit) - Serialized on-demand (flock serializes fenris-collect execution) Spec: §8.4, §8.5, §8.6, §8.7, §8.8, ADR 0008 """ import os import shutil import subprocess import sys from enum import Enum from pathlib import Path from typing import Any, Dict, Optional # --------------------------------------------------------------------------- # Init system detection # --------------------------------------------------------------------------- class InitSystem(Enum): SYSTEMD = "systemd" RUNIT = "runit" FENRIS_SV_DIR = Path("/etc/sv/fenris-collect") FENRIS_SERVICE_LINK = Path("/var/service/fenris-collect") FENRIS_LOG_DIR = Path("/var/log/fenris-collect") COLLECT_TIMEOUT_S = 90 def detect_init_system() -> InitSystem: """Detect the active init system. Checks for systemd first (PID 1 is systemd or /run/systemd/system exists), then falls back to runit (PID 1 is runsv or /etc/sv exists). """ # systemd detection: /run/systemd/system exists when systemd is PID 1 if Path("/run/systemd/system").exists(): return InitSystem.SYSTEMD # Check PID 1 name try: pid1_comm = Path("/proc/1/comm").read_text().strip() if pid1_comm == "systemd": return InitSystem.SYSTEMD if pid1_comm in ("runsv", "runsvdir"): return InitSystem.RUNIT except OSError: pass # Fallback: check for /etc/sv (Void Linux default) if Path("/etc/sv").is_dir(): return InitSystem.RUNIT # Default to systemd (existing behavior) return InitSystem.SYSTEMD def get_init_system() -> InitSystem: """Get the detected init system (cached).""" if not hasattr(get_init_system, "_cached"): get_init_system._cached = detect_init_system() return get_init_system._cached def reset_init_system_cache() -> None: """Reset the cached init system detection (for testing).""" if hasattr(get_init_system, "_cached"): delattr(get_init_system, "_cached") # --------------------------------------------------------------------------- # systemd backend # --------------------------------------------------------------------------- def _systemd_enable(now: bool) -> None: """Enable and optionally start the systemd timer.""" cmd = ["systemctl", "enable"] if now: cmd.append("--now") cmd.append("fenris-collect.timer") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print("Error enabling timer:", result.stderr, file=sys.stderr) sys.exit(1) print("Timer enabled" + (" and started" if now else "")) def _systemd_disable(now: bool) -> None: """Disable and optionally stop the systemd timer.""" cmd = ["systemctl", "disable"] if now: cmd.append("--now") cmd.append("fenris-collect.timer") result = subprocess.run(cmd, capture_output=True, text=True) if result.returncode != 0: print("Error disabling timer:", result.stderr, file=sys.stderr) sys.exit(1) print("Timer disabled" + (" and stopped" if now else "")) def _systemd_collect() -> None: """Trigger on-demand collection via systemd (blocking).""" result = subprocess.run( ["systemctl", "start", "fenris-collect.service"], capture_output=True, text=True, ) if result.returncode == 0: print("Collection completed successfully") else: print("Collection failed:", result.stderr, file=sys.stderr) sys.exit(1) def _systemctl_show(unit: str, *properties: str) -> Dict[str, str]: """Query systemctl show for specific properties.""" try: result = subprocess.run( ["systemctl", "show", unit, "--property=" + ",".join(properties)], capture_output=True, text=True, timeout=5, ) if result.returncode != 0: return {} out = {} for line in result.stdout.splitlines(): if "=" in line: key, _, value = line.partition("=") out[key.strip()] = value.strip() return out except (subprocess.TimeoutExpired, FileNotFoundError, OSError): return {} def _systemd_query_state() -> Dict[str, Any]: """Query systemd for the four separate service facts.""" from datetime import datetime, timezone timer_props = _systemctl_show( "fenris-collect.timer", "UnitFileState", "ActiveState", "LastTriggerUSec", ) service_props = _systemctl_show( "fenris-collect.service", "ActiveState", "ExecMainStatus", "ExecMainExitTimestamp", ) boot_enabled_str = timer_props.get("UnitFileState", "") boot_enabled = boot_enabled_str == "enabled" active_state = timer_props.get("ActiveState", "inactive") timer_active = active_state == "active" last_collect_ok = None last_collect_age_s = None last_collect_reason = None last_trigger = timer_props.get("LastTriggerUSec", "") if last_trigger and last_trigger != "n/a": try: trigger_dt = datetime.fromisoformat(last_trigger.replace("Z", "+00:00")) now = datetime.now(timezone.utc) last_collect_age_s = int((now - trigger_dt).total_seconds()) except (ValueError, TypeError): pass exec_status = service_props.get("ExecMainStatus", "") if exec_status: try: exit_code = int(exec_status) last_collect_ok = exit_code == 0 if exit_code != 0: last_collect_reason = "exit code %d" % exit_code except (ValueError, TypeError): pass return { "boot_enabled": boot_enabled, "timer_active": timer_active, "last_collect_ok": last_collect_ok, "last_collect_age_s": last_collect_age_s, "last_collect_reason": last_collect_reason, } def _systemd_journal_hint(lines: int = 5) -> Optional[str]: """Get the last N journal lines for fenris-collect.service.""" try: result = subprocess.run( ["journalctl", "-u", "fenris-collect.service", "--no-pager", "-n", str(lines), "--output=short-iso"], capture_output=True, text=True, timeout=5, ) if result.returncode != 0 or not result.stdout.strip(): return None return result.stdout.strip() except (subprocess.TimeoutExpired, FileNotFoundError, OSError): return None # --------------------------------------------------------------------------- # runit backend # --------------------------------------------------------------------------- def _runit_enable(_now: bool) -> None: """Enable the runit service by creating a symlink. runit activates the service immediately when the symlink appears. If the service is already enabled but stopped (e.g., via `sv stop`), restart it when `now=True`. """ if FENRIS_SERVICE_LINK.exists(): # Already enabled — check if we need to restart if _now and not _runit_is_running(): # Service is stopped but enabled — restart via sv try: subprocess.run( ["sv", "restart", "fenris-collect"], capture_output=True, text=True, timeout=5, ) except (FileNotFoundError, subprocess.TimeoutExpired): pass print("Service already enabled (idempotent)") return # Remove the 'down' file if present (dormant install marker) down_file = FENRIS_SV_DIR / "down" if down_file.exists(): down_file.unlink() FENRIS_SERVICE_LINK.symlink_to(FENRIS_SV_DIR) print("Service enabled") def _runit_disable(_now: bool) -> None: """Disable the runit service by removing the symlink. runit stops the service immediately when the symlink is removed. """ if not FENRIS_SERVICE_LINK.exists(): print("Service already disabled (idempotent)") return FENRIS_SERVICE_LINK.unlink() # Place 'down' file to mark as intentionally disabled (FENRIS_SV_DIR / "down").touch() print("Service disabled") def _runit_collect() -> None: """Trigger on-demand collection via direct execution with flock. Serializes against the scheduler using the same lock file. The timeout(1) command enforces bounded execution. """ lock_path = Path("/var/lib/fenris/fenris-collect.lock") collect_script = Path("/usr/libexec/fenris/fenris-collect") fallback_script = Path(__file__).parent.parent.parent / "src" / "fenris" / "collect.py" if collect_script.exists(): script = str(collect_script) elif fallback_script.exists(): script = str(fallback_script) else: print("Error: fenris-collect script not found", file=sys.stderr) sys.exit(1) try: result = subprocess.run( ["flock", "--nonblock", str(lock_path), "timeout", str(COLLECT_TIMEOUT_S), "nice", "ionice", "-c3", sys.executable, script], capture_output=True, text=True, ) if result.returncode == 0: print("Collection completed successfully") elif result.returncode == 124: print("Collection timed out after %ds" % COLLECT_TIMEOUT_S, file=sys.stderr) sys.exit(1) else: # exit code 1 from flock means lock is held (scheduled run in progress) if result.returncode == 1 and "Resource temporarily unavailable" in result.stderr: print("Collection already in progress (serialized)", file=sys.stderr) sys.exit(1) print("Collection failed:", result.stderr, file=sys.stderr) sys.exit(1) except FileNotFoundError: print("Error: flock/timeout not found", file=sys.stderr) sys.exit(1) def _runit_is_enabled() -> bool: """Check if the runit service is enabled (symlink exists).""" return FENRIS_SERVICE_LINK.exists() def _runit_is_running() -> bool: """Check if the runit service is currently running. Looks for a 'supervise/pid' file in the service directory. Void creates that directory root-only, so unprivileged dashboard reads fall back to the public process table when they cannot traverse it. """ def runsv_process_exists() -> bool: try: result = subprocess.run( ["pgrep", "-f", "^runsv fenris-collect$"], capture_output=True, text=True, timeout=5, ) return result.returncode == 0 except (FileNotFoundError, subprocess.TimeoutExpired, OSError): return False pid_file = FENRIS_SV_DIR / "supervise" / "pid" # On Void, Path.exists() is false for an unprivileged process when it # cannot traverse runit's root-only supervise directory. if not pid_file.exists(): return FENRIS_SERVICE_LINK.exists() and runsv_process_exists() try: pid = int(pid_file.read_text().strip()) # Check if the process is alive os.kill(pid, 0) return True except PermissionError: # runsv's supervisor state is root-only on Void. Its process command # is still observable, which gives the read-only UI the same runtime # fact without granting it service-control permissions. return FENRIS_SERVICE_LINK.exists() and runsv_process_exists() except (ValueError, OSError): return False def _runit_query_state() -> Dict[str, Any]: """Query runit for the four separate service facts. Checks: boot_enabled (symlink), timer_active (running), last_collect (store-based), freshness (store-based). """ from datetime import datetime, timezone from pathlib import Path boot_enabled = _runit_is_enabled() timer_active = _runit_is_running() last_collect_ok = None last_collect_age_s = None last_collect_reason = None # Try to get last collection info from the store store_path = Path("/var/lib/fenris/observations.db") if store_path.exists(): try: import sqlite3 conn = sqlite3.connect("file:%s?mode=ro" % store_path, uri=True) conn.row_factory = sqlite3.Row # Get newest sample cursor = conn.execute( "SELECT ts FROM samples ORDER BY id DESC LIMIT 1" ) row = cursor.fetchone() if row and row[0]: try: ts = datetime.fromisoformat(row[0]) if ts.tzinfo is None: ts = ts.replace(tzinfo=timezone.utc) now = datetime.now(timezone.utc) last_collect_age_s = int((now - ts).total_seconds()) last_collect_ok = True except (ValueError, TypeError): pass # Check for failed collections via monitoring_periods cursor = conn.execute( "SELECT end_cause FROM monitoring_periods " "WHERE ended_at IS NOT NULL ORDER BY ended_at DESC LIMIT 1" ) row = cursor.fetchone() if row and row[0] and row[0] not in ("user_disabled", None): last_collect_reason = row[0] conn.close() except Exception: pass # Check for timeout/exit failures via supervise exit status if timer_active: exit_file = FENRIS_SV_DIR / "supervise" / "exit" if exit_file.exists(): try: exit_code = int(exit_file.read_text().strip()) if exit_code != 0: last_collect_ok = False last_collect_reason = "exit code %d" % exit_code except (ValueError, OSError): pass return { "boot_enabled": boot_enabled, "timer_active": timer_active, "last_collect_ok": last_collect_ok, "last_collect_age_s": last_collect_age_s, "last_collect_reason": last_collect_reason, } def _runit_journal_hint(lines: int = 5) -> Optional[str]: """Get the last N log lines for fenris-collect from runit logging.""" log_current = FENRIS_LOG_DIR / "current" if not log_current.exists(): return None try: # Use tail to get the last N lines result = subprocess.run( ["tail", "-n", str(lines), str(log_current)], capture_output=True, text=True, timeout=5, ) if result.returncode != 0 or not result.stdout.strip(): return None return result.stdout.strip() except (subprocess.TimeoutExpired, FileNotFoundError, OSError): return None # --------------------------------------------------------------------------- # Public API — unified interface # --------------------------------------------------------------------------- def enable_timer(now: bool) -> None: """Enable the collection timer/service.""" init = get_init_system() if init == InitSystem.SYSTEMD: _systemd_enable(now) else: _runit_enable(now) def disable_timer(now: bool) -> None: """Disable the collection timer/service.""" init = get_init_system() if init == InitSystem.SYSTEMD: _systemd_disable(now) else: _runit_disable(now) def collect_now() -> None: """Trigger on-demand collection (blocking).""" init = get_init_system() if init == InitSystem.SYSTEMD: _systemd_collect() else: _runit_collect() def query_service_state() -> Dict[str, Any]: """Query the four separate service facts.""" init = get_init_system() if init == InitSystem.SYSTEMD: return _systemd_query_state() else: return _runit_query_state() def journal_hint(lines: int = 5, unit: str = "fenris-collect.service") -> Optional[str]: """Get journal/log hints for diagnostics. The unit parameter is accepted for backward compatibility with callers that pass 'fenris-collect.service'. For runit, the unit parameter is ignored since the log directory is always /var/log/fenris-collect/. """ init = get_init_system() if init == InitSystem.SYSTEMD: return _systemd_journal_hint(lines) else: return _runit_journal_hint(lines)