Coverage for lambda / kubectl-applier-simple / handler.py: 100.00%
1240 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""
2Lambda handler for applying Kubernetes manifests to EKS clusters.
4This Lambda function is triggered by CloudFormation Custom Resources during
5stack deployment. It applies Kubernetes manifests (namespaces, deployments,
6services, RBAC, Karpenter NodePools, etc.) to the EKS cluster.
8Key Features:
9- Pure Python implementation (no Docker/kubectl binary required)
10- Generates EKS authentication tokens using STS presigned URLs
11- Supports create/update operations with idempotent behavior
12- Handles placeholder replacement for dynamic values (image URIs, etc.)
13- Two-pass deployment: main pass then post-Helm pass for CRD-dependent resources
15Manifest Naming Convention:
16 NN-name.yaml Applied in the main pass (before Helm)
17 post-helm-*.yaml Applied in the post-Helm pass (after Helm installs CRDs)
19 Files with unreplaced {{PLACEHOLDER}} values are automatically skipped,
20 enabling optional features (FSx, Valkey, queue processor).
22Environment Variables:
23 CLUSTER_NAME: Name of the EKS cluster
24 REGION: AWS region where the cluster is deployed
26CloudFormation Properties:
27 ClusterName: EKS cluster name
28 Region: AWS region
29 ImageReplacements: Dict of placeholder -> value mappings
30 SkipDeletionOnStackDelete: If "true", don't delete resources on stack deletion
31 PostHelm: "true" to apply only post-helm-* manifests (after Helm installs CRDs)
32"""
34import base64
35import copy
36import json
37import logging
38import os
39import re
40import time
41from datetime import UTC
42from typing import Any
44import boto3
45import urllib3
46import yaml
47from kubernetes import client, dynamic
48from kubernetes.client.rest import ApiException
49from kubernetes.dynamic.exceptions import NotFoundError, ResourceNotFoundError
51# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
52# Generated at (UTC): 2026-09-12T12:46:59Z
53# Generated from Git commit: d77e920379da4b3fd56a9cf0b58cbd1fc45c1802
54# Flowchart(s) generated from this file:
55# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/kubectl-applier-simple/handler.lambda_handler.html``
56# (PNG: ``diagrams/code_diagrams/lambda/kubectl-applier-simple/handler.lambda_handler.png``)
57# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
58# <pyflowchart-code-diagram> END
61# Configure logging for CloudWatch
62# In Lambda, the root logger is already configured, so we need to set the level explicitly
63logger = logging.getLogger()
64logger.setLevel(logging.INFO)
66# Lazy-initialized AWS clients
67_eks_client = None
69# CloudFormation response status constants
70SUCCESS = "SUCCESS"
71FAILED = "FAILED"
73# Feature-gate placeholders follow an UPPER_SNAKE token convention
74# ({{CLUSTER_OBSERVABILITY_ENABLED}}, {{FSX_FILE_SYSTEM_ID}}, {{VALKEY_ENDPOINT}},
75# ...). A manifest that still contains one *after* substitution belongs to a
76# feature that is turned off, so the file is skipped. The character class is
77# deliberately restricted to A-Z/0-9/_ so the check never matches lower- or
78# mixed-case double-brace tokens that are legitimate *content* in an applied
79# manifest — e.g. Grafana dashboard legend fields ({{gpu}}, {{service}},
80# {{Hostname}}) in the observability dashboard ConfigMaps, which must survive
81# substitution untouched and be applied verbatim.
82_UNRESOLVED_PLACEHOLDER_RE = re.compile(r"\{\{[A-Z0-9_]+\}\}")
83_HPA_REPLICA_OWNERSHIP_ANNOTATION = "gco.aws/hpa-controls-replicas"
84_POST_HELM_PREFIX = "post-helm-"
85_CLUSTER_SCOPE = "<cluster>"
86_MAX_PLANNING_FAILURES = 20
87_MAX_VALIDATION_FAILURES = 20
88_GATEWAY_DELETE_WAIT_SECONDS = 270
89_GATEWAY_DELETE_POLL_SECONDS = 5
91# Gateway API resources are installed after the pinned CRD bootstrap. Keep the
92# exact group/version/plural/scope mapping in one place so apply and teardown
93# cannot drift onto different objects.
94_GATEWAY_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
95 "GatewayClass": ("gateway.networking.k8s.io", "v1", "gatewayclasses", True),
96 "Gateway": ("gateway.networking.k8s.io", "v1", "gateways", False),
97 "HTTPRoute": ("gateway.networking.k8s.io", "v1", "httproutes", False),
98 "LoadBalancerConfiguration": (
99 "gateway.k8s.aws",
100 "v1",
101 "loadbalancerconfigurations",
102 False,
103 ),
104 "TargetGroupConfiguration": (
105 "gateway.k8s.aws",
106 "v1",
107 "targetgroupconfigurations",
108 False,
109 ),
110}
112# Kueue queue-topology resources applied by post-helm-kueue-default-queues.yaml.
113# Kept separate from _GATEWAY_CUSTOM_OBJECTS because gateway kinds get the
114# ALB-finalizer teardown treatment while these are ordinary CRs; the same
115# group/version/plural/scope discipline applies so apply and pruning cannot
116# drift onto different objects. tests/test_kubectl_applier.py pins this map
117# against the manifests directory exactly like the gateway map.
118_QUEUEING_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
119 "ResourceFlavor": ("kueue.x-k8s.io", "v1beta1", "resourceflavors", True),
120 "ClusterQueue": ("kueue.x-k8s.io", "v1beta1", "clusterqueues", True),
121 "LocalQueue": ("kueue.x-k8s.io", "v1beta1", "localqueues", False),
122}
124# cert-manager resources that issue the TLS leaves mounted by ALB-facing API
125# workloads. They are ordinary namespaced CRs and are applied before the
126# Gateway resources in the post-Helm phase.
127_CERT_MANAGER_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
128 "Issuer": ("cert-manager.io", "v1", "issuers", False),
129 "Certificate": ("cert-manager.io", "v1", "certificates", False),
130}
132# Services annotated with this marker are validated for exact existence only;
133# a ready EndpointSlice endpoint is not required. Reserved for Services whose
134# backends schedule exclusively onto accelerator nodes that a fresh cluster
135# does not have yet (for example the DCGM exporter DaemonSet).
136_ALLOW_EMPTY_ENDPOINTS_ANNOTATION = "gco.io/allow-empty-endpoints"
138# This is the authoritative set of kinds the applier knows how to create or
139# patch. Planning rejects anything else before the first Kubernetes mutation,
140# which prevents a newly added raw manifest from being silently ignored.
141_SUPPORTED_MANIFEST_KINDS = frozenset(
142 {
143 "APIService",
144 "Certificate",
145 "ClusterRole",
146 "ClusterRoleBinding",
147 "ClusterTrainingRuntime",
148 "ConfigMap",
149 "CronJob",
150 "CustomResourceDefinition",
151 "DaemonSet",
152 "Deployment",
153 "DeviceClass",
154 "EC2NodeClass",
155 "Gateway",
156 "GatewayClass",
157 "HTTPRoute",
158 "HorizontalPodAutoscaler",
159 "Issuer",
160 "Job",
161 "Lease",
162 "LimitRange",
163 "LoadBalancerConfiguration",
164 "Namespace",
165 "NetworkPolicy",
166 "NodePool",
167 "PersistentVolume",
168 "PersistentVolumeClaim",
169 "Pod",
170 "ClusterQueue",
171 "LocalQueue",
172 "PodDisruptionBudget",
173 "PodMonitor",
174 "PriorityClass",
175 "ResourceFlavor",
176 "ResourceQuota",
177 "Role",
178 "RoleBinding",
179 "ScaledJob",
180 "ScaledObject",
181 "Secret",
182 "Service",
183 "ServiceAccount",
184 "ServiceMonitor",
185 "StatefulSet",
186 "StorageClass",
187 "TargetGroupConfiguration",
188 }
189)
190_CLUSTER_SCOPED_KINDS = frozenset(
191 {
192 "APIService",
193 "ClusterQueue",
194 "ClusterRole",
195 "ClusterRoleBinding",
196 "ClusterTrainingRuntime",
197 "CustomResourceDefinition",
198 "DeviceClass",
199 "EC2NodeClass",
200 "GatewayClass",
201 "Namespace",
202 "NodePool",
203 "PersistentVolume",
204 "PriorityClass",
205 "ResourceFlavor",
206 "StorageClass",
207 }
208)
209_IDENTITY_FIELDS = ("apiVersion", "kind", "namespace", "name", "sourceFile", "phase")
212def _public_manifest_identity(resource: dict[str, Any]) -> dict[str, str]:
213 """Return the serializable, normalized identity for a planned resource."""
214 return {field: resource[field] for field in _IDENTITY_FIELDS}
217def _planning_error_message(errors: list[str], failure_count: int) -> str:
218 """Build a bounded planning failure with enough source context to act on."""
219 hidden = failure_count - len(errors)
220 suffix = f"; ... {hidden} additional error(s)" if hidden else ""
221 return "Manifest planning failed: " + "; ".join(errors) + suffix
224# ---------------------------------------------------------------------------
225# Cross-phase ServiceAccount/token-projection consistency guard.
226#
227# The 2026-08 SQS submission-path outage: hardening a ServiceAccount with
228# ``automountServiceAccountToken: false`` (base phase) and projecting the
229# compensating kubernetes-audience token onto its workload (post-Helm phase)
230# are one logical change split across two apply invocations. A redeploy that
231# ran only the base pass left the live queue-processor pod with no way to
232# authenticate to the Kubernetes API, crash-looping before its first SQS
233# receive. The two files cannot apply in one transaction (the ScaledJob needs
234# KEDA CRDs that do not exist in the base phase), so the planner enforces the
235# pairing instead — and because BOTH phases are always planned before either
236# is applied, a violation fails the base pass too.
238# Standard in-cluster credential mount point the kubernetes client reads.
239_KUBE_SERVICEACCOUNT_MOUNT = "/var/run/secrets/kubernetes.io/serviceaccount"
241# Where each workload kind keeps its pod spec.
242_POD_SPEC_PATHS: dict[str, tuple[str, ...]] = {
243 "Pod": ("spec",),
244 "Deployment": ("spec", "template", "spec"),
245 "StatefulSet": ("spec", "template", "spec"),
246 "DaemonSet": ("spec", "template", "spec"),
247 "Job": ("spec", "template", "spec"),
248 "CronJob": ("spec", "jobTemplate", "spec", "template", "spec"),
249 "ScaledJob": ("spec", "jobTargetRef", "template", "spec"),
250}
253def _planned_pod_spec(document: dict[str, Any]) -> dict[str, Any] | None:
254 """Return the pod spec embedded in a planned workload document, if any."""
255 path = _POD_SPEC_PATHS.get(str(document.get("kind")))
256 if path is None:
257 return None
258 node: Any = document
259 for key in path:
260 if not isinstance(node, dict):
261 return None
262 node = node.get(key)
263 return node if isinstance(node, dict) else None
266def _pod_spec_projects_service_account_token(pod_spec: dict[str, Any]) -> bool:
267 """True when a projected serviceAccountToken is mounted at the standard path."""
268 token_volumes: set[str] = set()
269 for volume in pod_spec.get("volumes") or []:
270 if not isinstance(volume, dict):
271 continue
272 projected = volume.get("projected")
273 if not isinstance(projected, dict):
274 continue
275 sources = projected.get("sources") or []
276 has_token_source = any(
277 isinstance(source, dict) and "serviceAccountToken" in source for source in sources
278 )
279 if has_token_source and isinstance(volume.get("name"), str):
280 token_volumes.add(volume["name"])
281 if not token_volumes:
282 return False
283 for containers_key in ("containers", "initContainers", "ephemeralContainers"):
284 containers = pod_spec.get(containers_key) or []
285 if not isinstance(containers, list):
286 continue
287 for container in containers:
288 if not isinstance(container, dict):
289 continue
290 for mount in container.get("volumeMounts") or []:
291 if (
292 isinstance(mount, dict)
293 and mount.get("mountPath") == _KUBE_SERVICEACCOUNT_MOUNT
294 and mount.get("name") in token_volumes
295 ):
296 return True
297 return False
300def _automount_disabled_service_accounts(
301 planned: list[dict[str, Any]],
302) -> dict[tuple[str, str], str]:
303 """Planned ServiceAccounts with automount disabled, keyed by (namespace, name)."""
304 disabled: dict[tuple[str, str], str] = {}
305 for item in planned:
306 if item["kind"] != "ServiceAccount":
307 continue
308 if item["document"].get("automountServiceAccountToken") is False:
309 disabled[(item["namespace"], item["name"])] = item["sourceFile"]
310 return disabled
313def _rbac_bound_service_accounts(planned: list[dict[str, Any]]) -> set[tuple[str, str]]:
314 """ServiceAccounts referenced as subjects by planned (Cluster)RoleBindings.
316 An RBAC binding is the planned inventory's own declaration that the
317 ServiceAccount is expected to call the Kubernetes API. Accounts with
318 automount disabled and NO binding (workload identities that only hold
319 AWS credentials, e.g. the inference proxy) are deliberately exempt from
320 the token-projection invariant.
321 """
322 bound: set[tuple[str, str]] = set()
323 for item in planned:
324 if item["kind"] not in ("RoleBinding", "ClusterRoleBinding"):
325 continue
326 default_namespace = item["namespace"] if item["kind"] == "RoleBinding" else None
327 for subject in item["document"].get("subjects") or []:
328 if not isinstance(subject, dict) or subject.get("kind") != "ServiceAccount":
329 continue
330 name = subject.get("name")
331 namespace = subject.get("namespace") or default_namespace
332 if isinstance(name, str) and isinstance(namespace, str):
333 bound.add((namespace, name))
334 return bound
337def _service_account_token_projection_errors(
338 phases: dict[str, list[dict[str, Any]]],
339) -> list[str]:
340 """Cross-phase invariant: hardened, API-bound SAs need a projected token.
342 For every planned ServiceAccount (either phase) that sets
343 ``automountServiceAccountToken: false`` AND is bound to the Kubernetes
344 API by a planned RoleBinding/ClusterRoleBinding, every planned workload
345 whose pod spec runs as that account must mount a projected
346 serviceAccountToken at ``/var/run/secrets/kubernetes.io/serviceaccount``.
347 """
348 planned = phases["base"] + phases["post-helm"]
349 disabled = _automount_disabled_service_accounts(planned)
350 if not disabled:
351 return []
352 bound = _rbac_bound_service_accounts(planned)
353 guarded = {identity: source for identity, source in disabled.items() if identity in bound}
354 if not guarded:
355 return []
357 errors: list[str] = []
358 for item in planned:
359 pod_spec = _planned_pod_spec(item["document"])
360 if pod_spec is None:
361 continue
362 service_account = pod_spec.get("serviceAccountName")
363 if not isinstance(service_account, str) or not service_account:
364 continue
365 source_file = guarded.get((item["namespace"], service_account))
366 if source_file is None:
367 continue
368 if _pod_spec_projects_service_account_token(pod_spec):
369 continue
370 errors.append(
371 f"{item['sourceFile']}: {item['kind']}/{item['namespace']}/{item['name']} runs as "
372 f"ServiceAccount {service_account!r} ({source_file}), which sets "
373 "automountServiceAccountToken: false and is RBAC-bound to the Kubernetes API, "
374 f"but mounts no projected serviceAccountToken at {_KUBE_SERVICEACCOUNT_MOUNT} - "
375 "the pod would have no credentials. Project the token (see the "
376 "kubernetes-api-token volume in post-helm-sqs-consumer.yaml) or ship both "
377 "halves of the hardening in one release"
378 )
379 return errors
382def _log_service_account_automount_flip(
383 v1: Any,
384 document: dict[str, Any],
385 namespace: str,
386 name: str,
387 plan: dict[str, Any],
388) -> None:
389 """Name the workloads affected when automount is being flipped to false.
391 Clusters created by an older release are supported: their live
392 ServiceAccount may still automount tokens while the incoming manifest
393 disables it. This apply is the exact moment previously-running pods
394 lose ambient credentials, so log every planned workload that references
395 the account and whether each already carries the compensating projected
396 token — the diagnostic that would have named the queue processor on the
397 redeploy that caused the SQS outage. Best-effort: any read failure
398 skips the diagnostic, never the apply.
399 """
400 if document.get("automountServiceAccountToken") is not False:
401 return
402 try:
403 live = v1.read_namespaced_service_account(name, namespace)
404 except ApiException as e:
405 if e.status != 404:
406 logger.debug(
407 "Automount-flip check could not read live ServiceAccount %s/%s: %s",
408 namespace,
409 name,
410 e,
411 )
412 return
413 except Exception as e: # pragma: no cover - defensive; diagnostics never block
414 logger.debug("Automount-flip check failed for ServiceAccount %s/%s: %s", namespace, name, e)
415 return
416 if getattr(live, "automount_service_account_token", None) is False:
417 return # Already hardened on the live cluster; nothing is flipping.
419 references: list[str] = []
420 for item in plan["phases"]["base"] + plan["phases"]["post-helm"]:
421 pod_spec = _planned_pod_spec(item["document"])
422 if (
423 pod_spec is None
424 or item["namespace"] != namespace
425 or pod_spec.get("serviceAccountName") != name
426 ):
427 continue
428 projected = _pod_spec_projects_service_account_token(pod_spec)
429 references.append(
430 f"{item['phase']}:{item['sourceFile']} {item['kind']}/{item['name']} "
431 f"projected-token={'present' if projected else 'MISSING'}"
432 )
433 logger.warning(
434 "ServiceAccount %s/%s: automountServiceAccountToken is being flipped to false on a "
435 "live cluster; planned workloads running as it: %s",
436 namespace,
437 name,
438 "; ".join(references) or "<none>",
439 )
442def plan_manifests(
443 manifests_dir: str,
444 replacements: dict[str, str],
445) -> dict[str, Any]:
446 """Plan the complete raw-manifest inventory without mutating the cluster.
448 Files are scanned in lexical order. Replacements are literal string
449 replacements, then an unresolved UPPER_SNAKE placeholder gates the entire
450 file out. Every remaining nonempty YAML document must have an exact,
451 supported identity and be unique across both apply phases.
452 """
453 phases: dict[str, list[dict[str, Any]]] = {"base": [], "post-helm": []}
454 skipped: dict[str, list[str]] = {"base": [], "post-helm": []}
455 feature_gates: dict[str, set[str]] = {"base": set(), "post-helm": set()}
456 errors: list[str] = []
457 failure_count = 0
458 seen: dict[tuple[str, str, str, str], str] = {}
460 def add_error(message: str) -> None:
461 nonlocal failure_count
462 failure_count += 1
463 if len(errors) < _MAX_PLANNING_FAILURES:
464 errors.append(message[:500])
466 replacement_items: list[tuple[str, str]] = []
467 if not isinstance(replacements, dict):
468 add_error("ImageReplacements must be a string-to-string mapping")
469 else:
470 for key, value in replacements.items():
471 if not isinstance(key, str) or not key:
472 add_error("ImageReplacements contains an empty or non-string key")
473 continue
474 if not isinstance(value, str):
475 add_error(f"ImageReplacements value for {key!r} is not a string")
476 continue
477 replacement_items.append((key, value))
479 if failure_count:
480 raise ValueError(_planning_error_message(errors, failure_count))
482 for filename in sorted(os.listdir(manifests_dir)):
483 if not filename.endswith((".yaml", ".yml")):
484 continue
486 phase = "post-helm" if filename.startswith(_POST_HELM_PREFIX) else "base"
487 if phase == "post-helm":
488 skipped["base"].append(f"{filename}:deferred-to-post-helm")
490 filepath = os.path.join(manifests_dir, filename)
491 try:
492 with open(filepath, encoding="utf-8") as manifest_file:
493 content = manifest_file.read()
494 except OSError as exc:
495 add_error(f"{filename}: unable to read manifest: {exc}")
496 continue
498 for key, value in replacement_items:
499 content = content.replace(key, value)
501 unresolved = sorted(set(_UNRESOLVED_PLACEHOLDER_RE.findall(content)))
502 if unresolved:
503 skipped[phase].append(f"{filename}:unreplaced-placeholders")
504 feature_gates[phase].update(unresolved)
505 logger.info(
506 "Planning excludes %s - unresolved feature placeholder(s): %s",
507 filename,
508 ", ".join(unresolved),
509 )
510 continue
512 try:
513 documents = list(yaml.safe_load_all(content))
514 except yaml.YAMLError as exc:
515 add_error(f"{filename}: invalid YAML: {exc}")
516 continue
518 for document_index, document in enumerate(documents, start=1):
519 if document is None:
520 continue
521 location = f"{filename} document {document_index}"
522 if not isinstance(document, dict):
523 add_error(f"{location}: document must be a mapping")
524 continue
526 api_version = document.get("apiVersion")
527 kind = document.get("kind")
528 metadata = document.get("metadata")
529 if not isinstance(api_version, str) or not api_version.strip():
530 add_error(f"{location}: apiVersion must be a nonempty string")
531 continue
532 if not isinstance(kind, str) or not kind.strip():
533 add_error(f"{location}: kind must be a nonempty string")
534 continue
535 api_version = api_version.strip()
536 kind = kind.strip()
537 if kind not in _SUPPORTED_MANIFEST_KINDS:
538 add_error(f"{location}: unsupported kind {kind!r}")
539 continue
540 if not isinstance(metadata, dict):
541 add_error(f"{location}: metadata must be a mapping")
542 continue
544 name = metadata.get("name")
545 if not isinstance(name, str) or not name.strip():
546 add_error(f"{location}: metadata.name must be a nonempty string")
547 continue
548 name = name.strip()
550 if kind in _CLUSTER_SCOPED_KINDS:
551 namespace = _CLUSTER_SCOPE
552 else:
553 namespace_value = metadata.get("namespace", "default")
554 if not isinstance(namespace_value, str) or not namespace_value.strip():
555 add_error(
556 f"{location}: metadata.namespace must be a nonempty string when present"
557 )
558 continue
559 namespace = namespace_value.strip()
561 duplicate_key = (api_version, kind, namespace, name)
562 previous = seen.get(duplicate_key)
563 if previous is not None:
564 add_error(
565 f"{location}: duplicate {api_version}/{kind}/{namespace}/{name}; "
566 f"first declared in {previous}"
567 )
568 continue
569 seen[duplicate_key] = location
571 phases[phase].append(
572 {
573 "apiVersion": api_version,
574 "kind": kind,
575 "namespace": namespace,
576 "name": name,
577 "sourceFile": filename,
578 "phase": phase,
579 "document": document,
580 }
581 )
583 # Cross-phase consistency: a hardened ServiceAccount and the token
584 # projection that compensates for it must ship together. Runs over BOTH
585 # planned phases, so the base pass fails before its first Kubernetes
586 # mutation even when the violation lives in a post-Helm file.
587 for message in _service_account_token_projection_errors(phases):
588 add_error(message)
590 if failure_count:
591 raise ValueError(_planning_error_message(errors, failure_count))
593 return {
594 "phases": phases,
595 "skipped": skipped,
596 "featureGates": {
597 phase: sorted(placeholders) for phase, placeholders in feature_gates.items()
598 },
599 }
602def _deployment_patch_body(document: dict[str, Any]) -> dict[str, Any]:
603 """Copy a Deployment update without replicas when its HPA owns scale."""
604 patch_body = copy.deepcopy(document)
605 metadata = patch_body.get("metadata")
606 annotations = metadata.get("annotations") if isinstance(metadata, dict) else None
607 if not isinstance(annotations, dict) or (
608 annotations.get(_HPA_REPLICA_OWNERSHIP_ANNOTATION) != "true"
609 ):
610 return patch_body
612 spec = patch_body.get("spec")
613 if isinstance(spec, dict):
614 spec.pop("replicas", None)
615 return patch_body
618# Exact resources owned by optional features. Keys include the apply phase so
619# disabling one feature cannot delete similarly named or unrelated resources.
620# Tuples are (apiVersion, kind, namespace-or-None, name); order is deliberate
621# for dependent resources such as FSx claims before volumes before the class.
622_FEATURE_RESOURCE_INVENTORY: dict[
623 tuple[str, bool], tuple[tuple[str, str, str | None, str], ...]
624] = {
625 ("{{FSX_FILE_SYSTEM_ID}}", False): (
626 ("v1", "PersistentVolumeClaim", "default", "gco-fsx-storage"),
627 ("v1", "PersistentVolumeClaim", "gco-jobs", "gco-fsx-storage"),
628 ("v1", "PersistentVolumeClaim", "gco-system", "gco-fsx-storage"),
629 ("v1", "PersistentVolume", None, "gco-fsx-pv-default"),
630 ("v1", "PersistentVolume", None, "gco-fsx-pv-jobs"),
631 ("v1", "PersistentVolume", None, "gco-fsx-pv-system"),
632 ("storage.k8s.io/v1", "StorageClass", None, "fsx-sc"),
633 ),
634 ("{{VALKEY_ENDPOINT}}", False): tuple(
635 ("v1", "ConfigMap", namespace, "gco-valkey")
636 for namespace in ("gco-system", "gco-jobs", "gco-inference")
637 ),
638 ("{{AURORA_PGVECTOR_ENDPOINT}}", False): tuple(
639 ("v1", "ConfigMap", namespace, "gco-aurora-pgvector")
640 for namespace in ("gco-system", "gco-jobs", "gco-inference")
641 ),
642 ("{{VECTOR_STORE_TABLE_NAME}}", False): tuple(
643 ("v1", "ConfigMap", namespace, "gco-vector-store")
644 for namespace in ("gco-system", "gco-jobs", "gco-inference")
645 ),
646 ("{{CLUSTER_OBSERVABILITY_ENABLED}}", False): (
647 ("apps/v1", "DaemonSet", "kube-system", "dcgm-exporter"),
648 ("v1", "Service", "kube-system", "dcgm-exporter"),
649 ("v1", "ConfigMap", "kube-system", "dcgm-device-counters"),
650 ("storage.k8s.io/v1", "StorageClass", None, "gco-observability-gp3"),
651 ),
652 ("{{CLUSTER_OBSERVABILITY_ENABLED}}", True): (
653 ("batch/v1", "CronJob", "monitoring", "gco-grafana-admin-password-rotation"),
654 ("v1", "ConfigMap", "monitoring", "gco-dashboard-gpu"),
655 ("v1", "ConfigMap", "monitoring", "gco-dashboard-schedulers"),
656 ("v1", "ConfigMap", "monitoring", "gco-dashboard-keda"),
657 ("v1", "ConfigMap", "monitoring", "gco-dashboard-services"),
658 (
659 "rbac.authorization.k8s.io/v1",
660 "ClusterRoleBinding",
661 None,
662 "gco-prometheus-kueue-metrics",
663 ),
664 *tuple(
665 ("monitoring.coreos.com/v1", "ServiceMonitor", "monitoring", name)
666 for name in (
667 "gco-keda",
668 "gco-volcano",
669 "gco-kueue",
670 "gco-kuberay",
671 "gco-yunikorn",
672 "gco-dcgm-exporter",
673 )
674 ),
675 *tuple(
676 ("monitoring.coreos.com/v1", "PodMonitor", "monitoring", name)
677 for name in (
678 "gco-health-monitor",
679 "gco-manifest-processor",
680 "gco-inference-proxy",
681 "gco-inference-monitor",
682 )
683 ),
684 ("rbac.authorization.k8s.io/v1", "RoleBinding", "monitoring", "gco-grafana-rotator"),
685 ("rbac.authorization.k8s.io/v1", "Role", "monitoring", "gco-grafana-rotator"),
686 ("v1", "ServiceAccount", "monitoring", "gco-grafana-rotator"),
687 ),
688 ("{{QUEUE_PROCESSOR_IMAGE}}", True): (
689 ("keda.sh/v1alpha1", "ScaledJob", "gco-system", "sqs-queue-processor"),
690 ),
691 # Optional manifest-processor CPU autoscaler (cdk.json
692 # manifest_processor.autoscaling.enabled). Turning it off must also remove
693 # the HPA, otherwise the last scale value would keep fighting the
694 # Deployment's re-asserted replicas.
695 ("{{MP_HPA_ENABLED}}", False): (
696 ("autoscaling/v2", "HorizontalPodAutoscaler", "gco-system", "manifest-processor-hpa"),
697 ),
698 ("{{KUEUE_ENABLED}}", True): (
699 # Deletion order matters: the LocalQueue references the ClusterQueue,
700 # which references the ResourceFlavor.
701 ("kueue.x-k8s.io/v1beta1", "LocalQueue", "gco-jobs", "gco-default"),
702 ("kueue.x-k8s.io/v1beta1", "ClusterQueue", None, "gco-cluster-queue"),
703 ("kueue.x-k8s.io/v1beta1", "ResourceFlavor", None, "gco-default-flavor"),
704 ),
705 ("{{SLURM_ENABLED}}", True): (
706 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-cluster-internal"),
707 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-client-to-restapi"),
708 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-client-egress"),
709 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-operator-to-restapi"),
710 ),
711 ("{{KUBEFLOW_TRAINER_ENABLED}}", True): (
712 ("trainer.kubeflow.org/v1alpha1", "ClusterTrainingRuntime", None, "torch-distributed"),
713 ),
714 ("{{MLFLOW_ENABLED}}", True): (
715 # The claim is created BY THE CHART (storage.enabled), not by a
716 # shipped manifest — it appears here because helm uninstall never
717 # deletes chart PVCs, so disabling the feature would otherwise leak
718 # the volume forever. Deliberately destructive on disable: the claim
719 # holds the tracking server's SQLite run METADATA. Run artifacts
720 # live in S3 (untouched).
721 ("v1", "PersistentVolumeClaim", "monitoring", "mlflow"),
722 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-mlflow-clients"),
723 # The server's own network posture; GCO owns it because the chart's
724 # policy drops kubelet probes (post-helm-mlflow-network.yaml).
725 ("networking.k8s.io/v1", "NetworkPolicy", "monitoring", "mlflow-server"),
726 ),
727 ("{{COST_MONITORING_ENABLED}}", False): (
728 ("apps/v1", "Deployment", "gco-system", "cost-monitor"),
729 ("v1", "Service", "gco-system", "cost-monitor"),
730 ("v1", "ServiceAccount", "gco-system", "gco-cost-monitor-sa"),
731 (
732 "networking.k8s.io/v1",
733 "NetworkPolicy",
734 "gco-system",
735 "allow-manifest-processor-to-cost-monitor-ingress",
736 ),
737 (
738 "networking.k8s.io/v1",
739 "NetworkPolicy",
740 "gco-system",
741 "allow-cost-monitor-to-opencost",
742 ),
743 (
744 "networking.k8s.io/v1",
745 "NetworkPolicy",
746 "gco-system",
747 "allow-manifest-processor-to-cost-monitor-egress",
748 ),
749 ),
750 ("{{COST_MONITORING_ENABLED}}", True): (
751 ("v1", "ConfigMap", "monitoring", "gco-dashboard-cost"),
752 ),
753}
756# Resources GCO shipped in earlier releases that no longer appear in the
757# manifest set. The base apply pass deletes them exactly (missing = no-op) so
758# upgraded clusters do not keep orphaned objects running forever.
759#
760# nvidia-device-plugin-daemonset: GCO runs exclusively on EKS Auto Mode, which
761# ships its own NVIDIA device plugin built into the node ("runs automatically
762# and isn't visible as a daemon set" — the EKS auto-accelerated guide). The
763# community plugin GCO used to ship can never start on Auto Mode GPU nodes:
764# the runtime only injects the NVIDIA driver libraries for containers that
765# request them, so the plugin crash-loops with NVML ERROR_LIBRARY_NOT_FOUND
766# and permanently fails DaemonSet convergence (observed live the moment the
767# Slurm NodeSet provisioned the first GPU nodes). The built-in plugin
768# advertises nvidia.com/gpu on its own.
769#
770# allow-vpc-endpoint-egress / allow-ray-cluster-internal (gco-jobs): the
771# pre-v7.7 job-namespace model — HTTPS only to the VPC's own CIDR (there were
772# never VPC endpoints for that traffic to reach, so enforced it cut jobs off
773# from every AWS API) and a Ray-only peer rule. 03-network-policies.yaml now
774# ships allow-https-egress + allow-vpc-egress + allow-same-namespace, which
775# strictly contain both. NetworkPolicies union, so the leftovers would allow
776# nothing new — they are swept so the live policy set stays exactly the
777# shipped, documented one.
778_LEGACY_REMOVED_RESOURCES: tuple[tuple[str, str, str | None, str], ...] = (
779 ("apps/v1", "DaemonSet", "kube-system", "nvidia-device-plugin-daemonset"),
780 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-vpc-endpoint-egress"),
781 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-ray-cluster-internal"),
782)
785def _delete_exact_resources(
786 targets: tuple[tuple[str, str, str | None, str], ...],
787 context: str,
788) -> dict[str, list[str]]:
789 """Delete an exact list of (apiVersion, kind, namespace, name) resources.
791 Missing resources and missing CRDs/API resource types are successful no-ops.
792 Every other error is returned so convergence fails instead of silently
793 leaving stale resources running.
794 """
795 result: dict[str, list[str]] = {"pruned": [], "failed": []}
796 if not targets:
797 return result
799 dynamic_client = dynamic.DynamicClient(client.ApiClient())
800 delete_options = client.V1DeleteOptions(propagation_policy="Background")
801 for api_version, kind, namespace, name in targets:
802 identifier = f"{api_version}/{kind}/{namespace or '<cluster>'}/{name}"
803 try:
804 resource = dynamic_client.resources.get(api_version=api_version, kind=kind)
805 kwargs: dict[str, Any] = {"name": name, "body": delete_options}
806 if namespace is not None:
807 kwargs["namespace"] = namespace
808 resource.delete(**kwargs)
809 result["pruned"].append(identifier)
810 logger.info("Pruned %s resource %s", context, identifier)
811 except ResourceNotFoundError, NotFoundError:
812 logger.info("%s resource already absent: %s", context, identifier)
813 except ApiException as exc:
814 if exc.status == 404:
815 logger.info("%s resource already absent: %s", context, identifier)
816 else:
817 failure = f"{identifier}:{exc.status}:{exc.reason}"
818 result["failed"].append(failure)
819 logger.error("Failed pruning %s resource %s", context, failure)
820 except Exception as exc:
821 if getattr(exc, "status", None) == 404:
822 logger.info("%s resource already absent: %s", context, identifier)
823 else:
824 failure = f"{identifier}:{exc}"
825 result["failed"].append(failure)
826 logger.error("Failed pruning %s resource %s", context, failure)
828 return result
831def _prune_disabled_feature(placeholder: str, post_helm: bool) -> dict[str, list[str]]:
832 """Delete only the exact resources managed by a disabled optional feature."""
833 return _delete_exact_resources(
834 _FEATURE_RESOURCE_INVENTORY.get((placeholder, post_helm), ()),
835 "disabled-feature",
836 )
839def _prune_legacy_removed_resources() -> dict[str, list[str]]:
840 """Delete resources shipped by earlier GCO releases and since removed."""
841 return _delete_exact_resources(_LEGACY_REMOVED_RESOURCES, "legacy-removed")
844# ---------------------------------------------------------------------------
845# Tunables
846# ---------------------------------------------------------------------------
848# Maximum time to wait for a PersistentVolume or PersistentVolumeClaim to
849# disappear after we issue a delete. Needed when we're recreating a PV
850# whose ``volumeHandle`` changed (FSx/EFS ID rotated) or reconciling a
851# Lost PVC whose backing PV was just recreated. If you see this wait
852# consistently timing out, check for stuck finalizers (the handler
853# already clears the standard ``pv-protection`` / ``pvc-protection``
854# finalizers) or for an AWS control-plane issue on the underlying
855# volume. Raising the value is safe — it only blocks the re-create path,
856# not steady-state applies.
857PV_PVC_DELETE_WAIT_SECONDS = 30
859# Interval between PV/PVC existence polls during the delete wait.
860PV_PVC_DELETE_POLL_INTERVAL_SECONDS = 1
863def get_eks_client() -> Any:
864 """Get EKS client with lazy initialization."""
865 global _eks_client
866 if _eks_client is None:
867 _eks_client = boto3.client("eks")
868 return _eks_client
871def send_response(
872 event: dict[str, Any],
873 context: Any,
874 response_status: str,
875 response_data: dict[str, Any],
876 physical_resource_id: str,
877 reason: str | None = None,
878) -> None:
879 """Send response to CloudFormation."""
880 response_body = {
881 "Status": response_status,
882 "Reason": reason or f"See CloudWatch Log Stream: {context.log_stream_name}",
883 "PhysicalResourceId": physical_resource_id,
884 "StackId": event["StackId"],
885 "RequestId": event["RequestId"],
886 "LogicalResourceId": event["LogicalResourceId"],
887 "Data": response_data,
888 }
890 logger.info(f"Sending response: {json.dumps(response_data)}")
892 # Timeout is for the CFN response callback (HTTP PUT to S3 presigned URL),
893 # not for manifest application. K8s API calls have their own timeouts.
894 http = urllib3.PoolManager()
895 try:
896 http.request(
897 "PUT",
898 event["ResponseURL"],
899 body=json.dumps(response_body).encode("utf-8"),
900 headers={"Content-Type": "application/json"},
901 timeout=10.0,
902 )
903 except Exception as e:
904 logger.error(f"Failed to send response: {e}")
907def get_eks_token(cluster_name: str, region: str) -> str:
908 """Generate EKS authentication token using STS presigned URL."""
909 from botocore.signers import RequestSigner
911 # Create STS client
912 session = boto3.Session()
913 sts_client = session.client("sts", region_name=region)
914 service_id = sts_client.meta.service_model.service_id
916 # Create request signer
917 signer = RequestSigner(
918 service_id, region, "sts", "v4", session.get_credentials(), session.events
919 )
921 # Build the presigned URL for GetCallerIdentity
922 params = {
923 "method": "GET",
924 "url": f"https://sts.{region}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15",
925 "body": {},
926 "headers": {"x-k8s-aws-id": cluster_name},
927 "context": {},
928 }
930 # Generate presigned URL (valid for 60 seconds)
931 url = signer.generate_presigned_url(
932 params, region_name=region, expires_in=60, operation_name=""
933 )
935 # Encode as base64 and create the k8s-aws-v1 token
936 token_b64 = base64.urlsafe_b64encode(url.encode("utf-8")).decode("utf-8").rstrip("=")
937 return f"k8s-aws-v1.{token_b64}"
940def configure_k8s_client(cluster_name: str, region: str) -> None:
941 """Configure Kubernetes client for EKS cluster."""
942 eks = get_eks_client()
944 # Get cluster info
945 cluster_info = eks.describe_cluster(name=cluster_name)
946 cluster = cluster_info["cluster"]
948 # Configure Kubernetes client
949 configuration = client.Configuration()
950 configuration.host = cluster["endpoint"]
951 configuration.verify_ssl = True
953 # Set connection timeouts (important for Lambda!)
954 configuration.connection_pool_maxsize = 1
955 configuration.retries = 3
956 # Set socket timeout to 30 seconds
957 import socket
959 socket.setdefaulttimeout(30)
961 # Decode and write CA certificate to temp file using secure method
962 ca_cert = base64.b64decode(cluster["certificateAuthority"]["data"])
963 import tempfile
965 fd, ca_cert_path = tempfile.mkstemp(suffix=".crt")
966 try:
967 with os.fdopen(fd, "wb") as ca_file:
968 ca_file.write(ca_cert)
969 ca_file.flush()
970 configuration.ssl_ca_cert = ca_cert_path
971 except Exception:
972 os.close(fd)
973 raise
975 # Generate EKS authentication token
976 eks_token = get_eks_token(cluster_name, region)
978 # Set the bearer token
979 configuration.api_key = {"authorization": f"Bearer {eks_token}"}
981 logger.info(
982 f"✓ Configured Kubernetes client for cluster {cluster_name} at {cluster['endpoint']}"
983 )
985 client.Configuration.set_default(configuration)
988def restart_deployments(namespace: str, deployment_names: list[str]) -> dict[str, Any]:
989 """
990 Restart deployments by patching their spec with a restart annotation.
991 This forces Kubernetes to roll out new pods with the latest image.
992 """
993 from datetime import datetime
995 apps_v1 = client.AppsV1Api()
996 restarted = []
997 failed = []
999 restart_time = datetime.now(UTC).isoformat()
1001 for name in deployment_names:
1002 try:
1003 # Patch the deployment with a restart annotation
1004 # This is equivalent to `kubectl rollout restart deployment`
1005 patch = {
1006 "spec": {
1007 "template": {
1008 "metadata": {
1009 "annotations": {"kubectl.kubernetes.io/restartedAt": restart_time}
1010 }
1011 }
1012 }
1013 }
1014 apps_v1.patch_namespaced_deployment(name, namespace, body=patch)
1015 restarted.append(name)
1016 logger.info(f"✓ Restarted deployment {name} in namespace {namespace}")
1017 except ApiException as e:
1018 # 404 means the deployment isn't installed on this cluster
1019 # (e.g. fsx-csi when FSx is disabled) — that's not a failure.
1020 if e.status == 404:
1021 logger.info(f"Deployment {namespace}/{name} not found — skipping restart")
1022 else:
1023 logger.error(f"Failed to restart deployment {name}: {e.status} - {e.reason}")
1024 failed.append(name)
1026 return {"restarted": restarted, "failed": failed}
1029def restart_daemonsets(namespace: str, daemonset_names: list[str]) -> dict[str, Any]:
1030 """
1031 Restart daemonsets by patching their pod template with a restart annotation.
1032 This forces Kubernetes to roll out new pods with the latest image or latest
1033 service-account annotation set.
1035 Used for IRSA-annotated addons whose pods still need to be re-mutated by
1036 the EKS Pod Identity webhook after the addon's service account had its
1037 role ARN annotation patched post-install. Without this, DaemonSet pods
1038 like efs-csi-node, fsx-csi-node, and cloudwatch-agent keep their
1039 original (credential-less) pod spec and silently fail with IMDS 401s.
1040 """
1041 from datetime import datetime
1043 apps_v1 = client.AppsV1Api()
1044 restarted = []
1045 failed = []
1047 restart_time = datetime.now(UTC).isoformat()
1049 for name in daemonset_names:
1050 try:
1051 patch = {
1052 "spec": {
1053 "template": {
1054 "metadata": {
1055 "annotations": {"kubectl.kubernetes.io/restartedAt": restart_time}
1056 }
1057 }
1058 }
1059 }
1060 apps_v1.patch_namespaced_daemon_set(name, namespace, body=patch)
1061 restarted.append(name)
1062 logger.info(f"✓ Restarted daemonset {name} in namespace {namespace}")
1063 except ApiException as e:
1064 # 404 means the daemonset isn't installed on this cluster
1065 # (e.g. fsx-csi-node when FSx is disabled) — that's expected,
1066 # not a failure.
1067 if e.status == 404:
1068 logger.info(f"DaemonSet {namespace}/{name} not found — skipping restart")
1069 else:
1070 logger.error(f"Failed to restart daemonset {name}: {e.status} - {e.reason}")
1071 failed.append(name)
1073 return {"restarted": restarted, "failed": failed}
1076# Every platform Deployment GCO ships in gco-system, with the dedicated
1077# ServiceAccount it must run as: (namespace, deployment, service account).
1078# tests/test_platform_workload_contract.py pins this to the manifests, so a new
1079# service (or a renamed account) fails a unit test instead of silently escaping
1080# the post-apply credential verification below.
1081PLATFORM_DEPLOYMENTS: tuple[tuple[str, str, str], ...] = (
1082 ("gco-system", "health-monitor", "gco-health-monitor-sa"),
1083 ("gco-system", "manifest-processor", "gco-manifest-processor-sa"),
1084 ("gco-system", "inference-monitor", "gco-inference-monitor-sa"),
1085 ("gco-system", "inference-proxy", "gco-inference-proxy-sa"),
1086 # Feature-gated (cdk.json cost_monitoring); verified only when planned.
1087 ("gco-system", "cost-monitor", "gco-cost-monitor-sa"),
1088)
1090# User-workload accounts the manifests declare outside gco-system.
1091WORKLOAD_SERVICE_ACCOUNTS: tuple[tuple[str, str], ...] = (
1092 ("gco-jobs", "gco-service-account"),
1093 ("gco-inference", "gco-service-account"),
1094)
1097def _verify_workload_credentials(
1098 apps_v1: Any,
1099 planned: list[dict[str, Any]] | None = None,
1100) -> list[str]:
1101 """Verify that key GCO deployments have working IAM credential configuration.
1103 Checks that:
1104 1. Deployments use their dedicated service account (with IRSA annotation)
1105 2. The projected service-account token volume is mounted
1106 3. AWS_ROLE_ARN and AWS_WEB_IDENTITY_TOKEN_FILE env vars are set
1108 ``planned`` is the base-phase resource list from ``plan_manifests``. When
1109 given, a platform Deployment (and its ServiceAccount) is checked only if
1110 this apply planned it, so a feature-gated service that is switched off
1111 (cost-monitor) does not surface as "deployment not found".
1113 Returns a list of warning strings (empty = all good).
1114 """
1115 warnings: list[str] = []
1116 planned_identities: set[tuple[str, str, str]] | None = None
1117 if planned is not None:
1118 planned_identities = {
1119 (str(item["kind"]), str(item["namespace"]), str(item["name"])) for item in planned
1120 }
1122 def _is_planned(kind: str, namespace: str, name: str) -> bool:
1123 return planned_identities is None or (kind, namespace, name) in planned_identities
1125 expected_deployments = [
1126 (namespace, name, service_account)
1127 for namespace, name, service_account in PLATFORM_DEPLOYMENTS
1128 if _is_planned("Deployment", namespace, name)
1129 ]
1131 for namespace, name, expected_sa in expected_deployments:
1132 try:
1133 dep = apps_v1.read_namespaced_deployment(name, namespace)
1134 spec = dep.spec.template.spec
1136 # Check service account
1137 if spec.service_account_name != expected_sa:
1138 warnings.append(
1139 f"{namespace}/{name}: uses SA '{spec.service_account_name}' instead of {expected_sa}"
1140 )
1142 # Check for projected token volume
1143 has_token_volume = False
1144 if spec.volumes:
1145 for vol in spec.volumes:
1146 if vol.projected and vol.projected.sources:
1147 for src in vol.projected.sources:
1148 if (
1149 src.service_account_token
1150 and src.service_account_token.audience == "sts.amazonaws.com"
1151 ):
1152 has_token_volume = True
1153 break
1155 if not has_token_volume:
1156 warnings.append(
1157 f"{namespace}/{name}: missing projected service-account token volume for IRSA"
1158 )
1160 # Check env vars on first container
1161 container = spec.containers[0] if spec.containers else None
1162 if container and container.env:
1163 env_names = {e.name for e in container.env}
1164 if "AWS_ROLE_ARN" not in env_names:
1165 warnings.append(f"{namespace}/{name}: missing AWS_ROLE_ARN env var")
1166 if "AWS_WEB_IDENTITY_TOKEN_FILE" not in env_names:
1167 warnings.append(
1168 f"{namespace}/{name}: missing AWS_WEB_IDENTITY_TOKEN_FILE env var"
1169 )
1171 except ApiException as e:
1172 if e.status == 404:
1173 warnings.append(f"{namespace}/{name}: deployment not found")
1174 else:
1175 warnings.append(f"{namespace}/{name}: failed to read ({e.status})")
1176 except Exception as e:
1177 warnings.append(f"{namespace}/{name}: verification error ({e})")
1179 # Check that service accounts exist in all required namespaces: the
1180 # dedicated account of every planned platform Deployment, then the user
1181 # workload accounts.
1182 v1 = client.CoreV1Api()
1183 platform_sas = [(namespace, sa_name) for namespace, _name, sa_name in expected_deployments]
1184 workload_sas = [
1185 (namespace, sa_name)
1186 for namespace, sa_name in WORKLOAD_SERVICE_ACCOUNTS
1187 if _is_planned("ServiceAccount", namespace, sa_name)
1188 ]
1189 for namespace, sa_name in platform_sas + workload_sas:
1190 try:
1191 sa = v1.read_namespaced_service_account(sa_name, namespace)
1192 annotations = sa.metadata.annotations or {}
1193 if "eks.amazonaws.com/role-arn" not in annotations:
1194 warnings.append(
1195 f"{namespace}/{sa_name}: missing eks.amazonaws.com/role-arn annotation"
1196 )
1197 except ApiException as e:
1198 if e.status == 404:
1199 warnings.append(f"{namespace}/{sa_name}: ServiceAccount not found")
1200 else:
1201 warnings.append(f"{namespace}/{sa_name}: failed to read ({e.status})")
1203 if warnings:
1204 for w in warnings:
1205 logger.warning(f"⚠ Credential check: {w}")
1206 else:
1207 logger.info("✓ All workload IAM credential configurations verified")
1209 return warnings
1212def apply_manifests(
1213 cluster_name: str,
1214 region: str,
1215 manifests_dir: str,
1216 replacements: dict[str, str],
1217 post_helm: bool = False,
1218) -> dict[str, Any]:
1219 """Apply Kubernetes manifests.
1221 Args:
1222 cluster_name: EKS cluster name
1223 region: AWS region
1224 manifests_dir: Directory containing manifest YAML files
1225 replacements: Template variable substitutions
1226 post_helm: If True, apply only post-helm-* files (run after Helm installs CRDs).
1227 If False (default), apply all other files and skip post-helm-* ones.
1228 """
1229 plan = plan_manifests(manifests_dir, replacements)
1230 phase_name = "post-helm" if post_helm else "base"
1231 planned_resources = plan["phases"][phase_name]
1232 expected_resources = [_public_manifest_identity(item) for item in planned_resources]
1233 expected_count = len(planned_resources)
1235 configure_k8s_client(cluster_name, region)
1236 v1 = client.CoreV1Api()
1237 apps_v1 = client.AppsV1Api()
1238 autoscaling_v2 = client.AutoscalingV2Api()
1239 rbac_v1 = client.RbacAuthorizationV1Api()
1240 networking_v1 = client.NetworkingV1Api()
1241 custom_api = client.CustomObjectsApi()
1243 applied_count = 0
1244 failed: list[str] = []
1245 skipped: list[str] = list(plan["skipped"][phase_name])
1246 pruned: list[str] = []
1247 prune_failures: list[str] = []
1249 # A gated-out file represents an optional feature that is disabled. Preserve
1250 # the existing exact-resource pruning behavior, once per gate and phase.
1251 for placeholder in plan["featureGates"][phase_name]:
1252 if (placeholder, post_helm) not in _FEATURE_RESOURCE_INVENTORY:
1253 continue
1254 prune_result = _prune_disabled_feature(placeholder, post_helm)
1255 pruned.extend(prune_result["pruned"])
1256 prune_failures.extend(prune_result["failed"])
1257 failed.extend(f"prune:{failure}" for failure in prune_result["failed"])
1259 # Objects GCO used to ship that no longer exist in any manifest — delete
1260 # them exactly on upgraded clusters (fresh clusters: no-op). Base pass
1261 # only, so the sweep runs once per convergence.
1262 if not post_helm:
1263 legacy_result = _prune_legacy_removed_resources()
1264 pruned.extend(legacy_result["pruned"])
1265 prune_failures.extend(legacy_result["failed"])
1266 failed.extend(f"prune:{failure}" for failure in legacy_result["failed"])
1268 for planned_resource in planned_resources:
1269 filename = planned_resource["sourceFile"]
1270 try:
1271 for doc in (planned_resource["document"],):
1272 kind = planned_resource["kind"]
1273 api_version = planned_resource["apiVersion"]
1274 namespace = doc.get("metadata", {}).get("namespace", "default")
1275 name = planned_resource["name"]
1277 logger.info(f"Applying {kind}/{name} in namespace {namespace}")
1278 try:
1279 # Apply based on kind
1280 if kind == "Namespace":
1281 try:
1282 v1.create_namespace(body=doc)
1283 except ApiException as e:
1284 if e.status == 409: # Already exists
1285 v1.patch_namespace(name, body=doc)
1286 else:
1287 raise
1289 elif kind == "ServiceAccount":
1290 _log_service_account_automount_flip(v1, doc, namespace, name, plan)
1291 try:
1292 v1.create_namespaced_service_account(namespace, body=doc)
1293 except ApiException as e:
1294 if e.status == 409:
1295 v1.patch_namespaced_service_account(name, namespace, body=doc)
1296 else:
1297 raise
1299 elif kind == "ClusterRole":
1300 try:
1301 rbac_v1.create_cluster_role(body=doc)
1302 except ApiException as e:
1303 if e.status == 409:
1304 rbac_v1.patch_cluster_role(name, body=doc)
1305 else:
1306 raise
1308 elif kind == "ClusterRoleBinding":
1309 try:
1310 rbac_v1.create_cluster_role_binding(body=doc)
1311 except ApiException as e:
1312 if e.status == 409:
1313 rbac_v1.patch_cluster_role_binding(name, body=doc)
1314 else:
1315 raise
1317 elif kind == "Role":
1318 try:
1319 rbac_v1.create_namespaced_role(namespace, body=doc)
1320 except ApiException as e:
1321 if e.status == 409:
1322 rbac_v1.patch_namespaced_role(name, namespace, body=doc)
1323 else:
1324 raise
1326 elif kind == "RoleBinding":
1327 try:
1328 rbac_v1.create_namespaced_role_binding(namespace, body=doc)
1329 except ApiException as e:
1330 if e.status == 409:
1331 rbac_v1.patch_namespaced_role_binding(name, namespace, body=doc)
1332 else:
1333 raise
1335 elif kind == "Lease":
1336 coordination_v1 = client.CoordinationV1Api()
1337 try:
1338 coordination_v1.create_namespaced_lease(namespace, body=doc)
1339 except ApiException as e:
1340 if e.status == 409:
1341 coordination_v1.patch_namespaced_lease(name, namespace, body=doc)
1342 else:
1343 raise
1345 elif kind == "Deployment":
1346 try:
1347 apps_v1.create_namespaced_deployment(namespace, body=doc)
1348 except ApiException as e:
1349 if e.status == 409:
1350 apps_v1.patch_namespaced_deployment(
1351 name,
1352 namespace,
1353 body=_deployment_patch_body(doc),
1354 )
1355 else:
1356 raise
1358 elif kind == "StatefulSet":
1359 try:
1360 apps_v1.create_namespaced_stateful_set(namespace, body=doc)
1361 except ApiException as e:
1362 if e.status == 409:
1363 apps_v1.patch_namespaced_stateful_set(name, namespace, body=doc)
1364 else:
1365 raise
1367 elif kind == "DaemonSet":
1368 try:
1369 apps_v1.create_namespaced_daemon_set(namespace, body=doc)
1370 except ApiException as e:
1371 if e.status == 409:
1372 apps_v1.patch_namespaced_daemon_set(name, namespace, body=doc)
1373 else:
1374 raise
1376 elif kind == "Job":
1377 batch_v1 = client.BatchV1Api()
1378 try:
1379 batch_v1.create_namespaced_job(namespace, body=doc)
1380 except ApiException as e:
1381 if e.status == 409:
1382 batch_v1.patch_namespaced_job(name, namespace, body=doc)
1383 else:
1384 raise
1386 elif kind == "CronJob":
1387 # batch/v1 CronJob (e.g. the Grafana admin-password
1388 # rotation job in the observability post-Helm pass).
1389 batch_v1 = client.BatchV1Api()
1390 try:
1391 batch_v1.create_namespaced_cron_job(namespace, body=doc)
1392 except ApiException as e:
1393 if e.status == 409:
1394 batch_v1.patch_namespaced_cron_job(name, namespace, body=doc)
1395 else:
1396 raise
1398 elif kind == "HorizontalPodAutoscaler":
1399 try:
1400 autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(
1401 namespace, body=doc
1402 )
1403 except ApiException as e:
1404 if e.status == 409:
1405 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler(
1406 name, namespace, body=doc
1407 )
1408 else:
1409 raise
1411 elif kind == "PodDisruptionBudget":
1412 policy_v1 = client.PolicyV1Api()
1413 try:
1414 policy_v1.create_namespaced_pod_disruption_budget(namespace, body=doc)
1415 except ApiException as e:
1416 if e.status == 409:
1417 policy_v1.patch_namespaced_pod_disruption_budget(
1418 name, namespace, body=doc
1419 )
1420 else:
1421 raise
1423 elif kind == "Service":
1424 try:
1425 v1.create_namespaced_service(namespace, body=doc)
1426 except ApiException as e:
1427 if e.status == 409:
1428 v1.patch_namespaced_service(name, namespace, body=doc)
1429 else:
1430 raise
1432 elif kind == "Pod":
1433 try:
1434 v1.create_namespaced_pod(namespace, body=doc)
1435 except ApiException as e:
1436 if e.status == 409:
1437 v1.patch_namespaced_pod(name, namespace, body=doc)
1438 else:
1439 raise
1441 elif kind == "ConfigMap":
1442 try:
1443 v1.create_namespaced_config_map(namespace, body=doc)
1444 except ApiException as e:
1445 if e.status == 409:
1446 v1.patch_namespaced_config_map(name, namespace, body=doc)
1447 else:
1448 raise
1450 elif kind == "Secret":
1451 try:
1452 v1.create_namespaced_secret(namespace, body=doc)
1453 except ApiException as e:
1454 if e.status == 409:
1455 v1.patch_namespaced_secret(name, namespace, body=doc)
1456 else:
1457 raise
1459 elif (
1460 kind in _GATEWAY_CUSTOM_OBJECTS
1461 or kind in _QUEUEING_CUSTOM_OBJECTS
1462 or kind in _CERT_MANAGER_CUSTOM_OBJECTS
1463 ):
1464 group, version, plural, cluster_scoped = (
1465 _GATEWAY_CUSTOM_OBJECTS.get(kind)
1466 or _QUEUEING_CUSTOM_OBJECTS.get(kind)
1467 or _CERT_MANAGER_CUSTOM_OBJECTS[kind]
1468 )
1469 try:
1470 if cluster_scoped:
1471 custom_api.create_cluster_custom_object(
1472 group, version, plural, body=doc
1473 )
1474 else:
1475 custom_api.create_namespaced_custom_object(
1476 group, version, namespace, plural, body=doc
1477 )
1478 except ApiException as e:
1479 if e.status != 409:
1480 raise
1481 if cluster_scoped:
1482 custom_api.patch_cluster_custom_object(
1483 group, version, plural, name, body=doc
1484 )
1485 else:
1486 custom_api.patch_namespaced_custom_object(
1487 group, version, namespace, plural, name, body=doc
1488 )
1490 elif kind == "StorageClass":
1491 storage_v1 = client.StorageV1Api()
1492 try:
1493 storage_v1.create_storage_class(body=doc)
1494 except ApiException as e:
1495 if e.status == 409:
1496 # StorageClass already exists - skip patching as most fields are immutable
1497 logger.info(f"StorageClass {name} already exists, skipping update")
1498 else:
1499 raise
1500 elif kind == "PriorityClass":
1501 scheduling_v1 = client.SchedulingV1Api()
1502 try:
1503 scheduling_v1.create_priority_class(body=doc)
1504 except ApiException as e:
1505 if e.status == 409:
1506 # Converge the mutable fields (description,
1507 # labels, globalDefault). ``value`` and
1508 # ``preemptionPolicy`` are immutable: patching
1509 # them with an unchanged value is a no-op,
1510 # while a genuine change fails loudly (422)
1511 # instead of silently keeping the old
1512 # priority — delete the class and redeploy to
1513 # change a value.
1514 scheduling_v1.patch_priority_class(name, body=doc)
1515 else:
1516 raise
1518 elif kind == "PersistentVolume":
1519 try:
1520 v1.create_persistent_volume(body=doc)
1521 except ApiException as e:
1522 if e.status == 409:
1523 # PVs have immutable spec fields. Check if the existing PV
1524 # matches — if so, skip. If the volumeHandle changed (new
1525 # FSx file system), force-remove the old PV and recreate.
1526 existing = v1.read_persistent_volume(name)
1527 existing_handle = (
1528 existing.spec.csi.volume_handle if existing.spec.csi else None
1529 )
1530 new_handle = doc.get("spec", {}).get("csi", {}).get("volumeHandle")
1532 if existing_handle == new_handle:
1533 logger.info(f"PersistentVolume {name} unchanged, skipping")
1534 else:
1535 logger.info(
1536 f"PersistentVolume {name} volumeHandle changed "
1537 f"({existing_handle} → {new_handle}), recreating"
1538 )
1539 # Remove the protection finalizer so the PV can be deleted
1540 # even while bound to a PVC
1541 v1.patch_persistent_volume(
1542 name,
1543 body={"metadata": {"finalizers": None}},
1544 )
1545 v1.delete_persistent_volume(name)
1546 # Wait for the PV to actually disappear
1547 import time as _time
1549 _pv_iterations = max(
1550 1,
1551 PV_PVC_DELETE_WAIT_SECONDS
1552 // PV_PVC_DELETE_POLL_INTERVAL_SECONDS,
1553 )
1554 for _wait in range(_pv_iterations):
1555 try:
1556 v1.read_persistent_volume(name)
1557 _time.sleep(PV_PVC_DELETE_POLL_INTERVAL_SECONDS)
1558 except ApiException as read_e:
1559 if read_e.status == 404:
1560 break
1561 raise
1562 v1.create_persistent_volume(body=doc)
1563 else:
1564 raise
1566 elif kind == "PersistentVolumeClaim":
1567 try:
1568 v1.create_namespaced_persistent_volume_claim(namespace, body=doc)
1569 except ApiException as e:
1570 if e.status == 409:
1571 # Check if the PVC is in Lost state (bound PV was recreated).
1572 # A Lost PVC can't be patched back to health — it must be
1573 # deleted and recreated so it binds to the new PV.
1574 existing_pvc = v1.read_namespaced_persistent_volume_claim(
1575 name, namespace
1576 )
1577 if existing_pvc.status.phase == "Lost":
1578 logger.info(
1579 f"PVC {namespace}/{name} is Lost (bound PV was "
1580 f"recreated), deleting and recreating"
1581 )
1582 v1.delete_namespaced_persistent_volume_claim(name, namespace)
1583 import time as _time
1585 _pvc_iterations = max(
1586 1,
1587 PV_PVC_DELETE_WAIT_SECONDS
1588 // PV_PVC_DELETE_POLL_INTERVAL_SECONDS,
1589 )
1590 for _wait in range(_pvc_iterations):
1591 try:
1592 v1.read_namespaced_persistent_volume_claim(
1593 name, namespace
1594 )
1595 _time.sleep(PV_PVC_DELETE_POLL_INTERVAL_SECONDS)
1596 except ApiException as read_e:
1597 if read_e.status == 404:
1598 break
1599 raise
1600 v1.create_namespaced_persistent_volume_claim(
1601 namespace, body=doc
1602 )
1603 else:
1604 v1.patch_namespaced_persistent_volume_claim(
1605 name, namespace, body=doc
1606 )
1607 else:
1608 raise
1610 elif kind == "NodePool":
1611 # Karpenter NodePool CRD
1612 group = "karpenter.sh"
1613 version = api_version.split("/")[-1] if "/" in api_version else "v1"
1614 plural = "nodepools"
1615 try:
1616 custom_api.create_cluster_custom_object(
1617 group, version, plural, body=doc
1618 )
1619 except ApiException as e:
1620 if e.status == 409:
1621 custom_api.patch_cluster_custom_object(
1622 group, version, plural, name, body=doc
1623 )
1624 else:
1625 raise
1627 elif kind == "EC2NodeClass":
1628 # Karpenter EC2NodeClass CRD
1629 group = "karpenter.k8s.aws"
1630 version = api_version.split("/")[-1] if "/" in api_version else "v1"
1631 plural = "ec2nodeclasses"
1632 try:
1633 custom_api.create_cluster_custom_object(
1634 group, version, plural, body=doc
1635 )
1636 except ApiException as e:
1637 if e.status == 409:
1638 custom_api.patch_cluster_custom_object(
1639 group, version, plural, name, body=doc
1640 )
1641 else:
1642 raise
1644 elif kind == "APIService":
1645 # Kubernetes API aggregation layer
1646 api_reg_v1 = client.ApiregistrationV1Api()
1647 try:
1648 api_reg_v1.create_api_service(body=doc)
1649 except ApiException as e:
1650 if e.status == 409:
1651 api_reg_v1.patch_api_service(name, body=doc)
1652 else:
1653 raise
1655 elif kind == "CustomResourceDefinition":
1656 api_extensions_v1 = client.ApiextensionsV1Api()
1657 try:
1658 api_extensions_v1.create_custom_resource_definition(body=doc)
1659 except ApiException as e:
1660 if e.status == 409:
1661 api_extensions_v1.patch_custom_resource_definition(name, body=doc)
1662 else:
1663 raise
1665 elif kind == "DeviceClass":
1666 # Kubernetes DRA DeviceClass (resource.k8s.io)
1667 group = "resource.k8s.io"
1668 version = api_version.split("/")[-1] if "/" in api_version else "v1"
1669 plural = "deviceclasses"
1670 try:
1671 custom_api.create_cluster_custom_object(
1672 group, version, plural, body=doc
1673 )
1674 except ApiException as e:
1675 if e.status == 409:
1676 custom_api.patch_cluster_custom_object(
1677 group, version, plural, name, body=doc
1678 )
1679 else:
1680 raise
1681 elif kind == "ClusterTrainingRuntime":
1682 # Kubeflow Trainer v2 runtime blueprint (cluster-scoped;
1683 # the CRD is registered by the kubeflow-trainer chart, so
1684 # the shipped torch-distributed runtime lands in the
1685 # post-Helm pass).
1686 group = "trainer.kubeflow.org"
1687 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
1688 plural = "clustertrainingruntimes"
1689 try:
1690 custom_api.create_cluster_custom_object(
1691 group, version, plural, body=doc
1692 )
1693 except ApiException as e:
1694 if e.status == 409:
1695 custom_api.patch_cluster_custom_object(
1696 group, version, plural, name, body=doc
1697 )
1698 else:
1699 raise
1701 elif kind == "NetworkPolicy":
1702 try:
1703 networking_v1.create_namespaced_network_policy(namespace, body=doc)
1704 except ApiException as e:
1705 if e.status == 409:
1706 networking_v1.patch_namespaced_network_policy(
1707 name, namespace, body=doc
1708 )
1709 else:
1710 raise
1712 elif kind == "ResourceQuota":
1713 try:
1714 v1.create_namespaced_resource_quota(namespace, body=doc)
1715 except ApiException as e:
1716 if e.status == 409:
1717 v1.patch_namespaced_resource_quota(name, namespace, body=doc)
1718 else:
1719 raise
1721 elif kind == "LimitRange":
1722 try:
1723 v1.create_namespaced_limit_range(namespace, body=doc)
1724 except ApiException as e:
1725 if e.status == 409:
1726 v1.patch_namespaced_limit_range(name, namespace, body=doc)
1727 else:
1728 raise
1730 elif kind == "ScaledJob":
1731 # KEDA ScaledJob CRD
1732 group = "keda.sh"
1733 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
1734 plural = "scaledjobs"
1735 try:
1736 custom_api.create_namespaced_custom_object(
1737 group, version, namespace, plural, body=doc
1738 )
1739 except ApiException as e:
1740 if e.status == 409:
1741 custom_api.patch_namespaced_custom_object(
1742 group, version, namespace, plural, name, body=doc
1743 )
1744 else:
1745 raise
1747 elif kind == "ScaledObject":
1748 # KEDA ScaledObject CRD
1749 group = "keda.sh"
1750 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
1751 plural = "scaledobjects"
1752 try:
1753 custom_api.create_namespaced_custom_object(
1754 group, version, namespace, plural, body=doc
1755 )
1756 except ApiException as e:
1757 if e.status == 409:
1758 custom_api.patch_namespaced_custom_object(
1759 group, version, namespace, plural, name, body=doc
1760 )
1761 else:
1762 raise
1764 elif kind in ("ServiceMonitor", "PodMonitor"):
1765 # Prometheus Operator CRDs registered by the
1766 # kube-prometheus-stack chart, so these land in the
1767 # post-Helm pass. GCO uses ServiceMonitors for
1768 # components fronted by a Service (schedulers, DCGM)
1769 # and PodMonitors for its own multi-replica services.
1770 group = "monitoring.coreos.com"
1771 version = api_version.split("/")[-1] if "/" in api_version else "v1"
1772 plural = "servicemonitors" if kind == "ServiceMonitor" else "podmonitors"
1773 try:
1774 custom_api.create_namespaced_custom_object(
1775 group, version, namespace, plural, body=doc
1776 )
1777 except ApiException as e:
1778 if e.status == 409:
1779 custom_api.patch_namespaced_custom_object(
1780 group, version, namespace, plural, name, body=doc
1781 )
1782 else:
1783 raise
1785 else:
1786 # Defensive only: plan_manifests rejects unsupported kinds.
1787 raise ValueError(f"Planner admitted unsupported kind: {kind}")
1789 applied_count += 1
1790 logger.info(f"✓ Applied {kind}/{name}")
1792 except ApiException as e:
1793 logger.error(f"API error applying {kind}/{name}: {e.status} - {e.reason}")
1794 failed.append(f"{filename}:{kind}/{name}")
1796 except Exception as e:
1797 logger.error(f"Failed to apply {filename}: {e}")
1798 failed.append(filename)
1800 # Invariant, not a reachable path: every iteration above either increments
1801 # applied_count or appends to failed, so a clean pass always applies exactly
1802 # expected_count resources. Kept as a tripwire against future edits to the
1803 # loop that break that accounting.
1804 if not failed and applied_count != expected_count: # pragma: no cover - loop invariant
1805 raise RuntimeError(
1806 f"Manifest apply count mismatch: expected={expected_count} applied={applied_count}"
1807 )
1809 # Restart deployments and verify credentials only on the main (full) pass,
1810 # not on the post-Helm pass
1811 if post_helm:
1812 return {
1813 "AppliedCount": applied_count,
1814 "ExpectedCount": expected_count,
1815 "ExpectedResources": expected_resources,
1816 "FailedCount": len(failed),
1817 "SkippedCount": len(skipped),
1818 "Failed": ",".join(failed) if failed else "None",
1819 "Skipped": ",".join(skipped) if skipped else "None",
1820 "PrunedCount": len(pruned),
1821 "Pruned": ",".join(pruned) if pruned else "None",
1822 "PruneFailures": ",".join(prune_failures) if prune_failures else "None",
1823 }
1825 # GCO Deployments already roll exactly once through their
1826 # gco.aws/deployment-timestamp pod-template annotation (and any image patch).
1827 # Do not immediately add kubectl.kubernetes.io/restartedAt: a second
1828 # back-to-back revision can pin maxUnavailable=0/maxSurge=1 Deployments at
1829 # their surge ceiling while the first revision is still converging.
1831 # Restart the EFS CSI controller so it picks up the IRSA role-ARN
1832 # annotation that the EKS addon update patched onto its service account.
1833 #
1834 # Background: the managed aws-efs-csi-driver addon creates the
1835 # efs-csi-controller-sa ServiceAccount and the controller Deployment in
1836 # parallel. Our stack later calls UpdateAddon with a
1837 # serviceAccountRoleArn, which patches the SA's eks.amazonaws.com/role-arn
1838 # annotation — but EKS does NOT restart the controller pods. The existing
1839 # pods keep their original (un-mutated) pod spec: no AWS_ROLE_ARN,
1840 # no AWS_WEB_IDENTITY_TOKEN_FILE. They then fall back to IMDS for
1841 # credentials, which EKS Auto Mode blocks at the pod network level
1842 # (hop-limit / security policy), causing every EFS CreateAccessPoint
1843 # call to fail with HTTP 401. The visible symptom is a PVC that stays
1844 # Pending forever with "no EC2 IMDS role found".
1845 #
1846 # Restarting the deployment forces a new pod template to go through the
1847 # EKS Pod Identity / IRSA mutating webhook, which sees the annotation
1848 # this time and injects the projected-token volume and env vars.
1849 #
1850 # The same pattern applies to:
1851 # - aws-fsx-csi-driver: fsx-csi-controller Deployment +
1852 # fsx-csi-node DaemonSet
1853 # - amazon-cloudwatch-observability: cloudwatch-agent DaemonSet
1854 # - aws-efs-csi-driver: efs-csi-node DaemonSet (the controller already
1855 # above)
1856 #
1857 # Deployments/DaemonSets that don't exist on this cluster (e.g. FSx when
1858 # fsx_lustre.enabled=false) return 404 and are skipped gracefully — see
1859 # the 404 branch in restart_deployments/restart_daemonsets.
1860 kube_system_deployments = ["efs-csi-controller", "fsx-csi-controller"]
1861 logger.info(f"Restarting deployments in kube-system: {kube_system_deployments}")
1862 ks_deploy_restart = restart_deployments("kube-system", kube_system_deployments)
1864 kube_system_daemonsets = ["efs-csi-node", "fsx-csi-node"]
1865 logger.info(f"Restarting daemonsets in kube-system: {kube_system_daemonsets}")
1866 ks_ds_restart = restart_daemonsets("kube-system", kube_system_daemonsets)
1868 # CloudWatch agent daemonset runs in its own namespace when the
1869 # amazon-cloudwatch-observability addon is installed.
1870 cw_daemonsets = ["cloudwatch-agent"]
1871 logger.info(f"Restarting daemonsets in amazon-cloudwatch: {cw_daemonsets}")
1872 cw_ds_restart = restart_daemonsets("amazon-cloudwatch", cw_daemonsets)
1874 # Verify IAM credentials are available for workloads
1875 # Check that the projected service-account token volume is configured
1876 # on key deployments — if missing, IRSA won't work. Scoped to the
1877 # Deployments this pass planned so a gated-off service is not reported.
1878 credential_warnings = _verify_workload_credentials(apps_v1, planned_resources)
1880 # Combine the restart results for the return payload.
1881 all_restarted = (
1882 ks_deploy_restart["restarted"] + ks_ds_restart["restarted"] + cw_ds_restart["restarted"]
1883 )
1885 return {
1886 "AppliedCount": applied_count,
1887 "ExpectedCount": expected_count,
1888 "ExpectedResources": expected_resources,
1889 "FailedCount": len(failed),
1890 "SkippedCount": len(skipped),
1891 "Failed": ",".join(failed) if failed else "None",
1892 "Skipped": ",".join(skipped) if skipped else "None",
1893 "RestartedDeployments": (",".join(all_restarted) if all_restarted else "None"),
1894 "CredentialWarnings": ",".join(credential_warnings) if credential_warnings else "None",
1895 "PrunedCount": len(pruned),
1896 "Pruned": ",".join(pruned) if pruned else "None",
1897 "PruneFailures": ",".join(prune_failures) if prune_failures else "None",
1898 }
1901def _delete_gateway_resources(cluster_name: str, region: str) -> dict[str, Any]:
1902 """Delete Gateway objects while the LBC controller is still running.
1904 The Gateway is deleted only after its routes, and every object is polled to
1905 exact absence. Waiting for the Gateway's controller finalizer is what proves
1906 the ALB has been removed before the LBC Helm release is uninstalled.
1907 """
1908 configure_k8s_client(cluster_name, region)
1909 custom_api = client.CustomObjectsApi()
1910 namespace = "gco-system"
1911 delete_options = client.V1DeleteOptions(propagation_policy="Foreground")
1912 resources = (
1913 ("HTTPRoute", "gco-routes"),
1914 ("Gateway", "gco-gateway"),
1915 ("LoadBalancerConfiguration", "gco-gateway-load-balancer"),
1916 ("TargetGroupConfiguration", "gco-health-monitor-target-group"),
1917 ("TargetGroupConfiguration", "gco-manifest-processor-target-group"),
1918 ("TargetGroupConfiguration", "gco-inference-proxy-target-group"),
1919 ("TargetGroupConfiguration", "gco-default-target-group"),
1920 ("GatewayClass", "gco-aws-alb"),
1921 )
1922 deleted: list[str] = []
1923 deadline = time.monotonic() + _GATEWAY_DELETE_WAIT_SECONDS
1925 for kind, name in resources:
1926 group, version, plural, cluster_scoped = _GATEWAY_CUSTOM_OBJECTS[kind]
1927 label = f"{kind}/{name}" if cluster_scoped else f"{kind}/{namespace}/{name}"
1928 try:
1929 if cluster_scoped:
1930 custom_api.delete_cluster_custom_object(
1931 group, version, plural, name, body=delete_options
1932 )
1933 else:
1934 custom_api.delete_namespaced_custom_object(
1935 group, version, namespace, plural, name, body=delete_options
1936 )
1937 except ApiException as exc:
1938 if exc.status != 404:
1939 raise RuntimeError(
1940 f"failed to delete {label}: Kubernetes API {exc.status} ({exc.reason})"
1941 ) from exc
1942 deleted.append(f"{label}:already-absent")
1943 continue
1945 while True:
1946 try:
1947 if cluster_scoped:
1948 custom_api.get_cluster_custom_object(group, version, plural, name)
1949 else:
1950 custom_api.get_namespaced_custom_object(group, version, namespace, plural, name)
1951 except ApiException as exc:
1952 if exc.status == 404:
1953 deleted.append(label)
1954 break
1955 raise RuntimeError(
1956 f"failed while waiting for {label} deletion: "
1957 f"Kubernetes API {exc.status} ({exc.reason})"
1958 ) from exc
1959 if time.monotonic() >= deadline:
1960 raise RuntimeError(
1961 "timed out waiting for the complete Gateway resource set to delete "
1962 f"within {_GATEWAY_DELETE_WAIT_SECONDS}s; last resource was {label}"
1963 )
1964 time.sleep(_GATEWAY_DELETE_POLL_SECONDS)
1966 return {"status": "deleted", "DeletedCount": len(deleted), "Deleted": deleted}
1969def _as_plain_dict(value: Any) -> dict[str, Any]:
1970 """Convert a DynamicClient response to a plain mapping."""
1971 if isinstance(value, dict):
1972 return value
1973 converter = getattr(value, "to_dict", None)
1974 if callable(converter):
1975 converted = converter()
1976 if isinstance(converted, dict):
1977 return converted
1978 try:
1979 converted = dict(value)
1980 except (TypeError, ValueError) as exc:
1981 raise TypeError(f"Kubernetes response is not a mapping: {type(value).__name__}") from exc
1982 return converted
1985def _condition_matches(condition: dict[str, Any], expected_status: str) -> bool:
1986 value = condition.get("status")
1987 if isinstance(value, bool):
1988 value = "True" if value else "False"
1989 return str(value).lower() == expected_status.lower()
1992def _conditions(status: dict[str, Any]) -> list[dict[str, Any]]:
1993 value = status.get("conditions", [])
1994 if not isinstance(value, list):
1995 return []
1996 return [condition for condition in value if isinstance(condition, dict)]
1999def _required_condition_failure(
2000 status: dict[str, Any],
2001 condition_type: str,
2002) -> str | None:
2003 matching = [item for item in _conditions(status) if item.get("type") == condition_type]
2004 if any(_condition_matches(item, "True") for item in matching):
2005 return None
2006 if matching:
2007 details = matching[-1].get("message") or matching[-1].get("reason") or "status is not True"
2008 return f"condition {condition_type} is not True ({details})"
2009 return f"condition {condition_type} is missing"
2012def _generation_failure(resource: dict[str, Any]) -> str | None:
2013 metadata_value = resource.get("metadata")
2014 status_value = resource.get("status")
2015 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2016 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2017 generation = metadata.get("generation")
2018 observed = status.get("observedGeneration")
2019 if (
2020 not isinstance(generation, int)
2021 or isinstance(generation, bool)
2022 or not isinstance(observed, int)
2023 or isinstance(observed, bool)
2024 or observed != generation
2025 ):
2026 return f"generation not observed (generation={generation}, observedGeneration={observed})"
2027 return None
2030def _replica_failure(
2031 status: dict[str, Any],
2032 desired: int,
2033 fields: tuple[str, ...],
2034) -> str | None:
2035 for field in fields:
2036 # Kubernetes omits optional zero-valued counters. Treat omission as
2037 # zero while still rejecting malformed non-integer values.
2038 value = status.get(field, 0)
2039 if not isinstance(value, int) or isinstance(value, bool) or value != desired:
2040 return f"replicas not converged ({field}={value}, desired={desired})"
2041 unavailable = status.get("unavailableReplicas", status.get("numberUnavailable", 0))
2042 if not isinstance(unavailable, int) or isinstance(unavailable, bool) or unavailable != 0:
2043 return f"replicas unavailable ({unavailable})"
2044 return None
2047def _current_condition_failure(
2048 conditions: list[dict[str, Any]],
2049 condition_type: str,
2050 generation: Any,
2051 context: str,
2052) -> str | None:
2053 """Require a True condition that observed the object's current generation."""
2054 current = [
2055 condition
2056 for condition in conditions
2057 if condition.get("type") == condition_type
2058 and condition.get("observedGeneration") == generation
2059 ]
2060 if any(_condition_matches(condition, "True") for condition in current):
2061 return None
2062 matching = [condition for condition in conditions if condition.get("type") == condition_type]
2063 if current:
2064 detail = current[-1].get("message") or current[-1].get("reason") or "status is not True"
2065 return f"{context} condition {condition_type} is not True ({detail})"
2066 if matching:
2067 observed = matching[-1].get("observedGeneration")
2068 return (
2069 f"{context} condition {condition_type} is stale "
2070 f"(generation={generation}, observedGeneration={observed})"
2071 )
2072 return f"{context} condition {condition_type} is missing"
2075def _gateway_api_readiness_failure(kind: str, resource: dict[str, Any]) -> str | None:
2076 """Return strict, generation-aware Gateway API readiness evidence."""
2077 metadata_value = resource.get("metadata")
2078 spec_value = resource.get("spec")
2079 status_value = resource.get("status")
2080 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2081 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2082 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2083 generation = metadata.get("generation")
2084 if not isinstance(generation, int) or isinstance(generation, bool):
2085 return f"invalid metadata.generation ({generation})"
2087 if kind == "GatewayClass":
2088 return _current_condition_failure(
2089 _conditions(status), "Accepted", generation, "GatewayClass"
2090 )
2092 if kind == "Gateway":
2093 for condition_type in ("Accepted", "Programmed"):
2094 failure = _current_condition_failure(
2095 _conditions(status), condition_type, generation, "Gateway"
2096 )
2097 if failure:
2098 return failure
2099 addresses = status.get("addresses", [])
2100 if not isinstance(addresses, list) or not any(
2101 isinstance(address, dict) and str(address.get("value", "")).strip()
2102 for address in addresses
2103 ):
2104 return "Gateway has no address"
2106 intended_listeners: set[str] = {
2107 listener["name"]
2108 for listener in spec.get("listeners", [])
2109 if isinstance(listener, dict) and isinstance(listener.get("name"), str)
2110 }
2111 listener_statuses = status.get("listeners", [])
2112 if not isinstance(listener_statuses, list):
2113 return "Gateway listener status is missing"
2114 by_name = {
2115 listener.get("name"): listener
2116 for listener in listener_statuses
2117 if isinstance(listener, dict) and isinstance(listener.get("name"), str)
2118 }
2119 for listener_name in sorted(intended_listeners):
2120 listener = by_name.get(listener_name)
2121 if not isinstance(listener, dict):
2122 return f"Gateway listener {listener_name!r} status is missing"
2123 listener_conditions = listener.get("conditions", [])
2124 if not isinstance(listener_conditions, list):
2125 return f"Gateway listener {listener_name!r} conditions are missing"
2126 plain_conditions = [item for item in listener_conditions if isinstance(item, dict)]
2127 for condition_type in ("Accepted", "ResolvedRefs", "Programmed"):
2128 failure = _current_condition_failure(
2129 plain_conditions,
2130 condition_type,
2131 generation,
2132 f"Gateway listener {listener_name!r}",
2133 )
2134 if failure:
2135 return failure
2136 return None
2138 if kind == "HTTPRoute":
2139 namespace = str(metadata.get("namespace", "default"))
2141 def parent_key(reference: dict[str, Any]) -> tuple[str, str, str, str, str]:
2142 return (
2143 str(reference.get("group", "gateway.networking.k8s.io")),
2144 str(reference.get("kind", "Gateway")),
2145 str(reference.get("namespace", namespace)),
2146 str(reference.get("name", "")),
2147 str(reference.get("sectionName", "")),
2148 )
2150 intended: set[tuple[str, str, str, str, str]] = {
2151 parent_key(parent) for parent in spec.get("parentRefs", []) if isinstance(parent, dict)
2152 }
2153 parents = status.get("parents", [])
2154 if not isinstance(parents, list):
2155 return "HTTPRoute parent status is missing"
2156 live_by_ref = {
2157 parent_key(parent["parentRef"]): parent
2158 for parent in parents
2159 if isinstance(parent, dict) and isinstance(parent.get("parentRef"), dict)
2160 }
2161 for reference in sorted(intended):
2162 parent = live_by_ref.get(reference)
2163 if not isinstance(parent, dict):
2164 return f"HTTPRoute intended parent {reference} status is missing"
2165 parent_conditions = parent.get("conditions", [])
2166 if not isinstance(parent_conditions, list):
2167 return f"HTTPRoute intended parent {reference} conditions are missing"
2168 plain_conditions = [item for item in parent_conditions if isinstance(item, dict)]
2169 for condition_type in ("Accepted", "ResolvedRefs"):
2170 failure = _current_condition_failure(
2171 plain_conditions,
2172 condition_type,
2173 generation,
2174 f"HTTPRoute parent {reference}",
2175 )
2176 if failure:
2177 return failure
2178 return None
2180 return None
2183def _resource_readiness_failure(kind: str, resource: dict[str, Any]) -> str | None:
2184 """Return an actionable readiness reason, or None when the object is ready."""
2185 metadata_value = resource.get("metadata")
2186 spec_value = resource.get("spec")
2187 status_value = resource.get("status")
2188 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2189 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2190 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2192 if metadata.get("deletionTimestamp"):
2193 return f"object is terminating since {metadata['deletionTimestamp']}"
2195 for condition in _conditions(status):
2196 if condition.get("type") in {"Ready", "Available"} and _condition_matches(
2197 condition, "False"
2198 ):
2199 detail = condition.get("message") or condition.get("reason") or "no detail"
2200 return f"condition {condition.get('type')} is False ({detail})"
2202 if kind in {"GatewayClass", "Gateway", "HTTPRoute"}:
2203 return _gateway_api_readiness_failure(kind, resource)
2205 if kind in {"Deployment", "StatefulSet"}:
2206 generation_failure = _generation_failure(resource)
2207 if generation_failure:
2208 return generation_failure
2209 desired_value = spec.get("replicas", 1)
2210 if not isinstance(desired_value, int) or isinstance(desired_value, bool):
2211 return f"invalid desired replica count ({desired_value})"
2212 desired = desired_value
2213 fields = (
2214 ("replicas", "updatedReplicas", "readyReplicas", "availableReplicas")
2215 if kind == "Deployment"
2216 else ("currentReplicas", "updatedReplicas", "readyReplicas")
2217 )
2218 return _replica_failure(status, desired, fields)
2220 if kind == "DaemonSet":
2221 generation_failure = _generation_failure(resource)
2222 if generation_failure:
2223 return generation_failure
2224 desired_value = status.get("desiredNumberScheduled")
2225 if not isinstance(desired_value, int) or isinstance(desired_value, bool):
2226 return f"invalid desiredNumberScheduled ({desired_value})"
2227 desired = desired_value
2228 replica_failure = _replica_failure(
2229 status,
2230 desired,
2231 (
2232 "currentNumberScheduled",
2233 "updatedNumberScheduled",
2234 "numberReady",
2235 "numberAvailable",
2236 ),
2237 )
2238 if replica_failure:
2239 return replica_failure
2240 misscheduled = status.get("numberMisscheduled", 0)
2241 if not isinstance(misscheduled, int) or isinstance(misscheduled, bool) or misscheduled != 0:
2242 return f"pods are misscheduled (numberMisscheduled={misscheduled})"
2243 return None
2245 if kind == "Job":
2246 failed = _required_condition_failure(status, "Failed")
2247 if failed is None:
2248 return "condition Failed is True"
2249 return _required_condition_failure(status, "Complete")
2251 if kind == "Pod":
2252 return _required_condition_failure(status, "Ready")
2254 if kind == "PersistentVolumeClaim":
2255 phase = status.get("phase")
2256 return None if phase == "Bound" else f"PVC phase is {phase!r}, expected 'Bound'"
2258 if kind == "PersistentVolume":
2259 phase = status.get("phase")
2260 return (
2261 None
2262 if phase in {"Bound", "Available"}
2263 else f"PV phase is {phase!r}, expected 'Bound' or 'Available'"
2264 )
2266 if kind == "CustomResourceDefinition":
2267 return _required_condition_failure(status, "Established")
2269 if kind == "APIService":
2270 return _required_condition_failure(status, "Available")
2272 if kind == "HorizontalPodAutoscaler":
2273 for condition_type in ("AbleToScale", "ScalingActive"):
2274 failure = _required_condition_failure(status, condition_type)
2275 if failure:
2276 return failure
2277 return None
2279 if kind == "PodDisruptionBudget":
2280 generation_failure = _generation_failure(resource)
2281 if generation_failure:
2282 return generation_failure
2283 current_healthy = status.get("currentHealthy")
2284 desired_healthy = status.get("desiredHealthy")
2285 if (
2286 not isinstance(current_healthy, int)
2287 or isinstance(current_healthy, bool)
2288 or not isinstance(desired_healthy, int)
2289 or isinstance(desired_healthy, bool)
2290 ):
2291 return (
2292 "invalid PDB health "
2293 f"(currentHealthy={current_healthy}, desiredHealthy={desired_healthy})"
2294 )
2295 if current_healthy < desired_healthy:
2296 return (
2297 "PDB health below target "
2298 f"(currentHealthy={current_healthy}, desiredHealthy={desired_healthy})"
2299 )
2300 return None
2302 if kind in {"Certificate", "Issuer"}:
2303 generation = metadata.get("generation")
2304 if not isinstance(generation, int) or isinstance(generation, bool):
2305 return f"invalid metadata.generation ({generation})"
2306 return _current_condition_failure(
2307 _conditions(status),
2308 "Ready",
2309 generation,
2310 kind,
2311 )
2313 if kind in {"NodePool", "EC2NodeClass", "ScaledJob", "ScaledObject"}:
2314 return _required_condition_failure(status, "Ready")
2316 # Static configuration and RBAC resources have no rollout contract. Their
2317 # exact-object existence is sufficient unless a generic Ready/Available
2318 # condition explicitly reported False above.
2319 return None
2322def _dynamic_resource(
2323 dynamic_client: Any,
2324 cache: dict[tuple[str, str], Any],
2325 api_version: str,
2326 kind: str,
2327) -> Any:
2328 key = (api_version, kind)
2329 if key not in cache:
2330 cache[key] = dynamic_client.resources.get(api_version=api_version, kind=kind)
2331 return cache[key]
2334def _service_endpoint_failure(
2335 dynamic_client: Any,
2336 cache: dict[tuple[str, str], Any],
2337 planned_resource: dict[str, Any],
2338) -> str | None:
2339 document_value = planned_resource["document"]
2340 document: dict[str, Any] = document_value if isinstance(document_value, dict) else {}
2341 metadata_value = document.get("metadata")
2342 document_metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2343 annotations_value = document_metadata.get("annotations")
2344 annotations: dict[str, Any] = annotations_value if isinstance(annotations_value, dict) else {}
2345 if str(annotations.get(_ALLOW_EMPTY_ENDPOINTS_ANNOTATION)).lower() == "true":
2346 # Services backing accelerator-scheduled DaemonSets (for example the
2347 # DCGM exporter) legitimately have zero endpoints until the first GPU
2348 # node is provisioned; existence is their readiness contract.
2349 return None
2350 spec_value = document.get("spec")
2351 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2352 selector = spec.get("selector")
2353 if not isinstance(selector, dict) or not selector:
2354 return None
2356 endpoint_slices = _dynamic_resource(
2357 dynamic_client,
2358 cache,
2359 "discovery.k8s.io/v1",
2360 "EndpointSlice",
2361 )
2362 response = endpoint_slices.get(
2363 namespace=planned_resource["namespace"],
2364 label_selector=f"kubernetes.io/service-name={planned_resource['name']}",
2365 )
2366 items = _as_plain_dict(response).get("items", [])
2367 if not isinstance(items, list):
2368 return "EndpointSlice response does not contain an items list"
2369 for item in items:
2370 item_mapping = _as_plain_dict(item)
2371 metadata_value = item_mapping.get("metadata")
2372 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2373 if metadata.get("deletionTimestamp"):
2374 continue
2375 endpoints = item_mapping.get("endpoints", [])
2376 if not isinstance(endpoints, list):
2377 continue
2378 for endpoint in endpoints:
2379 if not isinstance(endpoint, dict):
2380 continue
2381 conditions = endpoint.get("conditions", {})
2382 if not isinstance(conditions, dict):
2383 continue
2384 if conditions.get("ready") is True and conditions.get("terminating") is not True:
2385 return None
2386 return "selector-backed Service has no ready, nonterminating EndpointSlice endpoint"
2389def _certificate_secret_failure(
2390 dynamic_client: Any,
2391 cache: dict[tuple[str, str], Any],
2392 planned_resource: dict[str, Any],
2393 live_resource: dict[str, Any],
2394) -> str | None:
2395 """Require cert-manager's referenced Secret to contain a nonempty TLS keypair."""
2396 spec_value = live_resource.get("spec")
2397 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2398 secret_name = spec.get("secretName")
2399 if not isinstance(secret_name, str) or not secret_name:
2400 return "Certificate spec.secretName is missing"
2401 secret_api = _dynamic_resource(dynamic_client, cache, "v1", "Secret")
2402 response = secret_api.get(
2403 namespace=planned_resource["namespace"],
2404 name=secret_name,
2405 )
2406 secret = _as_plain_dict(response)
2407 data_value = secret.get("data")
2408 data: dict[str, Any] = data_value if isinstance(data_value, dict) else {}
2409 missing = [
2410 key for key in ("tls.crt", "tls.key") if not isinstance(data.get(key), str) or not data[key]
2411 ]
2412 if missing:
2413 return f"Certificate Secret {secret_name!r} has no nonempty {', '.join(missing)}"
2414 return None
2417def _manifest_identity_label(resource: dict[str, Any]) -> str:
2418 return (
2419 f"{resource['apiVersion']}/{resource['kind']}/"
2420 f"{resource['namespace']}/{resource['name']}"
2421 f" [{resource['phase']}:{resource['sourceFile']}]"
2422 )
2425def validate_manifests(
2426 cluster_name: str,
2427 region: str,
2428 manifests_dir: str,
2429 replacements: dict[str, str],
2430 deployment_token: Any = None,
2431) -> dict[str, Any]:
2432 """Validate exact existence and readiness for the complete planned inventory."""
2433 plan = plan_manifests(manifests_dir, replacements)
2434 expected = plan["phases"]["base"] + plan["phases"]["post-helm"]
2435 expected_resources = [_public_manifest_identity(item) for item in expected]
2436 phase_counts = {
2437 phase: {"ExpectedCount": len(plan["phases"][phase]), "ValidatedCount": 0}
2438 for phase in ("base", "post-helm")
2439 }
2441 configure_k8s_client(cluster_name, region)
2442 dynamic_client = dynamic.DynamicClient(client.ApiClient())
2443 resource_cache: dict[tuple[str, str], Any] = {}
2444 validated_resources: list[dict[str, str]] = []
2445 failures: list[str] = []
2446 failure_count = 0
2448 def add_failure(message: str) -> None:
2449 nonlocal failure_count
2450 failure_count += 1
2451 if len(failures) < _MAX_VALIDATION_FAILURES:
2452 failures.append(message[:500])
2454 for planned_resource in expected:
2455 label = _manifest_identity_label(planned_resource)
2456 try:
2457 resource_api = _dynamic_resource(
2458 dynamic_client,
2459 resource_cache,
2460 planned_resource["apiVersion"],
2461 planned_resource["kind"],
2462 )
2463 get_kwargs: dict[str, Any] = {"name": planned_resource["name"]}
2464 if planned_resource["namespace"] != _CLUSTER_SCOPE:
2465 get_kwargs["namespace"] = planned_resource["namespace"]
2466 live_resource = _as_plain_dict(resource_api.get(**get_kwargs))
2467 failure = _resource_readiness_failure(planned_resource["kind"], live_resource)
2468 if failure is None and planned_resource["kind"] == "Service":
2469 failure = _service_endpoint_failure(
2470 dynamic_client,
2471 resource_cache,
2472 planned_resource,
2473 )
2474 if failure is None and planned_resource["kind"] == "Certificate":
2475 failure = _certificate_secret_failure(
2476 dynamic_client,
2477 resource_cache,
2478 planned_resource,
2479 live_resource,
2480 )
2481 if failure:
2482 add_failure(f"{label}: {failure}")
2483 continue
2484 except (NotFoundError, ResourceNotFoundError) as exc:
2485 add_failure(f"{label}: object or API resource not found ({exc})")
2486 continue
2487 except ApiException as exc:
2488 detail = exc.reason or str(exc)
2489 add_failure(f"{label}: Kubernetes API error {exc.status} ({detail})")
2490 continue
2491 except Exception as exc:
2492 add_failure(f"{label}: validation error ({exc})")
2493 continue
2495 identity = _public_manifest_identity(planned_resource)
2496 validated_resources.append(identity)
2497 phase_counts[planned_resource["phase"]]["ValidatedCount"] += 1
2499 if failure_count:
2500 hidden = failure_count - len(failures)
2501 suffix = f"; ... {hidden} additional failure(s)" if hidden else ""
2502 raise RuntimeError(
2503 f"Manifest validation failed: validated={len(validated_resources)} "
2504 f"expected={len(expected)}; " + "; ".join(failures) + suffix
2505 )
2507 return {
2508 "status": "validated",
2509 "DeploymentToken": deployment_token,
2510 "ExpectedCount": len(expected),
2511 "ValidatedCount": len(validated_resources),
2512 "BaseExpectedCount": phase_counts["base"]["ExpectedCount"],
2513 "BaseValidatedCount": phase_counts["base"]["ValidatedCount"],
2514 "PostHelmExpectedCount": phase_counts["post-helm"]["ExpectedCount"],
2515 "PostHelmValidatedCount": phase_counts["post-helm"]["ValidatedCount"],
2516 "PhaseCounts": phase_counts,
2517 "ExpectedResources": expected_resources,
2518 "ValidatedResources": validated_resources,
2519 }
2522def _record_phase_status(phase: str, status: str, message: str) -> None:
2523 """Record a convergence phase's outcome to SSM (best-effort).
2525 Mirrors the helm worker's per-chart status (``_record_addon_status``) so
2526 ``gco stacks addons status`` surfaces the base and post-Helm apply passes
2527 alongside the charts. Writes ``/<project>/addons/<region>/<phase>`` as a
2528 small JSON blob. Failures are swallowed — status reporting must never turn a
2529 successful apply into a failure (or vice versa). Reads PROJECT_NAME / REGION
2530 from the Lambda environment; a no-op if either is unset.
2531 """
2532 project = os.environ.get("PROJECT_NAME")
2533 region = os.environ.get("REGION")
2534 if not project or not region:
2535 return
2536 import contextlib
2537 import time as _time
2539 with contextlib.suppress(Exception):
2540 boto3.client("ssm").put_parameter(
2541 Name=f"/{project}/addons/{region}/{phase}",
2542 Value=json.dumps(
2543 {
2544 "phase": phase,
2545 "status": status,
2546 "message": message[:1024],
2547 "updated_at": int(_time.time()),
2548 }
2549 ),
2550 Type="String",
2551 Overwrite=True,
2552 )
2555def handle_task(event: dict[str, Any]) -> dict[str, Any]:
2556 """Run an apply or exhaustive manifest-validation Step Functions task."""
2557 cluster_name = event["ClusterName"]
2558 region = event["Region"]
2559 replacements = event.get("ImageReplacements", {})
2560 action = event.get("Action", "apply_manifests")
2561 manifests_dir = os.path.join(os.path.dirname(__file__), "manifests")
2563 if action == "delete_gateway_resources":
2564 phase = "gateway-teardown"
2565 try:
2566 result = _delete_gateway_resources(cluster_name, region)
2567 except Exception as exc:
2568 _record_phase_status(phase, "failed", str(exc))
2569 raise
2570 _record_phase_status(
2571 phase,
2572 "deleted",
2573 f"deleted={result['DeletedCount']}",
2574 )
2575 return result
2577 if action == "validate_manifests":
2578 phase = "manifest-validation"
2579 deployment_token = event.get("DeploymentToken")
2580 try:
2581 result = validate_manifests(
2582 cluster_name,
2583 region,
2584 manifests_dir,
2585 replacements,
2586 deployment_token,
2587 )
2588 except Exception as exc:
2589 _record_phase_status(phase, "failed", f"token={deployment_token} {exc}")
2590 raise
2591 _record_phase_status(
2592 phase,
2593 "validated",
2594 f"token={deployment_token} validated={result['ValidatedCount']} "
2595 f"expected={result['ExpectedCount']}",
2596 )
2597 return result
2599 if action != "apply_manifests":
2600 raise ValueError(f"Unsupported task action: {action}")
2602 post_helm = str(event.get("PostHelm", "false")).lower() == "true"
2603 phase = "post-helm-manifests" if post_helm else "base-manifests"
2604 try:
2605 result = apply_manifests(cluster_name, region, manifests_dir, replacements, post_helm)
2606 except Exception as exc:
2607 _record_phase_status(phase, "failed", str(exc))
2608 raise
2609 if result.get("FailedCount", 0):
2610 _record_phase_status(phase, "failed", str(result.get("Failed")))
2611 raise RuntimeError(
2612 f"kubectl apply failed (post_helm={post_helm}, "
2613 f"failed={result.get('FailedCount')}): {result.get('Failed')}"
2614 )
2615 if result.get("ExpectedCount") is not None and result.get("AppliedCount") != result.get(
2616 "ExpectedCount"
2617 ):
2618 message = (
2619 f"apply count mismatch: applied={result.get('AppliedCount')} "
2620 f"expected={result.get('ExpectedCount')}"
2621 )
2622 _record_phase_status(phase, "failed", message)
2623 raise RuntimeError(message)
2624 _record_phase_status(
2625 phase,
2626 "applied",
2627 f"applied={result.get('AppliedCount')} expected={result.get('ExpectedCount')} "
2628 f"skipped={result.get('SkippedCount')}",
2629 )
2630 return result
2633def lambda_handler(event: dict[str, Any], context: Any) -> Any:
2634 """Main Lambda handler.
2636 Two entrypoints share this function:
2638 - **Step Functions task** (the convergence pipeline): the event carries an
2639 ``Action`` key and is dispatched to :func:`handle_task`, which applies the
2640 manifests and raises on any failure.
2641 - **CloudFormation custom resource** (legacy/fallback): the event carries a
2642 ``RequestType`` and the result is POSTed back to CloudFormation.
2643 """
2644 if event.get("Action"):
2645 logger.info(f"Task event: {json.dumps(event)}")
2646 return handle_task(event)
2648 print(f"[HANDLER] Received event type: {event.get('RequestType')}")
2649 logger.info(f"Received event: {json.dumps(event)}")
2651 request_type = event["RequestType"]
2652 physical_resource_id = event.get("PhysicalResourceId", f"kubectl-{event['LogicalResourceId']}")
2654 try:
2655 properties = event["ResourceProperties"]
2656 cluster_name = properties["ClusterName"]
2657 region = properties["Region"]
2659 if request_type == "Create" or request_type == "Update":
2660 manifests_dir = os.path.join(os.path.dirname(__file__), "manifests")
2661 replacements = properties.get("ImageReplacements", {})
2662 # PostHelm: "true" means this is the post-Helm pass — apply only post-helm-* files
2663 post_helm = properties.get("PostHelm", "false").lower() == "true"
2664 response_data = apply_manifests(
2665 cluster_name, region, manifests_dir, replacements, post_helm
2666 )
2667 failed_count = int(response_data.get("FailedCount", 0))
2668 prune_failures = response_data.get("PruneFailures")
2669 if failed_count > 0 or prune_failures not in (None, "", "None"):
2670 raise RuntimeError(
2671 "Manifest application reported failures: "
2672 f"failed_count={failed_count}, failed={response_data.get('Failed', 'unknown')}, "
2673 f"prune_failures={prune_failures or 'None'}"
2674 )
2675 send_response(event, context, SUCCESS, response_data, physical_resource_id)
2677 elif request_type == "Delete":
2678 # Always succeed on delete to prevent stack from getting stuck
2679 skip_deletion = properties.get("SkipDeletionOnStackDelete", "false").lower() == "true"
2680 if skip_deletion:
2681 logger.info("Skipping deletion (SkipDeletionOnStackDelete=true)")
2682 response_data = {"Status": "Deleted"}
2683 send_response(event, context, SUCCESS, response_data, physical_resource_id)
2685 else:
2686 # CloudFormation only ever sends Create/Update/Delete. Anything else
2687 # must still get a FAILED callback: silently returning would leave
2688 # the stack waiting on this resource until its timeout.
2689 raise ValueError(f"Unsupported RequestType: {request_type}")
2691 except Exception as e:
2692 logger.error(f"Error: {e}", exc_info=True)
2693 # On delete, always return success to prevent stack from getting stuck
2694 if request_type == "Delete":
2695 send_response(
2696 event,
2697 context,
2698 SUCCESS,
2699 {"Status": "Forced success on delete"},
2700 physical_resource_id,
2701 )
2702 else:
2703 send_response(event, context, FAILED, {}, physical_resource_id, str(e))