diff --git a/docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md b/docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md index 35ab3d3..3aeb05c 100644 --- a/docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md +++ b/docs/SCALING_AND_MULTI_HOST_DEPLOYMENT.md @@ -150,8 +150,29 @@ versioning, and lifecycle controls. ## Capacity -- Scale API replicas only within the PostgreSQL connection budget. +- Set `GOVOPLAN_DB_CONNECTION_LIMIT` to the PostgreSQL role's effective + connection limit. The Kubernetes export reserves + `GOVOPLAN_DB_CONNECTION_RESERVE` connections and rejects a topology whose + calculated rolling-update peak would exceed the remainder. The calculation + includes API pools, every Celery parent and prefork child, the scheduler, + migration, and one surge replica per deployment. Role-specific pool and + overflow values are emitted into each workload rather than inherited from one + unconstrained global default. - Scale workers by queue, with upper bounds based on external provider limits. + `GOVOPLAN_WORKER_POOLS` may contain a JSON list of exact queue owners, for + example: + + ```json + [ + {"name":"delivery","queues":["send_email","append_sent"],"replicas":2,"concurrency":2}, + {"name":"platform","queues":["events","workflow","default"],"replicas":2,"concurrency":2} + ] + ``` + + Pool replica totals must equal `replicas.worker`, and the pools must cover + `CELERY_QUEUES` exactly without duplicate ownership. Each pool receives its + own Deployment, disruption budget, topology-spread selector, runtime identity, + and declared concurrency. - Keep one fenced scheduler rather than load-balancing schedulers. - Increase WebUI replicas for asset/proxy capacity. - Measure request latency, database query time and locks, active connections, @@ -181,3 +202,28 @@ availability, drill replica loss, rolling replacement, session continuity, job redelivery, scheduler failover, migration exclusion, object-store outage, and a coordinated database/object/key restore. Recovery rules and evidence are defined in [Recovery And Rollback Guarantees](RECOVERY_AND_ROLLBACK_GUARANTEES.md). + +## Live Multi-Host Evidence + +After deploying a pinned release on at least two Kubernetes nodes, create an API +key with Ops read scope and run: + +```bash +export GOVOPLAN_OPS_API_KEY='...' +python tools/deployment/govoplan-deploy.py verify-kubernetes \ + --directory /srv/govoplan/installation \ + --namespace govoplan +``` + +The command fails unless API and WebUI pods are ready on at least two nodes, +all rendered deployments are available, Ops reports a consistent release and +module composition, every declared worker queue is served, and the calculated +database peak remains below its budget. It writes a private, sanitized JSON +record under the installation evidence directory and never retains the API key. + +Use `--exercise-api-pod-loss` in an approved drill window to delete one API pod, +observe the public readiness path continuously, and record its replacement. +This proves the bounded stateless-node-loss slice only. Session continuity, +accepted-job redelivery, state-service failover, and coordinated restore remain +separate target exercises whose signed evidence is governed by +`docs/TARGET_MATURITY_EVIDENCE_RUNBOOK.md` and GovOPlaN #37. diff --git a/tests/test_deployment_installer.py b/tests/test_deployment_installer.py index 45a915e..9397a16 100644 --- a/tests/test_deployment_installer.py +++ b/tests/test_deployment_installer.py @@ -31,6 +31,9 @@ from govoplan_deploy.bundle import ( # noqa: E402 ) from govoplan_deploy.cli import _receipt_uses_direct_web_port, main # noqa: E402 import govoplan_deploy.cli as deployment_cli # noqa: E402 +from govoplan_deploy.cluster_evidence import ( # noqa: E402 + collect_kubernetes_evidence, +) from govoplan_deploy.kubernetes import render_kubernetes # noqa: E402 from govoplan_deploy.model import ( # noqa: E402 SpecError, @@ -55,7 +58,104 @@ def run_cli(arguments: list[str]) -> tuple[int, str, str]: return result, stdout.getvalue(), stderr.getvalue() +def _kubernetes_test_pod(name: str, component: str, node: str) -> dict: + return { + "metadata": { + "name": name, + "uid": f"uid-{name}", + "labels": {"app.kubernetes.io/component": component}, + }, + "spec": {"nodeName": node}, + "status": {"conditions": [{"type": "Ready", "status": "True"}]}, + } + + +def _kubernetes_test_deployment(component: str, replicas: int) -> dict: + return { + "metadata": { + "name": f"govoplan-cluster-{component}", + "labels": {"app.kubernetes.io/component": component}, + }, + "spec": {"replicas": replicas}, + "status": {"availableReplicas": replicas, "updatedReplicas": replicas}, + } + + class DeploymentInstallerTests(unittest.TestCase): + def test_kubernetes_evidence_requires_two_node_spread_and_safe_runtime( + self, + ) -> None: + resources = { + "nodes": { + "items": [ + { + "metadata": {"name": name}, + "spec": {}, + "status": {"conditions": [{"type": "Ready", "status": "True"}]}, + } + for name in ("node-a", "node-b") + ] + }, + "pods": { + "items": [ + _kubernetes_test_pod("api-a", "api", "node-a"), + _kubernetes_test_pod("api-b", "api", "node-b"), + _kubernetes_test_pod("web-a", "web", "node-a"), + _kubernetes_test_pod("web-b", "web", "node-b"), + _kubernetes_test_pod("worker-a", "worker", "node-a"), + ] + }, + "deployments": { + "items": [ + _kubernetes_test_deployment("api", 2), + _kubernetes_test_deployment("web", 2), + _kubernetes_test_deployment("worker", 1), + ] + }, + } + + def run(arguments): + return resources[ + "nodes" + if "nodes" in arguments + else "deployments" + if "deployments" in arguments + else "pods" + ] + + evidence = collect_kubernetes_evidence( + installation_id="govoplan-cluster", + namespace="govoplan", + ops_url="https://govoplan.example.test/api/v1/ops/status", + api_key="not-retained", + command_runner=run, + json_fetcher=lambda _url, _key: { + "readiness": {"ready": True}, + "runtime_cluster": { + "expected": {"api": 2, "worker": 1}, + "active": {"api": 2, "worker": 1}, + "composition": {"skewed": False}, + "software_versions": {"skewed": False}, + "queues": {"missing": []}, + }, + "checks": [ + { + "id": "database_capacity", + "state": "ok", + "detail": "Within budget", + "metrics": {"peak": 30, "available": 90}, + } + ], + }, + ) + + self.assertEqual("passed", evidence["result"]["state"]) + self.assertNotIn("not-retained", json.dumps(evidence)) + self.assertEqual( + ["node-a", "node-b"], + evidence["snapshot"]["ready_node_names"], + ) + def test_pre_migration_failure_restores_checksum_verified_applied_bundle( self, ) -> None: @@ -212,6 +312,7 @@ class DeploymentInstallerTests(unittest.TestCase): "FILE_STORAGE_S3_ACCESS_KEY_ID": "object-key", "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "object-secret", "FILE_STORAGE_S3_BUCKET": "govoplan", + "GOVOPLAN_DB_CONNECTION_LIMIT": "100", }, ) @@ -257,6 +358,103 @@ class DeploymentInstallerTests(unittest.TestCase): "containers" ][0]["readinessProbe"]["httpGet"]["httpHeaders"], ) + worker_command = deployments["govoplan-cluster-worker"]["spec"]["template"][ + "spec" + ]["containers"][0]["command"] + self.assertIn("--concurrency", worker_command) + self.assertEqual( + "62", + manifest["metadata"]["annotations"][ + "govoplan.add-ideas.de/database-connection-peak" + ], + ) + + def test_kubernetes_export_splits_worker_queues_and_rejects_capacity_overrun( + self, + ) -> None: + spec = default_spec( + installation_id="govoplan-cluster", + postgres_mode="external", + redis_mode="external", + storage_mode="s3", + api_replicas=2, + web_replicas=2, + worker_replicas=3, + api_image="registry.example.test/govoplan-api@sha256:" + "a" * 64, + web_image="registry.example.test/govoplan-web@sha256:" + "b" * 64, + ) + environment = initial_secrets( + spec, + supplied={ + "DATABASE_URL": "postgresql+psycopg://user:db-secret@postgres.example.test/govoplan", + "GOVOPLAN_DATABASE_URL_PGTOOLS": "postgresql://user:db-secret@postgres.example.test/govoplan", + "REDIS_URL": "rediss://:redis-secret@redis.example.test/0", + "FILE_STORAGE_S3_ENDPOINT_URL": "https://s3.example.test", + "FILE_STORAGE_S3_REGION": "eu-test-1", + "FILE_STORAGE_S3_ACCESS_KEY_ID": "object-key", + "FILE_STORAGE_S3_SECRET_ACCESS_KEY": "object-secret", + "FILE_STORAGE_S3_BUCKET": "govoplan", + "GOVOPLAN_DB_CONNECTION_LIMIT": "100", + }, + ) + queues = environment["CELERY_QUEUES"].split(",") + environment["GOVOPLAN_WORKER_POOLS"] = json.dumps( + [ + { + "name": "delivery", + "queues": queues[:3], + "replicas": 2, + "concurrency": 2, + }, + { + "name": "platform", + "queues": queues[3:], + "replicas": 1, + "concurrency": 1, + }, + ] + ) + + manifest = render_kubernetes(spec, environment) + workers = [ + item + for item in manifest["items"] + if item["kind"] == "Deployment" + and item["metadata"]["labels"].get("app.kubernetes.io/component") + == "worker" + ] + + self.assertEqual(2, len(workers)) + self.assertEqual( + {"delivery", "platform"}, + { + item["metadata"]["labels"]["govoplan.add-ideas.de/worker-pool"] + for item in workers + }, + ) + for deployment in workers: + container = deployment["spec"]["template"]["spec"]["containers"][0] + explicit_environment = { + item["name"]: item.get("value") + for item in container["env"] + if "value" in item + } + self.assertEqual( + deployment["metadata"]["labels"]["govoplan.add-ideas.de/worker-pool"], + explicit_environment["GOVOPLAN_WORKER_POOL"], + ) + self.assertEqual( + set( + container["command"][ + container["command"].index("--queues") + 1 + ].split(",") + ), + set(explicit_environment["CELERY_QUEUES"].split(",")), + ) + + environment["GOVOPLAN_DB_CONNECTION_LIMIT"] = "40" + with self.assertRaisesRegex(ValueError, "connection peak"): + render_kubernetes(spec, environment) def test_kubernetes_export_rejects_mutable_release_images(self) -> None: spec = default_spec( diff --git a/tools/deployment/govoplan_deploy/bundle.py b/tools/deployment/govoplan_deploy/bundle.py index a1d605c..3032cd4 100644 --- a/tools/deployment/govoplan_deploy/bundle.py +++ b/tools/deployment/govoplan_deploy/bundle.py @@ -40,6 +40,18 @@ RUNTIME_ENV_KEYS = ( "ENABLED_MODULES", "CELERY_ENABLED", "CELERY_QUEUES", + "CELERY_WORKER_CONCURRENCY", + "GOVOPLAN_DB_CONNECTION_LIMIT", + "GOVOPLAN_DB_CONNECTION_RESERVE", + "GOVOPLAN_API_DB_POOL_SIZE", + "GOVOPLAN_API_DB_MAX_OVERFLOW", + "GOVOPLAN_WORKER_DB_POOL_SIZE", + "GOVOPLAN_WORKER_DB_MAX_OVERFLOW", + "GOVOPLAN_SCHEDULER_DB_POOL_SIZE", + "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW", + "GOVOPLAN_MIGRATION_DB_POOL_SIZE", + "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW", + "GOVOPLAN_WORKER_POOLS", "REDIS_URL", "CORS_ORIGINS", "GOVOPLAN_TRUSTED_HOSTS", @@ -193,6 +205,46 @@ def reconcile_runtime_environment( "CELERY_QUEUES": ( "send_email,append_sent,notifications,mail,calendar,dataflow,workflow,postbox,events,idm,default" ), + "CELERY_WORKER_CONCURRENCY": values.get( + "CELERY_WORKER_CONCURRENCY", + "2", + ), + "GOVOPLAN_DB_CONNECTION_RESERVE": values.get( + "GOVOPLAN_DB_CONNECTION_RESERVE", + "10", + ), + "GOVOPLAN_API_DB_POOL_SIZE": values.get( + "GOVOPLAN_API_DB_POOL_SIZE", + "5", + ), + "GOVOPLAN_API_DB_MAX_OVERFLOW": values.get( + "GOVOPLAN_API_DB_MAX_OVERFLOW", + "2", + ), + "GOVOPLAN_WORKER_DB_POOL_SIZE": values.get( + "GOVOPLAN_WORKER_DB_POOL_SIZE", + "1", + ), + "GOVOPLAN_WORKER_DB_MAX_OVERFLOW": values.get( + "GOVOPLAN_WORKER_DB_MAX_OVERFLOW", + "1", + ), + "GOVOPLAN_SCHEDULER_DB_POOL_SIZE": values.get( + "GOVOPLAN_SCHEDULER_DB_POOL_SIZE", + "1", + ), + "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW": values.get( + "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW", + "0", + ), + "GOVOPLAN_MIGRATION_DB_POOL_SIZE": values.get( + "GOVOPLAN_MIGRATION_DB_POOL_SIZE", + "2", + ), + "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW": values.get( + "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW", + "0", + ), "CORS_ORIGINS": spec.public_url, "GOVOPLAN_TRUSTED_HOSTS": public.hostname or "", "FORWARDED_ALLOW_IPS": spec.network_subnet, @@ -212,6 +264,8 @@ def reconcile_runtime_environment( values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "true" else: values["GOVOPLAN_ALLOW_PROCESS_LOCAL_LOGIN_THROTTLE"] = "false" + if postgres.mode == "managed": + values.setdefault("GOVOPLAN_DB_CONNECTION_LIMIT", "100") storage = spec.components.storage if storage.mode == "local": diff --git a/tools/deployment/govoplan_deploy/cli.py b/tools/deployment/govoplan_deploy/cli.py index 737675e..0df22b5 100644 --- a/tools/deployment/govoplan_deploy/cli.py +++ b/tools/deployment/govoplan_deploy/cli.py @@ -35,6 +35,7 @@ from .bundle import ( service_names, write_env, ) +from .cluster_evidence import collect_kubernetes_evidence from .model import ( ComponentConfig, DEFAULT_GARAGE_IMAGE, @@ -141,6 +142,37 @@ def build_parser() -> argparse.ArgumentParser: help="Output JSON path; defaults to /kubernetes.json.", ) + verify_kubernetes = subparsers.add_parser( + "verify-kubernetes", + help="Collect sanitized readiness evidence from a live multi-host profile.", + ) + _directory_argument(verify_kubernetes) + verify_kubernetes.add_argument("--namespace", default="govoplan") + verify_kubernetes.add_argument( + "--ops-url", + help="Ops status URL; defaults to /api/v1/ops/status.", + ) + verify_kubernetes.add_argument( + "--api-key-env", + default="GOVOPLAN_OPS_API_KEY", + help="Environment variable containing an API key with Ops read scope.", + ) + verify_kubernetes.add_argument( + "--exercise-api-pod-loss", + action="store_true", + help="Delete one ready API pod and prove health-aware replacement.", + ) + verify_kubernetes.add_argument( + "--timeout-seconds", + type=float, + default=180.0, + ) + verify_kubernetes.add_argument( + "--output", + type=Path, + help="Private evidence output path.", + ) + operations = subparsers.add_parser( "operations", help="List durable deployment and recovery operations.", @@ -282,6 +314,8 @@ def main(argv: Sequence[str] | None = None) -> int: return _status(args) if args.command == "render-kubernetes": return _render_kubernetes(args) + if args.command == "verify-kubernetes": + return _verify_kubernetes(args) if args.command == "operations": return _operations(args) if args.command == "recover": @@ -598,9 +632,7 @@ def _render_kubernetes(args: argparse.Namespace) -> int: output = (args.output or (paths.root / "kubernetes.json")).expanduser().resolve() atomic_write(output, canonical_json(manifest), mode=0o600) print(f"Wrote stateless Kubernetes runtime manifest to {output}") - print( - "Required Secret keys: " + ", ".join(kubernetes_secret_contract()) - ) + print("Required Secret keys: " + ", ".join(kubernetes_secret_contract())) print( write_secret_creation_hint( paths.env, @@ -611,6 +643,41 @@ def _render_kubernetes(args: argparse.Namespace) -> int: return 0 +def _verify_kubernetes(args: argparse.Namespace) -> int: + paths = bundle_paths(args.directory) + spec = load_spec(paths.spec) + api_key = str(os.environ.get(args.api_key_env) or "").strip() + if not api_key: + raise ValueError( + f"{args.api_key_env} must contain an API key with Ops read scope" + ) + ops_url = args.ops_url or (spec.public_url.rstrip("/") + "/api/v1/ops/status") + evidence = collect_kubernetes_evidence( + installation_id=spec.installation_id, + namespace=args.namespace, + ops_url=ops_url, + api_key=api_key, + exercise_api_pod_loss=args.exercise_api_pod_loss, + timeout_seconds=args.timeout_seconds, + ) + timestamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + output = ( + (args.output or paths.root / "evidence" / f"kubernetes-{timestamp}.json") + .expanduser() + .resolve() + ) + ensure_private_directory(output.parent) + atomic_write(output, canonical_json(evidence), mode=0o600) + print(f"Wrote Kubernetes evidence to {output}") + if evidence["result"]["state"] != "passed": + print( + "Failed checks: " + ", ".join(evidence["result"]["failed_checks"]), + file=sys.stderr, + ) + return 1 + return 0 + + def _operations(args: argparse.Namespace) -> int: paths = bundle_paths(args.directory) payload = list_operations(paths) diff --git a/tools/deployment/govoplan_deploy/cluster_evidence.py b/tools/deployment/govoplan_deploy/cluster_evidence.py new file mode 100644 index 0000000..13a361f --- /dev/null +++ b/tools/deployment/govoplan_deploy/cluster_evidence.py @@ -0,0 +1,380 @@ +"""Collect sanitized evidence for the Kubernetes multi-host profile.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +from datetime import UTC, datetime +import json +import shutil +import subprocess +import time +from typing import Any +from urllib.request import Request, urlopen + + +JsonObject = dict[str, Any] +CommandRunner = Callable[[Sequence[str]], JsonObject] +JsonFetcher = Callable[[str, str], JsonObject] + + +def collect_kubernetes_evidence( + *, + installation_id: str, + namespace: str, + ops_url: str, + api_key: str, + exercise_api_pod_loss: bool = False, + timeout_seconds: float = 180.0, + command_runner: CommandRunner | None = None, + json_fetcher: JsonFetcher | None = None, +) -> JsonObject: + """Inspect a live cluster and optionally exercise one API pod replacement.""" + + run_json = command_runner or _kubectl_json + fetch_json = json_fetcher or _fetch_json + nodes = run_json(("get", "nodes", "-o", "json")) + pods = run_json( + ( + "-n", + namespace, + "get", + "pods", + "-l", + f"app.kubernetes.io/instance={installation_id}", + "-o", + "json", + ) + ) + deployments = run_json( + ( + "-n", + namespace, + "get", + "deployments", + "-l", + f"app.kubernetes.io/instance={installation_id}", + "-o", + "json", + ) + ) + ops = fetch_json(ops_url, api_key) + snapshot = _evaluate_snapshot( + installation_id=installation_id, + namespace=namespace, + nodes=nodes, + pods=pods, + deployments=deployments, + ops=ops, + ) + replacement: JsonObject | None = None + if exercise_api_pod_loss: + replacement = _exercise_api_pod_loss( + installation_id=installation_id, + namespace=namespace, + ops_url=ops_url, + api_key=api_key, + initial_pods=pods, + timeout_seconds=timeout_seconds, + run_json=run_json, + fetch_json=fetch_json, + ) + evidence = { + "schema_version": 1, + "evidence_kind": "govoplan.kubernetes-multi-host", + "collected_at": datetime.now(UTC).isoformat(), + "installation_id": installation_id, + "namespace": namespace, + "snapshot": snapshot, + "api_pod_loss": replacement, + } + failures = [ + check["id"] for check in snapshot["checks"] if check["state"] != "passed" + ] + if replacement and replacement["state"] != "passed": + failures.append("api_pod_loss") + evidence["result"] = { + "state": "passed" if not failures else "failed", + "failed_checks": failures, + } + return evidence + + +def _evaluate_snapshot( + *, + installation_id: str, + namespace: str, + nodes: Mapping[str, Any], + pods: Mapping[str, Any], + deployments: Mapping[str, Any], + ops: Mapping[str, Any], +) -> JsonObject: + ready_nodes = [ + item + for item in _items(nodes) + if not bool(item.get("spec", {}).get("unschedulable")) + and _condition(item, "Ready") == "True" + ] + pod_rows = [_pod_summary(item) for item in _items(pods)] + deployment_rows = [_deployment_summary(item) for item in _items(deployments)] + ready_api = [ + item for item in pod_rows if item["component"] == "api" and item["ready"] + ] + ready_web = [ + item for item in pod_rows if item["component"] == "web" and item["ready"] + ] + runtime = ops.get("runtime_cluster") + runtime = runtime if isinstance(runtime, Mapping) else {} + checks = ops.get("checks") + check_rows = checks if isinstance(checks, list) else [] + check_by_id = { + str(item.get("id")): item for item in check_rows if isinstance(item, Mapping) + } + ops_readiness = ops.get("readiness") + ops_readiness = ops_readiness if isinstance(ops_readiness, Mapping) else {} + snapshot_checks = [ + _check( + "ready_nodes", + len(ready_nodes) >= 2, + f"{len(ready_nodes)} ready schedulable node(s)", + ), + _check( + "api_node_spread", + len(ready_api) >= 2 and len({item["node"] for item in ready_api}) >= 2, + f"{len(ready_api)} ready API pod(s) on {len({item['node'] for item in ready_api})} node(s)", + ), + _check( + "web_node_spread", + len(ready_web) >= 2 and len({item["node"] for item in ready_web}) >= 2, + f"{len(ready_web)} ready WebUI pod(s) on {len({item['node'] for item in ready_web})} node(s)", + ), + _check( + "deployment_availability", + bool(deployment_rows) + and all(item["available"] >= item["desired"] for item in deployment_rows), + f"{len(deployment_rows)} deployment(s) inspected", + ), + _check( + "ops_readiness", + bool(ops_readiness.get("ready")), + str(ops_readiness.get("detail") or "Ops readiness response inspected"), + ), + _check( + "runtime_composition", + not bool(runtime.get("composition", {}).get("skewed")), + "Active runtime composition matches the loaded module graph", + ), + _check( + "runtime_versions", + not bool(runtime.get("software_versions", {}).get("skewed")), + "Active runtime software versions are consistent", + ), + _check( + "worker_queue_coverage", + not bool(runtime.get("queues", {}).get("missing")), + "Every configured queue has an active worker", + ), + _check( + "database_connection_budget", + str(check_by_id.get("database_capacity", {}).get("state")) == "ok", + str( + check_by_id.get("database_capacity", {}).get("detail") + or "Database capacity check is unavailable" + ), + ), + ] + return { + "installation_id": installation_id, + "namespace": namespace, + "ready_node_names": sorted(_name(item) for item in ready_nodes), + "pods": sorted(pod_rows, key=lambda item: item["name"]), + "deployments": sorted(deployment_rows, key=lambda item: item["name"]), + "runtime": { + "expected": dict(runtime.get("expected") or {}), + "active": dict(runtime.get("active") or {}), + "composition": dict(runtime.get("composition") or {}), + "software_versions": dict(runtime.get("software_versions") or {}), + "queues": dict(runtime.get("queues") or {}), + }, + "database_capacity": dict( + check_by_id.get("database_capacity", {}).get("metrics") or {} + ), + "checks": snapshot_checks, + } + + +def _exercise_api_pod_loss( + *, + installation_id: str, + namespace: str, + ops_url: str, + api_key: str, + initial_pods: Mapping[str, Any], + timeout_seconds: float, + run_json: CommandRunner, + fetch_json: JsonFetcher, +) -> JsonObject: + candidates = [ + _pod_summary(item) + for item in _items(initial_pods) + if item.get("metadata", {}).get("labels", {}).get("app.kubernetes.io/component") + == "api" + and _condition(item, "Ready") == "True" + ] + if len(candidates) < 2: + return { + "state": "failed", + "detail": "At least two ready API pods are required for the loss drill.", + } + victim = sorted(candidates, key=lambda item: item["name"])[0] + initial_uids = {item["uid"] for item in candidates} + desired_ready = len(candidates) + run_json( + ( + "-n", + namespace, + "delete", + "pod", + victim["name"], + "--wait=false", + "-o", + "json", + ) + ) + deadline = time.monotonic() + timeout_seconds + health_failures = 0 + replacement: JsonObject | None = None + while time.monotonic() < deadline: + try: + current_ops = fetch_json(ops_url, api_key) + if not bool(current_ops.get("readiness", {}).get("ready")): + health_failures += 1 + except (OSError, ValueError): + health_failures += 1 + current = run_json( + ( + "-n", + namespace, + "get", + "pods", + "-l", + f"app.kubernetes.io/instance={installation_id},app.kubernetes.io/component=api", + "-o", + "json", + ) + ) + ready = [ + _pod_summary(item) + for item in _items(current) + if _condition(item, "Ready") == "True" + ] + new_pods = [item for item in ready if item["uid"] not in initial_uids] + if len(ready) >= desired_ready and new_pods: + replacement = sorted(new_pods, key=lambda item: item["name"])[-1] + break + time.sleep(2.0) + passed = replacement is not None and health_failures == 0 + return { + "state": "passed" if passed else "failed", + "victim": victim, + "replacement": replacement, + "readiness_failures": health_failures, + "detail": ( + "API pod was replaced without an observed readiness outage." + if passed + else "API replacement timed out or readiness failed during replacement." + ), + } + + +def _kubectl_json(arguments: Sequence[str]) -> JsonObject: + kubectl = shutil.which("kubectl") + if kubectl is None: + raise ValueError("kubectl is required for Kubernetes evidence collection") + result = subprocess.run( + (kubectl, *arguments), + check=False, + capture_output=True, + text=True, + timeout=60, + ) + if result.returncode: + detail = result.stderr.strip() or result.stdout.strip() + raise ValueError(f"kubectl failed: {detail}") + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError as exc: + raise ValueError("kubectl did not return JSON") from exc + if not isinstance(payload, dict): + raise ValueError("kubectl returned a non-object JSON payload") + return payload + + +def _fetch_json(url: str, api_key: str) -> JsonObject: + request = Request( + url, + headers={"Accept": "application/json", "X-API-Key": api_key}, + ) + with urlopen(request, timeout=15) as response: + payload = json.load(response) + if not isinstance(payload, dict): + raise ValueError("Ops API returned a non-object JSON payload") + return payload + + +def _items(payload: Mapping[str, Any]) -> list[Mapping[str, Any]]: + items = payload.get("items") + if not isinstance(items, list): + raise ValueError("Kubernetes list response has no items") + return [item for item in items if isinstance(item, Mapping)] + + +def _condition(item: Mapping[str, Any], condition_type: str) -> str | None: + conditions = item.get("status", {}).get("conditions", []) + for condition in conditions if isinstance(conditions, list) else []: + if isinstance(condition, Mapping) and condition.get("type") == condition_type: + return str(condition.get("status")) + return None + + +def _name(item: Mapping[str, Any]) -> str: + return str(item.get("metadata", {}).get("name") or "") + + +def _pod_summary(item: Mapping[str, Any]) -> JsonObject: + metadata = item.get("metadata", {}) + labels = metadata.get("labels", {}) + return { + "name": _name(item), + "uid": str(metadata.get("uid") or ""), + "component": str(labels.get("app.kubernetes.io/component") or ""), + "worker_pool": labels.get("govoplan.add-ideas.de/worker-pool"), + "node": str(item.get("spec", {}).get("nodeName") or ""), + "ready": _condition(item, "Ready") == "True", + } + + +def _deployment_summary(item: Mapping[str, Any]) -> JsonObject: + status = item.get("status", {}) + return { + "name": _name(item), + "component": str( + item.get("metadata", {}) + .get("labels", {}) + .get("app.kubernetes.io/component") + or "" + ), + "desired": int(item.get("spec", {}).get("replicas") or 0), + "available": int(status.get("availableReplicas") or 0), + "updated": int(status.get("updatedReplicas") or 0), + } + + +def _check(check_id: str, passed: bool, detail: str) -> JsonObject: + return { + "id": check_id, + "state": "passed" if passed else "failed", + "detail": detail, + } + + +__all__ = ["collect_kubernetes_evidence"] diff --git a/tools/deployment/govoplan_deploy/kubernetes.py b/tools/deployment/govoplan_deploy/kubernetes.py index 3b88d77..07b6326 100644 --- a/tools/deployment/govoplan_deploy/kubernetes.py +++ b/tools/deployment/govoplan_deploy/kubernetes.py @@ -2,7 +2,9 @@ from __future__ import annotations +from dataclasses import dataclass from hashlib import sha256 +import json from pathlib import Path import re from typing import Any, Mapping @@ -26,6 +28,17 @@ _CONFIG_KEYS = ( "ENABLED_MODULES", "CELERY_ENABLED", "CELERY_QUEUES", + "GOVOPLAN_DB_CONNECTION_LIMIT", + "GOVOPLAN_DB_CONNECTION_RESERVE", + "GOVOPLAN_API_DB_POOL_SIZE", + "GOVOPLAN_API_DB_MAX_OVERFLOW", + "GOVOPLAN_WORKER_DB_POOL_SIZE", + "GOVOPLAN_WORKER_DB_MAX_OVERFLOW", + "GOVOPLAN_SCHEDULER_DB_POOL_SIZE", + "GOVOPLAN_SCHEDULER_DB_MAX_OVERFLOW", + "GOVOPLAN_MIGRATION_DB_POOL_SIZE", + "GOVOPLAN_MIGRATION_DB_MAX_OVERFLOW", + "GOVOPLAN_WORKER_POOLS", "CORS_ORIGINS", "GOVOPLAN_TRUSTED_HOSTS", "FORWARDED_ALLOW_IPS", @@ -43,6 +56,15 @@ _CONFIG_KEYS = ( "FILE_STORAGE_S3_DEPLOYMENT_MANAGED", "FILE_STORAGE_S3_ENDPOINT_TRUSTED", ) +_QUEUE_NAME = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$") + + +@dataclass(frozen=True, slots=True) +class _WorkerPool: + name: str + queues: tuple[str, ...] + replicas: int + concurrency: int def render_kubernetes( @@ -57,6 +79,8 @@ def render_kubernetes( """Render runtime roles only; shared state services stay externally managed.""" _validate_cluster_profile(spec, environment, namespace, secret_name) + worker_pools = _worker_pools(spec, environment) + database_capacity = _database_capacity(spec, environment, worker_pools) name = _resource_name(spec.installation_id) public_host = urlsplit(spec.public_url).hostname or "localhost" labels = {"app.kubernetes.io/name": "govoplan", "app.kubernetes.io/instance": name} @@ -75,6 +99,8 @@ def render_kubernetes( "GOVOPLAN_RUNTIME_STALE_AFTER_SECONDS": "60", "GOVOPLAN_EXPECTED_API_REPLICAS": str(spec.replicas.api), "GOVOPLAN_EXPECTED_WORKER_REPLICAS": str(spec.replicas.worker), + "GOVOPLAN_DB_CONNECTION_PEAK": str(database_capacity["peak"]), + "GOVOPLAN_DB_CONNECTION_AVAILABLE": str(database_capacity["available"]), "FILE_STORAGE_BACKEND": "s3", "FILE_STORAGE_S3_DEPLOYMENT_MANAGED": "false", "FILE_STORAGE_S3_ENDPOINT_TRUSTED": "true", @@ -130,6 +156,7 @@ def render_kubernetes( readiness_path="/health/ready", liveness_path="/health", probe_host=public_host, + extra_environment=_role_database_environment(environment, "API"), ), _service( name=f"{name}-api", @@ -168,16 +195,25 @@ def render_kubernetes( config_name=config_name, secret_name=secret_name, service_account=service_account, + database_environment=_role_database_environment( + environment, + "MIGRATION", + ), ), ] - if spec.replicas.worker: + for pool in worker_pools: + worker_name = ( + f"{name}-worker" + if len(worker_pools) == 1 and pool.name == "default" + else _name_with_suffix(name, f"worker-{pool.name}") + ) items.append( _deployment( - name=f"{name}-worker", + name=worker_name, namespace=namespace, labels=labels, role="worker", - replicas=spec.replicas.worker, + replicas=pool.replicas, image=spec.release.api_image, command=( "python", @@ -187,15 +223,39 @@ def render_kubernetes( "govoplan_core.celery_app:celery", "worker", "--queues", - str(environment["CELERY_QUEUES"]), + ",".join(pool.queues), + "--concurrency", + str(pool.concurrency), "--loglevel", "INFO", ), config_name=config_name, secret_name=secret_name, service_account=service_account, + extra_environment={ + **_role_database_environment(environment, "WORKER"), + "CELERY_QUEUES": ",".join(pool.queues), + "CELERY_WORKER_CONCURRENCY": str(pool.concurrency), + "GOVOPLAN_WORKER_POOL": pool.name, + }, + selector_labels={ + "govoplan.add-ideas.de/worker-pool": pool.name, + }, ) ) + if pool.replicas > 1: + items.append( + _pod_disruption_budget( + worker_name, + namespace, + labels, + "worker", + selector_labels={ + "govoplan.add-ideas.de/worker-pool": pool.name, + }, + ) + ) + if worker_pools: items.append( _deployment( name=f"{name}-scheduler", @@ -225,6 +285,10 @@ def render_kubernetes( config_name=config_name, secret_name=secret_name, service_account=service_account, + extra_environment=_role_database_environment( + environment, + "SCHEDULER", + ), ) ) if spec.replicas.api > 1: @@ -248,6 +312,15 @@ def render_kubernetes( "annotations": { "govoplan.add-ideas.de/profile": "stateless-shared-state", "govoplan.add-ideas.de/secret-contract": ",".join(_SECRET_KEYS), + "govoplan.add-ideas.de/database-connection-peak": str( + database_capacity["peak"] + ), + "govoplan.add-ideas.de/database-connection-available": str( + database_capacity["available"] + ), + "govoplan.add-ideas.de/worker-pools": ",".join( + pool.name for pool in worker_pools + ), } }, "items": items, @@ -290,7 +363,13 @@ def _validate_cluster_profile( + ", ".join(unpinned_images) ) missing = [ - key for key in (*_SECRET_KEYS, "CELERY_QUEUES") if not environment.get(key) + key + for key in ( + *_SECRET_KEYS, + "CELERY_QUEUES", + "GOVOPLAN_DB_CONNECTION_LIMIT", + ) + if not environment.get(key) ] if missing: raise ValueError( @@ -298,6 +377,245 @@ def _validate_cluster_profile( ) +def _worker_pools( + spec: InstallationSpec, + environment: Mapping[str, str], +) -> tuple[_WorkerPool, ...]: + if spec.replicas.worker == 0: + if str(environment.get("GOVOPLAN_WORKER_POOLS") or "").strip(): + raise ValueError("Worker pools require at least one worker replica") + return () + configured_queues = tuple( + item.strip() + for item in str(environment.get("CELERY_QUEUES") or "").split(",") + if item.strip() + ) + if not configured_queues or any( + _QUEUE_NAME.fullmatch(item) is None for item in configured_queues + ): + raise ValueError("CELERY_QUEUES must contain canonical queue names") + raw_pools = str(environment.get("GOVOPLAN_WORKER_POOLS") or "").strip() + if not raw_pools: + return ( + _WorkerPool( + name="default", + queues=configured_queues, + replicas=spec.replicas.worker, + concurrency=_bounded_environment_integer( + environment, + "CELERY_WORKER_CONCURRENCY", + default=2, + minimum=1, + maximum=64, + ), + ), + ) + try: + payload = json.loads(raw_pools) + except json.JSONDecodeError as exc: + raise ValueError("GOVOPLAN_WORKER_POOLS must be valid JSON") from exc + if not isinstance(payload, list) or not payload: + raise ValueError("GOVOPLAN_WORKER_POOLS must be a non-empty JSON list") + pools: list[_WorkerPool] = [] + seen_names: set[str] = set() + seen_queues: set[str] = set() + for index, item in enumerate(payload): + if not isinstance(item, dict): + raise ValueError(f"Worker pool {index} must be an object") + unknown = set(item) - {"name", "queues", "replicas", "concurrency"} + if unknown: + raise ValueError( + f"Worker pool {index} has unsupported keys: " + + ", ".join(sorted(unknown)) + ) + pool_name = item.get("name") + queues = item.get("queues") + replicas = item.get("replicas") + concurrency = item.get("concurrency") + if not isinstance(pool_name, str) or _DNS_LABEL.fullmatch(pool_name) is None: + raise ValueError(f"Worker pool {index} has an invalid name") + if pool_name in seen_names: + raise ValueError(f"Worker pool name is duplicated: {pool_name}") + if ( + not isinstance(queues, list) + or not queues + or any( + not isinstance(queue, str) or _QUEUE_NAME.fullmatch(queue) is None + for queue in queues + ) + ): + raise ValueError(f"Worker pool {pool_name} has invalid queues") + duplicate_queues = seen_queues.intersection(queues) + if duplicate_queues: + raise ValueError( + "Worker queues must have one owning pool: " + + ", ".join(sorted(duplicate_queues)) + ) + if not isinstance(replicas, int) or not 1 <= replicas <= 128: + raise ValueError( + f"Worker pool {pool_name} replicas must be between 1 and 128" + ) + if not isinstance(concurrency, int) or not 1 <= concurrency <= 64: + raise ValueError( + f"Worker pool {pool_name} concurrency must be between 1 and 64" + ) + pools.append( + _WorkerPool( + name=pool_name, + queues=tuple(queues), + replicas=replicas, + concurrency=concurrency, + ) + ) + seen_names.add(pool_name) + seen_queues.update(queues) + if sum(pool.replicas for pool in pools) != spec.replicas.worker: + raise ValueError("Worker pool replicas must equal installation replicas.worker") + configured_queue_set = set(configured_queues) + if seen_queues != configured_queue_set: + missing = configured_queue_set - seen_queues + unknown = seen_queues - configured_queue_set + detail = [] + if missing: + detail.append("missing " + ", ".join(sorted(missing))) + if unknown: + detail.append("unknown " + ", ".join(sorted(unknown))) + raise ValueError( + "Worker pools must cover CELERY_QUEUES exactly: " + "; ".join(detail) + ) + return tuple(sorted(pools, key=lambda pool: pool.name)) + + +def _database_capacity( + spec: InstallationSpec, + environment: Mapping[str, str], + worker_pools: tuple[_WorkerPool, ...], +) -> dict[str, int]: + limit = _bounded_environment_integer( + environment, + "GOVOPLAN_DB_CONNECTION_LIMIT", + minimum=10, + maximum=1_000_000, + ) + reserve = _bounded_environment_integer( + environment, + "GOVOPLAN_DB_CONNECTION_RESERVE", + default=10, + minimum=1, + maximum=999_999, + ) + if reserve >= limit: + raise ValueError("Database connection reserve must be below the limit") + api_ceiling = _role_database_ceiling(environment, "API", 5, 2) + worker_ceiling = _role_database_ceiling(environment, "WORKER", 1, 1) + scheduler_ceiling = _role_database_ceiling(environment, "SCHEDULER", 1, 0) + migration_ceiling = _role_database_ceiling(environment, "MIGRATION", 2, 0) + worker_processes = sum( + pool.replicas * (pool.concurrency + 1) for pool in worker_pools + ) + steady = ( + spec.replicas.api * api_ceiling + + worker_processes * worker_ceiling + + (scheduler_ceiling if worker_pools else 0) + ) + rollout_surge = api_ceiling + migration_ceiling + if worker_pools: + rollout_surge += scheduler_ceiling + rollout_surge += sum( + (pool.concurrency + 1) * worker_ceiling for pool in worker_pools + ) + peak = steady + rollout_surge + available = limit - reserve + if peak > available: + raise ValueError( + "Kubernetes database connection peak exceeds the configured budget: " + f"peak={peak}, available={available}, limit={limit}, reserve={reserve}" + ) + return { + "limit": limit, + "reserve": reserve, + "available": available, + "steady": steady, + "peak": peak, + } + + +def _role_database_ceiling( + environment: Mapping[str, str], + role: str, + default_pool_size: int, + default_overflow: int, +) -> int: + return _bounded_environment_integer( + environment, + f"GOVOPLAN_{role}_DB_POOL_SIZE", + default=default_pool_size, + minimum=1, + maximum=100, + ) + _bounded_environment_integer( + environment, + f"GOVOPLAN_{role}_DB_MAX_OVERFLOW", + default=default_overflow, + minimum=0, + maximum=200, + ) + + +def _role_database_environment( + environment: Mapping[str, str], + role: str, +) -> dict[str, str]: + defaults = { + "API": (5, 2), + "WORKER": (1, 1), + "SCHEDULER": (1, 0), + "MIGRATION": (2, 0), + } + pool_size, overflow = defaults[role] + return { + "GOVOPLAN_DB_POOL_SIZE": str( + _bounded_environment_integer( + environment, + f"GOVOPLAN_{role}_DB_POOL_SIZE", + default=pool_size, + minimum=1, + maximum=100, + ) + ), + "GOVOPLAN_DB_MAX_OVERFLOW": str( + _bounded_environment_integer( + environment, + f"GOVOPLAN_{role}_DB_MAX_OVERFLOW", + default=overflow, + minimum=0, + maximum=200, + ) + ), + } + + +def _bounded_environment_integer( + environment: Mapping[str, str], + name: str, + *, + default: int | None = None, + minimum: int, + maximum: int, +) -> int: + raw = environment.get(name) + if raw is None or not str(raw).strip(): + if default is None: + raise ValueError(f"{name} is required") + return default + try: + value = int(str(raw)) + except ValueError as exc: + raise ValueError(f"{name} must be an integer") from exc + if not minimum <= value <= maximum: + raise ValueError(f"{name} must be between {minimum} and {maximum}") + return value + + def _deployment( *, name: str, @@ -315,8 +633,13 @@ def _deployment( liveness_path: str | None = None, probe_host: str | None = None, extra_environment: Mapping[str, str] | None = None, + selector_labels: Mapping[str, str] | None = None, ) -> dict[str, Any]: - role_labels = {**labels, "app.kubernetes.io/component": role} + role_labels = { + **labels, + "app.kubernetes.io/component": role, + **dict(selector_labels or {}), + } environment: list[dict[str, Any]] = [ {"name": "TMPDIR", "value": "/tmp"}, {"name": "GOVOPLAN_RUNTIME_ROLE", "value": role}, @@ -400,6 +723,8 @@ def _deployment( "env": [ {"name": "TMPDIR", "value": "/tmp"}, {"name": "GOVOPLAN_RUNTIME_ROLE", "value": "migration-wait"}, + {"name": "GOVOPLAN_DB_POOL_SIZE", "value": "1"}, + {"name": "GOVOPLAN_DB_MAX_OVERFLOW", "value": "0"}, *_secret_environment(secret_name), ], "securityContext": { @@ -439,6 +764,7 @@ def _migration_job( config_name: str, secret_name: str, service_account: str, + database_environment: Mapping[str, str], ) -> dict[str, Any]: job_labels = {**labels, "app.kubernetes.io/component": "migration"} job_name = _name_with_suffix(name, f"migrate-{release_key}") @@ -481,6 +807,12 @@ def _migration_job( "env": [ {"name": "TMPDIR", "value": "/tmp"}, {"name": "GOVOPLAN_RUNTIME_ROLE", "value": "migration"}, + *( + {"name": key, "value": value} + for key, value in sorted( + database_environment.items() + ) + ), { "name": "GOVOPLAN_NODE_ID", "valueFrom": { @@ -541,8 +873,14 @@ def _pod_disruption_budget( namespace: str, labels: dict[str, str], role: str, + *, + selector_labels: Mapping[str, str] | None = None, ) -> dict[str, Any]: - role_labels = {**labels, "app.kubernetes.io/component": role} + role_labels = { + **labels, + "app.kubernetes.io/component": role, + **dict(selector_labels or {}), + } return { "apiVersion": "policy/v1", "kind": "PodDisruptionBudget",