Coverage for lambda / ga-registration / handler.py: 100.00%

1"""Gateway ALB registration and endpoint-publication Lambda handler. 

2 

3The handler converges the regional Gateway API ALB after the Gateway manifests 

4have been applied: 

5 

61. Read the exact ``gco-system/gco-gateway`` Gateway status address. 

72. Resolve that address to an active internal ALB, with an exact-tag fallback 

8 while the Gateway status is still being populated. 

93. When a Global Accelerator endpoint group is configured, register only that 

10 ALB, remove stale endpoints, and enforce the HTTPS health-check contract. 

114. Always publish the selected ALB hostname to the SSM endpoint registry. 

12 

13``EndpointGroupArn`` is optional. Deployments without Global Accelerator still 

14publish the Gateway hostname. 

15 

16Entrypoints: 

17 - ``handle_task`` is the final Step Functions convergence task. 

18 - ``lambda_handler`` preserves the legacy raw CloudFormation custom-resource 

19 protocol and dispatches Step Functions events carrying ``Action``. 

20 - ``on_delete_event`` is the CDK provider delete guard. 

21 

22The endpoint registry parameter is always: 

23``/{ProjectName}/alb-hostname-{Region}`` in ``RegistryRegion``. The legacy 

24``GlobalRegion`` property remains accepted for existing deployments. 

25""" 

26 

27import base64 

28import json 

29import logging 

30import os 

31import tempfile 

32import time 

33from contextlib import suppress 

34from typing import Any 

35 

36import boto3 

37import urllib3 

38from botocore.exceptions import ClientError 

39 

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

41# Generated at (UTC): 2026-09-01T14:42:56Z 

42# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc 

43# Flowchart(s) generated from this file: 

44# * ``lambda_handler`` -> ``diagrams/code_diagrams/lambda/ga-registration/handler.lambda_handler.html`` 

45# (PNG: ``diagrams/code_diagrams/lambda/ga-registration/handler.lambda_handler.png``) 

46# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

47# <pyflowchart-code-diagram> END 

48 

49 

50logger = logging.getLogger() 

51logger.setLevel(logging.INFO) 

52 

53MAX_WAIT_SECONDS = 840 

54ALB_POLL_INTERVAL = 5 

55GA_DEPLOYED_WAIT_SECONDS = 720 

56GA_DEPLOYED_POLL_INTERVAL = 15 

57DEFAULT_REGISTRY_REGION = "us-east-2" 

58 

59GATEWAY_NAMESPACE = "gco-system" 

60GATEWAY_NAME = "gco-gateway" 

61GATEWAY_REFERENCE = f"{GATEWAY_NAMESPACE}/{GATEWAY_NAME}" 

62GATEWAY_TAG = "gco.aws/gateway" 

63CLUSTER_TAG = "elbv2.k8s.aws/cluster" 

64GATEWAY_API_PATH = ( 

65 f"/apis/gateway.networking.k8s.io/v1/namespaces/{GATEWAY_NAMESPACE}/gateways/{GATEWAY_NAME}" 

66) 

67 

68 

69def send_response( 

70 event: dict[str, Any], 

71 context: Any, 

72 status: str, 

73 data: dict[str, Any], 

74 physical_id: str, 

75 reason: str | None = None, 

76) -> None: 

77 """Send a response to a raw CloudFormation custom resource.""" 

78 response_body = { 

79 "Status": status, 

80 "Reason": reason or f"See CloudWatch Log Stream: {context.log_stream_name}", 

81 "PhysicalResourceId": physical_id, 

82 "StackId": event["StackId"], 

83 "RequestId": event["RequestId"], 

84 "LogicalResourceId": event["LogicalResourceId"], 

85 "Data": data, 

86 } 

87 logger.info("Sending CFN response: Status=%s, PhysicalResourceId=%s", status, physical_id) 

88 http = urllib3.PoolManager() 

89 try: 

90 http.request( 

91 "PUT", 

92 event["ResponseURL"], 

93 body=json.dumps(response_body).encode("utf-8"), 

94 headers={"Content-Type": "application/json"}, 

95 timeout=10.0, 

96 ) 

97 except Exception as exc: # noqa: BLE001 - callback failure can only be logged 

98 logger.error("Failed to send CloudFormation response: %s", exc) 

99 

100 

101def _remove_temporary_ca_file(ca_path: str | None) -> None: 

102 """Unlink a temporary Kubernetes CA file without masking the real result.""" 

