Coverage for lambda / kubectl-applier-simple / handler.py: 100.00%
1349 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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-10-28T14:57:12Z
53# Generated from Git commit: fdfe31c154bbf84cdfeb034fee1256e0a838d1f1
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", "v1beta2", "resourceflavors", True),
120 "ClusterQueue": ("kueue.x-k8s.io", "v1beta2", "clusterqueues", True),
121 "LocalQueue": ("kueue.x-k8s.io", "v1beta2", "localqueues", False),
122}
124# cert-manager resources behind GCO's internal PKI
125# (post-helm-api-workload-certificates.yaml): the cluster-scoped bootstrap and
126# CA ClusterIssuers, and the namespaced Certificates they sign (the CA itself
127# and every in-cluster TLS leaf). cert-manager's chart installs the CRDs, so
128# these are POST-HELM kinds; the file sorts before the Gateway resources.
129# Issuer stays supported for namespaced issuers even though GCO ships none.
130_CERT_MANAGER_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
131 "ClusterIssuer": ("cert-manager.io", "v1", "clusterissuers", True),
132 "Issuer": ("cert-manager.io", "v1", "issuers", False),
133 "Certificate": ("cert-manager.io", "v1", "certificates", False),
134}
136# Argo CD objects GCO owns for the self-managed Argo CD (helm.argocd): the
137# fenced AppProject (post-helm-argocd-access.yaml) and the optional root
138# Application (post-helm-argocd-gitops.yaml). The argo-cd chart installs their
139# CRDs, so they belong to the POST-HELM pass. Ordinary namespaced CRs in the
140# ``argocd`` namespace; readiness is exact existence only — a tenant
141# repository that fails to sync must never fail a GCO deployment (``gco gitops
142# status`` and the Argo CD UI surface sync/health instead).
143_ARGOCD_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
144 "AppProject": ("argoproj.io", "v1alpha1", "appprojects", False),
145 "Application": ("argoproj.io", "v1alpha1", "applications", False),
146}
147# The Crossplane composition function GCO installs for the self-managed
148# Crossplane (helm.crossplane; post-helm-crossplane.yaml). Cluster-scoped; the
149# chart installs its CRD, so it is a POST-HELM kind. Readiness is exact
150# existence only — the package pull is Crossplane's to report.
151_CROSSPLANE_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
152 "Function": ("pkg.crossplane.io", "v1", "functions", True),
153}
154# The trust-manager Bundle that publishes the internal CA certificate to
155# MLflow's clients in gco-jobs (post-helm-mlflow-tls.yaml). Cluster-scoped; the
156# trust-manager chart installs its CRD, so it is a POST-HELM kind. Readiness
157# is a Synced condition for the current generation: trust-manager has written
158# the target ConfigMap from the current spec and source.
159_TRUST_MANAGER_CUSTOM_OBJECTS: dict[str, tuple[str, str, str, bool]] = {
160 "Bundle": ("trust.cert-manager.io", "v1alpha1", "bundles", True),
161}
163# Prometheus Operator kinds (monitoring.coreos.com) the applier creates or
164# patches through the custom-objects API, with their resource plurals.
165_MONITORING_CUSTOM_OBJECT_PLURALS: dict[str, str] = {
166 "ServiceMonitor": "servicemonitors",
167 "PodMonitor": "podmonitors",
168 "PrometheusRule": "prometheusrules",
169}
171# A Secret annotated with this marker names (comma-separated) the data keys
172# the applier fills with a fresh random value when it CREATES the Secret. An
173# existing Secret is never overwritten: the generated values are credentials
174# other objects already hold (a database's password), so a later apply only
175# adds keys the Secret is missing. The manifest itself carries no data for
176# those keys, so no credential ever appears in the repository, the Step
177# Functions input or the applier's logs. Used for the Slurm accounting
178# database (29-slurm-accounting-db.yaml).
179_GENERATE_SECRET_KEYS_ANNOTATION = "gco.io/generate-keys" # nosec B105 # annotation key
180_GENERATED_SECRET_BYTES = 32
182# Services annotated with this marker are validated for exact existence only;
183# a ready EndpointSlice endpoint is not required. Reserved for Services whose
184# backends schedule exclusively onto accelerator nodes that a fresh cluster
185# does not have yet (for example the DCGM exporter DaemonSet).
186_ALLOW_EMPTY_ENDPOINTS_ANNOTATION = "gco.io/allow-empty-endpoints"
188# This is the authoritative set of kinds the applier knows how to create or
189# patch. Planning rejects anything else before the first Kubernetes mutation,
190# which prevents a newly added raw manifest from being silently ignored.
191_SUPPORTED_MANIFEST_KINDS = frozenset(
192 {
193 "APIService",
194 "AppProject",
195 "Application",
196 "Bundle",
197 "Certificate",
198 "ClusterIssuer",
199 "ClusterRole",
200 "ClusterRoleBinding",
201 "ClusterTrainingRuntime",
202 "ConfigMap",
203 "CronJob",
204 "CustomResourceDefinition",
205 "DaemonSet",
206 "Deployment",
207 "DeviceClass",
208 "EC2NodeClass",
209 "Function",
210 "Gateway",
211 "GatewayClass",
212 "HTTPRoute",
213 "HorizontalPodAutoscaler",
214 "Issuer",
215 "Job",
216 "Lease",
217 "LimitRange",
218 "LoadBalancerConfiguration",
219 "Namespace",
220 "NetworkPolicy",
221 "NodePool",
222 "PersistentVolume",
223 "PersistentVolumeClaim",
224 "Pod",
225 "ClusterQueue",
226 "LocalQueue",
227 "PodDisruptionBudget",
228 "PodMonitor",
229 "PriorityClass",
230 "PrometheusRule",
231 "ResourceFlavor",
232 "ResourceQuota",
233 "Role",
234 "RoleBinding",
235 "ScaledJob",
236 "ScaledObject",
237 "Secret",
238 "Service",
239 "ServiceAccount",
240 "ServiceMonitor",
241 "StatefulSet",
242 "StorageClass",
243 "TargetGroupConfiguration",
244 "Token",
245 "ValidatingAdmissionPolicy",
246 "ValidatingAdmissionPolicyBinding",
247 }
248)
249_CLUSTER_SCOPED_KINDS = frozenset(
250 {
251 "APIService",
252 "Bundle",
253 "ClusterIssuer",
254 "ClusterQueue",
255 "ClusterRole",
256 "ClusterRoleBinding",
257 "ClusterTrainingRuntime",
258 "CustomResourceDefinition",
259 "DeviceClass",
260 "EC2NodeClass",
261 "Function",
262 "GatewayClass",
263 "Namespace",
264 "NodePool",
265 "PersistentVolume",
266 "PriorityClass",
267 "ResourceFlavor",
268 "StorageClass",
269 "ValidatingAdmissionPolicy",
270 "ValidatingAdmissionPolicyBinding",
271 }
272)
273_IDENTITY_FIELDS = ("apiVersion", "kind", "namespace", "name", "sourceFile", "phase")
276def _public_manifest_identity(resource: dict[str, Any]) -> dict[str, str]:
277 """Return the serializable, normalized identity for a planned resource."""
278 return {field: resource[field] for field in _IDENTITY_FIELDS}
281def _planning_error_message(errors: list[str], failure_count: int) -> str:
282 """Build a bounded planning failure with enough source context to act on."""
283 hidden = failure_count - len(errors)
284 suffix = f"; ... {hidden} additional error(s)" if hidden else ""
285 return "Manifest planning failed: " + "; ".join(errors) + suffix
288# ---------------------------------------------------------------------------
289# Cross-phase ServiceAccount/token-projection consistency guard.
290#
291# The 2026-08 SQS submission-path outage: hardening a ServiceAccount with
292# ``automountServiceAccountToken: false`` (base phase) and projecting the
293# compensating kubernetes-audience token onto its workload (post-Helm phase)
294# are one logical change split across two apply invocations. A redeploy that
295# ran only the base pass left the live queue-processor pod with no way to
296# authenticate to the Kubernetes API, crash-looping before its first SQS
297# receive. The two files cannot apply in one transaction (the ScaledJob needs
298# KEDA CRDs that do not exist in the base phase), so the planner enforces the
299# pairing instead — and because BOTH phases are always planned before either
300# is applied, a violation fails the base pass too.
302# Standard in-cluster credential mount point the kubernetes client reads.
303_KUBE_SERVICEACCOUNT_MOUNT = "/var/run/secrets/kubernetes.io/serviceaccount"
305# Where each workload kind keeps its pod spec.
306_POD_SPEC_PATHS: dict[str, tuple[str, ...]] = {
307 "Pod": ("spec",),
308 "Deployment": ("spec", "template", "spec"),
309 "StatefulSet": ("spec", "template", "spec"),
310 "DaemonSet": ("spec", "template", "spec"),
311 "Job": ("spec", "template", "spec"),
312 "CronJob": ("spec", "jobTemplate", "spec", "template", "spec"),
313 "ScaledJob": ("spec", "jobTargetRef", "template", "spec"),
314}
317def _planned_pod_spec(document: dict[str, Any]) -> dict[str, Any] | None:
318 """Return the pod spec embedded in a planned workload document, if any."""
319 path = _POD_SPEC_PATHS.get(str(document.get("kind")))
320 if path is None:
321 return None
322 node: Any = document
323 for key in path:
324 if not isinstance(node, dict):
325 return None
326 node = node.get(key)
327 return node if isinstance(node, dict) else None
330def _pod_spec_projects_service_account_token(pod_spec: dict[str, Any]) -> bool:
331 """True when a projected serviceAccountToken is mounted at the standard path."""
332 token_volumes: set[str] = set()
333 for volume in pod_spec.get("volumes") or []:
334 if not isinstance(volume, dict):
335 continue
336 projected = volume.get("projected")
337 if not isinstance(projected, dict):
338 continue
339 sources = projected.get("sources") or []
340 has_token_source = any(
341 isinstance(source, dict) and "serviceAccountToken" in source for source in sources
342 )
343 if has_token_source and isinstance(volume.get("name"), str):
344 token_volumes.add(volume["name"])
345 if not token_volumes:
346 return False
347 for containers_key in ("containers", "initContainers", "ephemeralContainers"):
348 containers = pod_spec.get(containers_key) or []
349 if not isinstance(containers, list):
350 continue
351 for container in containers:
352 if not isinstance(container, dict):
353 continue
354 for mount in container.get("volumeMounts") or []:
355 if (
356 isinstance(mount, dict)
357 and mount.get("mountPath") == _KUBE_SERVICEACCOUNT_MOUNT
358 and mount.get("name") in token_volumes
359 ):
360 return True
361 return False
364def _automount_disabled_service_accounts(
365 planned: list[dict[str, Any]],
366) -> dict[tuple[str, str], str]:
367 """Planned ServiceAccounts with automount disabled, keyed by (namespace, name)."""
368 disabled: dict[tuple[str, str], str] = {}
369 for item in planned:
370 if item["kind"] != "ServiceAccount":
371 continue
372 if item["document"].get("automountServiceAccountToken") is False:
373 disabled[(item["namespace"], item["name"])] = item["sourceFile"]
374 return disabled
377def _rbac_bound_service_accounts(planned: list[dict[str, Any]]) -> set[tuple[str, str]]:
378 """ServiceAccounts referenced as subjects by planned (Cluster)RoleBindings.
380 An RBAC binding is the planned inventory's own declaration that the
381 ServiceAccount is expected to call the Kubernetes API. Accounts with
382 automount disabled and NO binding (workload identities that only hold
383 AWS credentials, e.g. the inference proxy) are deliberately exempt from
384 the token-projection invariant.
385 """
386 bound: set[tuple[str, str]] = set()
387 for item in planned:
388 if item["kind"] not in ("RoleBinding", "ClusterRoleBinding"):
389 continue
390 default_namespace = item["namespace"] if item["kind"] == "RoleBinding" else None
391 for subject in item["document"].get("subjects") or []:
392 if not isinstance(subject, dict) or subject.get("kind") != "ServiceAccount":
393 continue
394 name = subject.get("name")
395 namespace = subject.get("namespace") or default_namespace
396 if isinstance(name, str) and isinstance(namespace, str):
397 bound.add((namespace, name))
398 return bound
401def _service_account_token_projection_errors(
402 phases: dict[str, list[dict[str, Any]]],
403) -> list[str]:
404 """Cross-phase invariant: hardened, API-bound SAs need a projected token.
406 For every planned ServiceAccount (either phase) that sets
407 ``automountServiceAccountToken: false`` AND is bound to the Kubernetes
408 API by a planned RoleBinding/ClusterRoleBinding, every planned workload
409 whose pod spec runs as that account must mount a projected
410 serviceAccountToken at ``/var/run/secrets/kubernetes.io/serviceaccount``.
411 """
412 planned = phases["base"] + phases["post-helm"]
413 disabled = _automount_disabled_service_accounts(planned)
414 if not disabled:
415 return []
416 bound = _rbac_bound_service_accounts(planned)
417 guarded = {identity: source for identity, source in disabled.items() if identity in bound}
418 if not guarded:
419 return []
421 errors: list[str] = []
422 for item in planned:
423 pod_spec = _planned_pod_spec(item["document"])
424 if pod_spec is None:
425 continue
426 service_account = pod_spec.get("serviceAccountName")
427 if not isinstance(service_account, str) or not service_account:
428 continue
429 source_file = guarded.get((item["namespace"], service_account))
430 if source_file is None:
431 continue
432 if _pod_spec_projects_service_account_token(pod_spec):
433 continue
434 errors.append(
435 f"{item['sourceFile']}: {item['kind']}/{item['namespace']}/{item['name']} runs as "
436 f"ServiceAccount {service_account!r} ({source_file}), which sets "
437 "automountServiceAccountToken: false and is RBAC-bound to the Kubernetes API, "
438 f"but mounts no projected serviceAccountToken at {_KUBE_SERVICEACCOUNT_MOUNT} - "
439 "the pod would have no credentials. Project the token (see the "
440 "kubernetes-api-token volume in post-helm-sqs-consumer.yaml) or ship both "
441 "halves of the hardening in one release"
442 )
443 return errors
446def _log_service_account_automount_flip(
447 v1: Any,
448 document: dict[str, Any],
449 namespace: str,
450 name: str,
451 plan: dict[str, Any],
452) -> None:
453 """Name the workloads affected when automount is being flipped to false.
455 Clusters created by an older release are supported: their live
456 ServiceAccount may still automount tokens while the incoming manifest
457 disables it. This apply is the exact moment previously-running pods
458 lose ambient credentials, so log every planned workload that references
459 the account and whether each already carries the compensating projected
460 token — the diagnostic that would have named the queue processor on the
461 redeploy that caused the SQS outage. Best-effort: any read failure
462 skips the diagnostic, never the apply.
463 """
464 if document.get("automountServiceAccountToken") is not False:
465 return
466 try:
467 live = v1.read_namespaced_service_account(name, namespace)
468 except ApiException as e:
469 if e.status != 404:
470 logger.debug(
471 "Automount-flip check could not read live ServiceAccount %s/%s: %s",
472 namespace,
473 name,
474 e,
475 )
476 return
477 except Exception as e: # pragma: no cover - defensive; diagnostics never block
478 logger.debug("Automount-flip check failed for ServiceAccount %s/%s: %s", namespace, name, e)
479 return
480 if getattr(live, "automount_service_account_token", None) is False:
481 return # Already hardened on the live cluster; nothing is flipping.
483 references: list[str] = []
484 for item in plan["phases"]["base"] + plan["phases"]["post-helm"]:
485 pod_spec = _planned_pod_spec(item["document"])
486 if (
487 pod_spec is None
488 or item["namespace"] != namespace
489 or pod_spec.get("serviceAccountName") != name
490 ):
491 continue
492 projected = _pod_spec_projects_service_account_token(pod_spec)
493 references.append(
494 f"{item['phase']}:{item['sourceFile']} {item['kind']}/{item['name']} "
495 f"projected-token={'present' if projected else 'MISSING'}"
496 )
497 logger.warning(
498 "ServiceAccount %s/%s: automountServiceAccountToken is being flipped to false on a "
499 "live cluster; planned workloads running as it: %s",
500 namespace,
501 name,
502 "; ".join(references) or "<none>",
503 )
506def plan_manifests(
507 manifests_dir: str,
508 replacements: object,
509) -> dict[str, Any]:
510 """Plan the complete raw-manifest inventory without mutating the cluster.
512 Files are scanned in lexical order. Replacements are literal string
513 replacements, then an unresolved UPPER_SNAKE placeholder gates the entire
514 file out. Every remaining nonempty YAML document must have an exact,
515 supported identity and be unique across both apply phases.
517 ``replacements`` is the ``ImageReplacements`` value exactly as it
518 arrived in the CloudFormation / Step Functions event, so it is typed
519 ``object`` and validated here: anything but a string-to-string mapping
520 is reported as a planning failure rather than trusted.
521 """
522 phases: dict[str, list[dict[str, Any]]] = {"base": [], "post-helm": []}
523 skipped: dict[str, list[str]] = {"base": [], "post-helm": []}
524 feature_gates: dict[str, set[str]] = {"base": set(), "post-helm": set()}
525 errors: list[str] = []
526 failure_count = 0
527 seen: dict[tuple[str, str, str, str], str] = {}
529 def add_error(message: str) -> None:
530 nonlocal failure_count
531 failure_count += 1
532 if len(errors) < _MAX_PLANNING_FAILURES:
533 errors.append(message[:500])
535 replacement_items: list[tuple[str, str]] = []
536 if not isinstance(replacements, dict):
537 add_error("ImageReplacements must be a string-to-string mapping")
538 else:
539 for key, value in replacements.items():
540 if not isinstance(key, str) or not key:
541 add_error("ImageReplacements contains an empty or non-string key")
542 continue
543 if not isinstance(value, str):
544 add_error(f"ImageReplacements value for {key!r} is not a string")
545 continue
546 replacement_items.append((key, value))
548 if failure_count:
549 raise ValueError(_planning_error_message(errors, failure_count))
551 for filename in sorted(os.listdir(manifests_dir)):
552 if not filename.endswith((".yaml", ".yml")):
553 continue
555 phase = "post-helm" if filename.startswith(_POST_HELM_PREFIX) else "base"
556 if phase == "post-helm":
557 skipped["base"].append(f"{filename}:deferred-to-post-helm")
559 filepath = os.path.join(manifests_dir, filename)
560 try:
561 with open(filepath, encoding="utf-8") as manifest_file:
562 content = manifest_file.read()
563 except OSError as exc:
564 add_error(f"{filename}: unable to read manifest: {exc}")
565 continue
567 for key, value in replacement_items:
568 content = content.replace(key, value)
570 unresolved = sorted(set(_UNRESOLVED_PLACEHOLDER_RE.findall(content)))
571 if unresolved:
572 skipped[phase].append(f"{filename}:unreplaced-placeholders")
573 feature_gates[phase].update(unresolved)
574 logger.info(
575 "Planning excludes %s - unresolved feature placeholder(s): %s",
576 filename,
577 ", ".join(unresolved),
578 )
579 continue
581 try:
582 documents = list(yaml.safe_load_all(content))
583 except yaml.YAMLError as exc:
584 add_error(f"{filename}: invalid YAML: {exc}")
585 continue
587 for document_index, document in enumerate(documents, start=1):
588 if document is None:
589 continue
590 location = f"{filename} document {document_index}"
591 if not isinstance(document, dict):
592 add_error(f"{location}: document must be a mapping")
593 continue
595 api_version = document.get("apiVersion")
596 kind = document.get("kind")
597 metadata = document.get("metadata")
598 if not isinstance(api_version, str) or not api_version.strip():
599 add_error(f"{location}: apiVersion must be a nonempty string")
600 continue
601 if not isinstance(kind, str) or not kind.strip():
602 add_error(f"{location}: kind must be a nonempty string")
603 continue
604 api_version = api_version.strip()
605 kind = kind.strip()
606 if kind not in _SUPPORTED_MANIFEST_KINDS:
607 add_error(f"{location}: unsupported kind {kind!r}")
608 continue
609 if not isinstance(metadata, dict):
610 add_error(f"{location}: metadata must be a mapping")
611 continue
613 name = metadata.get("name")
614 if not isinstance(name, str) or not name.strip():
615 add_error(f"{location}: metadata.name must be a nonempty string")
616 continue
617 name = name.strip()
619 if kind in _CLUSTER_SCOPED_KINDS:
620 namespace = _CLUSTER_SCOPE
621 else:
622 namespace_value = metadata.get("namespace", "default")
623 if not isinstance(namespace_value, str) or not namespace_value.strip():
624 add_error(
625 f"{location}: metadata.namespace must be a nonempty string when present"
626 )
627 continue
628 namespace = namespace_value.strip()
630 duplicate_key = (api_version, kind, namespace, name)
631 previous = seen.get(duplicate_key)
632 if previous is not None:
633 add_error(
634 f"{location}: duplicate {api_version}/{kind}/{namespace}/{name}; "
635 f"first declared in {previous}"
636 )
637 continue
638 seen[duplicate_key] = location
640 phases[phase].append(
641 {
642 "apiVersion": api_version,
643 "kind": kind,
644 "namespace": namespace,
645 "name": name,
646 "sourceFile": filename,
647 "phase": phase,
648 "document": document,
649 }
650 )
652 # Cross-phase consistency: a hardened ServiceAccount and the token
653 # projection that compensates for it must ship together. Runs over BOTH
654 # planned phases, so the base pass fails before its first Kubernetes
655 # mutation even when the violation lives in a post-Helm file.
656 for message in _service_account_token_projection_errors(phases):
657 add_error(message)
659 if failure_count:
660 raise ValueError(_planning_error_message(errors, failure_count))
662 return {
663 "phases": phases,
664 "skipped": skipped,
665 "featureGates": {
666 phase: sorted(placeholders) for phase, placeholders in feature_gates.items()
667 },
668 }
671def _hpa_owns_replicas(document: dict[str, Any]) -> bool:
672 """Whether a Deployment manifest hands its replica count to an HPA."""
673 metadata = document.get("metadata")
674 annotations = metadata.get("annotations") if isinstance(metadata, dict) else None
675 return (
676 isinstance(annotations, dict)
677 and annotations.get(_HPA_REPLICA_OWNERSHIP_ANNOTATION) == "true"
678 )
681def _generated_secret_keys(document: dict[str, Any]) -> list[str]:
682 """The data keys a Secret manifest asks the applier to generate, if any.
684 Read from the ``gco.io/generate-keys`` annotation (comma-separated). A
685 manifest that lists a key AND ships a value for it is contradictory and
686 fails planning-time-equivalent here, before any Kubernetes call.
687 """
688 metadata = document.get("metadata")
689 annotations = metadata.get("annotations") if isinstance(metadata, dict) else None
690 raw = (
691 annotations.get(_GENERATE_SECRET_KEYS_ANNOTATION) if isinstance(annotations, dict) else None
692 )
693 if not isinstance(raw, str) or not raw.strip():
694 return []
695 keys = [part.strip() for part in raw.split(",") if part.strip()]
696 shipped: set[str] = set()
697 for field in ("data", "stringData"):
698 values = document.get(field)
699 if isinstance(values, dict):
700 shipped.update(str(key) for key in values)
701 conflicting = sorted(set(keys) & shipped)
702 if conflicting:
703 name = metadata.get("name") if isinstance(metadata, dict) else None
704 raise ValueError(
705 f"Secret {name!r} lists generated key(s) {conflicting} in "
706 f"{_GENERATE_SECRET_KEYS_ANNOTATION} but also ships a value for them"
707 )
708 return keys
711def _apply_generated_secret(
712 v1: Any,
713 document: dict[str, Any],
714 namespace: str,
715 name: str,
716 generated_keys: list[str],
717) -> None:
718 """Create a Secret with fresh random values for its generated keys.
720 When the Secret already exists only the generated keys it lacks are
721 added, never replacing a value: the consumers (the Slurm accounting
722 database and slurmdbd) hold the current one, and rotating it is an
723 operator action, not a side effect of a redeploy. Values are URL-safe
724 random strings of ``_GENERATED_SECRET_BYTES`` bytes of entropy and are
725 never logged.
726 """
727 import secrets
729 try:
730 existing = v1.read_namespaced_secret(name, namespace)
731 except ApiException as e:
732 if e.status != 404:
733 raise
734 existing = None
736 # The three log lines below carry a targeted `nosemgrep`: the
737 # logger-credential-disclosure rule matches the word "Secret" in the
738 # message, but only the Secret's namespace/name and a key *count* are
739 # logged. The generated values go straight into the API call and are
740 # never bound to a name this function could log.
741 if existing is None:
742 body = copy.deepcopy(document)
743 string_data = body.setdefault("stringData", {})
744 for key in generated_keys:
745 string_data[key] = secrets.token_urlsafe(_GENERATED_SECRET_BYTES)
746 v1.create_namespaced_secret(namespace, body=body)
747 # nosemgrep: python-logger-credential-disclosure
748 logger.info(
749 "Generated %d key(s) for new Secret %s/%s", len(generated_keys), namespace, name
750 )
751 return
753 present = set((getattr(existing, "data", None) or {}).keys())
754 missing = [key for key in generated_keys if key not in present]
755 if not missing:
756 # nosemgrep: python-logger-credential-disclosure
757 logger.info(
758 "Secret %s/%s already carries its generated keys; left unchanged", namespace, name
759 )
760 return
761 patch = {"stringData": {key: secrets.token_urlsafe(_GENERATED_SECRET_BYTES) for key in missing}}
762 v1.patch_namespaced_secret(name, namespace, body=patch)
763 # nosemgrep: python-logger-credential-disclosure
764 logger.info("Added %d missing generated key(s) to Secret %s/%s", len(missing), namespace, name)
767def _deployment_patch_body(document: dict[str, Any]) -> dict[str, Any]:
768 """Copy a Deployment update without replicas when its HPA owns scale."""
769 patch_body = copy.deepcopy(document)
770 if not _hpa_owns_replicas(patch_body):
771 return patch_body
773 spec = patch_body.get("spec")
774 if isinstance(spec, dict):
775 spec.pop("replicas", None)
776 return patch_body
779def _api_error_detail(exc: ApiException) -> str:
780 """Return the API server's own message for a failed call, as a log suffix.
782 ``ApiException.reason`` is only the HTTP phrase ("Unprocessable Entity");
783 the field-level cause ("spec.template.spec.containers[0].livenessProbe
784 .tcpSocket: Forbidden: may not specify more than 1 handler type") is the
785 ``message`` of the Status object in the response body. Without it a
786 rejected manifest is a dead end in the convergence log. Returns "" when
787 the body is absent or not a Status document.
788 """
789 body = getattr(exc, "body", None)
790 if not body:
791 return ""
792 try:
793 status = json.loads(body)
794 except TypeError, ValueError:
795 return ""
796 message = status.get("message") if isinstance(status, dict) else None
797 return f": {message}" if message else ""
800def _patch_or_replace(
801 api: Any,
802 resource: str,
803 name: str,
804 namespace: str,
805 document: dict[str, Any],
806 *,
807 patch_body: dict[str, Any] | None = None,
808 keep_live_replicas: bool = False,
809) -> None:
810 """Strategic-merge-patch an existing object, replacing it when the merge is invalid.
812 A strategic merge keeps every live field the manifest no longer sets. For
813 fields that are mutually exclusive that produces an object the API server
814 rejects with 422, so an older install can never be patched forward:
816 - a container probe that moved from ``httpGet`` to ``exec`` (or, for the
817 Slurm accounting MariaDB StatefulSet, from ``exec`` to ``tcpSocket``)
818 ends up with both handlers ("may not specify more than 1 handler type");
819 - a PodDisruptionBudget that moved from ``minAvailable`` to
820 ``maxUnavailable`` ends up with both.
822 On 422 the manifest is the whole desired state, so this replaces the object
823 with it at the live ``resourceVersion`` (an optimistic-concurrency update,
824 not a delete). An HPA-owned Deployment keeps its live replica count. A
825 StatefulSet replace is still bound by the API server's immutability rules
826 (``selector``, ``serviceName``, ``volumeClaimTemplates`` ...): a manifest
827 that changes one of those fails loudly here, as it should.
829 ``resource`` is the client method suffix, e.g. ``namespaced_deployment``
830 for ``patch_/read_/replace_namespaced_deployment``.
831 """
832 try:
833 getattr(api, f"patch_{resource}")(
834 name, namespace, body=document if patch_body is None else patch_body
835 )
836 return
837 except ApiException as exc:
838 if exc.status != 422:
839 raise
840 patch_error = exc
842 live = getattr(api, f"read_{resource}")(name, namespace)
843 body = copy.deepcopy(document)
844 metadata = body.setdefault("metadata", {})
845 metadata["resourceVersion"] = live.metadata.resource_version
846 if keep_live_replicas:
847 live_replicas = getattr(getattr(live, "spec", None), "replicas", None)
848 spec = body.setdefault("spec", {})
849 if live_replicas is None:
850 spec.pop("replicas", None)
851 else:
852 spec["replicas"] = live_replicas
853 logger.warning(
854 "Strategic merge of %s %s/%s was rejected (%s); replacing it with the manifest",
855 resource,
856 namespace,
857 name,
858 patch_error.reason,
859 )
860 getattr(api, f"replace_{resource}")(name, namespace, body=body)
863# Exact resources owned by optional features. Keys include the apply phase so
864# disabling one feature cannot delete similarly named or unrelated resources.
865# Tuples are (apiVersion, kind, namespace-or-None, name); order is deliberate
866# for dependent resources such as FSx claims before volumes before the class.
867# post-helm-slurm-observability.yaml carries both the Slurm and the
868# observability gate, so it applies only when both are on and is pruned when
869# either is turned off: the same identities appear under both inventory keys.
870_SLURM_OBSERVABILITY_RESOURCES: tuple[tuple[str, str, str | None, str], ...] = (
871 ("v1", "ConfigMap", "monitoring", "gco-dashboard-slurm"),
872 ("monitoring.coreos.com/v1", "PrometheusRule", "monitoring", "gco-slurm-alerts"),
873)
875_FEATURE_RESOURCE_INVENTORY: dict[
876 tuple[str, bool], tuple[tuple[str, str, str | None, str], ...]
877] = {
878 ("{{FSX_FILE_SYSTEM_ID}}", False): (
879 ("v1", "PersistentVolumeClaim", "default", "gco-fsx-storage"),
880 ("v1", "PersistentVolumeClaim", "gco-jobs", "gco-fsx-storage"),
881 ("v1", "PersistentVolumeClaim", "gco-system", "gco-fsx-storage"),
882 ("v1", "PersistentVolume", None, "gco-fsx-pv-default"),
883 ("v1", "PersistentVolume", None, "gco-fsx-pv-jobs"),
884 ("v1", "PersistentVolume", None, "gco-fsx-pv-system"),
885 ("storage.k8s.io/v1", "StorageClass", None, "fsx-sc"),
886 ),
887 ("{{VALKEY_ENDPOINT}}", False): tuple(
888 ("v1", "ConfigMap", namespace, "gco-valkey")
889 for namespace in ("gco-system", "gco-jobs", "gco-inference")
890 ),
891 ("{{AURORA_PGVECTOR_ENDPOINT}}", False): tuple(
892 ("v1", "ConfigMap", namespace, "gco-aurora-pgvector")
893 for namespace in ("gco-system", "gco-jobs", "gco-inference")
894 ),
895 ("{{VECTOR_STORE_TABLE_NAME}}", False): tuple(
896 ("v1", "ConfigMap", namespace, "gco-vector-store")
897 for namespace in ("gco-system", "gco-jobs", "gco-inference")
898 ),
899 ("{{CLUSTER_OBSERVABILITY_ENABLED}}", False): (
900 ("apps/v1", "DaemonSet", "kube-system", "dcgm-exporter"),
901 ("v1", "Service", "kube-system", "dcgm-exporter"),
902 ("v1", "ConfigMap", "kube-system", "dcgm-device-counters"),
903 ("storage.k8s.io/v1", "StorageClass", None, "gco-observability-gp3"),
904 ),
905 ("{{CLUSTER_OBSERVABILITY_ENABLED}}", True): (
906 ("batch/v1", "CronJob", "monitoring", "gco-grafana-admin-password-rotation"),
907 ("v1", "ConfigMap", "monitoring", "gco-dashboard-gpu"),
908 ("v1", "ConfigMap", "monitoring", "gco-dashboard-schedulers"),
909 ("v1", "ConfigMap", "monitoring", "gco-dashboard-keda"),
910 ("v1", "ConfigMap", "monitoring", "gco-dashboard-services"),
911 (
912 "rbac.authorization.k8s.io/v1",
913 "ClusterRoleBinding",
914 None,
915 "gco-prometheus-kueue-metrics",
916 ),
917 *tuple(
918 ("monitoring.coreos.com/v1", "ServiceMonitor", "monitoring", name)
919 for name in (
920 "gco-keda",
921 "gco-volcano",
922 "gco-kueue",
923 "gco-kuberay",
924 "gco-yunikorn",
925 "gco-dcgm-exporter",
926 "gco-slurm-controller",
927 )
928 ),
929 *_SLURM_OBSERVABILITY_RESOURCES,
930 *tuple(
931 ("monitoring.coreos.com/v1", "PodMonitor", "monitoring", name)
932 for name in (
933 "gco-health-monitor",
934 "gco-manifest-processor",
935 "gco-inference-proxy",
936 "gco-inference-monitor",
937 )
938 ),
939 # The monitoring half of the internal PKI
940 # (post-helm-monitoring-tls.yaml). Each Certificate goes before its
941 # Secret: cert-manager does not delete a Secret with its Certificate,
942 # and a disabled feature must not leave key material behind; deleted
943 # in this order, nothing re-issues it.
944 ("v1", "Service", "monitoring", "grafana-tls"),
945 ("cert-manager.io/v1", "Certificate", "monitoring", "grafana-tls"),
946 ("v1", "Secret", "monitoring", "grafana-tls"),
947 ("cert-manager.io/v1", "Certificate", "monitoring", "gco-monitoring-trust"),
948 ("v1", "Secret", "monitoring", "gco-monitoring-trust"),
949 ("rbac.authorization.k8s.io/v1", "RoleBinding", "monitoring", "gco-grafana-rotator"),
950 ("rbac.authorization.k8s.io/v1", "Role", "monitoring", "gco-grafana-rotator"),
951 ("v1", "ServiceAccount", "monitoring", "gco-grafana-rotator"),
952 ),
953 ("{{QUEUE_PROCESSOR_IMAGE}}", True): (
954 ("keda.sh/v1alpha1", "ScaledJob", "gco-system", "sqs-queue-processor"),
955 ),
956 # Optional manifest-processor CPU autoscaler (cdk.json
957 # manifest_processor.autoscaling.enabled). Turning it off must also remove
958 # the HPA, otherwise the last scale value would keep fighting the
959 # Deployment's re-asserted replicas.
960 ("{{MP_HPA_ENABLED}}", False): (
961 ("autoscaling/v2", "HorizontalPodAutoscaler", "gco-system", "manifest-processor-hpa"),
962 ),
963 ("{{KUEUE_ENABLED}}", True): (
964 # Deletion order matters: the LocalQueue references the ClusterQueue,
965 # which references the ResourceFlavor.
966 ("kueue.x-k8s.io/v1beta2", "LocalQueue", "gco-jobs", "gco-default"),
967 ("kueue.x-k8s.io/v1beta2", "ClusterQueue", None, "gco-cluster-queue"),
968 ("kueue.x-k8s.io/v1beta2", "ResourceFlavor", None, "gco-default-flavor"),
969 ),
970 # Slurm (cdk.json helm.slurm). The base pass owns the gp3 StorageClass the
971 # controller's state volume and the accounting database bind to; the
972 # post-Helm pass owns the NetworkPolicies, the GCO API's Slurm credential,
973 # the metrics ServiceMonitor and the observability objects
974 # (28-storage-slurm-gp3.yaml, post-helm-slurm-network.yaml,
975 # post-helm-slurm-api-access.yaml, post-helm-slurm-observability.yaml).
976 ("{{SLURM_ENABLED}}", False): (("storage.k8s.io/v1", "StorageClass", None, "gco-slurm-gp3"),),
977 ("{{SLURM_ENABLED}}", True): (
978 # The API's credential first: the binding before the role it names,
979 # then the Token. The Secret the operator minted from the Token is
980 # owned by the chart's JWT key Secret rather than by the Token, so it
981 # is listed here explicitly instead of relying on garbage collection.
982 (
983 "rbac.authorization.k8s.io/v1",
984 "RoleBinding",
985 "gco-jobs",
986 "gco-manifest-processor-slurm-token",
987 ),
988 ("rbac.authorization.k8s.io/v1", "Role", "gco-jobs", "gco-manifest-processor-slurm-token"),
989 ("slinky.slurm.net/v1beta1", "Token", "gco-jobs", "slinky-slurm-token-gco-api"),
990 ("v1", "Secret", "gco-jobs", "slinky-slurm-token-gco-api"),
991 ("slinky.slurm.net/v1beta1", "Token", "gco-jobs", "slinky-slurm-token-client"),
992 ("v1", "Secret", "gco-jobs", "slinky-slurm-client-token"),
993 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-cluster-internal"),
994 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-client-to-restapi"),
995 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-client-egress"),
996 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-slurm-operator-to-restapi"),
997 (
998 "networking.k8s.io/v1",
999 "NetworkPolicy",
1000 "gco-jobs",
1001 "allow-prometheus-to-slurm-controller",
1002 ),
1003 (
1004 "networking.k8s.io/v1",
1005 "NetworkPolicy",
1006 "gco-jobs",
1007 "allow-manifest-processor-to-slurm-restapi",
1008 ),
1009 (
1010 "networking.k8s.io/v1",
1011 "NetworkPolicy",
1012 "gco-system",
1013 "allow-manifest-processor-to-slurm-restapi",
1014 ),
1015 *_SLURM_OBSERVABILITY_RESOURCES,
1016 ),
1017 # The in-cluster accounting database (29-slurm-accounting-db.yaml, base
1018 # pass so MariaDB is up before the Slinky chart starts slurmdbd and
1019 # slurmctld, whose first boot refuses to run without the dbd) and its
1020 # backup CronJob (post-helm-slurm-accounting-backup.yaml), each on its own
1021 # gate so an external database or backups-off keeps the rest. The
1022 # StatefulSet's persistentVolumeClaimRetentionPolicy deletes the data
1023 # claim with it: the nightly dumps in the regional bucket are the record,
1024 # exactly as the MLflow claim below is pruned while its artifacts stay in
1025 # S3. The password Secret is generated in-cluster by the applier
1026 # (gco.io/generate-keys) and goes with the database.
1027 ("{{SLURM_ACCOUNTING_DB_ENABLED}}", False): (
1028 ("apps/v1", "StatefulSet", "gco-jobs", "slinky-slurm-mariadb"),
1029 ("v1", "Service", "gco-jobs", "slinky-slurm-mariadb"),
1030 ("v1", "Secret", "gco-jobs", "slinky-slurm-mariadb"),
1031 ),
1032 ("{{SLURM_ACCOUNTING_BACKUP_ENABLED}}", True): (
1033 ("batch/v1", "CronJob", "gco-jobs", "slinky-slurm-accounting-backup"),
1034 ),
1035 ("{{KUBEFLOW_TRAINER_ENABLED}}", True): (
1036 ("trainer.kubeflow.org/v1alpha1", "ClusterTrainingRuntime", None, "torch-distributed"),
1037 ),
1038 ("{{MLFLOW_ENABLED}}", True): (
1039 # The claim is created BY THE CHART (storage.enabled), not by a
1040 # shipped manifest — it appears here because helm uninstall never
1041 # deletes chart PVCs, so disabling the feature would otherwise leak
1042 # the volume forever. Deliberately destructive on disable: the claim
1043 # holds the tracking server's SQLite run METADATA. Run artifacts
1044 # live in S3 (untouched).
1045 ("v1", "PersistentVolumeClaim", "monitoring", "mlflow"),
1046 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-mlflow-clients"),
1047 # The server's own network posture; GCO owns it because the chart's
1048 # policy drops kubelet probes (post-helm-mlflow-network.yaml).
1049 ("networking.k8s.io/v1", "NetworkPolicy", "monitoring", "mlflow-server"),
1050 # The HTTPS front door and its trust (post-helm-mlflow-tls.yaml). Each
1051 # Certificate goes before its Secret, as for the observability leaves.
1052 # The Bundle goes before the ConfigMap trust-manager writes from it,
1053 # so nothing rewrites the ConfigMap once it is deleted; the Bundle
1054 # owns it, but pruning it here does not wait on garbage collection.
1055 # helm uninstall keeps the Bundle CRD, so the Bundle delete resolves.
1056 ("v1", "Service", "monitoring", "mlflow-tls"),
1057 ("cert-manager.io/v1", "Certificate", "monitoring", "mlflow-tls"),
1058 ("v1", "Secret", "monitoring", "mlflow-tls"),
1059 ("trust.cert-manager.io/v1alpha1", "Bundle", None, "gco-internal-ca"),
1060 ("v1", "ConfigMap", "gco-jobs", "gco-internal-ca"),
1061 ("cert-manager.io/v1", "Certificate", "trust-manager", "gco-internal-ca-source"),
1062 ("v1", "Secret", "trust-manager", "gco-internal-ca-source"),
1063 ),
1064 ("{{COST_MONITORING_ENABLED}}", False): (
1065 ("apps/v1", "Deployment", "gco-system", "cost-monitor"),
1066 ("v1", "Service", "gco-system", "cost-monitor"),
1067 ("v1", "ServiceAccount", "gco-system", "gco-cost-monitor-sa"),
1068 (
1069 "networking.k8s.io/v1",
1070 "NetworkPolicy",
1071 "gco-system",
1072 "allow-manifest-processor-to-cost-monitor-ingress",
1073 ),
1074 (
1075 "networking.k8s.io/v1",
1076 "NetworkPolicy",
1077 "gco-system",
1078 "allow-cost-monitor-to-opencost",
1079 ),
1080 (
1081 "networking.k8s.io/v1",
1082 "NetworkPolicy",
1083 "gco-system",
1084 "allow-manifest-processor-to-cost-monitor-egress",
1085 ),
1086 ),
1087 ("{{COST_MONITORING_ENABLED}}", True): (
1088 ("v1", "ConfigMap", "monitoring", "gco-dashboard-cost"),
1089 # post-helm-cost-monitoring-tls.yaml; each Certificate before its
1090 # Secret, as for the observability leaves above.
1091 ("v1", "Service", "monitoring", "opencost-tls"),
1092 ("cert-manager.io/v1", "Certificate", "monitoring", "opencost-tls"),
1093 ("v1", "Secret", "monitoring", "opencost-tls"),
1094 ("cert-manager.io/v1", "Certificate", "gco-system", "cost-monitor-tls"),
1095 ("v1", "Secret", "gco-system", "cost-monitor-tls"),
1096 ),
1097 # Self-managed Argo CD (cdk.json helm.argocd; post-helm-argocd-access.yaml).
1098 # Bindings before the roles they reference. The AppProject is normally
1099 # gone already: the helm installer deletes every argoproj.io object before
1100 # it uninstalls the chart (the CRDs stay, so this delete is a clean no-op).
1101 ("{{ARGOCD_ENABLED}}", True): (
1102 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-jobs", "gco-argocd-deploy"),
1103 ("rbac.authorization.k8s.io/v1", "Role", "gco-jobs", "gco-argocd-deploy"),
1104 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-inference", "gco-argocd-deploy"),
1105 ("rbac.authorization.k8s.io/v1", "Role", "gco-inference", "gco-argocd-deploy"),
1106 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-jobs", "gco-argocd-read"),
1107 ("rbac.authorization.k8s.io/v1", "Role", "gco-jobs", "gco-argocd-read"),
1108 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-inference", "gco-argocd-read"),
1109 ("rbac.authorization.k8s.io/v1", "Role", "gco-inference", "gco-argocd-read"),
1110 ("argoproj.io/v1alpha1", "AppProject", "argocd", "gco-tenants"),
1111 ),
1112 # post-helm-argocd-gitops.yaml: the GitOps hand-off (helm.argocd.gitops).
1113 # The Application carries no resources-finalizer, so deleting it detaches
1114 # the tenant workloads from Git instead of deleting them.
1115 ("{{ARGOCD_GITOPS_REPO_URL}}", True): (
1116 ("argoproj.io/v1alpha1", "Application", "argocd", "gco-gitops-root"),
1117 ),
1118 # Self-managed Crossplane (cdk.json helm.crossplane; post-helm-crossplane.yaml).
1119 # Bindings before the roles they reference; the Function (and with it
1120 # the function's Deployment) goes last. As with Argo CD, the helm
1121 # installer has normally deleted every pkg.crossplane.io object already.
1122 ("{{CROSSPLANE_ENABLED}}", True): (
1123 ("rbac.authorization.k8s.io/v1", "ClusterRoleBinding", None, "gco-crossview-read"),
1124 ("rbac.authorization.k8s.io/v1", "ClusterRole", None, "gco-crossview-read"),
1125 (
1126 "rbac.authorization.k8s.io/v1",
1127 "ClusterRoleBinding",
1128 None,
1129 "gco-crossview-crossplane-view",
1130 ),
1131 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-jobs", "gco-crossplane-compose"),
1132 ("rbac.authorization.k8s.io/v1", "Role", "gco-jobs", "gco-crossplane-compose"),
1133 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-inference", "gco-crossplane-compose"),
1134 ("rbac.authorization.k8s.io/v1", "Role", "gco-inference", "gco-crossplane-compose"),
1135 ("rbac.authorization.k8s.io/v1", "ClusterRole", None, "gco-crossplane-read"),
1136 ("pkg.crossplane.io/v1", "Function", None, "crossplane-contrib-function-go-templating"),
1137 ),
1138 # kro EKS Capability tenant RBAC (cdk.json eks_capabilities.kro;
1139 # 07-kro-tenant-access.yaml). Bindings before the roles they reference.
1140 ("{{KRO_CAPABILITY_USERNAME}}", False): (
1141 ("rbac.authorization.k8s.io/v1", "ClusterRoleBinding", None, "gco-kro-read"),
1142 ("rbac.authorization.k8s.io/v1", "ClusterRole", None, "gco-kro-read"),
1143 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-jobs", "gco-kro-compose"),
1144 ("rbac.authorization.k8s.io/v1", "Role", "gco-jobs", "gco-kro-compose"),
1145 ("rbac.authorization.k8s.io/v1", "RoleBinding", "gco-inference", "gco-kro-compose"),
1146 ("rbac.authorization.k8s.io/v1", "Role", "gco-inference", "gco-kro-compose"),
1147 ),
1148}
1151# Resources GCO shipped in earlier releases that no longer appear in the
1152# manifest set. The base apply pass deletes them exactly (missing = no-op) so
1153# upgraded clusters do not keep orphaned objects running forever.
1154#
1155# nvidia-device-plugin-daemonset: GCO runs exclusively on EKS Auto Mode, which
1156# ships its own NVIDIA device plugin built into the node ("runs automatically
1157# and isn't visible as a daemon set" — the EKS auto-accelerated guide). The
1158# community plugin GCO used to ship can never start on Auto Mode GPU nodes:
1159# the runtime only injects the NVIDIA driver libraries for containers that
1160# request them, so the plugin crash-loops with NVML ERROR_LIBRARY_NOT_FOUND
1161# and permanently fails DaemonSet convergence (observed live the moment the
1162# Slurm NodeSet provisioned the first GPU nodes). The built-in plugin
1163# advertises nvidia.com/gpu on its own.
1164#
1165# allow-vpc-endpoint-egress / allow-ray-cluster-internal (gco-jobs): the
1166# pre-v7.7 job-namespace model — HTTPS only to the VPC's own CIDR (there were
1167# never VPC endpoints for that traffic to reach, so enforced it cut jobs off
1168# from every AWS API) and a Ray-only peer rule. 03-network-policies.yaml now
1169# ships allow-https-egress + allow-vpc-egress + allow-same-namespace, which
1170# strictly contain both. NetworkPolicies union, so the leftovers would allow
1171# nothing new — they are swept so the live policy set stays exactly the
1172# shipped, documented one.
1173#
1174# gco-api-selfsigned (gco-system): the namespaced selfSigned Issuer that signed
1175# each API workload leaf as its own root before the internal CA existed. The
1176# leaves now name the gco-internal-ca ClusterIssuer
1177# (post-helm-api-workload-certificates.yaml). Deleting the Issuer in the base
1178# pass, before the post-Helm pass re-points them, is safe: a Certificate whose
1179# issuer is briefly missing keeps its Secret, stays Ready while that Secret is
1180# valid, and is re-issued by the new issuer as soon as its issuerRef changes.
1181# On a fresh cluster the cert-manager CRDs do not exist yet in the base pass,
1182# which the sweep treats as already absent.
1183_LEGACY_REMOVED_RESOURCES: tuple[tuple[str, str, str | None, str], ...] = (
1184 ("apps/v1", "DaemonSet", "kube-system", "nvidia-device-plugin-daemonset"),
1185 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-vpc-endpoint-egress"),
1186 ("networking.k8s.io/v1", "NetworkPolicy", "gco-jobs", "allow-ray-cluster-internal"),
1187 ("cert-manager.io/v1", "Issuer", "gco-system", "gco-api-selfsigned"),
1188)
1191def _delete_exact_resources(
1192 targets: tuple[tuple[str, str, str | None, str], ...],
1193 context: str,
1194) -> dict[str, list[str]]:
1195 """Delete an exact list of (apiVersion, kind, namespace, name) resources.
1197 Missing resources and missing CRDs/API resource types are successful no-ops.
1198 Every other error is returned so convergence fails instead of silently
1199 leaving stale resources running.
1200 """
1201 result: dict[str, list[str]] = {"pruned": [], "failed": []}
1202 if not targets:
1203 return result
1205 dynamic_client = dynamic.DynamicClient(client.ApiClient())
1206 delete_options = client.V1DeleteOptions(propagation_policy="Background")
1207 for api_version, kind, namespace, name in targets:
1208 identifier = f"{api_version}/{kind}/{namespace or '<cluster>'}/{name}"
1209 try:
1210 resource = dynamic_client.resources.get(api_version=api_version, kind=kind)
1211 kwargs: dict[str, Any] = {"name": name, "body": delete_options}
1212 if namespace is not None:
1213 kwargs["namespace"] = namespace
1214 resource.delete(**kwargs)
1215 result["pruned"].append(identifier)
1216 logger.info("Pruned %s resource %s", context, identifier)
1217 except ResourceNotFoundError, NotFoundError:
1218 logger.info("%s resource already absent: %s", context, identifier)
1219 except ApiException as exc:
1220 if exc.status == 404:
1221 logger.info("%s resource already absent: %s", context, identifier)
1222 else:
1223 failure = f"{identifier}:{exc.status}:{exc.reason}"
1224 result["failed"].append(failure)
1225 logger.error("Failed pruning %s resource %s", context, failure)
1226 except Exception as exc:
1227 if getattr(exc, "status", None) == 404:
1228 logger.info("%s resource already absent: %s", context, identifier)
1229 else:
1230 failure = f"{identifier}:{exc}"
1231 result["failed"].append(failure)
1232 logger.error("Failed pruning %s resource %s", context, failure)
1234 return result
1237def _prune_disabled_feature(placeholder: str, post_helm: bool) -> dict[str, list[str]]:
1238 """Delete only the exact resources managed by a disabled optional feature."""
1239 return _delete_exact_resources(
1240 _FEATURE_RESOURCE_INVENTORY.get((placeholder, post_helm), ()),
1241 "disabled-feature",
1242 )
1245def _prune_legacy_removed_resources() -> dict[str, list[str]]:
1246 """Delete resources shipped by earlier GCO releases and since removed."""
1247 return _delete_exact_resources(_LEGACY_REMOVED_RESOURCES, "legacy-removed")
1250# ---------------------------------------------------------------------------
1251# Tunables
1252# ---------------------------------------------------------------------------
1254# Maximum time to wait for a PersistentVolume or PersistentVolumeClaim to
1255# disappear after we issue a delete. Needed when we're recreating a PV
1256# whose ``volumeHandle`` changed (FSx/EFS ID rotated) or reconciling a
1257# Lost PVC whose backing PV was just recreated. If you see this wait
1258# consistently timing out, check for stuck finalizers (the handler
1259# already clears the standard ``pv-protection`` / ``pvc-protection``
1260# finalizers) or for an AWS control-plane issue on the underlying
1261# volume. Raising the value is safe — it only blocks the re-create path,
1262# not steady-state applies.
1263PV_PVC_DELETE_WAIT_SECONDS = 30
1265# Interval between PV/PVC existence polls during the delete wait.
1266PV_PVC_DELETE_POLL_INTERVAL_SECONDS = 1
1269def get_eks_client() -> Any:
1270 """Get EKS client with lazy initialization."""
1271 global _eks_client
1272 if _eks_client is None:
1273 _eks_client = boto3.client("eks")
1274 return _eks_client
1277def send_response(
1278 event: dict[str, Any],
1279 context: Any,
1280 response_status: str,
1281 response_data: dict[str, Any],
1282 physical_resource_id: str,
1283 reason: str | None = None,
1284) -> None:
1285 """Send response to CloudFormation."""
1286 response_body = {
1287 "Status": response_status,
1288 "Reason": reason or f"See CloudWatch Log Stream: {context.log_stream_name}",
1289 "PhysicalResourceId": physical_resource_id,
1290 "StackId": event["StackId"],
1291 "RequestId": event["RequestId"],
1292 "LogicalResourceId": event["LogicalResourceId"],
1293 "Data": response_data,
1294 }
1296 logger.info(f"Sending response: {json.dumps(response_data)}")
1298 # Timeout is for the CFN response callback (HTTP PUT to S3 presigned URL),
1299 # not for manifest application. K8s API calls have their own timeouts.
1300 http = urllib3.PoolManager()
1301 try:
1302 http.request(
1303 "PUT",
1304 event["ResponseURL"],
1305 body=json.dumps(response_body).encode("utf-8"),
1306 headers={"Content-Type": "application/json"},
1307 timeout=10.0,
1308 )
1309 except Exception as e:
1310 logger.error(f"Failed to send response: {e}")
1313def get_eks_token(cluster_name: str, region: str) -> str:
1314 """Generate EKS authentication token using STS presigned URL."""
1315 from botocore.signers import RequestSigner
1317 # Create STS client
1318 session = boto3.Session()
1319 sts_client = session.client("sts", region_name=region)
1320 service_id = sts_client.meta.service_model.service_id
1322 # Create request signer
1323 signer = RequestSigner(
1324 service_id, region, "sts", "v4", session.get_credentials(), session.events
1325 )
1327 # Build the presigned URL for GetCallerIdentity
1328 params = {
1329 "method": "GET",
1330 "url": f"https://sts.{region}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15",
1331 "body": {},
1332 "headers": {"x-k8s-aws-id": cluster_name},
1333 "context": {},
1334 }
1336 # Generate presigned URL (valid for 60 seconds)
1337 url = signer.generate_presigned_url(
1338 params, region_name=region, expires_in=60, operation_name=""
1339 )
1341 # Encode as base64 and create the k8s-aws-v1 token
1342 token_b64 = base64.urlsafe_b64encode(url.encode("utf-8")).decode("utf-8").rstrip("=")
1343 return f"k8s-aws-v1.{token_b64}"
1346def configure_k8s_client(cluster_name: str, region: str) -> None:
1347 """Configure Kubernetes client for EKS cluster."""
1348 eks = get_eks_client()
1350 # Get cluster info
1351 cluster_info = eks.describe_cluster(name=cluster_name)
1352 cluster = cluster_info["cluster"]
1354 # Configure Kubernetes client
1355 configuration = client.Configuration()
1356 configuration.host = cluster["endpoint"]
1357 configuration.verify_ssl = True
1359 # Set connection timeouts (important for Lambda!)
1360 configuration.connection_pool_maxsize = 1
1361 configuration.retries = 3
1362 # Set socket timeout to 30 seconds
1363 import socket
1365 socket.setdefaulttimeout(30)
1367 # Decode and write CA certificate to temp file using secure method
1368 ca_cert = base64.b64decode(cluster["certificateAuthority"]["data"])
1369 import tempfile
1371 fd, ca_cert_path = tempfile.mkstemp(suffix=".crt")
1372 try:
1373 with os.fdopen(fd, "wb") as ca_file:
1374 ca_file.write(ca_cert)
1375 ca_file.flush()
1376 configuration.ssl_ca_cert = ca_cert_path
1377 except Exception:
1378 os.close(fd)
1379 raise
1381 # Generate EKS authentication token
1382 eks_token = get_eks_token(cluster_name, region)
1384 # Set the bearer token
1385 configuration.api_key = {"authorization": f"Bearer {eks_token}"}
1387 logger.info(
1388 f"✓ Configured Kubernetes client for cluster {cluster_name} at {cluster['endpoint']}"
1389 )
1391 client.Configuration.set_default(configuration)
1394def restart_deployments(namespace: str, deployment_names: list[str]) -> dict[str, Any]:
1395 """
1396 Restart deployments by patching their spec with a restart annotation.
1397 This forces Kubernetes to roll out new pods with the latest image.
1398 """
1399 from datetime import datetime
1401 apps_v1 = client.AppsV1Api()
1402 restarted = []
1403 failed = []
1405 restart_time = datetime.now(UTC).isoformat()
1407 for name in deployment_names:
1408 try:
1409 # Patch the deployment with a restart annotation
1410 # This is equivalent to `kubectl rollout restart deployment`
1411 patch = {
1412 "spec": {
1413 "template": {
1414 "metadata": {
1415 "annotations": {"kubectl.kubernetes.io/restartedAt": restart_time}
1416 }
1417 }
1418 }
1419 }
1420 apps_v1.patch_namespaced_deployment(name, namespace, body=patch)
1421 restarted.append(name)
1422 logger.info(f"✓ Restarted deployment {name} in namespace {namespace}")
1423 except ApiException as e:
1424 # 404 means the deployment isn't installed on this cluster
1425 # (e.g. fsx-csi when FSx is disabled) — that's not a failure.
1426 if e.status == 404:
1427 logger.info(f"Deployment {namespace}/{name} not found — skipping restart")
1428 else:
1429 logger.error(f"Failed to restart deployment {name}: {e.status} - {e.reason}")
1430 failed.append(name)
1432 return {"restarted": restarted, "failed": failed}
1435def restart_daemonsets(namespace: str, daemonset_names: list[str]) -> dict[str, Any]:
1436 """
1437 Restart daemonsets by patching their pod template with a restart annotation.
1438 This forces Kubernetes to roll out new pods with the latest image or latest
1439 service-account annotation set.
1441 Used for IRSA-annotated addons whose pods still need to be re-mutated by
1442 the EKS Pod Identity webhook after the addon's service account had its
1443 role ARN annotation patched post-install. Without this, DaemonSet pods
1444 like efs-csi-node, fsx-csi-node, and cloudwatch-agent keep their
1445 original (credential-less) pod spec and silently fail with IMDS 401s.
1446 """
1447 from datetime import datetime
1449 apps_v1 = client.AppsV1Api()
1450 restarted = []
1451 failed = []
1453 restart_time = datetime.now(UTC).isoformat()
1455 for name in daemonset_names:
1456 try:
1457 patch = {
1458 "spec": {
1459 "template": {
1460 "metadata": {
1461 "annotations": {"kubectl.kubernetes.io/restartedAt": restart_time}
1462 }
1463 }
1464 }
1465 }
1466 apps_v1.patch_namespaced_daemon_set(name, namespace, body=patch)
1467 restarted.append(name)
1468 logger.info(f"✓ Restarted daemonset {name} in namespace {namespace}")
1469 except ApiException as e:
1470 # 404 means the daemonset isn't installed on this cluster
1471 # (e.g. fsx-csi-node when FSx is disabled) — that's expected,
1472 # not a failure.
1473 if e.status == 404:
1474 logger.info(f"DaemonSet {namespace}/{name} not found — skipping restart")
1475 else:
1476 logger.error(f"Failed to restart daemonset {name}: {e.status} - {e.reason}")
1477 failed.append(name)
1479 return {"restarted": restarted, "failed": failed}
1482# Every platform Deployment GCO ships in gco-system, with the dedicated
1483# ServiceAccount it must run as: (namespace, deployment, service account).
1484# tests/test_platform_workload_contract.py pins this to the manifests, so a new
1485# service (or a renamed account) fails a unit test instead of silently escaping
1486# the post-apply credential verification below.
1487PLATFORM_DEPLOYMENTS: tuple[tuple[str, str, str], ...] = (
1488 ("gco-system", "health-monitor", "gco-health-monitor-sa"),
1489 ("gco-system", "manifest-processor", "gco-manifest-processor-sa"),
1490 ("gco-system", "inference-monitor", "gco-inference-monitor-sa"),
1491 ("gco-system", "inference-proxy", "gco-inference-proxy-sa"),
1492 # Feature-gated (cdk.json cost_monitoring); verified only when planned.
1493 ("gco-system", "cost-monitor", "gco-cost-monitor-sa"),
1494)
1496# User-workload accounts the manifests declare outside gco-system.
1497WORKLOAD_SERVICE_ACCOUNTS: tuple[tuple[str, str], ...] = (
1498 ("gco-jobs", "gco-service-account"),
1499 ("gco-inference", "gco-service-account"),
1500)
1503def _verify_workload_credentials(
1504 apps_v1: Any,
1505 planned: list[dict[str, Any]] | None = None,
1506) -> list[str]:
1507 """Verify that key GCO deployments have working IAM credential configuration.
1509 Checks that:
1510 1. Deployments use their dedicated service account (with IRSA annotation)
1511 2. The projected service-account token volume is mounted
1512 3. AWS_ROLE_ARN and AWS_WEB_IDENTITY_TOKEN_FILE env vars are set
1514 ``planned`` is the base-phase resource list from ``plan_manifests``. When
1515 given, a platform Deployment (and its ServiceAccount) is checked only if
1516 this apply planned it, so a feature-gated service that is switched off
1517 (cost-monitor) does not surface as "deployment not found".
1519 Returns a list of warning strings (empty = all good).
1520 """
1521 warnings: list[str] = []
1522 planned_identities: set[tuple[str, str, str]] | None = None
1523 if planned is not None:
1524 planned_identities = {
1525 (str(item["kind"]), str(item["namespace"]), str(item["name"])) for item in planned
1526 }
1528 def _is_planned(kind: str, namespace: str, name: str) -> bool:
1529 return planned_identities is None or (kind, namespace, name) in planned_identities
1531 expected_deployments = [
1532 (namespace, name, service_account)
1533 for namespace, name, service_account in PLATFORM_DEPLOYMENTS
1534 if _is_planned("Deployment", namespace, name)
1535 ]
1537 for namespace, name, expected_sa in expected_deployments:
1538 try:
1539 dep = apps_v1.read_namespaced_deployment(name, namespace)
1540 spec = dep.spec.template.spec
1542 # Check service account
1543 if spec.service_account_name != expected_sa:
1544 warnings.append(
1545 f"{namespace}/{name}: uses SA '{spec.service_account_name}' instead of {expected_sa}"
1546 )
1548 # Check for projected token volume
1549 has_token_volume = False
1550 if spec.volumes:
1551 for vol in spec.volumes:
1552 if vol.projected and vol.projected.sources:
1553 for src in vol.projected.sources:
1554 if (
1555 src.service_account_token
1556 and src.service_account_token.audience == "sts.amazonaws.com"
1557 ):
1558 has_token_volume = True
1559 break
1561 if not has_token_volume:
1562 warnings.append(
1563 f"{namespace}/{name}: missing projected service-account token volume for IRSA"
1564 )
1566 # Check env vars on first container
1567 container = spec.containers[0] if spec.containers else None
1568 if container and container.env:
1569 env_names = {e.name for e in container.env}
1570 if "AWS_ROLE_ARN" not in env_names:
1571 warnings.append(f"{namespace}/{name}: missing AWS_ROLE_ARN env var")
1572 if "AWS_WEB_IDENTITY_TOKEN_FILE" not in env_names:
1573 warnings.append(
1574 f"{namespace}/{name}: missing AWS_WEB_IDENTITY_TOKEN_FILE env var"
1575 )
1577 except ApiException as e:
1578 if e.status == 404:
1579 warnings.append(f"{namespace}/{name}: deployment not found")
1580 else:
1581 warnings.append(f"{namespace}/{name}: failed to read ({e.status})")
1582 except Exception as e:
1583 warnings.append(f"{namespace}/{name}: verification error ({e})")
1585 # Check that service accounts exist in all required namespaces: the
1586 # dedicated account of every planned platform Deployment, then the user
1587 # workload accounts.
1588 v1 = client.CoreV1Api()
1589 platform_sas = [(namespace, sa_name) for namespace, _name, sa_name in expected_deployments]
1590 workload_sas = [
1591 (namespace, sa_name)
1592 for namespace, sa_name in WORKLOAD_SERVICE_ACCOUNTS
1593 if _is_planned("ServiceAccount", namespace, sa_name)
1594 ]
1595 for namespace, sa_name in platform_sas + workload_sas:
1596 try:
1597 sa = v1.read_namespaced_service_account(sa_name, namespace)
1598 annotations = sa.metadata.annotations or {}
1599 if "eks.amazonaws.com/role-arn" not in annotations:
1600 warnings.append(
1601 f"{namespace}/{sa_name}: missing eks.amazonaws.com/role-arn annotation"
1602 )
1603 except ApiException as e:
1604 if e.status == 404:
1605 warnings.append(f"{namespace}/{sa_name}: ServiceAccount not found")
1606 else:
1607 warnings.append(f"{namespace}/{sa_name}: failed to read ({e.status})")
1609 if warnings:
1610 for w in warnings:
1611 logger.warning(f"⚠ Credential check: {w}")
1612 else:
1613 logger.info("✓ All workload IAM credential configurations verified")
1615 return warnings
1618def apply_manifests(
1619 cluster_name: str,
1620 region: str,
1621 manifests_dir: str,
1622 replacements: dict[str, str],
1623 post_helm: bool = False,
1624) -> dict[str, Any]:
1625 """Apply Kubernetes manifests.
1627 Args:
1628 cluster_name: EKS cluster name
1629 region: AWS region
1630 manifests_dir: Directory containing manifest YAML files
1631 replacements: Template variable substitutions
1632 post_helm: If True, apply only post-helm-* files (run after Helm installs CRDs).
1633 If False (default), apply all other files and skip post-helm-* ones.
1634 """
1635 plan = plan_manifests(manifests_dir, replacements)
1636 phase_name = "post-helm" if post_helm else "base"
1637 planned_resources = plan["phases"][phase_name]
1638 expected_resources = [_public_manifest_identity(item) for item in planned_resources]
1639 expected_count = len(planned_resources)
1641 configure_k8s_client(cluster_name, region)
1642 v1 = client.CoreV1Api()
1643 apps_v1 = client.AppsV1Api()
1644 autoscaling_v2 = client.AutoscalingV2Api()
1645 rbac_v1 = client.RbacAuthorizationV1Api()
1646 networking_v1 = client.NetworkingV1Api()
1647 custom_api = client.CustomObjectsApi()
1649 applied_count = 0
1650 failed: list[str] = []
1651 skipped: list[str] = list(plan["skipped"][phase_name])
1652 pruned: list[str] = []
1653 prune_failures: list[str] = []
1655 # A gated-out file represents an optional feature that is disabled. Preserve
1656 # the existing exact-resource pruning behavior, once per gate and phase.
1657 for placeholder in plan["featureGates"][phase_name]:
1658 if (placeholder, post_helm) not in _FEATURE_RESOURCE_INVENTORY:
1659 continue
1660 prune_result = _prune_disabled_feature(placeholder, post_helm)
1661 pruned.extend(prune_result["pruned"])
1662 prune_failures.extend(prune_result["failed"])
1663 failed.extend(f"prune:{failure}" for failure in prune_result["failed"])
1665 # Objects GCO used to ship that no longer exist in any manifest — delete
1666 # them exactly on upgraded clusters (fresh clusters: no-op). Base pass
1667 # only, so the sweep runs once per convergence.
1668 if not post_helm:
1669 legacy_result = _prune_legacy_removed_resources()
1670 pruned.extend(legacy_result["pruned"])
1671 prune_failures.extend(legacy_result["failed"])
1672 failed.extend(f"prune:{failure}" for failure in legacy_result["failed"])
1674 for planned_resource in planned_resources:
1675 filename = planned_resource["sourceFile"]
1676 try:
1677 for doc in (planned_resource["document"],):
1678 kind = planned_resource["kind"]
1679 api_version = planned_resource["apiVersion"]
1680 namespace = doc.get("metadata", {}).get("namespace", "default")
1681 name = planned_resource["name"]
1683 logger.info(f"Applying {kind}/{name} in namespace {namespace}")
1684 try:
1685 # Apply based on kind
1686 if kind == "Namespace":
1687 try:
1688 v1.create_namespace(body=doc)
1689 except ApiException as e:
1690 if e.status == 409: # Already exists
1691 v1.patch_namespace(name, body=doc)
1692 else:
1693 raise
1695 elif kind == "ServiceAccount":
1696 _log_service_account_automount_flip(v1, doc, namespace, name, plan)
1697 try:
1698 v1.create_namespaced_service_account(namespace, body=doc)
1699 except ApiException as e:
1700 if e.status == 409:
1701 v1.patch_namespaced_service_account(name, namespace, body=doc)
1702 else:
1703 raise
1705 elif kind == "ClusterRole":
1706 try:
1707 rbac_v1.create_cluster_role(body=doc)
1708 except ApiException as e:
1709 if e.status == 409:
1710 rbac_v1.patch_cluster_role(name, body=doc)
1711 else:
1712 raise
1714 elif kind == "ClusterRoleBinding":
1715 try:
1716 rbac_v1.create_cluster_role_binding(body=doc)
1717 except ApiException as e:
1718 if e.status == 409:
1719 rbac_v1.patch_cluster_role_binding(name, body=doc)
1720 else:
1721 raise
1723 elif kind == "Role":
1724 try:
1725 rbac_v1.create_namespaced_role(namespace, body=doc)
1726 except ApiException as e:
1727 if e.status == 409:
1728 rbac_v1.patch_namespaced_role(name, namespace, body=doc)
1729 else:
1730 raise
1732 elif kind == "RoleBinding":
1733 try:
1734 rbac_v1.create_namespaced_role_binding(namespace, body=doc)
1735 except ApiException as e:
1736 if e.status == 409:
1737 rbac_v1.patch_namespaced_role_binding(name, namespace, body=doc)
1738 else:
1739 raise
1741 elif kind == "Lease":
1742 coordination_v1 = client.CoordinationV1Api()
1743 try:
1744 coordination_v1.create_namespaced_lease(namespace, body=doc)
1745 except ApiException as e:
1746 if e.status == 409:
1747 coordination_v1.patch_namespaced_lease(name, namespace, body=doc)
1748 else:
1749 raise
1751 elif kind == "Deployment":
1752 try:
1753 apps_v1.create_namespaced_deployment(namespace, body=doc)
1754 except ApiException as e:
1755 if e.status == 409:
1756 _patch_or_replace(
1757 apps_v1,
1758 "namespaced_deployment",
1759 name,
1760 namespace,
1761 doc,
1762 patch_body=_deployment_patch_body(doc),
1763 keep_live_replicas=_hpa_owns_replicas(doc),
1764 )
1765 else:
1766 raise
1768 elif kind == "StatefulSet":
1769 try:
1770 apps_v1.create_namespaced_stateful_set(namespace, body=doc)
1771 except ApiException as e:
1772 if e.status == 409:
1773 _patch_or_replace(
1774 apps_v1, "namespaced_stateful_set", name, namespace, doc
1775 )
1776 else:
1777 raise
1779 elif kind == "DaemonSet":
1780 try:
1781 apps_v1.create_namespaced_daemon_set(namespace, body=doc)
1782 except ApiException as e:
1783 if e.status == 409:
1784 _patch_or_replace(
1785 apps_v1, "namespaced_daemon_set", name, namespace, doc
1786 )
1787 else:
1788 raise
1790 elif kind == "Job":
1791 batch_v1 = client.BatchV1Api()
1792 try:
1793 batch_v1.create_namespaced_job(namespace, body=doc)
1794 except ApiException as e:
1795 if e.status == 409:
1796 batch_v1.patch_namespaced_job(name, namespace, body=doc)
1797 else:
1798 raise
1800 elif kind == "CronJob":
1801 # batch/v1 CronJob (e.g. the Grafana admin-password
1802 # rotation job in the observability post-Helm pass).
1803 batch_v1 = client.BatchV1Api()
1804 try:
1805 batch_v1.create_namespaced_cron_job(namespace, body=doc)
1806 except ApiException as e:
1807 if e.status == 409:
1808 batch_v1.patch_namespaced_cron_job(name, namespace, body=doc)
1809 else:
1810 raise
1812 elif kind == "HorizontalPodAutoscaler":
1813 try:
1814 autoscaling_v2.create_namespaced_horizontal_pod_autoscaler(
1815 namespace, body=doc
1816 )
1817 except ApiException as e:
1818 if e.status == 409:
1819 autoscaling_v2.patch_namespaced_horizontal_pod_autoscaler(
1820 name, namespace, body=doc
1821 )
1822 else:
1823 raise
1825 elif kind == "PodDisruptionBudget":
1826 policy_v1 = client.PolicyV1Api()
1827 try:
1828 policy_v1.create_namespaced_pod_disruption_budget(namespace, body=doc)
1829 except ApiException as e:
1830 if e.status == 409:
1831 _patch_or_replace(
1832 policy_v1,
1833 "namespaced_pod_disruption_budget",
1834 name,
1835 namespace,
1836 doc,
1837 )
1838 else:
1839 raise
1841 elif kind == "Service":
1842 try:
1843 v1.create_namespaced_service(namespace, body=doc)
1844 except ApiException as e:
1845 if e.status == 409:
1846 v1.patch_namespaced_service(name, namespace, body=doc)
1847 else:
1848 raise
1850 elif kind == "Pod":
1851 try:
1852 v1.create_namespaced_pod(namespace, body=doc)
1853 except ApiException as e:
1854 if e.status == 409:
1855 v1.patch_namespaced_pod(name, namespace, body=doc)
1856 else:
1857 raise
1859 elif kind == "ConfigMap":
1860 try:
1861 v1.create_namespaced_config_map(namespace, body=doc)
1862 except ApiException as e:
1863 if e.status == 409:
1864 v1.patch_namespaced_config_map(name, namespace, body=doc)
1865 else:
1866 raise
1868 elif kind == "Secret":
1869 generated_keys = _generated_secret_keys(doc)
1870 if generated_keys:
1871 _apply_generated_secret(v1, doc, namespace, name, generated_keys)
1872 else:
1873 try:
1874 v1.create_namespaced_secret(namespace, body=doc)
1875 except ApiException as e:
1876 if e.status == 409:
1877 v1.patch_namespaced_secret(name, namespace, body=doc)
1878 else:
1879 raise
1881 elif (
1882 kind in _GATEWAY_CUSTOM_OBJECTS
1883 or kind in _QUEUEING_CUSTOM_OBJECTS
1884 or kind in _CERT_MANAGER_CUSTOM_OBJECTS
1885 or kind in _ARGOCD_CUSTOM_OBJECTS
1886 or kind in _CROSSPLANE_CUSTOM_OBJECTS
1887 or kind in _TRUST_MANAGER_CUSTOM_OBJECTS
1888 ):
1889 group, version, plural, cluster_scoped = (
1890 _GATEWAY_CUSTOM_OBJECTS.get(kind)
1891 or _QUEUEING_CUSTOM_OBJECTS.get(kind)
1892 or _CERT_MANAGER_CUSTOM_OBJECTS.get(kind)
1893 or _ARGOCD_CUSTOM_OBJECTS.get(kind)
1894 or _CROSSPLANE_CUSTOM_OBJECTS.get(kind)
1895 or _TRUST_MANAGER_CUSTOM_OBJECTS[kind]
1896 )
1897 try:
1898 if cluster_scoped:
1899 custom_api.create_cluster_custom_object(
1900 group, version, plural, body=doc
1901 )
1902 else:
1903 custom_api.create_namespaced_custom_object(
1904 group, version, namespace, plural, body=doc
1905 )
1906 except ApiException as e:
1907 if e.status != 409:
1908 raise
1909 if cluster_scoped:
1910 custom_api.patch_cluster_custom_object(
1911 group, version, plural, name, body=doc
1912 )
1913 else:
1914 custom_api.patch_namespaced_custom_object(
1915 group, version, namespace, plural, name, body=doc
1916 )
1918 elif kind == "StorageClass":
1919 storage_v1 = client.StorageV1Api()
1920 try:
1921 storage_v1.create_storage_class(body=doc)
1922 except ApiException as e:
1923 if e.status == 409:
1924 # StorageClass already exists - skip patching as most fields are immutable
1925 logger.info(f"StorageClass {name} already exists, skipping update")
1926 else:
1927 raise
1928 elif kind == "PriorityClass":
1929 scheduling_v1 = client.SchedulingV1Api()
1930 try:
1931 scheduling_v1.create_priority_class(body=doc)
1932 except ApiException as e:
1933 if e.status == 409:
1934 # Converge the mutable fields (description,
1935 # labels, globalDefault). ``value`` and
1936 # ``preemptionPolicy`` are immutable: patching
1937 # them with an unchanged value is a no-op,
1938 # while a genuine change fails loudly (422)
1939 # instead of silently keeping the old
1940 # priority — delete the class and redeploy to
1941 # change a value.
1942 scheduling_v1.patch_priority_class(name, body=doc)
1943 else:
1944 raise
1946 elif kind == "PersistentVolume":
1947 try:
1948 v1.create_persistent_volume(body=doc)
1949 except ApiException as e:
1950 if e.status == 409:
1951 # PVs have immutable spec fields. Check if the existing PV
1952 # matches — if so, skip. If the volumeHandle changed (new
1953 # FSx file system), force-remove the old PV and recreate.
1954 existing = v1.read_persistent_volume(name)
1955 existing_handle = (
1956 existing.spec.csi.volume_handle if existing.spec.csi else None
1957 )
1958 new_handle = doc.get("spec", {}).get("csi", {}).get("volumeHandle")
1960 if existing_handle == new_handle:
1961 logger.info(f"PersistentVolume {name} unchanged, skipping")
1962 else:
1963 logger.info(
1964 f"PersistentVolume {name} volumeHandle changed "
1965 f"({existing_handle} → {new_handle}), recreating"
1966 )
1967 # Remove the protection finalizer so the PV can be deleted
1968 # even while bound to a PVC
1969 v1.patch_persistent_volume(
1970 name,
1971 body={"metadata": {"finalizers": None}},
1972 )
1973 v1.delete_persistent_volume(name)
1974 # Wait for the PV to actually disappear
1975 import time as _time
1977 _pv_iterations = max(
1978 1,
1979 PV_PVC_DELETE_WAIT_SECONDS
1980 // PV_PVC_DELETE_POLL_INTERVAL_SECONDS,
1981 )
1982 for _wait in range(_pv_iterations):
1983 try:
1984 v1.read_persistent_volume(name)
1985 _time.sleep(PV_PVC_DELETE_POLL_INTERVAL_SECONDS)
1986 except ApiException as read_e:
1987 if read_e.status == 404:
1988 break
1989 raise
1990 v1.create_persistent_volume(body=doc)
1991 else:
1992 raise
1994 elif kind == "PersistentVolumeClaim":
1995 try:
1996 v1.create_namespaced_persistent_volume_claim(namespace, body=doc)
1997 except ApiException as e:
1998 if e.status == 409:
1999 # Check if the PVC is in Lost state (bound PV was recreated).
2000 # A Lost PVC can't be patched back to health — it must be
2001 # deleted and recreated so it binds to the new PV.
2002 existing_pvc = v1.read_namespaced_persistent_volume_claim(
2003 name, namespace
2004 )
2005 if existing_pvc.status.phase == "Lost":
2006 logger.info(
2007 f"PVC {namespace}/{name} is Lost (bound PV was "
2008 f"recreated), deleting and recreating"
2009 )
2010 v1.delete_namespaced_persistent_volume_claim(name, namespace)
2011 import time as _time
2013 _pvc_iterations = max(
2014 1,
2015 PV_PVC_DELETE_WAIT_SECONDS
2016 // PV_PVC_DELETE_POLL_INTERVAL_SECONDS,
2017 )
2018 for _wait in range(_pvc_iterations):
2019 try:
2020 v1.read_namespaced_persistent_volume_claim(
2021 name, namespace
2022 )
2023 _time.sleep(PV_PVC_DELETE_POLL_INTERVAL_SECONDS)
2024 except ApiException as read_e:
2025 if read_e.status == 404:
2026 break
2027 raise
2028 v1.create_namespaced_persistent_volume_claim(
2029 namespace, body=doc
2030 )
2031 else:
2032 v1.patch_namespaced_persistent_volume_claim(
2033 name, namespace, body=doc
2034 )
2035 else:
2036 raise
2038 elif kind == "NodePool":
2039 # Karpenter NodePool CRD
2040 group = "karpenter.sh"
2041 version = api_version.split("/")[-1] if "/" in api_version else "v1"
2042 plural = "nodepools"
2043 try:
2044 custom_api.create_cluster_custom_object(
2045 group, version, plural, body=doc
2046 )
2047 except ApiException as e:
2048 if e.status == 409:
2049 custom_api.patch_cluster_custom_object(
2050 group, version, plural, name, body=doc
2051 )
2052 else:
2053 raise
2055 elif kind == "EC2NodeClass":
2056 # Karpenter EC2NodeClass CRD
2057 group = "karpenter.k8s.aws"
2058 version = api_version.split("/")[-1] if "/" in api_version else "v1"
2059 plural = "ec2nodeclasses"
2060 try:
2061 custom_api.create_cluster_custom_object(
2062 group, version, plural, body=doc
2063 )
2064 except ApiException as e:
2065 if e.status == 409:
2066 custom_api.patch_cluster_custom_object(
2067 group, version, plural, name, body=doc
2068 )
2069 else:
2070 raise
2072 elif kind == "APIService":
2073 # Kubernetes API aggregation layer
2074 api_reg_v1 = client.ApiregistrationV1Api()
2075 try:
2076 api_reg_v1.create_api_service(body=doc)
2077 except ApiException as e:
2078 if e.status == 409:
2079 api_reg_v1.patch_api_service(name, body=doc)
2080 else:
2081 raise
2083 elif kind == "ValidatingAdmissionPolicy":
2084 # Built-in and cluster-scoped: the fence on who may
2085 # obtain a certificate from GCO's internal CA
2086 # (08-internal-ca-issuance.yaml) and the write fence on
2087 # GCO's objects in the tenant namespaces
2088 # (09-tenant-write-fence.yaml). Both are base-pass
2089 # objects, so they are enforced before the post-Helm
2090 # pass creates the CA and its leaves.
2091 admission_v1 = client.AdmissionregistrationV1Api()
2092 try:
2093 admission_v1.create_validating_admission_policy(body=doc)
2094 except ApiException as e:
2095 if e.status == 409:
2096 admission_v1.patch_validating_admission_policy(name, body=doc)
2097 else:
2098 raise
2100 elif kind == "ValidatingAdmissionPolicyBinding":
2101 admission_v1 = client.AdmissionregistrationV1Api()
2102 try:
2103 admission_v1.create_validating_admission_policy_binding(body=doc)
2104 except ApiException as e:
2105 if e.status == 409:
2106 admission_v1.patch_validating_admission_policy_binding(
2107 name, body=doc
2108 )
2109 else:
2110 raise
2112 elif kind == "CustomResourceDefinition":
2113 api_extensions_v1 = client.ApiextensionsV1Api()
2114 try:
2115 api_extensions_v1.create_custom_resource_definition(body=doc)
2116 except ApiException as e:
2117 if e.status == 409:
2118 api_extensions_v1.patch_custom_resource_definition(name, body=doc)
2119 else:
2120 raise
2122 elif kind == "DeviceClass":
2123 # Kubernetes DRA DeviceClass (resource.k8s.io)
2124 group = "resource.k8s.io"
2125 version = api_version.split("/")[-1] if "/" in api_version else "v1"
2126 plural = "deviceclasses"
2127 try:
2128 custom_api.create_cluster_custom_object(
2129 group, version, plural, body=doc
2130 )
2131 except ApiException as e:
2132 if e.status == 409:
2133 custom_api.patch_cluster_custom_object(
2134 group, version, plural, name, body=doc
2135 )
2136 else:
2137 raise
2138 elif kind == "Token":
2139 # slurm-operator Token: the operator mints a Slurm
2140 # auth/jwt token into a Secret and refreshes it
2141 # (post-helm-slurm-api-access.yaml). The CRD is
2142 # registered by the slinky-slurm-operator chart, so
2143 # Tokens land in the post-Helm pass.
2144 group = "slinky.slurm.net"
2145 version = api_version.split("/")[-1] if "/" in api_version else "v1beta1"
2146 plural = "tokens"
2147 try:
2148 custom_api.create_namespaced_custom_object(
2149 group, version, namespace, plural, body=doc
2150 )
2151 except ApiException as e:
2152 if e.status == 409:
2153 custom_api.patch_namespaced_custom_object(
2154 group, version, namespace, plural, name, body=doc
2155 )
2156 else:
2157 raise
2158 elif kind == "ClusterTrainingRuntime":
2159 # Kubeflow Trainer v2 runtime blueprint (cluster-scoped;
2160 # the CRD is registered by the kubeflow-trainer chart, so
2161 # the shipped torch-distributed runtime lands in the
2162 # post-Helm pass).
2163 group = "trainer.kubeflow.org"
2164 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
2165 plural = "clustertrainingruntimes"
2166 try:
2167 custom_api.create_cluster_custom_object(
2168 group, version, plural, body=doc
2169 )
2170 except ApiException as e:
2171 if e.status == 409:
2172 custom_api.patch_cluster_custom_object(
2173 group, version, plural, name, body=doc
2174 )
2175 else:
2176 raise
2178 elif kind == "NetworkPolicy":
2179 try:
2180 networking_v1.create_namespaced_network_policy(namespace, body=doc)
2181 except ApiException as e:
2182 if e.status == 409:
2183 networking_v1.patch_namespaced_network_policy(
2184 name, namespace, body=doc
2185 )
2186 else:
2187 raise
2189 elif kind == "ResourceQuota":
2190 try:
2191 v1.create_namespaced_resource_quota(namespace, body=doc)
2192 except ApiException as e:
2193 if e.status == 409:
2194 v1.patch_namespaced_resource_quota(name, namespace, body=doc)
2195 else:
2196 raise
2198 elif kind == "LimitRange":
2199 try:
2200 v1.create_namespaced_limit_range(namespace, body=doc)
2201 except ApiException as e:
2202 if e.status == 409:
2203 v1.patch_namespaced_limit_range(name, namespace, body=doc)
2204 else:
2205 raise
2207 elif kind == "ScaledJob":
2208 # KEDA ScaledJob CRD
2209 group = "keda.sh"
2210 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
2211 plural = "scaledjobs"
2212 try:
2213 custom_api.create_namespaced_custom_object(
2214 group, version, namespace, plural, body=doc
2215 )
2216 except ApiException as e:
2217 if e.status == 409:
2218 custom_api.patch_namespaced_custom_object(
2219 group, version, namespace, plural, name, body=doc
2220 )
2221 else:
2222 raise
2224 elif kind == "ScaledObject":
2225 # KEDA ScaledObject CRD
2226 group = "keda.sh"
2227 version = api_version.split("/")[-1] if "/" in api_version else "v1alpha1"
2228 plural = "scaledobjects"
2229 try:
2230 custom_api.create_namespaced_custom_object(
2231 group, version, namespace, plural, body=doc
2232 )
2233 except ApiException as e:
2234 if e.status == 409:
2235 custom_api.patch_namespaced_custom_object(
2236 group, version, namespace, plural, name, body=doc
2237 )
2238 else:
2239 raise
2241 elif kind in _MONITORING_CUSTOM_OBJECT_PLURALS:
2242 # Prometheus Operator CRDs registered by the
2243 # kube-prometheus-stack chart, so these land in the
2244 # post-Helm pass. GCO uses ServiceMonitors for
2245 # components fronted by a Service (schedulers, DCGM,
2246 # slurmctld), PodMonitors for its own multi-replica
2247 # services, and PrometheusRules for alerting rules.
2248 group = "monitoring.coreos.com"
2249 version = api_version.split("/")[-1] if "/" in api_version else "v1"
2250 plural = _MONITORING_CUSTOM_OBJECT_PLURALS[kind]
2251 try:
2252 custom_api.create_namespaced_custom_object(
2253 group, version, namespace, plural, body=doc
2254 )
2255 except ApiException as e:
2256 if e.status == 409:
2257 custom_api.patch_namespaced_custom_object(
2258 group, version, namespace, plural, name, body=doc
2259 )
2260 else:
2261 raise
2263 else:
2264 # Defensive only: plan_manifests rejects unsupported kinds.
2265 raise ValueError(f"Planner admitted unsupported kind: {kind}")
2267 applied_count += 1
2268 logger.info(f"✓ Applied {kind}/{name}")
2270 except ApiException as e:
2271 logger.error(
2272 f"API error applying {kind}/{name}: {e.status} - {e.reason}"
2273 f"{_api_error_detail(e)}"
2274 )
2275 failed.append(f"{filename}:{kind}/{name}")
2277 except Exception as e:
2278 logger.error(f"Failed to apply {filename}: {e}")
2279 failed.append(filename)
2281 # Invariant, not a reachable path: every iteration above either increments
2282 # applied_count or appends to failed, so a clean pass always applies exactly
2283 # expected_count resources. Kept as a tripwire against future edits to the
2284 # loop that break that accounting.
2285 if not failed and applied_count != expected_count: # pragma: no cover - loop invariant
2286 raise RuntimeError(
2287 f"Manifest apply count mismatch: expected={expected_count} applied={applied_count}"
2288 )
2290 # Restart deployments and verify credentials only on the main (full) pass,
2291 # not on the post-Helm pass
2292 if post_helm:
2293 return {
2294 "AppliedCount": applied_count,
2295 "ExpectedCount": expected_count,
2296 "ExpectedResources": expected_resources,
2297 "FailedCount": len(failed),
2298 "SkippedCount": len(skipped),
2299 "Failed": ",".join(failed) if failed else "None",
2300 "Skipped": ",".join(skipped) if skipped else "None",
2301 "PrunedCount": len(pruned),
2302 "Pruned": ",".join(pruned) if pruned else "None",
2303 "PruneFailures": ",".join(prune_failures) if prune_failures else "None",
2304 }
2306 # GCO Deployments already roll exactly once through their
2307 # gco.aws/deployment-timestamp pod-template annotation (and any image patch).
2308 # Do not immediately add kubectl.kubernetes.io/restartedAt: a second
2309 # back-to-back revision can pin maxUnavailable=0/maxSurge=1 Deployments at
2310 # their surge ceiling while the first revision is still converging.
2312 # Restart the EFS CSI controller so it picks up the IRSA role-ARN
2313 # annotation that the EKS addon update patched onto its service account.
2314 #
2315 # Background: the managed aws-efs-csi-driver addon creates the
2316 # efs-csi-controller-sa ServiceAccount and the controller Deployment in
2317 # parallel. Our stack later calls UpdateAddon with a
2318 # serviceAccountRoleArn, which patches the SA's eks.amazonaws.com/role-arn
2319 # annotation — but EKS does NOT restart the controller pods. The existing
2320 # pods keep their original (un-mutated) pod spec: no AWS_ROLE_ARN,
2321 # no AWS_WEB_IDENTITY_TOKEN_FILE. They then fall back to IMDS for
2322 # credentials, which EKS Auto Mode blocks at the pod network level
2323 # (hop-limit / security policy), causing every EFS CreateAccessPoint
2324 # call to fail with HTTP 401. The visible symptom is a PVC that stays
2325 # Pending forever with "no EC2 IMDS role found".
2326 #
2327 # Restarting the deployment forces a new pod template to go through the
2328 # EKS Pod Identity / IRSA mutating webhook, which sees the annotation
2329 # this time and injects the projected-token volume and env vars.
2330 #
2331 # The same pattern applies to:
2332 # - aws-fsx-csi-driver: fsx-csi-controller Deployment +
2333 # fsx-csi-node DaemonSet
2334 # - amazon-cloudwatch-observability: cloudwatch-agent DaemonSet
2335 # - aws-efs-csi-driver: efs-csi-node DaemonSet (the controller already
2336 # above)
2337 #
2338 # Deployments/DaemonSets that don't exist on this cluster (e.g. FSx when
2339 # fsx_lustre.enabled=false) return 404 and are skipped gracefully — see
2340 # the 404 branch in restart_deployments/restart_daemonsets.
2341 kube_system_deployments = ["efs-csi-controller", "fsx-csi-controller"]
2342 logger.info(f"Restarting deployments in kube-system: {kube_system_deployments}")
2343 ks_deploy_restart = restart_deployments("kube-system", kube_system_deployments)
2345 kube_system_daemonsets = ["efs-csi-node", "fsx-csi-node"]
2346 logger.info(f"Restarting daemonsets in kube-system: {kube_system_daemonsets}")
2347 ks_ds_restart = restart_daemonsets("kube-system", kube_system_daemonsets)
2349 # CloudWatch agent daemonset runs in its own namespace when the
2350 # amazon-cloudwatch-observability addon is installed.
2351 cw_daemonsets = ["cloudwatch-agent"]
2352 logger.info(f"Restarting daemonsets in amazon-cloudwatch: {cw_daemonsets}")
2353 cw_ds_restart = restart_daemonsets("amazon-cloudwatch", cw_daemonsets)
2355 # Verify IAM credentials are available for workloads
2356 # Check that the projected service-account token volume is configured
2357 # on key deployments — if missing, IRSA won't work. Scoped to the
2358 # Deployments this pass planned so a gated-off service is not reported.
2359 credential_warnings = _verify_workload_credentials(apps_v1, planned_resources)
2361 # Combine the restart results for the return payload.
2362 all_restarted = (
2363 ks_deploy_restart["restarted"] + ks_ds_restart["restarted"] + cw_ds_restart["restarted"]
2364 )
2366 return {
2367 "AppliedCount": applied_count,
2368 "ExpectedCount": expected_count,
2369 "ExpectedResources": expected_resources,
2370 "FailedCount": len(failed),
2371 "SkippedCount": len(skipped),
2372 "Failed": ",".join(failed) if failed else "None",
2373 "Skipped": ",".join(skipped) if skipped else "None",
2374 "RestartedDeployments": (",".join(all_restarted) if all_restarted else "None"),
2375 "CredentialWarnings": ",".join(credential_warnings) if credential_warnings else "None",
2376 "PrunedCount": len(pruned),
2377 "Pruned": ",".join(pruned) if pruned else "None",
2378 "PruneFailures": ",".join(prune_failures) if prune_failures else "None",
2379 }
2382def _delete_gateway_resources(cluster_name: str, region: str) -> dict[str, Any]:
2383 """Delete Gateway objects while the LBC controller is still running.
2385 The Gateway is deleted only after its routes, and every object is polled to
2386 exact absence. Waiting for the Gateway's controller finalizer is what proves
2387 the ALB has been removed before the LBC Helm release is uninstalled.
2388 """
2389 configure_k8s_client(cluster_name, region)
2390 custom_api = client.CustomObjectsApi()
2391 namespace = "gco-system"
2392 delete_options = client.V1DeleteOptions(propagation_policy="Foreground")
2393 resources = (
2394 ("HTTPRoute", "gco-routes"),
2395 ("Gateway", "gco-gateway"),
2396 ("LoadBalancerConfiguration", "gco-gateway-load-balancer"),
2397 ("TargetGroupConfiguration", "gco-health-monitor-target-group"),
2398 ("TargetGroupConfiguration", "gco-manifest-processor-target-group"),
2399 ("TargetGroupConfiguration", "gco-inference-proxy-target-group"),
2400 ("TargetGroupConfiguration", "gco-default-target-group"),
2401 ("GatewayClass", "gco-aws-alb"),
2402 )
2403 deleted: list[str] = []
2404 deadline = time.monotonic() + _GATEWAY_DELETE_WAIT_SECONDS
2406 for kind, name in resources:
2407 group, version, plural, cluster_scoped = _GATEWAY_CUSTOM_OBJECTS[kind]
2408 label = f"{kind}/{name}" if cluster_scoped else f"{kind}/{namespace}/{name}"
2409 try:
2410 if cluster_scoped:
2411 custom_api.delete_cluster_custom_object(
2412 group, version, plural, name, body=delete_options
2413 )
2414 else:
2415 custom_api.delete_namespaced_custom_object(
2416 group, version, namespace, plural, name, body=delete_options
2417 )
2418 except ApiException as exc:
2419 if exc.status != 404:
2420 raise RuntimeError(
2421 f"failed to delete {label}: Kubernetes API {exc.status} ({exc.reason})"
2422 ) from exc
2423 deleted.append(f"{label}:already-absent")
2424 continue
2426 while True:
2427 try:
2428 if cluster_scoped:
2429 custom_api.get_cluster_custom_object(group, version, plural, name)
2430 else:
2431 custom_api.get_namespaced_custom_object(group, version, namespace, plural, name)
2432 except ApiException as exc:
2433 if exc.status == 404:
2434 deleted.append(label)
2435 break
2436 raise RuntimeError(
2437 f"failed while waiting for {label} deletion: "
2438 f"Kubernetes API {exc.status} ({exc.reason})"
2439 ) from exc
2440 if time.monotonic() >= deadline:
2441 raise RuntimeError(
2442 "timed out waiting for the complete Gateway resource set to delete "
2443 f"within {_GATEWAY_DELETE_WAIT_SECONDS}s; last resource was {label}"
2444 )
2445 time.sleep(_GATEWAY_DELETE_POLL_SECONDS)
2447 return {"status": "deleted", "DeletedCount": len(deleted), "Deleted": deleted}
2450def _as_plain_dict(value: Any) -> dict[str, Any]:
2451 """Convert a DynamicClient response to a plain mapping."""
2452 if isinstance(value, dict):
2453 return value
2454 converter = getattr(value, "to_dict", None)
2455 if callable(converter):
2456 converted = converter()
2457 if isinstance(converted, dict):
2458 return converted
2459 try:
2460 converted = dict(value)
2461 except (TypeError, ValueError) as exc:
2462 raise TypeError(f"Kubernetes response is not a mapping: {type(value).__name__}") from exc
2463 return converted
2466def _condition_matches(condition: dict[str, Any], expected_status: str) -> bool:
2467 value = condition.get("status")
2468 if isinstance(value, bool):
2469 value = "True" if value else "False"
2470 return str(value).lower() == expected_status.lower()
2473def _conditions(status: dict[str, Any]) -> list[dict[str, Any]]:
2474 value = status.get("conditions", [])
2475 if not isinstance(value, list):
2476 return []
2477 return [condition for condition in value if isinstance(condition, dict)]
2480def _required_condition_failure(
2481 status: dict[str, Any],
2482 condition_type: str,
2483) -> str | None:
2484 matching = [item for item in _conditions(status) if item.get("type") == condition_type]
2485 if any(_condition_matches(item, "True") for item in matching):
2486 return None
2487 if matching:
2488 details = matching[-1].get("message") or matching[-1].get("reason") or "status is not True"
2489 return f"condition {condition_type} is not True ({details})"
2490 return f"condition {condition_type} is missing"
2493def _generation_failure(resource: dict[str, Any]) -> str | None:
2494 metadata_value = resource.get("metadata")
2495 status_value = resource.get("status")
2496 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2497 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2498 generation = metadata.get("generation")
2499 observed = status.get("observedGeneration")
2500 if (
2501 not isinstance(generation, int)
2502 or isinstance(generation, bool)
2503 or not isinstance(observed, int)
2504 or isinstance(observed, bool)
2505 or observed != generation
2506 ):
2507 return f"generation not observed (generation={generation}, observedGeneration={observed})"
2508 return None
2511def _replica_failure(
2512 status: dict[str, Any],
2513 desired: int,
2514 fields: tuple[str, ...],
2515) -> str | None:
2516 for field in fields:
2517 # Kubernetes omits optional zero-valued counters. Treat omission as
2518 # zero while still rejecting malformed non-integer values.
2519 value = status.get(field, 0)
2520 if not isinstance(value, int) or isinstance(value, bool) or value != desired:
2521 return f"replicas not converged ({field}={value}, desired={desired})"
2522 unavailable = status.get("unavailableReplicas", status.get("numberUnavailable", 0))
2523 if not isinstance(unavailable, int) or isinstance(unavailable, bool) or unavailable != 0:
2524 return f"replicas unavailable ({unavailable})"
2525 return None
2528def _current_condition_failure(
2529 conditions: list[dict[str, Any]],
2530 condition_type: str,
2531 generation: Any,
2532 context: str,
2533) -> str | None:
2534 """Require a True condition that observed the object's current generation."""
2535 current = [
2536 condition
2537 for condition in conditions
2538 if condition.get("type") == condition_type
2539 and condition.get("observedGeneration") == generation
2540 ]
2541 if any(_condition_matches(condition, "True") for condition in current):
2542 return None
2543 matching = [condition for condition in conditions if condition.get("type") == condition_type]
2544 if current:
2545 detail = current[-1].get("message") or current[-1].get("reason") or "status is not True"
2546 return f"{context} condition {condition_type} is not True ({detail})"
2547 if matching:
2548 observed = matching[-1].get("observedGeneration")
2549 return (
2550 f"{context} condition {condition_type} is stale "
2551 f"(generation={generation}, observedGeneration={observed})"
2552 )
2553 return f"{context} condition {condition_type} is missing"
2556def _gateway_api_readiness_failure(kind: str, resource: dict[str, Any]) -> str | None:
2557 """Return strict, generation-aware Gateway API readiness evidence."""
2558 metadata_value = resource.get("metadata")
2559 spec_value = resource.get("spec")
2560 status_value = resource.get("status")
2561 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2562 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2563 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2564 generation = metadata.get("generation")
2565 if not isinstance(generation, int) or isinstance(generation, bool):
2566 return f"invalid metadata.generation ({generation})"
2568 if kind == "GatewayClass":
2569 return _current_condition_failure(
2570 _conditions(status), "Accepted", generation, "GatewayClass"
2571 )
2573 if kind == "Gateway":
2574 for condition_type in ("Accepted", "Programmed"):
2575 failure = _current_condition_failure(
2576 _conditions(status), condition_type, generation, "Gateway"
2577 )
2578 if failure:
2579 return failure
2580 addresses = status.get("addresses", [])
2581 if not isinstance(addresses, list) or not any(
2582 isinstance(address, dict) and str(address.get("value", "")).strip()
2583 for address in addresses
2584 ):
2585 return "Gateway has no address"
2587 intended_listeners: set[str] = {
2588 listener["name"]
2589 for listener in spec.get("listeners", [])
2590 if isinstance(listener, dict) and isinstance(listener.get("name"), str)
2591 }
2592 listener_statuses = status.get("listeners", [])
2593 if not isinstance(listener_statuses, list):
2594 return "Gateway listener status is missing"
2595 by_name = {
2596 listener.get("name"): listener
2597 for listener in listener_statuses
2598 if isinstance(listener, dict) and isinstance(listener.get("name"), str)
2599 }
2600 for listener_name in sorted(intended_listeners):
2601 listener = by_name.get(listener_name)
2602 if not isinstance(listener, dict):
2603 return f"Gateway listener {listener_name!r} status is missing"
2604 listener_conditions = listener.get("conditions", [])
2605 if not isinstance(listener_conditions, list):
2606 return f"Gateway listener {listener_name!r} conditions are missing"
2607 plain_conditions = [item for item in listener_conditions if isinstance(item, dict)]
2608 for condition_type in ("Accepted", "ResolvedRefs", "Programmed"):
2609 failure = _current_condition_failure(
2610 plain_conditions,
2611 condition_type,
2612 generation,
2613 f"Gateway listener {listener_name!r}",
2614 )
2615 if failure:
2616 return failure
2617 return None
2619 if kind == "HTTPRoute":
2620 namespace = str(metadata.get("namespace", "default"))
2622 def parent_key(reference: dict[str, Any]) -> tuple[str, str, str, str, str]:
2623 return (
2624 str(reference.get("group", "gateway.networking.k8s.io")),
2625 str(reference.get("kind", "Gateway")),
2626 str(reference.get("namespace", namespace)),
2627 str(reference.get("name", "")),
2628 str(reference.get("sectionName", "")),
2629 )
2631 intended: set[tuple[str, str, str, str, str]] = {
2632 parent_key(parent) for parent in spec.get("parentRefs", []) if isinstance(parent, dict)
2633 }
2634 parents = status.get("parents", [])
2635 if not isinstance(parents, list):
2636 return "HTTPRoute parent status is missing"
2637 live_by_ref = {
2638 parent_key(parent["parentRef"]): parent
2639 for parent in parents
2640 if isinstance(parent, dict) and isinstance(parent.get("parentRef"), dict)
2641 }
2642 for reference in sorted(intended):
2643 parent = live_by_ref.get(reference)
2644 if not isinstance(parent, dict):
2645 return f"HTTPRoute intended parent {reference} status is missing"
2646 parent_conditions = parent.get("conditions", [])
2647 if not isinstance(parent_conditions, list):
2648 return f"HTTPRoute intended parent {reference} conditions are missing"
2649 plain_conditions = [item for item in parent_conditions if isinstance(item, dict)]
2650 for condition_type in ("Accepted", "ResolvedRefs"):
2651 failure = _current_condition_failure(
2652 plain_conditions,
2653 condition_type,
2654 generation,
2655 f"HTTPRoute parent {reference}",
2656 )
2657 if failure:
2658 return failure
2659 return None
2661 return None
2664def _resource_readiness_failure(kind: str, resource: dict[str, Any]) -> str | None:
2665 """Return an actionable readiness reason, or None when the object is ready."""
2666 metadata_value = resource.get("metadata")
2667 spec_value = resource.get("spec")
2668 status_value = resource.get("status")
2669 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2670 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2671 status: dict[str, Any] = status_value if isinstance(status_value, dict) else {}
2673 if metadata.get("deletionTimestamp"):
2674 return f"object is terminating since {metadata['deletionTimestamp']}"
2676 for condition in _conditions(status):
2677 if condition.get("type") in {"Ready", "Available"} and _condition_matches(
2678 condition, "False"
2679 ):
2680 detail = condition.get("message") or condition.get("reason") or "no detail"
2681 return f"condition {condition.get('type')} is False ({detail})"
2683 if kind in {"GatewayClass", "Gateway", "HTTPRoute"}:
2684 return _gateway_api_readiness_failure(kind, resource)
2686 if kind in {"Deployment", "StatefulSet"}:
2687 generation_failure = _generation_failure(resource)
2688 if generation_failure:
2689 return generation_failure
2690 desired_value = spec.get("replicas", 1)
2691 if not isinstance(desired_value, int) or isinstance(desired_value, bool):
2692 return f"invalid desired replica count ({desired_value})"
2693 desired = desired_value
2694 fields = (
2695 ("replicas", "updatedReplicas", "readyReplicas", "availableReplicas")
2696 if kind == "Deployment"
2697 else ("currentReplicas", "updatedReplicas", "readyReplicas")
2698 )
2699 return _replica_failure(status, desired, fields)
2701 if kind == "DaemonSet":
2702 generation_failure = _generation_failure(resource)
2703 if generation_failure:
2704 return generation_failure
2705 desired_value = status.get("desiredNumberScheduled")
2706 if not isinstance(desired_value, int) or isinstance(desired_value, bool):
2707 return f"invalid desiredNumberScheduled ({desired_value})"
2708 desired = desired_value
2709 replica_failure = _replica_failure(
2710 status,
2711 desired,
2712 (
2713 "currentNumberScheduled",
2714 "updatedNumberScheduled",
2715 "numberReady",
2716 "numberAvailable",
2717 ),
2718 )
2719 if replica_failure:
2720 return replica_failure
2721 misscheduled = status.get("numberMisscheduled", 0)
2722 if not isinstance(misscheduled, int) or isinstance(misscheduled, bool) or misscheduled != 0:
2723 return f"pods are misscheduled (numberMisscheduled={misscheduled})"
2724 return None
2726 if kind == "Job":
2727 failed = _required_condition_failure(status, "Failed")
2728 if failed is None:
2729 return "condition Failed is True"
2730 return _required_condition_failure(status, "Complete")
2732 if kind == "Pod":
2733 return _required_condition_failure(status, "Ready")
2735 if kind == "PersistentVolumeClaim":
2736 phase = status.get("phase")
2737 return None if phase == "Bound" else f"PVC phase is {phase!r}, expected 'Bound'"
2739 if kind == "PersistentVolume":
2740 phase = status.get("phase")
2741 return (
2742 None
2743 if phase in {"Bound", "Available"}
2744 else f"PV phase is {phase!r}, expected 'Bound' or 'Available'"
2745 )
2747 if kind == "CustomResourceDefinition":
2748 return _required_condition_failure(status, "Established")
2750 if kind == "APIService":
2751 return _required_condition_failure(status, "Available")
2753 if kind == "HorizontalPodAutoscaler":
2754 for condition_type in ("AbleToScale", "ScalingActive"):
2755 failure = _required_condition_failure(status, condition_type)
2756 if failure:
2757 return failure
2758 return None
2760 if kind == "PodDisruptionBudget":
2761 generation_failure = _generation_failure(resource)
2762 if generation_failure:
2763 return generation_failure
2764 current_healthy = status.get("currentHealthy")
2765 desired_healthy = status.get("desiredHealthy")
2766 if (
2767 not isinstance(current_healthy, int)
2768 or isinstance(current_healthy, bool)
2769 or not isinstance(desired_healthy, int)
2770 or isinstance(desired_healthy, bool)
2771 ):
2772 return (
2773 "invalid PDB health "
2774 f"(currentHealthy={current_healthy}, desiredHealthy={desired_healthy})"
2775 )
2776 if current_healthy < desired_healthy:
2777 return (
2778 "PDB health below target "
2779 f"(currentHealthy={current_healthy}, desiredHealthy={desired_healthy})"
2780 )
2781 return None
2783 if kind in {"Certificate", "ClusterIssuer", "Issuer", "Bundle"}:
2784 # cert-manager stamps observedGeneration on these Ready conditions,
2785 # and trust-manager on a Bundle's Synced condition, so a condition
2786 # left over from a previous spec (a re-pointed issuerRef, a new
2787 # dnsName, another bundle source) is stale rather than ready.
2788 generation = metadata.get("generation")
2789 if not isinstance(generation, int) or isinstance(generation, bool):
2790 return f"invalid metadata.generation ({generation})"
2791 return _current_condition_failure(
2792 _conditions(status),
2793 "Synced" if kind == "Bundle" else "Ready",
2794 generation,
2795 kind,
2796 )
2798 if kind in {"NodePool", "EC2NodeClass", "ScaledJob", "ScaledObject"}:
2799 return _required_condition_failure(status, "Ready")
2801 # Static configuration and RBAC resources have no rollout contract. Their
2802 # exact-object existence is sufficient unless a generic Ready/Available
2803 # condition explicitly reported False above. That includes the
2804 # ValidatingAdmissionPolicy and its binding: the API server rejects a
2805 # policy whose CEL does not compile, and enforces a stored one without
2806 # waiting for the status controller (whose type check is advisory).
2807 return None
2810def _dynamic_resource(
2811 dynamic_client: Any,
2812 cache: dict[tuple[str, str], Any],
2813 api_version: str,
2814 kind: str,
2815) -> Any:
2816 key = (api_version, kind)
2817 if key not in cache:
2818 cache[key] = dynamic_client.resources.get(api_version=api_version, kind=kind)
2819 return cache[key]
2822def _service_endpoint_failure(
2823 dynamic_client: Any,
2824 cache: dict[tuple[str, str], Any],
2825 planned_resource: dict[str, Any],
2826) -> str | None:
2827 document_value = planned_resource["document"]
2828 document: dict[str, Any] = document_value if isinstance(document_value, dict) else {}
2829 metadata_value = document.get("metadata")
2830 document_metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2831 annotations_value = document_metadata.get("annotations")
2832 annotations: dict[str, Any] = annotations_value if isinstance(annotations_value, dict) else {}
2833 if str(annotations.get(_ALLOW_EMPTY_ENDPOINTS_ANNOTATION)).lower() == "true":
2834 # Services backing accelerator-scheduled DaemonSets (for example the
2835 # DCGM exporter) legitimately have zero endpoints until the first GPU
2836 # node is provisioned; existence is their readiness contract.
2837 return None
2838 spec_value = document.get("spec")
2839 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2840 selector = spec.get("selector")
2841 if not isinstance(selector, dict) or not selector:
2842 return None
2844 endpoint_slices = _dynamic_resource(
2845 dynamic_client,
2846 cache,
2847 "discovery.k8s.io/v1",
2848 "EndpointSlice",
2849 )
2850 response = endpoint_slices.get(
2851 namespace=planned_resource["namespace"],
2852 label_selector=f"kubernetes.io/service-name={planned_resource['name']}",
2853 )
2854 items = _as_plain_dict(response).get("items", [])
2855 if not isinstance(items, list):
2856 return "EndpointSlice response does not contain an items list"
2857 for item in items:
2858 item_mapping = _as_plain_dict(item)
2859 metadata_value = item_mapping.get("metadata")
2860 metadata: dict[str, Any] = metadata_value if isinstance(metadata_value, dict) else {}
2861 if metadata.get("deletionTimestamp"):
2862 continue
2863 endpoints = item_mapping.get("endpoints", [])
2864 if not isinstance(endpoints, list):
2865 continue
2866 for endpoint in endpoints:
2867 if not isinstance(endpoint, dict):
2868 continue
2869 conditions = endpoint.get("conditions", {})
2870 if not isinstance(conditions, dict):
2871 continue
2872 if conditions.get("ready") is True and conditions.get("terminating") is not True:
2873 return None
2874 return "selector-backed Service has no ready, nonterminating EndpointSlice endpoint"
2877def _certificate_secret_failure(
2878 dynamic_client: Any,
2879 cache: dict[tuple[str, str], Any],
2880 planned_resource: dict[str, Any],
2881 live_resource: dict[str, Any],
2882) -> str | None:
2883 """Require cert-manager's referenced Secret to contain a nonempty TLS keypair."""
2884 spec_value = live_resource.get("spec")
2885 spec: dict[str, Any] = spec_value if isinstance(spec_value, dict) else {}
2886 secret_name = spec.get("secretName")
2887 if not isinstance(secret_name, str) or not secret_name:
2888 return "Certificate spec.secretName is missing"
2889 secret_api = _dynamic_resource(dynamic_client, cache, "v1", "Secret")
2890 response = secret_api.get(
2891 namespace=planned_resource["namespace"],
2892 name=secret_name,
2893 )
2894 secret = _as_plain_dict(response)
2895 data_value = secret.get("data")
2896 data: dict[str, Any] = data_value if isinstance(data_value, dict) else {}
2897 missing = [
2898 key for key in ("tls.crt", "tls.key") if not isinstance(data.get(key), str) or not data[key]
2899 ]
2900 if missing:
2901 return f"Certificate Secret {secret_name!r} has no nonempty {', '.join(missing)}"
2902 return None
2905def _manifest_identity_label(resource: dict[str, Any]) -> str:
2906 return (
2907 f"{resource['apiVersion']}/{resource['kind']}/"
2908 f"{resource['namespace']}/{resource['name']}"
2909 f" [{resource['phase']}:{resource['sourceFile']}]"
2910 )
2913def validate_manifests(
2914 cluster_name: str,
2915 region: str,
2916 manifests_dir: str,
2917 replacements: dict[str, str],
2918 deployment_token: Any = None,
2919) -> dict[str, Any]:
2920 """Validate exact existence and readiness for the complete planned inventory."""
2921 plan = plan_manifests(manifests_dir, replacements)
2922 expected = plan["phases"]["base"] + plan["phases"]["post-helm"]
2923 expected_resources = [_public_manifest_identity(item) for item in expected]
2924 phase_counts = {
2925 phase: {"ExpectedCount": len(plan["phases"][phase]), "ValidatedCount": 0}
2926 for phase in ("base", "post-helm")
2927 }
2929 configure_k8s_client(cluster_name, region)
2930 dynamic_client = dynamic.DynamicClient(client.ApiClient())
2931 resource_cache: dict[tuple[str, str], Any] = {}
2932 validated_resources: list[dict[str, str]] = []
2933 failures: list[str] = []
2934 failure_count = 0
2936 def add_failure(message: str) -> None:
2937 nonlocal failure_count
2938 failure_count += 1
2939 if len(failures) < _MAX_VALIDATION_FAILURES:
2940 failures.append(message[:500])
2942 for planned_resource in expected:
2943 label = _manifest_identity_label(planned_resource)
2944 try:
2945 resource_api = _dynamic_resource(
2946 dynamic_client,
2947 resource_cache,
2948 planned_resource["apiVersion"],
2949 planned_resource["kind"],
2950 )
2951 get_kwargs: dict[str, Any] = {"name": planned_resource["name"]}
2952 if planned_resource["namespace"] != _CLUSTER_SCOPE:
2953 get_kwargs["namespace"] = planned_resource["namespace"]
2954 live_resource = _as_plain_dict(resource_api.get(**get_kwargs))
2955 failure = _resource_readiness_failure(planned_resource["kind"], live_resource)
2956 if failure is None and planned_resource["kind"] == "Service":
2957 failure = _service_endpoint_failure(
2958 dynamic_client,
2959 resource_cache,
2960 planned_resource,
2961 )
2962 if failure is None and planned_resource["kind"] == "Certificate":
2963 failure = _certificate_secret_failure(
2964 dynamic_client,
2965 resource_cache,
2966 planned_resource,
2967 live_resource,
2968 )
2969 if failure:
2970 add_failure(f"{label}: {failure}")
2971 continue
2972 except (NotFoundError, ResourceNotFoundError) as exc:
2973 add_failure(f"{label}: object or API resource not found ({exc})")
2974 continue
2975 except ApiException as exc:
2976 detail = exc.reason or str(exc)
2977 add_failure(f"{label}: Kubernetes API error {exc.status} ({detail})")
2978 continue
2979 except Exception as exc:
2980 add_failure(f"{label}: validation error ({exc})")
2981 continue
2983 identity = _public_manifest_identity(planned_resource)
2984 validated_resources.append(identity)
2985 phase_counts[planned_resource["phase"]]["ValidatedCount"] += 1
2987 if failure_count:
2988 hidden = failure_count - len(failures)
2989 suffix = f"; ... {hidden} additional failure(s)" if hidden else ""
2990 raise RuntimeError(
2991 f"Manifest validation failed: validated={len(validated_resources)} "
2992 f"expected={len(expected)}; " + "; ".join(failures) + suffix
2993 )
2995 return {
2996 "status": "validated",
2997 "DeploymentToken": deployment_token,
2998 "ExpectedCount": len(expected),
2999 "ValidatedCount": len(validated_resources),
3000 "BaseExpectedCount": phase_counts["base"]["ExpectedCount"],
3001 "BaseValidatedCount": phase_counts["base"]["ValidatedCount"],
3002 "PostHelmExpectedCount": phase_counts["post-helm"]["ExpectedCount"],
3003 "PostHelmValidatedCount": phase_counts["post-helm"]["ValidatedCount"],
3004 "PhaseCounts": phase_counts,
3005 "ExpectedResources": expected_resources,
3006 "ValidatedResources": validated_resources,
3007 }
3010def _record_phase_status(phase: str, status: str, message: str) -> None:
3011 """Record a convergence phase's outcome to SSM (best-effort).
3013 Mirrors the helm worker's per-chart status (``_record_addon_status``) so
3014 ``gco stacks addons status`` surfaces the base and post-Helm apply passes
3015 alongside the charts. Writes ``/<project>/addons/<region>/<phase>`` as a
3016 small JSON blob. Failures are swallowed — status reporting must never turn a
3017 successful apply into a failure (or vice versa). Reads PROJECT_NAME / REGION
3018 from the Lambda environment; a no-op if either is unset.
3019 """
3020 project = os.environ.get("PROJECT_NAME")
3021 region = os.environ.get("REGION")
3022 if not project or not region:
3023 return
3024 import contextlib
3025 import time as _time
3027 with contextlib.suppress(Exception):
3028 boto3.client("ssm").put_parameter(
3029 Name=f"/{project}/addons/{region}/{phase}",
3030 Value=json.dumps(
3031 {
3032 "phase": phase,
3033 "status": status,
3034 "message": message[:1024],
3035 "updated_at": int(_time.time()),
3036 }
3037 ),
3038 Type="String",
3039 Overwrite=True,
3040 )
3043def handle_task(event: dict[str, Any]) -> dict[str, Any]:
3044 """Run an apply or exhaustive manifest-validation Step Functions task."""
3045 cluster_name = event["ClusterName"]
3046 region = event["Region"]
3047 replacements = event.get("ImageReplacements", {})
3048 action = event.get("Action", "apply_manifests")
3049 manifests_dir = os.path.join(os.path.dirname(__file__), "manifests")
3051 if action == "delete_gateway_resources":
3052 phase = "gateway-teardown"
3053 try:
3054 result = _delete_gateway_resources(cluster_name, region)
3055 except Exception as exc:
3056 _record_phase_status(phase, "failed", str(exc))
3057 raise
3058 _record_phase_status(
3059 phase,
3060 "deleted",
3061 f"deleted={result['DeletedCount']}",
3062 )
3063 return result
3065 if action == "validate_manifests":
3066 phase = "manifest-validation"
3067 deployment_token = event.get("DeploymentToken")
3068 try:
3069 result = validate_manifests(
3070 cluster_name,
3071 region,
3072 manifests_dir,
3073 replacements,
3074 deployment_token,
3075 )
3076 except Exception as exc:
3077 _record_phase_status(phase, "failed", f"token={deployment_token} {exc}")
3078 raise
3079 _record_phase_status(
3080 phase,
3081 "validated",
3082 f"token={deployment_token} validated={result['ValidatedCount']} "
3083 f"expected={result['ExpectedCount']}",
3084 )
3085 return result
3087 if action != "apply_manifests":
3088 raise ValueError(f"Unsupported task action: {action}")
3090 post_helm = str(event.get("PostHelm", "false")).lower() == "true"
3091 phase = "post-helm-manifests" if post_helm else "base-manifests"
3092 try:
3093 result = apply_manifests(cluster_name, region, manifests_dir, replacements, post_helm)
3094 except Exception as exc:
3095 _record_phase_status(phase, "failed", str(exc))
3096 raise
3097 if result.get("FailedCount", 0):
3098 _record_phase_status(phase, "failed", str(result.get("Failed")))
3099 raise RuntimeError(
3100 f"kubectl apply failed (post_helm={post_helm}, "
3101 f"failed={result.get('FailedCount')}): {result.get('Failed')}"
3102 )
3103 if result.get("ExpectedCount") is not None and result.get("AppliedCount") != result.get(
3104 "ExpectedCount"
3105 ):
3106 message = (
3107 f"apply count mismatch: applied={result.get('AppliedCount')} "
3108 f"expected={result.get('ExpectedCount')}"
3109 )
3110 _record_phase_status(phase, "failed", message)
3111 raise RuntimeError(message)
3112 _record_phase_status(
3113 phase,
3114 "applied",
3115 f"applied={result.get('AppliedCount')} expected={result.get('ExpectedCount')} "
3116 f"skipped={result.get('SkippedCount')}",
3117 )
3118 return result
3121def lambda_handler(event: dict[str, Any], context: Any) -> Any:
3122 """Main Lambda handler.
3124 Two entrypoints share this function:
3126 - **Step Functions task** (the convergence pipeline): the event carries an
3127 ``Action`` key and is dispatched to :func:`handle_task`, which applies the
3128 manifests and raises on any failure.
3129 - **CloudFormation custom resource** (legacy/fallback): the event carries a
3130 ``RequestType`` and the result is POSTed back to CloudFormation.
3131 """
3132 if event.get("Action"):
3133 logger.info(f"Task event: {json.dumps(event)}")
3134 return handle_task(event)
3136 print(f"[HANDLER] Received event type: {event.get('RequestType')}")
3137 logger.info(f"Received event: {json.dumps(event)}")
3139 request_type = event["RequestType"]
3140 physical_resource_id = event.get("PhysicalResourceId", f"kubectl-{event['LogicalResourceId']}")
3142 try:
3143 properties = event["ResourceProperties"]
3144 cluster_name = properties["ClusterName"]
3145 region = properties["Region"]
3147 if request_type == "Create" or request_type == "Update":
3148 manifests_dir = os.path.join(os.path.dirname(__file__), "manifests")
3149 replacements = properties.get("ImageReplacements", {})
3150 # PostHelm: "true" means this is the post-Helm pass — apply only post-helm-* files
3151 post_helm = properties.get("PostHelm", "false").lower() == "true"
3152 response_data = apply_manifests(
3153 cluster_name, region, manifests_dir, replacements, post_helm
3154 )
3155 failed_count = int(response_data.get("FailedCount", 0))
3156 prune_failures = response_data.get("PruneFailures")
3157 if failed_count > 0 or prune_failures not in (None, "", "None"):
3158 raise RuntimeError(
3159 "Manifest application reported failures: "
3160 f"failed_count={failed_count}, failed={response_data.get('Failed', 'unknown')}, "
3161 f"prune_failures={prune_failures or 'None'}"
3162 )
3163 send_response(event, context, SUCCESS, response_data, physical_resource_id)
3165 elif request_type == "Delete":
3166 # Always succeed on delete to prevent stack from getting stuck
3167 skip_deletion = properties.get("SkipDeletionOnStackDelete", "false").lower() == "true"
3168 if skip_deletion:
3169 logger.info("Skipping deletion (SkipDeletionOnStackDelete=true)")
3170 response_data = {"Status": "Deleted"}
3171 send_response(event, context, SUCCESS, response_data, physical_resource_id)
3173 else:
3174 # CloudFormation only ever sends Create/Update/Delete. Anything else
3175 # must still get a FAILED callback: silently returning would leave
3176 # the stack waiting on this resource until its timeout.
3177 raise ValueError(f"Unsupported RequestType: {request_type}")
3179 except Exception as e:
3180 logger.error(f"Error: {e}", exc_info=True)
3181 # On delete, always return success to prevent stack from getting stuck
3182 if request_type == "Delete":
3183 send_response(
3184 event,
3185 context,
3186 SUCCESS,
3187 {"Status": "Forced success on delete"},
3188 physical_resource_id,
3189 )
3190 else:
3191 send_response(event, context, FAILED, {}, physical_resource_id, str(e))