From c3a9a43b66480e1479a6077cedb2b2b2e93b0de1 Mon Sep 17 00:00:00 2001 From: Tang Vu <145498528+tang-vu@users.noreply.github.com> Date: Fri, 21 Aug 2026 22:38:28 +0700 Subject: [PATCH 1/2] feat(observability): stream service health transitions --- .../services/dashboard-api/README.md | 1 + ods/extensions/services/dashboard-api/main.py | 2 + .../services/dashboard-api/routers/events.py | 130 ++++++++++++++++++ .../tests/test_service_events.py | 130 ++++++++++++++++++ 4 files changed, 263 insertions(+) create mode 100644 ods/extensions/services/dashboard-api/routers/events.py create mode 100644 ods/extensions/services/dashboard-api/tests/test_service_events.py diff --git a/ods/extensions/services/dashboard-api/README.md b/ods/extensions/services/dashboard-api/README.md index b709b7aed..01fc11434 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 snapshots and transitions | ### 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..3be4f241b --- /dev/null +++ b/ods/extensions/services/dashboard-api/routers/events.py @@ -0,0 +1,130 @@ +"""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, +): + tracker = ServiceStateTracker() + emitted = 0 + last_heartbeat = time.monotonic() + + while not await request.is_disconnected(): + cached = get_cached_services() + observation = tracker.observe(cached or [], 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), + 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, + ) + 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..60e6f24cd --- /dev/null +++ b/ods/extensions/services/dashboard-api/tests/test_service_events.py @@ -0,0 +1,130 @@ +"""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, + ) + 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_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 From d8aed4c6bb3c4f9a7941a25c7a52b99e2e4abd6e Mon Sep 17 00:00:00 2001 From: Tang Vu <145498528+tang-vu@users.noreply.github.com> Date: Mon, 24 Aug 2026 00:05:05 +0700 Subject: [PATCH 2/2] feat(observability): filter service health streams --- .../services/dashboard-api/README.md | 2 +- .../services/dashboard-api/routers/events.py | 12 ++++++++- .../tests/test_service_events.py | 26 +++++++++++++++++++ 3 files changed, 38 insertions(+), 2 deletions(-) diff --git a/ods/extensions/services/dashboard-api/README.md b/ods/extensions/services/dashboard-api/README.md index 01fc11434..94d0f129b 100644 --- a/ods/extensions/services/dashboard-api/README.md +++ b/ods/extensions/services/dashboard-api/README.md @@ -52,7 +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 snapshots and transitions | +| `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/routers/events.py b/ods/extensions/services/dashboard-api/routers/events.py index 3be4f241b..9eea9fb6c 100644 --- a/ods/extensions/services/dashboard-api/routers/events.py +++ b/ods/extensions/services/dashboard-api/routers/events.py @@ -82,6 +82,7 @@ async def service_event_stream( heartbeat_seconds: float, last_event_id: str | None, max_events: int | None, + service_ids: frozenset[str] | None, ): tracker = ServiceStateTracker() emitted = 0 @@ -89,7 +90,14 @@ async def service_event_stream( while not await request.is_disconnected(): cached = get_cached_services() - observation = tracker.observe(cached or [], cache_ready=cached is not None) + 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 @@ -109,6 +117,7 @@ async def service_events( 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), ): @@ -119,6 +128,7 @@ async def service_events( 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, diff --git a/ods/extensions/services/dashboard-api/tests/test_service_events.py b/ods/extensions/services/dashboard-api/tests/test_service_events.py index 60e6f24cd..89b4198a3 100644 --- a/ods/extensions/services/dashboard-api/tests/test_service_events.py +++ b/ods/extensions/services/dashboard-api/tests/test_service_events.py @@ -85,6 +85,7 @@ async def is_disconnected(self): heartbeat_seconds=15, last_event_id=previous["id"], max_events=1, + service_ids=None, ) resumed = await anext(stream) @@ -115,6 +116,31 @@ def test_http_stream_can_emit_one_bounded_snapshot(test_client, monkeypatch): ] +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")