103 if not ca_path: 

104 return 

105 try: 

106 os.unlink(ca_path) 

107 except FileNotFoundError: 

108 return 

109 except OSError as exc: 

110 logger.warning("Failed to remove temporary Kubernetes CA file: %s", exc) 

111 

112 

113def get_k8s_client(cluster_name: str, region: str) -> tuple[str, str, str]: 

114 """Return ``(endpoint, bearer_token, temporary_ca_path)`` for an EKS cluster. 

115 

116 The caller owns the returned CA path and must remove it in a ``finally`` 

117 block. ``mkstemp`` prevents name races; the explicit mode keeps the 

118 certificate readable only by the Lambda process. 

119 """ 

120 eks = boto3.client("eks", region_name=region) 

121 cluster_info = eks.describe_cluster(name=cluster_name)["cluster"] 

122 

123 session = boto3.Session() 

124 sts_client = session.client("sts", region_name=region) 

125 sts_endpoint = str(sts_client.meta.endpoint_url).rstrip("/") 

126 sts_url = f"{sts_endpoint}/?Action=GetCallerIdentity&Version=2011-06-15" 

127 signed_url = sts_client._request_signer.generate_presigned_url( # noqa: SLF001 

128 request_dict={ 

129 "method": "GET", 

130 "url": sts_url, 

131 "body": {}, 

132 "headers": {"x-k8s-aws-id": cluster_name}, 

133 "context": {}, 

134 }, 

135 operation_name="GetCallerIdentity", 

136 expires_in=60, 

137 ) 

138 token = "k8s-aws-v1." + base64.urlsafe_b64encode(signed_url.encode()).decode().rstrip("=") 

139 

140 ca_cert = base64.b64decode(cluster_info["certificateAuthority"]["data"]) 

141 fd, ca_path = tempfile.mkstemp(suffix=".crt") 

142 try: 

143 os.fchmod(fd, 0o600) 

144 with os.fdopen(fd, "wb") as ca_file: 

145 ca_file.write(ca_cert) 

146 except Exception: 

147 with suppress(OSError): 

148 os.close(fd) 

149 _remove_temporary_ca_file(ca_path) 

150 raise 

151 

152 return str(cluster_info["endpoint"]), token, ca_path 

153 

154 

155def _response_json(response: Any) -> dict[str, Any]: 

156 """Decode a Kubernetes JSON response, returning an object mapping.""" 

157 raw_data = response.data 

158 if isinstance(raw_data, bytes): 

159 raw_data = raw_data.decode("utf-8") 

160 document = json.loads(raw_data) 

161 return document if isinstance(document, dict) else {} 

162 

163 

164def find_gateway_address( 

165 http: urllib3.PoolManager, 

166 endpoint: str, 

167 headers: dict[str, str], 

168) -> str | None: 

169 """Read the nonempty hostname from the exact GCO Gateway status.""" 

170 try: 

171 response = http.request( 

172 "GET", 

173 f"{endpoint}{GATEWAY_API_PATH}", 

174 headers=headers, 

175 timeout=10.0, 

176 ) 

177 if response.status == 404: 

178 logger.debug("Gateway %s not found yet", GATEWAY_REFERENCE) 

179 return None 

180 if response.status != 200: 

181 logger.warning("Gateway status request returned HTTP %s", response.status) 

182 return None 

183 

184 addresses = _response_json(response).get("status", {}).get("addresses", []) 

185 for address in addresses: 

186 if not isinstance(address, dict): 

187 continue 

188 address_type = address.get("type", "Hostname") 

189 value = address.get("value") 

190 if address_type == "Hostname" and isinstance(value, str) and value.strip(): 

191 return value.strip() 

192 except Exception as exc: # noqa: BLE001 - polling falls back to exact tags 

193 logger.warning("Error checking Gateway status: %s", exc) 

194 return None 

195 

196 

197def _list_load_balancers(elb_client: Any) -> list[dict[str, Any]]: 

198 """List every load balancer, following ELBv2 marker pagination.""" 

199 load_balancers: list[dict[str, Any]] = [] 

200 marker: str | None = None 

201 seen_markers: set[str] = set() 

202 while True: 

203 kwargs = {"Marker": marker} if marker else {} 

204 response = elb_client.describe_load_balancers(**kwargs) 

205 load_balancers.extend(response.get("LoadBalancers", [])) 

206 next_marker = response.get("NextMarker") 

