Coverage for lambda / helm-orchestrator / handler.py: 100.00%
135 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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-01T14:42:56Z
55# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
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 _encode_replay_input(execution_input_json: str) -> str:
87 """Encode the replay input as zlib+base64 for SSM Parameter Store.
89 The raw execution input embeds ``{{PLACEHOLDER}}`` image-replacement keys,
90 and SSM rejects any String value containing ``{{}}`` ("Parameter value
91 can't nest another parameter"). Base64 is brace-free by construction and
92 zlib keeps the highly repetitive JSON inside the 8 KiB Advanced-tier bound.
93 Consumers (``gco stacks addons install`` and live release validation)
94 reverse this exact encoding before use.
95 """
96 compressed = zlib.compress(execution_input_json.encode("utf-8"), 9)
97 return base64.b64encode(compressed).decode("ascii")
100def _bounded_error_text(exc: BaseException) -> str:
101 """Render an exception for a CloudFormation response without overflowing it.
103 Botocore validation errors echo the full offending parameter value, and a
104 custom-resource response larger than 4 KiB is rejected wholesale with the
105 unhelpful "Response object is too long" — masking the real failure.
106 """
107 text = f"{type(exc).__name__}: {exc}"
108 if len(text) > _BOUNDED_ERROR_CHARS:
109 text = text[:_BOUNDED_ERROR_CHARS] + " ...[truncated; see provider logs]"
110 return text
113def _execution_name(request_id: Any, generation: int = 0) -> str:
114 """Return a retry-stable, Step-Functions-safe generation name."""
115 request_digest = hashlib.sha256(str(request_id).encode("utf-8")).hexdigest()[:48]
116 suffix = "" if generation == 0 else f"-{generation}"
117 return f"helm-install-{request_digest}{suffix}"
120def _execution_arn(state_machine_arn: str, execution_name: str) -> str:
121 """Build the execution ARN corresponding to a state-machine ARN."""
122 prefix, separator, state_machine_name = state_machine_arn.partition(":stateMachine:")
123 if not separator or not state_machine_name:
124 raise ValueError(f"Invalid Step Functions state machine ARN: {state_machine_arn}")
125 return f"{prefix}:execution:{state_machine_name}:{execution_name}"
128def _start_or_adopt_execution(
129 sfn_client: Any,
130 *,
131 state_machine_arn: str,
132 execution_input_json: str,
133 request_id: Any,
134) -> dict[str, Any]:
135 """Start one retry-safe execution or adopt an identical running attempt."""
136 if request_id is None:
137 return dict(
138 sfn_client.start_execution(
139 stateMachineArn=state_machine_arn,
140 input=execution_input_json,
141 )
142 )
144 for generation in range(_MAX_EXECUTION_NAME_GENERATIONS):
145 execution_name = _execution_name(request_id, generation)
146 try:
147 return dict(
148 sfn_client.start_execution(
149 stateMachineArn=state_machine_arn,
150 input=execution_input_json,
151 name=execution_name,
152 )
153 )
154 except ClientError as exc:
155 if exc.response.get("Error", {}).get("Code") != "ExecutionAlreadyExists":
156 raise
158 execution_arn = _execution_arn(state_machine_arn, execution_name)
159 existing = sfn_client.describe_execution(executionArn=execution_arn)
160 if existing.get("status") == "RUNNING":
161 if existing.get("input") != execution_input_json:
162 raise RuntimeError(
163 f"Running retry execution {execution_arn} has non-identical input"
164 )
165 return {
166 "executionArn": execution_arn,
167 "startDate": existing.get("startDate"),
168 }
170 raise RuntimeError(
171 f"Exhausted {_MAX_EXECUTION_NAME_GENERATIONS} retry-safe execution generations"
172 )
175def _stop_execution_and_wait(sfn_client: Any, execution_arn: str) -> None:
176 """Stop an untracked execution and require terminal confirmation."""
177 sfn_client.stop_execution(executionArn=execution_arn)
178 deadline = time.monotonic() + _STOP_CONFIRMATION_SECONDS
179 while time.monotonic() < deadline:
180 status = sfn_client.describe_execution(executionArn=execution_arn).get("status")
181 if status != "RUNNING":
182 return
183 time.sleep(_STOP_CONFIRMATION_POLL_SECONDS)
184 raise TimeoutError(f"Execution {execution_arn} remained RUNNING after StopExecution")
187def _prepare_teardown_fence(ssm_client: Any, *, request_type: str, fence_name: str) -> None:
188 """Clear a stale fence on create and reject convergence during deletion."""
189 if request_type == "Create":
190 try:
191 ssm_client.delete_parameter(Name=fence_name)
192 except ClientError as exc:
193 if exc.response.get("Error", {}).get("Code") != "ParameterNotFound":
194 raise
195 return
197 try:
198 ssm_client.get_parameter(Name=fence_name)
199 except ClientError as exc:
200 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound":
201 return
202 raise
203 raise RuntimeError(f"Regional add-on teardown fence is active: {fence_name}")
206def _started_at(response: dict[str, Any]) -> int:
207 """Return the execution start time as Unix seconds."""
208 start_date = response.get("startDate")
209 if start_date is None:
210 return int(time.time())
211 if isinstance(start_date, int | float):
212 return int(start_date)
213 return int(start_date.timestamp())
216def on_event(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
217 """Start a state-machine execution for Create/Update; no-op for Delete.
219 Returns immediately after starting the execution — the custom resource is
220 fire-and-forget, so CloudFormation considers the resource created as soon as
221 the helm-install state machine has been *kicked off*, never waiting for the
222 charts to finish. The started ``ExecutionArn`` is returned in ``Data`` purely
223 as an observability attribute (it does not gate resource completion).
224 """
225 request_type = event["RequestType"]
226 logger.info("on_event %s: %s", request_type, json.dumps(event.get("ResourceProperties", {})))
228 physical_id = event.get("PhysicalResourceId") or "helm-install-charts"
230 if request_type == "Delete":
231 # Charts are torn down with the cluster; nothing to do here.
232 return {"PhysicalResourceId": physical_id}
234 props = event["ResourceProperties"]
235 state_machine_arn = os.environ["STATE_MACHINE_ARN"]
236 deployment_token = props["DeploymentTimestamp"]
238 # The execution input carries everything the convergence pipeline's tasks
239 # need, including the registry identity consumed by the unconditional final
240 # Gateway endpoint-publication task. EndpointGroupArn is present only where
241 # Global Accelerator exists.
242 execution_input = {
243 "ClusterName": props["ClusterName"],
244 "Region": props["Region"],
245 "RegistryRegion": props["RegistryRegion"],
246 "ProjectName": props["ProjectName"],
247 "EnabledCharts": props.get("EnabledCharts", []),
248 "Charts": props.get("Charts", {}),
249 "KedaOperatorRoleArn": props.get("KedaOperatorRoleArn"),
250 "ImageReplacements": props.get("ImageReplacements", {}),
251 "DeploymentToken": deployment_token,
252 }
253 endpoint_group_arn = props.get("EndpointGroupArn")
254 if endpoint_group_arn:
255 execution_input["EndpointGroupArn"] = endpoint_group_arn
256 execution_input_json = _canonical_json(execution_input)
257 execution_input_bytes = execution_input_json.encode("utf-8")
258 # No raw-JSON size gate here: what SSM stores is the zlib+base64 encoding,
259 # whose own bound is enforced below before any write. Gating on the raw
260 # bytes rejected deployments whose encoded form fit comfortably — caught
261 # live by example-job validation run ex241-edf33111-r2, where enabling
262 # every optional chart pushed the raw input to 8771 bytes while its
263 # encoded form stayed under half the Advanced-tier limit.
265 project = str(props["ProjectName"])
266 region = str(props["Region"])
267 parameter_root = f"/{project}/addons/{region}"
268 input_sha256 = hashlib.sha256(execution_input_bytes).hexdigest()
269 ssm = _ssm()
270 fence_name = f"{parameter_root}/_teardown"
271 _prepare_teardown_fence(
272 ssm,
273 request_type=request_type,
274 fence_name=fence_name,
275 )
277 # Persist the exact replay input before convergence can mutate the cluster.
278 # The value is zlib+base64 encoded because SSM rejects raw ``{{}}`` tokens;
279 # ``input_sha256`` below is always computed over the raw canonical JSON.
280 # Intelligent-Tiering selects an Advanced parameter only when the payload
281 # exceeds the Standard tier's 4 KiB limit, while retaining the 8 KiB bound
282 # enforced above.
283 encoded_input = _encode_replay_input(execution_input_json)
284 if len(encoded_input) > _SSM_PARAMETER_MAX_BYTES:
285 raise ValueError(
286 f"Encoded convergence replay input is {len(encoded_input)} bytes; "
287 f"SSM Parameter Store supports at most {_SSM_PARAMETER_MAX_BYTES} bytes"
288 )
289 try:
290 ssm.put_parameter(
291 Name=f"{parameter_root}/_input",
292 Value=encoded_input,
293 Type="String",
294 Tier="Intelligent-Tiering",
295 Overwrite=True,
296 )
297 except Exception as exc:
298 logger.error("Replay-input persistence failed", exc_info=True)
299 raise RuntimeError(
300 "Could not persist the convergence replay input: " + _bounded_error_text(exc)
301 ) from exc
303 request_id = event.get("RequestId")
304 sfn_client = _sfn()
305 resp = _start_or_adopt_execution(
306 sfn_client,
307 state_machine_arn=state_machine_arn,
308 execution_input_json=execution_input_json,
309 request_id=request_id,
310 )
311 execution_arn = resp["executionArn"]
313 execution_metadata = {
314 "execution_arn": execution_arn,
315 "state_machine_arn": state_machine_arn,
316 "deployment_token": deployment_token,
317 "cluster_name": props["ClusterName"],
318 "region": region,
319 "input_sha256": input_sha256,
320 "started_at": _started_at(resp),
321 }
322 try:
323 ssm.put_parameter(
324 Name=f"{parameter_root}/_execution",
325 Value=_canonical_json(execution_metadata),
326 Type="String",
327 Overwrite=True,
328 )
329 except Exception as exc:
330 # Do not leave an untracked execution mutating a stack whose provider
331 # is about to fail and potentially roll back. The replay input remains
332 # available for an explicit retry.
333 logger.error("Execution-metadata persistence failed", exc_info=True)
334 try:
335 _stop_execution_and_wait(sfn_client, execution_arn)
336 except Exception as stop_exc: # noqa: BLE001 - surface unsafe rollback state
337 raise RuntimeError(
338 f"Could not confirm untracked execution {execution_arn} stopped"
339 ) from stop_exc
340 raise RuntimeError(
341 "Could not persist the convergence execution identity: " + _bounded_error_text(exc)
342 ) from exc
344 logger.info("Started and recorded execution %s (fire-and-forget)", execution_arn)
346 return {
347 "PhysicalResourceId": physical_id,
348 "ExecutionArn": execution_arn,
349 "Data": {"ExecutionArn": execution_arn},
350 }