Coverage for scripts / live_release_validation / actions / topology.py: 100.00%
101 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""topology: verify stacks, EKS, HTTPS ALB targets, APIs, queues, and DynamoDB."""
3from __future__ import annotations
5import json
6from typing import Any
8from ..checks.alb_tls import _alb_https_target_evidence
9from ..checks.topology import (
10 _bounded_topology_evidence,
11 _converge_region_addons,
12 _health_stability_samples,
13 _health_warmup_samples,
14 _metrics_reachability_samples,
15 _queue_counts,
16)
17from ..constants import (
18 _HEALTHY_STACK_STATUSES,
19)
20from ..context import (
21 _direct_regional_access_enabled,
22)
23from ..inventory import (
24 describe_stack,
25)
26from ..models import RunContext, utc_now
27from ..ownership.stacks import (
28 _reconcile_stack_ownership,
29 _record_stack_identity,
30)
33def _expected_kubernetes_version(ctx: RunContext) -> str:
34 """The ``kubernetes_version`` the run deployed with, from cdk.json's context.
36 The runner loads ``cdk.json`` once (``RunContext.cdk_context``); a context
37 without the key, or with a non-string value, is a broken checkout rather
38 than something to default, so it fails the action instead of letting the
39 cluster-version comparison pass vacuously.
40 """
41 version = ctx.cdk_context.get("kubernetes_version")
42 if not isinstance(version, str) or not version.strip():
43 raise RuntimeError("cdk.json context.kubernetes_version is missing or not a string")
44 return version.strip()
47def action_topology(ctx: RunContext) -> dict[str, Any]:
48 """Verify deterministic add-on convergence before stable API and data-plane health."""
49 _reconcile_stack_ownership(ctx)
50 stack_details: dict[str, Any] = {}
51 target_stack_regions = ctx.checkpoint.state["target_stack_regions"]
52 for stack_name, region in target_stack_regions.items():
53 stack = describe_stack(ctx.session, str(region), stack_name)
54 if stack is None:
55 raise RuntimeError(f"Expected deployed stack is absent: {stack_name} ({region})")
56 _record_stack_identity(ctx, stack_name, str(region), stack)
57 if stack["status"] not in _HEALTHY_STACK_STATUSES:
58 raise RuntimeError(
59 f"Stack {stack_name} is {stack['status']}, expected one of "
60 f"{sorted(_HEALTHY_STACK_STATUSES)}"
61 )
62 stack_details[stack_name] = stack
64 convergence: dict[str, Any] = {
65 "started_at": utc_now(),
66 "status": "running",
67 "regions": {},
68 }
69 ctx.checkpoint.state["topology_convergence"] = convergence
70 ctx.persist()
71 for region in ctx.deployment_regions:
72 stack_name = f"{ctx.config.project_name}-{region}"
73 evidence: dict[str, Any] = {
74 "region": region,
75 "stack_name": stack_name,
76 "result": "pending",
77 "observations": [],
78 }
79 convergence["regions"][region] = evidence
80 try:
81 if target_stack_regions.get(stack_name) != region:
82 raise RuntimeError(
83 f"Checkpoint does not bind exact regional stack {stack_name} to {region}"
84 )
85 stack = stack_details.get(stack_name)
86 if not isinstance(stack, dict):
87 raise RuntimeError(f"Exact regional stack was not described: {region}:{stack_name}")
88 _converge_region_addons(
89 ctx,
90 region=region,
91 stack_name=stack_name,
92 stack=stack,
93 evidence=evidence,
94 )
95 except Exception as exc:
96 evidence["result"] = "failed"
97 evidence["completed_at"] = utc_now()
98 evidence["error"] = _bounded_topology_evidence(f"{type(exc).__name__}: {exc}")
99 convergence["status"] = "failed"
100 convergence["completed_at"] = utc_now()
101 ctx.persist()
102 raise
103 evidence["result"] = "succeeded"
104 evidence["completed_at"] = utc_now()
105 ctx.persist()
106 convergence["status"] = "succeeded"
107 convergence["completed_at"] = utc_now()
108 ctx.persist()
110 clusters: dict[str, Any] = {}
111 expected_version = _expected_kubernetes_version(ctx)
112 for region in ctx.deployment_regions:
113 name = f"{ctx.config.project_name}-{region}"
114 cluster = ctx.session.client("eks", region_name=region).describe_cluster(name=name)[
115 "cluster"
116 ]
117 if cluster.get("status") != "ACTIVE":
118 raise RuntimeError(f"EKS cluster {name} is not ACTIVE: {cluster.get('status')}")
119 # The cluster minor is the one thing a Kubernetes upgrade PR changes
120 # that nothing else in this run would notice: every add-on, probe and
121 # Job below passes just as well on the previous minor. Compare it with
122 # cdk.json so the run proves the version it deployed, not just that a
123 # cluster came up.
124 if cluster.get("version") != expected_version:
125 raise RuntimeError(
126 f"EKS cluster {name} runs Kubernetes {cluster.get('version')!r}; "
127 f"cdk.json kubernetes_version is {expected_version!r}"
128 )
129 clusters[region] = {
130 "name": name,
131 "arn": cluster.get("arn"),
132 "status": cluster.get("status"),
133 "version": cluster.get("version"),
134 "endpoint_public_access": (cluster.get("resourcesVpcConfig") or {}).get(
135 "endpointPublicAccess"
136 ),
137 "endpoint_private_access": (cluster.get("resourcesVpcConfig") or {}).get(
138 "endpointPrivateAccess"
139 ),
140 }
142 alb_https_targets = {
143 region: _alb_https_target_evidence(
144 ctx,
145 region=region,
146 cluster_name=f"{ctx.config.project_name}-{region}",
147 )
148 for region in ctx.deployment_regions
149 }
151 global_endpoint = ctx.aws_client.get_api_endpoint(force_refresh=True)
152 global_url = str(getattr(global_endpoint, "url", "") or "")
153 if not global_url:
154 raise RuntimeError("Global API endpoint has no URL")
155 direct_regional_access = _direct_regional_access_enabled(ctx)
156 regional_urls: dict[str, str] = {}
157 if direct_regional_access:
158 for region in ctx.deployment_regions:
159 endpoint = ctx.aws_client.get_regional_api_endpoint(region, force_refresh=True)
160 endpoint_url = str(getattr(endpoint, "url", "") or "")
161 if not endpoint_url:
162 raise RuntimeError(f"Direct regional API endpoint is absent in {region}")
163 regional_urls[region] = endpoint_url
165 health_warmup_samples = _health_warmup_samples(
166 ctx,
167 global_url=global_url,
168 regional_urls=regional_urls,
169 )
170 health_samples = _health_stability_samples(
171 ctx,
172 global_url=global_url,
173 regional_urls=regional_urls,
174 )
175 metrics_samples = _metrics_reachability_samples(
176 ctx,
177 global_url=global_url,
178 regional_urls=regional_urls,
179 )
180 global_samples = [sample for sample in health_samples if sample["scope"] == "global"]
181 global_api = {
182 "url": global_url,
183 "health": global_samples[-1]["payload"],
184 "samples": global_samples,
185 }
186 if direct_regional_access:
187 regional_endpoints = {
188 region: {
189 "url": regional_urls[region],
190 "health": next(
191 sample["payload"]
192 for sample in reversed(health_samples)
193 if sample["region"] == region
194 ),
195 "samples": [sample for sample in health_samples if sample["region"] == region],
196 }
197 for region in ctx.deployment_regions
198 }
199 else:
200 regional_endpoints = {
201 region: {
202 "skipped": True,
203 "reason": "direct caller access is disabled by cdk.json",
204 "samples": [],
205 }
206 for region in ctx.deployment_regions
207 }
209 queue_baseline: dict[str, Any] = {}
210 for region in ctx.deployment_regions:
211 status = ctx.job_manager.get_queue_status(region)
212 counts = _queue_counts(status)
213 if any(counts.values()):
214 raise RuntimeError(
215 f"Fresh queue in {region} is not empty: {json.dumps(counts, sort_keys=True)}"
216 )
217 queue_baseline[region] = status
219 table_name = f"{ctx.config.project_name}-jobs"
220 table = ctx.session.client("dynamodb", region_name=ctx.config.global_region).describe_table(
221 TableName=table_name
222 )["Table"]
223 if table.get("TableStatus") != "ACTIVE":
224 raise RuntimeError(f"DynamoDB table {table_name} is not ACTIVE")
226 ctx.checkpoint.state["queue_baseline"] = queue_baseline
227 ctx.persist()
228 return {
229 "stacks": stack_details,
230 "clusters": clusters,
231 "convergence": convergence,
232 "alb_https_targets": alb_https_targets,
233 "health_warmup_samples": health_warmup_samples,
234 "health_samples": health_samples,
235 "metrics_samples": metrics_samples,
236 "global_api": global_api,
237 "regional_apis": regional_endpoints,
238 "queue_baseline": queue_baseline,
239 "jobs_table": {
240 "name": table_name,
241 "arn": table.get("TableArn"),
242 "status": table.get("TableStatus"),
243 },
244 }