From b10a7d150a5b7a837195bd4443e86fd06c57543f Mon Sep 17 00:00:00 2001 From: ogt Date: Tue, 14 Jul 2026 16:23:30 +0800 Subject: [PATCH] fix(api): delegate production bootstrap ddl --- apps/api/src/core/config.py | 2 +- apps/api/src/main.py | 28 +- .../tests/test_runtime_bootstrap_guards.py | 82 ++++ k8s/awoooi-prod/06-deployment-api.yaml | 18 +- .../test_cd_controlled_runtime_profile.py | 7 + ...erify_awoooi_api_liveness_stabilization.py | 100 +++++ ...erify_awoooi_api_liveness_stabilization.py | 383 ++++++++++++++++++ 7 files changed, 608 insertions(+), 12 deletions(-) create mode 100644 ops/runner/test_verify_awoooi_api_liveness_stabilization.py create mode 100644 ops/runner/verify_awoooi_api_liveness_stabilization.py diff --git a/apps/api/src/core/config.py b/apps/api/src/core/config.py index bb8edbeed..e34d9496b 100644 --- a/apps/api/src/core/config.py +++ b/apps/api/src/core/config.py @@ -302,7 +302,7 @@ class Settings(BaseSettings): default=True, description=( "Run idempotent schema bootstrap DDL during process startup. " - "Production workers disable this after CD migration preflight." + "Production workloads disable this after CD migration preflight." ), ) diff --git a/apps/api/src/main.py b/apps/api/src/main.py index c23f906fe..640a2cc85 100644 --- a/apps/api/src/main.py +++ b/apps/api/src/main.py @@ -155,6 +155,25 @@ def _close_unscheduled_coroutine(coro: Coroutine[Any, Any, Any]) -> None: logger.warning("api_background_task_close_failed", error=str(exc)) +async def initialize_api_database() -> bool: + """Run local/dev bootstrap DDL or emit the production ownership receipt.""" + if not settings.DATABASE_BOOTSTRAP_DDL_ENABLED: + logger.info( + "database_bootstrap_ddl_skipped", + owner="cd_migration_preflight", + workload="api", + reason="database_bootstrap_ddl_disabled", + ) + return False + + await init_db() + db_url = settings.DATABASE_URL + logger.info( + "database_initialized", url=db_url.split("@")[-1] if "@" in db_url else db_url + ) + return True + + def _resolve_request_project_context(request: Request) -> tuple[str | None, str]: """Resolve tenant context for RLS while keeping non-webhook routes fail-closed.""" for candidate in ( @@ -293,12 +312,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]: return None return asyncio.create_task(coro, name=task_name) - # CTO-201: Initialize PostgreSQL database (統帥鐵律: 禁止 SQLite) - await init_db() - db_url = settings.DATABASE_URL - logger.info( - "database_initialized", url=db_url.split("@")[-1] if "@" in db_url else db_url - ) + # CTO-201: production DDL belongs to CD migration preflight; local/dev keeps + # the default bootstrap path (統帥鐵律: 禁止 SQLite). + await initialize_api_database() # Phase 5: Initialize HTTP Clients (ClickHouse, Ollama) # 統帥鐵律: 連線池在啟動時建立，關閉時回收 diff --git a/apps/api/tests/test_runtime_bootstrap_guards.py b/apps/api/tests/test_runtime_bootstrap_guards.py index cdc406b33..f4b36be32 100644 --- a/apps/api/tests/test_runtime_bootstrap_guards.py +++ b/apps/api/tests/test_runtime_bootstrap_guards.py @@ -84,6 +84,88 @@ def test_production_worker_delegates_bootstrap_ddl_to_cd() -> None: assert 'owner="cd_migration_preflight"' in source +def test_production_api_delegates_bootstrap_ddl_to_cd() -> None: + from src import main as api_main + + root = Path(__file__).resolve().parents[3] + documents = yaml.safe_load_all( + (root / "k8s/awoooi-prod/06-deployment-api.yaml").read_text(encoding="utf-8") + ) + manifest = next( + document for document in documents if document.get("kind") == "Deployment" + ) + container = manifest["spec"]["template"]["spec"]["containers"][0] + env = {item["name"]: item.get("value") for item in container["env"]} + source = inspect.getsource(api_main.initialize_api_database) + + assert ( + type(api_main.settings).model_fields["DATABASE_BOOTSTRAP_DDL_ENABLED"].default + is True + ) + assert env["DATABASE_BOOTSTRAP_DDL_ENABLED"] == "false" + assert "if not settings.DATABASE_BOOTSTRAP_DDL_ENABLED" in source + assert 'owner="cd_migration_preflight"' in source + assert 'workload="api"' in source + + +@pytest.mark.asyncio +async def test_api_database_bootstrap_runs_when_enabled(monkeypatch) -> None: + from src import main as api_main + + calls: list[str] = [] + logs: list[tuple[str, dict[str, object]]] = [] + + async def fake_init_db() -> None: + calls.append("init_db") + + class _FakeLogger: + def info(self, event: str, **fields: object) -> None: + logs.append((event, fields)) + + monkeypatch.setattr(api_main.settings, "DATABASE_BOOTSTRAP_DDL_ENABLED", True) + monkeypatch.setattr( + api_main.settings, + "DATABASE_URL", + "postgresql+asyncpg://test:test@127.0.0.1/test", + ) + monkeypatch.setattr(api_main, "init_db", fake_init_db) + monkeypatch.setattr(api_main, "logger", _FakeLogger()) + + assert await api_main.initialize_api_database() is True + assert calls == ["init_db"] + assert logs == [("database_initialized", {"url": "127.0.0.1/test"})] + + +@pytest.mark.asyncio +async def test_api_database_bootstrap_skips_with_cd_owner_receipt(monkeypatch) -> None: + from src import main as api_main + + logs: list[tuple[str, dict[str, object]]] = [] + + async def fail_if_called() -> None: + raise AssertionError("production API must not execute bootstrap DDL") + + class _FakeLogger: + def info(self, event: str, **fields: object) -> None: + logs.append((event, fields)) + + monkeypatch.setattr(api_main.settings, "DATABASE_BOOTSTRAP_DDL_ENABLED", False) + monkeypatch.setattr(api_main, "init_db", fail_if_called) + monkeypatch.setattr(api_main, "logger", _FakeLogger()) + + assert await api_main.initialize_api_database() is False + assert logs == [ + ( + "database_bootstrap_ddl_skipped", + { + "owner": "cd_migration_preflight", + "workload": "api", + "reason": "database_bootstrap_ddl_disabled", + }, + ) + ] + + def test_get_engine_uses_database_pool_budget(monkeypatch): from src.db import base as db_base diff --git a/k8s/awoooi-prod/06-deployment-api.yaml b/k8s/awoooi-prod/06-deployment-api.yaml index 7792645c5..35f10302f 100644 --- a/k8s/awoooi-prod/06-deployment-api.yaml +++ b/k8s/awoooi-prod/06-deployment-api.yaml @@ -14,11 +14,14 @@ metadata: system: awoooi environment: prod spec: - # 2026-07-01 Codex: temporary post-reboot recovery cap. The production - # database role `awoooi` currently has CONNECTION LIMIT 2, so a two-replica - # API rollout deadlocks when the old pod still holds DB connections. Restore - # replicas=2 only after the DB role limit / pool budget is raised and verified. - replicas: 1 + # 2026-07-14 Codex: keep the bounded two-replica serving topology only with + # startup DDL delegated to CD migration preflight below. The initial canary + # briefly reached 2/2 Ready and 200m/238m CPU, then failed when both API pods + # re-entered init_db and saturated at the 1 CPU limit. The independent + # verifier therefore requires the DDL gate, DB CONNECTION LIMIT 12 receipt, + # 2/2 Ready, zero restart delta, and unsaturated CPU for a full probe window. + # Roll back to replicas=1 if any post-deploy condition fails. + replicas: 2 revisionHistoryLimit: 3 selector: matchLabels: @@ -174,6 +177,11 @@ spec: # 2026-07-11 Codex: release each API connection after use while # workload-specific DB identities are being split and verified. value: "true" + - name: DATABASE_BOOTSTRAP_DDL_ENABLED + # 2026-07-14 Codex: production schema DDL has one owner. CD + # migration preflight applies/verifies it before API rollout; + # API startup emits an explicit no-write ownership receipt. + value: "false" - name: AWOOOI_API_BACKGROUND_TASKS_ENABLED value: "false" - name: IWOOOS_WAZUH_READONLY_ENABLED diff --git a/ops/runner/test_cd_controlled_runtime_profile.py b/ops/runner/test_cd_controlled_runtime_profile.py index 3a9935431..54ff89fa0 100644 --- a/ops/runner/test_cd_controlled_runtime_profile.py +++ b/ops/runner/test_cd_controlled_runtime_profile.py @@ -124,6 +124,13 @@ def test_prod_db_pool_budget_and_non_overlap_rollouts_are_source_controlled() -> assert "value: \"1\"" in api assert "DATABASE_MAX_OVERFLOW" in api assert "value: \"0\"" in api + assert "DATABASE_BOOTSTRAP_DDL_ENABLED" in api + assert "CD migration preflight" in api + assert "replicas: 2" in api + assert "CONNECTION" in api and "LIMIT 12" in api + assert "Roll back to replicas=1" in api + assert "maxSurge: 0" in api + assert "maxUnavailable: 1" in api assert "DATABASE_POOL_SIZE" in worker assert "DATABASE_MAX_OVERFLOW" in worker diff --git a/ops/runner/test_verify_awoooi_api_liveness_stabilization.py b/ops/runner/test_verify_awoooi_api_liveness_stabilization.py new file mode 100644 index 000000000..5428f27c7 --- /dev/null +++ b/ops/runner/test_verify_awoooi_api_liveness_stabilization.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import importlib.util +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[2] +SCRIPT = ROOT / "ops" / "runner" / "verify_awoooi_api_liveness_stabilization.py" +SPEC = importlib.util.spec_from_file_location("api_liveness_verifier", SCRIPT) +assert SPEC and SPEC.loader +verifier = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(verifier) + + +def _live_deployment() -> dict: + deployment = verifier._source_deployment() + deployment["status"] = { + "replicas": 2, + "updatedReplicas": 2, + "availableReplicas": 2, + "readyReplicas": 2, + } + return deployment + + +def _receipt() -> dict[str, str]: + return { + "connection_budget_verified": "true", + "api_database_null_pool": "true", + "api_representative_db_preflight_passed": "true", + "api_http_concurrency_probe_passed": "true", + "api_db_connection_limit": "12", + "api_required_connection_budget": "8", + "api_available_connection_headroom": "11", + } + + +def _pods() -> list[dict]: + return [ + { + "metadata": {"name": name}, + "status": {"containerStatuses": [{"name": "api", "ready": True}]}, + } + for name in ("awoooi-api-a", "awoooi-api-b") + ] + + +def test_source_contract_is_bounded_and_rollback_ready() -> None: + deployment = verifier._source_deployment() + assert verifier.evaluate_source_contract(deployment) == [] + edge_config = (verifier.ROOT / "ops/nginx/awoooi.wooo.work.conf").read_text() + assert verifier.evaluate_routing_contract( + verifier._source_service(), edge_config + ) == [] + assert verifier.ROLLBACK_COMMAND.endswith("--replicas=1") + + +def test_live_contract_accepts_two_ready_unsaturated_stable_pods() -> None: + pods = _pods() + restarts = {pod["metadata"]["name"]: 0 for pod in pods} + cpu = {pod["metadata"]["name"]: 410 for pod in pods} + assert verifier.evaluate_live_state( + _live_deployment(), _receipt(), pods, restarts, restarts, cpu, 900 + ) == [] + + +def test_live_contract_fails_closed_on_budget_restart_or_cpu_regression() -> None: + pods = _pods() + receipt = _receipt() + receipt["connection_budget_verified"] = "false" + failures = verifier.evaluate_live_state( + _live_deployment(), + receipt, + pods, + {"awoooi-api-a": 0, "awoooi-api-b": 0}, + {"awoooi-api-a": 1, "awoooi-api-b": 0}, + {"awoooi-api-a": 1000, "awoooi-api-b": 300}, + 900, + ) + assert "receipt_connection_budget_verified_not_verified" in failures + assert "live_restart_delta:1" in failures + assert "live_cpu_saturated:awoooi-api-a:1000m" in failures + + +def test_edge_receipt_requires_a_qualified_production_vantage() -> None: + assert verifier.evaluate_edge_receipt( + "192.168.0.120", + {"vip125": 200, "node120": 200, "node121": 200}, + ) == [] + assert verifier.evaluate_edge_receipt( + None, + {"vip125": None, "node120": None, "node121": None}, + ) == ["edge_production_vantage_unavailable"] + + +def test_local_vip_failure_is_not_used_as_production_edge_receipt() -> None: + qualified = {"vip125": 200, "node120": 200, "node121": 200} + local_unqualified_status = None + assert local_unqualified_status is None + assert verifier.evaluate_edge_receipt("192.168.0.110", qualified) == [] diff --git a/ops/runner/verify_awoooi_api_liveness_stabilization.py b/ops/runner/verify_awoooi_api_liveness_stabilization.py new file mode 100644 index 000000000..d3106e41e --- /dev/null +++ b/ops/runner/verify_awoooi_api_liveness_stabilization.py @@ -0,0 +1,383 @@ +#!/usr/bin/env python3 +"""Fail-closed verifier for the bounded AWOOOI API two-replica stabilization.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import time +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + + +ROOT = Path(__file__).resolve().parents[2] +MANIFEST = ROOT / "k8s" / "awoooi-prod" / "06-deployment-api.yaml" +NAMESPACE = "awoooi-prod" +DEPLOYMENT = "awoooi-api" +BOUNDARY_RECEIPT = "awoooi-executor-boundary-verification" +PUBLIC_LIVENESS_URL = "https://awoooi.wooo.work/api/v1/health/live" +EDGE_VIP_LIVENESS_URL = "http://192.168.0.125:32334/api/v1/health/live" +EDGE_LIVENESS_URLS = { + "vip125": EDGE_VIP_LIVENESS_URL, + "node120": "http://192.168.0.120:32334/api/v1/health/live", + "node121": "http://192.168.0.121:32334/api/v1/health/live", +} +EDGE_VANTAGE_HOSTS = ("192.168.0.120", "192.168.0.110") +ROLLBACK_COMMAND = ( + "kubectl -n awoooi-prod scale deployment/awoooi-api --replicas=1" +) + + +def _source_documents() -> list[dict[str, Any]]: + """Render the manifest with kubectl's client parser; never contact the API.""" + completed = subprocess.run( + ["kubectl", "create", "--dry-run=client", "-f", str(MANIFEST), "-o", "json"], + check=True, + capture_output=True, + text=True, + timeout=20, + ) + decoder = json.JSONDecoder() + remaining = completed.stdout + documents: list[dict[str, Any]] = [] + while remaining.strip(): + document, offset = decoder.raw_decode(remaining.lstrip()) + remaining = remaining.lstrip()[offset:] + documents.append(document) + return documents + + +def _source_deployment() -> dict[str, Any]: + return next( + document + for document in _source_documents() + if document.get("kind") == "Deployment" + ) + + +def _source_service() -> dict[str, Any]: + return next( + document + for document in _source_documents() + if document.get("kind") == "Service" + ) + + +def _env(container: dict[str, Any]) -> dict[str, str | None]: + return {entry["name"]: entry.get("value") for entry in container.get("env", [])} + + +def evaluate_source_contract(deployment: dict[str, Any]) -> list[str]: + failures: list[str] = [] + spec = deployment.get("spec") or {} + strategy = (spec.get("strategy") or {}).get("rollingUpdate") or {} + container = spec["template"]["spec"]["containers"][0] + env = _env(container) + + expectations = { + "replicas": (spec.get("replicas"), 2), + "max_surge": (strategy.get("maxSurge"), 0), + "max_unavailable": (strategy.get("maxUnavailable"), 1), + "cpu_limit": ((container.get("resources") or {}).get("limits", {}).get("cpu"), "1"), + "memory_limit": ((container.get("resources") or {}).get("limits", {}).get("memory"), "1Gi"), + "database_pool_size": (env.get("DATABASE_POOL_SIZE"), "1"), + "database_max_overflow": (env.get("DATABASE_MAX_OVERFLOW"), "0"), + "database_null_pool": (env.get("DATABASE_NULL_POOL"), "true"), + "database_bootstrap_ddl_enabled": ( + env.get("DATABASE_BOOTSTRAP_DDL_ENABLED"), + "false", + ), + "api_background_tasks": (env.get("AWOOOI_API_BACKGROUND_TASKS_ENABLED"), "false"), + } + for name, (actual, expected) in expectations.items(): + if actual != expected: + failures.append(f"source_{name}_mismatch:{actual!r}!={expected!r}") + + probes = { + name: container.get(name) or {} + for name in ("startupProbe", "livenessProbe", "readinessProbe") + } + for name, probe in probes.items(): + expected_path = ( + "/api/v1/health/ready" + if name == "readinessProbe" + else "/api/v1/health/live" + ) + if (probe.get("httpGet") or {}).get("path") != expected_path: + failures.append(f"source_{name}_not_lightweight") + return failures + + +def evaluate_routing_contract(service: dict[str, Any], edge_config: str) -> list[str]: + failures: list[str] = [] + spec = service.get("spec") or {} + ports = spec.get("ports") or [] + api_port = next((item for item in ports if item.get("name") == "http"), {}) + if service.get("metadata", {}).get("name") != "awoooi-api-svc": + failures.append("source_api_service_name_mismatch") + if spec.get("type") != "NodePort": + failures.append("source_api_service_not_nodeport") + if api_port.get("port") != 8000 or api_port.get("targetPort") != 8000: + failures.append("source_api_service_port_mismatch") + if api_port.get("nodePort") != 32334: + failures.append("source_api_nodeport_mismatch") + if "server 192.168.0.125:32334;" not in edge_config: + failures.append("source_edge_vip_upstream_mismatch") + return failures + + +def evaluate_live_state( + deployment: dict[str, Any], + receipt: dict[str, str], + pods: list[dict[str, Any]], + restart_before: dict[str, int], + restart_after: dict[str, int], + cpu_millicores: dict[str, int], + max_cpu_millicores: int, +) -> list[str]: + failures = evaluate_source_contract(deployment) + status = deployment.get("status") or {} + for name, actual in { + "desired_replicas": status.get("replicas", 0), + "updated_replicas": status.get("updatedReplicas", 0), + "available_replicas": status.get("availableReplicas", 0), + "ready_replicas": status.get("readyReplicas", 0), + }.items(): + if actual != 2: + failures.append(f"live_{name}_not_two:{actual}") + if status.get("unavailableReplicas", 0) not in (0, None): + failures.append(f"live_unavailable_replicas:{status['unavailableReplicas']}") + + expected_receipt = { + "connection_budget_verified": "true", + "api_database_null_pool": "true", + "api_representative_db_preflight_passed": "true", + "api_http_concurrency_probe_passed": "true", + } + for key, expected in expected_receipt.items(): + if receipt.get(key) != expected: + failures.append(f"receipt_{key}_not_verified") + for key, minimum in { + "api_db_connection_limit": 12, + "api_required_connection_budget": 8, + "api_available_connection_headroom": 8, + }.items(): + try: + actual = int(receipt.get(key, "")) + except ValueError: + actual = -1 + if actual < minimum: + failures.append(f"receipt_{key}_below_{minimum}:{actual}") + + ready_pods = 0 + for pod in pods: + statuses = (pod.get("status") or {}).get("containerStatuses") or [] + api = next((item for item in statuses if item.get("name") == "api"), None) + if api and api.get("ready") is True: + ready_pods += 1 + if ready_pods != 2: + failures.append(f"live_ready_pod_count_not_two:{ready_pods}") + + all_names = set(restart_before) | set(restart_after) + restart_delta = sum( + max(0, restart_after.get(name, 0) - restart_before.get(name, 0)) + for name in all_names + ) + if restart_delta: + failures.append(f"live_restart_delta:{restart_delta}") + if set(cpu_millicores) != {pod["metadata"]["name"] for pod in pods}: + failures.append("live_cpu_sample_incomplete") + for pod_name, cpu in cpu_millicores.items(): + if cpu >= max_cpu_millicores: + failures.append(f"live_cpu_saturated:{pod_name}:{cpu}m") + return failures + + +def _kubectl_json(*args: str) -> dict[str, Any]: + completed = subprocess.run( + ["kubectl", *args], + check=True, + capture_output=True, + text=True, + timeout=20, + ) + return json.loads(completed.stdout) + + +def _restart_counts(pods: list[dict[str, Any]]) -> dict[str, int]: + counts: dict[str, int] = {} + for pod in pods: + statuses = (pod.get("status") or {}).get("containerStatuses") or [] + api = next((item for item in statuses if item.get("name") == "api"), None) + counts[pod["metadata"]["name"]] = int((api or {}).get("restartCount", 0)) + return counts + + +def _cpu_samples() -> dict[str, int]: + completed = subprocess.run( + [ + "kubectl", "-n", NAMESPACE, "top", "pod", "-l", f"app={DEPLOYMENT}", + "--containers", "--no-headers", + ], + check=True, + capture_output=True, + text=True, + timeout=20, + ) + samples: dict[str, int] = {} + for line in completed.stdout.splitlines(): + fields = line.split() + if len(fields) >= 3 and fields[1] == "api" and fields[2].endswith("m"): + samples[fields[0]] = int(fields[2][:-1]) + return samples + + +def _http_status(url: str) -> int | None: + try: + with urllib.request.urlopen(url, timeout=8) as response: + return int(response.status) + except urllib.error.HTTPError as exc: + return int(exc.code) + except Exception: + return None + + +def evaluate_edge_receipt( + vantage_host: str | None, + statuses: dict[str, int | None], +) -> list[str]: + if not vantage_host: + return ["edge_production_vantage_unavailable"] + failures: list[str] = [] + for endpoint in EDGE_LIVENESS_URLS: + if statuses.get(endpoint) != 200: + failures.append( + f"edge_{endpoint}_liveness_unavailable:{statuses.get(endpoint)}" + ) + return failures + + +def _edge_status_from_production_vantage() -> tuple[str | None, dict[str, int | None]]: + for host in EDGE_VANTAGE_HOSTS: + statuses: dict[str, int | None] = {} + vantage_reachable = True + for endpoint, url in EDGE_LIVENESS_URLS.items(): + completed = subprocess.run( + [ + "ssh", + "-o", "BatchMode=yes", + "-o", "ConnectTimeout=5", + "-o", "ConnectionAttempts=1", + "-o", "ServerAliveInterval=3", + "-o", "ServerAliveCountMax=1", + f"wooo@{host}", + ( + "curl -sS -o /dev/null -w '%{http_code}' " + f"--connect-timeout 3 --max-time 6 {url}" + ), + ], + check=False, + capture_output=True, + text=True, + timeout=12, + ) + if completed.returncode == 255: + vantage_reachable = False + break + try: + statuses[endpoint] = int(completed.stdout.strip()) + except ValueError: + statuses[endpoint] = None + if vantage_reachable: + return host, statuses + return None, {} + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--live", action="store_true") + parser.add_argument("--sample-seconds", type=int, default=35) + parser.add_argument("--max-pod-cpu-millicores", type=int, default=900) + args = parser.parse_args() + + source = _source_deployment() + failures = evaluate_source_contract(source) + edge_config = (ROOT / "ops" / "nginx" / "awoooi.wooo.work.conf").read_text( + encoding="utf-8" + ) + failures.extend(evaluate_routing_contract(_source_service(), edge_config)) + result: dict[str, Any] = { + "schema_version": "awoooi_api_liveness_stabilization_verifier_v1", + "mode": "live" if args.live else "source", + "rollback": {"target_replicas": 1, "command": ROLLBACK_COMMAND}, + "routing_truth": { + "kubernetes_service": "awoooi-api-svc:8000", + "node_port": 32334, + "edge_vip": "192.168.0.125:32334", + "public_liveness": PUBLIC_LIVENESS_URL, + "edge_probe_vantage": "production_host_120_then_110", + "local_edge_probe_classification": "vantage_unqualified", + "edge_reachability_is_separate_from_replica_rollback": True, + }, + } + try: + if args.live: + deployment_before = _kubectl_json( + "-n", NAMESPACE, "get", "deployment", DEPLOYMENT, "-o", "json" + ) + receipt = _kubectl_json( + "-n", NAMESPACE, "get", "configmap", BOUNDARY_RECEIPT, "-o", "json" + ).get("data") or {} + pods_before = _kubectl_json( + "-n", NAMESPACE, "get", "pods", "-l", f"app={DEPLOYMENT}", "-o", "json" + ).get("items") or [] + restart_before = _restart_counts(pods_before) + time.sleep(max(0, args.sample_seconds)) + deployment_after = _kubectl_json( + "-n", NAMESPACE, "get", "deployment", DEPLOYMENT, "-o", "json" + ) + pods_after = _kubectl_json( + "-n", NAMESPACE, "get", "pods", "-l", f"app={DEPLOYMENT}", "-o", "json" + ).get("items") or [] + failures = evaluate_live_state( + deployment_after, + receipt, + pods_after, + restart_before, + _restart_counts(pods_after), + _cpu_samples(), + args.max_pod_cpu_millicores, + ) + result["sample_seconds"] = max(0, args.sample_seconds) + desired_before = (deployment_before.get("spec") or {}).get("replicas") + desired_after = (deployment_after.get("spec") or {}).get("replicas") + result["desired_replicas_before"] = desired_before + result["desired_replicas_after"] = desired_after + if desired_before != desired_after: + failures.append( + f"live_desired_replica_drift:{desired_before}->{desired_after}" + ) + public_status = _http_status(PUBLIC_LIVENESS_URL) + local_edge_vip_status = _http_status(EDGE_VIP_LIVENESS_URL) + edge_vantage, edge_statuses = _edge_status_from_production_vantage() + result["routing_truth"]["public_status"] = public_status + result["routing_truth"]["local_edge_vip_status"] = local_edge_vip_status + result["routing_truth"]["qualified_vantage_host"] = edge_vantage + result["routing_truth"]["qualified_edge_statuses"] = edge_statuses + if public_status != 200: + failures.append(f"public_liveness_unavailable:{public_status}") + failures.extend(evaluate_edge_receipt(edge_vantage, edge_statuses)) + except Exception as exc: # fail closed without leaking command output + failures.append(f"verifier_error:{type(exc).__name__}") + + result["status"] = "verified" if not failures else "blocked_with_safe_rollback" + result["failures"] = failures + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 if not failures else 1 + + +if __name__ == "__main__": + raise SystemExit(main())