diff --git a/src/govoplan_ops/backend/api/v1/routes.py b/src/govoplan_ops/backend/api/v1/routes.py index 9c23ef6..a050b19 100644 --- a/src/govoplan_ops/backend/api/v1/routes.py +++ b/src/govoplan_ops/backend/api/v1/routes.py @@ -1,6 +1,9 @@ from __future__ import annotations import os +import threading +import time +from collections.abc import Mapping from pathlib import Path from typing import Any from urllib.parse import urlsplit @@ -11,13 +14,23 @@ from sqlalchemy.exc import SQLAlchemyError from govoplan_core.auth import ApiPrincipal, require_any_scope from govoplan_core.core.maintenance import saved_maintenance_mode +from govoplan_core.core.module_installer import ( + default_installer_runtime_dir, + list_module_installer_runs, + read_module_installer_run, +) +from govoplan_core.core.operations import OperationalCheckProviderRegistration from govoplan_core.core.registry import PlatformRegistry from govoplan_core.db.session import get_database from govoplan_core.settings import settings as core_settings -from govoplan_ops.backend.manifest import OPS_READ_SCOPES +from govoplan_ops.backend.manifest import OPS_READ_SCOPES, OPS_RUN_SCOPES router = APIRouter(prefix="/ops", tags=["ops"]) +_module_check_cache: dict[str, tuple[float, dict[str, Any]]] = {} +_module_check_cache_lock = threading.Lock() +_worker_check_cache: tuple[float, dict[str, Any]] | None = None +_worker_check_cache_lock = threading.Lock() @router.get("/status") @@ -42,20 +55,40 @@ def ops_readiness( return readiness -def _ops_status_payload(request: Request) -> dict[str, Any]: +@router.post("/checks/run") +def run_ops_checks( + request: Request, + principal: ApiPrincipal = Depends(require_any_scope(*OPS_RUN_SCOPES)), +) -> dict[str, Any]: + del principal + return _ops_status_payload(request, force_module_checks=True) + + +def _ops_status_payload( + request: Request, + *, + force_module_checks: bool = False, +) -> dict[str, Any]: registry = _registry(request) database = _database_status() maintenance_mode = database.get("maintenance_mode") if isinstance(database.get("maintenance_mode"), dict) else {"enabled": False, "message": None} redis_check = _redis_check() current_profile = _current_profile() + worker_check = _worker_check() + module_checks = _module_operational_checks( + registry, + force=force_module_checks, + ) checks = [ _check("module_registry", "Module registry", "ok", f"{len(registry.manifests())} modules enabled."), _check("database", "Database", "ok" if database["ok"] else "error", str(database["detail"])), _maintenance_check(maintenance_mode), redis_check, - _worker_check(), + worker_check, _storage_check(), + _backup_restore_check(), _deployment_security_check(current_profile), + *module_checks, ] readiness = _readiness(checks, maintenance_mode) governance = _governance_inventory(registry) @@ -70,6 +103,8 @@ def _ops_status_payload(request: Request) -> dict[str, Any]: "maintenance_mode": maintenance_mode, "database_url": _redact_url(core_settings.database_url), "file_storage_backend": core_settings.file_storage_backend, + "worker_metrics": worker_check.get("metrics", {}), + "operational_probe_count": len(module_checks), }, "readiness": readiness, "checks": checks, @@ -161,18 +196,108 @@ def _redis_check() -> dict[str, Any]: def _worker_check() -> dict[str, Any]: if not core_settings.celery_enabled: - return _check("worker_split", "Background workers", "warning", "Celery is disabled; long-running work executes only through synchronous or dev paths.") + return _check( + "worker_split", + "Background workers", + "warning", + "Celery is disabled; long-running work executes only through synchronous or dev paths.", + metrics={"workers": 0, "active_tasks": 0, "missing_queues": []}, + ) + global _worker_check_cache + now = time.monotonic() + with _worker_check_cache_lock: + if _worker_check_cache is not None and now - _worker_check_cache[0] < 15: + return dict(_worker_check_cache[1]) try: from govoplan_core.celery_app import celery inspector = celery.control.inspect(timeout=0.75) replies = inspector.ping() or {} + active_queues_by_worker = inspector.active_queues() or {} + active_by_worker = inspector.active() or {} except Exception as exc: # noqa: BLE001 - diagnostic endpoint should report the concrete failure. - return _check("worker_split", "Background workers", "error", f"Worker heartbeat check failed: {exc}", readiness_critical=True) + result = _check( + "worker_split", + "Background workers", + "error", + f"Worker heartbeat check failed: {exc}", + readiness_critical=True, + metrics={"workers": 0, "active_tasks": 0}, + ) + with _worker_check_cache_lock: + _worker_check_cache = (now, result) + return result if replies: worker_names = ", ".join(sorted(replies)) - return _check("worker_split", "Background workers", "ok", f"{len(replies)} worker(s) replied: {worker_names}.") - return _check("worker_split", "Background workers", "error", "Celery is enabled, but no workers replied to heartbeat.", readiness_critical=True) + active_queues = sorted({ + str(queue.get("name")) + for queues in active_queues_by_worker.values() + if isinstance(queues, list) + for queue in queues + if isinstance(queue, Mapping) and queue.get("name") + }) + expected_queues = _celery_queues() + missing_queues = sorted(set(expected_queues) - set(active_queues)) + active_tasks = sum( + len(tasks) + for tasks in active_by_worker.values() + if isinstance(tasks, list) + ) + queue_depths = _queue_depths(expected_queues) + state = "warning" if missing_queues else "ok" + detail = f"{len(replies)} worker(s) replied: {worker_names}." + if missing_queues: + detail += " No worker consumes configured queues: " + ", ".join(missing_queues) + "." + result = _check( + "worker_split", + "Background workers", + state, + detail, + readiness_critical=bool(missing_queues), + metrics={ + "workers": len(replies), + "active_tasks": active_tasks, + "expected_queues": expected_queues, + "active_queues": active_queues, + "missing_queues": missing_queues, + "queue_depths": queue_depths, + }, + ) + with _worker_check_cache_lock: + _worker_check_cache = (now, result) + return result + result = _check( + "worker_split", + "Background workers", + "error", + "Celery is enabled, but no workers replied to heartbeat.", + readiness_critical=True, + metrics={"workers": 0, "active_tasks": 0}, + ) + with _worker_check_cache_lock: + _worker_check_cache = (now, result) + return result + + +def _queue_depths(queues: list[str]) -> dict[str, int]: + try: + from redis import Redis + + client = Redis.from_url( + core_settings.redis_url, + socket_connect_timeout=0.75, + socket_timeout=0.75, + ) + pipeline = client.pipeline(transaction=False) + for queue in queues: + pipeline.llen(queue) + values = pipeline.execute() + except Exception: # noqa: BLE001 - worker heartbeat remains the readiness source. + return {} + return { + queue: int(value) + for queue, value in zip(queues, values, strict=True) + } def _storage_check() -> dict[str, Any]: @@ -189,6 +314,149 @@ def _storage_check() -> dict[str, Any]: return _check("file_storage", "File storage", "warning", f"Local file storage root does not exist yet: {root}", readiness_critical=False) +def _module_operational_checks( + registry: PlatformRegistry, + *, + force: bool, +) -> list[dict[str, Any]]: + registrations: list[OperationalCheckProviderRegistration] = [] + for manifest in registry.manifests(): + registrations.extend(manifest.operational_check_providers) + + results: list[dict[str, Any]] = [] + seen: set[str] = set() + for registration in registrations: + cache_key = f"{registration.module_id}:{registration.check_id}" + if registration.check_id in seen: + results.append(_check( + f"ops.duplicate.{registration.check_id}", + "Operational check registry", + "error", + f"Operational check id {registration.check_id!r} is registered more than once.", + readiness_critical=True, + )) + continue + seen.add(registration.check_id) + cached = _cached_module_check( + cache_key, + max_age_seconds=max(0, registration.cache_seconds), + ) + if not force and cached is not None: + results.append(cached) + continue + try: + check = registration.provider() + if check.id != registration.check_id: + raise ValueError( + f"provider returned {check.id!r}, expected {registration.check_id!r}" + ) + result = check.as_dict() + except Exception as exc: # noqa: BLE001 - one optional module must not hide all Ops status. + result = _check( + registration.check_id, + f"{registration.module_id} operational check", + "error", + f"The module-owned check failed unexpectedly ({type(exc).__name__}).", + readiness_critical=True, + ) + with _module_check_cache_lock: + _module_check_cache[cache_key] = (time.monotonic(), result) + results.append(dict(result)) + return results + + +def _cached_module_check( + cache_key: str, + *, + max_age_seconds: int, +) -> dict[str, Any] | None: + if max_age_seconds <= 0: + return None + with _module_check_cache_lock: + cached = _module_check_cache.get(cache_key) + if cached is None or time.monotonic() - cached[0] >= max_age_seconds: + return None + return dict(cached[1]) + + +def _backup_restore_check() -> dict[str, Any]: + runtime_dir = default_installer_runtime_dir(core_settings.database_url) + runs = list_module_installer_runs(runtime_dir=runtime_dir, limit=25) + latest_evidence: tuple[dict[str, object], Mapping[str, object]] | None = None + for summary in runs: + try: + record = read_module_installer_run( + runtime_dir=runtime_dir, + run_id=str(summary["run_id"]), + ) + except Exception: # noqa: BLE001 - malformed historical evidence is skipped. + continue + snapshot = record.get("snapshot") + backup = snapshot.get("database_backup") if isinstance(snapshot, Mapping) else None + if isinstance(backup, Mapping): + latest_evidence = (record, backup) + break + + drill = _restore_drill_evidence(runtime_dir) + if latest_evidence is None: + return _check( + "backup_restore_evidence", + "Backup and restore evidence", + "warning", + "No database backup evidence is present in installer history. External deployment backups may exist, but are not evidenced here.", + metrics={"restore_drill_ok": bool(drill and drill.get("ok") is True)}, + ) + + record, backup = latest_evidence + restore_check = backup.get("restore_check") + restore_check_ok = ( + isinstance(restore_check, Mapping) + and ( + restore_check.get("return_code") == 0 + or str(restore_check.get("result") or "").lower() == "ok" + ) + ) + drill_ok = bool(drill and drill.get("ok") is True) + state = "ok" if restore_check_ok and drill_ok else "warning" + missing: list[str] = [] + if not restore_check_ok: + missing.append("backup restore-readiness check") + if not drill_ok: + missing.append("recorded rollback drill") + detail = ( + f"Installer run {record.get('run_id')} records a {backup.get('type', 'database')} backup." + ) + if missing: + detail += " Missing evidence: " + ", ".join(missing) + "." + else: + detail += " The backup check and rollback drill both passed." + return _check( + "backup_restore_evidence", + "Backup and restore evidence", + state, + detail, + metrics={ + "run_id": record.get("run_id"), + "backup_type": backup.get("type"), + "restore_check_ok": restore_check_ok, + "restore_drill_ok": drill_ok, + "drill_completed_at": drill.get("completed_at") if drill else None, + }, + ) + + +def _restore_drill_evidence(runtime_dir: Path) -> dict[str, object] | None: + configured = os.environ.get("GOVOPLAN_RESTORE_DRILL_EVIDENCE_PATH") + path = Path(configured).expanduser() if configured else runtime_dir / "restore-drill-evidence.json" + try: + import json + + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + return payload if isinstance(payload, dict) else None + + def _deployment_security_check(current_profile: str) -> dict[str, Any]: app_env = str(core_settings.app_env or "").lower() if app_env in {"dev", "test", "local"}: @@ -217,8 +485,23 @@ def _deployment_security_check(current_profile: str) -> dict[str, Any]: ) -def _check(check_id: str, label: str, state: str, detail: str, *, readiness_critical: bool = False) -> dict[str, Any]: - return {"id": check_id, "label": label, "state": state, "detail": detail, "readiness_critical": readiness_critical} +def _check( + check_id: str, + label: str, + state: str, + detail: str, + *, + readiness_critical: bool = False, + metrics: Mapping[str, object] | None = None, +) -> dict[str, Any]: + return { + "id": check_id, + "label": label, + "state": state, + "detail": detail, + "readiness_critical": readiness_critical, + "metrics": dict(metrics or {}), + } def _readiness(checks: list[dict[str, Any]], maintenance_mode: dict[str, Any]) -> dict[str, Any]: diff --git a/src/govoplan_ops/backend/manifest.py b/src/govoplan_ops/backend/manifest.py index d79e922..bd8af6e 100644 --- a/src/govoplan_ops/backend/manifest.py +++ b/src/govoplan_ops/backend/manifest.py @@ -6,6 +6,8 @@ from govoplan_core.core.views import ViewSurface OPS_READ_SCOPE = "ops:operations:read" OPS_READ_SCOPES = (OPS_READ_SCOPE, "system:settings:read", "admin:settings:read") +OPS_RUN_SCOPE = "ops:operations:run" +OPS_RUN_SCOPES = (OPS_RUN_SCOPE, "system:settings:write") def _permission(scope: str, label: str, description: str) -> PermissionDefinition: @@ -41,6 +43,11 @@ manifest = ModuleManifest( "View operations status", "Read runtime health, deployment profile, and sizing information.", ), + _permission( + OPS_RUN_SCOPE, + "Run operational checks", + "Run bounded module-owned persistence and integration probes.", + ), ), role_templates=( RoleTemplate( @@ -50,6 +57,13 @@ manifest = ModuleManifest( permissions=(OPS_READ_SCOPE,), level="system", ), + RoleTemplate( + slug="ops_operator", + name="Operations operator", + description="Read platform health and run bounded operational probes.", + permissions=(OPS_READ_SCOPE, OPS_RUN_SCOPE), + level="system", + ), ), route_factory=_route_factory, nav_items=(NavItem(path="/ops", label="Ops", icon="activity", required_any=OPS_READ_SCOPES, order=890),), diff --git a/tests/test_operational_checks.py b/tests/test_operational_checks.py new file mode 100644 index 0000000..4c41b7c --- /dev/null +++ b/tests/test_operational_checks.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from govoplan_core.core.operations import ( + OperationalCheck, + OperationalCheckProviderRegistration, +) +from govoplan_ops.backend.api.v1 import routes + + +@dataclass +class _Manifest: + operational_check_providers: tuple[OperationalCheckProviderRegistration, ...] + + +class _Registry: + def __init__(self, *registrations: OperationalCheckProviderRegistration): + self._manifest = _Manifest(tuple(registrations)) + + def manifests(self): + return (self._manifest,) + + +def test_module_operational_checks_cache_and_force() -> None: + routes._module_check_cache.clear() + calls = 0 + + def provider() -> OperationalCheck: + nonlocal calls + calls += 1 + return OperationalCheck("example.roundtrip", "Example", "ok", "Passed") + + registration = OperationalCheckProviderRegistration( + module_id="example", + check_id="example.roundtrip", + provider=provider, + ) + registry = _Registry(registration) + + assert routes._module_operational_checks(registry, force=False)[0]["state"] == "ok" + assert routes._module_operational_checks(registry, force=False)[0]["state"] == "ok" + assert calls == 1 + routes._module_operational_checks(registry, force=True) + assert calls == 2 + + +def test_module_operational_check_failure_is_isolated() -> None: + routes._module_check_cache.clear() + + def provider() -> OperationalCheck: + raise RuntimeError("secret detail") + + result = routes._module_operational_checks( + _Registry(OperationalCheckProviderRegistration( + module_id="example", + check_id="example.failed", + provider=provider, + )), + force=True, + )[0] + + assert result["state"] == "error" + assert "secret detail" not in result["detail"] + diff --git a/webui/src/api/ops.ts b/webui/src/api/ops.ts index 90af109..dc8b729 100644 --- a/webui/src/api/ops.ts +++ b/webui/src/api/ops.ts @@ -6,6 +6,7 @@ export type OpsCheck = { state: "ok" | "warning" | "error" | string; detail: string; readiness_critical?: boolean; + metrics?: Record; }; export type OpsDeploymentProfile = { @@ -52,6 +53,15 @@ export type OpsStatus = { }; database_url: string; file_storage_backend: string; + worker_metrics: { + workers?: number; + active_tasks?: number; + expected_queues?: string[]; + active_queues?: string[]; + missing_queues?: string[]; + queue_depths?: Record; + }; + operational_probe_count: number; }; readiness: { ready: boolean; @@ -85,3 +95,7 @@ export type OpsStatus = { export function fetchOpsStatus(settings: ApiSettings): Promise { return apiFetch(settings, "/api/v1/ops/status"); } + +export function runOpsChecks(settings: ApiSettings): Promise { + return apiFetch(settings, "/api/v1/ops/checks/run", { method: "POST" }); +} diff --git a/webui/src/features/ops/OpsHealthWidget.tsx b/webui/src/features/ops/OpsHealthWidget.tsx index d103ac1..0b32606 100644 --- a/webui/src/features/ops/OpsHealthWidget.tsx +++ b/webui/src/features/ops/OpsHealthWidget.tsx @@ -29,13 +29,21 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap const warningCount = checks.filter((item) => item.state === "warning").length; const errorCount = checks.filter((item) => item.state === "error").length; const ready = status?.readiness.ready ?? false; + const workerMetrics = status?.summary.worker_metrics; + const queueDepths = workerMetrics?.queue_depths ?? {}; + const queuedTasks = Object.values(queueDepths).reduce((sum, value) => sum + value, 0); + const missingQueues = workerMetrics?.missing_queues ?? []; return ( {error && {error}}
- + + + + +
{status?.readiness.blockers.length ? @@ -51,4 +59,3 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap } ); } - diff --git a/webui/src/features/ops/OpsPage.tsx b/webui/src/features/ops/OpsPage.tsx index 29c2030..a238ac7 100644 --- a/webui/src/features/ops/OpsPage.tsx +++ b/webui/src/features/ops/OpsPage.tsx @@ -11,11 +11,14 @@ import { PageTitle, StatusBadge, adminErrorMessage, + hasAnyScope, type ApiSettings, + type AuthInfo, type DataGridColumn } from "@govoplan/core-webui"; import { fetchOpsStatus, + runOpsChecks, type OpsCheck, type OpsDeploymentProfile, type OpsGovernanceModule, @@ -23,7 +26,7 @@ import { type OpsStatus } from "../../api/ops"; -export default function OpsPage({ settings }: {settings: ApiSettings;}) { +export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: AuthInfo;}) { const [status, setStatus] = useState(null); const [loading, setLoading] = useState(true); const [error, setError] = useState(""); @@ -40,12 +43,25 @@ export default function OpsPage({ settings }: {settings: ApiSettings;}) { } } + async function runChecks() { + setLoading(true); + setError(""); + try { + setStatus(await runOpsChecks(settings)); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + setLoading(false); + } + } + useEffect(() => {void load();}, [settings.apiBaseUrl, settings.apiKey, settings.accessToken]); const checks = status?.checks ?? []; const warningCount = checks.filter((item) => item.state === "warning").length; const errorCount = checks.filter((item) => item.state === "error").length; const ready = status?.readiness.ready ?? false; + const canRunChecks = hasAnyScope(auth, ["ops:operations:run", "system:settings:write"]); return ( @@ -56,6 +72,7 @@ export default function OpsPage({ settings }: {settings: ApiSettings;}) {

i18n:govoplan-ops.runtime_health_deployment_profile_worker_split_a.55340156

+ {canRunChecks ? : null}
@@ -69,7 +86,7 @@ export default function OpsPage({ settings }: {settings: ApiSettings;}) { - + @@ -209,3 +226,11 @@ function stateTone(state: string): string { if (state === "error") return "error"; return "inactive"; } + +function workerMetricDetail(status: OpsStatus | null): string { + if (!status?.summary.celery_enabled) return "i18n:govoplan-ops.celery_worker_setting.323d7737"; + const metrics = status.summary.worker_metrics; + if (metrics.missing_queues?.length) return `Missing queues: ${metrics.missing_queues.join(", ")}`; + const queued = Object.values(metrics.queue_depths ?? {}).reduce((total, value) => total + value, 0); + return `${metrics.active_tasks ?? 0} active ยท ${queued} queued`; +} diff --git a/webui/src/module.ts b/webui/src/module.ts index 0953da7..645eea3 100644 --- a/webui/src/module.ts +++ b/webui/src/module.ts @@ -40,7 +40,7 @@ export const opsModule: PlatformWebModule = { ], navItems: [{ to: "/ops", label: "i18n:govoplan-ops.ops.907a54c2", iconName: "activity", anyOf: opsReadScopes, order: 890 }], routes: [ - { path: "/ops", anyOf: opsReadScopes, order: 890, render: ({ settings }) => createElement(OpsPage, { settings }) }], + { path: "/ops", anyOf: opsReadScopes, order: 890, render: ({ settings, auth }) => createElement(OpsPage, { settings, auth }) }], uiCapabilities: { "dashboard.widgets": dashboardWidgets }