import socket import time from datetime import datetime, timezone from urllib.request import urlopen from app.config import settings from app.models import CheckResult, HealthResponse, RuntimeResponse from app.services.systemd_manager import SystemdManager START_TIME = time.time() HEALTH_TARGETS = { 'backend': 'http://127.0.0.1:8000/health', 'worker': None, 'frontend': 'http://127.0.0.1:5173/', } def probe_http(url: str | None) -> CheckResult: if not url: return CheckResult(status='warn', detail='no health endpoint configured') try: started = time.time() with urlopen(url, timeout=2) as resp: latency = round((time.time() - started) * 1000, 2) return CheckResult(status='pass' if resp.status < 400 else 'fail', latency_ms=latency, detail=f'HTTP {resp.status}') except Exception as exc: return CheckResult(status='fail', detail=str(exc)) def build_health() -> HealthResponse: return HealthResponse( status='pass', service=settings.service_name, version=settings.version, time=datetime.now(timezone.utc), uptime_s=round(time.time() - START_TIME, 2), checks={'dbus': CheckResult(status='pass', detail='system bus reachable')}, ) def build_runtime(systemd_manager: SystemdManager) -> RuntimeResponse: services = systemd_manager.list_services(settings.monitored_services) checks = {'dbus': CheckResult(status='pass', detail='system bus reachable')} checks.update({k: probe_http(v) for k, v in HEALTH_TARGETS.items()}) overall = 'pass' if any(c.status == 'fail' for c in checks.values()) or any(s.status == 'fail' for s in services): overall = 'fail' elif any(c.status == 'warn' for c in checks.values()) or any(s.status == 'warn' for s in services): overall = 'warn' return RuntimeResponse( status=overall, service=settings.service_name, version=settings.version, time=datetime.now(timezone.utc), hostname=socket.gethostname(), uptime_s=round(time.time() - START_TIME, 2), services=services, checks=checks, )