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

feat(ops): show deployment capability receipt

Browse Source
This commit is contained in:
zemion
2026-08-07 01:59:57 +02:00
parent 71666db45c
commit d26a601dda
7 changed files with 389 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/govoplan_ops/backend/api/v1/routes.py
+40
View File
@@ -43,6 +43,7 @@ from govoplan_core.db.session import get_database
from govoplan_core.settings import settings as core_settings
from govoplan_ops.backend.manifest import OPS_READ_SCOPES, OPS_RUN_SCOPES
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]]] = {}
@@ -207,6 +208,7 @@ def _ops_status_payload(
)
storage_check = _storage_check()
backup_check = _backup_restore_check()
infrastructure = deployment_capability_status()
recovery_metrics = runtime_cluster.get("recovery", {}).get("metrics", {})
checks = [
_check(
@@ -229,6 +231,7 @@ def _ops_status_payload(
storage_check,
backup_check,
_deployment_security_check(current_profile),
_infrastructure_capability_check(infrastructure),
*module_checks,
]
readiness = _readiness(checks, maintenance_mode)
@@ -262,6 +265,12 @@ def _ops_status_payload(
"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),
"infrastructure_capability_count": len(
infrastructure.get("capabilities", [])
),
"pending_post_install_task_count": len(
infrastructure.get("post_install_tasks", [])
),
},
"readiness": readiness,
"checks": checks,
@@ -269,9 +278,40 @@ def _ops_status_payload(
"deployment_profiles": _deployment_profiles(current_profile),
"sizing": _sizing_assumptions(),
"runtime_cluster": runtime_cluster,
"infrastructure": infrastructure,
}
def _infrastructure_capability_check(
infrastructure: Mapping[str, object],
) -> dict[str, Any]:
if infrastructure.get("available") is True:
capabilities = infrastructure.get("capabilities")
count = len(capabilities) if isinstance(capabilities, list) else 0
return _check(
"infrastructure_capability_receipt",
"Infrastructure capabilities",
"ok",
f"A validated non-secret deployment receipt reports {count} capabilities.",
)
if infrastructure.get("configured") is True:
return _check(
"infrastructure_capability_receipt",
"Infrastructure capabilities",
"warning",
str(
infrastructure.get("error")
or "The configured deployment capability receipt is unavailable."
),
)
return _check(
"infrastructure_capability_receipt",
"Infrastructure capabilities",
"ok",
"No deployment capability receipt is mounted in this runtime profile.",
)
def _runtime_cluster_status(
*,
expected_composition_hash: str | None = None,
src/govoplan_ops/backend/infrastructure.py
+171
View File
@@ -0,0 +1,171 @@
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any, Mapping
MAX_CAPABILITY_DOCUMENT_BYTES = 256 * 1024
CAPABILITY_STATES = {
"configured",
"available_unconfigured",
"externally_supplied",
"unavailable",
}
def deployment_capability_status(
path: Path | None = None,
) -> dict[str, object]:
configured_path = path
if configured_path is None:
raw_path = os.getenv("GOVOPLAN_DEPLOYMENT_CAPABILITIES_PATH", "").strip()
if not raw_path:
return _unavailable(configured=False, error=None)
configured_path = Path(raw_path)
try:
document = _read_document(configured_path)
return {
"configured": True,
"available": True,
"schema_version": 1,
"installation_id": document["installation_id"],
"profile": document["profile"],
"capabilities": document["capabilities"],
"post_install_tasks": document["post_install_tasks"],
"error": None,
}
except (OSError, ValueError, json.JSONDecodeError) as exc:
return _unavailable(configured=True, error=str(exc))
def _read_document(path: Path) -> dict[str, object]:
if path.is_symlink() or not path.is_file():
raise ValueError("Deployment capability receipt is not a regular file.")
size = path.stat().st_size
if size > MAX_CAPABILITY_DOCUMENT_BYTES:
raise ValueError("Deployment capability receipt exceeds 256 KiB.")
raw = path.read_bytes()
if len(raw) != size:
raise ValueError("Deployment capability receipt changed while being read.")
payload = json.loads(raw.decode("utf-8"))
if not isinstance(payload, Mapping) or payload.get("schema_version") != 1:
raise ValueError("Deployment capability receipt has an unsupported schema.")
installation_id = _required_text(payload, "installation_id", maximum=100)
profile = _required_text(payload, "profile", maximum=100)
raw_capabilities = payload.get("capabilities")
if not isinstance(raw_capabilities, list) or len(raw_capabilities) > 100:
raise ValueError("Deployment capability receipt has invalid capabilities.")
capabilities = [_capability(item) for item in raw_capabilities]
ids = [str(item["id"]) for item in capabilities]
if len(ids) != len(set(ids)):
raise ValueError("Deployment capability receipt repeats a capability id.")
raw_tasks = payload.get("post_install_tasks", [])
if not isinstance(raw_tasks, list) or len(raw_tasks) > 100:
raise ValueError("Deployment capability receipt has invalid post-install tasks.")
tasks = [_task(item) for item in raw_tasks]
return {
"installation_id": installation_id,
"profile": profile,
"capabilities": capabilities,
"post_install_tasks": tasks,
}
def _capability(value: object) -> dict[str, object]:
if not isinstance(value, Mapping):
raise ValueError("Deployment capability entries must be objects.")
state = _required_text(value, "state", maximum=40)
if state not in CAPABILITY_STATES:
raise ValueError(f"Deployment capability state is unsupported: {state!r}.")
endpoint = value.get("endpoint", {})
if not isinstance(endpoint, Mapping) or len(endpoint) > 10:
raise ValueError("Deployment capability endpoint metadata is invalid.")
normalized_endpoint: dict[str, object] = {}
for key, raw in endpoint.items():
if not isinstance(key, str) or len(key) > 50:
raise ValueError("Deployment capability endpoint key is invalid.")
if isinstance(raw, bool) or raw is None:
normalized_endpoint[key] = raw
elif isinstance(raw, int):
normalized_endpoint[key] = raw
elif isinstance(raw, str) and len(raw) <= 500:
normalized_endpoint[key] = raw
else:
raise ValueError("Deployment capability endpoint value is invalid.")
secret_refs = _string_list(value.get("secret_refs"), maximum_items=30)
if any(not item.startswith("env:") for item in secret_refs):
raise ValueError("Deployment capability secrets must use environment references.")
return {
"id": _required_text(value, "id", maximum=120),
"label": _required_text(value, "label", maximum=200),
"state": state,
"source": _required_text(value, "source", maximum=120),
"detail": _required_text(value, "detail", maximum=1000),
"endpoint": normalized_endpoint,
"secret_refs": secret_refs,
"dependent_modules": _string_list(
value.get("dependent_modules"),
maximum_items=100,
),
}
def _task(value: object) -> dict[str, object]:
if not isinstance(value, Mapping):
raise ValueError("Deployment post-install task entries must be objects.")
return {
"id": _required_text(value, "id", maximum=120),
"resume_key": _required_text(value, "resume_key", maximum=240),
"capability_id": _required_text(value, "capability_id", maximum=120),
"state": _required_text(value, "state", maximum=40),
"owner_module": _required_text(value, "owner_module", maximum=120),
"summary": _required_text(value, "summary", maximum=1000),
"required_inputs": _string_list(
value.get("required_inputs"),
maximum_items=30,
),
"secret_boundary": _required_text(
value,
"secret_boundary",
maximum=120,
),
}
def _required_text(
value: Mapping[str, Any],
key: str,
*,
maximum: int,
) -> str:
raw = value.get(key)
text = str(raw).strip() if raw is not None else ""
if not text or len(text) > maximum:
raise ValueError(f"Deployment capability field {key!r} is invalid.")
return text
def _string_list(value: object, *, maximum_items: int) -> list[str]:
if not isinstance(value, list) or len(value) > maximum_items:
raise ValueError("Deployment capability list field is invalid.")
result: list[str] = []
for item in value:
if not isinstance(item, str) or not item.strip() or len(item) > 500:
raise ValueError("Deployment capability list item is invalid.")
result.append(item.strip())
return result
def _unavailable(*, configured: bool, error: str | None) -> dict[str, object]:
return {
"configured": configured,
"available": False,
"schema_version": None,
"installation_id": None,
"profile": None,
"capabilities": [],
"post_install_tasks": [],
"error": error,
}
src/govoplan_ops/backend/manifest.py
+1 -1
View File
@@ -129,7 +129,7 @@ manifest = ModuleManifest(
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. 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. 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"),