207 if not isinstance(next_marker, str) or not next_marker: 

208 return load_balancers 

209 if next_marker in seen_markers: 

210 raise RuntimeError(f"ELB pagination repeated marker {next_marker!r}") 

211 seen_markers.add(next_marker) 

212 marker = next_marker 

213 

214 

215def find_alb_by_gateway_hostname( 

216 elb_client: Any, hostname: str, cluster_name: str 

217) -> tuple[str | None, str | None, str | None]: 

218 """Resolve a Gateway hostname only to its exactly owned internal ALB.""" 

219 try: 

220 candidates = [ 

221 load_balancer 

222 for load_balancer in _list_load_balancers(elb_client) 

223 if load_balancer.get("Type") == "application" 

224 and load_balancer.get("Scheme") == "internal" 

225 and load_balancer.get("DNSName") == hostname 

226 ] 

227 if not candidates: 

228 return None, None, None 

229 

230 arns = [str(load_balancer["LoadBalancerArn"]) for load_balancer in candidates] 

231 tags_by_arn = _describe_tags(elb_client, arns) 

232 for load_balancer in candidates: 

233 arn = str(load_balancer["LoadBalancerArn"]) 

234 tags = tags_by_arn.get(arn, {}) 

235 if not ( 

236 tags.get(GATEWAY_TAG) == GATEWAY_REFERENCE and tags.get(CLUSTER_TAG) == cluster_name 

237 ): 

238 logger.warning("Rejecting hostname-matched ALB without exact ownership: %s", arn) 

239 continue 

240 state = str(load_balancer.get("State", {}).get("Code", "unknown")) 

241 logger.info( 

242 "Found exactly owned Gateway ALB by hostname: %s (state: %s)", 

243 load_balancer.get("LoadBalancerName", "<unknown>"), 

244 state, 

245 ) 

246 return str(load_balancer["DNSName"]), arn, state 

247 except Exception as exc: # noqa: BLE001 - discovery polling retries 

248 logger.warning("Error finding Gateway ALB by hostname: %s", exc) 

249 return None, None, None 

250 

251 

252def _describe_tags(elb_client: Any, load_balancer_arns: list[str]) -> dict[str, dict[str, str]]: 

253 """Return ELB tags by ARN, respecting the API's 20-resource limit.""" 

254 tags_by_arn: dict[str, dict[str, str]] = {} 

255 for index in range(0, len(load_balancer_arns), 20): 

256 response = elb_client.describe_tags(ResourceArns=load_balancer_arns[index : index + 20]) 

257 for description in response.get("TagDescriptions", []): 

258 arn = description.get("ResourceArn") 

259 if not isinstance(arn, str): 

260 continue 

261 tags_by_arn[arn] = { 

262 str(tag["Key"]): str(tag["Value"]) 

263 for tag in description.get("Tags", []) 

264 if "Key" in tag and "Value" in tag 

265 } 

266 return tags_by_arn 

267 

268 

269def find_platform_alb_by_tags( 

270 elb_client: Any, cluster_name: str 

271) -> tuple[str | None, str | None, str | None]: 

272 """Find the Gateway ALB only when both exact ownership tags match. 

273 

274 This fallback is used only while the exact Gateway status has no address. 

275 Cluster-only matches, alternative cluster tags, NLBs, and internet-facing 

276 ALBs are deliberately rejected. 

277 """ 

278 try: 

279 load_balancers = [ 

280 load_balancer 

281 for load_balancer in _list_load_balancers(elb_client) 

282 if load_balancer.get("Type") == "application" 

283 and load_balancer.get("Scheme") == "internal" 

284 ] 

285 if not load_balancers: 

286 return None, None, None 

287 

288 arns = [str(load_balancer["LoadBalancerArn"]) for load_balancer in load_balancers] 

289 tags_by_arn = _describe_tags(elb_client, arns) 

290 for load_balancer in load_balancers: 

291 arn = str(load_balancer["LoadBalancerArn"]) 

292 tags = tags_by_arn.get(arn, {}) 

293 if not ( 

294 tags.get(GATEWAY_TAG) == GATEWAY_REFERENCE and tags.get(CLUSTER_TAG) == cluster_name 

295 ): 

296 continue 

297 state = str(load_balancer.get("State", {}).get("Code", "unknown")) 

298 logger.info( 

299 "Found Gateway ALB by exact tags: %s (state: %s)", 

300 load_balancer.get("LoadBalancerName", "<unknown>"), 

301 state, 

302 ) 

