84 lines
3.7 KiB
Python
Executable File
84 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Generic read-only Linux health adapter for Pegasus Monitor 1.1.
|
|
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
|
|
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()
|
|
try:
|
|
load = os.getloadavg()
|
|
cpus = os.cpu_count() or 1
|
|
check('System', 'CPU load', 'warn' if load[0] > cpus * 1.5 else 'ok',
|
|
f'1m load {load[0]:.2f}; {cpus} CPUs')
|
|
except OSError as exc:
|
|
check('System', 'CPU load', 'warn', str(exc))
|
|
|
|
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()]
|
|
check('Services', 'Failed systemd units', 'fail' if failed else 'ok',
|
|
', '.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')
|
|
|
|
for service in ('nginx', 'mysql', 'postgresql', 'redis-server', 'gitea'):
|
|
result = run(['systemctl', 'is-enabled', service])
|
|
if not result or result.returncode != 0:
|
|
continue
|
|
active = run(['systemctl', 'is-active', service])
|
|
working = active is not None and active.stdout.strip() == 'active'
|
|
check('Services', service, 'ok' if working else 'fail', 'active' if working else 'inactive')
|
|
|
|
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)))
|