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

feat(ops): show worker and queue readiness

Browse Source
This commit is contained in:
zemion
2026-08-19 20:52:44 +02:00
parent f84097224b
commit 6ecb94c99e
12 changed files with 559 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_ops/backend/api/v1/routes.py
+179 -128
View File
@@ -23,7 +23,11 @@ from govoplan_core.core.module_installer import (
list_module_installer_runs,
read_module_installer_run,
)
from govoplan_core.core.operations import OperationalCheckProviderRegistration
from govoplan_core.core.operations import (
OperationalCheckProviderRegistration,
RuntimeWorkStatusContext,
RuntimeWorkStatusProviderRegistration,
)
from govoplan_core.core.recovery import (
RecoveryOperation,
)
@@ -48,8 +52,8 @@ from govoplan_ops.backend.infrastructure import deployment_capability_status
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()
_runtime_work_cache: dict[str, tuple[float, dict[str, Any]]] = {}
_runtime_work_cache_lock = threading.Lock()
class RuntimeDrainRequest(BaseModel):
@@ -192,9 +196,7 @@ def _ops_status_payload(
if isinstance(database.get("maintenance_mode"), dict)
else {"enabled": False, "message": None}
)
redis_check = _redis_check()
current_profile = _current_profile()
worker_check = _worker_check()
database_capacity = _database_capacity_check()
expected_composition_hash = runtime_composition_hash(
tuple(manifest.id for manifest in registry.manifests())
@@ -202,6 +204,18 @@ def _ops_status_payload(
runtime_cluster = _runtime_cluster_status(
expected_composition_hash=expected_composition_hash
)
runtime_work = _runtime_work_statuses(
registry,
RuntimeWorkStatusContext(
profile=current_profile,
observed_at=datetime.now(UTC),
stale_after_seconds=core_settings.runtime_stale_after_seconds,
runtime_nodes=tuple(runtime_cluster.get("nodes", [])),
),
force=force_module_checks,
)
worker_check = _runtime_work_check(runtime_work, current_profile)
redis_check = _runtime_backend_check(runtime_work)
module_checks = _module_operational_checks(
registry,
force=force_module_checks,
@@ -278,6 +292,7 @@ def _ops_status_payload(
"deployment_profiles": _deployment_profiles(current_profile),
"sizing": _sizing_assumptions(),
"runtime_cluster": runtime_cluster,
"runtime_work": runtime_work,
"infrastructure": infrastructure,
}
@@ -753,142 +768,178 @@ def _maintenance_check(maintenance_mode: dict[str, Any]) -> dict[str, Any]:
)
def _redis_check() -> dict[str, Any]:
if not core_settings.celery_enabled:
return _check(
"redis_broker",
"Redis broker",
"inactive",
"Redis is not required while Celery is disabled.",
)
try:
from redis import Redis
def _runtime_work_statuses(
registry: PlatformRegistry,
context: RuntimeWorkStatusContext,
*,
force: bool = False,
) -> list[dict[str, Any]]:
registrations: list[RuntimeWorkStatusProviderRegistration] = []
for manifest in registry.manifests():
registrations.extend(manifest.runtime_work_status_providers)
client = Redis.from_url(
core_settings.redis_url, socket_connect_timeout=0.75, socket_timeout=0.75
)
client.ping()
return _check(
"redis_broker",
"Redis broker",
"ok",
f"Redis broker reachable at {_redact_url(core_settings.redis_url)}.",
)
except Exception as exc: # noqa: BLE001 - diagnostic endpoint should report the concrete failure.
return _check(
"redis_broker",
"Redis broker",
"error",
f"Redis broker check failed: {exc}",
readiness_critical=True,
)
results: list[dict[str, Any]] = []
seen: set[str] = set()
now = time.monotonic()
for registration in registrations:
key = f"{registration.module_id}:{registration.provider_id}"
if registration.provider_id in seen:
results.append(_unavailable_runtime_work(registration, "Duplicate provider id."))
continue
seen.add(registration.provider_id)
with _runtime_work_cache_lock:
cached = _runtime_work_cache.get(key)
if (
not force
and cached is not None
and now - cached[0] < max(1, registration.cache_seconds)
):
results.append(dict(cached[1]))
continue
try:
status_result = registration.provider(context)
if status_result.provider_id != registration.provider_id:
raise ValueError("provider id mismatch")
result = status_result.as_dict()
except Exception: # noqa: BLE001 - provider failures remain isolated and sanitized.
result = _unavailable_runtime_work(
registration,
"The runtime-work provider failed without usable evidence.",
)
with _runtime_work_cache_lock:
_runtime_work_cache[key] = (now, result)
results.append(dict(result))
return results
def _worker_check() -> dict[str, Any]:
if not core_settings.celery_enabled:
def _unavailable_runtime_work(
registration: RuntimeWorkStatusProviderRegistration,
detail: str,
) -> dict[str, Any]:
return {
"provider_id": registration.provider_id,
"label": registration.provider_id,
"backend": "unavailable",
"enabled": None,
"configured": None,
"state": "unreachable",
"detail": detail,
"observed_at": datetime.now(UTC).isoformat(),
"active_workers": None,
"last_heartbeat_at": None,
"queue_depths": {},
"active_work": None,
"reserved_work": None,
"failures": None,
"stale_after_seconds": None,
"guidance": "Restore or configure the owning status provider.",
}
def _runtime_work_check(
statuses: list[dict[str, Any]],
profile: str,
) -> dict[str, Any]:
if not statuses:
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": []},
"No runtime-work status provider is installed; worker and queue health is unavailable.",
readiness_critical=profile not in {"local-dev", "development"},
metrics={"workers": None, "active_tasks": None, "reserved_tasks": None},
)
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.
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))
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)
+ "."
development_profile = profile in {"local-dev", "development"}
state_rank = {
"unreachable": 8,
"stale": 7,
"degraded": 6,
"unconfigured": 5,
"starting": 4,
"busy": 3,
"healthy": 2,
"idle": 1,
"disabled": 0 if development_profile else 6,
}
worst = max(statuses, key=lambda item: state_rank.get(str(item.get("state")), 8))
runtime_state = str(worst.get("state") or "unreachable")
disabled_in_dev = runtime_state == "disabled" and development_profile
check_state = (
"inactive"
if disabled_in_dev
else "ok"
if runtime_state in {"healthy", "idle", "busy"}
else "warning"
if runtime_state in {"disabled", "starting", "degraded"}
else "error"
)
queue_depths: dict[str, int | None] = {}
for item in statuses:
depths = item.get("queue_depths")
if isinstance(depths, Mapping):
queue_depths.update(
{
str(queue): int(value) if isinstance(value, int) else None
for queue, value in depths.items()
}
)
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(
return _check(
"worker_split",
"Background workers",
"error",
"Celery is enabled, but no workers replied to heartbeat.",
readiness_critical=True,
metrics={"workers": 0, "active_tasks": 0},
check_state,
str(worst.get("detail") or "Runtime-work status is unavailable."),
readiness_critical=(
not disabled_in_dev
and runtime_state
in {"disabled", "unconfigured", "degraded", "stale", "unreachable"}
),
metrics={
"state": runtime_state,
"workers": _sum_known(statuses, "active_workers"),
"active_tasks": _sum_known(statuses, "active_work"),
"reserved_tasks": _sum_known(statuses, "reserved_work"),
"failures": _sum_known(statuses, "failures"),
"queue_depths": queue_depths,
},
)
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,
def _runtime_backend_check(statuses: list[dict[str, Any]]) -> dict[str, Any]:
if not statuses or all(item.get("enabled") is False for item in statuses):
return _check(
"redis_broker",
"Work backend",
"inactive",
"No enabled runtime-work backend requires a connectivity assertion.",
)
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)}
if any(item.get("state") == "unreachable" for item in statuses):
return _check(
"redis_broker",
"Work backend",
"error",
"An enabled runtime-work backend is unreachable.",
readiness_critical=True,
)
if any(item.get("configured") is not True for item in statuses if item.get("enabled") is True):
return _check(
"redis_broker",
"Work backend",
"warning",
"An enabled runtime-work backend is not fully configured.",
readiness_critical=True,
)
return _check(
"redis_broker",
"Work backend",
"ok",
"Enabled runtime-work providers returned bounded status evidence.",
)
def _sum_known(statuses: list[dict[str, Any]], field: str) -> int | None:
values = [item.get(field) for item in statuses]
known = [int(value) for value in values if isinstance(value, int)]
return sum(known) if known else None
def _storage_check() -> dict[str, Any]:
@@ -1346,9 +1397,9 @@ def _readiness(
def _current_profile() -> str:
app_env = str(core_settings.app_env or "").lower()
if app_env == "dev" and core_settings.celery_enabled:
if app_env in {"dev", "test"} and core_settings.celery_enabled:
return "production-like-dev"
if app_env == "dev":
if app_env in {"dev", "test"}:
return "local-dev"
if core_settings.celery_enabled:
return "split-worker"