GovOPlaN/govoplan-ops
2
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Add module probes and worker telemetry

Browse Source
This commit is contained in:
zemion
2026-07-31 22:48:07 +02:00
parent b84cab1aa4
commit 05ce4dc8ec
7 changed files with 422 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_ops/backend/api/v1/routes.py
+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]:
src/govoplan_ops/backend/manifest.py
+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),),