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