97 lines
4.5 KiB
Python
97 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only JSON adapter for an existing pegasus-health --verbose command."""
|
|
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 re
|
|
import subprocess
|
|
import sys
|
|
|
|
COMMAND = ['/usr/local/bin/pegasus-health', '--verbose']
|
|
GROUPS = [
|
|
('Ubuntu 26.04', ['Failed systemd units']),
|
|
('Nginx and PHP-FPM', ['nginx','php8.3-fpm','php8.4-fpm','Nginx configuration']),
|
|
('MySQL, PostgreSQL and Redis', ['mysql','postgresql','redis-server','MySQL','PostgreSQL','Redis']),
|
|
('Docker', ['docker','Docker containers discovered']),
|
|
('Gitea and Gitea MCP', ['gitea','gitea-mcp','Gitea HTTP']),
|
|
('Cloudflare Tunnel', ['cloudflared']),
|
|
('Cockpit', ['cockpit.socket','Cockpit']),
|
|
('Domino VM and NRPC', ['domino651 VM','domino651 autostart','Domino NRPC']),
|
|
('OpenVPN and IPsec', ['openvpn','OpenVPN UDP','IKEv2 UDP','IPsec NAT-T UDP']),
|
|
('PPPoE networking', ['Interface ppp0','Default route','DNS resolution']),
|
|
('Backup drive', ['/backup mounted','Backup filesystem UUID']),
|
|
('Backup timer', ['pegasus-backup.timer']),
|
|
('Time Machine removed', ['Time Machine removed']),
|
|
('Public websites', ['absdev.net','git.absdev.net','saturn.absdev.net','cloud.absdev.net']),
|
|
]
|
|
|
|
def output():
|
|
try:
|
|
p = subprocess.run(COMMAND, capture_output=True, text=True, timeout=180, check=False)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
print(json.dumps({'schema': 1, 'error': str(exc), 'overall': 'fail', 'groups': [], 'checks': [], 'counts': {'ok':0,'warn':0,'fail':1}}))
|
|
return 2
|
|
checks = []
|
|
section = ''
|
|
plain = re.sub(r'\x1b\[[0-9;]*m', '', p.stdout)
|
|
for line in plain.splitlines():
|
|
head = re.match(r'^=== (.+) ===$', line)
|
|
if head:
|
|
section = head.group(1)
|
|
match = re.match(r'^\[(OK\s*|WARN|FAIL)\]\s+(.+?)(?:\s+—\s+(.*))?$', line)
|
|
if match:
|
|
checks.append({'name': match[2].strip(), 'status': match[1].strip().lower(), 'detail': (match[3] or '').strip(), 'section': section})
|
|
# Noncritical firmware metadata failures are warnings.
|
|
# Other failed systemd units remain critical.
|
|
for item in checks:
|
|
if item['name'] != 'Failed systemd units':
|
|
continue
|
|
if item['status'] != 'fail':
|
|
continue
|
|
|
|
failed_units = re.findall(
|
|
r'([A-Za-z0-9@_.:-]+\\.service)\\s+loaded\\s+failed\\s+failed',
|
|
item['detail']
|
|
)
|
|
|
|
if failed_units and all(
|
|
unit == 'fwupd-refresh.service'
|
|
for unit in failed_units
|
|
):
|
|
item['status'] = 'warn'
|
|
|
|
lookup = {c['name']: c['status'] for c in checks}
|
|
groups = []
|
|
for title, names in GROUPS:
|
|
relevant = names + ([c['name'] for c in checks if c['name'].startswith('Container ')] if title == 'Docker' else [])
|
|
states = [lookup.get(name, 'fail') for name in relevant]
|
|
state = 'fail' if 'fail' in states else 'warn' if 'warn' in states else 'ok'
|
|
label = title
|
|
if title == 'Docker':
|
|
running = sum(c['status'] == 'ok' for c in checks if c['name'].startswith('Container '))
|
|
label = f'Docker — {running} containers'
|
|
groups.append({'name':label, 'status':state, 'missing':[n for n in relevant if n not in lookup]})
|
|
counts_match = re.search(r'SUMMARY:\s*(\d+) OK\s*\|\s*(\d+) WARN\s*\|\s*(\d+) FAIL', plain)
|
|
counts = dict(zip(('ok','warn','fail'), (int(x) for x in counts_match.groups()))) if counts_match else {'ok':0,'warn':0,'fail':1}
|
|
if checks:
|
|
counts = {
|
|
state: sum(c['status'] == state for c in checks)
|
|
for state in ('ok', 'warn', 'fail')
|
|
}
|
|
else:
|
|
counts['fail'] = max(1, counts['fail'])
|
|
overall = 'fail' if counts['fail'] or any(g['status']=='fail' for g in groups) else 'warn' if counts['warn'] or any(g['status']=='warn' for g in groups) else 'ok'
|
|
print(json.dumps({'schema':1, 'host':'pegasus', 'uptimeSeconds':system_uptime_seconds(), 'generatedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(), 'overall':overall, 'counts':counts,'groups':groups,'checks':checks, 'diagnostics':plain[-24000:], 'commandExitCode':p.returncode}, ensure_ascii=False))
|
|
return {'ok':0,'warn':1,'fail':2}[overall]
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(output())
|