303 return str(load_balancer["DNSName"]), arn, state 

304 except Exception as exc: # noqa: BLE001 - discovery polling retries 

305 logger.warning("Error finding Gateway ALB by tags: %s", exc) 

306 return None, None, None 

307 

308 

309def find_active_alb( 

310 elb_client: Any, 

311 http: urllib3.PoolManager, 

312 k8s_endpoint: str, 

313 k8s_headers: dict[str, str], 

314 cluster_name: str, 

315) -> tuple[str | None, str | None]: 

316 """Find the active ALB owned by the exact GCO Gateway. 

317 

318 A nonempty Gateway status address is authoritative. Tag fallback is allowed 

319 only when that address is absent; it never overrides a provisioning or 

320 otherwise unresolved ALB named by Gateway status. 

321 """ 

322 hostname = find_gateway_address(http, k8s_endpoint, k8s_headers) 

323 if hostname: 

324 dns_name, arn, state = find_alb_by_gateway_hostname(elb_client, hostname, cluster_name) 

325 if arn and state == "active": 

326 logger.info("Found active ALB from Gateway status: %s", hostname) 

327 return dns_name, arn 

328 if arn: 

329 logger.info("Gateway ALB state is %r; waiting for 'active'", state) 

330 return None, None 

331 

332 dns_name, arn, state = find_platform_alb_by_tags(elb_client, cluster_name) 

333 if arn and state == "active": 

334 return dns_name, arn 

335 if arn: 

336 logger.info("Gateway ALB found by tags but state is %r; waiting for 'active'", state) 

337 return None, None 

338 

339 

340def check_existing_ga_endpoint(ga_client: Any, endpoint_group_arn: str, alb_arn: str) -> bool: 

341 """Return whether the exact ALB is already registered with GA.""" 

342 try: 

343 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

344 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

345 if any(endpoint.get("EndpointId") == alb_arn for endpoint in endpoints): 

346 logger.info("ALB %s is already registered with GA", alb_arn) 

347 return True 

348 except Exception as exc: # noqa: BLE001 - add_endpoints remains authoritative 

349 logger.warning("Error checking existing GA endpoints: %s", exc) 

350 return False 

351 

352 

353def scrub_stale_ga_endpoints(ga_client: Any, endpoint_group_arn: str, correct_alb_arn: str) -> None: 

354 """Remove every GA endpoint other than the exact Gateway ALB.""" 

355 try: 

356 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

357 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

358 for endpoint in endpoints: 

359 endpoint_id = endpoint.get("EndpointId", "") 

360 if not endpoint_id or endpoint_id == correct_alb_arn: 

361 continue 

362 logger.warning( 

363 "Removing stale GA endpoint %s; exact Gateway ALB is %s", 

364 endpoint_id, 

365 correct_alb_arn, 

366 ) 

367 try: 

368 ga_client.remove_endpoints( 

369 EndpointGroupArn=endpoint_group_arn, 

370 EndpointIdentifiers=[{"EndpointId": endpoint_id}], 

371 ) 

372 except ClientError as exc: 

373 error_code = exc.response.get("Error", {}).get("Code", "") 

374 if error_code == "EndpointNotFoundException": 

375 logger.info("GA endpoint %s was already absent", endpoint_id) 

376 else: 

377 logger.error("Failed to remove stale GA endpoint %s: %s", endpoint_id, exc) 

378 raise 

379 except Exception as exc: 

380 logger.error("Error scrubbing stale GA endpoints: %s", exc) 

381 raise 

382 

383 

384def register_alb_with_ga(ga_client: Any, endpoint_group_arn: str, alb_arn: str) -> None: 

385 """Register the exact Gateway ALB with GA, idempotently.""" 

386 if check_existing_ga_endpoint(ga_client, endpoint_group_arn, alb_arn): 

387 return 

388 try: 

389 ga_client.add_endpoints( 

390 EndpointGroupArn=endpoint_group_arn, 

391 EndpointConfigurations=[ 

392 { 

393 "EndpointId": alb_arn, 

394 "Weight": 100, 

395 "ClientIPPreservationEnabled": True, 

396 } 

397 ], 

398 ) 

399 logger.info("Registered Gateway ALB %s with Global Accelerator", alb_arn) 

400 except ClientError as exc: 

401 if exc.response.get("Error", {}).get("Code") == "EndpointAlreadyExists": 

