Coverage for lambda / helm-installer / teardown_provider.py: 100.00%

1"""Delete-only custom-resource provider for ordered Helm teardown. 

2 

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. 

8 

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""" 

14 

15from __future__ import annotations 

16 

17import hashlib 

18import json 

19import logging 

20import os 

21from typing import Any 

22 

23import boto3 

24from botocore.exceptions import ClientError 

25 

26# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

27# Generated at (UTC): 2026-09-27T04:10:54Z 

28# Generated from Git commit: 590c275a9cd8dc0e0ba9f8ea74acefb63754df46 

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 

36 

37 

38logger = logging.getLogger() 

39logger.setLevel(logging.INFO) 

40 

41_IN_FLIGHT_LAMBDA_DRAIN_SECONDS = 16 * 60 

42 

43 

44def _sfn() -> Any: 

45 return boto3.client("stepfunctions") 

46 

47 

48def _ssm(region: str) -> Any: 

49 return boto3.client("ssm", region_name=region) 

50 

51 

52def _chart_overrides(props: dict[str, Any]) -> dict[str, Any]: 

53 """Return the ``Charts`` overrides with the JSON types the stack gave them. 

54 

55 CloudFormation delivers custom-resource properties with every number and 

56 boolean as a string, so the regional stack sends ``Charts`` as one JSON 

57 string and it is decoded here, the same way the Helm orchestrator decodes 

58 it. An object (a stack synthesized before that change) is taken as it is. 

59 """ 

60 charts = props.get("Charts", {}) 

61 if isinstance(charts, str): 

62 try: 

63 charts = json.loads(charts) 

64 except json.JSONDecodeError as exc: 

65 raise ValueError(f"Charts is not valid JSON: {exc}") from exc 

66 if not isinstance(charts, dict): 

67 raise ValueError("Charts must be a JSON object of per-chart overrides") 

68 return charts 

69 

70 

71def _execution_name(event: dict[str, Any]) -> str: 

72 """Return a retry-stable, Step-Functions-safe execution name.""" 

73 identity = "|".join( 

74 str(event.get(key, "")) for key in ("StackId", "RequestId", "LogicalResourceId") 

75 ) 

76 return f"helm-delete-{hashlib.sha256(identity.encode()).hexdigest()[:32]}" 

77 

78 

79def _execution_arn(state_machine_arn: str, execution_name: str) -> str: 

80 """Build the execution ARN corresponding to a state-machine ARN.""" 

81 prefix, separator, state_machine_name = state_machine_arn.partition(":stateMachine:") 

82 if not separator or not state_machine_name: 

83 raise ValueError(f"Invalid Step Functions state machine ARN: {state_machine_arn}") 

84 return f"{prefix}:execution:{state_machine_name}:{execution_name}" 

85 

86 

87def _stop_running_install_executions(sfn_client: Any, state_machine_arn: str) -> int: 

88 """Stop every visible convergence execution and return the count stopped. 

89 

90 ``ListExecutions`` is eventually consistent. The delete provider invokes 

91 this once before the teardown state machine starts its unconditional drain; 

92 the state machine then invokes ``drain_install_executions`` after each full 

93 drain interval and loops whenever a late execution is discovered. 

94 """ 

95 execution_arns: list[str] = [] 

96 request: dict[str, Any] = { 

97 "stateMachineArn": state_machine_arn, 

98 "statusFilter": "RUNNING", 

99 "maxResults": 100, 

100 } 

101 while True: 

102 response = sfn_client.list_executions(**request) 

103 execution_arns.extend( 

104 execution["executionArn"] for execution in response.get("executions", []) 

105 ) 

106 next_token = response.get("nextToken") 

107 if not next_token: 

108 break 

109 request["nextToken"] = next_token 

110 

111 for execution_arn in execution_arns: 

112 try: 

113 sfn_client.stop_execution(executionArn=execution_arn) 

114 except ClientError as exc: 

115 # An execution can finish between ListExecutions and StopExecution. 

116 # Re-read only a ValidationException and suppress it only when the 

117 # execution is now terminal; every other failure blocks teardown. 

118 code = exc.response.get("Error", {}).get("Code") 

119 if code in {"ExecutionDoesNotExist", "ExecutionNotRunning"}: 

120 continue 

121 if code == "ValidationException": 

122 try: 

123 status = sfn_client.describe_execution(executionArn=execution_arn).get("status") 

124 except ClientError as describe_exc: 

125 describe_code = describe_exc.response.get("Error", {}).get("Code") 

