Coverage for lambda / helm-installer / handler.py: 100.00%
1075 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Helm Installer Lambda Handler
4Installs and manages Helm charts on EKS clusters via CloudFormation Custom Resources.
5Supports KEDA and other Helm-based installations.
7Features:
8- Automatic Helm repo management
9- Idempotent install/upgrade operations
10- Configurable chart values via CloudFormation properties
11- EKS authentication via IAM
13Environment Variables:
14 CLUSTER_NAME: Name of the EKS cluster
15 REGION: AWS region
17CloudFormation Properties:
18 ClusterName: EKS cluster name
19 Region: AWS region
20 Charts: Dict of chart configurations to override defaults
21 EnabledCharts: List of chart names to enable (overrides charts.yaml)
22"""
24import base64
25import contextlib
26import copy
27import hashlib
28import json
29import logging
30import os
31import re
32import subprocess
33import tempfile
34import time
35from collections import Counter
36from collections.abc import Iterator
37from dataclasses import dataclass
38from pathlib import Path
39from typing import Any
41import boto3
42import urllib3
43import yaml
45# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
46# Generated at (UTC): 2026-08-30T12:00:00Z
47# Generated from Git commit: 8cb8e766461f5b7f35205e12db980988598cf6d5
48# Flowchart(s) generated from this file:
49# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/helm-installer/handler.lambda_handler.html``
50# (PNG: ``diagrams/code_diagrams/lambda/helm-installer/handler.lambda_handler.png``)
51# * ``handle_task`` -> ``diagrams/code_diagrams/lambda/helm-installer/handler.handle_task.html``
52# (PNG: ``diagrams/code_diagrams/lambda/helm-installer/handler.handle_task.png``)
53# * ``validate_releases`` -> ``diagrams/code_diagrams/lambda/helm-installer/handler.validate_releases.html``
54# (PNG: ``diagrams/code_diagrams/lambda/helm-installer/handler.validate_releases.png``)
55# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
56# <pyflowchart-code-diagram> END
59logger = logging.getLogger()
60logger.setLevel(logging.INFO)
62SUCCESS = "SUCCESS"
63FAILED = "FAILED"
65# ---------------------------------------------------------------------------
66# Tunables
67# ---------------------------------------------------------------------------
69# Maximum number of install/upgrade attempts for each chart. Some charts
70# (e.g. cert-manager, NVIDIA operators) depend on admission webhooks or
71# CRDs that briefly flap during cluster bring-up, and a single retry often
72# clears those transient failures. Raise this if you see persistent retry
73# exhaustion in the logs; lower it for faster feedback in local testing.
74HELM_INSTALL_MAX_RETRIES = 3
76# Seconds to wait between failed chart attempts. Sized to give the EKS
77# control plane time to stabilise (webhook endpoints coming up, API
78# server throttling clearing) without dragging CloudFormation custom
79# resource completion beyond its 15-minute timeout.
80HELM_INSTALL_RETRY_DELAY_SECONDS = 30
82# Delete is a synchronous CloudFormation custom-resource operation with a
83# one-hour ceiling. Ordinary releases get a 60-second Helm deadline and a
84# 75-second process cap so each state-machine task fits its two-minute slot.
85# LBC receives a dedicated four-minute Helm deadline because it must remove
86# controller webhooks and finalizers after every Gateway-owned ALB is gone.
87HELM_UNINSTALL_TIMEOUT = "60s"
88HELM_UNINSTALL_COMMAND_TIMEOUT_SECONDS = 75
89LBC_CHART_NAME = "aws-load-balancer-controller"
90LBC_UNINSTALL_TIMEOUT = "4m"
91LBC_UNINSTALL_COMMAND_TIMEOUT_SECONDS = 270
93# KEDA's Helm release owns CRDs whose instances carry operator-managed
94# finalizers. Four bounded discovery calls plus two ordered deletion calls and
95# the final Helm uninstall fit inside the dedicated four-minute KEDA task.
96KEDA_API_GROUPS = ("keda.sh", "eventing.keda.sh")
97KUEUE_API_GROUPS = ("kueue.x-k8s.io",)
98# Argo CD Applications carry the resources finalizer when their author asked
99# for cascading deletes; the application controller must still be running to
100# honor it (and to release AppProjects), so they go before the chart.
101ARGOCD_API_GROUPS = ("argoproj.io",)
102# Crossplane: usages first (they block deletion of what they protect), then
103# operations, then composite definitions and compositions — deleting an XRD
104# makes Crossplane delete every composite resource of that kind and its
105# composed resources — and the packages (functions, providers) last.
106CROSSPLANE_API_GROUPS = (
107 "protection.crossplane.io",
108 "ops.crossplane.io",
109 "apiextensions.crossplane.io",
110 "pkg.crossplane.io",
111)
112#: Charts whose controllers attach finalizers to their custom resources.
113#: Their instances must be deleted BEFORE ``helm uninstall`` removes the
114#: controller — uninstalling first leaves finalizer-bearing objects that
115#: nothing can clear, wedging CRD deletion (kueue: caught live by release
116#: validation run sched241-350ffc7d, where the default gco-cluster-queue /
117#: gco-default-flavor objects deadlocked teardown).
118#: Slinky: the slurm chart is nothing but custom resources (Controller,
119#: NodeSet, RestApi, LoginSet, Accounting, Token). The operator drains a
120#: NodeSet's Slurm nodes and clears the finalizers it holds on them only while
121#: it runs, and the worker PDBs it creates would otherwise outlive the
122#: release; the operator chart itself (slinky-slurm-operator) installs later
123#: in teardown order and needs nothing.
124SLINKY_API_GROUPS = ("slinky.slurm.net",)
125CHART_CUSTOM_RESOURCE_API_GROUPS: dict[str, tuple[str, ...]] = {
126 "keda": KEDA_API_GROUPS,
127 "kueue": KUEUE_API_GROUPS,
128 "argocd": ARGOCD_API_GROUPS,
129 "crossplane": CROSSPLANE_API_GROUPS,
130 "slinky-slurm": SLINKY_API_GROUPS,
131}
132#: Resource types (``plural.group``) or whole API groups that a chart's
133#: cleanup deletes in a pass of their own, once every other type of the same
134#: scope is gone: one ``kubectl delete`` issues its deletes in order but
135#: waits for all of them together, so order alone is not enough. Argo CD's
136#: application controller looks an Application's AppProject up before it
137#: processes the resources finalizer and gives up while the project is
138#: missing; deleting the projects in the same call as the Applications stalls
139#: the cascade until the wait times out and the finalizers are stripped,
140#: which can leave the synced workloads behind (caught by
141#: integration:kind:platform-addons). Crossplane's packages (the functions
142#: and providers) go once the definitions and Compositions are gone.
143CHART_CUSTOM_RESOURCES_DELETED_LAST: dict[str, frozenset[str]] = {
144 "argocd": frozenset({"appprojects.argoproj.io"}),
145 "crossplane": frozenset({"pkg.crossplane.io"}),
146}
147KEDA_CUSTOM_RESOURCE_DELETE_TIMEOUT = "45s"
148KEDA_CUSTOM_RESOURCE_COMMAND_TIMEOUT_SECONDS = 55
149KEDA_CUSTOM_RESOURCE_DISCOVERY_TIMEOUT_SECONDS = 10
150CUSTOM_RESOURCE_FINALIZER_STRIP_TIMEOUT_SECONDS = 30
152# Validation intentionally has tighter command caps than chart installation:
153# these are read-only convergence checks and should never consume an entire
154# Lambda invocation when the API server or a Helm storage backend is wedged.
155HELM_VALIDATION_COMMAND_TIMEOUT_SECONDS = 120
156HELM_VALIDATION_TOTAL_TIMEOUT_SECONDS = 780
157KUBECTL_VALIDATION_COMMAND_TIMEOUT_SECONDS = 120
158# Service endpoint readiness is polled (it is a convergence condition); every
159# other validation dimension stays single-shot within the shared deadline.
160ENDPOINT_READINESS_POLL_SECONDS = 10.0
161KUBECTL_VALIDATION_REQUEST_TIMEOUT = "30s"
162MAX_VALIDATION_DIAGNOSTIC_CHARS = 2048
163_HELM_RELEASE_NOT_FOUND = "Error: release: not found"
166@dataclass(frozen=True)
167class _PinnedManifestBundle:
168 """One remotely hosted manifest whose bytes and inventory are immutable."""
170 name: str
171 url: str
172 size: int
173 sha256: str
174 object_count: int
175 crd_count: int
178PINNED_GATEWAY_CRD_BUNDLES = (
179 # Controller v3.6.0 is built against Gateway API v1.6.0 and requires the
180 # CRDs to be updated BEFORE the controller (release notes: upgrading the
181 # controller first silently disables gateway reconciliation until the
182 # CRDs catch up — exactly the stale-Accepted failure the 2026-08 live
183 # validation caught under the v1.5.0 bundle). Keep these two bundles and
184 # the aws-load-balancer-controller chart version in charts.yaml in
185 # lockstep when bumping either.
186 _PinnedManifestBundle(
187 name="gateway-api-standard-v1.6.0",
188 url=(
189 "https://github.com/kubernetes-sigs/gateway-api/releases/download/"
190 "v1.6.0/standard-install.yaml"
191 ),
192 size=1_170_953,
193 sha256="a557172e8348f758479e9ee4000bbbb4b4aa48302a6b73461823ea5349bad56d",
194 object_count=12,
195 crd_count=10,
196 ),
197 _PinnedManifestBundle(
198 name="aws-lbc-gateway-v3.6.0",
199 url=(
200 "https://raw.githubusercontent.com/kubernetes-sigs/"
201 "aws-load-balancer-controller/v3.6.0/config/crd/gateway/gateway-crds.yaml"
202 ),
203 size=129_552,
204 sha256="3a4264e2a7cf132944fd095b04ce86299385f26ea31b1e89d0ac21761e1865a0",
205 object_count=3,
206 crd_count=3,
207 ),
208)
209GATEWAY_CRD_HTTP_CONNECT_TIMEOUT_SECONDS = 5
210GATEWAY_CRD_HTTP_READ_TIMEOUT_SECONDS = 45
211GATEWAY_CRD_HTTP_MAX_REDIRECTS = 3
212GATEWAY_CRD_APPLY_COMMAND_TIMEOUT_SECONDS = 180
215class _ValidationTimeout(RuntimeError):
216 """A systemic command/budget timeout that should stop further release checks."""
219def _validation_command_timeout(deadline: float, cap: int) -> int:
220 """Cap one command to both its normal limit and the invocation-wide budget."""
221 remaining = deadline - time.monotonic()
222 if remaining < 1:
223 raise _ValidationTimeout("Helm validation exhausted its invocation-wide time budget")
224 return min(cap, max(1, int(remaining)))
227def _bounded_diagnostic(value: Any, limit: int = MAX_VALIDATION_DIAGNOSTIC_CHARS) -> str:
228 """Return useful subprocess/error text without emitting unbounded payloads."""
229 text = str(value).strip()
230 if not text:
231 return "<empty>"
232 if len(text) <= limit:
233 return text
234 return f"{text[:limit]}... [truncated {len(text) - limit} chars]"
237def _record_addon_status(chart_name: str, status: str, message: str) -> None:
238 """Record a single chart's install outcome to SSM (best-effort).
240 Writes ``/<project>/addons/<region>/<chart>`` as a small JSON blob so the
241 add-on layer's health is observable out-of-band — decoupled from the
242 CloudFormation rollback path. Read back via ``gco stacks addons-status``.
243 Failures here are swallowed: status reporting must never turn a successful
244 install into a failure (or vice versa).
245 """
246 project = os.environ.get("PROJECT_NAME")
247 region = os.environ.get("REGION")
248 if not project or not region:
249 return
250 import contextlib
251 import time as _time
253 with contextlib.suppress(Exception):
254 boto3.client("ssm").put_parameter(
255 Name=f"/{project}/addons/{region}/{chart_name}",
256 Value=json.dumps(
257 {
258 "chart": chart_name,
259 "status": status,
260 "message": message[:1024],
261 "updated_at": int(_time.time()),
262 }
263 ),
264 Type="String",
265 Overwrite=True,
266 )
269# Load default chart configurations
270CHARTS_CONFIG_PATH = Path(__file__).parent / "charts.yaml"
273def load_charts_config() -> dict[str, Any]:
274 """Load chart configurations from charts.yaml."""
275 if CHARTS_CONFIG_PATH.exists():
276 with open(CHARTS_CONFIG_PATH, encoding="utf-8") as f:
277 loaded = yaml.safe_load(f)
278 return loaded if isinstance(loaded, dict) else {"charts": {}}
279 return {"charts": {}}
282def send_response(
283 event: dict[str, Any],
284 context: Any,
285 status: str,
286 data: dict[str, Any],
287 physical_id: str,
288 reason: str | None = None,
289) -> None:
290 """Send response to CloudFormation."""
291 body = {
292 "Status": status,
293 "Reason": reason or f"See CloudWatch Log Stream: {context.log_stream_name}",
294 "PhysicalResourceId": physical_id,
295 "StackId": event["StackId"],
296 "RequestId": event["RequestId"],
297 "LogicalResourceId": event["LogicalResourceId"],
298 "Data": data,
299 }
301 logger.info(f"Sending response: {json.dumps(data)}")
303 # Timeout is for the CFN response callback (HTTP PUT to S3 presigned URL),
304 # not for Helm chart installation. Helm installs use subprocess with --timeout 10m.
305 http = urllib3.PoolManager()
306 try:
307 http.request(
308 "PUT",
309 event["ResponseURL"],
310 body=json.dumps(body).encode("utf-8"),
311 headers={"Content-Type": "application/json"},
312 timeout=10.0,
313 )
314 except Exception as e:
315 logger.error(f"Failed to send response: {e}")
318def get_eks_token(cluster_name: str, region: str) -> str:
319 """Generate EKS authentication token."""
320 from botocore.signers import RequestSigner
322 session = boto3.Session()
323 sts = session.client("sts", region_name=region)
324 service_id = sts.meta.service_model.service_id
326 signer = RequestSigner(
327 service_id, region, "sts", "v4", session.get_credentials(), session.events
328 )
330 params = {
331 "method": "GET",
332 "url": f"https://sts.{region}.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15",
333 "body": {},
334 "headers": {"x-k8s-aws-id": cluster_name},
335 "context": {},
336 }
338 url = signer.generate_presigned_url(
339 params, region_name=region, expires_in=60, operation_name=""
340 )
341 token = base64.urlsafe_b64encode(url.encode()).decode().rstrip("=")
342 return f"k8s-aws-v1.{token}"
345def configure_kubeconfig(cluster_name: str, region: str) -> str:
346 """Configure kubeconfig for EKS cluster and return path."""
347 eks = boto3.client("eks", region_name=region)
348 cluster = eks.describe_cluster(name=cluster_name)["cluster"]
350 # Create kubeconfig
351 kubeconfig = {
352 "apiVersion": "v1",
353 "kind": "Config",
354 "clusters": [
355 {
356 "name": cluster_name,
357 "cluster": {
358 "server": cluster["endpoint"],
359 "certificate-authority-data": cluster["certificateAuthority"]["data"],
360 },
361 }
362 ],
363 "contexts": [
364 {
365 "name": cluster_name,
366 "context": {
367 "cluster": cluster_name,
368 "user": cluster_name,
369 },
370 }
371 ],
372 "current-context": cluster_name,
373 "users": [
374 {
375 "name": cluster_name,
376 "user": {
377 "token": get_eks_token(cluster_name, region),
378 },
379 }
380 ],
381 }
383 # Write kubeconfig to temp file using secure method
384 fd, kubeconfig_path = tempfile.mkstemp(suffix=".yaml")
385 try:
386 with os.fdopen(fd, "w", encoding="utf-8") as f:
387 yaml.dump(kubeconfig, f)
388 except Exception:
389 # ``fdopen`` owns and normally closes the descriptor. Suppress a
390 # possible EBADF here, but always remove a partially-written credential
391 # file before propagating the original error.
392 with contextlib.suppress(OSError):
393 os.close(fd)
394 with contextlib.suppress(OSError):
395 os.remove(kubeconfig_path)
396 raise
398 return kubeconfig_path
401def run_helm(
402 args: list[str],
403 kubeconfig: str,
404 env: dict[str, str] | None = None,
405 command_timeout_seconds: int | None = None,
406 log_output: bool = True,
407) -> tuple[int, str, str]:
408 """Run helm command with kubeconfig.
410 Returns ``(returncode, stdout, stderr)``. A subprocess timeout is mapped
411 to ``(-1, "", "timeout: ...")`` so callers get a uniform failure contract
412 and can branch on the return code instead of wrapping every invocation in
413 ``try: ... except subprocess.TimeoutExpired``. This matters because
414 ``helm ... --wait`` can block on operator reconciliation; without this
415 mapping a single stuck release would crash the Lambda past the outer
416 ``except Exception`` and fail the whole retry loop.
418 ``command_timeout_seconds`` lets synchronous stack deletion use a tighter,
419 provable bound than create/update without changing the latter's 13-minute
420 allowance.
421 """
422 cmd = ["helm", *args]
424 helm_env = os.environ.copy()
425 helm_env["KUBECONFIG"] = kubeconfig
426 # Lambda has read-only filesystem except /tmp
427 helm_env["HELM_CACHE_HOME"] = "/tmp/.helm/cache" # nosec B108 # Lambda runtime requires /tmp for writable storage
428 helm_env["HELM_CONFIG_HOME"] = "/tmp/.helm/config" # nosec B108 # Lambda runtime requires /tmp for writable storage
429 helm_env["HELM_DATA_HOME"] = "/tmp/.helm/data" # nosec B108 # Lambda runtime requires /tmp for writable storage
430 if env:
431 helm_env.update(env)
433 logger.info(f"Running: {' '.join(cmd)}")
435 # Subprocess wall-clock cap. This MUST be >= helm's own ``--timeout`` (10m)
436 # below, otherwise a legitimately-slow install (e.g. a cold NVIDIA operator
437 # image pull) gets SIGKILLed by Python before helm's own deadline and a
438 # would-succeed install is reported as a failure. Each chart now runs in its
439 # own Step Functions task / Lambda invocation, so this can safely approach
440 # the per-invocation Lambda limit; retries are handled at the state-machine
441 # level. Override with HELM_CMD_TIMEOUT_SECONDS.
442 cmd_timeout = (
443 command_timeout_seconds
444 if command_timeout_seconds is not None
445 else int(os.environ.get("HELM_CMD_TIMEOUT_SECONDS", "780"))
446 )
448 try:
449 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - cmd is ["helm"] + static args list; helm_env is a controlled copy of os.environ, no shell=True
450 cmd,
451 capture_output=True,
452 text=True,
453 env=helm_env,
454 timeout=cmd_timeout,
455 check=False,
456 )
457 except subprocess.TimeoutExpired as exc:
458 logger.warning(f"helm subprocess timed out after {exc.timeout}s: {' '.join(cmd)}")
459 return -1, "", f"timeout: helm command exceeded {exc.timeout}s"
461 if log_output and result.stdout:
462 logger.info(f"stdout: {_bounded_diagnostic(result.stdout, 4096)}")
463 if log_output and result.stderr:
464 logger.warning(f"stderr: {_bounded_diagnostic(result.stderr, 4096)}")
466 return result.returncode, result.stdout, result.stderr
469def _autoscaled_scale_targets(values: dict[str, Any]) -> set[tuple[str, str]]:
470 """Return ``(kind, name)`` of every KEDA ScaledObject target in ``extraObjects``."""
471 targets: set[tuple[str, str]] = set()
472 extra_objects = values.get("extraObjects")
473 if not isinstance(extra_objects, list):
474 return targets
475 for obj in extra_objects:
476 if not isinstance(obj, dict) or obj.get("kind") != "ScaledObject":
477 continue
478 spec = obj.get("spec")
479 ref = spec.get("scaleTargetRef") if isinstance(spec, dict) else None
480 if isinstance(ref, dict) and ref.get("name"):
481 targets.add((str(ref.get("kind", "")), str(ref["name"])))
482 return targets
485def _carry_live_autoscaled_replicas(
486 config: dict[str, Any], values: dict[str, Any], namespace: str, kubeconfig: str
487) -> dict[str, Any]:
488 """Return the chart values with each autoscaled object's live ``spec.replicas``.
490 Helm applies server-side. A field another field manager has set to a
491 different value is a conflict, and one conflict fails the whole upgrade:
492 with the Slurm cluster scaled up by KEDA, a deploy died on ``conflict with
493 "keda" with subresource "scale" ... .spec.replicas`` (observed live,
494 2026-10-28). The chart renders ``replicas`` unconditionally and the CRD
495 defaults it to 1, so leaving the field out is not an option; applying the
496 live value is — it is what the autoscaler decided, nothing is scaled in or
497 out by the deploy, and the two managers share the field without conflict.
499 The chart entry declares what to carry under ``autoscaled_replicas``:
500 ``kind``/``resource`` of the scaled object, ``values_key`` (the values map
501 whose entries carry ``replicas``) and ``object_name`` (the live object's
502 name for a map key, ``{name}`` substituted). Only entries that are the
503 target of a KEDA ScaledObject among the values' ``extraObjects`` are
504 touched: a partition with autoscaling off keeps Helm as the owner of
505 ``replicas``, so a changed ``min_nodes`` still applies. Objects that do not
506 exist yet (first install, a new partition) and lookup failures keep the
507 rendered value; a failure is logged, and the upgrade then either succeeds
508 (nobody else owns the field) or fails loudly on the conflict.
510 ``values`` is not modified: ``deep_merge`` shares nested maps with its
511 inputs, so writing into them would reach back into the event's overrides.
512 """
513 rule = config.get("autoscaled_replicas")
514 if not rule:
515 return values
516 entries = values.get(rule["values_key"])
517 if not isinstance(entries, dict):
518 return values
519 values = copy.deepcopy(values)
520 entries = values[rule["values_key"]]
521 targets = _autoscaled_scale_targets(values)
522 for key, entry in entries.items():
523 if not isinstance(entry, dict):
524 continue
525 object_name = str(rule["object_name"]).format(name=key)
526 if (rule["kind"], object_name) not in targets:
527 continue
528 code, stdout, stderr = run_kubectl(
529 [
530 "get",
531 rule["resource"],
532 object_name,
533 "-n",
534 namespace,
535 "-o",
536 "jsonpath={.spec.replicas}",
537 "--ignore-not-found",
538 ],
539 kubeconfig,
540 log_output=False,
541 )
542 live = stdout.strip()
543 if code != 0 or not live.isdigit():
544 if code != 0 or live:
545 logger.warning(
546 f"Could not read live replicas of {rule['kind']} {namespace}/{object_name} "
547 f"(exit {code}, {_bounded_diagnostic(stderr or stdout, 512)!r}); "
548 f"applying the rendered value {entry.get('replicas')!r}"
549 )
550 continue
551 live_replicas = int(live)
552 if entry.get("replicas") != live_replicas:
553 logger.info(
554 f"{rule['kind']} {namespace}/{object_name} is autoscaled: carrying its live "
555 f"replicas={live_replicas} into values.{rule['values_key']}.{key}.replicas "
556 f"(rendered {entry.get('replicas')!r})"
557 )
558 entry["replicas"] = live_replicas
559 return values
562def _clear_stuck_release(chart_name: str, namespace: str, kubeconfig: str) -> bool:
563 """Delete release secrets for revisions stuck in ``pending-*`` state.
565 When a previous ``helm upgrade --wait`` is interrupted (timeout, Lambda
566 crash, network blip, operator reconciliation stall), Helm leaves the
567 revision's release secret in ``pending-upgrade``, ``pending-install``,
568 or ``pending-rollback`` status. That status acts as an exclusive lock:
569 every subsequent ``helm upgrade`` / ``helm rollback`` against the same
570 release fails with ``another operation (install/upgrade/rollback) is in
571 progress`` until the lock is cleared.
573 ``helm rollback --wait`` would normally clear it, but it can hang
574 indefinitely when the target chart's own operator (e.g. a CRD
575 controller) is stuck reconciling the half-applied state — which is
576 exactly the failure mode that got us here. Deleting the stuck secret
577 is the reliable recovery: Helm's view
578 of the release reverts to the previous ``deployed`` revision, and the
579 next upgrade proceeds normally.
581 Returns ``True`` if any stuck secrets were deleted.
582 """
583 status_code, status_out, _ = run_helm(
584 ["status", chart_name, "-n", namespace, "-o", "json"], kubeconfig
585 )
586 if status_code != 0:
587 # Release not installed yet (first install) — nothing to clear.
588 return False
590 try:
591 status = json.loads(status_out).get("info", {}).get("status", "")
592 except json.JSONDecodeError, AttributeError:
593 return False
595 if status not in ("pending-install", "pending-upgrade", "pending-rollback"):
596 return False
598 logger.warning(
599 f"Release {chart_name} in namespace {namespace} is stuck in {status!r}; "
600 f"clearing the stuck release secret so the next upgrade can proceed."
601 )
603 env = os.environ.copy()
604 env["KUBECONFIG"] = kubeconfig
606 # Only delete secrets matching the exact stuck status. ``deployed`` /
607 # ``superseded`` / ``failed`` history is preserved so ``helm history``
608 # still shows the prior revisions for debugging.
609 try:
610 list_result = (
611 subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv, no shell=True
612 [
613 "kubectl",
614 "get",
615 "secrets",
616 "-n",
617 namespace,
618 "-l",
619 f"owner=helm,name={chart_name},status={status}",
620 "-o",
621 "jsonpath={.items[*].metadata.name}",
622 ],
623 capture_output=True,
624 text=True,
625 env=env,
626 timeout=15,
627 check=False,
628 )
629 )
630 except subprocess.TimeoutExpired:
631 logger.warning(f"kubectl get secrets timed out while clearing {chart_name}")
632 return False
634 if list_result.returncode != 0 or not list_result.stdout.strip():
635 return False
637 cleared = False
638 for secret in list_result.stdout.split():
639 try:
640 del_result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv, no shell=True
641 ["kubectl", "delete", "secret", "-n", namespace, secret, "--ignore-not-found"],
642 capture_output=True,
643 text=True,
644 env=env,
645 timeout=15,
646 check=False,
647 )
648 except subprocess.TimeoutExpired:
649 logger.warning(f"kubectl delete timed out for {secret}")
650 continue
651 if del_result.returncode == 0:
652 cleared = True
653 logger.info(f"Deleted stuck release secret {secret}")
655 return cleared
658def add_helm_repo(repo_name: str, repo_url: str, kubeconfig: str) -> bool:
659 """Add Helm repository."""
660 code, _, _ = run_helm(["repo", "add", repo_name, repo_url, "--force-update"], kubeconfig)
661 if code != 0:
662 return False
664 code, _, _ = run_helm(["repo", "update", repo_name], kubeconfig)
665 return code == 0
668def install_chart(
669 chart_name: str,
670 config: dict[str, Any],
671 kubeconfig: str,
672 value_overrides: dict[str, Any] | None = None,
673) -> tuple[bool, str]:
674 """Install or upgrade a Helm chart."""
675 repo_name = config["repo_name"]
676 repo_url = config["repo_url"]
677 chart = config["chart"]
678 version = config.get("version")
679 namespace = config.get("namespace", "default")
680 create_ns = config.get("create_namespace", True)
681 values = config.get("values", {})
682 use_oci = config.get("use_oci", False)
684 # Per-chart readiness gate.
685 # ``wait`` (default True) -> ``helm --wait`` (block until the
686 # release's resources report Ready).
687 # ``wait_timeout`` (default 10m) -> ``helm --timeout``.
688 # A chart whose components converge asynchronously — e.g. one that pulls
689 # large images from a slow/rate-limited registry — can set ``wait: false``
690 # so the install returns as soon as manifests are applied instead of
691 # blocking the whole invocation on readiness. That keeps a single slow
692 # chart from burning the Lambda wall-clock guard (HELM_CMD_TIMEOUT_SECONDS)
693 # and lets the Step Functions state machine move on to the next chart; the
694 # release still converges in the background and its status is recorded to
695 # SSM either way. ``wait_timeout`` must stay below HELM_CMD_TIMEOUT_SECONDS
696 # (default 780s) or the subprocess guard SIGKILLs helm before its own
697 # deadline and a would-succeed install is reported as a failure.
698 wait = config.get("wait", True)
699 wait_timeout = config.get("wait_timeout", "10m")
701 # Merge value overrides
702 if value_overrides:
703 values = deep_merge(values, value_overrides)
705 # Fields an in-cluster autoscaler owns keep their live value (see
706 # _carry_live_autoscaled_replicas).
707 values = _carry_live_autoscaled_replicas(config, values, namespace, kubeconfig)
709 # For OCI registries, we don't need to add a repo
710 if not use_oci:
711 # Add repo
712 if not add_helm_repo(repo_name, repo_url, kubeconfig):
713 return False, f"Failed to add repo {repo_name}"
714 chart_ref = f"{repo_name}/{chart}"
715 else:
716 # For OCI, use the full OCI URL
717 chart_ref = f"{repo_url}/{chart}"
719 # Build helm upgrade --install command
720 args = [
721 "upgrade",
722 "--install",
723 chart_name,
724 chart_ref,
725 "--namespace",
726 namespace,
727 "--timeout",
728 wait_timeout,
729 ]
731 # ``--wait`` blocks until the release's resources are Ready. Opt-out per
732 # chart via ``wait: false`` for asynchronously-converging charts.
733 if wait:
734 args.append("--wait")
736 if version:
737 args.extend(["--version", version])
739 if create_ns:
740 args.append("--create-namespace")
742 # Write values to a mode-0600 temp file and remove it on every return or
743 # exception path. Chart values can contain credentials and role details.
744 values_file: str | None = None
745 try:
746 if values:
747 fd, values_file = tempfile.mkstemp(suffix=".yaml")
748 try:
749 with os.fdopen(fd, "w", encoding="utf-8") as f:
750 yaml.dump(values, f)
751 except Exception:
752 # ``fdopen`` normally owns and closes the descriptor, but also
753 # cover failures before ownership is transferred.
754 with contextlib.suppress(OSError):
755 os.close(fd)
756 raise
757 args.extend(["--values", values_file])
759 # Preflight: if a previous upgrade was interrupted, the release is
760 # wedged in ``pending-*`` and blocks all subsequent operations. Clear
761 # the stuck secret before attempting the upgrade so we don't have to
762 # rely on rollback-after-failure (which itself hangs when the chart's
763 # operator is stuck reconciling the half-applied state).
764 _clear_stuck_release(chart_name, namespace, kubeconfig)
766 code, _stdout, stderr = run_helm(args, kubeconfig)
768 if code == 0:
769 return True, f"Successfully installed {chart_name}"
771 # If we still hit "another operation in progress" despite the
772 # preflight (e.g. a concurrent operation started between the check
773 # and the upgrade), clear the stuck state and retry once. Unlike
774 # the previous ``rollback --wait`` approach, this never blocks on
775 # operator reconciliation.
776 if "another operation" in stderr.lower() and "in progress" in stderr.lower():
777 logger.warning(
778 f"Release {chart_name} reports 'another operation in progress' "
779 f"after preflight; clearing stuck state and retrying once."
780 )
781 _clear_stuck_release(chart_name, namespace, kubeconfig)
782 code2, _, stderr2 = run_helm(args, kubeconfig)
783 if code2 == 0:
784 return True, f"Successfully installed {chart_name} (after clearing stuck state)"
785 return False, f"Failed to install {chart_name}: {stderr2}"
786 return False, f"Failed to install {chart_name}: {stderr}"
787 finally:
788 if values_file:
789 with contextlib.suppress(FileNotFoundError):
790 os.unlink(values_file)
793def _strip_custom_resource_finalizers(
794 kubeconfig: str, resource_types: list[str], namespaced: bool
795) -> str | None:
796 """Remove finalizers from every remaining instance of the given types.
798 Recovery path for a delete that stalled because the finalizer-clearing
799 controller is already gone (e.g. a teardown retry after a partial
800 uninstall). Returns an error string, or ``None`` on success.
801 """
802 env = os.environ.copy()
803 env["KUBECONFIG"] = kubeconfig
804 common = ["kubectl", "--kubeconfig", kubeconfig, "--request-timeout=30s"]
805 for resource_type in resource_types:
806 if namespaced:
807 list_command = [
808 *common,
809 "get",
810 resource_type,
811 "--all-namespaces",
812 "-o",
813 'jsonpath={range .items[*]}{.metadata.namespace}{","}{.metadata.name}{"\\n"}{end}',
814 ]
815 else:
816 list_command = [*common, "get", resource_type, "-o", "name"]
817 try:
818 listing = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv, no shell=True
819 list_command,
820 capture_output=True,
821 text=True,
822 env=env,
823 timeout=CUSTOM_RESOURCE_FINALIZER_STRIP_TIMEOUT_SECONDS,
824 check=False,
825 )
826 except subprocess.TimeoutExpired:
827 return f"Timed out listing {resource_type} instances for finalizer removal"
828 if listing.returncode != 0:
829 # The whole resource type may already be gone with its CRD.
830 continue
831 for line in listing.stdout.split():
832 patch_command = [*common, "patch"]
833 if namespaced:
834 namespace, _, name = line.partition(",")
835 if not name:
836 continue
837 patch_command += [resource_type, name, "-n", namespace]
838 else:
839 patch_command += [line]
840 patch_command += ["--type", "merge", "-p", '{"metadata":{"finalizers":[]}}']
841 try:
842 patched = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - discovered resource names, no shell=True
843 patch_command,
844 capture_output=True,
845 text=True,
846 env=env,
847 timeout=CUSTOM_RESOURCE_FINALIZER_STRIP_TIMEOUT_SECONDS,
848 check=False,
849 )
850 except subprocess.TimeoutExpired:
851 return f"Timed out removing finalizers from {line}"
852 if patched.returncode != 0 and "not found" not in (patched.stderr or "").lower():
853 error = (patched.stderr or patched.stdout).strip()
854 return f"Failed to remove finalizers from {line}: {error}"
855 return None
858def _delete_chart_custom_resources(chart_name: str, kubeconfig: str) -> tuple[bool, str]:
859 """Delete a chart's custom resources before uninstalling its controller.
861 Applies to every chart in ``CHART_CUSTOM_RESOURCE_API_GROUPS``. Resource
862 discovery keeps this compatible with the exact chart version in use
863 instead of maintaining a second CRD list here. Namespaced resources are
864 deleted first across every namespace, then cluster-scoped ones; within a
865 scope, the chart's ``CHART_CUSTOM_RESOURCES_DELETED_LAST`` types get a
866 pass of their own after the rest. ``kubectl delete --wait`` does not
867 return until controller-owned finalizers are gone, so Helm can safely
868 remove the controller and CRDs afterwards (see
869 ``_delete_custom_resource_pass`` for the stalled-wait recovery).
870 """
871 env = os.environ.copy()
872 env["KUBECONFIG"] = kubeconfig
873 common = ["kubectl", "--kubeconfig", kubeconfig, "--request-timeout=30s"]
874 resources_by_scope: dict[bool, list[str]] = {True: [], False: []}
876 api_groups = CHART_CUSTOM_RESOURCE_API_GROUPS[chart_name]
877 for api_group in api_groups:
878 for namespaced in (True, False):
879 try:
880 discovery = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - fixed argv, no shell=True
881 [
882 *common,
883 "api-resources",
884 f"--api-group={api_group}",
885 "--verbs=list,delete",
886 f"--namespaced={'true' if namespaced else 'false'}",
887 "-o",
888 "name",
889 ],
890 capture_output=True,
891 text=True,
892 env=env,
893 timeout=KEDA_CUSTOM_RESOURCE_DISCOVERY_TIMEOUT_SECONDS,
894 check=False,
895 )
896 except subprocess.TimeoutExpired:
897 return False, f"Timed out discovering {api_group} custom resources"
899 if discovery.returncode != 0:
900 error = (discovery.stderr or discovery.stdout).strip()
901 return False, f"Failed to discover {api_group} custom resources: {error}"
902 resources_by_scope[namespaced].extend(discovery.stdout.split())
904 deleted_last = CHART_CUSTOM_RESOURCES_DELETED_LAST.get(chart_name, frozenset())
905 deleted_types = 0
906 for namespaced in (True, False):
907 resources = list(dict.fromkeys(resources_by_scope[namespaced]))
908 last = [
909 name
910 for name in resources
911 if name in deleted_last or name.partition(".")[2] in deleted_last
912 ]
913 passes = ([name for name in resources if name not in last], last)
914 for batch in passes:
915 if not batch:
916 continue
917 pass_error = _delete_custom_resource_pass(chart_name, kubeconfig, batch, namespaced)
918 if pass_error is not None:
919 return False, pass_error
920 deleted_types += len(batch)
921 return True, f"Deleted and waited for {deleted_types} {chart_name} custom resource type(s)"
924def _delete_custom_resource_pass(
925 chart_name: str, kubeconfig: str, resources: list[str], namespaced: bool
926) -> str | None:
927 """Delete every instance of ``resources`` and wait until they are gone.
929 If the wait stalls — the finalizer-clearing controller may already be
930 gone on a teardown retry — finalizers are stripped from the survivors and
931 the delete retried once, so teardown self-heals instead of wedging CRDs
932 in Terminating. Returns an error string, or ``None`` on success.
933 """
934 env = os.environ.copy()
935 env["KUBECONFIG"] = kubeconfig
936 command = [
937 "kubectl",
938 "--kubeconfig",
939 kubeconfig,
940 "--request-timeout=30s",
941 "delete",
942 ",".join(resources),
943 "--all",
944 "--ignore-not-found=true",
945 "--wait=true",
946 f"--timeout={KEDA_CUSTOM_RESOURCE_DELETE_TIMEOUT}",
947 ]
948 if namespaced:
949 command.append("--all-namespaces")
951 scope = "namespaced" if namespaced else "cluster-scoped"
952 failure: str | None = None
953 try:
954 deletion = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - discovered resource names, no shell=True
955 command,
956 capture_output=True,
957 text=True,
958 env=env,
959 timeout=KEDA_CUSTOM_RESOURCE_COMMAND_TIMEOUT_SECONDS,
960 check=False,
961 )
962 if deletion.returncode != 0:
963 failure = (deletion.stderr or deletion.stdout).strip()
964 except subprocess.TimeoutExpired:
965 failure = "delete --wait timed out"
966 if failure is None:
967 return None
968 logger.warning(
969 f"{scope} {chart_name} custom-resource delete stalled ({failure}); "
970 "stripping finalizers and retrying once"
971 )
972 strip_error = _strip_custom_resource_finalizers(kubeconfig, resources, namespaced)
973 if strip_error is not None:
974 return (
975 f"Failed to delete {scope} {chart_name} custom resources "
976 f"({failure}); finalizer removal also failed: {strip_error}"
977 )
978 try:
979 retry = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - discovered resource names, no shell=True
980 command,
981 capture_output=True,
982 text=True,
983 env=env,
984 timeout=KEDA_CUSTOM_RESOURCE_COMMAND_TIMEOUT_SECONDS,
985 check=False,
986 )
987 except subprocess.TimeoutExpired:
988 return (
989 f"Timed out deleting {scope} {chart_name} custom resources even after finalizer removal"
990 )
991 if retry.returncode != 0:
992 error = (retry.stderr or retry.stdout).strip()
993 return (
994 f"Failed to delete {scope} {chart_name} custom resources "
995 f"even after finalizer removal: {error}"
996 )
997 return None
1000def uninstall_chart(chart_name: str, namespace: str, kubeconfig: str) -> tuple[bool, str]:
1001 """Uninstall a Helm chart within the synchronous teardown budget."""
1002 if chart_name in CHART_CUSTOM_RESOURCE_API_GROUPS:
1003 cleaned, cleanup_message = _delete_chart_custom_resources(chart_name, kubeconfig)
1004 if not cleaned:
1005 return False, f"{chart_name} pre-uninstall cleanup failed: {cleanup_message}"
1006 logger.info(cleanup_message)
1008 helm_timeout = LBC_UNINSTALL_TIMEOUT if chart_name == LBC_CHART_NAME else HELM_UNINSTALL_TIMEOUT
1009 command_timeout = (
1010 LBC_UNINSTALL_COMMAND_TIMEOUT_SECONDS
1011 if chart_name == LBC_CHART_NAME
1012 else HELM_UNINSTALL_COMMAND_TIMEOUT_SECONDS
1013 )
1014 args = [
1015 "uninstall",
1016 chart_name,
1017 "--namespace",
1018 namespace,
1019 "--wait",
1020 "--timeout",
1021 helm_timeout,
1022 ]
1023 code, _, stderr = run_helm(
1024 args,
1025 kubeconfig,
1026 command_timeout_seconds=command_timeout,
1027 )
1029 if code == 0:
1030 return True, f"Successfully uninstalled {chart_name}"
1032 # Helm's explicit release-absence signature is idempotent success. Do not
1033 # accept a generic "not found": Kubernetes API/resource failures can carry
1034 # that text while the release is still live and must block teardown.
1035 if "release: not found" in stderr.lower():
1036 return True, f"Chart {chart_name} not found (already uninstalled)"
1037 return False, f"Failed to uninstall {chart_name}: {stderr}"
1040def quiesce_health_monitor(kubeconfig: str, namespace: str = "gco-system") -> tuple[bool, str]:
1041 """Scale health-monitor to zero and wait until every replica is gone.
1043 This is the first synchronous stack-delete task. It prevents the monitor
1044 from recreating the ALB-hostname SSM parameter after GA deregistration.
1045 Kubernetes' exact Deployment ``NotFound`` response is idempotent; every
1046 other scale/wait failure is surfaced to the teardown state machine.
1047 """
1048 env = os.environ.copy()
1049 env["KUBECONFIG"] = kubeconfig
1050 common = ["kubectl", "--kubeconfig", kubeconfig, "--request-timeout=30s"]
1052 try:
1053 scale = subprocess.run(
1054 [
1055 *common,
1056 "scale",
1057 "deployment/health-monitor",
1058 "--namespace",
1059 namespace,
1060 "--replicas=0",
1061 ],
1062 capture_output=True,
1063 text=True,
1064 env=env,
1065 timeout=30,
1066 check=False,
1067 )
1068 except subprocess.TimeoutExpired:
1069 return False, "Timed out scaling health-monitor deployment to zero"
1071 if scale.returncode != 0:
1072 scale_error = (scale.stderr or scale.stdout).strip()
1073 lowered = scale_error.lower()
1074 exact_absence = (
1075 'deployments.apps "health-monitor" not found',
1076 'deployment.apps "health-monitor" not found',
1077 'deployment "health-monitor" not found',
1078 # A deploy that failed before the base manifests applied never
1079 # created the namespace at all; there is nothing to quiesce, and
1080 # treating this as fatal wedged a stack DELETE_FAILED on the
1081 # HelmTeardown custom resource (2026-09 live validation, run
1082 # sched241-1ae7c0d3). Nothing-was-ever-there is as idempotent as
1083 # deployment-already-gone.
1084 f'namespaces "{namespace}" not found',
1085 )
1086 if not any(signature in lowered for signature in exact_absence):
1087 return False, f"Failed to scale health-monitor to zero: {scale_error}"
1089 try:
1090 wait = subprocess.run(
1091 [
1092 *common,
1093 "wait",
1094 "--for=delete",
1095 "pod",
1096 "--selector=app=health-monitor",
1097 "--namespace",
1098 namespace,
1099 "--timeout=120s",
1100 ],
1101 capture_output=True,
1102 text=True,
1103 env=env,
1104 timeout=135,
1105 check=False,
1106 )
1107 except subprocess.TimeoutExpired:
1108 return False, "Timed out waiting for health-monitor pods to terminate"
1110 if wait.returncode != 0:
1111 wait_error = (wait.stderr or wait.stdout).strip()
1112 lowered_wait = wait_error.lower()
1113 wait_absence = (
1114 "no matching resources found",
1115 # Same never-created-namespace case as the scale step above.
1116 f'namespaces "{namespace}" not found',
1117 )
1118 if not any(signature in lowered_wait for signature in wait_absence):
1119 return False, f"Failed waiting for health-monitor pods: {wait_error}"
1121 return True, "Health monitor quiesced"
1124def _cfn_chart_overrides(props: dict[str, Any]) -> dict[str, Any]:
1125 """Return a custom resource's ``Charts`` overrides with their JSON types.
1127 CloudFormation delivers custom-resource properties with every number and
1128 boolean as a string, so the regional stack sends ``Charts`` as one JSON
1129 string (decoded here); an object is taken as it is. Step Functions task
1130 events carry the orchestrator's already-decoded object and never pass
1131 through this.
1132 """
1133 charts = props.get("Charts", {})
1134 if isinstance(charts, str):
1135 try:
1136 charts = json.loads(charts)
1137 except json.JSONDecodeError as exc:
1138 raise ValueError(f"Charts is not valid JSON: {exc}") from exc
1139 if not isinstance(charts, dict):
1140 raise ValueError("Charts must be a JSON object of per-chart overrides")
1141 return charts
1144def deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]:
1145 """Deep merge two dictionaries."""
1146 result = base.copy()
1147 for key, value in override.items():
1148 if key in result and isinstance(result[key], dict) and isinstance(value, dict):
1149 result[key] = deep_merge(result[key], value)
1150 else:
1151 result[key] = value
1152 return result
1155def run_kubectl(
1156 args: list[str],
1157 kubeconfig: str,
1158 command_timeout_seconds: int = KUBECTL_VALIDATION_COMMAND_TIMEOUT_SECONDS,
1159 log_output: bool = True,
1160) -> tuple[int, str, str]:
1161 """Run a bounded, argument-vector-only kubectl command for validation."""
1162 cmd = [
1163 "kubectl",
1164 "--kubeconfig",
1165 kubeconfig,
1166 f"--request-timeout={KUBECTL_VALIDATION_REQUEST_TIMEOUT}",
1167 *args,
1168 ]
1169 env = os.environ.copy()
1170 env["KUBECONFIG"] = kubeconfig
1171 logger.info(f"Running: {' '.join(cmd)}")
1173 try:
1174 result = (
1175 subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - argv only, no shell=True
1176 cmd,
1177 capture_output=True,
1178 text=True,
1179 env=env,
1180 timeout=command_timeout_seconds,
1181 check=False,
1182 )
1183 )
1184 except subprocess.TimeoutExpired as exc:
1185 logger.warning(f"kubectl subprocess timed out after {exc.timeout}s")
1186 return -1, "", f"timeout: kubectl command exceeded {exc.timeout}s"
1188 if log_output and result.stdout:
1189 logger.info(f"stdout: {_bounded_diagnostic(result.stdout, 4096)}")
1190 if log_output and result.stderr:
1191 logger.warning(f"stderr: {_bounded_diagnostic(result.stderr, 4096)}")
1192 return result.returncode, result.stdout, result.stderr
1195def _release_configurations(
1196 event: dict[str, Any],
1197) -> tuple[list[tuple[str, dict[str, Any]]], set[str]]:
1198 """Return ordered, deeply-merged release configs and the enabled set."""
1199 defaults = load_charts_config().get("charts", {})
1200 overrides = event.get("Charts", {})
1201 enabled_charts = event.get("EnabledCharts", [])
1202 if overrides is None:
1203 overrides = {}
1204 if enabled_charts is None:
1205 enabled_charts = []
1207 if not isinstance(defaults, dict):
1208 raise RuntimeError("charts.yaml field 'charts' must be a mapping")
1209 if not isinstance(overrides, dict):
1210 raise RuntimeError("Charts must be a mapping")
1211 if not isinstance(enabled_charts, list) or not all(
1212 isinstance(name, str) and name for name in enabled_charts
1213 ):
1214 raise RuntimeError("EnabledCharts must be a list of non-empty release names")
1216 merged: dict[str, dict[str, Any]] = {}
1217 for release, config in defaults.items():
1218 if not isinstance(release, str) or not release or not isinstance(config, dict):
1219 raise RuntimeError("charts.yaml contains an invalid release configuration")
1220 merged[release] = deep_merge({}, config)
1222 # Existing releases retain charts.yaml order. Runtime-only releases append
1223 # in the JSON mapping's insertion order, while known releases receive the
1224 # exact same recursive override semantics used by installation.
1225 for release, override in overrides.items():
1226 if not isinstance(release, str) or not release or not isinstance(override, dict):
1227 raise RuntimeError("Charts contains an invalid release override")
1228 merged[release] = deep_merge(merged.get(release, {}), override)
1230 unknown_enabled = [name for name in enabled_charts if name not in merged]
1231 if unknown_enabled:
1232 names = ", ".join(unknown_enabled[:5])
1233 raise RuntimeError(f"EnabledCharts has no chart configuration for: {names}")
1235 return list(merged.items()), set(enabled_charts)
1238def _release_metadata(release: str, config: dict[str, Any]) -> tuple[str, str, str]:
1239 """Extract the chart, version, and namespace needed for exact validation."""
1240 chart = config.get("chart")
1241 version = config.get("version")
1242 namespace = config.get("namespace", "default")
1243 if not isinstance(chart, str) or not chart:
1244 raise RuntimeError(f"release {release!r} has no valid chart name")
1245 if version is None or not str(version):
1246 raise RuntimeError(f"release {release!r} has no configured chart version")
1247 if not isinstance(namespace, str) or not namespace:
1248 raise RuntimeError(f"release {release!r} has no valid namespace")
1249 return chart, str(version), namespace
1252def _parse_json_object(output: str, description: str) -> dict[str, Any]:
1253 """Parse command output as a JSON object with a bounded failure message."""
1254 try:
1255 parsed = json.loads(output)
1256 except json.JSONDecodeError as exc:
1257 raise RuntimeError(f"{description} returned invalid JSON: {exc.msg}") from exc
1258 except TypeError as exc:
1259 raise RuntimeError(f"{description} returned invalid JSON: {exc}") from exc
1260 if not isinstance(parsed, dict):
1261 raise RuntimeError(f"{description} returned {type(parsed).__name__}, expected object")
1262 return parsed
1265def _flatten_resources(value: Any, description: str) -> list[dict[str, Any]]:
1266 """Flatten Kubernetes ``kind: List`` documents into individual objects."""
1267 documents = value if isinstance(value, list) else [value]
1269 resources: list[dict[str, Any]] = []
1270 for document in documents:
1271 if document is None:
1272 continue
1273 if not isinstance(document, dict):
1274 raise RuntimeError(f"{description} contains a non-object document")
1275 if document.get("kind") == "List":
1276 items = document.get("items")
1277 if not isinstance(items, list):
1278 raise RuntimeError(f"{description} contains kind List without an items list")
1279 resources.extend(_flatten_resources(items, description))
1280 else:
1281 resources.append(document)
1282 return resources
1285def _resource_core_identity(resource: dict[str, Any], description: str) -> tuple[str, str, str]:
1286 """Return the immutable API-version/kind/name portion of an identity."""
1287 api_version = resource.get("apiVersion")
1288 kind = resource.get("kind")
1289 metadata = resource.get("metadata")
1290 name = metadata.get("name") if isinstance(metadata, dict) else None
1291 if not isinstance(api_version, str) or not api_version:
1292 raise RuntimeError(f"{description} contains an object without apiVersion/kind/name")
1293 if not isinstance(kind, str) or not kind:
1294 raise RuntimeError(f"{description} contains an object without apiVersion/kind/name")
1295 if not isinstance(name, str) or not name:
1296 raise RuntimeError(f"{description} contains an object without apiVersion/kind/name")
1297 return api_version, kind, name
1300def _resource_namespace(resource: dict[str, Any]) -> str | None:
1301 metadata = resource.get("metadata")
1302 namespace = metadata.get("namespace") if isinstance(metadata, dict) else None
1303 return namespace if isinstance(namespace, str) and namespace else None
1306def _display_identity(identity: tuple[str, str, str], namespace: str | None = None) -> str:
1307 api_version, kind, name = identity
1308 object_name = f"{namespace}/{name}" if namespace else name
1309 return f"{api_version}/{kind} {object_name}"
1312def _compare_resource_identities(
1313 expected: list[dict[str, Any]],
1314 actual: list[dict[str, Any]],
1315 release: str,
1316 release_namespace: str,
1317) -> None:
1318 """Require exact counts and API/kind/name/namespace identities."""
1319 expected_core = Counter(
1320 _resource_core_identity(resource, f"manifest for {release}") for resource in expected
1321 )
1322 actual_core = Counter(
1323 _resource_core_identity(resource, f"kubectl output for {release}") for resource in actual
1324 )
1325 if len(expected) != len(actual) or expected_core != actual_core:
1326 missing = list((expected_core - actual_core).elements())[:5]
1327 unexpected = list((actual_core - expected_core).elements())[:5]
1328 details = []
1329 if missing:
1330 details.append("missing=" + ", ".join(_display_identity(item) for item in missing))
1331 if unexpected:
1332 details.append(
1333 "unexpected=" + ", ".join(_display_identity(item) for item in unexpected)
1334 )
1335 detail = "; ".join(details) or "duplicate resource identities differ"
1336 raise RuntimeError(
1337 f"release {release!r} rendered {len(expected)} resources but kubectl returned "
1338 f"{len(actual)} ({detail})"
1339 )
1341 # Explicit manifest namespaces must match exactly for namespaced kinds. A
1342 # namespace omitted by a namespaced Helm object is defaulted by
1343 # ``kubectl -n`` and must return from the release namespace; a
1344 # cluster-scoped object legitimately returns no namespace even when the
1345 # chart templates ``metadata.namespace`` onto it (for example kueue's
1346 # MutatingWebhookConfiguration) because the API server discards the field
1347 # on cluster-scoped kinds. Namespaced kinds always return with their
1348 # namespace, so accepting a cluster-scoped return never weakens the check
1349 # for them. Consume counters so duplicate identities are also exact.
1350 actual_namespaced = Counter(
1351 (_resource_core_identity(resource, "kubectl output"), _resource_namespace(resource))
1352 for resource in actual
1353 )
1354 for resource in expected:
1355 namespace = _resource_namespace(resource)
1356 if namespace is None:
1357 continue
1358 identity = _resource_core_identity(resource, "manifest")
1359 key = (identity, namespace)
1360 cluster_scoped_key = (identity, None)
1361 if actual_namespaced[key] > 0:
1362 actual_namespaced[key] -= 1
1363 elif actual_namespaced[cluster_scoped_key] > 0:
1364 actual_namespaced[cluster_scoped_key] -= 1
1365 else:
1366 wrong_namespaces = sorted(
1367 actual_namespace or "<cluster-scoped>"
1368 for (actual_identity, actual_namespace), count in actual_namespaced.items()
1369 if actual_identity == identity and count > 0
1370 )
1371 raise RuntimeError(
1372 f"release {release!r} returned the wrong namespace for "
1373 f"{_display_identity(identity, namespace)}: expected {namespace!r} or "
1374 f"cluster scope, got {wrong_namespaces}"
1375 )
1377 for resource in expected:
1378 if _resource_namespace(resource) is not None:
1379 continue
1380 identity = _resource_core_identity(resource, "manifest")
1381 namespaced_key = (identity, release_namespace)
1382 cluster_scoped_key = (identity, None)
1383 if actual_namespaced[namespaced_key] > 0:
1384 actual_namespaced[namespaced_key] -= 1
1385 elif actual_namespaced[cluster_scoped_key] > 0:
1386 actual_namespaced[cluster_scoped_key] -= 1
1387 else:
1388 wrong_namespaces = sorted(
1389 namespace or "<cluster-scoped>"
1390 for (actual_identity, namespace), count in actual_namespaced.items()
1391 if actual_identity == identity and count > 0
1392 )
1393 raise RuntimeError(
1394 f"release {release!r} returned the wrong namespace for "
1395 f"{_display_identity(identity)}: expected {release_namespace!r} or "
1396 f"cluster scope, got {wrong_namespaces}"
1397 )
1400def _condition_status(resource: dict[str, Any], condition_type: str) -> Any:
1401 status = resource.get("status")
1402 conditions = status.get("conditions", []) if isinstance(status, dict) else []
1403 if not isinstance(conditions, list):
1404 return None
1405 for condition in conditions:
1406 if isinstance(condition, dict) and condition.get("type") == condition_type:
1407 return condition.get("status")
1408 return None
1411def _is_true(value: Any) -> bool:
1412 return value is True or value == "True"
1415def _is_false(value: Any) -> bool:
1416 return value is False or value == "False" or value == "false"
1419def _replica_value(status: dict[str, Any], field: str) -> int | None:
1420 # Kubernetes omits optional integer counters when their value is zero.
1421 value = status.get(field, 0)
1422 return value if isinstance(value, int) and not isinstance(value, bool) else None
1425def _require_observed_generation(resource: dict[str, Any], identity: str) -> None:
1426 metadata = resource.get("metadata", {})
1427 status = resource.get("status", {})
1428 generation = metadata.get("generation") if isinstance(metadata, dict) else None
1429 observed = status.get("observedGeneration") if isinstance(status, dict) else None
1430 if not isinstance(generation, int) or observed != generation:
1431 raise RuntimeError(
1432 f"{identity} has stale generation: observed={observed!r}, expected={generation!r}"
1433 )
1436def _require_replica_convergence(
1437 resource: dict[str, Any], identity: str, fields: tuple[str, ...]
1438) -> None:
1439 spec = resource.get("spec", {})
1440 status = resource.get("status", {})
1441 desired = spec.get("replicas", 1) if isinstance(spec, dict) else None
1442 if not isinstance(desired, int) or isinstance(desired, bool) or not isinstance(status, dict):
1443 raise RuntimeError(f"{identity} has invalid desired/status replica data")
1444 mismatches = {
1445 field: _replica_value(status, field)
1446 for field in fields
1447 if _replica_value(status, field) != desired
1448 }
1449 if mismatches:
1450 raise RuntimeError(f"{identity} is not converged: desired={desired}, replicas={mismatches}")
1453def _validate_resource_readiness(resource: dict[str, Any]) -> None:
1454 """Apply kind-specific readiness gates plus generic custom conditions."""
1455 core = _resource_core_identity(resource, "kubectl output")
1456 namespace = _resource_namespace(resource)
1457 identity = _display_identity(core, namespace)
1458 _, kind, _ = core
1459 metadata = resource.get("metadata", {})
1460 if isinstance(metadata, dict) and metadata.get("deletionTimestamp") is not None:
1461 raise RuntimeError(f"{identity} is terminating")
1462 status = resource.get("status", {})
1463 if not isinstance(status, dict):
1464 status = {}
1466 if kind == "Deployment":
1467 _require_observed_generation(resource, identity)
1468 _require_replica_convergence(
1469 resource,
1470 identity,
1471 ("replicas", "updatedReplicas", "readyReplicas", "availableReplicas"),
1472 )
1473 if not _is_true(_condition_status(resource, "Available")):
1474 raise RuntimeError(f"{identity} does not report Available=True")
1475 elif kind == "StatefulSet":
1476 _require_observed_generation(resource, identity)
1477 _require_replica_convergence(
1478 resource, identity, ("currentReplicas", "updatedReplicas", "readyReplicas")
1479 )
1480 elif kind == "DaemonSet":
1481 _require_observed_generation(resource, identity)
1482 if "desiredNumberScheduled" not in status:
1483 raise RuntimeError(f"{identity} has no desiredNumberScheduled")
1484 desired = _replica_value(status, "desiredNumberScheduled")
1485 if desired is None:
1486 raise RuntimeError(f"{identity} has invalid desiredNumberScheduled")
1487 mismatches = {
1488 field: _replica_value(status, field)
1489 for field in (
1490 "currentNumberScheduled",
1491 "updatedNumberScheduled",
1492 "numberReady",
1493 "numberAvailable",
1494 )
1495 if _replica_value(status, field) != desired
1496 }
1497 misscheduled = _replica_value(status, "numberMisscheduled")
1498 if misscheduled != 0:
1499 mismatches["numberMisscheduled"] = misscheduled
1500 if mismatches:
1501 raise RuntimeError(
1502 f"{identity} is not converged: desired={desired}, replicas={mismatches}"
1503 )
1504 elif kind == "Job":
1505 if not _is_true(_condition_status(resource, "Complete")):
1506 raise RuntimeError(f"{identity} does not report Complete=True")
1507 elif kind == "Pod":
1508 if not _is_true(_condition_status(resource, "Ready")):
1509 raise RuntimeError(f"{identity} does not report Ready=True")
1510 elif kind == "PersistentVolumeClaim":
1511 if status.get("phase") != "Bound":
1512 raise RuntimeError(f"{identity} is not Bound (phase={status.get('phase')!r})")
1513 elif kind == "PersistentVolume":
1514 if status.get("phase") not in ("Bound", "Available"):
1515 raise RuntimeError(
1516 f"{identity} is neither Bound nor Available (phase={status.get('phase')!r})"
1517 )
1518 elif kind == "Ingress":
1519 load_balancer = status.get("loadBalancer", {})
1520 ingress = load_balancer.get("ingress", []) if isinstance(load_balancer, dict) else []
1521 has_address = isinstance(ingress, list) and any(
1522 isinstance(item, dict) and (item.get("ip") or item.get("hostname")) for item in ingress
1523 )
1524 if not has_address:
1525 raise RuntimeError(f"{identity} has no load-balancer address")
1526 elif kind == "CustomResourceDefinition":
1527 if not _is_true(_condition_status(resource, "Established")):
1528 raise RuntimeError(f"{identity} does not report Established=True")
1529 elif kind == "APIService":
1530 if not _is_true(_condition_status(resource, "Available")):
1531 raise RuntimeError(f"{identity} does not report Available=True")
1532 elif kind == "HorizontalPodAutoscaler":
1533 _require_observed_generation(resource, identity)
1534 if not _is_true(_condition_status(resource, "AbleToScale")):
1535 raise RuntimeError(f"{identity} does not report AbleToScale=True")
1536 if not _is_true(_condition_status(resource, "ScalingActive")):
1537 raise RuntimeError(f"{identity} does not report ScalingActive=True")
1538 elif kind == "PodDisruptionBudget":
1539 _require_observed_generation(resource, identity)
1540 current = status.get("currentHealthy")
1541 desired = status.get("desiredHealthy")
1542 if (
1543 not isinstance(current, int)
1544 or isinstance(current, bool)
1545 or not isinstance(desired, int)
1546 or isinstance(desired, bool)
1547 or current < desired
1548 ):
1549 raise RuntimeError(
1550 f"{identity} is unhealthy: currentHealthy={current!r}, desiredHealthy={desired!r}"
1551 )
1553 conditions = status.get("conditions", [])
1554 if isinstance(conditions, list):
1555 for condition in conditions:
1556 if (
1557 isinstance(condition, dict)
1558 and condition.get("type") in ("Ready", "Available")
1559 and _is_false(condition.get("status"))
1560 ):
1561 raise RuntimeError(
1562 f"{identity} reports {condition.get('type')}=False: "
1563 f"{_bounded_diagnostic(condition.get('message', 'no message'), 300)}"
1564 )
1567def _service_has_ready_endpoint(
1568 name: str,
1569 namespace: str,
1570 display: str,
1571 kubeconfig: str,
1572 deadline: float,
1573) -> bool:
1574 """Return whether one ready, non-terminating endpoint backs the Service."""
1575 code, stdout, stderr = run_kubectl(
1576 [
1577 "get",
1578 "endpointslices.discovery.k8s.io",
1579 "-n",
1580 namespace,
1581 "-l",
1582 f"kubernetes.io/service-name={name}",
1583 "-o",
1584 "json",
1585 ],
1586 kubeconfig,
1587 command_timeout_seconds=_validation_command_timeout(
1588 deadline, KUBECTL_VALIDATION_COMMAND_TIMEOUT_SECONDS
1589 ),
1590 log_output=False,
1591 )
1592 if code == -1:
1593 raise _ValidationTimeout(f"{display} EndpointSlice query timed out")
1594 if code != 0:
1595 raise RuntimeError(
1596 f"{display} EndpointSlice query failed: {_bounded_diagnostic(stderr or stdout)}"
1597 )
1599 payload = _parse_json_object(stdout, f"EndpointSlice query for {display}")
1600 items = payload.get("items")
1601 # A list response is an EndpointSliceList; anything else is a single slice.
1602 slices = items if isinstance(items, list) else [payload]
1603 for endpoint_slice in slices:
1604 if not isinstance(endpoint_slice, dict):
1605 continue
1606 metadata = endpoint_slice.get("metadata", {})
1607 if isinstance(metadata, dict) and metadata.get("deletionTimestamp") is not None:
1608 continue
1609 endpoints = endpoint_slice.get("endpoints", [])
1610 if not isinstance(endpoints, list):
1611 continue
1612 for endpoint in endpoints:
1613 if not isinstance(endpoint, dict):
1614 continue
1615 conditions = endpoint.get("conditions", {})
1616 if not isinstance(conditions, dict):
1617 continue
1618 if _is_true(conditions.get("ready")) and not _is_true(conditions.get("terminating")):
1619 return True
1620 return False
1623def _validate_service_endpoints(
1624 resource: dict[str, Any],
1625 kubeconfig: str,
1626 release_namespace: str,
1627 deadline: float,
1628) -> None:
1629 """Wait, within the validation deadline, for one ready Service endpoint.
1631 Endpoint readiness is a convergence condition, not an instant contract:
1632 slow-starting workloads (for example Grafana running its first-boot
1633 database migrations on a fresh PersistentVolume) legitimately publish
1634 their ready endpoint minutes after installation. A single-shot check
1635 failed a healthy live deployment for exactly that reason, so poll until
1636 the shared validation deadline; a Service that never converges still
1637 fails with the exact object named. Query and parse failures are not
1638 convergence conditions and surface immediately.
1639 """
1640 spec = resource.get("spec", {})
1641 selector = spec.get("selector") if isinstance(spec, dict) else None
1642 if resource.get("kind") != "Service" or not isinstance(selector, dict) or not selector:
1643 return
1645 identity = _resource_core_identity(resource, "kubectl output")
1646 name = identity[2]
1647 namespace = _resource_namespace(resource) or release_namespace
1648 display = _display_identity(identity, namespace)
1650 while not _service_has_ready_endpoint(name, namespace, display, kubeconfig, deadline):
1651 if deadline - time.monotonic() <= ENDPOINT_READINESS_POLL_SECONDS:
1652 raise RuntimeError(f"{display} has no ready, non-terminating EndpointSlice endpoint")
1653 # nosemgrep: arbitrary-sleep - bounded convergence polling within the validation deadline
1654 time.sleep(ENDPOINT_READINESS_POLL_SECONDS)
1657def _remove_validation_file(path: str) -> None:
1658 """Remove validation material, ignoring only an already-absent path."""
1659 try:
1660 os.remove(path)
1661 except FileNotFoundError:
1662 return
1665@contextlib.contextmanager
1666def _secure_manifest_file(manifest: str) -> Iterator[str]:
1667 """Write a mode-0600 manifest in the system temporary directory and always remove it."""
1668 fd, path = tempfile.mkstemp(
1669 prefix="helm-validation-",
1670 suffix=".yaml",
1671 dir=tempfile.gettempdir(),
1672 )
1673 try:
1674 os.fchmod(fd, 0o600)
1675 with os.fdopen(fd, "w", encoding="utf-8") as manifest_file:
1676 manifest_file.write(manifest)
1677 yield path
1678 finally:
1679 with contextlib.suppress(OSError):
1680 os.close(fd)
1681 _remove_validation_file(path)
1684@contextlib.contextmanager
1685def _verified_gateway_crd_bundle(
1686 bundle: _PinnedManifestBundle,
1687) -> Iterator[tuple[str, list[dict[str, Any]]]]:
1688 """Download one pinned bundle, verify exact bytes, and expose a mode-0600 file."""
1689 # A failed request leaves nothing to release, so the connection guard
1690 # only needs to cover the checks that run once a response exists.
1691 response = urllib3.PoolManager().request(
1692 "GET",
1693 bundle.url,
1694 headers={"User-Agent": "gco-helm-installer/1"},
1695 timeout=urllib3.Timeout(
1696 connect=GATEWAY_CRD_HTTP_CONNECT_TIMEOUT_SECONDS,
1697 read=GATEWAY_CRD_HTTP_READ_TIMEOUT_SECONDS,
1698 ),
1699 retries=urllib3.Retry(
1700 total=GATEWAY_CRD_HTTP_MAX_REDIRECTS,
1701 connect=0,
1702 read=0,
1703 redirect=GATEWAY_CRD_HTTP_MAX_REDIRECTS,
1704 status=0,
1705 other=0,
1706 raise_on_redirect=True,
1707 raise_on_status=True,
1708 ),
1709 redirect=True,
1710 )
1711 try:
1712 if response.status != 200:
1713 raise RuntimeError(
1714 f"{bundle.name} download returned HTTP {response.status}, expected 200"
1715 )
1716 body = response.data
1717 if not isinstance(body, bytes):
1718 raise RuntimeError(f"{bundle.name} download returned a non-byte body")
1719 finally:
1720 response.release_conn()
1722 if len(body) != bundle.size:
1723 raise RuntimeError(f"{bundle.name} size mismatch: got {len(body)}, expected {bundle.size}")
1724 actual_sha256 = hashlib.sha256(body).hexdigest()
1725 if actual_sha256 != bundle.sha256:
1726 raise RuntimeError(
1727 f"{bundle.name} SHA-256 mismatch: got {actual_sha256}, expected {bundle.sha256}"
1728 )
1730 try:
1731 documents = list(yaml.safe_load_all(body.decode("utf-8")))
1732 resources = _flatten_resources(documents, bundle.name)
1733 except (UnicodeDecodeError, yaml.YAMLError) as exc:
1734 raise RuntimeError(f"{bundle.name} is not valid UTF-8 YAML: {exc}") from exc
1736 identities = [_resource_core_identity(resource, bundle.name) for resource in resources]
1737 crd_count = sum(kind == "CustomResourceDefinition" for _, kind, _ in identities)
1738 if len(resources) != bundle.object_count or crd_count != bundle.crd_count:
1739 raise RuntimeError(
1740 f"{bundle.name} inventory mismatch: objects={len(resources)}/"
1741 f"{bundle.object_count}, CRDs={crd_count}/{bundle.crd_count}"
1742 )
1743 if len(identities) != len(set(identities)):
1744 raise RuntimeError(f"{bundle.name} contains duplicate object identities")
1746 fd, path = tempfile.mkstemp(
1747 prefix=f"{bundle.name}-",
1748 suffix=".yaml",
1749 dir=tempfile.gettempdir(),
1750 )
1751 try:
1752 os.fchmod(fd, 0o600)
1753 with os.fdopen(fd, "wb") as manifest_file:
1754 manifest_file.write(body)
1755 yield path, resources
1756 finally:
1757 with contextlib.suppress(OSError):
1758 os.close(fd)
1759 _remove_validation_file(path)
1762def _apply_gateway_crds(kubeconfig: str) -> list[dict[str, Any]]:
1763 """Server-side apply both verified Gateway API bundles before LBC install."""
1764 evidence: list[dict[str, Any]] = []
1765 for bundle in PINNED_GATEWAY_CRD_BUNDLES:
1766 with _verified_gateway_crd_bundle(bundle) as (manifest_path, resources):
1767 code, stdout, stderr = run_kubectl(
1768 [
1769 "apply",
1770 "--server-side=true",
1771 "--force-conflicts",
1772 "--field-manager=gco-helm-installer",
1773 "-f",
1774 manifest_path,
1775 ],
1776 kubeconfig,
1777 command_timeout_seconds=GATEWAY_CRD_APPLY_COMMAND_TIMEOUT_SECONDS,
1778 )
1779 if code != 0:
1780 raise RuntimeError(
1781 f"failed to apply {bundle.name}: {_bounded_diagnostic(stderr or stdout)}"
1782 )
1783 evidence.append(
1784 {
1785 "bundle": bundle.name,
1786 "object_count": len(resources),
1787 "crd_count": bundle.crd_count,
1788 "sha256": bundle.sha256,
1789 }
1790 )
1791 return evidence
1794def _validate_gateway_crds(kubeconfig: str, deadline: float) -> list[dict[str, Any]]:
1795 """Redownload and prove exact live identities plus Established=True CRDs."""
1796 evidence: list[dict[str, Any]] = []
1797 for bundle in PINNED_GATEWAY_CRD_BUNDLES:
1798 with _verified_gateway_crd_bundle(bundle) as (manifest_path, expected):
1799 code, stdout, stderr = run_kubectl(
1800 ["get", "-f", manifest_path, "-o", "json"],
1801 kubeconfig,
1802 command_timeout_seconds=_validation_command_timeout(
1803 deadline, KUBECTL_VALIDATION_COMMAND_TIMEOUT_SECONDS
1804 ),
1805 log_output=False,
1806 )
1807 if code == -1:
1808 raise _ValidationTimeout(f"kubectl get timed out for {bundle.name}")
1809 if code != 0:
1810 raise RuntimeError(
1811 f"kubectl could not retrieve {bundle.name}: "
1812 f"{_bounded_diagnostic(stderr or stdout)}"
1813 )
1814 live_payload = _parse_json_object(stdout, f"kubectl get for {bundle.name}")
1815 live_resources = _flatten_resources(live_payload, f"kubectl output for {bundle.name}")
1816 _compare_resource_identities(
1817 expected,
1818 live_resources,
1819 bundle.name,
1820 "default",
1821 )
1822 for resource in live_resources:
1823 _validate_resource_readiness(resource)
1824 evidence.append(
1825 {
1826 "bundle": bundle.name,
1827 "object_count": len(live_resources),
1828 "crd_count": bundle.crd_count,
1829 "sha256": bundle.sha256,
1830 }
1831 )
1832 return evidence
1835def _validate_enabled_release(
1836 release: str,
1837 chart: str,
1838 version: str,
1839 namespace: str,
1840 kubeconfig: str,
1841 deadline: float,
1842) -> int:
1843 code, stdout, stderr = run_helm(
1844 ["status", release, "-n", namespace, "-o", "json"],
1845 kubeconfig,
1846 command_timeout_seconds=_validation_command_timeout(
1847 deadline, HELM_VALIDATION_COMMAND_TIMEOUT_SECONDS
1848 ),
1849 log_output=False,
1850 )
1851 if code == -1:
1852 raise _ValidationTimeout(f"helm status timed out for release {release!r}")
1853 if code != 0:
1854 raise RuntimeError(f"helm status failed: {_bounded_diagnostic(stderr or stdout)}")
1855 status_payload = _parse_json_object(stdout, f"helm status for {release}")
1856 release_status = status_payload.get("info", {}).get("status")
1857 if release_status != "deployed":
1858 raise RuntimeError(f"helm status is {release_status!r}, expected exactly 'deployed'")
1860 # Release names are DNS labels, so '-' is literal outside a character
1861 # class. ``re.escape`` handles all regex metacharacters; undoing its
1862 # unnecessary hyphen escape keeps the expression valid for Helm's Go regex.
1863 escaped_release = re.escape(release).replace(r"\-", "-")
1864 code, stdout, stderr = run_helm(
1865 ["list", "-n", namespace, "--filter", f"^{escaped_release}$", "-o", "json"],
1866 kubeconfig,
1867 command_timeout_seconds=_validation_command_timeout(
1868 deadline, HELM_VALIDATION_COMMAND_TIMEOUT_SECONDS
1869 ),
1870 log_output=False,
1871 )
1872 if code == -1:
1873 raise _ValidationTimeout(f"helm list timed out for release {release!r}")
1874 if code != 0:
1875 raise RuntimeError(f"helm list failed: {_bounded_diagnostic(stderr or stdout)}")
1876 try:
1877 listed = json.loads(stdout)
1878 except json.JSONDecodeError as exc:
1879 raise RuntimeError(f"helm list for {release} returned invalid JSON: {exc.msg}") from exc
1880 except TypeError as exc:
1881 raise RuntimeError(f"helm list for {release} returned invalid JSON: {exc}") from exc
1882 if not isinstance(listed, list) or len(listed) != 1 or not isinstance(listed[0], dict):
1883 count = len(listed) if isinstance(listed, list) else "non-list"
1884 raise RuntimeError(f"helm list returned {count} entries, expected exactly one")
1885 entry = listed[0]
1886 expected_chart = f"{chart}-{version}"
1887 mismatches = {
1888 field: (entry.get(field), expected)
1889 for field, expected in (
1890 ("name", release),
1891 ("namespace", namespace),
1892 ("status", "deployed"),
1893 ("chart", expected_chart),
1894 )
1895 if entry.get(field) != expected
1896 }
1897 if mismatches:
1898 raise RuntimeError(f"helm list metadata mismatch: {mismatches}")
1900 code, manifest, stderr = run_helm(
1901 ["get", "manifest", release, "-n", namespace],
1902 kubeconfig,
1903 command_timeout_seconds=_validation_command_timeout(
1904 deadline, HELM_VALIDATION_COMMAND_TIMEOUT_SECONDS
1905 ),
1906 log_output=False,
1907 )
1908 if code == -1:
1909 raise _ValidationTimeout(f"helm get manifest timed out for release {release!r}")
1910 if code != 0:
1911 raise RuntimeError(f"helm get manifest failed: {_bounded_diagnostic(stderr or manifest)}")
1912 if not manifest.strip():
1913 raise RuntimeError("helm get manifest returned empty output")
1914 try:
1915 rendered = _flatten_resources(list(yaml.safe_load_all(manifest)), f"manifest for {release}")
1916 except yaml.YAMLError as exc:
1917 raise RuntimeError(
1918 f"helm manifest is invalid YAML: {_bounded_diagnostic(exc, 400)}"
1919 ) from exc
1920 if not rendered:
1921 raise RuntimeError("helm get manifest yielded no Kubernetes objects")
1923 # Charts legitimately render objects into other namespaces (kube-system
1924 # auth-reader RoleBindings from KEDA/cert-manager/kueue, control-plane
1925 # metric Services from kube-prometheus-stack). kubectl refuses a single
1926 # ``-n`` covering mixed namespaces, so retrieval is grouped by each
1927 # object's effective namespace; objects without an explicit namespace
1928 # resolve to the release namespace, and kubectl ignores ``-n`` for
1929 # cluster-scoped kinds.
1930 grouped: dict[str, list[dict[str, Any]]] = {}
1931 for resource in rendered:
1932 grouped.setdefault(_resource_namespace(resource) or namespace, []).append(resource)
1933 live_resources: list[dict[str, Any]] = []
1934 for group_namespace in sorted(grouped):
1935 group_manifest = yaml.safe_dump_all(grouped[group_namespace], sort_keys=False)
1936 with _secure_manifest_file(group_manifest) as manifest_path:
1937 code, live_output, stderr = run_kubectl(
1938 ["get", "-f", manifest_path, "-n", group_namespace, "-o", "json"],
1939 kubeconfig,
1940 command_timeout_seconds=_validation_command_timeout(
1941 deadline, KUBECTL_VALIDATION_COMMAND_TIMEOUT_SECONDS
1942 ),
1943 log_output=False,
1944 )
1945 if code == -1:
1946 raise _ValidationTimeout(f"kubectl get timed out for release {release!r}")
1947 if code != 0:
1948 raise RuntimeError(
1949 "kubectl could not retrieve every rendered object: "
1950 f"{_bounded_diagnostic(stderr or live_output)}"
1951 )
1952 live_payload = _parse_json_object(live_output, f"kubectl get for {release}")
1953 live_resources.extend(_flatten_resources(live_payload, f"kubectl output for {release}"))
1954 _compare_resource_identities(rendered, live_resources, release, namespace)
1956 for resource in live_resources:
1957 _validate_resource_readiness(resource)
1958 _validate_service_endpoints(resource, kubeconfig, namespace, deadline)
1959 return len(live_resources)
1962def _validate_disabled_release(
1963 release: str, namespace: str, kubeconfig: str, deadline: float
1964) -> None:
1965 code, stdout, stderr = run_helm(
1966 ["status", release, "-n", namespace, "-o", "json"],
1967 kubeconfig,
1968 command_timeout_seconds=_validation_command_timeout(
1969 deadline, HELM_VALIDATION_COMMAND_TIMEOUT_SECONDS
1970 ),
1971 log_output=False,
1972 )
1973 if code == -1:
1974 raise _ValidationTimeout(f"helm status timed out for disabled release {release!r}")
1975 if code == 0:
1976 raise RuntimeError("disabled release is still present")
1977 if stdout.strip() or stderr.strip() != _HELM_RELEASE_NOT_FOUND:
1978 raise RuntimeError(
1979 "disabled release absence is ambiguous; expected exact "
1980 f"{_HELM_RELEASE_NOT_FOUND!r}, got {_bounded_diagnostic(stderr or stdout)!r}"
1981 )
1984def validate_releases(event: dict[str, Any], kubeconfig: str) -> dict[str, Any]:
1985 """Validate exact Helm state and Kubernetes convergence for every release."""
1986 configurations, enabled_releases = _release_configurations(event)
1987 release_evidence: list[dict[str, Any]] = []
1988 failures: list[str] = []
1989 expected_resources = 0
1990 validated_resources = 0
1991 validated_releases = 0
1992 deadline = time.monotonic() + HELM_VALIDATION_TOTAL_TIMEOUT_SECONDS
1993 gateway_crd_evidence: list[dict[str, Any]] = []
1994 if LBC_CHART_NAME in enabled_releases:
1995 try:
1996 gateway_crd_evidence = _validate_gateway_crds(kubeconfig, deadline)
1997 except _ValidationTimeout:
1998 raise
1999 except Exception as exc:
2000 raise RuntimeError(
2001 f"pinned Gateway CRD validation failed: {_bounded_diagnostic(exc, 800)}"
2002 ) from exc
2003 gateway_resource_count = sum(item["object_count"] for item in gateway_crd_evidence)
2004 expected_resources += gateway_resource_count
2005 validated_resources += gateway_resource_count
2007 for release, config in configurations:
2008 enabled = release in enabled_releases
2009 try:
2010 chart, version, namespace = _release_metadata(release, config)
2011 if enabled:
2012 resource_count = _validate_enabled_release(
2013 release, chart, version, namespace, kubeconfig, deadline
2014 )
2015 state = "deployed"
2016 expected_resources += resource_count
2017 validated_resources += resource_count
2018 else:
2019 _validate_disabled_release(release, namespace, kubeconfig, deadline)
2020 resource_count = 0
2021 state = "absent"
2022 release_evidence.append(
2023 {
2024 "release": release,
2025 "namespace": namespace,
2026 "chart": chart,
2027 "version": version,
2028 "enabled": enabled,
2029 "status": state,
2030 "resource_count": resource_count,
2031 }
2032 )
2033 validated_releases += 1
2034 except _ValidationTimeout as exc:
2035 failures.append(f"{release}: {_bounded_diagnostic(exc, 600)}")
2036 # A timed-out control plane/storage backend is systemic. Continuing
2037 # would only consume the remaining Lambda budget and risk skipping
2038 # status recording and secure-file cleanup at the hard deadline.
2039 break
2040 except Exception as exc:
2041 failures.append(f"{release}: {_bounded_diagnostic(exc, 600)}")
2043 if failures:
2044 shown = failures[:8]
2045 if len(failures) > len(shown):
2046 shown.append(f"... and {len(failures) - len(shown)} more failure(s)")
2047 summary = f"validated {validated_releases}/{len(configurations)} releases; " + "; ".join(
2048 shown
2049 )
2050 raise RuntimeError(_bounded_diagnostic(summary))
2052 return {
2053 "status": "validated",
2054 "DeploymentToken": event.get("DeploymentToken"),
2055 "expected_release_count": len(configurations),
2056 "validated_release_count": validated_releases,
2057 "expected_resource_count": expected_resources,
2058 "validated_resource_count": validated_resources,
2059 "enabled_release_count": len(enabled_releases),
2060 "disabled_release_count": len(configurations) - len(enabled_releases),
2061 "gateway_crd_bundles": gateway_crd_evidence,
2062 "releases": release_evidence,
2063 }
2066def _cleanup_stale_webhooks(kubeconfig: str) -> None:
2067 """Remove MutatingWebhookConfigurations whose service endpoints are unavailable.
2069 When a webhook's backing pod is down (evicted, pending, crashed), the webhook
2070 blocks all API mutations for the resources it intercepts. This function detects
2071 and temporarily removes such webhooks so other Helm charts can upgrade.
2072 The webhook will be recreated when its chart is successfully reinstalled.
2073 """
2074 try:
2075 # Use kubectl to check for stale webhooks (simpler than kubernetes
2076 # Python client). Everything below builds its own KUBECONFIG env; an
2077 # earlier revision also spawned a throwaway `helm --kubeconfig` here
2078 # "for the env" and discarded every result, which was one helm
2079 # process per cleanup for nothing.
2080 import subprocess
2082 env = os.environ.copy()
2083 env["KUBECONFIG"] = kubeconfig
2085 result = subprocess.run(
2086 [
2087 "kubectl",
2088 "get",
2089 "mutatingwebhookconfigurations",
2090 "-o",
2091 "jsonpath={range .items[*]}{.metadata.name}{'\\n'}{end}",
2092 ],
2093 capture_output=True,
2094 text=True,
2095 env=env,
2096 timeout=30,
2097 check=False,
2098 )
2099 if result.returncode != 0:
2100 logger.warning(f"Failed to list webhooks: {result.stderr}")
2101 return
2103 for webhook_name in result.stdout.strip().split("\n"):
2104 if not webhook_name:
2105 continue
2107 # Check if the webhook's service has ready endpoints
2108 svc_result = subprocess.run(
2109 [
2110 "kubectl",
2111 "get",
2112 "mutatingwebhookconfiguration",
2113 webhook_name,
2114 "-o",
2115 "jsonpath={.webhooks[0].clientConfig.service.namespace}/{.webhooks[0].clientConfig.service.name}",
2116 ],
2117 capture_output=True,
2118 text=True,
2119 env=env,
2120 timeout=15,
2121 check=False,
2122 )
2123 if svc_result.returncode != 0 or "/" not in svc_result.stdout:
2124 continue
2126 ns, svc = svc_result.stdout.strip().split("/", 1)
2128 # Check if the service has ready endpoints
2129 ep_result = subprocess.run(
2130 [
2131 "kubectl",
2132 "get",
2133 "endpoints",
2134 svc,
2135 "-n",
2136 ns,
2137 "-o",
2138 "jsonpath={.subsets[*].addresses[*].ip}",
2139 ],
2140 capture_output=True,
2141 text=True,
2142 env=env,
2143 timeout=15,
2144 check=False,
2145 )
2147 if not ep_result.stdout.strip():
2148 logger.warning(
2149 f"Webhook {webhook_name} has no ready endpoints "
2150 f"(service {ns}/{svc}), temporarily removing..."
2151 )
2152 subprocess.run(
2153 ["kubectl", "delete", "mutatingwebhookconfiguration", webhook_name],
2154 capture_output=True,
2155 text=True,
2156 env=env,
2157 timeout=15,
2158 check=False,
2159 )
2161 except Exception as e:
2162 logger.warning(f"Webhook cleanup failed (non-fatal): {e}")
2165def handle_task(event: dict[str, Any]) -> dict[str, Any]:
2166 """Step Functions task entrypoint: install or uninstall a single chart.
2168 Each chart is its own state-machine task, so this performs exactly one
2169 helm operation per invocation and raises on failure — retries and ordering
2170 are owned by the state machine, not this function. That keeps every
2171 invocation comfortably under the Lambda timeout and gives per-chart retry
2172 and observability in the Step Functions console.
2174 Event shape (from the state machine task payload)::
2176 {
2177 "Action": "install_chart" | "uninstall_chart" |
2178 "quiesce_health_monitor" | "validate_releases",
2179 "Chart": "<chart name as keyed in charts.yaml>", # omitted for quiesce/validate
2180 "ClusterName": "...", "Region": "...",
2181 "EnabledCharts": ["keda", ...],
2182 "KedaOperatorRoleArn": "arn:...", # optional
2183 "Charts": { "<name>": { ...overrides... } } # optional
2184 }
2186 Returns a small status dict on success; raises on failure so the state
2187 machine's Retry/Catch handles it.
2188 """
2189 action = event["Action"]
2190 cluster_name = event.get("ClusterName") or os.environ["CLUSTER_NAME"]
2191 region = event.get("Region") or os.environ["REGION"]
2193 if action == "quiesce_health_monitor":
2194 kubeconfig = configure_kubeconfig(cluster_name, region)
2195 try:
2196 success, message = quiesce_health_monitor(kubeconfig)
2197 if not success:
2198 raise RuntimeError(f"health-monitor quiesce failed: {message}")
2199 return {"status": "quiesced", "message": message}
2200 finally:
2201 with contextlib.suppress(Exception):
2202 os.remove(kubeconfig)
2204 if action == "validate_releases":
2205 try:
2206 kubeconfig = configure_kubeconfig(cluster_name, region)
2207 try:
2208 evidence = validate_releases(event, kubeconfig)
2209 finally:
2210 # Validation must not report success while credentials remain
2211 # in a reusable warm Lambda filesystem. An unlink failure is a
2212 # validation failure and is recorded by the outer handler.
2213 _remove_validation_file(kubeconfig)
2214 except Exception as exc:
2215 diagnostic = _bounded_diagnostic(exc)
2216 _record_addon_status("helm-validation", "failed", diagnostic)
2217 raise RuntimeError(f"helm release validation failed: {diagnostic}") from exc
2219 message = (
2220 f"validated {evidence['validated_release_count']}/"
2221 f"{evidence['expected_release_count']} releases and "
2222 f"{evidence['validated_resource_count']}/"
2223 f"{evidence['expected_resource_count']} resources"
2224 )
2225 _record_addon_status("helm-validation", "validated", message)
2226 return evidence
2228 chart_name = event["Chart"]
2229 enabled_charts = event.get("EnabledCharts") or []
2230 chart_overrides = event.get("Charts") or {}
2231 keda_operator_role_arn = event.get("KedaOperatorRoleArn")
2233 default_config = load_charts_config().get("charts", {})
2234 config = dict(default_config.get(chart_name, {}))
2235 if chart_name in chart_overrides:
2236 config = deep_merge(config, chart_overrides[chart_name])
2238 is_enabled = chart_name in enabled_charts
2240 # Inject the KEDA operator IAM role ARN for IRSA, mirroring the legacy
2241 # custom-resource path.
2242 if chart_name == "keda" and keda_operator_role_arn:
2243 keda_values = config.setdefault("values", {})
2244 service_account = keda_values.setdefault("serviceAccount", {})
2245 operator = service_account.setdefault("operator", {})
2246 annotations = operator.setdefault("annotations", {})
2247 annotations["eks.amazonaws.com/role-arn"] = keda_operator_role_arn
2249 namespace = config.get("namespace", "default")
2250 kubeconfig = configure_kubeconfig(cluster_name, region)
2251 try:
2252 disabled_install = action == "install_chart" and not is_enabled
2253 if action == "install_chart" and is_enabled and chart_name == LBC_CHART_NAME:
2254 _apply_gateway_crds(kubeconfig)
2255 if action == "uninstall_chart" or disabled_install:
2256 # Disabled chart on an install pass: ensure it's gone (idempotent).
2257 # Helm's explicit "release: not found" is already reported as
2258 # success by uninstall_chart; every other error must escape so
2259 # CloudFormation teardown cannot continue against a live release.
2260 success, message = uninstall_chart(chart_name, namespace, kubeconfig)
2261 if not success:
2262 _record_addon_status(chart_name, "failed", message)
2263 raise RuntimeError(f"helm uninstall {chart_name} failed: {message}")
2264 if disabled_install:
2265 message = f"uninstalled (disabled): {message}"
2266 _record_addon_status(chart_name, "uninstalled", message)
2267 return {
2268 "chart": chart_name,
2269 "status": "uninstalled",
2270 "message": message,
2271 }
2273 if action == "install_chart":
2274 value_overrides = chart_overrides.get(chart_name, {}).get("values", {})
2275 success, message = install_chart(chart_name, config, kubeconfig, value_overrides)
2276 if not success:
2277 _record_addon_status(chart_name, "failed", message)
2278 # Raise so the state machine retries this single chart with
2279 # backoff rather than failing the whole deploy.
2280 raise RuntimeError(f"helm install {chart_name} failed: {message}")
2281 _record_addon_status(chart_name, "installed", message)
2282 return {"chart": chart_name, "status": "installed", "message": message}
2284 raise ValueError(f"Unknown Action: {action!r}")
2285 finally:
2286 with contextlib.suppress(Exception):
2287 os.remove(kubeconfig)
2290def lambda_handler(event: dict[str, Any], context: Any) -> Any:
2291 """Main Lambda handler.
2293 Two entrypoints share this function:
2295 - **Step Functions task** (the current install path): the event carries an
2296 ``Action`` key and is dispatched to :func:`handle_task`, which operates on
2297 a single chart and raises on failure.
2298 - **CloudFormation custom resource** (legacy/fallback): the event carries a
2299 ``RequestType`` and the whole-chart-set loop below runs.
2300 """
2301 if event.get("Action"):
2302 logger.info(
2303 "Task event: action=%s chart=%s cluster=%s region=%s",
2304 event.get("Action"),
2305 event.get("Chart"),
2306 event.get("ClusterName"),
2307 event.get("Region"),
2308 )
2309 return handle_task(event)
2311 logger.info(f"Received event: {json.dumps(event)}")
2313 request_type = event["RequestType"]
2314 physical_id = event.get("PhysicalResourceId", f"helm-{event['LogicalResourceId']}")
2316 try:
2317 props = event["ResourceProperties"]
2318 cluster_name = props["ClusterName"]
2319 region = props["Region"]
2321 # Load default config and merge with overrides
2322 default_config = load_charts_config()
2323 charts_config = default_config.get("charts", {})
2325 # Apply chart overrides from CloudFormation, which the regional stack
2326 # sends as one JSON string so their types survive (see
2327 # _cfn_chart_overrides).
2328 chart_overrides = _cfn_chart_overrides(props)
2329 for chart_name, overrides in chart_overrides.items():
2330 if chart_name in charts_config:
2331 charts_config[chart_name] = deep_merge(charts_config[chart_name], overrides)
2332 else:
2333 charts_config[chart_name] = overrides
2335 # Apply enabled charts list
2336 enabled_charts = props.get("EnabledCharts", [])
2337 if enabled_charts:
2338 for chart_name in charts_config:
2339 charts_config[chart_name]["enabled"] = chart_name in enabled_charts
2341 # Inject KEDA operator IAM role ARN for IRSA if provided
2342 keda_operator_role_arn = props.get("KedaOperatorRoleArn")
2343 if keda_operator_role_arn and "keda" in charts_config:
2344 logger.info(f"Injecting KEDA operator role ARN: {keda_operator_role_arn}")
2345 keda_values = charts_config["keda"].setdefault("values", {})
2346 service_account = keda_values.setdefault("serviceAccount", {})
2347 operator = service_account.setdefault("operator", {})
2348 annotations = operator.setdefault("annotations", {})
2349 annotations["eks.amazonaws.com/role-arn"] = keda_operator_role_arn
2351 # Configure kubeconfig
2352 kubeconfig = configure_kubeconfig(cluster_name, region)
2354 results = {}
2355 failed = []
2356 uninstall_failed = []
2358 if request_type in ("Create", "Update"):
2359 # Install/upgrade enabled charts with retry for transient failures
2360 # (e.g., webhook not ready yet, API server temporarily unavailable).
2361 # Tunables at the top of this module.
2362 max_retries = HELM_INSTALL_MAX_RETRIES
2363 retry_delay = HELM_INSTALL_RETRY_DELAY_SECONDS
2365 # First pass: uninstall disabled charts that were previously installed.
2366 # A genuine uninstall error is a failed convergence operation; only
2367 # Helm's explicit "not found" result is idempotent success.
2368 for chart_name, config in charts_config.items():
2369 if not config.get("enabled", False):
2370 namespace = config.get("namespace", "default")
2371 logger.info(f"Chart {chart_name} is disabled, checking if installed...")
2372 success, message = uninstall_chart(chart_name, namespace, kubeconfig)
2373 if success:
2374 message = f"uninstalled (disabled): {message}"
2375 results[chart_name] = message
2376 if not success:
2377 uninstall_failed.append(chart_name)
2379 # Second pass: install/upgrade enabled charts
2380 for chart_name, config in charts_config.items():
2381 if not config.get("enabled", False):
2382 continue
2384 if chart_name == LBC_CHART_NAME:
2385 _apply_gateway_crds(kubeconfig)
2386 value_overrides = chart_overrides.get(chart_name, {}).get("values", {})
2387 success, message = install_chart(chart_name, config, kubeconfig, value_overrides)
2388 results[chart_name] = message
2390 if not success:
2391 failed.append(chart_name)
2393 # Retry failed charts — transient issues (webhook races, API timeouts)
2394 # often resolve after other charts finish installing
2395 for attempt in range(1, max_retries + 1):
2396 if not failed:
2397 break
2399 # If failures look like webhook issues, temporarily remove stale
2400 # MutatingWebhookConfigurations whose endpoints are unavailable.
2401 # This breaks the deadlock where a down webhook blocks all upgrades.
2402 if any(
2403 "webhook" in results.get(c, "").lower()
2404 or "no endpoints" in results.get(c, "").lower()
2405 for c in failed
2406 ):
2407 logger.info("Detected webhook-related failures, cleaning stale webhooks...")
2408 _cleanup_stale_webhooks(kubeconfig)
2410 logger.info(
2411 f"Retrying {len(failed)} failed chart(s) "
2412 f"(attempt {attempt}/{max_retries}, waiting {retry_delay}s)..."
2413 )
2414 import time
2416 time.sleep(retry_delay)
2418 retry_list = failed.copy()
2419 failed = []
2420 for chart_name in retry_list:
2421 config = charts_config[chart_name]
2422 value_overrides = chart_overrides.get(chart_name, {}).get("values", {})
2423 success, message = install_chart(
2424 chart_name, config, kubeconfig, value_overrides
2425 )
2426 results[chart_name] = message
2427 if not success:
2428 failed.append(chart_name)
2429 else:
2430 logger.info(f"Retry succeeded for {chart_name}")
2432 if failed:
2433 logger.warning(f"Charts still failing after {max_retries} retries: {failed}")
2435 # Keep uninstall failures out of the install retry loop: retrying
2436 # one as an install would recreate the disabled release. They still
2437 # participate in the final FAILED response.
2438 failed.extend(uninstall_failed)
2440 elif request_type == "Delete":
2441 # Uninstall charts (in reverse order)
2442 for chart_name, config in reversed(list(charts_config.items())):
2443 if not config.get("enabled", False):
2444 continue
2446 namespace = config.get("namespace", "default")
2447 success, message = uninstall_chart(chart_name, namespace, kubeconfig)
2448 results[chart_name] = message
2450 if not success:
2451 failed.append(chart_name)
2453 # Clean up kubeconfig
2454 import contextlib
2456 with contextlib.suppress(Exception):
2457 os.remove(kubeconfig)
2459 # Prepare response
2460 response_data = {
2461 "Results": json.dumps(results),
2462 "InstalledCharts": ",".join(
2463 [k for k, v in results.items() if "Successfully" in str(v)]
2464 ),
2465 "FailedCharts": ",".join(failed),
2466 }
2468 if failed:
2469 send_response(
2470 event,
2471 context,
2472 FAILED,
2473 response_data,
2474 physical_id,
2475 f"Failed charts: {', '.join(failed)}",
2476 )
2477 else:
2478 send_response(event, context, SUCCESS, response_data, physical_id)
2480 except Exception as e:
2481 logger.error(f"Error: {e}", exc_info=True)
2482 # Delete errors are intentionally failures. Reporting success here lets
2483 # CloudFormation remove the EKS/access resources while Helm releases
2484 # (and their external load balancers/webhooks) are still live.
2485 send_response(event, context, FAILED, {}, physical_id, str(e))