Skip to content

Analyze API

Functions wrapping systemd-analyze for boot analysis, security scoring, unit verification, and critical-chain inspection.

These are available as methods on SystemdClient / AsyncSystemdClient (e.g., client.analyze_blame()), but are implemented in the systemd_client._analyze module.

Module functions

systemd_client._analyze

systemd-analyze wrapper for boot analysis and security scoring.

analyze_blame(scope) async

Parse systemd-analyze blame output into BlameEntry list.

Source code in src/systemd_client/_analyze.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
async def analyze_blame(scope: SystemdScope) -> list[BlameEntry]:
    """Parse systemd-analyze blame output into BlameEntry list."""
    stdout = await _run_analyze(scope, "blame", "--no-pager")
    entries: list[BlameEntry] = []
    for line in stdout.splitlines():
        line = line.strip()
        if not line:
            continue
        # Format: "32.875s pmlogger.service" or "245ms dbus.service"
        match = re.match(r"^\s*([\d.]+)(min|s|ms|us)\s+(.+)$", line)
        if match:
            val, unit_str, name = match.groups()
            multipliers = {"us": 1, "ms": 1000, "s": 1_000_000, "min": 60_000_000}
            time_us = int(float(val) * multipliers.get(unit_str, 1))
            entries.append(BlameEntry(time_us=time_us, unit=name))
    return entries

analyze_critical_chain(scope, unit=None) async

Return systemd-analyze critical-chain output as raw text.

Source code in src/systemd_client/_analyze.py 
47
48
49
50
51
52
async def analyze_critical_chain(scope: SystemdScope, unit: str | None = None) -> str:
    """Return systemd-analyze critical-chain output as raw text."""
    args = ["critical-chain", "--no-pager"]
    if unit:
        args.append(unit)
    return await _run_analyze(scope, *args)

analyze_security(scope, unit) async

Parse systemd-analyze security output for a specific unit.

Source code in src/systemd_client/_analyze.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
async def analyze_security(scope: SystemdScope, unit: str) -> SecurityAnalysis:
    """Parse systemd-analyze security output for a specific unit."""
    stdout = await _run_analyze(scope, "security", "--json=short", unit)
    data = json.loads(stdout) if stdout.strip() else []

    # JSON output is a list of unit analyses
    for entry in data:
        if entry.get("unit", "") == unit or len(data) == 1:
            exposure = float(entry.get("exposure", 10.0))
            issues: list[SecurityIssue] = []
            for pred in entry.get("predicates", []):
                issues.append(SecurityIssue(
                    id=pred.get("set_name", ""),
                    description=pred.get("description", ""),
                    severity=pred.get("badness_description", ""),
                    value=pred.get("value", ""),
                ))
            return SecurityAnalysis(unit=unit, exposure=exposure, issues=issues)

    return SecurityAnalysis(unit=unit, exposure=10.0)

analyze_verify(scope, unit) async

Run systemd-analyze verify and return diagnostic messages.

Source code in src/systemd_client/_analyze.py 
77
78
79
80
81
82
83
84
async def analyze_verify(scope: SystemdScope, unit: str) -> list[str]:
    """Run systemd-analyze verify and return diagnostic messages."""
    try:
        stdout = await _run_analyze(scope, "verify", unit)
    except SubprocessError as exc:
        # verify returns non-zero on warnings, but stderr has the messages
        return [line for line in exc.stderr.splitlines() if line.strip()]
    return [line for line in stdout.splitlines() if line.strip()]