diff --git a/.gitignore b/.gitignore index 2a33867..0046af2 100644 --- a/.gitignore +++ b/.gitignore @@ -16,11 +16,13 @@ webui/dist/ .policy-test-build/ .template-preview-test-build/ .import-test-build/ +.runtime-status-test-build/ webui/.component-test-build/ webui/.module-test-build/ webui/.policy-test-build/ webui/.template-preview-test-build/ webui/.import-test-build/ +webui/.runtime-status-test-build/ # GovOPlaN shared ignore rules from govoplan-core # ---> Node diff --git a/docs/SCALABILITY_PROFILES.md b/docs/SCALABILITY_PROFILES.md index 2bad360..1424219 100644 --- a/docs/SCALABILITY_PROFILES.md +++ b/docs/SCALABILITY_PROFILES.md @@ -40,6 +40,11 @@ The Ops API reports: stale state, and drain state - configured versus active API and worker replica counts - active worker-pool names, exact queue coverage, and missing queue owners +- provider-neutral worker state (`disabled`, `unconfigured`, `starting`, + `healthy`, `idle`, `busy`, `degraded`, `stale`, or `unreachable`), with the + configured backend, latest heartbeat age, and stale threshold +- queue depth, active/reserved work, and failure count only when reported by + the provider; unavailable values are never interpreted as zero or healthy - release/module-composition skew and software-version skew across active nodes - rendered PostgreSQL connection peak, declared server limit, and operator reserve - recovery operation status, mode, checkpoint count, and last update @@ -90,6 +95,15 @@ 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. +The WebUI polls this read-only projection every 15 seconds only while its page +is visible. It permits one request at a time, stops the timer when the document +is hidden, and performs one refresh when visibility returns. Runtime providers +register through the Core contract; Ops itself does not import Celery, Redis, +or module-owned job implementations. A local development profile may +intentionally disable workers without becoming unready. A production or other +non-development profile treats a disabled, unconfigured, stale, or unreachable +provider as readiness-critical. + 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 27171a5..3f3639d 100644 --- a/src/govoplan_ops/backend/api/v1/routes.py +++ b/src/govoplan_ops/backend/api/v1/routes.py @@ -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" diff --git a/src/govoplan_ops/backend/manifest.py b/src/govoplan_ops/backend/manifest.py index 66bfb13..c66069d 100644 --- a/src/govoplan_ops/backend/manifest.py +++ b/src/govoplan_ops/backend/manifest.py @@ -20,6 +20,8 @@ from govoplan_core.core.provider_governance import ( ModuleArchitectureDocumentation, ModuleMaturityEvidence, ) +from govoplan_core.core.operations import RuntimeWorkStatusProviderRegistration +from govoplan_core.core.runtime_work import celery_runtime_work_status from govoplan_core.core.views import ViewSurface OPS_READ_SCOPE = "ops:operations:read" @@ -124,12 +126,20 @@ manifest = ModuleManifest( level="system", ), ), + runtime_work_status_providers=( + RuntimeWorkStatusProviderRegistration( + module_id="ops", + provider_id="core.celery", + provider=celery_runtime_work_status, + cache_seconds=15, + ), + ), documentation=( DocumentationTopic( id="ops.health-governance-and-sizing", title="Inspect platform health and deployment posture", summary="Ops combines module-owned health checks with deployment profile, governance inventory, worker assumptions, and sizing guidance.", - body="Read-only status distinguishes configured capabilities from healthy integrations. When the deployment mounts a signed or locally generated non-secret infrastructure capability receipt, Ops shows whether PostgreSQL, Redis, SMTP, file storage, load balancing, and ingress are configured, externally supplied, available but unconfigured, or unavailable. Secret values never cross this boundary; only stable environment or credential-envelope references may be disclosed. Pending post-install tasks remain visible with a stable resume key. Authorized operators can run bounded probes; a probe must not perform unbounded business work or silently repair data. Use readiness and worker results when diagnosing a node, and use the deployment profile and sizing assumptions when planning horizontal capacity.", + body="Read-only status distinguishes configured capabilities from healthy integrations. Worker and queue providers use a Core runtime-status contract, so Ops never imports a provider backend. The surface distinguishes intentionally disabled, unconfigured, starting, healthy with unsupported queue depth, measured idle, busy, degraded, stale, and unreachable states. It shows enabled/configured state, backend, workers, heartbeat age and stale threshold, queue depth, active/reserved work, and failures only when each value is actually reported; unavailable values are never rendered as zero or healthy. Local development treats intentionally disabled workers as expected, while production profiles require an enabled, configured, reachable provider before queue-backed work is accepted. Polling is bounded to one request, pauses while the page is hidden, and refreshes on return. When the deployment mounts a signed or locally generated non-secret infrastructure capability receipt, Ops shows whether PostgreSQL, Redis, SMTP, file storage, load balancing, and ingress are configured, externally supplied, available but unconfigured, or unavailable. Secret values never cross this boundary; only stable environment or credential-envelope references may be disclosed. Pending post-install tasks remain visible with a stable resume key. Authorized operators can run bounded probes; a probe must not perform unbounded business work or silently repair data. Use readiness and worker results when diagnosing a node, and use the deployment profile and sizing assumptions when planning horizontal capacity.", documentation_types=("admin", "user"), audience=("operator", "system_admin"), related_modules=("audit", "docs", "notifications"), @@ -139,6 +149,7 @@ manifest = ModuleManifest( "ops.page", "ops.page.summary", "ops.page.health", + "ops.page.runtime", "ops.page.governance", "ops.page.deployment", "ops.page.sizing", diff --git a/tests/test_operational_checks.py b/tests/test_operational_checks.py index 88522f9..98495e7 100644 --- a/tests/test_operational_checks.py +++ b/tests/test_operational_checks.py @@ -7,6 +7,9 @@ from pathlib import Path from govoplan_core.core.operations import ( OperationalCheck, OperationalCheckProviderRegistration, + RuntimeWorkStatus, + RuntimeWorkStatusContext, + RuntimeWorkStatusProviderRegistration, ) from govoplan_ops.backend.api.v1 import routes @@ -14,6 +17,7 @@ from govoplan_ops.backend.api.v1 import routes @dataclass class _Manifest: operational_check_providers: tuple[OperationalCheckProviderRegistration, ...] + runtime_work_status_providers: tuple[RuntimeWorkStatusProviderRegistration, ...] = () class _Registry: @@ -68,6 +72,98 @@ def test_module_operational_check_failure_is_isolated() -> None: assert "secret detail" not in result["detail"] +def test_runtime_work_provider_cache_force_and_unknown_metrics() -> None: + routes._runtime_work_cache.clear() + calls = 0 + + def provider(context: RuntimeWorkStatusContext) -> RuntimeWorkStatus: + nonlocal calls + calls += 1 + return RuntimeWorkStatus( + provider_id="example.queue", + label="Example queue", + backend="Example", + enabled=True, + configured=True, + state="healthy", + detail="Workers answered; queue depth unsupported.", + observed_at=context.observed_at, + active_workers=1, + queue_depths={"example": None}, + ) + + registry = _Registry() + registry._manifest.runtime_work_status_providers = ( # type: ignore[misc] + RuntimeWorkStatusProviderRegistration( + module_id="example", + provider_id="example.queue", + provider=provider, + ), + ) + context = RuntimeWorkStatusContext( + profile="split-worker", + observed_at=datetime.now(UTC), + stale_after_seconds=60, + ) + + first = routes._runtime_work_statuses(registry, context) + second = routes._runtime_work_statuses(registry, context) + forced = routes._runtime_work_statuses(registry, context, force=True) + + assert first[0]["queue_depths"] == {"example": None} + assert second == first + assert forced[0]["state"] == "healthy" + assert calls == 2 + + +def test_runtime_work_provider_failure_is_sanitized() -> None: + routes._runtime_work_cache.clear() + + def provider(context: RuntimeWorkStatusContext) -> RuntimeWorkStatus: + del context + raise RuntimeError("redis://user:secret@example.test") + + registry = _Registry() + registry._manifest.runtime_work_status_providers = ( # type: ignore[misc] + RuntimeWorkStatusProviderRegistration( + module_id="example", + provider_id="example.failed", + provider=provider, + ), + ) + result = routes._runtime_work_statuses( + registry, + RuntimeWorkStatusContext( + profile="split-worker", + observed_at=datetime.now(UTC), + stale_after_seconds=60, + ), + force=True, + )[0] + + assert result["state"] == "unreachable" + assert "secret" not in result["detail"] + + +def test_disabled_workers_are_expected_only_in_development() -> None: + disabled = { + "provider_id": "example.queue", + "state": "disabled", + "detail": "Intentionally disabled.", + "enabled": False, + "configured": True, + "queue_depths": {}, + } + + development = routes._runtime_work_check([disabled], "local-dev") + production = routes._runtime_work_check([disabled], "single-process") + + assert development["state"] == "inactive" + assert development["readiness_critical"] is False + assert production["state"] == "warning" + assert production["readiness_critical"] is True + + def test_shared_runtime_cluster_missing_replicas_blocks_readiness() -> None: check = routes._runtime_cluster_check( { diff --git a/webui/package.json b/webui/package.json index 4db04a6..96ee268 100644 --- a/webui/package.json +++ b/webui/package.json @@ -12,6 +12,9 @@ "import": "./src/index.ts" } }, + "scripts": { + "test:runtime-status": "rm -rf .runtime-status-test-build && mkdir -p .runtime-status-test-build && printf '{\"type\":\"commonjs\"}\\n' > .runtime-status-test-build/package.json && ../../govoplan-core/webui/node_modules/.bin/tsc -p tsconfig.runtime-status-tests.json && node .runtime-status-test-build/tests/runtime-status.test.js" + }, "peerDependencies": { "@govoplan/core-webui": "^0.1.18", "lucide-react": "^1.23.0", diff --git a/webui/src/api/ops.ts b/webui/src/api/ops.ts index 57366cd..572e4f1 100644 --- a/webui/src/api/ops.ts +++ b/webui/src/api/ops.ts @@ -169,6 +169,25 @@ export type OpsRuntimeCluster = { }; }; +export type OpsRuntimeWorkStatus = { + provider_id: string; + label: string; + backend: string; + enabled: boolean | null; + configured: boolean | null; + state: "disabled" | "unconfigured" | "starting" | "healthy" | "idle" | "busy" | "degraded" | "stale" | "unreachable" | string; + detail: string; + observed_at: string; + active_workers?: number | null; + last_heartbeat_at?: string | null; + queue_depths: Record; + active_work?: number | null; + reserved_work?: number | null; + failures?: number | null; + stale_after_seconds?: number | null; + guidance: string; +}; + export type OpsStatus = { summary: { app_env: string; @@ -195,12 +214,15 @@ export type OpsStatus = { }; backup_state?: string; worker_metrics: { - workers?: number; - active_tasks?: number; + state?: string; + workers?: number | null; + active_tasks?: number | null; + reserved_tasks?: number | null; + failures?: number | null; expected_queues?: string[]; active_queues?: string[]; missing_queues?: string[]; - queue_depths?: Record; + queue_depths?: Record; }; operational_probe_count: number; runtime_node_count: number; @@ -244,6 +266,7 @@ export type OpsStatus = { deployment_profiles: OpsDeploymentProfile[]; sizing: OpsSizingAssumption[]; runtime_cluster: OpsRuntimeCluster; + runtime_work: OpsRuntimeWorkStatus[]; infrastructure: { configured: boolean; available: boolean; diff --git a/webui/src/features/ops/OpsHealthWidget.tsx b/webui/src/features/ops/OpsHealthWidget.tsx index 882067c..9414c59 100644 --- a/webui/src/features/ops/OpsHealthWidget.tsx +++ b/webui/src/features/ops/OpsHealthWidget.tsx @@ -12,6 +12,7 @@ import { } from "@govoplan/core-webui"; import { fetchOpsStatus, type OpsStatus } from "../../api/ops"; import { OPS_DOCUMENTATION } from "./interfacePatterns"; +import { knownMetric, knownQueueDepthTotal, runtimeWorkTone } from "./runtimeStatus"; export default function OpsHealthWidget({ settings, refreshKey }: { settings: ApiSettings; refreshKey: number }) { const [status, setStatus] = useState(null); @@ -35,8 +36,7 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap 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 ?? []; + const queuedTasks = knownQueueDepthTotal(queueDepths); const storageUsage = status?.summary.storage_metrics?.capacity_used_percent; return ( @@ -47,9 +47,9 @@ export default function OpsHealthWidget({ settings, refreshKey }: { settings: Ap - - - + + + typeof value === "number").length} measured queue(s)`} /> = 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 80fe5ca..6949bbc 100644 --- a/webui/src/features/ops/OpsPage.tsx +++ b/webui/src/features/ops/OpsPage.tsx @@ -1,6 +1,6 @@ import { DescriptionList } from "@govoplan/core-webui"; import { MetricGrid } from "@govoplan/core-webui"; -import { useEffect, useState } from "react"; +import { useCallback, useEffect, useRef, useState } from "react"; import { PauseCircle, PlayCircle, RefreshCw } from "lucide-react"; import { ContentGrid, ActionBlockerHint, @@ -33,6 +33,7 @@ import { type OpsInfrastructureCapability, type OpsRecoveryOperation, type OpsRuntimeNode, + type OpsRuntimeWorkStatus, type OpsSizingAssumption, type OpsStatus } from "../../api/ops"; @@ -41,6 +42,14 @@ import { OPS_I18N, OPS_RECOVERY_DOCUMENTATION, } from "./interfacePatterns"; +import { + heartbeatAgeLabel, + heartbeatAgeSeconds, + knownMetric, + knownQueueDepthTotal, + runtimeWorkTone, + shouldPollRuntimeStatus +} from "./runtimeStatus"; export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: AuthInfo;}) { const [status, setStatus] = useState(null); @@ -49,18 +58,27 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: const [error, setError] = useState(""); const [drainTarget, setDrainTarget] = useState(null); const [nodeActionId, setNodeActionId] = useState(""); + const loadInFlight = useRef | null>(null); - async function load() { - setLoading(true); - setError(""); - try { - setStatus(await fetchOpsStatus(settings)); - } catch (err) { - setError(adminErrorMessage(err)); - } finally { - setLoading(false); - } - } + const load = useCallback((background = false): Promise => { + if (loadInFlight.current) return loadInFlight.current; + const request = (async () => { + if (!background) setLoading(true); + setError(""); + try { + setStatus(await fetchOpsStatus(settings)); + } catch (err) { + setError(adminErrorMessage(err)); + } finally { + if (!background) setLoading(false); + } + })(); + loadInFlight.current = request; + void request.finally(() => { + if (loadInFlight.current === request) loadInFlight.current = null; + }); + return request; + }, [settings.accessToken, settings.apiBaseUrl, settings.apiKey]); async function runChecks() { setRunningProbes(true); @@ -104,7 +122,37 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: } } - useEffect(() => {void load();}, [settings.apiBaseUrl, settings.apiKey, settings.accessToken]); + useEffect(() => { + let intervalId: number | null = null; + const stopPolling = () => { + if (intervalId !== null) window.clearInterval(intervalId); + intervalId = null; + }; + const startPolling = () => { + stopPolling(); + if (document.visibilityState === "hidden") return; + intervalId = window.setInterval(() => { + if (shouldPollRuntimeStatus(document.visibilityState === "hidden", Boolean(loadInFlight.current))) { + void load(true); + } + }, 15_000); + }; + const handleVisibility = () => { + if (document.visibilityState === "hidden") { + stopPolling(); + return; + } + if (shouldPollRuntimeStatus(false, Boolean(loadInFlight.current))) void load(true); + startPolling(); + }; + void load(); + startPolling(); + document.addEventListener("visibilitychange", handleVisibility); + return () => { + stopPolling(); + document.removeEventListener("visibilitychange", handleVisibility); + }; + }, [load]); const checks = status?.checks ?? []; const warningCount = checks.filter((item) => item.state === "warning").length; @@ -119,7 +167,7 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: ? OPS_I18N.runPermissionRequired : undefined; const queueDepths = status?.summary.worker_metrics.queue_depths ?? {}; - const queuedTasks = Object.values(queueDepths).reduce((total, value) => total + value, 0); + const queuedTasks = knownQueueDepthTotal(queueDepths); const storageUsage = status?.summary.storage_metrics?.capacity_used_percent; return ( @@ -156,8 +204,8 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: - - + + typeof value === "number").length} measured queue(s)`} /> = 90 ? "danger" : storageUsage !== undefined && storageUsage >= 75 ? "warning" : "neutral"} detail={storageMetricDetail(status)} /> @@ -186,6 +234,10 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: /> + + + + @@ -226,6 +278,70 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: } +function RuntimeWorkTable({ items }: { items: OpsRuntimeWorkStatus[] }) { + const columns: DataGridColumn[] = [ + { + id: "provider", + header: "Backend", + width: "minmax(200px, 1fr)", + minWidth: 180, + resizable: true, + sortable: true, + filterable: true, + value: (item) => `${item.label} ${item.backend}`, + render: (item) =>
{item.label}{item.backend} · {item.provider_id}
+ }, + { + id: "state", + header: "State", + width: 150, + sortable: true, + filterable: true, + value: (item) => item.state, + render: (item) => + }, + { + id: "activity", + header: "Activity", + width: "minmax(180px, .7fr)", + minWidth: 170, + value: (item) => `${item.active_workers ?? ""} ${item.active_work ?? ""} ${item.reserved_work ?? ""}`, + render: (item) => `${knownMetric(item.active_workers)} workers · ${knownMetric(item.active_work)} active · ${knownMetric(item.reserved_work)} reserved` + }, + { + id: "queues", + header: "Queue depth", + width: "minmax(220px, 1fr)", + minWidth: 190, + resizable: true, + value: (item) => Object.entries(item.queue_depths).map(([queue, depth]) => `${queue}:${depth ?? "unavailable"}`).join(" "), + render: (item) => runtimeQueueSummary(item.queue_depths) + }, + { + id: "heartbeat", + header: "Last heartbeat", + width: "minmax(180px, .7fr)", + minWidth: 170, + sortable: true, + value: (item) => item.last_heartbeat_at ?? "", + render: (item) => { + const age = heartbeatAgeSeconds(Date.now(), item.last_heartbeat_at); + return
{heartbeatAgeLabel(age)}stale after {item.stale_after_seconds ?? "unavailable"}s
; + } + }, + { + id: "guidance", + header: "Guidance", + width: "minmax(260px, 1.3fr)", + minWidth: 220, + resizable: true, + value: (item) => `${item.detail} ${item.guidance}`, + render: (item) =>
{item.detail}{item.guidance}
+ } + ]; + return item.provider_id} emptyText="Worker and queue status unavailable." />; +} + function RuntimeNodeTable({ nodes, canManage, @@ -268,7 +384,7 @@ function RuntimeNodeTable({ resizable: true, sortable: true, value: (node) => node.last_heartbeat_at, - render: (node) => new Date(node.last_heartbeat_at).toLocaleString() + render: (node) =>
{heartbeatAgeLabel(heartbeatAgeSeconds(Date.now(), node.last_heartbeat_at))}{new Date(node.last_heartbeat_at).toLocaleString()}
}, { id: "queues", @@ -545,9 +661,10 @@ function capabilityTone(state: OpsInfrastructureCapability["state"]): string { } function stateTone(state: string): string { - if (state === "ok") return "success"; - if (state === "warning") return "warning"; - if (state === "error") return "error"; + if (["ok", "healthy", "idle"].includes(state)) return "success"; + if (state === "busy") return "info"; + if (["warning", "starting", "degraded", "unconfigured"].includes(state)) return "warning"; + if (["error", "stale", "unreachable"].includes(state)) return "error"; return "inactive"; } @@ -586,11 +703,17 @@ function databaseCapacityTone(status: OpsStatus | null): "good" | "warning" | "d } function workerMetricDetail(status: OpsStatus | null): string { - if (!status?.summary.celery_enabled) return "i18n:govoplan-ops.celery_worker_setting.323d7737"; + if (!status) return "Worker status unavailable"; 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`; + if (metrics.state === "disabled") return "Workers intentionally disabled"; + const queued = knownQueueDepthTotal(metrics.queue_depths ?? {}); + return `${knownMetric(metrics.active_tasks)} active · ${knownMetric(metrics.reserved_tasks)} reserved · ${queued ?? "unavailable"} queued`; +} + +function runtimeQueueSummary(depths: Record): string { + const entries = Object.entries(depths); + if (!entries.length) return "unavailable"; + return entries.map(([queue, depth]) => `${queue}: ${depth ?? "unavailable"}`).join(" · "); } function storageMetricDetail(status: OpsStatus | null): string { diff --git a/webui/src/features/ops/runtimeStatus.ts b/webui/src/features/ops/runtimeStatus.ts new file mode 100644 index 0000000..8883463 --- /dev/null +++ b/webui/src/features/ops/runtimeStatus.ts @@ -0,0 +1,34 @@ +export function knownMetric(value: number | null | undefined): number | "unavailable" { + return typeof value === "number" && Number.isFinite(value) ? value : "unavailable"; +} + +export function knownQueueDepthTotal(depths: Record): number | null { + const values = Object.values(depths).filter((value): value is number => typeof value === "number" && Number.isFinite(value)); + return values.length ? values.reduce((total, value) => total + value, 0) : null; +} + +export function runtimeWorkTone(state: string): "good" | "info" | "warning" | "danger" | "neutral" { + if (state === "healthy" || state === "idle") return "good"; + if (state === "busy") return "info"; + if (state === "starting" || state === "degraded" || state === "unconfigured") return "warning"; + if (state === "stale" || state === "unreachable") return "danger"; + return "neutral"; +} + +export function heartbeatAgeSeconds(now: number, value?: string | null): number | null { + if (!value) return null; + const timestamp = Date.parse(value); + if (!Number.isFinite(timestamp)) return null; + return Math.max(0, Math.floor((now - timestamp) / 1000)); +} + +export function heartbeatAgeLabel(ageSeconds: number | null): string { + if (ageSeconds === null) return "unavailable"; + if (ageSeconds < 60) return `${ageSeconds}s ago`; + if (ageSeconds < 3600) return `${Math.floor(ageSeconds / 60)}m ago`; + return `${Math.floor(ageSeconds / 3600)}h ago`; +} + +export function shouldPollRuntimeStatus(hidden: boolean, requestInFlight: boolean): boolean { + return !hidden && !requestInFlight; +} diff --git a/webui/tests/runtime-status.test.ts b/webui/tests/runtime-status.test.ts new file mode 100644 index 0000000..e2a764a --- /dev/null +++ b/webui/tests/runtime-status.test.ts @@ -0,0 +1,24 @@ +import assert from "node:assert/strict"; +import { + heartbeatAgeLabel, + heartbeatAgeSeconds, + knownMetric, + knownQueueDepthTotal, + runtimeWorkTone, + shouldPollRuntimeStatus +} from "../src/features/ops/runtimeStatus"; + +assert.equal(knownMetric(null), "unavailable"); +assert.equal(knownMetric(0), 0); +assert.equal(knownQueueDepthTotal({ mail: null }), null); +assert.equal(knownQueueDepthTotal({ mail: 0, calendar: 2 }), 2); +assert.equal(runtimeWorkTone("stale"), "danger"); +assert.equal(runtimeWorkTone("disabled"), "neutral"); +assert.equal(heartbeatAgeSeconds(Date.parse("2026-08-19T12:01:00Z"), "2026-08-19T12:00:00Z"), 60); +assert.equal(heartbeatAgeLabel(60), "1m ago"); +assert.equal(heartbeatAgeLabel(null), "unavailable"); +assert.equal(shouldPollRuntimeStatus(true, false), false); +assert.equal(shouldPollRuntimeStatus(false, true), false); +assert.equal(shouldPollRuntimeStatus(false, false), true); + +console.log("Ops runtime status and polling model tests passed."); diff --git a/webui/tsconfig.runtime-status-tests.json b/webui/tsconfig.runtime-status-tests.json new file mode 100644 index 0000000..a0f2270 --- /dev/null +++ b/webui/tsconfig.runtime-status-tests.json @@ -0,0 +1,17 @@ +{ + "compilerOptions": { + "module": "CommonJS", + "moduleResolution": "Node", + "target": "ES2022", + "strict": true, + "esModuleInterop": true, + "skipLibCheck": true, + "typeRoots": ["../../govoplan-core/webui/node_modules/@types"], + "types": ["node"], + "outDir": ".runtime-status-test-build" + }, + "include": [ + "src/features/ops/runtimeStatus.ts", + "tests/runtime-status.test.ts" + ] +}