Perplexity Bot
60 lines
2.1 KiB
Python
60 lines
2.1 KiB
Python
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,
|
|
)
|