Coverage for lambda / helm-orchestrator / handler.py: 100.00%
145 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Helm Orchestrator — CloudFormation custom-resource provider for the helm
3install Step Functions state machine.
5This is a thin **fire-and-forget** provider (CDK ``cr.Provider`` with only an
6``onEvent`` handler — no ``isComplete`` waiter). It does no Helm or Kubernetes
7work itself; that all happens in the per-chart Step Functions tasks. Its only
8job is:
10- ``on_event``: on Create/Update, start a state-machine execution whose input
11 carries the chart configuration, persist its exact convergence identity, and
12 return success immediately; on Delete, no-op (the cluster teardown removes
13 the charts).
15Add-on installation is intentionally decoupled from the CloudFormation lifecycle
16entirely. Earlier this provider polled the execution to completion via an
17``isComplete`` handler, but that re-coupled the cluster's create to the helm
18batch: a slow chart (e.g. volcano retrying docker.io image pulls) keeps the
19execution ``RUNNING`` past CloudFormation's ~1-hour custom-resource ceiling, at
20which point CloudFormation declares "did not receive a response" and rolls back
21— destroying the freshly-created EKS cluster over a recoverable add-on problem.
23So the custom resource now reports success as soon as the execution is *started*.
24The state machine then converges every chart it can in the background (each chart
25task catches its own failure and continues, so one broken chart never blocks the
26rest). The real per-chart outcome is recorded out-of-band in SSM by the installer
27tasks (``/<project>/addons/<region>/<chart>``) and surfaced via
28``gco stacks addons status``; a degraded add-on layer is re-converged with
29``gco stacks addons install`` rather than by tearing the cluster down.
31Keeping the heavy lifting in Step Functions (one task per chart, with per-chart
32retry) also means no single Lambda invocation is bound by the 15-minute Lambda
33limit.
35Environment Variables:
36 STATE_MACHINE_ARN: ARN of the helm-install state machine.
37"""
39from __future__ import annotations
41import base64
42import hashlib
43import json
44import logging
45import os
46import time
47import zlib
48from typing import Any
50import boto3
51from botocore.exceptions import ClientError
53# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
54# Generated at (UTC): 2026-09-27T04:10:54Z
55# Generated from Git commit: 590c275a9cd8dc0e0ba9f8ea74acefb63754df46
56# Flowchart(s) generated from this file:
57# * ``on_event`` -> ``diagrams/code_diagrams/lambda/helm-orchestrator/handler.on_event.html``
58# (PNG: ``diagrams/code_diagrams/lambda/helm-orchestrator/handler.on_event.png``)
59# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
60# <pyflowchart-code-diagram> END
63logger = logging.getLogger()
64logger.setLevel(logging.INFO)
66_SSM_PARAMETER_MAX_BYTES = 8 * 1024
67_MAX_EXECUTION_NAME_GENERATIONS = 100
68_STOP_CONFIRMATION_SECONDS = 10
69_STOP_CONFIRMATION_POLL_SECONDS = 0.2
70_BOUNDED_ERROR_CHARS = 512
73def _sfn() -> Any:
74 return boto3.client("stepfunctions")
77def _ssm() -> Any:
78 return boto3.client("ssm")
81def _canonical_json(value: Any) -> str:
82 """Serialize *value* deterministically for execution and persistence."""
83 return json.dumps(value, sort_keys=True, separators=(",", ":"))
86def _chart_overrides(props: dict[str, Any]) -> dict[str, Any]:
87 """Return the ``Charts`` overrides with the JSON types the stack gave them.
89 CloudFormation delivers custom-resource properties with every number and
90 boolean as a string, so the regional stack sends ``Charts`` as one JSON
91 string and it is decoded here. An object (a stack synthesized before
92 that change) is taken as it is.
93 """
94 charts = props.get("Charts", {})
95 if isinstance(charts, str):
96 try:
97 charts = json.loads(charts)
98 except json.JSONDecodeError as exc:
99 raise ValueError(f"Charts is not valid JSON: {exc}") from exc
100 if not isinstance(charts, dict):
101 raise ValueError("Charts must be a JSON object of per-chart overrides")
102 return charts
105def _encode_replay_input(execution_input_json: str) -> str:
106 """Encode the replay input as zlib+base64 for SSM Parameter Store.
108 The raw execution input embeds ``{{PLACEHOLDER}}`` image-replacement keys,
109 and SSM rejects any String value containing ``{{}}`` ("Parameter value
110 can't nest another parameter"). Base64 is brace-free by construction and
111 zlib keeps the highly repetitive JSON inside the 8 KiB Advanced-tier bound.
112 Consumers (``gco stacks addons install`` and live release validation)
113 reverse this exact encoding before use.
114 """
115 compressed = zlib.compress(execution_input_json.encode("utf-8"), 9)
116 return base64.b64encode(compressed).decode("ascii")
119def _bounded_error_text(exc: BaseException) -> str:
120 """Render an exception for a CloudFormation response without overflowing it.
122 Botocore validation errors echo the full offending parameter value, and a
123 custom-resource response larger than 4 KiB is rejected wholesale with the
124 unhelpful "Response object is too long" — masking the real failure.
125 """
126 text = f"{type(exc).__name__}: {exc}"
127 if len(text) > _BOUNDED_ERROR_CHARS:
128 text = text[:_BOUNDED_ERROR_CHARS] + " ...[truncated; see provider logs]"
129 return text
132def _execution_name(request_id: Any, generation: int = 0) -> str:
133 """Return a retry-stable, Step-Functions-safe generation name."""
134 request_digest = hashlib.sha256(str(request_id).encode("utf-8")).hexdigest()[:48]
135 suffix = "" if generation == 0 else f"-{generation}"
136 return f"helm-install-{request_digest}{suffix}"
139def _execution_arn(state_machine_arn: str, execution_name: str) -> str:
140 """Build the execution ARN corresponding to a state-machine ARN."""
141 prefix, separator, state_machine_name = state_machine_arn.partition(":stateMachine:")
142 if not separator or not state_machine_name:
143 raise ValueError(f"Invalid Step Functions state machine ARN: {state_machine_arn}")
144 return f"{prefix}:execution:{state_machine_name}:{execution_name}"
147def _start_or_adopt_execution(
148 sfn_client: Any,
149 *,
150 state_machine_arn: str,
151 execution_input_json: str,
152 request_id: Any,
153) -> dict[str, Any]:
154 """Start one retry-safe execution or adopt an identical running attempt."""
155 if request_id is None:
156 return dict(
157 sfn_client.start_execution(
158 stateMachineArn=state_machine_arn,
159 input=execution_input_json,
160 )
161 )
163 for generation in range(_MAX_EXECUTION_NAME_GENERATIONS):
164 execution_name = _execution_name(request_id, generation)
165 try:
166 return dict(
167 sfn_client.start_execution(
168 stateMachineArn=state_machine_arn,
169 input=execution_input_json,
170 name=execution_name,
171 )
172 )
173 except ClientError as exc:
174 if exc.response.get("Error", {}).get("Code") != "ExecutionAlreadyExists":
175 raise
177 execution_arn = _execution_arn(state_machine_arn, execution_name)
178 existing = sfn_client.describe_execution(executionArn=execution_arn)
179 if existing.get("status") == "RUNNING":
180 if existing.get("input") != execution_input_json:
181 raise RuntimeError(
182 f"Running retry execution {execution_arn} has non-identical input"
183 )
184 return {
185 "executionArn": execution_arn,
186 "startDate": existing.get("startDate"),
187 }
189 raise RuntimeError(
190 f"Exhausted {_MAX_EXECUTION_NAME_GENERATIONS} retry-safe execution generations"
191 )
194def _stop_execution_and_wait(sfn_client: Any, execution_arn: str) -> None:
195 """Stop an untracked execution and require terminal confirmation."""
196 sfn_client.stop_execution(executionArn=execution_arn)
197 deadline = time.monotonic() + _STOP_CONFIRMATION_SECONDS
198 while time.monotonic() < deadline:
199 status = sfn_client.describe_execution(executionArn=execution_arn).get("status")
200 if status != "RUNNING":
201 return
202 time.sleep(_STOP_CONFIRMATION_POLL_SECONDS)
203 raise TimeoutError(f"Execution {execution_arn} remained RUNNING after StopExecution")
206def _prepare_teardown_fence(ssm_client: Any, *, request_type: str, fence_name: str) -> None:
207 """Clear a stale fence on create and reject convergence during deletion."""
208 if request_type == "Create":
209 try:
210 ssm_client.delete_parameter(Name=fence_name)
211 except ClientError as exc:
212 if exc.response.get("Error", {}).get("Code") != "ParameterNotFound":
213 raise
214 return
216 try:
217 ssm_client.get_parameter(Name=fence_name)
218 except ClientError as exc:
219 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound":
220 return
221 raise
222 raise RuntimeError(f"Regional add-on teardown fence is active: {fence_name}")
225def _started_at(response: dict[str, Any]) -> int:
226 """Return the execution start time as Unix seconds."""
227 start_date = response.get("startDate")
228 if start_date is None:
229 return int(time.time())
230 if isinstance(start_date, int | float):
231 return int(start_date)
232 return int(start_date.timestamp())
235def on_event(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
236 """Start a state-machine execution for Create/Update; no-op for Delete.
238 Returns immediately after starting the execution — the custom resource is
239 fire-and-forget, so CloudFormation considers the resource created as soon as
240 the helm-install state machine has been *kicked off*, never waiting for the
241 charts to finish. The started ``ExecutionArn`` is returned in ``Data`` purely
242 as an observability attribute (it does not gate resource completion).
243 """
244 request_type = event["RequestType"]
245 logger.info("on_event %s: %s", request_type, json.dumps(event.get("ResourceProperties", {})))
247 physical_id = event.get("PhysicalResourceId") or "helm-install-charts"
249 if request_type == "Delete":
250 # Charts are torn down with the cluster; nothing to do here.
251 return {"PhysicalResourceId": physical_id}
253 props = event["ResourceProperties"]
254 state_machine_arn = os.environ["STATE_MACHINE_ARN"]
255 deployment_token = props["DeploymentTimestamp"]
257 # The execution input carries everything the convergence pipeline's tasks
258 # need, including the registry identity consumed by the unconditional final
259 # Gateway endpoint-publication task. EndpointGroupArn is present only where
260 # Global Accelerator exists.
261 execution_input = {
262 "ClusterName": props["ClusterName"],
263 "Region": props["Region"],
264 "RegistryRegion": props["RegistryRegion"],
265 "ProjectName": props["ProjectName"],
266 "EnabledCharts": props.get("EnabledCharts", []),
267 "Charts": _chart_overrides(props),
268 "KedaOperatorRoleArn": props.get("KedaOperatorRoleArn"),
269 "ImageReplacements": props.get("ImageReplacements", {}),
270 "DeploymentToken": deployment_token,
271 }
272 endpoint_group_arn = props.get("EndpointGroupArn")
273 if endpoint_group_arn:
274 execution_input["EndpointGroupArn"] = endpoint_group_arn
275 execution_input_json = _canonical_json(execution_input)
276 execution_input_bytes = execution_input_json.encode("utf-8")
277 # No raw-JSON size gate here: what SSM stores is the zlib+base64 encoding,
278 # whose own bound is enforced below before any write. Gating on the raw
279 # bytes rejected deployments whose encoded form fit comfortably — caught
280 # live by example-job validation run ex241-edf33111-r2, where enabling
281 # every optional chart pushed the raw input to 8771 bytes while its
282 # encoded form stayed under half the Advanced-tier limit.
284 project = str(props["ProjectName"])
285 region = str(props["Region"])
286 parameter_root = f"/{project}/addons/{region}"
287 input_sha256 = hashlib.sha256(execution_input_bytes).hexdigest()
288 ssm = _ssm()
289 fence_name = f"{parameter_root}/_teardown"
290 _prepare_teardown_fence(
291 ssm,
292 request_type=request_type,
293 fence_name=fence_name,
294 )
296 # Persist the exact replay input before convergence can mutate the cluster.
297 # The value is zlib+base64 encoded because SSM rejects raw ``{{}}`` tokens;
298 # ``input_sha256`` below is always computed over the raw canonical JSON.
299 # Intelligent-Tiering selects an Advanced parameter only when the payload
300 # exceeds the Standard tier's 4 KiB limit, while retaining the 8 KiB bound
301 # enforced above.
302 encoded_input = _encode_replay_input(execution_input_json)
303 if len(encoded_input) > _SSM_PARAMETER_MAX_BYTES:
304 raise ValueError(
305 f"Encoded convergence replay input is {len(encoded_input)} bytes; "
306 f"SSM Parameter Store supports at most {_SSM_PARAMETER_MAX_BYTES} bytes"
307 )
308 try:
309 ssm.put_parameter(
310 Name=f"{parameter_root}/_input",
311 Value=encoded_input,
312 Type="String",
313 Tier="Intelligent-Tiering",
314 Overwrite=True,
315 )
316 except Exception as exc:
317 logger.error("Replay-input persistence failed", exc_info=True)
318 raise RuntimeError(
319 "Could not persist the convergence replay input: " + _bounded_error_text(exc)
320 ) from exc
322 request_id = event.get("RequestId")
323 sfn_client = _sfn()
324 resp = _start_or_adopt_execution(
325 sfn_client,
326 state_machine_arn=state_machine_arn,
327 execution_input_json=execution_input_json,
328 request_id=request_id,
329 )
330 execution_arn = resp["executionArn"]
332 execution_metadata = {
333 "execution_arn": execution_arn,
334 "state_machine_arn": state_machine_arn,
335 "deployment_token": deployment_token,
336 "cluster_name": props["ClusterName"],
337 "region": region,
338 "input_sha256": input_sha256,
339 "started_at": _started_at(resp),
340 }
341 try:
342 ssm.put_parameter(
343 Name=f"{parameter_root}/_execution",
344 Value=_canonical_json(execution_metadata),
345 Type="String",
346 Overwrite=True,
347 )
348 except Exception as exc:
349 # Do not leave an untracked execution mutating a stack whose provider
350 # is about to fail and potentially roll back. The replay input remains
351 # available for an explicit retry.
352 logger.error("Execution-metadata persistence failed", exc_info=True)
353 try:
354 _stop_execution_and_wait(sfn_client, execution_arn)
355 except Exception as stop_exc: # surface unsafe rollback state
356 raise RuntimeError(
357 f"Could not confirm untracked execution {execution_arn} stopped"
358 ) from stop_exc
359 raise RuntimeError(
360 "Could not persist the convergence execution identity: " + _bounded_error_text(exc)
361 ) from exc
363 logger.info("Started and recorded execution %s (fire-and-forget)", execution_arn)
365 return {
366 "PhysicalResourceId": physical_id,
367 "ExecutionArn": execution_arn,
368 "Data": {"ExecutionArn": execution_arn},
369 }