diff --git a/ods/extensions/services/dashboard-api/README.md b/ods/extensions/services/dashboard-api/README.md index b709b7aed..94d0f129b 100644 --- a/ods/extensions/services/dashboard-api/README.md +++ b/ods/extensions/services/dashboard-api/README.md @@ -52,6 +52,7 @@ Environment variables (set in `.env`): | `GET` | `/status` | Yes | Full system status (all above combined) | | `GET` | `/api/status` | Yes | Dashboard-formatted status with inference metrics | | `GET` | `/api/host-agent/diagnostics` | Yes | Host-agent URL, gateway, auth, and live probe diagnostics | +| `GET` | `/api/events/services` | Yes | SSE stream of deduplicated service-health transitions; repeat `service=` to scope the stream | ### Preflight diff --git a/ods/extensions/services/dashboard-api/main.py b/ods/extensions/services/dashboard-api/main.py index 8b5e06d2b..2364ed142 100644 --- a/ods/extensions/services/dashboard-api/main.py +++ b/ods/extensions/services/dashboard-api/main.py @@ -72,6 +72,7 @@ tailscale, usage, node, + events, ) from settings import ( _ENV_ASSIGNMENT_RE, _ENV_COMMENTED_ASSIGNMENT_RE, _SETTINGS_APPLY_ALLOWED_SERVICES, _parse_env_text, _read_env_map_from_path, @@ -1114,6 +1115,7 @@ def get_allowed_origins(): app.include_router(tailscale.router) app.include_router(usage.router) app.include_router(node.router) +app.include_router(events.router) # ================================================================ diff --git a/ods/extensions/services/dashboard-api/routers/events.py b/ods/extensions/services/dashboard-api/routers/events.py new file mode 100644 index 000000000..9eea9fb6c --- /dev/null +++ b/ods/extensions/services/dashboard-api/routers/events.py @@ -0,0 +1,140 @@ +"""Server-sent service health events for operator automation.""" + +import asyncio +import hashlib +import json +import time +from datetime import datetime, timezone + +from fastapi import APIRouter, Depends, Header, Query, Request +from fastapi.responses import StreamingResponse + +from helpers import get_cached_services +from security import verify_api_key + + +router = APIRouter(tags=["events"]) + + +class ServiceStateTracker: + """Convert cached health snapshots into deduplicated state transitions.""" + + def __init__(self) -> None: + self._fingerprint: str | None = None + self._states: dict[str, str] | None = None + + def observe(self, statuses: list, *, cache_ready: bool) -> dict | None: + states = { + str(service.id): str(service.status) + for service in statuses + } + canonical = json.dumps( + {"cache_ready": cache_ready, "states": states}, + sort_keys=True, + separators=(",", ":"), + ) + fingerprint = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20] + if fingerprint == self._fingerprint: + return None + + changes = [] + if self._states is not None: + for service_id in sorted(set(self._states) | set(states)): + previous = self._states.get(service_id) + current = states.get(service_id) + if previous != current: + changes.append({ + "id": service_id, + "from": previous, + "to": current or "removed", + }) + + self._fingerprint = fingerprint + self._states = states + return { + "id": fingerprint, + "event": "services", + "data": { + "observed_at": datetime.now(timezone.utc).isoformat(), + "cache_ready": cache_ready, + "services": [ + {"id": service_id, "status": states[service_id]} + for service_id in sorted(states) + ], + "changes": changes, + }, + } + + +def format_sse(observation: dict) -> str: + payload = json.dumps(observation["data"], sort_keys=True, separators=(",", ":")) + return ( + f"id: {observation['id']}\n" + f"event: {observation['event']}\n" + f"data: {payload}\n\n" + ) + + +async def service_event_stream( + request: Request, + *, + poll_seconds: float, + heartbeat_seconds: float, + last_event_id: str | None, + max_events: int | None, + service_ids: frozenset[str] | None, +): + tracker = ServiceStateTracker() + emitted = 0 + last_heartbeat = time.monotonic() + + while not await request.is_disconnected(): + cached = get_cached_services() + statuses = cached or [] + if service_ids is not None: + statuses = [ + service + for service in statuses + if str(service.id) in service_ids + ] + observation = tracker.observe(statuses, cache_ready=cached is not None) + if observation is not None and observation["id"] != last_event_id: + yield format_sse(observation) + emitted += 1 + last_event_id = observation["id"] + last_heartbeat = time.monotonic() + if max_events is not None and emitted >= max_events: + return + elif time.monotonic() - last_heartbeat >= heartbeat_seconds: + yield ": keep-alive\n\n" + last_heartbeat = time.monotonic() + await asyncio.sleep(poll_seconds) + + +@router.get("/api/events/services") +async def service_events( + request: Request, + poll_seconds: float = Query(2.0, ge=0.5, le=30.0), + heartbeat_seconds: float = Query(15.0, ge=5.0, le=120.0), + max_events: int | None = Query(None, ge=1, le=1000), + service: list[str] | None = Query(None), + last_event_id: str | None = Header(None, alias="Last-Event-ID"), + api_key: str = Depends(verify_api_key), +): + """Stream deduplicated service-status snapshots and transitions as SSE.""" + stream = service_event_stream( + request, + poll_seconds=poll_seconds, + heartbeat_seconds=heartbeat_seconds, + last_event_id=last_event_id, + max_events=max_events, + service_ids=frozenset(service) if service else None, + ) + return StreamingResponse( + stream, + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }, + ) diff --git a/ods/extensions/services/dashboard-api/tests/test_service_events.py b/ods/extensions/services/dashboard-api/tests/test_service_events.py new file mode 100644 index 000000000..89b4198a3 --- /dev/null +++ b/ods/extensions/services/dashboard-api/tests/test_service_events.py @@ -0,0 +1,156 @@ +"""Tests for the authenticated service-health event stream.""" + +import json +from types import SimpleNamespace + +import pytest + +from routers.events import ServiceStateTracker, format_sse, service_event_stream + + +def _service(service_id, status, response_time_ms=1): + return SimpleNamespace(id=service_id, status=status, response_time_ms=response_time_ms) + + +def _event_data(event_text): + data_line = next(line for line in event_text.splitlines() if line.startswith("data: ")) + return json.loads(data_line.removeprefix("data: ")) + + +def test_tracker_emits_only_state_changes_and_reports_transitions(): + tracker = ServiceStateTracker() + + initial = tracker.observe([_service("api", "healthy", 10)], cache_ready=True) + latency_only = tracker.observe([_service("api", "healthy", 900)], cache_ready=True) + changed = tracker.observe( + [_service("api", "unhealthy"), _service("worker", "healthy")], + cache_ready=True, + ) + + assert initial["data"]["changes"] == [] + assert latency_only is None + assert changed["data"]["changes"] == [ + {"id": "api", "from": "healthy", "to": "unhealthy"}, + {"id": "worker", "from": None, "to": "healthy"}, + ] + + +def test_tracker_reports_removed_services_and_cache_readiness(): + tracker = ServiceStateTracker() + waiting = tracker.observe([], cache_ready=False) + ready = tracker.observe([_service("api", "healthy")], cache_ready=True) + removed = tracker.observe([], cache_ready=True) + + assert waiting["data"]["cache_ready"] is False + assert ready["data"]["changes"] == [ + {"id": "api", "from": None, "to": "healthy"}, + ] + assert removed["data"]["changes"] == [ + {"id": "api", "from": "healthy", "to": "removed"}, + ] + + +def test_sse_format_has_resumable_id_event_and_compact_json(): + observation = ServiceStateTracker().observe( + [_service("api", "healthy")], + cache_ready=True, + ) + + rendered = format_sse(observation) + + assert rendered.startswith(f"id: {observation['id']}\nevent: services\n") + assert rendered.endswith("\n\n") + assert _event_data(rendered)["services"] == [{"id": "api", "status": "healthy"}] + + +@pytest.mark.asyncio +async def test_stream_resumes_after_last_event_id_and_waits_for_a_change(monkeypatch): + snapshots = iter([ + [_service("api", "healthy")], + [_service("api", "unhealthy")], + ]) + monkeypatch.setattr("routers.events.get_cached_services", lambda: next(snapshots)) + previous = ServiceStateTracker().observe( + [_service("api", "healthy")], + cache_ready=True, + ) + + class ConnectedRequest: + async def is_disconnected(self): + return False + + stream = service_event_stream( + ConnectedRequest(), + poll_seconds=0, + heartbeat_seconds=15, + last_event_id=previous["id"], + max_events=1, + service_ids=None, + ) + resumed = await anext(stream) + + assert _event_data(resumed)["changes"] == [ + {"id": "api", "from": "healthy", "to": "unhealthy"}, + ] + + +def test_http_stream_can_emit_one_bounded_snapshot(test_client, monkeypatch): + monkeypatch.setattr( + "routers.events.get_cached_services", + lambda: [_service("dashboard-api", "healthy"), _service("llama-server", "starting")], + ) + + response = test_client.get( + "/api/events/services?max_events=1", + headers=test_client.auth_headers, + ) + + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + assert response.headers["cache-control"] == "no-cache" + assert response.headers["x-accel-buffering"] == "no" + assert "event: services" in response.text + assert _event_data(response.text)["services"] == [ + {"id": "dashboard-api", "status": "healthy"}, + {"id": "llama-server", "status": "starting"}, + ] + + +def test_http_stream_can_scope_snapshots_to_repeated_service_filters( + test_client, + monkeypatch, +): + monkeypatch.setattr( + "routers.events.get_cached_services", + lambda: [ + _service("dashboard-api", "healthy"), + _service("llama-server", "starting"), + _service("open-webui", "healthy"), + ], + ) + + response = test_client.get( + "/api/events/services?max_events=1&service=open-webui&service=llama-server", + headers=test_client.auth_headers, + ) + + assert response.status_code == 200 + assert _event_data(response.text)["services"] == [ + {"id": "llama-server", "status": "starting"}, + {"id": "open-webui", "status": "healthy"}, + ] + + +def test_http_stream_requires_authentication(test_client): + response = test_client.get("/api/events/services?max_events=1") + + assert response.status_code == 401 + + +def test_http_stream_validates_timing_and_event_bounds(test_client): + response = test_client.get( + "/api/events/services?poll_seconds=0.1&heartbeat_seconds=1&max_events=0", + headers=test_client.auth_headers, + ) + + assert response.status_code == 422