402 logger.info("Gateway ALB was already registered with Global Accelerator") 

403 return 

404 raise 

405 

406 

407def ensure_https_health_check( 

408 ga_client: Any, 

409 endpoint_group_arn: str, 

410 health_check_path: str = "/api/v1/health", 

411 expected_alb_arn: str | None = None, 

412 health_check_interval: int = 30, 

413 health_check_threshold: int = 3, 

414) -> None: 

415 """Enforce the endpoint group's configured HTTPS/443 health-check contract. 

416 

417 Interval and threshold come from the cdk.json ``global_accelerator`` block 

418 (baked into the convergence task payload at synth time); the defaults 

419 match the values this handler historically hardcoded, so legacy payloads 

420 without the keys keep converging identically. Comparing them here also 

421 means a non-default configured interval is no longer silently reset to 30 

422 whenever the path or protocol drifts. 

423 

424 When ``expected_alb_arn`` is supplied, only that endpoint is preserved in an 

425 update, preventing an eventually consistent stale endpoint description from 

426 being reintroduced after the scrub. 

427 """ 

428 try: 

429 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

430 group = endpoint_group.get("EndpointGroup", {}) 

431 current_protocol = group.get("HealthCheckProtocol", "TCP") 

432 current_port = int(group.get("HealthCheckPort", 0)) 

433 current_path = group.get("HealthCheckPath", "") 

434 current_interval = int(group.get("HealthCheckIntervalSeconds", 0)) 

435 current_threshold = int(group.get("ThresholdCount", 0)) 

436 if ( 

437 current_protocol == "HTTPS" 

438 and current_port == 443 

439 and current_path == health_check_path 

440 and current_interval == health_check_interval 

441 and current_threshold == health_check_threshold 

442 ): 

443 return 

444 

445 existing_endpoints = [ 

446 { 

447 "EndpointId": endpoint["EndpointId"], 

448 "Weight": endpoint.get("Weight", 100), 

449 "ClientIPPreservationEnabled": endpoint.get("ClientIPPreservationEnabled", True), 

450 } 

451 for endpoint in group.get("EndpointDescriptions", []) 

452 if endpoint.get("EndpointId") 

453 and (expected_alb_arn is None or endpoint.get("EndpointId") == expected_alb_arn) 

454 ] 

455 ga_client.update_endpoint_group( 

456 EndpointGroupArn=endpoint_group_arn, 

457 HealthCheckPort=443, 

458 HealthCheckProtocol="HTTPS", 

459 HealthCheckPath=health_check_path, 

460 HealthCheckIntervalSeconds=health_check_interval, 

461 ThresholdCount=health_check_threshold, 

462 EndpointConfigurations=existing_endpoints, 

463 ) 

464 logger.info( 

465 "Global Accelerator health check set to HTTPS/443 %s (interval %ss, threshold %s)", 

466 health_check_path, 

467 health_check_interval, 

468 health_check_threshold, 

469 ) 

470 except ClientError as exc: 

471 logger.error("Failed to enforce GA health-check configuration: %s", exc) 

472 raise 

473 

474 

475def _get_registry_region(properties: dict[str, Any], default: str | None = None) -> str | None: 

476 """Return ``RegistryRegion``, accepting ``GlobalRegion`` for compatibility.""" 

477 value = properties.get("RegistryRegion") or properties.get("GlobalRegion") or default 

478 return str(value) if value else None 

479 

480 

481def store_alb_hostname_in_ssm( 

482 region: str, alb_hostname: str, registry_region: str, project_name: str 

483) -> None: 

484 """Publish the Gateway ALB hostname to the regional endpoint registry.""" 

485 ssm_client = boto3.client("ssm", region_name=registry_region) 

486 parameter_name = f"/{project_name}/alb-hostname-{region}" 

487 ssm_client.put_parameter( 

488 Name=parameter_name, 

489 Value=alb_hostname, 

490 Type="String", 

491 Overwrite=True, 

492 Description=f"ALB hostname for {region} regional cluster", 

493 ) 

494 logger.info("Stored Gateway ALB hostname in SSM: %s = %s", parameter_name, alb_hostname) 

495 

496 

497def delete_alb_hostname_from_ssm( 

498 region: str, 

499 registry_region: str, 

500 project_name: str, 

501 *, 

502 strict: bool = False, 

503) -> None: 

504 """Remove this region's endpoint-registry parameter during cleanup.""" 

