SystemsMonitor/server/remote-health-json

#!/usr/bin/env python3
"""Generic read-only Linux health adapter for Pegasus Monitor 1.2.
Install as /usr/local/bin/pegasus-monitor-health-json and run as an ordinary user.
No third-party Python packages required.
"""
import datetime
import json

def system_uptime_seconds():
    """Read Linux kernel uptime without privilege or external commands."""
    try:
        with open('/proc/uptime', encoding='ascii') as handle:
            return max(0, int(float(handle.read().split()[0])))
    except (OSError, ValueError, IndexError):
        return None

import os
import shutil
import socket
import subprocess

checks = []

def run(args, timeout=8):
    try:
        return subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
    except (OSError, subprocess.TimeoutExpired):
        return None

def check(section, name, status, detail):
    checks.append(dict(section=section, name=name, status=status, detail=str(detail)[:500]))

host = socket.gethostname()
# CPU load is informational only.
# Do not generate health warnings without identifying
# the process responsible for excessive CPU usage.
try:
    load = os.getloadavg()
    cpus = os.cpu_count() or 1
    check('System', 'CPU load', 'ok',
          f'1m load {load[0]:.2f}; {cpus} CPUs (informational)')
except OSError:
    pass

try:
    with open('/proc/meminfo', encoding='utf-8') as f:
        values = {parts[0].rstrip(':'): int(parts[1]) for line in f if (parts := line.split()) and len(parts) > 1}
    total = values['MemTotal']
    available = values.get('MemAvailable', values.get('MemFree', 0))
    pct = round(100 * (total - available) / total)
    check('System', 'Memory', 'fail' if pct >= 95 else 'warn' if pct >= 85 else 'ok', f'{pct}% used')
except (OSError, KeyError, ZeroDivisionError, ValueError) as exc:
    check('System', 'Memory', 'warn', str(exc))

for directory in ('/', '/backup'):
    if directory != '/' and not os.path.ismount(directory):
        continue
    try:
        usage = shutil.disk_usage(directory)
        pct = round(100 * usage.used / usage.total)
        check('Storage', f'Disk {directory}', 'fail' if pct >= 95 else 'warn' if pct >= 85 else 'ok', f'{pct}% used')
    except (OSError, ZeroDivisionError) as exc:
        check('Storage', f'Disk {directory}', 'warn', str(exc))

systemctl = run(['systemctl', '--failed', '--no-legend', '--plain'])
if systemctl is not None:
    failed = [line for line in systemctl.stdout.splitlines() if line.strip()]
    # Firmware metadata refresh is noncritical.
    # Any other failed systemd unit remains critical.
    optional_units = {'fwupd-refresh.service'}

    failed_names = [
        line.split()[0].lstrip('●')
        for line in failed
        if line.split()
    ]

    if not failed:
        severity = 'ok'
    elif failed_names and all(
        unit in optional_units for unit in failed_names
    ):
        severity = 'warn'
    else:
        severity = 'fail'

    check('Services', 'Failed systemd units', severity,
          ', '.join(failed[:5]) if failed else 'None')

if shutil.which('docker'):
    docker = run(['docker', 'ps', '--format', '{{.Names}}'], timeout=12)
    if docker and docker.returncode == 0:
        names = [name for name in docker.stdout.splitlines() if name]
        check('Docker', 'Docker containers', 'ok', f'{len(names)} running: {", ".join(names[:8])}')
    else:
        check('Docker', 'Docker access', 'warn', 'Docker is installed but unavailable to this SSH user')

# Discover common services without assuming they are installed or enabled.
# Each state is explicit so optional missing software never counts as a failure.
SERVICES = (
    ('nginx', 'nginx'), ('mysql', 'mysql'), ('mariadb', 'mariadb'),
    ('postgresql', 'postgresql'), ('redis-server', 'redis-server'),
    ('gitea', 'gitea'), ('docker', 'docker'), ('containerd', 'containerd'),
    ('php8.3-fpm', 'php8.3-fpm'), ('php8.4-fpm', 'php8.4-fpm'),
    ('php8.5-fpm', 'php8.5-fpm'), ('ssh', 'ssh'), ('cloudflared', 'cloudflared'),
    ('cockpit.socket', 'cockpit.socket'), ('libvirtd', 'libvirtd'),
)
services = []
for display, unit in SERVICES:
    full_unit = unit if unit.endswith(('.service', '.socket')) else unit + '.service'
    result = run(['systemctl', 'show', full_unit, '--no-pager',
                  '--property=LoadState', '--property=ActiveState',
                  '--property=UnitFileState'], timeout=5)
    if result is None or not result.stdout.strip():
        state, detail = 'unknown', 'systemctl unavailable'
    else:
        properties = dict(line.split('=', 1) for line in result.stdout.splitlines() if '=' in line)
        load = properties.get('LoadState', 'not-found')
        active = properties.get('ActiveState', 'unknown')
        unit_state = properties.get('UnitFileState', 'unknown')
        detail = f'Load: {load}; active: {active}; enablement: {unit_state}'
        if load == 'not-found':
            state = 'not_installed'
        elif active == 'active':
            state = 'running'
        elif active == 'failed':
            state = 'failed'
        elif unit_state in ('disabled', 'masked'):
            state = 'disabled'
        elif active in ('inactive', 'deactivating'):
            state = 'stopped'
        else:
            state = 'unknown'
    services.append(dict(name=display, state=state, detail=detail))
    # Uninstalled/disabled services are informational, not warnings or failures.
    if state == 'failed':
        check('Services', display, 'fail', detail)
    elif state == 'stopped':
        check('Services', display, 'warn', detail)

groups = []
for section in dict.fromkeys(c['section'] for c in checks):
    relevant = [c for c in checks if c['section'] == section]
    status = 'fail' if any(c['status'] == 'fail' for c in relevant) else 'warn' if any(c['status'] == 'warn' for c in relevant) else 'ok'
    groups.append(dict(name=section, status=status, missing=[]))
counts = {s: sum(c['status'] == s for c in checks) for s in ('ok', 'warn', 'fail')}
overall = 'fail' if counts['fail'] else 'warn' if counts['warn'] else 'ok'
print(json.dumps(dict(schema=1, host=host, generatedAt=datetime.datetime.now(datetime.timezone.utc).isoformat(),
                      overall=overall, counts=counts, groups=groups, checks=checks, diagnostics=None, services=services,
                      uptimeSeconds=system_uptime_seconds())))