Add module probes and worker telemetry

This commit is contained in:
2026-07-31 22:48:07 +02:00
parent b84cab1aa4
commit 05ce4dc8ec
7 changed files with 422 additions and 14 deletions
+292 -9
View File
@@ -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]:
+14
View File
@@ -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),),
+65
View File
@@ -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"]
+14
View File
@@ -6,6 +6,7 @@ export type OpsCheck = {
state: "ok" | "warning" | "error" | string;
detail: string;
readiness_critical?: boolean;
metrics?: Record<string, unknown>;
};
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<string, number>;
};
operational_probe_count: number;
};
readiness: {
ready: boolean;
@@ -85,3 +95,7 @@ export type OpsStatus = {
export function fetchOpsStatus(settings: ApiSettings): Promise<OpsStatus> {
return apiFetch(settings, "/api/v1/ops/status");
}
export function runOpsChecks(settings: ApiSettings): Promise<OpsStatus> {
return apiFetch(settings, "/api/v1/ops/checks/run", { method: "POST" });
}
+9 -2
View File
@@ -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 (
<LoadingFrame loading={loading} label="Loading operations status">
{error && <DismissibleAlert tone="warning" resetKey={error}>{error}</DismissibleAlert>}
<div className="metric-grid inside dashboard-widget-metrics">
<MetricCard label="Readiness" value={ready ? "ready" : "blocked"} tone={ready ? "good" : "danger"} detail={status?.readiness.profile ?? "-"} />
<MetricCard label="Workers" value={status?.summary.celery_enabled ? "split" : "off"} tone={status?.summary.celery_enabled ? "good" : "warning"} detail={status?.summary.celery_queues?.length ? status.summary.celery_queues.join(", ") : "single-process"} />
<MetricCard label="Workers" value={status?.summary.celery_enabled ? workerMetrics?.workers ?? 0 : "off"} tone={status?.summary.celery_enabled && !missingQueues.length ? "good" : "warning"} detail={missingQueues.length ? `${missingQueues.length} queue(s) without a consumer` : status?.summary.celery_enabled ? "All configured queues covered" : "Single-process mode"} />
<MetricCard label="Active tasks" value={workerMetrics?.active_tasks ?? 0} tone="info" detail={status?.summary.celery_enabled ? "Reported by live workers" : "Workers disabled"} />
<MetricCard label="Queued tasks" value={queuedTasks} tone={queuedTasks ? "warning" : "neutral"} detail={Object.keys(queueDepths).length ? `${Object.keys(queueDepths).length} measured queue(s)` : "Queue depth unavailable"} />
<MetricCard label="Storage" value={status?.summary.file_storage_backend ?? "-"} tone="neutral" detail="Managed Files backend" />
<MetricCard label="Probes" value={status?.summary.operational_probe_count ?? 0} tone="neutral" detail="Module-owned operational checks" />
<MetricCard label="Warnings" value={warningCount + errorCount} tone={errorCount ? "danger" : warningCount ? "warning" : "good"} detail="Current health checks" />
</div>
{status?.readiness.blockers.length ?
@@ -51,4 +59,3 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap
}
</LoadingFrame>);
}
+27 -2
View File
@@ -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<OpsStatus | null>(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 (
<PageScrollViewport>
@@ -56,6 +72,7 @@ export default function OpsPage({ settings }: {settings: ApiSettings;}) {
<p>i18n:govoplan-ops.runtime_health_deployment_profile_worker_split_a.55340156</p>
</div>
<div className="button-row compact-actions">
{canRunChecks ? <Button variant="primary" onClick={() => void runChecks()} disabled={loading}>Run probes</Button> : null}
<Button onClick={() => void load()} disabled={loading}><RefreshCw size={16} /> i18n:govoplan-ops.reload.cce71553</Button>
</div>
</div>
@@ -69,7 +86,7 @@ export default function OpsPage({ settings }: {settings: ApiSettings;}) {
<MetricCard label="i18n:govoplan-ops.modules.04e9462c" value={status?.summary.module_count ?? 0} tone="neutral" detail="i18n:govoplan-ops.enabled_in_the_runtime_registry.d2c6142d" />
<MetricCard label="i18n:govoplan-ops.permissions.842c35eb" value={status?.governance.summary.permission_count ?? 0} tone="neutral" detail="i18n:govoplan-ops.declared_governance_permissions.d08d3bf1" />
<MetricCard label="i18n:govoplan-ops.policies.e7800f56" value={status?.governance.summary.policy_count ?? 0} tone="neutral" detail="i18n:govoplan-ops.registered_policy_capabilities.112a2b64" />
<MetricCard label="i18n:govoplan-ops.workers.b6ef3acd" value={status?.summary.celery_enabled ? "split" : "off"} tone={status?.summary.celery_enabled ? "good" : "warning"} detail={status?.summary.celery_queues?.length ? status.summary.celery_queues.join(", ") : "i18n:govoplan-ops.celery_worker_setting.323d7737"} />
<MetricCard label="i18n:govoplan-ops.workers.b6ef3acd" value={status?.summary.celery_enabled ? status.summary.worker_metrics.workers ?? 0 : "off"} tone={status?.summary.celery_enabled && !(status.summary.worker_metrics.missing_queues?.length) ? "good" : "warning"} detail={workerMetricDetail(status)} />
<MetricCard label="i18n:govoplan-ops.redis.5eaa1f2f" value={status?.summary.redis_url ? "configured" : "-"} tone={status?.summary.celery_enabled ? "info" : "neutral"} detail={status?.summary.redis_url ?? "-"} />
<MetricCard label="i18n:govoplan-ops.warnings.1430f976" value={warningCount + errorCount} tone={errorCount ? "danger" : warningCount ? "warning" : "good"} detail="i18n:govoplan-ops.current_health_checks.7830bccf" />
</div>
@@ -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`;
}
+1 -1
View File
@@ -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
}