505 ssm_client = boto3.client("ssm", region_name=registry_region) 

506 parameter_name = f"/{project_name}/alb-hostname-{region}" 

507 try: 

508 ssm_client.delete_parameter(Name=parameter_name) 

509 logger.info("Deleted Gateway ALB hostname from SSM: %s", parameter_name) 

510 except ClientError as exc: 

511 if exc.response.get("Error", {}).get("Code") == "ParameterNotFound": 

512 logger.info("SSM parameter %s was already absent", parameter_name) 

513 elif strict: 

514 raise 

515 else: 

516 logger.warning("Failed to delete Gateway ALB hostname from SSM: %s", exc) 

517 

518 

519def remove_ga_endpoints( 

520 ga_client: Any, 

521 endpoint_group_arn: str, 

522 *, 

523 strict: bool = False, 

524) -> None: 

525 """Remove every endpoint from one regional GA group.""" 

526 try: 

527 endpoint_group = ga_client.describe_endpoint_group(EndpointGroupArn=endpoint_group_arn) 

528 endpoints = endpoint_group.get("EndpointGroup", {}).get("EndpointDescriptions", []) 

529 for endpoint in endpoints: 

530 endpoint_id = endpoint.get("EndpointId") 

531 if not endpoint_id: 

532 continue 

533 try: 

534 ga_client.remove_endpoints( 

535 EndpointGroupArn=endpoint_group_arn, 

536 EndpointIdentifiers=[{"EndpointId": endpoint_id}], 

537 ) 

538 except ClientError as exc: 

539 if exc.response.get("Error", {}).get("Code") == "EndpointNotFoundException": 

540 continue 

541 if strict: 

542 raise 

543 logger.warning("Failed to remove GA endpoint %s: %s", endpoint_id, exc) 

544 except ClientError as exc: 

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

546 if code in {"EndpointGroupNotFoundException", "AcceleratorNotFoundException"}: 

547 logger.info("Global Accelerator endpoint group was already absent") 

548 return 

549 if strict: 

550 raise 

551 logger.warning("Failed to clean up GA endpoints: %s", exc) 

552 except Exception as exc: # noqa: BLE001 - delete guards remain best effort 

553 if strict: 

554 raise 

555 logger.warning("Failed to clean up GA endpoints: %s", exc) 

556 

557 

558def _accelerator_arn_from_endpoint_group(endpoint_group_arn: str) -> str: 

559 """Derive an accelerator ARN from one of its endpoint-group ARNs.""" 

560 return endpoint_group_arn.split("/listener/")[0] 

561 

562 

563def wait_for_accelerator_deployed( 

564 ga_client: Any, 

565 endpoint_group_arn: str, 

566 timeout_seconds: int = GA_DEPLOYED_WAIT_SECONDS, 

567 *, 

568 strict: bool = False, 

569) -> bool: 

570 """Wait until GA finishes redeploying and releases its managed ENIs. 

571 

572 In strict mode a describe failure raises immediately instead of being 

573 reported as a timeout; a permissions gap must surface as itself. 

574 """ 

575 accelerator_arn = _accelerator_arn_from_endpoint_group(endpoint_group_arn) 

576 start_time = time.time() 

577 while time.time() - start_time < timeout_seconds: 

578 try: 

579 accelerator = ga_client.describe_accelerator(AcceleratorArn=accelerator_arn) 

580 status = accelerator.get("Accelerator", {}).get("Status", "") 

581 except ClientError as exc: 

582 if exc.response.get("Error", {}).get("Code") == "AcceleratorNotFoundException": 

583 return True 

584 if strict: 

585 raise 

586 logger.warning("Failed to describe accelerator status: %s", exc) 

587 return False 

588 if status == "DEPLOYED": 

589 return True 

590 logger.info("Accelerator status=%r; waiting for DEPLOYED", status) 

591 # nosemgrep: arbitrary-sleep - intentional GA redeployment polling 

592 time.sleep(GA_DEPLOYED_POLL_INTERVAL) 

593 logger.warning("Timed out waiting for Global Accelerator to reach DEPLOYED") 

594 return False 

595 

596 

597def deregister_alb_from_ga( 

598 ga_client: Any, 

599 endpoint_group_arn: str, 

600 *, 

601 strict: bool = False, 

602) -> None: 

603 """Remove regional GA endpoints, then wait for managed ENI release.""" 

604 if strict: 

605 remove_ga_endpoints(ga_client, endpoint_group_arn, strict=True) 

