Coverage for lambda / helm-installer / teardown_provider.py: 100.00%
96 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Delete-only custom-resource provider for ordered Helm teardown.
3Create and update events are deliberate no-ops. On delete, the provider starts
4an ordered Step Functions execution that invokes the Helm worker once per chart
5and then waits for that execution to finish. A failed uninstall therefore fails
6the CloudFormation delete instead of allowing EKS access entries or the cluster
7to disappear underneath a still-live release.
9Environment variables:
10 TEARDOWN_STATE_MACHINE_ARN: Ordered uninstall state machine.
11 INSTALL_STATE_MACHINE_ARN: Fire-and-forget convergence state machine to
12 stop and drain before teardown starts.
13"""
15from __future__ import annotations
17import hashlib
18import json
19import logging
20import os
21from typing import Any
23import boto3
24from botocore.exceptions import ClientError
26# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
27# Generated at (UTC): 2026-09-01T14:42:56Z
28# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc
29# Flowchart(s) generated from this file:
30# * ``on_event`` -> ``diagrams/code_diagrams/lambda/helm-installer/teardown_provider.on_event.html``
31# (PNG: ``diagrams/code_diagrams/lambda/helm-installer/teardown_provider.on_event.png``)
32# * ``is_complete`` -> ``diagrams/code_diagrams/lambda/helm-installer/teardown_provider.is_complete.html``
33# (PNG: ``diagrams/code_diagrams/lambda/helm-installer/teardown_provider.is_complete.png``)
34# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
35# <pyflowchart-code-diagram> END
38logger = logging.getLogger()
39logger.setLevel(logging.INFO)
41_IN_FLIGHT_LAMBDA_DRAIN_SECONDS = 16 * 60
44def _sfn() -> Any:
45 return boto3.client("stepfunctions")
48def _ssm(region: str) -> Any:
49 return boto3.client("ssm", region_name=region)
52def _execution_name(event: dict[str, Any]) -> str:
53 """Return a retry-stable, Step-Functions-safe execution name."""
54 identity = "|".join(
55 str(event.get(key, "")) for key in ("StackId", "RequestId", "LogicalResourceId")
56 )
57 return f"helm-delete-{hashlib.sha256(identity.encode()).hexdigest()[:32]}"
60def _execution_arn(state_machine_arn: str, execution_name: str) -> str:
61 """Build the execution ARN corresponding to a state-machine ARN."""
62 prefix, separator, state_machine_name = state_machine_arn.partition(":stateMachine:")
63 if not separator or not state_machine_name:
64 raise ValueError(f"Invalid Step Functions state machine ARN: {state_machine_arn}")
65 return f"{prefix}:execution:{state_machine_name}:{execution_name}"
68def _stop_running_install_executions(sfn_client: Any, state_machine_arn: str) -> int:
69 """Stop every visible convergence execution and return the count stopped.
71 ``ListExecutions`` is eventually consistent. The delete provider invokes
72 this once before the teardown state machine starts its unconditional drain;
73 the state machine then invokes ``drain_install_executions`` after each full
74 drain interval and loops whenever a late execution is discovered.
75 """
76 execution_arns: list[str] = []
77 request: dict[str, Any] = {
78 "stateMachineArn": state_machine_arn,
79 "statusFilter": "RUNNING",
80 "maxResults": 100,
81 }
82 while True:
83 response = sfn_client.list_executions(**request)
84 execution_arns.extend(
85 execution["executionArn"] for execution in response.get("executions", [])
86 )
87 next_token = response.get("nextToken")
88 if not next_token:
89 break
90 request["nextToken"] = next_token
92 for execution_arn in execution_arns:
93 try:
94 sfn_client.stop_execution(executionArn=execution_arn)
95 except ClientError as exc:
96 # An execution can finish between ListExecutions and StopExecution.
97 # Re-read only a ValidationException and suppress it only when the
98 # execution is now terminal; every other failure blocks teardown.
99 code = exc.response.get("Error", {}).get("Code")
100 if code in {"ExecutionDoesNotExist", "ExecutionNotRunning"}:
101 continue
102 if code == "ValidationException":
103 try:
104 status = sfn_client.describe_execution(executionArn=execution_arn).get("status")
105 except ClientError as describe_exc:
106 describe_code = describe_exc.response.get("Error", {}).get("Code")
107 if describe_code == "ExecutionDoesNotExist":
108 continue
109 raise
110 if status and status != "RUNNING":
111 continue
112 raise
114 if execution_arns:
115 logger.info("Stopped %d install execution(s)", len(execution_arns))
116 return len(execution_arns)
119def drain_install_executions(
120 _event: dict[str, Any],
121 _context: Any = None,
122) -> dict[str, int]:
123 """State-machine task that stops late work and requests another full drain."""
124 stopped = _stop_running_install_executions(
125 _sfn(),
126 os.environ["INSTALL_STATE_MACHINE_ARN"],
127 )
128 return {"StoppedExecutions": stopped}
131def on_event(event: dict[str, Any], _context: Any = None) -> dict[str, Any]:
132 """Start ordered teardown on Delete; no-op on Create and Update."""
133 request_type = event["RequestType"]
134 physical_id = event.get("PhysicalResourceId") or "helm-teardown"
136 if request_type != "Delete":
137 return {"PhysicalResourceId": physical_id}
139 props = event["ResourceProperties"]
140 state_machine_arn = os.environ["TEARDOWN_STATE_MACHINE_ARN"]
141 install_state_machine_arn = os.environ["INSTALL_STATE_MACHINE_ARN"]
142 execution_name = _execution_name(event)
143 sfn_client = _sfn()
144 fence_name = f"/{props['ProjectName']}/addons/{props['Region']}/_teardown"
145 _ssm(str(props["Region"])).put_parameter(
146 Name=fence_name,
147 Value=execution_name,
148 Type="String",
149 Overwrite=True,
150 )
151 _stop_running_install_executions(sfn_client, install_state_machine_arn)
152 execution_input = {
153 "ClusterName": props["ClusterName"],
154 "Region": props["Region"],
155 "RegistryRegion": props["RegistryRegion"],
156 "ProjectName": props["ProjectName"],
157 "EnabledCharts": props.get("EnabledCharts", []),
158 "Charts": props.get("Charts", {}),
159 "KedaOperatorRoleArn": props.get("KedaOperatorRoleArn"),
160 # ListExecutions is eventually consistent and StopExecution cannot
161 # cancel an in-flight Lambda. Always drain the full invocation bound;
162 # the workflow checker loops if late-visible work is stopped.
163 "WaitForInFlightSeconds": _IN_FLIGHT_LAMBDA_DRAIN_SECONDS,
164 }
165 endpoint_group_arn = props.get("EndpointGroupArn")
166 if endpoint_group_arn:
167 execution_input["EndpointGroupArn"] = endpoint_group_arn
169 try:
170 sfn_client.start_execution(
171 stateMachineArn=state_machine_arn,
172 name=execution_name,
173 input=json.dumps(execution_input),
174 )
175 except ClientError as exc:
176 # Provider retries can replay the same Delete request. Reusing the
177 # deterministic name is idempotent when that execution already exists.
178 if exc.response.get("Error", {}).get("Code") != "ExecutionAlreadyExists":
179 raise
181 logger.info("Started ordered Helm teardown execution %s", execution_name)
182 return {"PhysicalResourceId": physical_id}
185def is_complete(event: dict[str, Any], _context: Any = None) -> dict[str, bool]:
186 """Surface teardown terminal status; the workflow owns the drain loop."""
187 if event["RequestType"] != "Delete":
188 return {"IsComplete": True}
190 state_machine_arn = os.environ["TEARDOWN_STATE_MACHINE_ARN"]
191 execution_name = _execution_name(event)
192 execution_arn = _execution_arn(state_machine_arn, execution_name)
193 sfn_client = _sfn()
195 response = sfn_client.describe_execution(executionArn=execution_arn)
196 status = response["status"]
198 if status == "RUNNING":
199 return {"IsComplete": False}
200 if status == "SUCCEEDED":
201 return {"IsComplete": True}
203 detail = response.get("cause") or response.get("error") or "no failure detail"
204 raise RuntimeError(f"Helm teardown execution {status}: {detail}")