Coverage for scripts / live_release_validation / checks / platform_workloads.py: 100.00%
156 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Platform workload hosting checks for the live cluster.
3The ``platform-workloads`` action proves, on every deployed Region, that the
4``gco-system`` services are hosted the way the manifests promise
5(``lambda/kubectl-applier-simple/manifests/README.md``, "Platform Workload
6Contract") and that the run left them healthy:
8* every platform Deployment has converged at its current generation with all
9 replicas updated, available, and ready, and no live container has restarted
10 since its pod started — a crash loop or an OOM kill during the run is a
11 finding, not noise;
12* every multi-replica service carries a PodDisruptionBudget with
13 ``maxUnavailable: 1`` that currently allows a voluntary disruption, so a node
14 drain can proceed without emptying the service;
15* the shipped autoscalers exist exactly when cdk.json says so — the
16 inference-proxy HPA always, the manifest-processor HPA only when
17 ``manifest_processor.autoscaling.enabled`` — target the right Deployment,
18 and can read their metrics; and
19* the Auto Mode network-policy switch (``kube-system/amazon-vpc-cni``) carries
20 the value rendered from ``eks_cluster.network_policy_enforcement``.
22States that heal by waiting (a rollout in flight, a budget still counting
23healthy pods, an HPA awaiting its first metric sample) are polled for a bounded
24time. Anything that cannot heal — a missing object, a restarted container, a
25wrong budget or autoscaler shape, a wrong switch value — fails immediately with
26the snapshot as evidence.
27"""
29from __future__ import annotations
31import time
32from typing import Any
34from ..models import RunContext
35from .cluster import KubectlRunner, kubectl_json
36from .opencost import _cost_monitoring_configured
38PLATFORM_NAMESPACE = "gco-system"
39#: Deployments every topology ships, in manifest order (30-33).
40PLATFORM_DEPLOYMENTS: tuple[str, ...] = (
41 "health-monitor",
42 "manifest-processor",
43 "inference-monitor",
44 "inference-proxy",
45)
46#: Present only when cdk.json ``cost_monitoring`` (and ``cluster_observability``,
47#: its data source) is enabled (34).
48COST_MONITOR_DEPLOYMENT = "cost-monitor"
49#: The multi-replica services and their budgets; ``<name>-pdb`` in the manifests.
50DISRUPTION_BUDGETS: tuple[str, ...] = tuple(f"{name}-pdb" for name in PLATFORM_DEPLOYMENTS)
51#: ``(HPA name, Deployment it must target)``.
52INFERENCE_PROXY_AUTOSCALER = ("inference-proxy-hpa", "inference-proxy")
53MANIFEST_PROCESSOR_AUTOSCALER = ("manifest-processor-hpa", "manifest-processor")
54ENFORCEMENT_SWITCH_NAMESPACE = "kube-system"
55ENFORCEMENT_SWITCH_NAME = "amazon-vpc-cni"
56#: Both spellings the manifest renders (see 06-network-policy-controller.yaml).
57ENFORCEMENT_SWITCH_KEYS: tuple[str, ...] = (
58 "enable-network-policy-controller",
59 "enable-network-policy",
60)
61#: Deployments were rolled out long before this action runs; ten minutes covers
62#: a node consolidation that happens to be rescheduling a pod as we look.
63_CONVERGENCE_TIMEOUT_SECONDS = 600
66class PlatformWorkloadValidationError(RuntimeError):
67 """The platform services are not hosted as the manifests promise."""
70def expected_deployments(ctx: RunContext) -> tuple[str, ...]:
71 """Return the gco-system Deployments this cdk.json deploys."""
72 if _cost_monitoring_configured(ctx):
73 return (*PLATFORM_DEPLOYMENTS, COST_MONITOR_DEPLOYMENT)
74 return PLATFORM_DEPLOYMENTS
77def manifest_processor_autoscaling_enabled(ctx: RunContext) -> bool:
78 """Return whether cdk.json opts the manifest processor into its HPA (default off)."""
79 block = ctx.cdk_context.get("manifest_processor")
80 autoscaling = block.get("autoscaling") if isinstance(block, dict) else None
81 if isinstance(autoscaling, dict) and "enabled" in autoscaling:
82 return bool(autoscaling["enabled"])
83 return False
86def network_policy_enforcement_enabled(ctx: RunContext) -> bool:
87 """Return whether cdk.json keeps the Auto Mode policy controller on (default on)."""
88 block = ctx.cdk_context.get("eks_cluster")
89 if isinstance(block, dict) and "network_policy_enforcement" in block:
90 return bool(block["network_policy_enforcement"])
91 return True
94def _int(value: Any) -> int:
95 return int(value or 0)
98def _dict(value: Any) -> dict[str, Any]:
99 return value if isinstance(value, dict) else {}
102def _list(value: Any) -> list[Any]:
103 return value if isinstance(value, list) else []
106def _deployment_snapshot(
107 kubectl: KubectlRunner,
108 record: dict[str, Any],
109 name: str,
110 *,
111 timeout: float,
112) -> dict[str, Any]:
113 deployment = kubectl_json(
114 kubectl,
115 record,
116 "get",
117 "deployment",
118 name,
119 "--namespace",
120 PLATFORM_NAMESPACE,
121 timeout=timeout,
122 )
123 if deployment is None:
124 return {"exists": False}
125 metadata = _dict(deployment.get("metadata"))
126 spec = _dict(deployment.get("spec"))
127 status = _dict(deployment.get("status"))
128 desired = _int(spec.get("replicas"))
129 ready = _int(status.get("readyReplicas"))
130 available = _int(status.get("availableReplicas"))
131 updated = _int(status.get("updatedReplicas"))
132 generation = _int(metadata.get("generation"))
133 observed = _int(status.get("observedGeneration"))
135 pods_payload = kubectl_json(
136 kubectl,
137 record,
138 "get",
139 "pods",
140 "--namespace",
141 PLATFORM_NAMESPACE,
142 "--selector",
143 f"app={name}",
144 timeout=timeout,
145 )
146 pods: list[dict[str, Any]] = []
147 for item in _list(_dict(pods_payload).get("items")):
148 pod_metadata = _dict(_dict(item).get("metadata"))
149 if pod_metadata.get("deletionTimestamp"):
150 # A pod on its way out (scale-down, drain, rollout) is no longer
151 # part of the service; its readiness and restarts are history.
152 continue
153 pod_status = _dict(_dict(item).get("status"))
154 containers = [_dict(entry) for entry in _list(pod_status.get("containerStatuses"))]
155 init_containers = [_dict(entry) for entry in _list(pod_status.get("initContainerStatuses"))]
156 pods.append(
157 {
158 "name": pod_metadata.get("name"),
159 "phase": pod_status.get("phase"),
160 "ready": (
161 pod_status.get("phase") == "Running"
162 and bool(containers)
163 and all(entry.get("ready") is True for entry in containers)
164 ),
165 "restarts": sum(
166 _int(entry.get("restartCount")) for entry in (*init_containers, *containers)
167 ),
168 }
169 )
170 converged = (
171 desired >= 1
172 and observed >= generation
173 and ready == desired
174 and available == desired
175 and updated == desired
176 and len(pods) == desired
177 and all(pod["ready"] for pod in pods)
178 )
179 return {
180 "exists": True,
181 "desired": desired,
182 "ready": ready,
183 "available": available,
184 "updated": updated,
185 "generation": generation,
186 "observed_generation": observed,
187 "pods": pods,
188 "restarts": sum(pod["restarts"] for pod in pods),
189 "failed_pods": [pod["name"] for pod in pods if pod["phase"] == "Failed"],
190 "converged": converged,
191 }
194def _budget_snapshot(
195 kubectl: KubectlRunner,
196 record: dict[str, Any],
197 name: str,
198 *,
199 timeout: float,
200) -> dict[str, Any]:
201 budget = kubectl_json(
202 kubectl,
203 record,
204 "get",
205 "poddisruptionbudget",
206 name,
207 "--namespace",
208 PLATFORM_NAMESPACE,
209 timeout=timeout,
210 )
211 if budget is None:
212 return {"exists": False}
213 spec = _dict(budget.get("spec"))
214 status = _dict(budget.get("status"))
215 return {
216 "exists": True,
217 "max_unavailable": spec.get("maxUnavailable"),
218 "disruptions_allowed": _int(status.get("disruptionsAllowed")),
219 "current_healthy": _int(status.get("currentHealthy")),
220 "desired_healthy": _int(status.get("desiredHealthy")),
221 "expected_pods": _int(status.get("expectedPods")),
222 }
225def _autoscaler_snapshot(
226 kubectl: KubectlRunner,
227 record: dict[str, Any],
228 name: str,
229 *,
230 timeout: float,
231) -> dict[str, Any]:
232 autoscaler = kubectl_json(
233 kubectl,
234 record,
235 "get",
236 "horizontalpodautoscaler",
237 name,
238 "--namespace",
239 PLATFORM_NAMESPACE,
240 timeout=timeout,
241 )
242 if autoscaler is None:
243 return {"exists": False}
244 spec = _dict(autoscaler.get("spec"))
245 status = _dict(autoscaler.get("status"))
246 conditions = {
247 _dict(entry).get("type"): _dict(entry).get("status")
248 for entry in _list(status.get("conditions"))
249 }
250 return {
251 "exists": True,
252 "target": _dict(spec.get("scaleTargetRef")).get("name"),
253 "min_replicas": spec.get("minReplicas"),
254 "max_replicas": spec.get("maxReplicas"),
255 "current_replicas": _int(status.get("currentReplicas")),
256 "desired_replicas": _int(status.get("desiredReplicas")),
257 "able_to_scale": conditions.get("AbleToScale") == "True",
258 "scaling_active": conditions.get("ScalingActive") == "True",
259 }
262def _enforcement_switch_snapshot(
263 kubectl: KubectlRunner,
264 record: dict[str, Any],
265 *,
266 timeout: float,
267) -> dict[str, Any]:
268 configmap = kubectl_json(
269 kubectl,
270 record,
271 "get",
272 "configmap",
273 ENFORCEMENT_SWITCH_NAME,
274 "--namespace",
275 ENFORCEMENT_SWITCH_NAMESPACE,
276 timeout=timeout,
277 )
278 if configmap is None:
279 return {"exists": False, "values": {}}
280 data = _dict(configmap.get("data"))
281 return {"exists": True, "values": {key: data.get(key) for key in ENFORCEMENT_SWITCH_KEYS}}
284def _snapshot(ctx: RunContext, kubectl: KubectlRunner, record: dict[str, Any]) -> dict[str, Any]:
285 timeout = float(ctx.settings.command_timeout_seconds)
286 deployments = {
287 name: _deployment_snapshot(kubectl, record, name, timeout=timeout)
288 for name in expected_deployments(ctx)
289 }
290 budgets = {
291 name: _budget_snapshot(kubectl, record, name, timeout=timeout)
292 for name in DISRUPTION_BUDGETS
293 }
294 autoscalers = {
295 name: _autoscaler_snapshot(kubectl, record, name, timeout=timeout)
296 for name, _target in (INFERENCE_PROXY_AUTOSCALER, MANIFEST_PROCESSOR_AUTOSCALER)
297 }
298 return {
299 "deployments": deployments,
300 "budgets": budgets,
301 "autoscalers": autoscalers,
302 "manifest_processor_autoscaling": manifest_processor_autoscaling_enabled(ctx),
303 "network_policy_enforcement": network_policy_enforcement_enabled(ctx),
304 "enforcement_switch": _enforcement_switch_snapshot(kubectl, record, timeout=timeout),
305 }
308def _violations(snapshot: dict[str, Any]) -> list[str]:
309 """Return contract breaches that waiting cannot heal."""
310 problems: list[str] = []
311 for name, deployment in snapshot["deployments"].items():
312 if not deployment["exists"]:
313 problems.append(f"Deployment {name} is missing")
314 continue
315 if deployment["restarts"]:
316 problems.append(
317 f"Deployment {name} has {deployment['restarts']} container restart(s) "
318 "across its live pods"
319 )
320 if deployment["failed_pods"]:
321 problems.append(f"Deployment {name} has failed pod(s): {deployment['failed_pods']}")
322 for name, budget in snapshot["budgets"].items():
323 if not budget["exists"]:
324 problems.append(f"PodDisruptionBudget {name} is missing")
325 elif budget["max_unavailable"] not in (1, "1"):
326 problems.append(
327 f"PodDisruptionBudget {name} has maxUnavailable {budget['max_unavailable']!r}, "
328 "expected 1"
329 )
330 expected_autoscalers = {INFERENCE_PROXY_AUTOSCALER[0]: INFERENCE_PROXY_AUTOSCALER[1]}
331 if snapshot["manifest_processor_autoscaling"]:
332 expected_autoscalers[MANIFEST_PROCESSOR_AUTOSCALER[0]] = MANIFEST_PROCESSOR_AUTOSCALER[1]
333 for name, autoscaler in snapshot["autoscalers"].items():
334 target = expected_autoscalers.get(name)
335 if target is None:
336 if autoscaler["exists"]:
337 problems.append(
338 f"HorizontalPodAutoscaler {name} exists although cdk.json leaves "
339 "manifest_processor.autoscaling disabled"
340 )
341 elif not autoscaler["exists"]:
342 problems.append(f"HorizontalPodAutoscaler {name} is missing")
343 elif autoscaler["target"] != target:
344 problems.append(
345 f"HorizontalPodAutoscaler {name} targets {autoscaler['target']!r}, "
346 f"expected {target!r}"
347 )
348 switch = snapshot["enforcement_switch"]
349 expected_value = "true" if snapshot["network_policy_enforcement"] else "false"
350 if not switch["exists"]:
351 problems.append(
352 f"ConfigMap {ENFORCEMENT_SWITCH_NAMESPACE}/{ENFORCEMENT_SWITCH_NAME} is missing"
353 )
354 else:
355 for key in ENFORCEMENT_SWITCH_KEYS:
356 if switch["values"].get(key) != expected_value:
357 problems.append(
358 f"ConfigMap {ENFORCEMENT_SWITCH_NAME} key {key} is "
359 f"{switch['values'].get(key)!r}, expected {expected_value!r}"
360 )
361 return problems
364def _pending(snapshot: dict[str, Any]) -> list[str]:
365 """Return conditions that a bounded wait may still satisfy."""
366 waiting: list[str] = []
367 for name, deployment in snapshot["deployments"].items():
368 if deployment["exists"] and not deployment["converged"]:
369 waiting.append(
370 f"Deployment {name} not converged "
371 f"(desired={deployment['desired']} ready={deployment['ready']} "
372 f"available={deployment['available']} updated={deployment['updated']} "
373 f"live_pods={len(deployment['pods'])} "
374 f"generation={deployment['generation']}/{deployment['observed_generation']})"
375 )
376 for name, budget in snapshot["budgets"].items():
377 if budget["exists"] and budget["disruptions_allowed"] < 1:
378 waiting.append(
379 f"PodDisruptionBudget {name} allows no disruption "
380 f"(healthy={budget['current_healthy']} desired={budget['desired_healthy']})"
381 )
382 for name, autoscaler in snapshot["autoscalers"].items():
383 if autoscaler["exists"] and not (
384 autoscaler["able_to_scale"] and autoscaler["scaling_active"]
385 ):
386 waiting.append(
387 f"HorizontalPodAutoscaler {name} not active "
388 f"(AbleToScale={autoscaler['able_to_scale']} "
389 f"ScalingActive={autoscaler['scaling_active']})"
390 )
391 return waiting
394def verify_platform_workloads(
395 ctx: RunContext,
396 region: str,
397 kubectl: KubectlRunner,
398) -> dict[str, Any]:
399 """Poll one Region until every platform workload meets the hosting contract.
401 Returns the converged snapshot. Raises ``PlatformWorkloadValidationError``
402 on the first snapshot that breaches the contract, or when the transient
403 conditions are still pending at the deadline; every observation is
404 checkpointed under ``platform_workloads.<region>`` first.
405 """
406 with ctx.state_lock:
407 record = ctx.checkpoint.state.setdefault("platform_workloads", {}).setdefault(region, {})
408 deadline = time.monotonic() + _CONVERGENCE_TIMEOUT_SECONDS
409 while True:
410 snapshot = _snapshot(ctx, kubectl, record)
411 snapshot["violations"] = _violations(snapshot)
412 snapshot["pending"] = _pending(snapshot)
413 with ctx.state_lock:
414 record["last_snapshot"] = snapshot
415 record["observations"] = _int(record.get("observations")) + 1
416 ctx.persist()
417 if snapshot["violations"]:
418 raise PlatformWorkloadValidationError(
419 f"platform workloads in {region} breach the hosting contract: "
420 + "; ".join(snapshot["violations"])
421 )
422 if not snapshot["pending"]:
423 return snapshot
424 if time.monotonic() >= deadline:
425 raise PlatformWorkloadValidationError(
426 f"platform workloads in {region} did not converge within "
427 f"{_CONVERGENCE_TIMEOUT_SECONDS}s: " + "; ".join(snapshot["pending"])
428 )
429 time.sleep(ctx.settings.poll_interval_seconds)