diff --git a/README.md b/README.md index d4b2eec..b7af98d 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,8 @@ This repository owns: - operator-facing status APIs - runtime-node registration, heartbeat, composition, stale-node, and expected replica projection +- queue-specific worker-pool coverage, release/composition skew, and the + deployment-rendered PostgreSQL connection budget - audited API and worker drain/cancel controls - recovery-operation status and evidence-chain summaries - governance inventory for module-declared permissions, roles, capabilities, diff --git a/docs/SCALABILITY_PROFILES.md b/docs/SCALABILITY_PROFILES.md index 70b2751..fbac8e8 100644 --- a/docs/SCALABILITY_PROFILES.md +++ b/docs/SCALABILITY_PROFILES.md @@ -39,6 +39,9 @@ The Ops API reports: - runtime node identity, role, software/module composition, queues, heartbeat, stale state, and drain state - configured versus active API and worker replica counts +- active worker-pool names, exact queue coverage, and missing queue owners +- 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 These values are intentionally diagnostic. They do not replace deployment @@ -54,6 +57,13 @@ manual-intervention record means an operator must repair the current release or restore a separately verified coordinated backup. Ops does not convert that state into a safe rollback. +For the `shared` profile, missing expected replicas, release/composition skew, +unserved queues, or an invalid PostgreSQL connection budget are readiness +errors. Stale historical records remain visible, but cannot downgrade one of +those errors to a warning. A database budget at or above 80 percent is reported +as a warning so operators retain room for measured bursts and administrative +connections. + `deployment_security` is inactive for local/test profiles. In staging or pilot profiles it warns when secure cookies or CORS origins still look local. In production it becomes readiness-critical because TLS certificates, proxy diff --git a/src/govoplan_ops/backend/api/v1/routes.py b/src/govoplan_ops/backend/api/v1/routes.py index 11f49aa..281f32c 100644 --- a/src/govoplan_ops/backend/api/v1/routes.py +++ b/src/govoplan_ops/backend/api/v1/routes.py @@ -35,6 +35,7 @@ from govoplan_core.core.runtime_coordination import ( cancel_runtime_node_drain, list_runtime_nodes, request_runtime_node_drain, + runtime_composition_hash, ) from govoplan_core.db.session import get_database from govoplan_core.settings import settings as core_settings @@ -70,7 +71,9 @@ def ops_readiness( payload = _ops_status_payload(request) readiness = payload["readiness"] if not readiness["ready"]: - raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=readiness) + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail=readiness + ) return readiness @@ -85,10 +88,16 @@ def run_ops_checks( @router.get("/runtime/nodes") def runtime_nodes( + request: Request, principal: ApiPrincipal = Depends(require_any_scope(*OPS_READ_SCOPES)), ) -> dict[str, Any]: del principal - return _runtime_cluster_status() + registry = _registry(request) + return _runtime_cluster_status( + expected_composition_hash=runtime_composition_hash( + tuple(manifest.id for manifest in registry.manifests()) + ) + ) @router.post("/runtime/nodes/{node_id}/drain") @@ -175,21 +184,42 @@ def _ops_status_payload( ) -> 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} + 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() - runtime_cluster = _runtime_cluster_status() + database_capacity = _database_capacity_check() + expected_composition_hash = runtime_composition_hash( + tuple(manifest.id for manifest in registry.manifests()) + ) + runtime_cluster = _runtime_cluster_status( + expected_composition_hash=expected_composition_hash + ) 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"])), + _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, + database_capacity, _runtime_cluster_check(runtime_cluster), _storage_check(), _backup_restore_check(), @@ -211,11 +241,17 @@ def _ops_status_payload( "redis_url": _redact_url(core_settings.redis_url), "maintenance_mode": maintenance_mode, "database_url": _redact_url(core_settings.database_url), + "database_connection_peak": core_settings.database_connection_peak, + "database_connection_available": ( + core_settings.database_connection_available + ), "file_storage_backend": core_settings.file_storage_backend, "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), + "recovery_required_count": runtime_cluster.get("recovery", {}).get( + "requires_attention", 0 + ), }, "readiness": readiness, "checks": checks, @@ -226,7 +262,10 @@ def _ops_status_payload( } -def _runtime_cluster_status() -> dict[str, Any]: +def _runtime_cluster_status( + *, + expected_composition_hash: str | None = None, +) -> dict[str, Any]: try: with get_database().SessionLocal() as session: nodes = list_runtime_nodes( @@ -237,8 +276,7 @@ def _runtime_cluster_status() -> dict[str, Any]: operations = ( session.query(RecoveryOperation) .filter( - RecoveryOperation.installation_id - == core_settings.installation_id + RecoveryOperation.installation_id == core_settings.installation_id ) .order_by(RecoveryOperation.updated_at.desc()) .limit(50) @@ -253,12 +291,48 @@ def _runtime_cluster_status() -> dict[str, Any]: "recovery": {"operations": [], "requires_attention": 0}, } active_nodes = [ - node - for node in nodes - if node["state"] != "stopped" and not node["stale"] + node for node in nodes if node["state"] == "active" and not node["stale"] ] active_api = sum(node["role"] == "api" for node in active_nodes) active_workers = sum(node["role"] == "worker" for node in active_nodes) + runtime_nodes = [node for node in active_nodes if node["role"] in {"api", "worker"}] + composition_hashes = sorted( + { + str(node["composition_hash"]) + for node in runtime_nodes + if node.get("composition_hash") + } + ) + software_versions = sorted( + { + str(node["software_version"]) + for node in runtime_nodes + if node.get("software_version") + } + ) + unexpected_composition_nodes = [ + str(node["node_id"]) + for node in runtime_nodes + if expected_composition_hash + and node.get("composition_hash") != expected_composition_hash + ] + active_worker_nodes = [node for node in active_nodes if node["role"] == "worker"] + active_queues = sorted( + { + str(queue) + for node in active_worker_nodes + for queue in (node.get("queues") or []) + if queue + } + ) + expected_queues = _celery_queues() if core_settings.celery_enabled else [] + missing_queues = sorted(set(expected_queues) - set(active_queues)) + worker_pools = sorted( + { + str(node.get("metadata", {}).get("worker_pool") or "default") + for node in active_worker_nodes + } + ) operation_payloads = [ { "id": operation.id, @@ -295,6 +369,22 @@ def _runtime_cluster_status() -> dict[str, Any]: "worker": core_settings.runtime_expected_worker_replicas, }, "active": {"api": active_api, "worker": active_workers}, + "composition": { + "expected_hash": expected_composition_hash, + "active_hashes": composition_hashes, + "unexpected_nodes": unexpected_composition_nodes, + "skewed": len(composition_hashes) > 1 or bool(unexpected_composition_nodes), + }, + "software_versions": { + "active": software_versions, + "skewed": len(software_versions) > 1, + }, + "queues": { + "expected": expected_queues, + "active": active_queues, + "missing": missing_queues, + "worker_pools": worker_pools, + }, "nodes": nodes, "recovery": { "operations": operation_payloads, @@ -317,15 +407,36 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]: expected = cluster.get("expected") or {} active = cluster.get("active") or {} missing_api = max(0, int(expected.get("api") or 0) - int(active.get("api") or 0)) - missing_workers = max(0, int(expected.get("worker") or 0) - int(active.get("worker") or 0)) + missing_workers = max( + 0, int(expected.get("worker") or 0) - int(active.get("worker") or 0) + ) + composition = cluster.get("composition") or {} + versions = cluster.get("software_versions") or {} + queues = cluster.get("queues") or {} + composition_skewed = bool(composition.get("skewed")) + version_skewed = bool(versions.get("skewed")) + missing_queues = list(queues.get("missing") or []) state = "ok" detail = f"{len(nodes)} runtime node record(s); {len(stale)} stale." readiness_critical = False if missing_api or missing_workers: state = "error" if cluster.get("state_profile") == "shared" else "warning" readiness_critical = state == "error" - detail += f" Missing expected replicas: api={missing_api}, worker={missing_workers}." - elif stale: + detail += ( + f" Missing expected replicas: api={missing_api}, worker={missing_workers}." + ) + if composition_skewed or version_skewed or missing_queues: + state = "error" if cluster.get("state_profile") == "shared" else "warning" + readiness_critical = readiness_critical or state == "error" + if composition_skewed: + detail += " Active runtime composition differs from the loaded graph." + if version_skewed: + detail += " Active runtime software versions differ." + if missing_queues: + detail += ( + " Queues without an active worker: " + ", ".join(missing_queues) + "." + ) + if stale and state == "ok": state = "warning" return _check( "runtime_cluster", @@ -338,6 +449,9 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]: "stale": len(stale), "active_api": int(active.get("api") or 0), "active_workers": int(active.get("worker") or 0), + "composition_skewed": composition_skewed, + "version_skewed": version_skewed, + "missing_queues": len(missing_queues), "recovery_required": int( cluster.get("recovery", {}).get("requires_attention") or 0 ), @@ -345,10 +459,70 @@ def _runtime_cluster_check(cluster: dict[str, Any]) -> dict[str, Any]: ) +def _database_capacity_check() -> dict[str, Any]: + limit = core_settings.database_connection_limit + reserve = core_settings.database_connection_reserve + peak = core_settings.database_connection_peak + available = core_settings.database_connection_available + shared = core_settings.state_profile == "shared" + if limit is None or peak is None or available is None: + return _check( + "database_capacity", + "Database connection capacity", + "error" if shared else "inactive", + ( + "Shared deployments must declare the PostgreSQL connection " + "limit and rendered peak demand." + if shared + else "No deployment-level PostgreSQL connection budget is configured." + ), + readiness_critical=shared, + ) + expected_available = limit - reserve + valid = reserve < limit and available == expected_available and peak <= available + utilization = peak / available if available else 1.0 + if not valid: + return _check( + "database_capacity", + "Database connection capacity", + "error", + ( + f"Rendered peak {peak} does not fit the declared available " + f"capacity {available} (limit {limit}, reserve {reserve})." + ), + readiness_critical=shared, + metrics={ + "limit": limit, + "reserve": reserve, + "available": available, + "peak": peak, + }, + ) + return _check( + "database_capacity", + "Database connection capacity", + "warning" if utilization >= 0.8 else "ok", + ( + f"Rendered peak {peak} uses {utilization:.0%} of {available} " + f"connections available after the {reserve}-connection reserve." + ), + metrics={ + "limit": limit, + "reserve": reserve, + "available": available, + "peak": peak, + "utilization": round(utilization, 4), + }, + ) + + def _registry(request: Request) -> PlatformRegistry: registry = getattr(request.app.state, "govoplan_registry", None) if not isinstance(registry, PlatformRegistry): - raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="GovOPlaN module registry is not configured") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail="GovOPlaN module registry is not configured", + ) return registry @@ -376,26 +550,33 @@ def _governance_inventory( access_control_count = ( len(manifest.resource_acl_providers) + len(manifest.ownership_providers) - + sum(len(providers) for providers in manifest.delete_veto_providers.values()) + + sum( + len(providers) for providers in manifest.delete_veto_providers.values() + ) + len(manifest.uninstall_guard_providers) ) - modules.append({ - "module_id": manifest.id, - "name": manifest.name, - "version": manifest.version, - "permission_count": len(manifest.permissions), - "role_template_count": len(manifest.role_templates), - "capability_count": len(capability_names), - "policy_count": sum(name.startswith("policy.") for name in capability_names), - "documentation_count": len(manifest.documentation), - "documentation_provider_count": len(manifest.documentation_providers), - "access_control_count": access_control_count, - "search_provider_count": len(manifest.search_providers) + len(manifest.search_sources), - "migration_managed": manifest.migration_spec is not None, - "architecture": architecture, - "external_provider_count": len(external_providers), - "external_providers": external_providers, - }) + modules.append( + { + "module_id": manifest.id, + "name": manifest.name, + "version": manifest.version, + "permission_count": len(manifest.permissions), + "role_template_count": len(manifest.role_templates), + "capability_count": len(capability_names), + "policy_count": sum( + name.startswith("policy.") for name in capability_names + ), + "documentation_count": len(manifest.documentation), + "documentation_provider_count": len(manifest.documentation_providers), + "access_control_count": access_control_count, + "search_provider_count": len(manifest.search_providers) + + len(manifest.search_sources), + "migration_managed": manifest.migration_spec is not None, + "architecture": architecture, + "external_provider_count": len(external_providers), + "external_providers": external_providers, + } + ) return { "summary": { "module_count": len(modules), @@ -404,12 +585,20 @@ def _governance_inventory( "capability_count": sum(item["capability_count"] for item in modules), "policy_count": sum(item["policy_count"] for item in modules), "documented_module_count": sum( - bool(item["documentation_count"] or item["documentation_provider_count"]) + bool( + item["documentation_count"] or item["documentation_provider_count"] + ) for item in modules ), - "access_control_count": sum(item["access_control_count"] for item in modules), - "search_provider_count": sum(item["search_provider_count"] for item in modules), - "migration_module_count": sum(bool(item["migration_managed"]) for item in modules), + "access_control_count": sum( + item["access_control_count"] for item in modules + ), + "search_provider_count": sum( + item["search_provider_count"] for item in modules + ), + "migration_module_count": sum( + bool(item["migration_managed"]) for item in modules + ), "architecture_declared_module_count": sum( bool(item["architecture"]) for item in modules ), @@ -471,29 +660,63 @@ def _database_status() -> dict[str, Any]: with get_database().session() as session: session.execute(text("select 1")) maintenance = saved_maintenance_mode(session).as_dict() - return {"ok": True, "detail": "Database session succeeded.", "maintenance_mode": maintenance} + return { + "ok": True, + "detail": "Database session succeeded.", + "maintenance_mode": maintenance, + } except (RuntimeError, SQLAlchemyError) as exc: - return {"ok": False, "detail": f"Database check failed: {exc}", "maintenance_mode": {"enabled": False, "message": None}} + return { + "ok": False, + "detail": f"Database check failed: {exc}", + "maintenance_mode": {"enabled": False, "message": None}, + } def _maintenance_check(maintenance_mode: dict[str, Any]) -> dict[str, Any]: if maintenance_mode.get("enabled"): message = str(maintenance_mode.get("message") or "Maintenance mode is enabled.") - return _check("maintenance_mode", "Maintenance mode", "warning", message, readiness_critical=True) - return _check("maintenance_mode", "Maintenance mode", "ok", "Maintenance mode is off.") + return _check( + "maintenance_mode", + "Maintenance mode", + "warning", + message, + readiness_critical=True, + ) + return _check( + "maintenance_mode", "Maintenance mode", "ok", "Maintenance mode is off." + ) 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.") + return _check( + "redis_broker", + "Redis broker", + "inactive", + "Redis is not required while Celery is disabled.", + ) try: from redis import Redis - client = Redis.from_url(core_settings.redis_url, socket_connect_timeout=0.75, socket_timeout=0.75) + 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)}.") + 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) + return _check( + "redis_broker", + "Redis broker", + "error", + f"Redis broker check failed: {exc}", + readiness_critical=True, + ) def _worker_check() -> dict[str, Any]: @@ -531,25 +754,29 @@ def _worker_check() -> dict[str, Any]: 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") - }) + 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) + 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) + "." + detail += ( + " No worker consumes configured queues: " + + ", ".join(missing_queues) + + "." + ) result = _check( "worker_split", "Background workers", @@ -596,24 +823,48 @@ def _queue_depths(queues: list[str]) -> dict[str, int]: 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) - } + return {queue: int(value) for queue, value in zip(queues, values, strict=True)} def _storage_check() -> dict[str, Any]: backend = str(core_settings.file_storage_backend or "local").lower() if backend == "s3": - configured = bool(core_settings.file_storage_s3_endpoint_url and core_settings.file_storage_s3_bucket) - return _check("file_storage", "File storage", "ok" if configured else "warning", "S3 file storage is configured." if configured else "S3 file storage needs endpoint and bucket settings.", readiness_critical=not configured) + configured = bool( + core_settings.file_storage_s3_endpoint_url + and core_settings.file_storage_s3_bucket + ) + return _check( + "file_storage", + "File storage", + "ok" if configured else "warning", + "S3 file storage is configured." + if configured + else "S3 file storage needs endpoint and bucket settings.", + readiness_critical=not configured, + ) root = Path(str(core_settings.file_storage_local_root or "runtime/files")) if root.exists() and root.is_dir(): writable = os.access(root, os.W_OK) state = "ok" if writable else "warning" - detail = f"Local file storage root: {root}" if writable else f"Local file storage root is not writable: {root}" - return _check("file_storage", "File storage", state, detail, readiness_critical=not writable) - return _check("file_storage", "File storage", "warning", f"Local file storage root does not exist yet: {root}", readiness_critical=False) + detail = ( + f"Local file storage root: {root}" + if writable + else f"Local file storage root is not writable: {root}" + ) + return _check( + "file_storage", + "File storage", + state, + detail, + readiness_critical=not writable, + ) + return _check( + "file_storage", + "File storage", + "warning", + f"Local file storage root does not exist yet: {root}", + readiness_critical=False, + ) def _module_operational_checks( @@ -630,13 +881,15 @@ def _module_operational_checks( 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, - )) + 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( @@ -694,7 +947,9 @@ def _backup_restore_check() -> dict[str, Any]: 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 + backup = ( + snapshot.get("database_backup") if isinstance(snapshot, Mapping) else None + ) if isinstance(backup, Mapping): latest_evidence = (record, backup) break @@ -711,12 +966,9 @@ def _backup_restore_check() -> dict[str, Any]: 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" - ) + 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" @@ -725,9 +977,7 @@ def _backup_restore_check() -> dict[str, Any]: 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." - ) + detail = f"Installer run {record.get('run_id')} records a {backup.get('type', 'database')} backup." if missing: detail += " Missing evidence: " + ", ".join(missing) + "." else: @@ -749,7 +999,11 @@ def _backup_restore_check() -> dict[str, Any]: 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" + path = ( + Path(configured).expanduser() + if configured + else runtime_dir / "restore-drill-evidence.json" + ) try: import json @@ -762,7 +1016,12 @@ def _restore_drill_evidence(runtime_dir: Path) -> dict[str, object] | 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"}: - return _check("deployment_security", "HTTP and certificates", "inactive", "Local/test profile; HTTPS and secure-cookie checks are deployment-owned.") + return _check( + "deployment_security", + "HTTP and certificates", + "inactive", + "Local/test profile; HTTPS and secure-cookie checks are deployment-owned.", + ) blockers: list[str] = [] if not core_settings.auth_cookie_secure: @@ -770,13 +1029,27 @@ def _deployment_security_check(current_profile: str) -> dict[str, Any]: cors_origins = _cors_origins() if not cors_origins: blockers.append("CORS_ORIGINS is empty") - elif any(origin == "*" or origin.startswith("http://localhost") or origin.startswith("http://127.0.0.1") for origin in cors_origins): + elif any( + origin == "*" + or origin.startswith("http://localhost") + or origin.startswith("http://127.0.0.1") + for origin in cors_origins + ): blockers.append("CORS_ORIGINS contains local or wildcard origins") if not blockers: - return _check("deployment_security", "HTTP and certificates", "ok", f"{current_profile} uses secure cookies and explicit non-local CORS origins.") + return _check( + "deployment_security", + "HTTP and certificates", + "ok", + f"{current_profile} uses secure cookies and explicit non-local CORS origins.", + ) - profile_label = "production" if app_env in {"prod", "production"} else app_env or current_profile + profile_label = ( + "production" + if app_env in {"prod", "production"} + else app_env or current_profile + ) detail = f"{profile_label} profile needs deployment HTTP/certificate hardening: {', '.join(blockers)}." return _check( "deployment_security", @@ -806,14 +1079,32 @@ def _check( } -def _readiness(checks: list[dict[str, Any]], maintenance_mode: dict[str, Any]) -> dict[str, Any]: +def _readiness( + checks: list[dict[str, Any]], maintenance_mode: dict[str, Any] +) -> dict[str, Any]: blockers = [ - {"id": check["id"], "label": check["label"], "state": check["state"], "detail": check["detail"]} + { + "id": check["id"], + "label": check["label"], + "state": check["state"], + "detail": check["detail"], + } for check in checks if check.get("readiness_critical") and check.get("state") != "ok" ] - if maintenance_mode.get("enabled") and not any(item["id"] == "maintenance_mode" for item in blockers): - blockers.append({"id": "maintenance_mode", "label": "Maintenance mode", "state": "warning", "detail": str(maintenance_mode.get("message") or "Maintenance mode is enabled.")}) + if maintenance_mode.get("enabled") and not any( + item["id"] == "maintenance_mode" for item in blockers + ): + blockers.append( + { + "id": "maintenance_mode", + "label": "Maintenance mode", + "state": "warning", + "detail": str( + maintenance_mode.get("message") or "Maintenance mode is enabled." + ), + } + ) return { "ready": not blockers, "blockers": blockers, @@ -833,11 +1124,19 @@ def _current_profile() -> str: def _celery_queues() -> list[str]: - return [item.strip() for item in str(core_settings.celery_queues or "").split(",") if item.strip()] + return [ + item.strip() + for item in str(core_settings.celery_queues or "").split(",") + if item.strip() + ] def _cors_origins() -> list[str]: - return [item.strip() for item in str(core_settings.cors_origins or "").split(",") if item.strip()] + return [ + item.strip() + for item in str(core_settings.cors_origins or "").split(",") + if item.strip() + ] def _deployment_profiles(current_profile: str) -> list[dict[str, Any]]: @@ -846,28 +1145,53 @@ def _deployment_profiles(current_profile: str) -> list[dict[str, Any]]: "id": "local-dev", "name": "Local development", "current": current_profile == "local-dev", - "components": ["FastAPI", "SQLite or local database", "Vite dev server", "local file storage"], + "components": [ + "FastAPI", + "SQLite or local database", + "Vite dev server", + "local file storage", + ], "fit": "Developer workstation, demos, and module integration work.", }, { "id": "production-like-dev", "name": "Production-like development", "current": current_profile == "production-like-dev", - "components": ["FastAPI", "Vite dev server", "PostgreSQL", "Redis broker", "Celery worker", "durable local file storage"], + "components": [ + "FastAPI", + "Vite dev server", + "PostgreSQL", + "Redis broker", + "Celery worker", + "durable local file storage", + ], "fit": "Local validation of the production dependency shape without publishing packages.", }, { "id": "single-process", "name": "Small institution", "current": current_profile == "single-process", - "components": ["FastAPI", "PostgreSQL", "reverse proxy", "scheduled backups", "local or S3-compatible file storage"], + "components": [ + "FastAPI", + "PostgreSQL", + "reverse proxy", + "scheduled backups", + "local or S3-compatible file storage", + ], "fit": "Low-volume internal administration without heavy background jobs.", }, { "id": "split-worker", "name": "Institution platform", "current": current_profile == "split-worker", - "components": ["FastAPI", "worker process", "Redis broker", "PostgreSQL", "object storage", "health probes"], + "components": [ + "FastAPI", + "worker process", + "Redis broker", + "PostgreSQL", + "object storage", + "health probes", + ], "fit": "Campaigns, imports, exports, scheduled work, and larger tenant counts.", }, ] diff --git a/tests/test_operational_checks.py b/tests/test_operational_checks.py index 92bee4c..96765a3 100644 --- a/tests/test_operational_checks.py +++ b/tests/test_operational_checks.py @@ -52,11 +52,13 @@ def test_module_operational_check_failure_is_isolated() -> None: raise RuntimeError("secret detail") result = routes._module_operational_checks( - _Registry(OperationalCheckProviderRegistration( - module_id="example", - check_id="example.failed", - provider=provider, - )), + _Registry( + OperationalCheckProviderRegistration( + module_id="example", + check_id="example.failed", + provider=provider, + ) + ), force=True, )[0] @@ -82,3 +84,43 @@ def test_shared_runtime_cluster_missing_replicas_blocks_readiness() -> None: assert check["state"] == "error" assert check["readiness_critical"] is True assert check["metrics"]["recovery_required"] == 1 + + +def test_shared_runtime_cluster_skew_or_unserved_queue_blocks_readiness() -> None: + check = routes._runtime_cluster_check( + { + "available": True, + "state_profile": "shared", + "nodes": [], + "expected": {"api": 2, "worker": 2}, + "active": {"api": 2, "worker": 2}, + "composition": {"skewed": True}, + "software_versions": {"skewed": False}, + "queues": {"missing": ["calendar"]}, + "recovery": {"requires_attention": 0}, + } + ) + + assert check["state"] == "error" + assert check["readiness_critical"] is True + assert check["metrics"]["composition_skewed"] is True + assert check["metrics"]["missing_queues"] == 1 + + +def test_database_capacity_check_enforces_shared_rendered_budget( + monkeypatch, +) -> None: + monkeypatch.setattr(routes.core_settings, "state_profile", "shared") + monkeypatch.setattr(routes.core_settings, "database_connection_limit", 100) + monkeypatch.setattr(routes.core_settings, "database_connection_reserve", 10) + monkeypatch.setattr(routes.core_settings, "database_connection_available", 90) + monkeypatch.setattr(routes.core_settings, "database_connection_peak", 70) + + healthy = routes._database_capacity_check() + assert healthy["state"] == "ok" + assert healthy["metrics"]["peak"] == 70 + + monkeypatch.setattr(routes.core_settings, "database_connection_peak", 91) + overrun = routes._database_capacity_check() + assert overrun["state"] == "error" + assert overrun["readiness_critical"] is True diff --git a/webui/src/api/ops.ts b/webui/src/api/ops.ts index a3b72fe..7afaf45 100644 --- a/webui/src/api/ops.ts +++ b/webui/src/api/ops.ts @@ -117,6 +117,22 @@ export type OpsRuntimeCluster = { state_profile: string; expected?: { api: number; worker: number }; active?: { api: number; worker: number }; + composition?: { + expected_hash?: string | null; + active_hashes: string[]; + unexpected_nodes: string[]; + skewed: boolean; + }; + software_versions?: { + active: string[]; + skewed: boolean; + }; + queues?: { + expected: string[]; + active: string[]; + missing: string[]; + worker_pools: string[]; + }; nodes: OpsRuntimeNode[]; recovery: { operations: OpsRecoveryOperation[]; @@ -137,6 +153,8 @@ export type OpsStatus = { message?: string | null; }; database_url: string; + database_connection_peak?: number | null; + database_connection_available?: number | null; file_storage_backend: string; worker_metrics: { workers?: number; diff --git a/webui/src/features/ops/OpsPage.tsx b/webui/src/features/ops/OpsPage.tsx index 004719f..fe7b05d 100644 --- a/webui/src/features/ops/OpsPage.tsx +++ b/webui/src/features/ops/OpsPage.tsx @@ -125,6 +125,7 @@ export default function OpsPage({ settings, auth }: {settings: ApiSettings;auth: + @@ -480,6 +481,24 @@ function runtimeNodeMetricDetail(status: OpsStatus | null): string { return `${active.api}/${expected.api} API ยท ${active.worker}/${expected.worker} workers`; } +function databaseCapacityValue(status: OpsStatus | null): string { + const peak = status?.summary.database_connection_peak; + const available = status?.summary.database_connection_available; + return typeof peak === "number" && typeof available === "number" + ? `${peak} / ${available}` + : "not declared"; +} + +function databaseCapacityTone(status: OpsStatus | null): "good" | "warning" | "danger" { + const peak = status?.summary.database_connection_peak; + const available = status?.summary.database_connection_available; + if (typeof peak !== "number" || typeof available !== "number") { + return status?.runtime_cluster.state_profile === "shared" ? "danger" : "warning"; + } + if (peak > available) return "danger"; + return peak / available >= 0.8 ? "warning" : "good"; +} + 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;