126 if describe_code == "ExecutionDoesNotExist": 

127 continue 

128 raise 

129 if status and status != "RUNNING": 

130 continue 

131 raise 

132 

133 if execution_arns: 

134 logger.info("Stopped %d install execution(s)", len(execution_arns)) 

135 return len(execution_arns) 

136 

137 

138def drain_install_executions( 

139 _event: dict[str, Any], 

140 _context: Any = None, 

141) -> dict[str, int]: 

142 """State-machine task that stops late work and requests another full drain.""" 

143 stopped = _stop_running_install_executions( 

144 _sfn(), 

145 os.environ["INSTALL_STATE_MACHINE_ARN"], 

146 ) 

147 return {"StoppedExecutions": stopped} 

148 

149 

150def on_event(event: dict[str, Any], _context: Any = None) -> dict[str, Any]: 

151 """Start ordered teardown on Delete; no-op on Create and Update.""" 

152 request_type = event["RequestType"] 

153 physical_id = event.get("PhysicalResourceId") or "helm-teardown" 

154 

155 if request_type != "Delete": 

156 return {"PhysicalResourceId": physical_id} 

157 

158 props = event["ResourceProperties"] 

159 # Decoded before anything is written, so a malformed payload fails the 

160 # Delete without leaving a fence or stopped executions behind. 

161 chart_overrides = _chart_overrides(props) 

162 state_machine_arn = os.environ["TEARDOWN_STATE_MACHINE_ARN"] 

163 install_state_machine_arn = os.environ["INSTALL_STATE_MACHINE_ARN"] 

164 execution_name = _execution_name(event) 

165 sfn_client = _sfn() 

166 fence_name = f"/{props['ProjectName']}/addons/{props['Region']}/_teardown" 

167 _ssm(str(props["Region"])).put_parameter( 

168 Name=fence_name, 

169 Value=execution_name, 

170 Type="String", 

171 Overwrite=True, 

172 ) 

173 _stop_running_install_executions(sfn_client, install_state_machine_arn) 

174 execution_input = { 

175 "ClusterName": props["ClusterName"], 

176 "Region": props["Region"], 

177 "RegistryRegion": props["RegistryRegion"], 

178 "ProjectName": props["ProjectName"], 

179 "EnabledCharts": props.get("EnabledCharts", []), 

180 "Charts": chart_overrides, 

181 "KedaOperatorRoleArn": props.get("KedaOperatorRoleArn"), 

182 # ListExecutions is eventually consistent and StopExecution cannot 

183 # cancel an in-flight Lambda. Always drain the full invocation bound; 

184 # the workflow checker loops if late-visible work is stopped. 

185 "WaitForInFlightSeconds": _IN_FLIGHT_LAMBDA_DRAIN_SECONDS, 

186 } 

187 endpoint_group_arn = props.get("EndpointGroupArn") 

188 if endpoint_group_arn: 

189 execution_input["EndpointGroupArn"] = endpoint_group_arn 

190 

191 try: 

192 sfn_client.start_execution( 

193 stateMachineArn=state_machine_arn, 

194 name=execution_name, 

195 input=json.dumps(execution_input), 

196 ) 

197 except ClientError as exc: 

198 # Provider retries can replay the same Delete request. Reusing the 

199 # deterministic name is idempotent when that execution already exists. 

200 if exc.response.get("Error", {}).get("Code") != "ExecutionAlreadyExists": 

201 raise 

202 

203 logger.info("Started ordered Helm teardown execution %s", execution_name) 

204 return {"PhysicalResourceId": physical_id} 

205 

206 

207def is_complete(event: dict[str, Any], _context: Any = None) -> dict[str, bool]: 

208 """Surface teardown terminal status; the workflow owns the drain loop.""" 

209 if event["RequestType"] != "Delete": 

210 return {"IsComplete": True} 

211 

212 state_machine_arn = os.environ["TEARDOWN_STATE_MACHINE_ARN"] 

213 execution_name = _execution_name(event) 

214 execution_arn = _execution_arn(state_machine_arn, execution_name) 

215 sfn_client = _sfn() 

216 

217 response = sfn_client.describe_execution(executionArn=execution_arn) 

218 status = response["status"] 

219 

220 if status == "RUNNING": 

221 return {"IsComplete": False} 

222 if status == "SUCCEEDED": 

223 return {"IsComplete": True} 

224 

225 detail = response.get("cause") or response.get("error") or "no failure detail" 

226 raise RuntimeError(f"Helm teardown execution {status}: {detail}")