606 else: 

607 remove_ga_endpoints(ga_client, endpoint_group_arn) 

608 deployed = wait_for_accelerator_deployed(ga_client, endpoint_group_arn, strict=strict) 

609 if strict and not deployed: 

610 raise TimeoutError("Global Accelerator did not reach DEPLOYED after endpoint removal") 

611 

612 

613def _optional_endpoint_group_arn(properties: dict[str, Any]) -> str | None: 

614 """Normalize an optional endpoint-group property.""" 

615 value = properties.get("EndpointGroupArn") 

616 if value is None: 

617 return None 

618 normalized = str(value).strip() 

619 return normalized or None 

620 

621 

622def _health_check_contract(properties: dict[str, Any]) -> dict[str, Any]: 

623 """Extract the optional ``GaHealthCheck*`` payload keys with legacy defaults. 

624 

625 The regional stack bakes the configured contract into the convergence 

626 payload at synth time. Payloads persisted before these keys existed (and 

627 raw CloudFormation properties from older deployments) fall back to the 

628 values this handler historically hardcoded. 

629 """ 

630 return { 

631 "health_check_path": str(properties.get("GaHealthCheckPath", "/api/v1/health")), 

632 "health_check_interval": int(properties.get("GaHealthCheckInterval", 30)), 

633 "health_check_threshold": int(properties.get("GaHealthCheckThreshold", 3)), 

634 } 

635 

636 

637def register_ga_endpoint( 

638 cluster_name: str, 

639 region: str, 

640 endpoint_group_arn: str | None = None, 

641 registry_region: str = DEFAULT_REGISTRY_REGION, 

642 project_name: str = "gco", 

643 *, 

644 health_check_path: str = "/api/v1/health", 

645 health_check_interval: int = 30, 

646 health_check_threshold: int = 3, 

647) -> dict[str, str]: 

648 """Converge the Gateway ALB, optional GA endpoint, registry, and migration.""" 

649 ca_path: str | None = None 

650 try: 

651 k8s_endpoint, token, ca_path = get_k8s_client(cluster_name, region) 

652 http = urllib3.PoolManager(cert_reqs="CERT_REQUIRED", ca_certs=ca_path) 

653 k8s_headers = {"Authorization": f"Bearer {token}"} 

654 elb_client = boto3.client("elbv2", region_name=region) 

655 

656 logger.info("Waiting for active %s ALB", GATEWAY_REFERENCE) 

657 start_time = time.time() 

658 last_log_time = start_time 

659 alb_hostname: str | None = None 

660 alb_arn: str | None = None 

661 while time.time() - start_time < MAX_WAIT_SECONDS: 

662 alb_hostname, alb_arn = find_active_alb( 

663 elb_client, 

664 http, 

665 k8s_endpoint, 

666 k8s_headers, 

667 cluster_name, 

668 ) 

669 if alb_arn: 

670 break 

671 if time.time() - last_log_time >= 30: 

672 elapsed = int(time.time() - start_time) 

673 logger.info("Still waiting for Gateway ALB (%ss elapsed)", elapsed) 

674 last_log_time = time.time() 

675 # nosemgrep: arbitrary-sleep - intentional Gateway/ALB polling 

676 time.sleep(ALB_POLL_INTERVAL) 

677 

678 if not alb_arn or not alb_hostname: 

679 elapsed = int(time.time() - start_time) 

680 raise TimeoutError( 

681 f"Timed out waiting for active {GATEWAY_REFERENCE} ALB after {elapsed} seconds" 

682 ) 

683 

684 normalized_endpoint_group = ( 

685 str(endpoint_group_arn).strip() if endpoint_group_arn is not None else "" 

686 ) 

687 if normalized_endpoint_group: 

688 ga_client = boto3.client("globalaccelerator", region_name="us-west-2") 

689 register_alb_with_ga(ga_client, normalized_endpoint_group, alb_arn) 

690 scrub_stale_ga_endpoints(ga_client, normalized_endpoint_group, alb_arn) 

691 ensure_https_health_check( 

692 ga_client, 

693 normalized_endpoint_group, 

694 health_check_path=health_check_path, 

695 expected_alb_arn=alb_arn, 

696 health_check_interval=health_check_interval, 

697 health_check_threshold=health_check_threshold, 

698 ) 

699 # AddEndpoints/RemoveEndpoints/UpdateEndpointGroup only submit a 

