diff --git a/docs/SCALABILITY_PROFILES.md b/docs/SCALABILITY_PROFILES.md index fbac8e8..782e0b4 100644 --- a/docs/SCALABILITY_PROFILES.md +++ b/docs/SCALABILITY_PROFILES.md @@ -76,6 +76,14 @@ Promote from local development to a production-like profile when a feature depends on PostgreSQL, Redis, Celery, module package lifecycle, or durable file storage. +The Operations page is the canonical monitoring surface. Its summary reports +worker and queue coverage, queue depth, active tasks, local filesystem capacity +when observable, backup/restore evidence, runtime-node skew, and recovery-ledger +operations split into active, failed, recovery-required, and outcome-unknown +states. S3 capacity remains provider-owned unless a configured module check +supplies bounded usage metrics; Ops must not enumerate an object store merely to +render a dashboard. + Promote from a single-process profile to a split-worker profile when queued work becomes part of normal operation: diff --git a/src/govoplan_ops/backend/api/v1/routes.py b/src/govoplan_ops/backend/api/v1/routes.py index 281f32c..3db733d 100644 --- a/src/govoplan_ops/backend/api/v1/routes.py +++ b/src/govoplan_ops/backend/api/v1/routes.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +import shutil import threading import time from collections.abc import Mapping @@ -203,6 +204,9 @@ def _ops_status_payload( registry, force=force_module_checks, ) + storage_check = _storage_check() + backup_check = _backup_restore_check() + recovery_metrics = runtime_cluster.get("recovery", {}).get("metrics", {}) checks = [ _check( "module_registry", @@ -221,8 +225,8 @@ def _ops_status_payload( worker_check, database_capacity, _runtime_cluster_check(runtime_cluster), - _storage_check(), - _backup_restore_check(), + storage_check, + backup_check, _deployment_security_check(current_profile), *module_checks, ] @@ -246,12 +250,19 @@ def _ops_status_payload( core_settings.database_connection_available ), "file_storage_backend": core_settings.file_storage_backend, + "storage_metrics": storage_check.get("metrics", {}), + "backup_state": backup_check.get("state", "unknown"), "worker_metrics": worker_check.get("metrics", {}), "operational_probe_count": len(module_checks), "runtime_node_count": len(runtime_cluster.get("nodes", [])), "recovery_required_count": runtime_cluster.get("recovery", {}).get( "requires_attention", 0 ), + "failed_operation_count": int(recovery_metrics.get("failed") or 0), + "outcome_unknown_count": int( + recovery_metrics.get("outcome_unknown") or 0 + ), + "active_operation_count": int(recovery_metrics.get("active") or 0), }, "readiness": readiness, "checks": checks, @@ -288,7 +299,11 @@ def _runtime_cluster_status( "detail": f"Runtime coordination query failed: {exc}", "state_profile": core_settings.state_profile, "nodes": [], - "recovery": {"operations": [], "requires_attention": 0}, + "recovery": { + "operations": [], + "requires_attention": 0, + "metrics": _recovery_metrics([]), + }, } active_nodes = [ node for node in nodes if node["state"] == "active" and not node["stale"] @@ -349,16 +364,7 @@ def _runtime_cluster_status( } for operation in operations ] - requires_attention = sum( - operation["status"] - in { - "outcome_unknown", - "recovery_required", - "recovering", - "manual_intervention", - } - for operation in operation_payloads - ) + recovery_metrics = _recovery_metrics(operation_payloads) return { "available": True, "detail": "Shared runtime directory is available.", @@ -388,11 +394,32 @@ def _runtime_cluster_status( "nodes": nodes, "recovery": { "operations": operation_payloads, - "requires_attention": requires_attention, + "requires_attention": recovery_metrics["requires_attention"], + "metrics": recovery_metrics, }, } +def _recovery_metrics(operations: list[dict[str, Any]]) -> dict[str, int]: + statuses = [str(operation.get("status") or "") for operation in operations] + failed = sum(value in {"failed", "manual_intervention"} for value in statuses) + outcome_unknown = statuses.count("outcome_unknown") + recovery_required = sum( + value in {"recovery_required", "recovering"} for value in statuses + ) + active = sum( + value in {"planned", "prepared", "running", "recovering"} + for value in statuses + ) + return { + "failed": failed, + "outcome_unknown": outcome_unknown, + "recovery_required": recovery_required, + "active": active, + "requires_attention": failed + outcome_unknown + recovery_required, + } + + def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]: if not cluster.get("available"): return _check( @@ -841,6 +868,10 @@ def _storage_check() -> dict[str, Any]: if configured else "S3 file storage needs endpoint and bucket settings.", readiness_critical=not configured, + metrics={ + "backend": "s3", + "capacity_observable": False, + }, ) root = Path(str(core_settings.file_storage_local_root or "runtime/files")) if root.exists() and root.is_dir(): @@ -851,12 +882,14 @@ def _storage_check() -> dict[str, Any]: if writable else f"Local file storage root is not writable: {root}" ) + metrics = _local_storage_capacity(root) return _check( "file_storage", "File storage", state, detail, readiness_critical=not writable, + metrics=metrics, ) return _check( "file_storage", @@ -864,9 +897,28 @@ def _storage_check() -> dict[str, Any]: "warning", f"Local file storage root does not exist yet: {root}", readiness_critical=False, + metrics={"backend": "local", "capacity_observable": False}, ) +def _local_storage_capacity(root: Path) -> dict[str, Any]: + try: + usage = shutil.disk_usage(root) + except OSError: + return {"backend": "local", "capacity_observable": False} + used_percent = ( + round((usage.used / usage.total) * 100, 1) if usage.total else 0.0 + ) + return { + "backend": "local", + "capacity_observable": True, + "capacity_total_bytes": int(usage.total), + "capacity_used_bytes": int(usage.used), + "capacity_free_bytes": int(usage.free), + "capacity_used_percent": used_percent, + } + + def _module_operational_checks( registry: PlatformRegistry, *, diff --git a/tests/test_operational_checks.py b/tests/test_operational_checks.py index 96765a3..cd12c47 100644 --- a/tests/test_operational_checks.py +++ b/tests/test_operational_checks.py @@ -1,6 +1,7 @@ from __future__ import annotations from dataclasses import dataclass +from pathlib import Path from govoplan_core.core.operations import ( OperationalCheck, @@ -124,3 +125,34 @@ def test_database_capacity_check_enforces_shared_rendered_budget( overrun = routes._database_capacity_check() assert overrun["state"] == "error" assert overrun["readiness_critical"] is True + + +def test_recovery_metrics_separate_failure_unknown_and_active_work() -> None: + metrics = routes._recovery_metrics( + [ + {"status": "running"}, + {"status": "failed"}, + {"status": "outcome_unknown"}, + {"status": "recovery_required"}, + {"status": "manual_intervention"}, + ] + ) + + assert metrics == { + "failed": 2, + "outcome_unknown": 1, + "recovery_required": 1, + "active": 1, + "requires_attention": 4, + } + + +def test_local_storage_capacity_reports_bounded_filesystem_metrics( + tmp_path: Path, +) -> None: + metrics = routes._local_storage_capacity(tmp_path) + + assert metrics["backend"] == "local" + assert metrics["capacity_observable"] is True + assert metrics["capacity_total_bytes"] > 0 + assert 0 <= metrics["capacity_used_percent"] <= 100 diff --git a/webui/src/api/ops.ts b/webui/src/api/ops.ts index 7afaf45..61a11a4 100644 --- a/webui/src/api/ops.ts +++ b/webui/src/api/ops.ts @@ -137,6 +137,13 @@ export type OpsRuntimeCluster = { recovery: { operations: OpsRecoveryOperation[]; requires_attention: number; + metrics?: { + failed?: number; + outcome_unknown?: number; + recovery_required?: number; + active?: number; + requires_attention?: number; + }; }; }; @@ -156,6 +163,15 @@ export type OpsStatus = { database_connection_peak?: number | null; database_connection_available?: number | null; file_storage_backend: string; + storage_metrics?: { + backend?: string; + capacity_observable?: boolean; + capacity_total_bytes?: number; + capacity_used_bytes?: number; + capacity_free_bytes?: number; + capacity_used_percent?: number; + }; + backup_state?: string; worker_metrics: { workers?: number; active_tasks?: number; @@ -167,6 +183,9 @@ export type OpsStatus = { operational_probe_count: number; runtime_node_count: number; recovery_required_count: number; + failed_operation_count?: number; + outcome_unknown_count?: number; + active_operation_count?: number; }; readiness: { ready: boolean; diff --git a/webui/src/features/ops/OpsHealthWidget.tsx b/webui/src/features/ops/OpsHealthWidget.tsx index 0b32606..bfae614 100644 --- a/webui/src/features/ops/OpsHealthWidget.tsx +++ b/webui/src/features/ops/OpsHealthWidget.tsx @@ -33,6 +33,7 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap const queueDepths = workerMetrics?.queue_depths ?? {}; const queuedTasks = Object.values(queueDepths).reduce((sum, value) => sum + value, 0); const missingQueues = workerMetrics?.missing_queues ?? []; + const storageUsage = status?.summary.storage_metrics?.capacity_used_percent; return ( @@ -42,7 +43,10 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap - + = 90 ? "danger" : storageUsage !== undefined && storageUsage >= 75 ? "warning" : "neutral"} detail="Managed Files backend capacity" /> + + + diff --git a/webui/src/features/ops/OpsPage.tsx b/webui/src/features/ops/OpsPage.tsx index fe7b05d..53fc087 100644 --- a/webui/src/features/ops/OpsPage.tsx +++ b/webui/src/features/ops/OpsPage.tsx @@ -97,6 +97,9 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: 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"]); + const queueDepths = status?.summary.worker_metrics.queue_depths ?? {}; + const queuedTasks = Object.values(queueDepths).reduce((total, value) => total + value, 0); + const storageUsage = status?.summary.storage_metrics?.capacity_used_percent; return ( @@ -122,6 +125,11 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: + + = 90 ? "danger" : storageUsage !== undefined && storageUsage >= 75 ? "warning" : "neutral"} detail={storageMetricDetail(status)} /> + + + @@ -506,3 +514,18 @@ function workerMetricDetail(status: OpsStatus | null): string { const queued = Object.values(metrics.queue_depths ?? {}).reduce((total, value) => total + value, 0); return `${metrics.active_tasks ?? 0} active ยท ${queued} queued`; } + +function storageMetricDetail(status: OpsStatus | null): string { + const metrics = status?.summary.storage_metrics; + if (!metrics?.capacity_observable) { + return `${status?.summary.file_storage_backend ?? "Storage"} capacity is provider-managed or unavailable`; + } + return `${formatBytes(metrics.capacity_used_bytes ?? 0)} used of ${formatBytes(metrics.capacity_total_bytes ?? 0)}`; +} + +function formatBytes(value: number): string { + if (!Number.isFinite(value) || value <= 0) return "0 B"; + const units = ["B", "KB", "MB", "GB", "TB", "PB"]; + const index = Math.min(Math.floor(Math.log(value) / Math.log(1024)), units.length - 1); + return `${(value / 1024 ** index).toFixed(index ? 1 : 0)} ${units[index]}`; +}