700 # configuration change; the accelerator serves it from its edge 

701 # locations only after returning to DEPLOYED. Returning success 

702 # earlier reports the deployment as complete while brand-new 

703 # connections to the global endpoint still black-hole for several 

704 # minutes, which a live release run proved by timing out on the 

705 # first health probe after deploy. Wait strictly, within whatever 

706 # remains of this handler's wall-clock budget. 

707 remaining_budget = min( 

708 GA_DEPLOYED_WAIT_SECONDS, 

709 int(MAX_WAIT_SECONDS - (time.time() - start_time)), 

710 ) 

711 if remaining_budget <= 0 or not wait_for_accelerator_deployed( 

712 ga_client, 

713 normalized_endpoint_group, 

714 timeout_seconds=remaining_budget, 

715 strict=True, 

716 ): 

717 raise TimeoutError( 

718 "Global Accelerator did not reach DEPLOYED after endpoint registration" 

719 ) 

720 else: 

721 logger.info("EndpointGroupArn is not configured; skipping Global Accelerator") 

722 

723 # Publication is mandatory even when Global Accelerator is disabled. 

724 store_alb_hostname_in_ssm( 

725 region, 

726 alb_hostname, 

727 registry_region, 

728 project_name, 

729 ) 

730 return {"AlbArn": alb_arn, "AlbHostname": alb_hostname} 

731 finally: 

732 _remove_temporary_ca_file(ca_path) 

733 

734 

735def cleanup_gateway_endpoint( 

736 *, 

737 region: str, 

738 endpoint_group_arn: str | None, 

739 registry_region: str, 

740 project_name: str, 

741) -> dict[str, bool]: 

742 """Strictly fence endpoint publication before Gateway deletion.""" 

743 delete_alb_hostname_from_ssm( 

744 region, 

745 registry_region, 

746 project_name, 

747 strict=True, 

748 ) 

749 if endpoint_group_arn: 

750 ga_client = boto3.client("globalaccelerator", region_name="us-west-2") 

751 deregister_alb_from_ga(ga_client, endpoint_group_arn, strict=True) 

752 return { 

753 "RegistryParameterDeleted": True, 

754 "GlobalAcceleratorDeregistered": bool(endpoint_group_arn), 

755 } 

756 

757 

758def handle_task(event: dict[str, Any]) -> dict[str, Any]: 

759 """Handle Step Functions convergence and teardown task invocations.""" 

760 if event.get("Action") == "cleanup_gateway_endpoint": 

761 return cleanup_gateway_endpoint( 

762 region=str(event["Region"]), 

763 endpoint_group_arn=_optional_endpoint_group_arn(event), 

764 registry_region=_get_registry_region(event, DEFAULT_REGISTRY_REGION) 

765 or DEFAULT_REGISTRY_REGION, 

766 project_name=str(event.get("ProjectName", "gco")), 

767 ) 

768 

769 return register_ga_endpoint( 

770 cluster_name=str(event["ClusterName"]), 

771 region=str(event["Region"]), 

772 endpoint_group_arn=_optional_endpoint_group_arn(event), 

773 registry_region=_get_registry_region(event, DEFAULT_REGISTRY_REGION) 

774 or DEFAULT_REGISTRY_REGION, 

775 project_name=str(event.get("ProjectName", "gco")), 

776 **_health_check_contract(event), 

777 ) 

778 

779 

780def handle_create_update( 

781 event: dict[str, Any], context: Any, props: dict[str, Any], physical_id: str 

782) -> None: 

783 """Handle Create/Update through the raw CloudFormation protocol.""" 

784 data = register_ga_endpoint( 

785 cluster_name=str(props["ClusterName"]), 

786 region=str(props["Region"]), 

787 endpoint_group_arn=_optional_endpoint_group_arn(props), 

788 registry_region=_get_registry_region(props, DEFAULT_REGISTRY_REGION) 

789 or DEFAULT_REGISTRY_REGION, 

790 project_name=str(props.get("ProjectName", "gco")), 

791 **_health_check_contract(props), 

792 ) 

793 send_response(event, context, "SUCCESS", data, physical_id) 

794 

795 

796def handle_delete( 

797 event: dict[str, Any], context: Any, props: dict[str, Any], physical_id: str 

798) -> None: 

799 """Always remove SSM and conditionally deregister/wait for GA.""" 

800 endpoint_group_arn = _optional_endpoint_group_arn(props) 

801 if endpoint_group_arn: