Coverage for cli/stacks.py: 81.50%
2539 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +0000
1"""
2Stack management for GCO CLI.
4Provides commands for deploying, updating, and managing CDK stacks.
5This is the largest CLI module (~1600 lines) because it orchestrates the
6full deployment lifecycle including container runtime detection, CDK
7bootstrapping, Lambda source synchronization, and parallel regional deploys.
9This module handles:
10 - Container runtime detection (Docker, Finch, Podman) with automatic fallback
11 - CDK bootstrap across all target regions (idempotent)
12 - Lambda source synchronization (copies handler code + dependencies before synth)
13 - CDK stack deployment with proper dependency ordering:
14 1. Global stack (partition-wide state, plus Global Accelerator in `aws`)
15 2. API Gateway stack (auth secret, Lambda proxy)
16 3. Regional stacks in parallel (EKS, VPC, ALB per region)
17 4. Monitoring stack (CloudWatch dashboards, alarms)
18 - Parallel deployment of regional stacks via ThreadPoolExecutor
19 - Stack destruction in reverse dependency order
20 - FSx for Lustre enable/disable toggle
21 - kubectl access configuration (EKS access entries + kubeconfig)
23Key Design Decisions:
24 - Regional stacks deploy in parallel for speed; global/API/monitoring are sequential
25 - Lambda build directories are synced before every deploy to avoid stale code
26 - Container runtime is auto-detected; CDK_DOCKER env var overrides
27 - All destructive operations require -y/--yes confirmation
28 - Stack status is read from CloudFormation, not cached locally
30Environment Variables:
31 CDK_DOCKER: Override container runtime (default: auto-detect Docker/Finch/Podman)
32 AWS_REGION: Default region for single-region operations
33"""
35from __future__ import annotations
37import errno
38import hashlib
39import importlib.util
40import json
41import logging
42import math
43import os
44import shutil
45import signal
46import site
47import stat
48import subprocess
49import sys
50import tempfile
51import time
52import uuid
53from collections.abc import Callable, Collection, Iterator, Mapping
54from concurrent.futures import ThreadPoolExecutor, as_completed
55from contextlib import ExitStack, contextmanager
56from dataclasses import dataclass, field
57from datetime import UTC, datetime
58from functools import lru_cache
59from pathlib import Path
60from threading import Event, Lock, Thread, local
61from typing import TYPE_CHECKING, Any, BinaryIO, TypedDict
63from botocore.exceptions import ClientError
65from gco.stacks.constants import (
66 known_cloudformation_regions,
67 validated_deployment_partition,
68 validated_regional_deployment_regions,
69)
71# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
72# Generated at (UTC): 2026-07-18T01:03:40Z
73# Flowchart(s) generated from this file:
74# * ``StackManager.deploy_orchestrated`` -> ``diagrams/code_diagrams/cli/stacks.StackManager_deploy_orchestrated.html``
75# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager_deploy_orchestrated.png``)
76# * ``StackManager.destroy_orchestrated`` -> ``diagrams/code_diagrams/cli/stacks.StackManager_destroy_orchestrated.html``
77# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager_destroy_orchestrated.png``)
78# * ``StackManager._mirror_images_if_enabled`` -> ``diagrams/code_diagrams/cli/stacks.StackManager__mirror_images_if_enabled.html``
79# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager__mirror_images_if_enabled.png``)
80# Regenerate with ``python diagrams/code_diagrams/generate.py``.
81# <pyflowchart-code-diagram> END
84if TYPE_CHECKING:
85 from .config import GCOConfig
87logger = logging.getLogger(__name__)
89# Python packages ``app.py`` imports at CDK synth time. They ship in the
90# optional ``[cdk]`` extra (see pyproject.toml), NOT the base install, so a
91# lightweight ``uvx`` / ``pip install`` of ``gco-cli`` that skips the extra
92# cannot synthesize or deploy. ``StackManager._ensure_cdk_toolchain`` checks
93# for these before invoking ``cdk`` so a missing toolchain is actionable.
94_CDK_TOOLCHAIN_MODULES = ("aws_cdk", "cdk_nag")
95_INFERENCE_STREAMING_PACKAGE_FILES = ("index.mjs", "package.json", "package-lock.json")
96_KUBECTL_PACKAGE_INPUTS = ("handler.py", "requirements.txt", "manifests")
97_LAMBDA_BUILD_MANIFEST = ".gco-build-manifest.json"
98_LAMBDA_BUILD_MANIFEST_VERSION = 1
99_LAMBDA_SOURCE_IGNORED_DIRECTORIES = frozenset({"__pycache__", ".mypy_cache", ".pytest_cache"})
100_LAMBDA_SOURCE_IGNORED_FILES = frozenset({".DS_Store"})
101_LAMBDA_SOURCE_COPY_IGNORE_PATTERNS = (
102 "__pycache__",
103 ".mypy_cache",
104 ".pytest_cache",
105 ".DS_Store",
106 "*.pyc",
107 "*.pyo",
108)
109_ASSET_LOCK_RETRY_SECONDS = 0.05
110# 15 minutes: comfortably above the longest legitimate hold (a cold publisher
111# rebuild, minutes) while bounding the pathological one (an abandoned pytest
112# session's session-long shared locks, indefinite).
113_ASSET_LOCK_TIMEOUT_SECONDS_DEFAULT = 900.0
114_CDK_ASSET_CONSUMER_MAX_ATTEMPTS = 3
115_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS = 7200.0
116_CLOUDFORMATION_DELETE_POLL_SECONDS = 15.0
117_CLOUDFORMATION_DELETE_HEARTBEAT_SECONDS = 60.0
118_BOOTSTRAP_HEALTHY_STATUSES = frozenset({"CREATE_COMPLETE", "UPDATE_COMPLETE"})
119_LIVE_VALIDATION_PROVIDER_LOG_CONTEXT = "gco_live_validation_retain_provider_log_groups"
120StackAuthorizationCallback = Callable[[str, str, str], None]
121CleanupOutcomeCallback = Callable[[str, dict[str, Any]], None]
122ChangeSetPreparedCallback = Callable[[str, str, str, str, str], None]
123PreparedChangeSetAuthority = Mapping[str, Mapping[str, Mapping[str, str]]]
124EcrRepositoryCreatedCallback = Callable[[str, Mapping[str, Any]], None]
127class _StackOperationSafetyKwargs(TypedDict):
128 """Type-preserving keyword bundle shared by strict deploy and destroy calls."""
130 allow_bootstrap: bool
131 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None
132 expected_stack_ids: Mapping[str, str | None] | None
133 prepared_change_sets: PreparedChangeSetAuthority | None
134 authorize_stack: StackAuthorizationCallback | None
135 strict_deployment_token: str | None
136 on_change_set_prepared: ChangeSetPreparedCallback | None
137 on_ecr_repository_created: EcrRepositoryCreatedCallback | None
140@dataclass(frozen=True)
141class _CdkAssetSpec:
142 """One canonical generated asset consumed by the CDK application."""
144 name: str
145 source_directory: str
146 build_directory: str
147 source_inputs: tuple[str, ...] | None
149 def paths(self, project_root: Path) -> tuple[Path, Path]:
150 lambda_dir = project_root / "lambda"
151 return lambda_dir / self.source_directory, lambda_dir / self.build_directory
154_KUBECTL_CDK_ASSET = _CdkAssetSpec(
155 name="kubectl-applier-simple",
156 source_directory="kubectl-applier-simple",
157 build_directory="kubectl-applier-simple-build",
158 source_inputs=_KUBECTL_PACKAGE_INPUTS,
159)
160_HELM_CDK_ASSET = _CdkAssetSpec(
161 name="helm-installer",
162 source_directory="helm-installer",
163 build_directory="helm-installer-build",
164 source_inputs=None,
165)
166_INFERENCE_STREAMING_CDK_ASSET = _CdkAssetSpec(
167 name="inference-streaming-proxy",
168 source_directory="inference-streaming-proxy",
169 build_directory="inference-streaming-proxy-build",
170 source_inputs=_INFERENCE_STREAMING_PACKAGE_FILES,
171)
172_CDK_ASSET_SPECS = (
173 _KUBECTL_CDK_ASSET,
174 _HELM_CDK_ASSET,
175 _INFERENCE_STREAMING_CDK_ASSET,
176)
179class _AssetThreadState(local):
180 """Per-thread nesting state; each OS lock still spans the full process."""
182 def __init__(self) -> None:
183 self.held: dict[str, tuple[bool, int]] = {}
184 self.active_consumers: dict[str, int] = {}
187_asset_thread_state = _AssetThreadState()
190def _asset_tree_paths(root: Path, source_inputs: tuple[str, ...] | None) -> Iterator[Path]:
191 """Yield deterministic source or build-tree entries below ``root``."""
192 selected: set[Path] = set()
193 if source_inputs is None:
194 selected.update(root.rglob("*"))
195 else:
196 for relative_name in source_inputs:
197 path = root / relative_name
198 if not path.exists() and not path.is_symlink(): 198 ↛ 199line 198 didn't jump to line 199 because the condition on line 198 was never true
199 raise FileNotFoundError(path)
200 selected.add(path)
201 if path.is_dir() and not path.is_symlink():
202 selected.update(path.rglob("*"))
203 yield from sorted(selected, key=lambda path: path.relative_to(root).as_posix())
206def _asset_tree_digest(
207 root: Path,
208 *,
209 source_inputs: tuple[str, ...] | None = None,
210) -> str | None:
211 """Hash every deployable entry in a source selection or complete build tree.
213 The completion manifest and local cache files are excluded. Regular-file
214 content, paths, modes, directory entries, and symlink targets are included
215 so removing any installed transitive dependency invalidates the build.
216 """
217 if not root.is_dir(): 217 ↛ 218line 217 didn't jump to line 218 because the condition on line 217 was never true
218 return None
219 digest = hashlib.sha256()
220 try:
221 for path in _asset_tree_paths(root, source_inputs):
222 relative = path.relative_to(root)
223 if any(part in _LAMBDA_SOURCE_IGNORED_DIRECTORIES for part in relative.parts):
224 continue
225 if (
226 path.name in _LAMBDA_SOURCE_IGNORED_FILES
227 or path.name == _LAMBDA_BUILD_MANIFEST
228 or path.suffix in {".pyc", ".pyo"}
229 ):
230 continue
232 metadata = path.lstat()
233 relative_bytes = relative.as_posix().encode("utf-8")
234 digest.update(len(relative_bytes).to_bytes(8, "big"))
235 digest.update(relative_bytes)
236 digest.update(stat.S_IMODE(metadata.st_mode).to_bytes(4, "big"))
238 if path.is_symlink(): 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true
239 target = os.readlink(path).encode("utf-8")
240 digest.update(b"L")
241 digest.update(len(target).to_bytes(8, "big"))
242 digest.update(target)
243 elif path.is_dir():
244 digest.update(b"D")
245 elif path.is_file(): 245 ↛ 253line 245 didn't jump to line 253 because the condition on line 245 was always true
246 file_digest = hashlib.sha256()
247 with path.open("rb") as handle:
248 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
249 file_digest.update(chunk)
250 digest.update(b"F")
251 digest.update(file_digest.digest())
252 else:
253 return None
254 except OSError, UnicodeError:
255 return None
256 return digest.hexdigest()
259def _read_build_manifest(build_dir: Path) -> dict[str, Any] | None:
260 try:
261 value = json.loads((build_dir / _LAMBDA_BUILD_MANIFEST).read_text(encoding="utf-8"))
262 except OSError, UnicodeError, json.JSONDecodeError:
263 return None
264 return value if isinstance(value, dict) else None
267def _write_build_manifest(build_dir: Path, source_digest: str) -> None:
268 """Write the completion marker only after the staged build is complete."""
269 build_digest = _asset_tree_digest(build_dir)
270 if build_digest is None: 270 ↛ 271line 270 didn't jump to line 271 because the condition on line 270 was never true
271 raise RuntimeError(f"Unable to hash completed Lambda asset {build_dir.name}")
272 manifest = {
273 "schema_version": _LAMBDA_BUILD_MANIFEST_VERSION,
274 "source_digest": source_digest,
275 "build_digest": build_digest,
276 }
277 manifest_path = build_dir / _LAMBDA_BUILD_MANIFEST
278 with manifest_path.open("x", encoding="utf-8") as handle:
279 json.dump(manifest, handle, sort_keys=True, separators=(",", ":"))
280 handle.write("\n")
281 handle.flush()
282 os.fsync(handle.fileno())
285def _asset_build_is_fresh_unlocked(
286 source_dir: Path,
287 build_dir: Path,
288 *,
289 source_inputs: tuple[str, ...] | None,
290) -> bool:
291 manifest = _read_build_manifest(build_dir)
292 if manifest is None or manifest.get("schema_version") != _LAMBDA_BUILD_MANIFEST_VERSION:
293 return False
294 source_digest = _asset_tree_digest(source_dir, source_inputs=source_inputs)
295 build_digest = _asset_tree_digest(build_dir)
296 return (
297 source_digest is not None
298 and build_digest is not None
299 and manifest.get("source_digest") == source_digest
300 and manifest.get("build_digest") == build_digest
301 )
304def _thread_asset_locks() -> dict[str, tuple[bool, int]]:
305 """Return locks held by the current thread for safe nested consumers."""
306 return _asset_thread_state.held
309def _ensure_windows_lock_byte(lock_file: BinaryIO) -> None:
310 """Ensure msvcrt has a real byte range to lock."""
311 lock_file.seek(0, os.SEEK_END)
312 if lock_file.tell() == 0:
313 lock_file.write(b"\0")
314 lock_file.flush()
315 lock_file.seek(0)
318def _windows_lock_is_contended(exc: OSError) -> bool:
319 return exc.errno in {errno.EACCES, errno.EAGAIN, errno.EDEADLK} or getattr(
320 exc,
321 "winerror",
322 None,
323 ) in {32, 33, 36}
326def _asset_lock_timeout_seconds() -> float:
327 """Bounded wait for a contended asset lock, env-tunable.
329 Contended acquisitions used to block forever with no diagnostics; a
330 destroy was once observed frozen for 40 minutes inside ``flock`` because
331 two abandoned pytest sessions still held their session-long shared locks.
332 The bound turns that silence into a warning at contention time and an
333 actionable error at the deadline.
334 """
335 raw = os.environ.get("GCO_ASSET_LOCK_TIMEOUT_SECONDS", "")
336 try:
337 value = float(raw)
338 except ValueError:
339 return _ASSET_LOCK_TIMEOUT_SECONDS_DEFAULT
340 if not math.isfinite(value) or value <= 0: 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 return _ASSET_LOCK_TIMEOUT_SECONDS_DEFAULT
342 return value
345def _warn_asset_lock_contended(lock_file: BinaryIO, *, exclusive: bool, timeout: float) -> None:
346 lock_name = getattr(lock_file, "name", "<unknown>")
347 mode = "exclusive" if exclusive else "shared"
348 logger.warning(
349 "Waiting up to %.0fs for the %s asset lock on %s — another process holds "
350 "it (a pytest session holds shared locks for its whole run; a "
351 "deploy/synth/destroy holds the exclusive lock while rebuilding). "
352 "Find the holder with `lsof %s`; tune via GCO_ASSET_LOCK_TIMEOUT_SECONDS.",
353 timeout,
354 mode,
355 lock_name,
356 lock_name,
357 )
360def _raise_asset_lock_timeout(lock_file: BinaryIO, *, timeout: float) -> None:
361 lock_name = getattr(lock_file, "name", "<unknown>")
362 raise TimeoutError(
363 f"Timed out after {timeout:.0f}s waiting for the asset lock on {lock_name}. "
364 "Another process still holds it — often an abandoned pytest session, which "
365 f"keeps shared locks until it exits. Find it with `lsof {lock_name}`, stop "
366 "it, and retry; raise GCO_ASSET_LOCK_TIMEOUT_SECONDS to wait longer."
367 )
370def _acquire_asset_file_lock(
371 lock_file: BinaryIO,
372 *,
373 exclusive: bool,
374) -> None:
375 """Acquire a platform-native interprocess lock, loudly and boundedly.
377 The first attempt is non-blocking. On contention a warning names the lock
378 file and the likely holder class, then acquisition polls until the
379 env-tunable deadline so a stuck holder produces an actionable error
380 instead of an indefinite silent hang.
381 """
382 if os.name == "nt": 382 ↛ 383line 382 didn't jump to line 383 because the condition on line 382 was never true
383 import msvcrt
385 msvcrt_api: Any = msvcrt
386 _ensure_windows_lock_byte(lock_file)
387 warned = False
388 deadline: float | None = None
389 while True:
390 lock_file.seek(0)
391 try:
392 # msvcrt exposes only exclusive byte-range locks. Serializing
393 # Windows readers and writers preserves correctness while POSIX
394 # keeps true shared-reader concurrency through flock below.
395 msvcrt_api.locking(lock_file.fileno(), msvcrt_api.LK_NBLCK, 1)
396 return
397 except OSError as exc:
398 if not _windows_lock_is_contended(exc):
399 raise
400 if not warned:
401 timeout = _asset_lock_timeout_seconds()
402 deadline = time.monotonic() + timeout
403 _warn_asset_lock_contended(lock_file, exclusive=exclusive, timeout=timeout)
404 warned = True
405 assert deadline is not None
406 if time.monotonic() >= deadline:
407 _raise_asset_lock_timeout(lock_file, timeout=_asset_lock_timeout_seconds())
408 time.sleep(_ASSET_LOCK_RETRY_SECONDS)
410 import fcntl
412 operation = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
413 try:
414 fcntl.flock(lock_file.fileno(), operation | fcntl.LOCK_NB)
415 return
416 except BlockingIOError:
417 pass
418 timeout = _asset_lock_timeout_seconds()
419 deadline = time.monotonic() + timeout
420 _warn_asset_lock_contended(lock_file, exclusive=exclusive, timeout=timeout)
421 while True:
422 try:
423 fcntl.flock(lock_file.fileno(), operation | fcntl.LOCK_NB)
424 return
425 except BlockingIOError:
426 if time.monotonic() >= deadline:
427 _raise_asset_lock_timeout(lock_file, timeout=timeout)
428 time.sleep(_ASSET_LOCK_RETRY_SECONDS)
431def _release_asset_file_lock(lock_file: BinaryIO) -> None:
432 """Release the matching platform-native interprocess lock."""
433 if os.name == "nt": 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 import msvcrt
436 msvcrt_api: Any = msvcrt
437 lock_file.seek(0)
438 msvcrt_api.locking(lock_file.fileno(), msvcrt_api.LK_UNLCK, 1)
439 return
441 import fcntl
443 fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
446@contextmanager
447def _lambda_asset_lock(build_dir: Path, *, exclusive: bool) -> Iterator[None]:
448 """Serialize publishers and keep freshness reads off rename windows."""
449 build_dir.parent.mkdir(parents=True, exist_ok=True)
450 lock_path = build_dir.with_name(f".{build_dir.name}.lock")
451 lock_key = os.path.normcase(os.path.abspath(lock_path))
452 held = _thread_asset_locks()
453 existing = held.get(lock_key)
454 if existing is not None: 454 ↛ 455line 454 didn't jump to line 455 because the condition on line 454 was never true
455 held_exclusive, depth = existing
456 if exclusive and not held_exclusive:
457 raise RuntimeError(f"Cannot upgrade shared asset lock to exclusive: {lock_path}")
458 held[lock_key] = (held_exclusive, depth + 1)
459 try:
460 yield
461 finally:
462 held[lock_key] = (held_exclusive, depth)
463 return
465 with lock_path.open("a+b") as lock_file:
466 _acquire_asset_file_lock(lock_file, exclusive=exclusive)
467 held[lock_key] = (exclusive, 1)
468 try:
469 yield
470 finally:
471 held.pop(lock_key, None)
472 _release_asset_file_lock(lock_file)
475def _asset_build_is_fresh(
476 source_dir: Path,
477 build_dir: Path,
478 *,
479 source_inputs: tuple[str, ...] | None,
480) -> bool:
481 with _lambda_asset_lock(build_dir, exclusive=False):
482 return _asset_build_is_fresh_unlocked(
483 source_dir,
484 build_dir,
485 source_inputs=source_inputs,
486 )
489def _remove_asset_tree(path: Path) -> None:
490 if path.exists() or path.is_symlink():
491 _safe_rmtree(path)
494def _recover_interrupted_asset_publish(build_dir: Path) -> None:
495 """Restore a prior final tree and discard abandoned staging directories."""
496 staging_dirs = list(build_dir.parent.glob(f".{build_dir.name}.staging-*"))
497 backup_dirs = list(build_dir.parent.glob(f".{build_dir.name}.backup-*"))
499 if not build_dir.exists() and backup_dirs: 499 ↛ 500line 499 didn't jump to line 500 because the condition on line 499 was never true
500 try:
501 newest_backup = max(backup_dirs, key=lambda path: path.stat().st_mtime_ns)
502 except OSError:
503 newest_backup = backup_dirs[0]
504 os.replace(newest_backup, build_dir)
506 for path in [*staging_dirs, *backup_dirs]: 506 ↛ 507line 506 didn't jump to line 507 because the loop on line 506 never started
507 _remove_asset_tree(path)
510def _publish_staged_asset(staging_dir: Path, build_dir: Path) -> None:
511 """Publish one complete staged tree with rollback to the previous final."""
512 backup_dir = build_dir.with_name(f".{build_dir.name}.backup-{uuid.uuid4().hex}")
513 had_previous = build_dir.exists()
514 if had_previous:
515 os.replace(build_dir, backup_dir)
516 try:
517 os.replace(staging_dir, build_dir)
518 except Exception:
519 if had_previous and backup_dir.exists() and not build_dir.exists():
520 os.replace(backup_dir, build_dir)
521 raise
522 if backup_dir.exists():
523 _remove_asset_tree(backup_dir)
526def _prepare_lambda_asset(
527 source_dir: Path,
528 build_dir: Path,
529 *,
530 source_inputs: tuple[str, ...] | None,
531 display_name: str,
532 builder: Callable[[Path], None],
533) -> bool:
534 """Build and atomically publish an asset when its completion proof is stale.
536 Freshness is checked under a *shared* lock first, so the common case —
537 the asset is already source-current — never contends: concurrent pytest
538 workers validate in parallel instead of serialising behind one writer,
539 and a deploy/destroy against fresh assets never blocks on a pytest
540 session's session-long shared locks. Only a genuinely stale asset
541 escalates to the exclusive publisher lock, which re-checks freshness
542 after acquisition (another publisher may have finished the same rebuild
543 while this one waited).
544 """
545 if _asset_build_is_fresh(source_dir, build_dir, source_inputs=source_inputs):
546 return False
547 with _lambda_asset_lock(build_dir, exclusive=True):
548 _recover_interrupted_asset_publish(build_dir)
549 source_digest = _asset_tree_digest(source_dir, source_inputs=source_inputs)
550 if source_digest is None: 550 ↛ 551line 550 didn't jump to line 551 because the condition on line 550 was never true
551 raise RuntimeError(f"{display_name} source inputs are incomplete or unreadable")
552 if _asset_build_is_fresh_unlocked( 552 ↛ 557line 552 didn't jump to line 557 because the condition on line 552 was never true
553 source_dir,
554 build_dir,
555 source_inputs=source_inputs,
556 ):
557 return False
559 print(f" Building {display_name}...")
560 staging_dir = Path(
561 tempfile.mkdtemp(prefix=f".{build_dir.name}.staging-", dir=build_dir.parent)
562 )
563 try:
564 builder(staging_dir)
565 if _asset_tree_digest(source_dir, source_inputs=source_inputs) != source_digest: 565 ↛ 566line 565 didn't jump to line 566 because the condition on line 565 was never true
566 raise RuntimeError(f"{display_name} sources changed while packaging")
567 _write_build_manifest(staging_dir, source_digest)
568 if not _asset_build_is_fresh_unlocked( 568 ↛ 573line 568 didn't jump to line 573 because the condition on line 568 was never true
569 source_dir,
570 staging_dir,
571 source_inputs=source_inputs,
572 ):
573 raise RuntimeError(f"{display_name} completion manifest verification failed")
574 _publish_staged_asset(staging_dir, build_dir)
575 finally:
576 _remove_asset_tree(staging_dir)
577 print(f" {display_name} built successfully")
578 return True
581def _atomic_copy_file(source: Path, target: Path) -> None:
582 """Replace one checked-in Lambda source copy without exposing partial bytes."""
583 temporary = target.with_name(f".{target.name}.tmp-{uuid.uuid4().hex}")
584 try:
585 shutil.copy2(source, temporary)
586 os.replace(temporary, target)
587 finally:
588 temporary.unlink(missing_ok=True)
591def _atomic_write_bytes(target: Path, content: bytes, *, mode: int | None = None) -> None:
592 """Atomically restore exact bytes without exposing a partial configuration."""
593 temporary = target.with_name(f".{target.name}.tmp-{uuid.uuid4().hex}")
594 try:
595 temporary.write_bytes(content)
596 if mode is not None: 596 ↛ 598line 596 didn't jump to line 598 because the condition on line 596 was always true
597 os.chmod(temporary, mode)
598 os.replace(temporary, target)
599 finally:
600 temporary.unlink(missing_ok=True)
603@lru_cache(maxsize=1)
604def _known_cloudformation_regions() -> frozenset[str]:
605 """Return every AWS SDK-known Region that exposes CloudFormation."""
606 return known_cloudformation_regions()
609class CdkToolchainError(RuntimeError):
610 """The CDK Python toolchain (``aws-cdk-lib`` / ``cdk-nag``) is not
611 importable in the environment that will run ``cdk``.
613 Raised before shelling out to ``cdk`` so operators get a clear install
614 hint instead of the cryptic ``ImportError: cannot import name 'App' from
615 'aws_cdk'`` that the ``python3 app.py`` synth subprocess would otherwise
616 emit from a base (extra-less) install.
617 """
620@dataclass
621class StackInfo:
622 """Information about a CDK stack."""
624 name: str
625 status: str
626 region: str
627 created_time: datetime | None = None
628 updated_time: datetime | None = None
629 outputs: dict[str, str] = field(default_factory=dict)
630 tags: dict[str, str] = field(default_factory=dict)
632 def to_dict(self) -> dict[str, Any]:
633 return {
634 "name": self.name,
635 "status": self.status,
636 "region": self.region,
637 "created_time": self.created_time.isoformat() if self.created_time else None,
638 "updated_time": self.updated_time.isoformat() if self.updated_time else None,
639 "outputs": self.outputs,
640 "tags": self.tags,
641 }
644def _safe_rmtree(path: Path) -> None:
645 """Remove a directory tree, handling broken symlinks on macOS.
647 shutil.rmtree can fail with ``OSError: [Errno 66] Directory not empty``
648 on macOS when pip-installed packages (e.g. botocore) contain broken
649 symlinks or extended-attribute resource forks.
651 Falls back to ``rm -rf`` via subprocess, but only after validating the
652 path is a real directory under the project tree to avoid accidents.
653 """
654 resolved = path.resolve()
656 # Safety: refuse to remove anything that isn't clearly a final, staging,
657 # or rollback Lambda build artifact inside the project tree.
658 artifact_name = resolved.name
659 is_final = artifact_name.endswith("-build")
660 is_ephemeral = artifact_name.startswith(".") and (
661 ".staging-" in artifact_name or ".backup-" in artifact_name
662 )
663 if "lambda" not in resolved.parts or not (is_final or is_ephemeral):
664 raise ValueError(f"Refusing to remove unexpected path: {resolved}")
666 try:
667 shutil.rmtree(str(resolved))
668 except OSError:
669 subprocess.run(["rm", "-rf", "--", str(resolved)], check=True)
672# Container runtime detection lives in cli/_container_runtime.py so it can
673# be shared between StackManager (CDK asset bundling) and ImageManager
674# (gco images build/push). The uncached probe is imported from there;
675# this module keeps its own small cache so existing tests that reset
676# ``cli.stacks._container_runtime_cache`` continue to work without
677# touching the new module's cache.
678from cli._container_runtime import ( # noqa: E402
679 _detect_container_runtime_uncached,
680)
682# Cached result for container runtime detection (None = not yet checked)
683_container_runtime_cache: str | None = None
684_container_runtime_checked: bool = False
687def _detect_container_runtime() -> str | None:
688 """
689 Detect available container runtime for CDK asset bundling.
691 Thin caching wrapper around the shared
692 ``cli._container_runtime._detect_container_runtime_uncached`` probe.
693 The cache state is held on this module so tests that patch or reset
694 ``cli.stacks._container_runtime_cache`` keep working unchanged.
695 """
696 global _container_runtime_cache, _container_runtime_checked
697 if _container_runtime_checked: 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 return _container_runtime_cache
700 _container_runtime_cache = _detect_container_runtime_uncached()
701 _container_runtime_checked = True
702 return _container_runtime_cache
705def prepare_cdk_assets(project_root: str | Path) -> None:
706 """Prepare every ignored Lambda asset consumed by the CDK application.
708 This is the shared entry point for build-only callers. CDK consumers must
709 use :func:`cdk_asset_consumer` so the resulting paths remain immutable
710 until app construction and synthesis finish.
711 """
712 manager = StackManager.__new__(StackManager)
713 manager.project_root = Path(project_root)
714 manager._ensure_lambda_build()
717def _thread_asset_consumers() -> dict[str, int]:
718 return _asset_thread_state.active_consumers
721@contextmanager
722def cdk_asset_consumer(project_root: str | Path) -> Iterator[None]:
723 """Hold source-current generated assets stable through CDK synthesis.
725 Preparation runs before any shared locks are acquired. The complete set of
726 canonical paths is then locked in deterministic order and every completion
727 manifest is revalidated while publishers are excluded. A stale observation
728 releases all locks and retries preparation; repeated source churn fails
729 closed instead of exposing CDK to a missing or mixed-version tree.
730 """
731 root = Path(project_root)
732 root_key = os.path.normcase(os.path.abspath(root))
733 active = _thread_asset_consumers()
734 if root_key in active:
735 active[root_key] += 1
736 try:
737 yield
738 finally:
739 active[root_key] -= 1
740 return
742 stale_assets: list[str] = []
743 # Attempt 0 validates under shared locks without preparing anything: when
744 # every asset is already source-current (always true in CI, where the
745 # composite build action runs first, and true locally on any second run)
746 # the consumer takes no exclusive lock and does one hash pass. Concurrent
747 # consumers — xdist workers — therefore proceed in parallel instead of
748 # serialising behind the publisher lock. Later attempts keep the original
749 # prepare-then-revalidate budget for genuinely stale trees.
750 for attempt in range(_CDK_ASSET_CONSUMER_MAX_ATTEMPTS + 1): 750 ↛ 787line 750 didn't jump to line 787 because the loop on line 750 didn't complete
751 if attempt: 751 ↛ 752line 751 didn't jump to line 752 because the condition on line 751 was never true
752 prepare_cdk_assets(root)
753 resolved_assets = []
754 for spec in _CDK_ASSET_SPECS:
755 source_dir, build_dir = spec.paths(root)
756 # Include a source-backed path even during the publisher's
757 # final-to-backup rename gap, when the canonical build is absent.
758 if source_dir.exists() or build_dir.exists():
759 resolved_assets.append((spec, source_dir, build_dir))
761 with ExitStack() as locks:
762 for _spec, _source_dir, build_dir in sorted(
763 resolved_assets,
764 key=lambda item: str(item[2]),
765 ):
766 locks.enter_context(_lambda_asset_lock(build_dir, exclusive=False))
768 stale_assets = [
769 spec.name
770 for spec, source_dir, build_dir in resolved_assets
771 if not _asset_build_is_fresh_unlocked(
772 source_dir,
773 build_dir,
774 source_inputs=spec.source_inputs,
775 )
776 ]
777 if stale_assets: 777 ↛ 778line 777 didn't jump to line 778 because the condition on line 777 was never true
778 continue
780 active[root_key] = 1
781 try:
782 yield
783 finally:
784 active.pop(root_key, None)
785 return
787 names = ", ".join(stale_assets) or "unknown assets"
788 raise RuntimeError(
789 "Generated CDK assets changed repeatedly while acquiring consumer locks: "
790 f"{names}. Stop concurrent source edits and retry."
791 )
794class StackManager:
795 """Manages CDK stack operations."""
797 def __init__(self, config: GCOConfig, project_root: Path | None = None):
798 self.config = config
799 self.project_root = project_root or self._find_project_root()
800 # Resolve CDK only when a CDK-backed operation runs. CloudFormation-only
801 # status/output commands must not require a local Node/CDK installation.
802 self._cdk_path: str | None = None
803 self._active_cdk_processes: dict[int, Any] = {}
804 self._active_cdk_lock = Lock()
805 self._cdk_cancel_event = Event()
807 def _find_project_root(self) -> Path:
808 """Find the project root by looking for cdk.json."""
809 current = Path.cwd()
810 for parent in [current] + list(current.parents):
811 if (parent / "cdk.json").exists():
812 return parent
813 return current
815 def _find_cdk(self) -> str:
816 """Find the dependency-locked CDK executable when available."""
817 # Prefer the repository's locked tool when ``npm ci`` has populated it.
818 local_cdk = self.project_root / "node_modules" / ".bin" / "cdk"
819 if local_cdk.is_file(): 819 ↛ 820line 819 didn't jump to line 820 because the condition on line 819 was never true
820 return str(local_cdk)
822 # Fall back to PATH for installed distributions that do not include
823 # the repository's root npm graph.
824 try:
825 result = subprocess.run(["which", "cdk"], capture_output=True, text=True, check=True)
826 return result.stdout.strip()
827 except subprocess.CalledProcessError:
828 pass
830 # Check common global-install locations.
831 for path in ["/usr/local/bin/cdk", "~/.npm-global/bin/cdk"]:
832 expanded = os.path.expanduser(path)
833 if os.path.exists(expanded):
834 return expanded
836 raise CdkToolchainError(
837 "AWS CDK CLI is not installed. Run "
838 "'npm ci --ignore-scripts --no-audit --no-fund' at the project root "
839 "to install the dependency-locked CLI."
840 )
842 @staticmethod
843 def _kubectl_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
844 """Return whether the kubectl build has a valid full-tree completion proof."""
845 return _asset_build_is_fresh(
846 source_dir,
847 build_dir,
848 source_inputs=_KUBECTL_CDK_ASSET.source_inputs,
849 )
851 @staticmethod
852 def _helm_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
853 """Return whether the Helm build has a valid full-tree completion proof."""
854 return _asset_build_is_fresh(
855 source_dir,
856 build_dir,
857 source_inputs=_HELM_CDK_ASSET.source_inputs,
858 )
860 @staticmethod
861 def _inference_streaming_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
862 """Return whether the Node build has a valid full-tree completion proof."""
863 return _asset_build_is_fresh(
864 source_dir,
865 build_dir,
866 source_inputs=_INFERENCE_STREAMING_CDK_ASSET.source_inputs,
867 )
869 def _ensure_lambda_build(self) -> None:
870 """Atomically prepare every generated Lambda asset when source-stale.
872 Every builder takes its per-asset interprocess lock, repairs an
873 interrupted publish, and rechecks freshness before doing installation
874 work. Concurrent app evaluations therefore either reuse one complete
875 final tree or publish another complete tree; they never share a
876 directory while pip/npm/copy operations are mutating it.
877 """
878 for spec, builder in (
879 (_KUBECTL_CDK_ASSET, self._build_kubectl_lambda),
880 (_HELM_CDK_ASSET, self._build_helm_installer_lambda),
881 (_INFERENCE_STREAMING_CDK_ASSET, self._build_inference_streaming_proxy_lambda),
882 ):
883 source_dir, _build_dir = spec.paths(self.project_root)
884 if source_dir.exists():
885 builder()
887 def _check_and_fix_stuck_stack(
888 self,
889 stack_name: str,
890 *,
891 expected_stack_id: str | None = None,
892 authorize_stack: StackAuthorizationCallback | None = None,
893 strict_ownership: bool = False,
894 ) -> None:
895 """Delete a stuck stack only after revalidating its immutable identity."""
896 import boto3
898 region = self._get_deploy_region(stack_name)
899 if not region:
900 if strict_ownership: 900 ↛ 901line 900 didn't jump to line 901 because the condition on line 900 was never true
901 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
902 return
904 cfn = boto3.client("cloudformation", region_name=region)
905 try:
906 response = cfn.describe_stacks(StackName=stack_name)
907 except ClientError as exc:
908 error = exc.response.get("Error", {})
909 if (
910 error.get("Code") == "ValidationError"
911 and "does not exist" in str(error.get("Message", "")).lower()
912 ):
913 return
914 if strict_ownership:
915 raise
916 logger.debug("Stack pre-check for %s failed: %s", stack_name, exc)
917 return
918 except Exception as exc:
919 if strict_ownership: 919 ↛ 920line 919 didn't jump to line 920 because the condition on line 919 was never true
920 raise
921 logger.debug("Stack pre-check for %s failed: %s", stack_name, exc)
922 return
924 stacks = response.get("Stacks", [])
925 if len(stacks) != 1: 925 ↛ 926line 925 didn't jump to line 926 because the condition on line 925 was never true
926 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
927 stack = stacks[0]
928 stack_id = str(stack.get("StackId") or "")
929 if stack.get("StackName") != stack_name or not stack_id: 929 ↛ 930line 929 didn't jump to line 930 because the condition on line 929 was never true
930 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
931 if strict_ownership and expected_stack_id is None:
932 raise RuntimeError(
933 f"Refusing to adopt uncheckpointed stack {region}:{stack_name} ({stack_id})"
934 )
935 if expected_stack_id is not None and stack_id != expected_stack_id: 935 ↛ 936line 935 didn't jump to line 936 because the condition on line 935 was never true
936 raise RuntimeError(
937 f"Stack identity changed for {region}:{stack_name}; expected {expected_stack_id}, "
938 f"found {stack_id}"
939 )
941 stuck_states = {
942 "REVIEW_IN_PROGRESS",
943 "ROLLBACK_COMPLETE",
944 "ROLLBACK_FAILED",
945 "CREATE_FAILED",
946 "DELETE_FAILED",
947 }
948 status = str(stack.get("StackStatus") or "")
949 if status not in stuck_states:
950 return
951 if authorize_stack is not None: 951 ↛ 952line 951 didn't jump to line 952 because the condition on line 951 was never true
952 authorize_stack(stack_name, region, stack_id)
954 print(f" Stack {stack_name} is in {status} state, cleaning up...")
955 cfn.delete_stack(StackName=stack_id)
956 waiter = cfn.get_waiter("stack_delete_complete")
957 waiter.wait(StackName=stack_id, WaiterConfig={"Delay": 10, "MaxAttempts": 60})
958 print(f" Stack {stack_name} cleaned up, will recreate on deploy")
960 def _diagnose_deploy_failure(self, stack_name: str) -> None:
961 """Fetch CloudFormation events after a failed deploy and print diagnostics.
963 Gives users actionable information instead of just the CDK error message.
964 """
965 import boto3
967 region = self._get_deploy_region(stack_name)
968 if not region:
969 return
971 try:
972 cfn = boto3.client("cloudformation", region_name=region)
974 # Get recent events
975 response = cfn.describe_stack_events(StackName=stack_name)
976 events = response.get("StackEvents", [])
978 # Filter to failed events
979 failed = [
980 e
981 for e in events[:20]
982 if "FAILED" in e.get("ResourceStatus", "")
983 or "ROLLBACK" in e.get("ResourceStatus", "")
984 ]
986 if failed:
987 print(f"\n CloudFormation failure details for {stack_name}:")
988 for event in failed[:5]:
989 resource = event.get("LogicalResourceId", "unknown")
990 status = event.get("ResourceStatus", "unknown")
991 reason = event.get("ResourceStatusReason", "no reason given")
992 print(f" {resource}: {status}")
993 print(f" {reason}")
995 # Check stack status for actionable advice
996 try:
997 stack_resp = cfn.describe_stacks(StackName=stack_name)
998 status = stack_resp["Stacks"][0]["StackStatus"]
1000 advice = {
1001 "REVIEW_IN_PROGRESS": (
1002 "Stack is stuck in REVIEW_IN_PROGRESS. "
1003 "Run: aws cloudformation delete-stack "
1004 f"--stack-name {stack_name} --region {region}"
1005 ),
1006 "ROLLBACK_COMPLETE": (
1007 "Stack rolled back. Delete it and retry: "
1008 f"aws cloudformation delete-stack "
1009 f"--stack-name {stack_name} --region {region}"
1010 ),
1011 "ROLLBACK_FAILED": (
1012 "Stack rollback failed. Delete with --retain: "
1013 f"aws cloudformation delete-stack "
1014 f"--stack-name {stack_name} --region {region}"
1015 ),
1016 "UPDATE_ROLLBACK_COMPLETE": (
1017 "Update rolled back but stack is stable. "
1018 "Check the events above and retry the deploy."
1019 ),
1020 }
1022 if status in advice: 1022 ↛ exitline 1022 didn't return from function '_diagnose_deploy_failure' because the condition on line 1022 was always true
1023 print(f"\n Suggested fix: {advice[status]}")
1025 except Exception as e:
1026 logger.debug("Failed to parse stack events: %s", e)
1028 except Exception as e:
1029 logger.debug("Failed to diagnose deploy failure for %s: %s", stack_name, e)
1030 # Best effort — don't fail the deploy further
1032 def _sync_lambda_sources(self) -> None:
1033 """Atomically synchronize canonical shared files before asset ensures.
1035 Checked-in copies keep raw CDK evaluation deterministic. Deploy updates
1036 those copies before generated assets are checked, and never mutates a
1037 generated final build tree in place.
1038 """
1039 if getattr(self, "_lambda_sources_synced", False):
1040 return
1042 lambda_dir = self.project_root / "lambda"
1043 shared_source_targets = {
1044 lambda_dir / "proxy-shared" / "proxy_utils.py": [
1045 lambda_dir / "api-gateway-proxy" / "proxy_utils.py",
1046 lambda_dir / "regional-api-proxy" / "proxy_utils.py",
1047 ],
1048 lambda_dir / "tls-shared" / "backend_tls.py": [
1049 lambda_dir / "proxy-shared" / "backend_tls.py",
1050 lambda_dir / "api-gateway-proxy" / "backend_tls.py",
1051 lambda_dir / "regional-api-proxy" / "backend_tls.py",
1052 ],
1053 }
1054 for shared_source, targets in shared_source_targets.items():
1055 if not shared_source.exists():
1056 continue
1057 for target in targets:
1058 if target.parent.exists(): 1058 ↛ 1057line 1058 didn't jump to line 1057 because the condition on line 1058 was always true
1059 _atomic_copy_file(shared_source, target)
1060 self._lambda_sources_synced = True
1062 def _rebuild_lambda_packages(self) -> None:
1063 """Compatibility wrapper for a source-current atomic asset ensure."""
1064 if getattr(self, "_lambda_packages_rebuilt", False):
1065 return
1066 self._ensure_lambda_build()
1067 self._lambda_packages_rebuilt = True
1069 def _build_lambda_packages(self) -> None:
1070 """Source-check and atomically publish all generated Lambda packages."""
1071 self._build_kubectl_lambda()
1072 self._build_helm_installer_lambda()
1073 self._build_inference_streaming_proxy_lambda()
1075 def _build_kubectl_lambda(self) -> None:
1076 """Build the kubectl-applier-simple Lambda package."""
1077 source_dir, build_dir = _KUBECTL_CDK_ASSET.paths(self.project_root)
1078 requirements = source_dir / "requirements.txt"
1079 if not source_dir.is_dir() or not requirements.is_file():
1080 return
1082 def build(staging_dir: Path) -> None:
1083 shutil.copy2(source_dir / "handler.py", staging_dir / "handler.py")
1084 shutil.copy2(requirements, staging_dir / "requirements.txt")
1085 shutil.copytree(source_dir / "manifests", staging_dir / "manifests")
1086 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - static pip arguments and project-owned paths
1087 [
1088 sys.executable,
1089 "-m",
1090 "pip",
1091 "install",
1092 "-r",
1093 str(requirements),
1094 "-t",
1095 str(staging_dir),
1096 "--upgrade",
1097 "--platform",
1098 "manylinux2014_x86_64",
1099 "--only-binary=:all:",
1100 "--quiet",
1101 ],
1102 capture_output=True,
1103 text=True,
1104 )
1105 if result.returncode != 0: 1105 ↛ 1106line 1105 didn't jump to line 1106 because the condition on line 1105 was never true
1106 raise RuntimeError(
1107 "kubectl Lambda dependency installation failed: " + result.stderr[:200]
1108 )
1110 _prepare_lambda_asset(
1111 source_dir,
1112 build_dir,
1113 source_inputs=_KUBECTL_CDK_ASSET.source_inputs,
1114 display_name="kubectl-applier-simple Lambda package",
1115 builder=build,
1116 )
1118 def _build_helm_installer_lambda(self) -> None:
1119 """Build the complete helm-installer Lambda Docker context."""
1120 source_dir, build_dir = _HELM_CDK_ASSET.paths(self.project_root)
1121 if not source_dir.is_dir():
1122 return
1124 def build(staging_dir: Path) -> None:
1125 shutil.copytree(
1126 source_dir,
1127 staging_dir,
1128 ignore=shutil.ignore_patterns(*_LAMBDA_SOURCE_COPY_IGNORE_PATTERNS),
1129 dirs_exist_ok=True,
1130 )
1132 _prepare_lambda_asset(
1133 source_dir,
1134 build_dir,
1135 source_inputs=_HELM_CDK_ASSET.source_inputs,
1136 display_name="helm-installer Lambda package",
1137 builder=build,
1138 )
1140 def _build_inference_streaming_proxy_lambda(self) -> None:
1141 """Build the Node.js streaming Lambda with its pinned AWS SDK clients."""
1142 source_dir, build_dir = _INFERENCE_STREAMING_CDK_ASSET.paths(self.project_root)
1143 if not source_dir.is_dir():
1144 return
1146 package_files = _INFERENCE_STREAMING_CDK_ASSET.source_inputs
1147 assert package_files is not None
1148 missing = [name for name in package_files if not (source_dir / name).is_file()]
1149 if missing: 1149 ↛ 1150line 1149 didn't jump to line 1150 because the condition on line 1149 was never true
1150 raise RuntimeError(
1151 "Inference streaming Lambda package is incomplete; missing: " + ", ".join(missing)
1152 )
1153 try:
1154 package_manager = str(
1155 json.loads((source_dir / "package.json").read_text(encoding="utf-8")).get(
1156 "packageManager", ""
1157 )
1158 )
1159 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
1160 raise RuntimeError("Unable to read the inference streaming Lambda npm pin") from exc
1161 required_npm = package_manager.removeprefix("npm@")
1162 version_parts = required_npm.split(".")
1163 if (
1164 not package_manager.startswith("npm@")
1165 or len(version_parts) != 3
1166 or any(not part.isdigit() for part in version_parts)
1167 ):
1168 raise RuntimeError(
1169 "Inference streaming Lambda packageManager must pin an exact npm version"
1170 )
1172 def build(staging_dir: Path) -> None:
1173 npm = shutil.which("npm")
1174 if npm is None: 1174 ↛ 1175line 1174 didn't jump to line 1175 because the condition on line 1174 was never true
1175 raise RuntimeError(
1176 f"npm {required_npm} is required to package the inference streaming Lambda; "
1177 "install the Node.js version pinned in .nvmrc"
1178 )
1179 try:
1180 version_result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved executable and project-owned cwd
1181 [npm, "--version"],
1182 cwd=source_dir,
1183 capture_output=True,
1184 text=True,
1185 timeout=30,
1186 )
1187 except (OSError, subprocess.TimeoutExpired) as exc:
1188 raise RuntimeError("Unable to verify the npm packaging version") from exc
1189 actual_npm = version_result.stdout.strip()
1190 if version_result.returncode != 0 or actual_npm != required_npm:
1191 found = actual_npm or "unavailable"
1192 raise RuntimeError(
1193 f"npm {required_npm} is required to package the inference streaming Lambda; "
1194 f"found {found}. Run: npm install --global npm@{required_npm}"
1195 )
1197 for name in package_files:
1198 shutil.copy2(source_dir / name, staging_dir / name)
1199 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved npm path, static arguments, and project-owned cwd
1200 [
1201 npm,
1202 "ci",
1203 "--omit=dev",
1204 "--ignore-scripts",
1205 "--no-audit",
1206 "--no-fund",
1207 ],
1208 cwd=staging_dir,
1209 capture_output=True,
1210 text=True,
1211 )
1212 if result.returncode != 0:
1213 raise RuntimeError(
1214 "Failed to install pinned inference streaming Lambda dependencies: "
1215 + result.stderr[:500]
1216 )
1218 _prepare_lambda_asset(
1219 source_dir,
1220 build_dir,
1221 source_inputs=package_files,
1222 display_name="inference-streaming-proxy Lambda package",
1223 builder=build,
1224 )
1226 def _get_python_path(self) -> str:
1227 """
1228 Get PYTHONPATH that includes the current Python's site-packages.
1230 This is critical for pipx installations where CDK runs `python3 app.py`
1231 using the system Python, which doesn't have aws_cdk installed.
1232 By setting PYTHONPATH, we ensure CDK's subprocess can find our modules.
1233 """
1234 # Get all site-packages directories from the current Python
1235 site_packages = site.getsitepackages()
1237 # Also include user site-packages if available
1238 user_site = site.getusersitepackages()
1239 if user_site and os.path.isdir(user_site): 1239 ↛ 1240line 1239 didn't jump to line 1240 because the condition on line 1239 was never true
1240 site_packages.append(user_site)
1242 # Include the directory containing the current module (for editable installs)
1243 current_module_dir = Path(__file__).parent.parent
1244 if current_module_dir.exists(): 1244 ↛ 1248line 1244 didn't jump to line 1248 because the condition on line 1244 was always true
1245 site_packages.append(str(current_module_dir))
1247 # Combine with existing PYTHONPATH if any
1248 existing_path = os.environ.get("PYTHONPATH", "")
1249 all_paths = site_packages + ([existing_path] if existing_path else [])
1251 return os.pathsep.join(all_paths)
1253 def _ensure_cdk_toolchain(self) -> None:
1254 """Preflight the CDK Python toolchain before invoking ``cdk``.
1256 Infra operations run ``python3 app.py`` (via the Node ``cdk`` CLI),
1257 which imports ``aws_cdk`` and ``cdk_nag``. Those ship in the optional
1258 ``[cdk]`` extra — a base ``uvx`` / ``pip`` install of ``gco-cli`` does
1259 not include them, so the synth subprocess fails with a cryptic
1260 ``ImportError: cannot import name 'App' from 'aws_cdk'``. Detect the
1261 missing toolchain up front and raise :class:`CdkToolchainError` with an
1262 actionable install hint instead.
1263 """
1264 missing = [m for m in _CDK_TOOLCHAIN_MODULES if importlib.util.find_spec(m) is None]
1265 if not missing:
1266 return
1267 raise CdkToolchainError(
1268 "CDK toolchain not available: cannot import "
1269 + ", ".join(missing)
1270 + ".\nInfrastructure operations (deploy / synth / diff / list / destroy / "
1271 "bootstrap) need the CDK Python packages installed in the SAME "
1272 "environment as the `gco` CLI, plus a repository checkout providing "
1273 "`app.py` and `cdk.json`.\n"
1274 "Install the `[cdk]` extra one of these ways:\n"
1275 ' - uv: uv tool install "gco-cli[cdk] @ '
1276 'git+https://github.com/awslabs/global-capacity-orchestrator-on-aws.git@<tag>"\n'
1277 ' - pip: pip install -e ".[cdk,mcp]" (from a clone)\n'
1278 " - or use the dev container (see QUICKSTART.md), which bundles the "
1279 "full toolchain.\n"
1280 "See gco_mcp/README.md (Setup) for the deploy-capable configuration."
1281 )
1283 @staticmethod
1284 def _terminate_cdk_process(process: Any) -> None:
1285 """Terminate one complete CDK process tree with a bounded grace period."""
1286 if process.poll() is not None: 1286 ↛ 1287line 1286 didn't jump to line 1287 because the condition on line 1286 was never true
1287 return
1289 if os.name == "nt":
1290 taskkill = shutil.which("taskkill.exe") or shutil.which("taskkill")
1292 def terminate_tree(*, force: bool) -> bool:
1293 if taskkill is None: 1293 ↛ 1294line 1293 didn't jump to line 1294 because the condition on line 1293 was never true
1294 return False
1295 command = [taskkill, "/PID", str(process.pid), "/T"]
1296 if force:
1297 command.append("/F")
1298 try:
1299 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved Windows system utility and numeric child PID
1300 command,
1301 capture_output=True,
1302 text=True,
1303 check=False,
1304 timeout=30,
1305 creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
1306 )
1307 except OSError, subprocess.TimeoutExpired:
1308 return False
1309 return result.returncode == 0
1311 terminate_tree(force=False)
1312 try:
1313 process.wait(timeout=30)
1314 except OSError, subprocess.TimeoutExpired:
1315 terminate_tree(force=True)
1316 if process.poll() is None: 1316 ↛ 1317line 1316 didn't jump to line 1317 because the condition on line 1316 was never true
1317 process.kill()
1318 process.wait()
1319 return
1321 try:
1322 os.killpg(process.pid, signal.SIGTERM)
1323 process.wait(timeout=30)
1324 except OSError, subprocess.TimeoutExpired:
1325 if process.poll() is None: 1325 ↛ exitline 1325 didn't return from function '_terminate_cdk_process' because the condition on line 1325 was always true
1326 try:
1327 os.killpg(process.pid, signal.SIGKILL)
1328 finally:
1329 process.wait()
1331 def cancel_active_cdk_processes(self) -> None:
1332 """Prevent new CDK work and terminate every process group currently registered."""
1333 self._cdk_cancel_event.set()
1334 with self._active_cdk_lock:
1335 processes = list(self._active_cdk_processes.values())
1336 for process in processes:
1337 self._terminate_cdk_process(process)
1339 def _run_cdk(
1340 self,
1341 command: list[str],
1342 capture_output: bool = False,
1343 env: dict[str, str] | None = None,
1344 timeout: float | None = None,
1345 ) -> subprocess.CompletedProcess[str]:
1346 """Run a CDK command.
1348 Args:
1349 command: CDK subcommand argv (e.g. ``["destroy", "gco-us-east-1", "--force"]``).
1350 capture_output: Capture stdout / stderr instead of streaming.
1351 env: Extra env vars merged onto the parent process environment.
1352 timeout: Wall-clock timeout in seconds. ``None`` (default) waits
1353 forever — preserving the old behaviour for ``synth`` / ``list``.
1354 When set, on timeout we send SIGTERM, give the CDK process up
1355 to 30 seconds to exit cleanly, then SIGKILL, and finally
1356 re-raise ``subprocess.TimeoutExpired`` so callers can decide
1357 how to handle a hung subprocess. ``deploy()`` and ``destroy()``
1358 pass a per-stack budget so a wedged ``cdk destroy`` (e.g. its
1359 post-delete polling loop hanging after CloudFormation has
1360 already finished) can't block the orchestrator forever.
1361 """
1362 # Fail fast with an actionable message when the CDK Python toolchain
1363 # isn't importable (e.g. a base uvx/pip install without the [cdk]
1364 # extra), instead of letting the ``python3 app.py`` subprocess surface
1365 # a cryptic ImportError.
1366 self._ensure_cdk_toolchain()
1368 # These commands all evaluate app.py, including list and destroy. The
1369 # stack graph references ignored generated Lambda assets, so prepare
1370 # them centrally rather than relying on individual command wrappers.
1371 if command and command[0] in {"deploy", "destroy", "diff", "list", "synth"}: 1371 ↛ 1374line 1371 didn't jump to line 1374 because the condition on line 1371 was always true
1372 self._ensure_lambda_build()
1374 full_env = os.environ.copy()
1376 # Inject PYTHONPATH so CDK's python3 subprocess can find aws_cdk
1377 # This is essential for pipx installations
1378 full_env["PYTHONPATH"] = self._get_python_path()
1380 if env:
1381 full_env.update(env)
1383 cdk_path = self._cdk_path
1384 if cdk_path is None: 1384 ↛ 1385line 1384 didn't jump to line 1385 because the condition on line 1384 was never true
1385 cdk_path = self._find_cdk()
1386 self._cdk_path = cdk_path
1387 cdk_cmd = [cdk_path, *command]
1389 if self._cdk_cancel_event.is_set(): 1389 ↛ 1390line 1389 didn't jump to line 1390 because the condition on line 1389 was never true
1390 raise RuntimeError("CDK operation cancelled before process start")
1391 popen_kwargs: dict[str, Any] = {
1392 "cwd": self.project_root,
1393 "stdout": subprocess.PIPE if capture_output else None,
1394 "stderr": subprocess.PIPE if capture_output else None,
1395 "text": True,
1396 "env": full_env,
1397 "start_new_session": os.name == "posix",
1398 }
1399 if os.name == "nt":
1400 popen_kwargs["creationflags"] = getattr(
1401 subprocess,
1402 "CREATE_NEW_PROCESS_GROUP",
1403 0,
1404 )
1405 process = subprocess.Popen( # nosemgrep: dangerous-subprocess-use-audit - static CDK argv, no shell
1406 cdk_cmd,
1407 **popen_kwargs,
1408 )
1409 with self._active_cdk_lock:
1410 self._active_cdk_processes[process.pid] = process
1411 if self._cdk_cancel_event.is_set(): 1411 ↛ 1412line 1411 didn't jump to line 1412 because the condition on line 1411 was never true
1412 self._terminate_cdk_process(process)
1413 with self._active_cdk_lock:
1414 self._active_cdk_processes.pop(process.pid, None)
1415 raise RuntimeError("CDK operation cancelled during process start")
1417 try:
1418 stdout, stderr = process.communicate(timeout=timeout)
1419 except subprocess.TimeoutExpired as exc:
1420 self._terminate_cdk_process(process)
1421 logger.warning(
1422 "cdk command timed out after %ss: %s",
1423 timeout,
1424 " ".join(cdk_cmd),
1425 )
1426 raise subprocess.TimeoutExpired(
1427 cdk_cmd,
1428 exc.timeout,
1429 output=exc.output,
1430 stderr=exc.stderr,
1431 ) from exc
1432 except BaseException:
1433 self._terminate_cdk_process(process)
1434 raise
1435 finally:
1436 with self._active_cdk_lock:
1437 self._active_cdk_processes.pop(process.pid, None)
1438 return subprocess.CompletedProcess(
1439 cdk_cmd,
1440 process.returncode,
1441 stdout=stdout or "",
1442 stderr=stderr or "",
1443 )
1445 def list_stacks(self) -> list[str]:
1446 """List all available CDK stacks."""
1447 result = self._run_cdk(["list"], capture_output=True)
1448 if result.returncode != 0:
1449 raise RuntimeError(f"Failed to list stacks: {result.stderr}")
1450 return [s.strip() for s in result.stdout.strip().split("\n") if s.strip()]
1452 def synth(self, stack_name: str | None = None, quiet: bool = True) -> str:
1453 """Synthesize CloudFormation templates from source-current assets."""
1454 self._ensure_lambda_build()
1455 cmd = ["synth"]
1456 if stack_name: 1456 ↛ 1458line 1456 didn't jump to line 1458 because the condition on line 1456 was always true
1457 cmd.append(stack_name)
1458 if quiet: 1458 ↛ 1461line 1458 didn't jump to line 1461 because the condition on line 1458 was always true
1459 cmd.append("--quiet")
1461 result = self._run_cdk(cmd, capture_output=True)
1462 if result.returncode != 0:
1463 raise RuntimeError(f"CDK synth failed: {result.stderr}")
1464 return str(result.stdout)
1466 def diff(self, stack_name: str | None = None) -> str:
1467 """Show diff between deployed and source-current local stacks."""
1468 self._ensure_lambda_build()
1469 cmd = ["diff", "--no-color"]
1470 if stack_name: 1470 ↛ 1473line 1470 didn't jump to line 1473 because the condition on line 1470 was always true
1471 cmd.append(stack_name)
1473 result = self._run_cdk(cmd, capture_output=True)
1474 # diff returns non-zero if there are differences, which is expected
1475 return str(result.stdout or result.stderr)
1477 def deploy(
1478 self,
1479 stack_name: str | None = None,
1480 require_approval: bool = True,
1481 all_stacks: bool = False,
1482 outputs_file: str | None = None,
1483 parameters: dict[str, str] | None = None,
1484 tags: dict[str, str] | None = None,
1485 progress: str = "events",
1486 output_dir: str | None = None,
1487 exclusively: bool = False,
1488 allow_bootstrap: bool = True,
1489 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
1490 expected_stack_ids: Mapping[str, str | None] | None = None,
1491 prepared_change_sets: PreparedChangeSetAuthority | None = None,
1492 authorize_stack: StackAuthorizationCallback | None = None,
1493 strict_deployment_token: str | None = None,
1494 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
1495 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
1496 ) -> bool:
1497 """Deploy CDK stacks.
1499 Args:
1500 stack_name: Name of the stack to deploy
1501 require_approval: Whether to require approval for changes
1502 all_stacks: Deploy all stacks
1503 outputs_file: File to write outputs to
1504 parameters: CDK parameters
1505 tags: Tags to apply to stacks
1506 progress: Progress display type
1507 output_dir: Custom CDK output directory (for parallel deployments)
1508 exclusively: Pass ``--exclusively`` to CDK so only the named
1509 stack is evaluated, not its transitive dependencies. Used by
1510 ``deploy_orchestrated`` once earlier phases have already
1511 deployed the globals — re-synthesizing them every phase
1512 forces custom resources (notably KubectlApplyManifests)
1513 to re-run each time, adding minutes per phase for no
1514 actual change.
1515 """
1516 # Synchronize canonical checked-in copies first, then source-check and
1517 # atomically publish only stale generated assets. A deploy must never
1518 # destructively rebuild a fresh tree while another CDK process may be
1519 # fingerprinting it.
1520 self._sync_lambda_sources()
1521 self._ensure_lambda_build()
1523 strict_deployment = (
1524 strict_deployment_token is not None or on_change_set_prepared is not None
1525 )
1526 expected_stack_id: str | None = None
1527 prepared_change_set_records: Mapping[str, Mapping[str, str]] = {}
1528 change_set_name: str | None = None
1529 if strict_deployment:
1530 if not stack_name or all_stacks: 1530 ↛ 1531line 1530 didn't jump to line 1531 because the condition on line 1530 was never true
1531 raise RuntimeError("Strict deployment requires exactly one named stack")
1532 if not strict_deployment_token or on_change_set_prepared is None: 1532 ↛ 1533line 1532 didn't jump to line 1533 because the condition on line 1532 was never true
1533 raise RuntimeError(
1534 "Strict deployment requires both a run token and a prepared-change-set callback"
1535 )
1536 if allow_bootstrap: 1536 ↛ 1537line 1536 didn't jump to line 1537 because the condition on line 1536 was never true
1537 raise RuntimeError("Strict deployment cannot auto-bootstrap a Region")
1538 if authorize_stack is None: 1538 ↛ 1539line 1538 didn't jump to line 1539 because the condition on line 1538 was never true
1539 raise RuntimeError("Strict deployment requires an exact stack authorizer")
1540 if expected_stack_ids is None or stack_name not in expected_stack_ids: 1540 ↛ 1541line 1540 didn't jump to line 1541 because the condition on line 1540 was never true
1541 raise RuntimeError(
1542 f"Strict deployment lacks authoritative target state for {stack_name}"
1543 )
1544 expected_stack_id = expected_stack_ids[stack_name]
1545 if prepared_change_sets is None or stack_name not in prepared_change_sets: 1545 ↛ 1546line 1545 didn't jump to line 1546 because the condition on line 1545 was never true
1546 raise RuntimeError(
1547 f"Strict deployment lacks prepared change-set history for {stack_name}"
1548 )
1549 prepared_change_set_records = prepared_change_sets[stack_name]
1550 change_set_name = self._strict_change_set_name(
1551 stack_name,
1552 strict_deployment_token,
1553 )
1555 # Validate bootstrap identity before any AWS mutation. In strict mode
1556 # this also revalidates the expected stack ARN (or authoritative
1557 # absence) before image mirroring or change-set preparation.
1558 if stack_name:
1559 region = self._get_deploy_region(stack_name)
1560 if not region: 1560 ↛ 1561line 1560 didn't jump to line 1561 because the condition on line 1560 was never true
1561 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
1562 if allow_bootstrap:
1563 if not self.ensure_bootstrapped(region):
1564 raise RuntimeError(
1565 f"Region {region} could not be bootstrapped. "
1566 "Run 'gco stacks bootstrap --region "
1567 f"{region}' manually to diagnose."
1568 )
1569 else:
1570 expected_bootstrap = (bootstrap_stacks or {}).get(region)
1571 if expected_bootstrap is None: 1571 ↛ 1572line 1571 didn't jump to line 1572 because the condition on line 1571 was never true
1572 raise RuntimeError(
1573 f"Strict deployment lacks a checkpointed CDKToolkit identity for {region}"
1574 )
1575 self._validate_bootstrap_stack(region, expected_bootstrap)
1576 if strict_deployment:
1577 target = self._describe_stack_target(
1578 stack_name,
1579 expected_stack_id=expected_stack_id,
1580 require_expected_identity=True,
1581 )
1582 if expected_stack_id is not None and target is None: 1582 ↛ 1583line 1582 didn't jump to line 1583 because the condition on line 1582 was never true
1583 raise RuntimeError(
1584 f"Checkpointed stack {expected_stack_id} is absent; refusing recreation"
1585 )
1586 assert change_set_name is not None
1587 self._preflight_strict_change_set(
1588 stack_name=stack_name,
1589 change_set_name=change_set_name,
1590 expected_stack_id=expected_stack_id,
1591 prepared_change_sets=prepared_change_set_records,
1592 )
1594 # Name-based stuck-stack recovery is intentionally disabled for strict
1595 # deployments. A prepared change set must establish CREATE-vs-UPDATE
1596 # authority without deleting or adopting anything by name.
1597 if stack_name and not strict_deployment:
1598 self._check_and_fix_stuck_stack(
1599 stack_name,
1600 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
1601 authorize_stack=authorize_stack,
1602 strict_ownership=not allow_bootstrap,
1603 )
1605 # Ensure container runtime is available for building images
1606 runtime = _detect_container_runtime()
1607 if not runtime:
1608 raise RuntimeError(
1609 "No container runtime found. Please install Docker, Finch, or Podman.\n"
1610 " - Docker: https://docs.docker.com/get-docker/\n"
1611 " - Finch: brew install finch && finch vm init\n"
1612 " - Podman: https://podman.io/getting-started/installation"
1613 )
1615 # Mirror third-party images into ECR only after strict bootstrap and
1616 # target checks. Repository creation acknowledgements are persisted
1617 # synchronously by the live-validation callback before any image copy.
1618 self._mirror_images_if_enabled(
1619 stack_name=stack_name,
1620 all_stacks=all_stacks,
1621 repository_tags=tags,
1622 on_repository_created=on_ecr_repository_created,
1623 )
1625 cmd = ["deploy"]
1627 if all_stacks:
1628 cmd.append("--all")
1629 elif stack_name: 1629 ↛ 1637line 1629 didn't jump to line 1637 because the condition on line 1629 was always true
1630 cmd.append(stack_name)
1632 # --exclusively tells CDK to deploy *only* the named stack, not its
1633 # transitive dependencies. deploy_orchestrated sets this once the
1634 # earlier phases (global, api-gateway) are already in place so that
1635 # the regional and monitoring phases don't re-synthesize and
1636 # re-evaluate globals on every pass.
1637 if exclusively and stack_name and not all_stacks:
1638 cmd.append("--exclusively")
1640 if strict_deployment:
1641 assert change_set_name is not None
1642 cmd.extend(
1643 [
1644 "--method",
1645 "prepare-change-set",
1646 "--change-set-name",
1647 change_set_name,
1648 "--context",
1649 f"{_LIVE_VALIDATION_PROVIDER_LOG_CONTEXT}=true",
1650 ]
1651 )
1653 if not require_approval:
1654 cmd.extend(["--require-approval", "never"])
1656 if outputs_file:
1657 cmd.extend(["--outputs-file", outputs_file])
1659 if parameters:
1660 for key, value in parameters.items():
1661 cmd.extend(["--parameters", f"{key}={value}"])
1663 if tags:
1664 for key, value in tags.items():
1665 cmd.extend(["--tags", f"{key}={value}"])
1667 cmd.extend(["--progress", progress])
1669 # Use custom output directory for parallel deployments
1670 if output_dir:
1671 cmd.extend(["--output", output_dir])
1673 # Set CDK_DOCKER env var if not already set
1674 env = {"CDK_DOCKER": runtime} if not os.environ.get("CDK_DOCKER") else None
1676 # Per-stack wall-clock cap so a wedged ``cdk deploy`` (e.g. an
1677 # IAM eventual-consistency wait that never completes) can't block
1678 # the orchestrator forever. Default 60 minutes — long enough for
1679 # a fresh EKS cluster cold start. Override via
1680 # GCO_CDK_DEPLOY_TIMEOUT_SECONDS.
1681 timeout_s = float(os.environ.get("GCO_CDK_DEPLOY_TIMEOUT_SECONDS", "3600"))
1683 # Timestamp (UTC) marking the start of this deploy attempt. The failure
1684 # reconciliation below uses it to tell a *fresh* CloudFormation
1685 # completion (cdk's client-side polling gave up just after CFN finished
1686 # — a real success) apart from a *stale* terminal state left by a
1687 # previous deploy (cdk failed before touching CloudFormation — a real
1688 # failure that must not be masked).
1689 deploy_start = datetime.now(UTC)
1691 try:
1692 result = self._run_cdk(cmd, env=env, timeout=timeout_s)
1693 success = result.returncode == 0
1694 except subprocess.TimeoutExpired:
1695 print(
1696 f" cdk deploy timed out after {timeout_s}s for "
1697 f"{stack_name or 'all stacks'}; verifying CloudFormation state..."
1698 )
1699 success = False
1701 if self._cdk_cancel_event.is_set(): 1701 ↛ 1702line 1701 didn't jump to line 1702 because the condition on line 1701 was never true
1702 raise RuntimeError("CDK deployment cancelled before AWS-side reconciliation")
1704 if strict_deployment:
1705 assert stack_name is not None
1706 assert change_set_name is not None
1707 assert on_change_set_prepared is not None
1708 try:
1709 success = self._execute_prepared_change_set(
1710 stack_name=stack_name,
1711 change_set_name=change_set_name,
1712 expected_stack_id=expected_stack_id,
1713 expected_tags=tags,
1714 prepared_change_sets=prepared_change_set_records,
1715 preparation_succeeded=success,
1716 authorize_stack=authorize_stack,
1717 on_change_set_prepared=on_change_set_prepared,
1718 allow_noop=success,
1719 timeout=timeout_s,
1720 )
1721 except Exception:
1722 self._diagnose_deploy_failure(stack_name)
1723 raise
1724 if not success: 1724 ↛ 1725line 1724 didn't jump to line 1725 because the condition on line 1724 was never true
1725 self._diagnose_deploy_failure(stack_name)
1727 if success and "analytics" in stack_name: 1727 ↛ 1728line 1727 didn't jump to line 1728 because the condition on line 1727 was never true
1728 api_gateway_stack = f"{self.config.project_name}-api-gateway"
1729 print(f" Updating {api_gateway_stack} with analytics routes...")
1730 success = self.deploy(
1731 stack_name=api_gateway_stack,
1732 require_approval=require_approval,
1733 outputs_file=outputs_file,
1734 parameters=parameters,
1735 tags=tags,
1736 progress=progress,
1737 exclusively=True,
1738 allow_bootstrap=allow_bootstrap,
1739 bootstrap_stacks=bootstrap_stacks,
1740 expected_stack_ids=expected_stack_ids,
1741 prepared_change_sets=prepared_change_sets,
1742 authorize_stack=authorize_stack,
1743 strict_deployment_token=(f"{strict_deployment_token}-analytics-routes"),
1744 on_change_set_prepared=on_change_set_prepared,
1745 on_ecr_repository_created=on_ecr_repository_created,
1746 )
1747 return success
1749 # Reconcile a cdk failure/timeout against CloudFormation. cdk's
1750 # client-side polling can give up (a transient ``read EADDRNOTAVAIL``
1751 # socket error, or our wall-clock timeout) while CloudFormation keeps
1752 # working server-side, so a non-zero exit does not always mean the
1753 # deploy failed. The trick is to reconcile without masking a *real*
1754 # failure by mistaking a stale terminal state for a fresh success.
1755 if stack_name and not all_stacks and not success:
1756 cfn_status = self._get_stack_status(stack_name)
1757 if cfn_status is not None and cfn_status.endswith("_IN_PROGRESS"):
1758 # CloudFormation is still mid-operation — observing that is
1759 # itself proof it ran an operation for this attempt. Wait for it
1760 # to settle and accept a terminal COMPLETE as a genuine success.
1761 print(
1762 f" cdk exited non-zero but {stack_name} is {cfn_status} in "
1763 "CloudFormation; waiting for the operation to settle..."
1764 )
1765 settled_status = self._wait_for_stack_settle(stack_name)
1766 if settled_status in ("CREATE_COMPLETE", "UPDATE_COMPLETE"):
1767 print(
1768 f" cdk reported a non-zero exit but {stack_name} settled "
1769 f"to {settled_status} in CloudFormation — treating as "
1770 "success."
1771 )
1772 success = True
1773 elif cfn_status in ("CREATE_COMPLETE", "UPDATE_COMPLETE"):
1774 # The stack is already terminal and CloudFormation is not
1775 # mid-flight. Two very different situations look identical on
1776 # status alone; only the stack's last-operation time tells them
1777 # apart:
1778 # * cdk's polling gave up just *after* CloudFormation finished
1779 # this attempt's operation — a genuine success whose
1780 # last-update time is newer than when we started.
1781 # * cdk failed *before* it ever touched CloudFormation (a
1782 # synth error, a cloud-assembly schema mismatch, an
1783 # asset/image build failure); the stack is merely sitting in
1784 # a *previous* deploy's COMPLETE state, whose last-update
1785 # time predates this attempt. Masking this is the
1786 # false-success bug this guards against.
1787 last_op = self._get_stack_last_update_time(stack_name)
1788 if last_op is not None and last_op >= deploy_start:
1789 print(
1790 f" cdk reported a non-zero exit but {stack_name} shows a "
1791 f"fresh {cfn_status} in CloudFormation — treating as "
1792 "success."
1793 )
1794 success = True
1795 else:
1796 print(
1797 f" cdk failed and {stack_name} is {cfn_status}, but no "
1798 "new CloudFormation operation ran for this attempt — cdk "
1799 "failed before touching CloudFormation. Treating as a "
1800 "failed deploy."
1801 )
1803 # Conversely, when cdk reports success, confirm CloudFormation actually
1804 # landed in a terminal success state. A zero cdk exit can still mask a
1805 # stack that silently rolled back (e.g. UPDATE_ROLLBACK_COMPLETE) or is
1806 # otherwise not in a healthy COMPLETE state — verifying the AWS-side
1807 # truth keeps deploy() from reporting a rolled-back stack as deployed.
1808 # A None status (lookup failed / transient) leaves cdk's verdict intact;
1809 # we only override on a *known* non-success state. No-op deploys stay in
1810 # CREATE_COMPLETE/UPDATE_COMPLETE, so this never false-fails them — we
1811 # deliberately don't require LastUpdatedTime to advance.
1812 if success and stack_name and not all_stacks:
1813 cfn_status = self._get_stack_status(stack_name)
1814 if cfn_status is not None and cfn_status not in (
1815 "CREATE_COMPLETE",
1816 "UPDATE_COMPLETE",
1817 ):
1818 print(
1819 f" cdk reported success but {stack_name} is in {cfn_status} "
1820 f"in CloudFormation — treating as a failed deploy."
1821 )
1822 success = False
1824 if not success and stack_name:
1825 self._diagnose_deploy_failure(stack_name)
1827 # After deploying gco-analytics, automatically redeploy
1828 # gco-api-gateway to wire in the /studio/* routes (the API gateway
1829 # imports the Cognito pool ARN and presigned-URL Lambda ARN from
1830 # the analytics stack).
1831 if success and stack_name and "analytics" in stack_name and not all_stacks:
1832 api_gateway_stack = f"{self.config.project_name}-api-gateway"
1833 print(f" Updating {api_gateway_stack} with analytics routes...")
1834 success = self.deploy(
1835 stack_name=api_gateway_stack,
1836 require_approval=require_approval,
1837 outputs_file=outputs_file,
1838 parameters=parameters,
1839 tags=tags,
1840 progress=progress,
1841 exclusively=True,
1842 allow_bootstrap=allow_bootstrap,
1843 bootstrap_stacks=bootstrap_stacks,
1844 expected_stack_ids=expected_stack_ids,
1845 authorize_stack=authorize_stack,
1846 on_ecr_repository_created=on_ecr_repository_created,
1847 )
1849 return success
1851 def destroy(
1852 self,
1853 stack_name: str | None = None,
1854 all_stacks: bool = False,
1855 force: bool = False,
1856 output_dir: str | None = None,
1857 expected_stack_id: str | None = None,
1858 expected_stack_ids: Mapping[str, str | None] | None = None,
1859 prepared_change_sets: PreparedChangeSetAuthority | None = None,
1860 authorize_stack: StackAuthorizationCallback | None = None,
1861 allow_bootstrap: bool = True,
1862 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
1863 strict_deployment_token: str | None = None,
1864 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
1865 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
1866 ) -> bool:
1867 """Destroy stacks while restoring any temporary config mutation exactly."""
1868 config_path: Path | None = None
1869 original_bytes: bytes | None = None
1870 original_mode: int | None = None
1871 if stack_name and not all_stacks and "analytics" in stack_name:
1872 config_path = _find_cdk_json()
1873 if config_path is None: 1873 ↛ 1874line 1873 didn't jump to line 1874 because the condition on line 1873 was never true
1874 raise RuntimeError("cdk.json not found before analytics destroy")
1875 original_bytes = config_path.read_bytes()
1876 original_mode = stat.S_IMODE(config_path.stat().st_mode)
1877 try:
1878 return self._destroy(
1879 stack_name=stack_name,
1880 all_stacks=all_stacks,
1881 force=force,
1882 output_dir=output_dir,
1883 expected_stack_id=expected_stack_id,
1884 expected_stack_ids=expected_stack_ids,
1885 prepared_change_sets=prepared_change_sets,
1886 authorize_stack=authorize_stack,
1887 allow_bootstrap=allow_bootstrap,
1888 bootstrap_stacks=bootstrap_stacks,
1889 strict_deployment_token=strict_deployment_token,
1890 on_change_set_prepared=on_change_set_prepared,
1891 on_ecr_repository_created=on_ecr_repository_created,
1892 )
1893 finally:
1894 if config_path is not None and original_bytes is not None:
1895 _atomic_write_bytes(config_path, original_bytes, mode=original_mode)
1897 def _destroy(
1898 self,
1899 stack_name: str | None = None,
1900 all_stacks: bool = False,
1901 force: bool = False,
1902 output_dir: str | None = None,
1903 expected_stack_id: str | None = None,
1904 expected_stack_ids: Mapping[str, str | None] | None = None,
1905 prepared_change_sets: PreparedChangeSetAuthority | None = None,
1906 authorize_stack: StackAuthorizationCallback | None = None,
1907 allow_bootstrap: bool = True,
1908 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
1909 strict_deployment_token: str | None = None,
1910 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
1911 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
1912 ) -> bool:
1913 """Destroy CDK stacks.
1915 If the target stack exists in CloudFormation but isn't in the CDK
1916 app (e.g. because a toggle was disabled), temporarily enables the
1917 toggle so CDK can synthesize and destroy the stack properly. This
1918 ensures custom resource cleanup handlers (like the analytics
1919 cleanup Lambda) fire during deletion.
1921 Args:
1922 stack_name: Name of the stack to destroy
1923 all_stacks: Destroy all stacks
1924 force: Skip confirmation prompts
1925 output_dir: Custom CDK output directory (for parallel deployments)
1926 """
1927 if all_stacks and (expected_stack_id is not None or expected_stack_ids is not None):
1928 raise RuntimeError("Identity-fenced teardown cannot use all_stacks=True")
1929 if stack_name is None and (expected_stack_id is not None or expected_stack_ids is not None): 1929 ↛ 1930line 1929 didn't jump to line 1930 because the condition on line 1929 was never true
1930 raise RuntimeError("Identity-fenced teardown requires exactly one named stack")
1932 strict_identity = expected_stack_id is not None or expected_stack_ids is not None
1933 if stack_name is not None and expected_stack_ids is not None:
1934 if stack_name not in expected_stack_ids: 1934 ↛ 1935line 1934 didn't jump to line 1935 because the condition on line 1934 was never true
1935 raise RuntimeError(
1936 f"Strict teardown lacks authoritative target state for {stack_name}"
1937 )
1938 mapped_stack_id = expected_stack_ids[stack_name]
1939 if expected_stack_id is not None and expected_stack_id != mapped_stack_id: 1939 ↛ 1940line 1939 didn't jump to line 1940 because the condition on line 1939 was never true
1940 raise RuntimeError(f"Conflicting expected stack identities for {stack_name}")
1941 expected_stack_id = mapped_stack_id
1942 if strict_identity and authorize_stack is None: 1942 ↛ 1943line 1942 didn't jump to line 1943 because the condition on line 1942 was never true
1943 raise RuntimeError("Identity-fenced teardown requires an exact stack authorizer")
1945 # Image-registry pre-destroy guards. Only fires for the global
1946 # stack (where the registry lives) and only when the operator
1947 # has explicitly chosen ``removal_policy: "destroy"``. The
1948 # default ``retain`` posture is a no-op here. See
1949 # ``_image_registry_destroy_preflight`` for the exact rules.
1950 if (
1951 stack_name is not None
1952 and stack_name.endswith("-global")
1953 and not all_stacks
1954 and not self._image_registry_destroy_preflight(force=force)
1955 ):
1956 return False
1958 # Strict callers never enter CDK's name-based destroy or toggle-based
1959 # recovery paths. Analytics may first require one strict prepared
1960 # change set on the exact API stack to remove cross-stack imports.
1961 if strict_identity and stack_name and not all_stacks:
1962 if self._cdk_cancel_event.is_set(): 1962 ↛ 1963line 1962 didn't jump to line 1963 because the condition on line 1962 was never true
1963 raise RuntimeError(f"Strict teardown cancelled before deleting {stack_name}")
1964 if "analytics" in stack_name:
1965 if expected_stack_ids is None:
1966 raise RuntimeError(
1967 "Identity-fenced analytics teardown requires the complete expected "
1968 "stack identity map"
1969 )
1970 safe_to_destroy = self._remove_api_gateway_analytics_dependency(
1971 allow_bootstrap=allow_bootstrap,
1972 bootstrap_stacks=bootstrap_stacks,
1973 expected_stack_ids=expected_stack_ids,
1974 prepared_change_sets=prepared_change_sets,
1975 authorize_stack=authorize_stack,
1976 strict_deployment_token=(
1977 f"{strict_deployment_token}-drop-analytics-routes"
1978 if strict_deployment_token is not None
1979 else None
1980 ),
1981 on_change_set_prepared=on_change_set_prepared,
1982 on_ecr_repository_created=on_ecr_repository_created,
1983 )
1984 if not safe_to_destroy: 1984 ↛ 1985line 1984 didn't jump to line 1985 because the condition on line 1984 was never true
1985 return False
1986 return self._cloudformation_delete_stack(
1987 stack_name,
1988 expected_stack_id=expected_stack_id,
1989 authorize_stack=authorize_stack,
1990 require_expected_identity=True,
1991 )
1993 # A regional API bridge disappears from the CDK app when its Region is
1994 # removed from configuration. Only this exact project-scoped shape with
1995 # an SDK-known CloudFormation Region may bypass CDK; configured bridges,
1996 # arbitrary suffixes, and every other stack keep the normal CDK path.
1997 if stack_name and not all_stacks:
1998 orphan_region = self._get_orphan_regional_api_region(stack_name)
1999 if orphan_region is not None:
2000 if not self._stack_exists_in_cloudformation(stack_name): 2000 ↛ 2001line 2000 didn't jump to line 2001 because the condition on line 2000 was never true
2001 return True
2002 print(
2003 f" {stack_name} is absent from the configured CDK app; "
2004 f"deleting it directly in {orphan_region}..."
2005 )
2006 return self._cloudformation_delete_stack(
2007 stack_name,
2008 expected_stack_id=expected_stack_id,
2009 authorize_stack=authorize_stack,
2010 )
2012 # If destroying a specific stack that exists in CloudFormation but
2013 # might not be in the CDK app, temporarily enable its toggle.
2014 toggle_restored = False
2015 if (
2016 stack_name
2017 and not all_stacks
2018 and "analytics" in stack_name
2019 and self._stack_exists_in_cloudformation(stack_name)
2020 ):
2021 toggle_restored = self._ensure_analytics_enabled_for_destroy()
2023 # The analytics stack exports values (e.g. Cognito pool ARN) that
2024 # gco-api-gateway imports. CloudFormation blocks deletion of stacks
2025 # with consumed exports. To break the dependency, redeploy the API
2026 # gateway with analytics disabled first, then destroy analytics.
2027 if stack_name and not all_stacks and "analytics" in stack_name:
2028 safe_to_destroy = self._remove_api_gateway_analytics_dependency(
2029 allow_bootstrap=allow_bootstrap,
2030 bootstrap_stacks=bootstrap_stacks,
2031 expected_stack_ids=expected_stack_ids,
2032 prepared_change_sets=prepared_change_sets,
2033 authorize_stack=authorize_stack,
2034 strict_deployment_token=strict_deployment_token,
2035 on_change_set_prepared=on_change_set_prepared,
2036 on_ecr_repository_created=on_ecr_repository_created,
2037 )
2038 if not safe_to_destroy:
2039 # Restore analytics toggle before bailing out.
2040 if toggle_restored: 2040 ↛ 2041line 2040 didn't jump to line 2041 because the condition on line 2040 was never true
2041 self._restore_analytics_disabled()
2042 project = self.config.project_name
2043 print(
2044 f" Aborting {project}-analytics destroy: {project}-api-gateway "
2045 "still imports analytics exports. Fix the API gateway and retry."
2046 )
2047 return False
2049 # Non-strict callers may still use CDK's name-based path. Strict calls
2050 # returned above after exact-ARN deletion.
2051 cmd = ["destroy"]
2053 if all_stacks:
2054 cmd.append("--all")
2055 elif stack_name: 2055 ↛ 2063line 2055 didn't jump to line 2063 because the condition on line 2055 was always true
2056 cmd.append(stack_name)
2057 # --exclusively prevents CDK from cascading the destroy to
2058 # dependent stacks (e.g. destroying gco-analytics should not
2059 # also destroy gco-api-gateway just because it references the
2060 # presigned-URL Lambda ARN).
2061 cmd.append("--exclusively")
2063 if force:
2064 cmd.append("--force")
2066 if output_dir:
2067 cmd.extend(["--output", output_dir])
2069 # Per-stack wall-clock cap so a wedged ``cdk destroy`` (its
2070 # post-delete polling loop hanging after CloudFormation has
2071 # already finished) can't block the orchestrator forever. Default
2072 # 90 minutes: a healthy EKS regional teardown has been observed
2073 # needing ~60 (the VPC Lambda ENI detach alone can serialise for
2074 # 20+ while CloudFormation keeps making progress), and the prior
2075 # 45-minute cap killed the poller mid-delete — the AWS-side
2076 # reconciliation below recovered, but the timeout should mark a
2077 # wedged CDK, not a normal teardown. Override via
2078 # GCO_CDK_DESTROY_TIMEOUT_SECONDS.
2079 timeout_s = float(os.environ.get("GCO_CDK_DESTROY_TIMEOUT_SECONDS", "5400"))
2081 try:
2082 result = self._run_cdk(cmd, timeout=timeout_s)
2083 cdk_succeeded = result.returncode == 0
2084 except subprocess.TimeoutExpired:
2085 # CDK hung. Verify the AWS-side state below — if the stack
2086 # is gone in CloudFormation, the destroy actually succeeded
2087 # and the timeout was just CDK's polling loop wedged.
2088 print(
2089 f" cdk destroy timed out after {timeout_s}s; verifying "
2090 f"CloudFormation state for {stack_name}..."
2091 )
2092 cdk_succeeded = False
2094 if self._cdk_cancel_event.is_set(): 2094 ↛ 2095line 2094 didn't jump to line 2095 because the condition on line 2094 was never true
2095 raise RuntimeError("CDK teardown cancelled before AWS-side reconciliation")
2097 # Restore the toggle if we changed it
2098 if toggle_restored:
2099 self._restore_analytics_disabled()
2101 # Reconcile against CloudFormation. A local CDK timeout/failure is not
2102 # an AWS failure when the delete operation is still healthy. Once AWS
2103 # reports DELETE_IN_PROGRESS, wait for bounded server-side convergence
2104 # instead of letting the orchestrator advance into dependent stacks.
2105 if stack_name and not all_stacks:
2106 still_present = self._stack_exists_in_cloudformation(stack_name)
2107 if not still_present:
2108 if not cdk_succeeded:
2109 print(
2110 f" cdk reported a non-zero exit but {stack_name} is "
2111 "already deleted in CloudFormation — treating as success."
2112 )
2113 return True
2115 status = self._get_stack_status(stack_name, expected_stack_id)
2116 if status == "DELETE_IN_PROGRESS":
2117 return self._wait_for_stack_delete_convergence(
2118 stack_name,
2119 initial_status=status,
2120 )
2121 if status == "DELETE_FAILED":
2122 self._print_stack_delete_heartbeat(
2123 stack_name,
2124 status,
2125 expected_stack_id,
2126 )
2127 print(f" {stack_name} reached DELETE_FAILED; refusing to continue teardown.")
2128 return False
2130 # A zero CDK exit with a still-present, non-deleting stack is a rare
2131 # client-side false success. Start deletion directly and then use
2132 # the same bounded convergence loop. A non-zero exit in any other
2133 # state means CDK failed before it started a delete operation.
2134 if cdk_succeeded:
2135 return self._cloudformation_delete_stack(stack_name)
2136 print(
2137 f" cdk failed and {stack_name} is still {status or 'in an unknown state'}; "
2138 "no active CloudFormation delete operation was confirmed."
2139 )
2140 return False
2142 return cdk_succeeded
2144 # ------------------------------------------------------------------
2145 # Image registry pre-destroy guards
2146 # ------------------------------------------------------------------
2147 def _read_images_config(self) -> dict[str, Any]:
2148 """Read the ``images`` block from cdk.json with defaults applied.
2150 Mirrors the parser in ``gco/stacks/global_stack.py`` so the CLI
2151 can reason about the same fields without importing the CDK
2152 module (which pulls aws_cdk and the full constructs surface).
2153 Defaults stay aligned with the global-stack parser; any value
2154 that fails validation (e.g. an unexpected ``removal_policy``)
2155 is silently coerced to ``"retain"`` here so the CLI never blocks
2156 on a typo — the actual deploy-time validation is the global
2157 stack's responsibility.
2158 """
2159 import json
2161 cdk_json_path = _find_cdk_json()
2162 if not cdk_json_path:
2163 return {
2164 "removal_policy": "retain",
2165 "empty_on_delete": False,
2166 }
2167 try:
2168 with open(cdk_json_path, encoding="utf-8") as f:
2169 ctx = json.load(f).get("context", {}) or {}
2170 except (OSError, json.JSONDecodeError) as exc:
2171 logger.debug("Failed to read cdk.json for images config: %s", exc)
2172 return {"removal_policy": "retain", "empty_on_delete": False}
2174 raw = ctx.get("images") or {}
2175 removal_policy = str(raw.get("removal_policy", "retain")).strip().lower()
2176 if removal_policy not in ("retain", "destroy"):
2177 removal_policy = "retain"
2178 return {
2179 "removal_policy": removal_policy,
2180 "empty_on_delete": bool(raw.get("empty_on_delete", False)),
2181 }
2183 def _build_image_registry_inventory(self) -> dict[str, Any]:
2184 """Aggregate repo / tag / size / reference counts for the registry.
2186 Returns a dict shape suitable for printing to the operator. Best
2187 effort: a missing ImageManager dependency or an AWS error
2188 produces a partially-populated dict rather than raising.
2189 """
2190 inventory: dict[str, Any] = {
2191 "repo_count": 0,
2192 "tag_count": 0,
2193 "total_bytes": 0,
2194 "endpoint_refs": 0,
2195 "job_refs": 0,
2196 }
2197 try:
2198 from cli.images import ImageManager
2199 except Exception as exc: # noqa: BLE001
2200 logger.debug("ImageManager import failed during preflight: %s", exc)
2201 return inventory
2203 try:
2204 manager = ImageManager(config=self.config)
2205 repos = manager.list_repos()
2206 inventory["repo_count"] = len(repos)
2207 # Repos this deployment owns live under ``<project_name>/`` (#139).
2208 repo_prefix = f"{self.config.project_name}/"
2209 for repo in repos:
2210 repo_name = repo.get("name", "")
2211 if not repo_name.startswith(repo_prefix):
2212 continue
2213 short = repo_name.removeprefix(repo_prefix)
2214 try:
2215 tags = manager.list_tags(short)
2216 except Exception as exc: # noqa: BLE001
2217 logger.debug("list_tags failed for %s: %s", repo_name, exc)
2218 continue
2219 inventory["tag_count"] += len(tags)
2220 for row in tags:
2221 size = row.get("size_bytes")
2222 if isinstance(size, int): 2222 ↛ 2220line 2222 didn't jump to line 2220 because the condition on line 2222 was always true
2223 inventory["total_bytes"] += size
2224 try:
2225 inventory["endpoint_refs"] = len(manager._collect_inference_image_refs())
2226 except Exception as exc: # noqa: BLE001
2227 logger.debug("inference ref collection failed: %s", exc)
2228 try:
2229 inventory["job_refs"] = len(manager._collect_recent_job_image_refs())
2230 except Exception as exc: # noqa: BLE001
2231 logger.debug("job ref collection failed: %s", exc)
2232 except Exception as exc: # noqa: BLE001
2233 logger.debug("Image registry inventory failed: %s", exc)
2234 return inventory
2236 def _image_registry_destroy_preflight(self, *, force: bool) -> bool:
2237 """Validate the image-registry destroy posture before invoking CFN.
2239 Two rules:
2241 1. ``removal_policy: "destroy"`` AND ``empty_on_delete: false``
2242 → refuse with the literal helpful-error message pointing
2243 the operator at ``gco images cleanup --all`` or at flipping
2244 ``empty_on_delete: true``.
2246 2. ``removal_policy: "destroy"`` AND ``empty_on_delete: true``
2247 → print the inventory summary first. On a TTY the operator
2248 is also prompted for confirmation; non-TTY runs proceed
2249 (the operator presumably passed ``-y`` or is automating).
2251 Returns True when the destroy may proceed, False when it has
2252 been refused or declined.
2253 """
2254 cfg = self._read_images_config()
2255 if cfg["removal_policy"] != "destroy":
2256 return True
2258 if not cfg["empty_on_delete"]:
2259 print(
2260 f"Repos under {self.config.project_name}/* are not empty and "
2261 "empty_on_delete is false. Run 'gco images cleanup --all' "
2262 "first, or set images.empty_on_delete: true in cdk.json."
2263 )
2264 return False
2266 inventory = self._build_image_registry_inventory()
2267 gib = inventory["total_bytes"] / (1024**3) if inventory["total_bytes"] else 0.0
2268 print("Image registry inventory before destroy:")
2269 print(f" repos: {inventory['repo_count']}")
2270 print(f" tags: {inventory['tag_count']}")
2271 print(f" total size: {gib:.2f} GiB")
2272 print(f" referencing endpoints: {inventory['endpoint_refs']}")
2273 print(f" recent job refs: {inventory['job_refs']}")
2275 # Already confirmed via -y, or non-interactive — proceed.
2276 if force or not sys.stdin.isatty():
2277 return True
2279 try:
2280 response = input(
2281 f"Destroy {self.config.project_name}-global and delete every "
2282 f"{self.config.project_name}/* repo? [y/N]: "
2283 )
2284 except EOFError, KeyboardInterrupt:
2285 print("Aborted.")
2286 return False
2287 if response.strip().lower() not in ("y", "yes"):
2288 print("Aborted.")
2289 return False
2290 return True
2292 @staticmethod
2293 def _stack_missing(exc: ClientError) -> bool:
2294 error = exc.response.get("Error", {})
2295 return bool(
2296 error.get("Code") == "ValidationError"
2297 and "does not exist" in str(error.get("Message", "")).lower()
2298 )
2300 @staticmethod
2301 def _change_set_missing(exc: ClientError) -> bool:
2302 """Return whether CloudFormation authoritatively reports an absent change set."""
2303 return bool(exc.response.get("Error", {}).get("Code") == "ChangeSetNotFound")
2305 def _describe_stack_target(
2306 self,
2307 stack_name: str,
2308 *,
2309 expected_stack_id: str | None = None,
2310 require_expected_identity: bool = False,
2311 ) -> tuple[str, Any, dict[str, Any]] | None:
2312 """Resolve live/absent/tombstone/replacement state without name adoption."""
2313 import boto3
2315 region = self._get_destroy_region(stack_name)
2316 cfn = boto3.client("cloudformation", region_name=region)
2318 def describe(identifier: str) -> dict[str, Any] | None:
2319 try:
2320 response = cfn.describe_stacks(StackName=identifier)
2321 except ClientError as exc:
2322 if self._stack_missing(exc):
2323 return None
2324 raise
2325 stacks = response.get("Stacks", [])
2326 if len(stacks) != 1: 2326 ↛ 2327line 2326 didn't jump to line 2327 because the condition on line 2326 was never true
2327 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2328 stack = stacks[0]
2329 if not isinstance(stack, dict): 2329 ↛ 2330line 2329 didn't jump to line 2330 because the condition on line 2329 was never true
2330 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2331 stack_id = str(stack.get("StackId") or "")
2332 if stack.get("StackName") != stack_name or not stack_id: 2332 ↛ 2333line 2332 didn't jump to line 2333 because the condition on line 2332 was never true
2333 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2334 return stack
2336 exact = describe(expected_stack_id) if expected_stack_id else None
2337 if exact is not None and str(exact.get("StackStatus") or "") != "DELETE_COMPLETE":
2338 if str(exact.get("StackId") or "") != expected_stack_id: 2338 ↛ 2339line 2338 didn't jump to line 2339 because the condition on line 2338 was never true
2339 raise RuntimeError(f"Stack identity changed for {region}:{stack_name}")
2340 return region, cfn, exact
2342 by_name = describe(stack_name)
2343 if by_name is None or str(by_name.get("StackStatus") or "") == "DELETE_COMPLETE":
2344 return None
2345 actual_id = str(by_name.get("StackId") or "")
2346 if expected_stack_id is not None and actual_id != expected_stack_id:
2347 raise RuntimeError(
2348 f"Checkpointed stack {expected_stack_id} is absent or deleted but same-name "
2349 f"replacement {actual_id} exists; refusing adoption"
2350 )
2351 if expected_stack_id is None and require_expected_identity:
2352 raise RuntimeError(
2353 f"Refusing name-authorized access to uncheckpointed stack "
2354 f"{region}:{stack_name} ({actual_id})"
2355 )
2356 return region, cfn, by_name
2358 def _stack_exists_in_cloudformation(
2359 self,
2360 stack_name: str,
2361 expected_stack_id: str | None = None,
2362 *,
2363 require_expected_identity: bool = False,
2364 ) -> bool:
2365 """Return whether the exact live target exists, rejecting replacements."""
2366 target = self._describe_stack_target(
2367 stack_name,
2368 expected_stack_id=expected_stack_id,
2369 require_expected_identity=require_expected_identity,
2370 )
2371 return target is not None
2373 def _get_stack_status(
2374 self,
2375 stack_name: str,
2376 stack_identifier: str | None = None,
2377 ) -> str | None:
2378 """Return the live CloudFormation status of ``stack_name`` or None.
2380 Used by ``deploy()`` to reconcile against AWS-side state when ``cdk
2381 deploy`` returns a non-zero exit code or times out — if the stack
2382 actually finished CREATE_COMPLETE or UPDATE_COMPLETE on the AWS
2383 side, the deploy succeeded regardless of what cdk reported.
2384 Returns None when the stack does not exist or the lookup itself
2385 fails (network blip, perms, etc.) so callers can treat the
2386 unknown case as 'cdk's verdict stands'.
2387 """
2388 import boto3
2390 try:
2391 region = self._get_destroy_region(stack_name)
2392 cfn = boto3.client("cloudformation", region_name=region)
2393 resp = cfn.describe_stacks(StackName=stack_identifier or stack_name)
2394 return str(resp["Stacks"][0]["StackStatus"])
2395 except Exception:
2396 return None
2398 def _get_stack_last_update_time(self, stack_name: str) -> datetime | None:
2399 """Return the UTC time of ``stack_name``'s most recent CloudFormation
2400 operation, or None if the stack is absent or the lookup fails.
2402 Uses ``LastUpdatedTime`` when the stack has been updated at least once,
2403 falling back to ``CreationTime`` for a stack that has only ever been
2404 created. ``deploy()`` compares this against the moment the deploy
2405 attempt started to decide whether a cdk failure/timeout that leaves the
2406 stack ``*_COMPLETE`` reflects a *fresh* operation (cdk's polling merely
2407 gave up early — success) or a *stale* one left by a previous deploy
2408 (cdk failed before touching CloudFormation — a real failure). A None
2409 return keeps the conservative 'cdk's failure stands' verdict.
2410 """
2411 import boto3
2413 try:
2414 region = self._get_destroy_region(stack_name)
2415 cfn = boto3.client("cloudformation", region_name=region)
2416 resp = cfn.describe_stacks(StackName=stack_name)
2417 stack = resp["Stacks"][0]
2418 last_op = stack.get("LastUpdatedTime") or stack.get("CreationTime")
2419 return last_op if isinstance(last_op, datetime) else None
2420 except Exception:
2421 return None
2423 def _wait_for_stack_settle(
2424 self,
2425 stack_name: str,
2426 timeout: float | None = None,
2427 stack_identifier: str | None = None,
2428 ) -> str | None:
2429 """Poll CloudFormation until ``stack_name`` leaves a ``*_IN_PROGRESS`` state.
2431 When ``cdk deploy`` dies on a transient client-side error (e.g. a
2432 ``read EADDRNOTAVAIL`` socket failure) the CloudFormation operation it
2433 started usually keeps running server-side. ``deploy()`` reconciles
2434 against CloudFormation, but a single status read taken the instant cdk
2435 exits can catch the stack mid-flight (``CREATE_IN_PROGRESS``) and give
2436 up only seconds before it would have reached ``CREATE_COMPLETE``. This
2437 helper waits out the in-progress window so the reconcile judges the
2438 *terminal* state instead of a transient one.
2440 Returns the terminal status string, the last status seen on timeout, or
2441 ``None`` if the status could not be read (so callers treat the unknown
2442 case as 'cdk's verdict stands').
2443 """
2444 import time
2446 if timeout is None:
2447 timeout = float(os.environ.get("GCO_CDK_SETTLE_TIMEOUT_SECONDS", "1200"))
2448 deadline = time.monotonic() + timeout
2449 status = self._get_stack_status(stack_name, stack_identifier)
2450 while status is not None and status.endswith("_IN_PROGRESS"):
2451 if self._cdk_cancel_event.is_set(): 2451 ↛ 2452line 2451 didn't jump to line 2452 because the condition on line 2451 was never true
2452 return status
2453 if time.monotonic() >= deadline:
2454 break
2455 time.sleep(15.0)
2456 status = self._get_stack_status(stack_name, stack_identifier)
2457 return status
2459 def _get_latest_stack_event(
2460 self,
2461 stack_name: str,
2462 stack_identifier: str | None = None,
2463 ) -> dict[str, Any] | None:
2464 """Return the newest CloudFormation event for delete heartbeats."""
2465 import boto3
2467 try:
2468 region = self._get_destroy_region(stack_name)
2469 cfn = boto3.client("cloudformation", region_name=region)
2470 events = cfn.describe_stack_events(StackName=stack_identifier or stack_name).get(
2471 "StackEvents", []
2472 )
2473 return events[0] if events else None
2474 except Exception:
2475 logger.debug("Could not read delete events for %s", stack_name, exc_info=True)
2476 return None
2478 def _print_stack_delete_heartbeat(
2479 self,
2480 stack_name: str,
2481 status: str | None,
2482 stack_identifier: str | None = None,
2483 ) -> None:
2484 """Print the latest AWS-side state while a long delete converges."""
2485 event = self._get_latest_stack_event(stack_name, stack_identifier)
2486 if not event: 2486 ↛ 2487line 2486 didn't jump to line 2487 because the condition on line 2486 was never true
2487 print(f" {stack_name}: CloudFormation status {status or 'unknown'}")
2488 return
2490 timestamp = event.get("Timestamp")
2491 timestamp_text = (
2492 timestamp.isoformat() if isinstance(timestamp, datetime) else str(timestamp or "")
2493 )
2494 logical_id = str(event.get("LogicalResourceId") or stack_name)
2495 resource_status = str(event.get("ResourceStatus") or status or "unknown")
2496 reason = " ".join(str(event.get("ResourceStatusReason") or "").split())
2497 if len(reason) > 400: 2497 ↛ 2498line 2497 didn't jump to line 2498 because the condition on line 2497 was never true
2498 reason = reason[:397] + "..."
2499 suffix = f" — {reason}" if reason else ""
2500 print(
2501 f" {stack_name}: {status or 'unknown'}; latest event "
2502 f"{timestamp_text} {logical_id} {resource_status}{suffix}"
2503 )
2505 def _wait_for_stack_delete_convergence(
2506 self,
2507 stack_name: str,
2508 *,
2509 timeout: float | None = None,
2510 poll_interval: float = _CLOUDFORMATION_DELETE_POLL_SECONDS,
2511 heartbeat_interval: float = _CLOUDFORMATION_DELETE_HEARTBEAT_SECONDS,
2512 initial_status: str = "DELETE_IN_PROGRESS",
2513 expected_stack_id: str | None = None,
2514 require_expected_identity: bool = False,
2515 ) -> bool:
2516 """Wait for an AWS-side stack delete to finish without trusting CDK polling.
2518 The caller must already have evidence that a delete operation started.
2519 Transient status-read failures are tolerated after that proof, but a
2520 terminal ``DELETE_FAILED`` or the overall deadline fails closed.
2521 """
2522 if timeout is None: 2522 ↛ 2523line 2522 didn't jump to line 2523 because the condition on line 2522 was never true
2523 try:
2524 timeout = float(
2525 os.environ.get(
2526 "GCO_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS",
2527 str(_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS),
2528 )
2529 )
2530 except ValueError:
2531 timeout = _CLOUDFORMATION_DELETE_TIMEOUT_SECONDS
2532 if not math.isfinite(timeout) or timeout <= 0:
2533 raise ValueError("CloudFormation delete timeout must be positive and finite")
2534 if (
2535 not math.isfinite(poll_interval)
2536 or not math.isfinite(heartbeat_interval)
2537 or poll_interval <= 0
2538 or heartbeat_interval <= 0
2539 ):
2540 raise ValueError("CloudFormation delete polling intervals must be positive and finite")
2542 deadline = time.monotonic() + timeout
2543 next_heartbeat = time.monotonic()
2544 status: str | None = initial_status
2545 last_printed_status: str | None = None
2547 while True:
2548 if self._cdk_cancel_event.is_set(): 2548 ↛ 2549line 2548 didn't jump to line 2549 because the condition on line 2548 was never true
2549 logger.warning("CloudFormation delete wait cancelled for %s", stack_name)
2550 return False
2551 try:
2552 if not self._stack_exists_in_cloudformation( 2552 ↛ 2557line 2552 didn't jump to line 2557 because the condition on line 2552 was never true
2553 stack_name,
2554 expected_stack_id=expected_stack_id,
2555 require_expected_identity=require_expected_identity,
2556 ):
2557 print(f" {stack_name} is absent from CloudFormation.")
2558 return True
2559 except RuntimeError:
2560 raise
2561 except Exception:
2562 logger.debug(
2563 "CloudFormation presence check failed for %s",
2564 stack_name,
2565 exc_info=True,
2566 )
2568 now = time.monotonic()
2569 if status == "DELETE_COMPLETE": 2569 ↛ 2570line 2569 didn't jump to line 2570 because the condition on line 2569 was never true
2570 return True
2571 if status == "DELETE_FAILED": 2571 ↛ 2572line 2571 didn't jump to line 2572 because the condition on line 2571 was never true
2572 self._print_stack_delete_heartbeat(
2573 stack_name,
2574 status,
2575 expected_stack_id,
2576 )
2577 return False
2578 if status not in (None, "DELETE_IN_PROGRESS"): 2578 ↛ 2579line 2578 didn't jump to line 2579 because the condition on line 2578 was never true
2579 self._print_stack_delete_heartbeat(
2580 stack_name,
2581 status,
2582 expected_stack_id,
2583 )
2584 print(
2585 f" {stack_name} left DELETE_IN_PROGRESS without being deleted; "
2586 "refusing to continue teardown."
2587 )
2588 return False
2589 if now >= deadline: 2589 ↛ 2600line 2589 didn't jump to line 2600 because the condition on line 2589 was always true
2590 self._print_stack_delete_heartbeat(
2591 stack_name,
2592 status,
2593 expected_stack_id,
2594 )
2595 print(
2596 f" Timed out after {timeout:.0f}s waiting for {stack_name} "
2597 "to disappear from CloudFormation."
2598 )
2599 return False
2600 if status != last_printed_status or now >= next_heartbeat:
2601 self._print_stack_delete_heartbeat(
2602 stack_name,
2603 status,
2604 expected_stack_id,
2605 )
2606 last_printed_status = status
2607 next_heartbeat = now + heartbeat_interval
2609 time.sleep(min(poll_interval, max(0.0, deadline - now)))
2610 status = self._get_stack_status(stack_name, expected_stack_id)
2612 def _cloudformation_delete_stack(
2613 self,
2614 stack_name: str,
2615 *,
2616 expected_stack_id: str | None = None,
2617 authorize_stack: StackAuthorizationCallback | None = None,
2618 require_expected_identity: bool = False,
2619 ) -> bool:
2620 """Delete an immediately revalidated stack by immutable ARN."""
2621 if self._cdk_cancel_event.is_set(): 2621 ↛ 2622line 2621 didn't jump to line 2622 because the condition on line 2621 was never true
2622 raise RuntimeError(f"CloudFormation deletion cancelled before {stack_name}")
2623 target = self._describe_stack_target(
2624 stack_name,
2625 expected_stack_id=expected_stack_id,
2626 require_expected_identity=require_expected_identity,
2627 )
2628 if target is None: 2628 ↛ 2629line 2628 didn't jump to line 2629 because the condition on line 2628 was never true
2629 return True
2630 region, cfn, stack = target
2631 stack_id = str(stack["StackId"])
2632 status = str(stack.get("StackStatus") or "")
2633 if authorize_stack is not None:
2634 authorize_stack(stack_name, region, stack_id)
2635 if status == "DELETE_IN_PROGRESS": 2635 ↛ 2636line 2635 didn't jump to line 2636 because the condition on line 2635 was never true
2636 return self._wait_for_stack_delete_convergence(
2637 stack_name,
2638 initial_status=status,
2639 expected_stack_id=stack_id,
2640 require_expected_identity=require_expected_identity,
2641 )
2642 try:
2643 cfn.delete_stack(StackName=stack_id)
2644 except Exception:
2645 logger.debug("Direct CloudFormation delete failed for %s", stack_id, exc_info=True)
2646 return False
2647 return self._wait_for_stack_delete_convergence(
2648 stack_name,
2649 expected_stack_id=stack_id,
2650 require_expected_identity=require_expected_identity,
2651 )
2653 def _validated_regional_api_region(self, stack_name: str) -> str | None:
2654 """Return an exact project bridge's SDK-known CloudFormation Region."""
2655 bridge_prefix = f"{self.config.project_name}-regional-api-"
2656 if not stack_name.startswith(bridge_prefix):
2657 return None
2659 region = stack_name[len(bridge_prefix) :]
2660 if not region: 2660 ↛ 2661line 2660 didn't jump to line 2661 because the condition on line 2660 was never true
2661 return None
2662 try:
2663 return region if region in _known_cloudformation_regions() else None
2664 except Exception:
2665 logger.debug(
2666 "Could not validate regional API bridge Region for %s",
2667 stack_name,
2668 exc_info=True,
2669 )
2670 return None
2672 def _configured_regional_api_regions(
2673 self,
2674 ) -> tuple[frozenset[str], str] | None:
2675 """Read valid root regions and their partition for orphan deletion.
2677 Returning ``None`` means the configuration could not prove anything:
2678 missing, unreadable, malformed, wrong-project, incomplete, empty, and
2679 duplicate Region configurations all fail closed under the same contract
2680 used by :class:`ConfigLoader`.
2681 """
2682 path = self.project_root / "cdk.json"
2683 try:
2684 data = json.loads(path.read_text(encoding="utf-8"))
2685 if not isinstance(data, dict): 2685 ↛ 2686line 2685 didn't jump to line 2686 because the condition on line 2685 was never true
2686 return None
2687 context = data.get("context")
2688 if not isinstance(context, dict): 2688 ↛ 2689line 2688 didn't jump to line 2689 because the condition on line 2688 was never true
2689 return None
2690 configured_project = context.get("project_name")
2691 if (
2692 not isinstance(configured_project, str)
2693 or not configured_project
2694 or configured_project != self.config.project_name
2695 ):
2696 return None
2697 deployment_regions = context.get("deployment_regions")
2698 if not isinstance(deployment_regions, dict): 2698 ↛ 2699line 2698 didn't jump to line 2699 because the condition on line 2698 was never true
2699 return None
2701 known_regions = _known_cloudformation_regions()
2702 for key in ("global", "api_gateway", "monitoring"):
2703 region = deployment_regions.get(key)
2704 if not isinstance(region, str) or region not in known_regions:
2705 return None
2707 try:
2708 regional = validated_regional_deployment_regions(
2709 deployment_regions.get("regional"),
2710 known_regions=known_regions,
2711 )
2712 deployment_partition = validated_deployment_partition(
2713 (
2714 deployment_regions["global"],
2715 deployment_regions["api_gateway"],
2716 deployment_regions["monitoring"],
2717 *regional,
2718 )
2719 )
2720 except RuntimeError, ValueError:
2721 return None
2722 return frozenset(regional), deployment_partition
2723 except OSError, UnicodeError, json.JSONDecodeError, TypeError:
2724 logger.debug(
2725 "Could not read authoritative regional configuration from %s",
2726 path,
2727 exc_info=True,
2728 )
2729 return None
2731 def _get_orphan_regional_api_region(self, stack_name: str) -> str | None:
2732 """Return a bridge Region only when valid root config proves it absent.
2734 This result authorizes bypassing CDK and deleting a stack directly via
2735 CloudFormation. Merely failing a normal configuration lookup can never
2736 be interpreted as proof that the stack is orphaned.
2737 """
2738 region = self._validated_regional_api_region(stack_name)
2739 if region is None:
2740 return None
2741 configured = self._configured_regional_api_regions()
2742 if configured is None:
2743 return None
2744 configured_regions, deployment_partition = configured
2745 if region in configured_regions:
2746 return None
2747 try:
2748 candidate_partition = validated_deployment_partition((region,))
2749 except RuntimeError, ValueError:
2750 return None
2751 if candidate_partition != deployment_partition:
2752 return None
2753 return region
2755 def _get_destroy_region(self, stack_name: str) -> str:
2756 """Determine a configured or cryptographically bounded destroy Region.
2758 Deploy resolution intentionally requires bridge Regions to remain in
2759 ``cdk.json``. A removed bridge may still resolve through the orphan
2760 path, but that path validates the exact project-scoped name, SDK-known
2761 CloudFormation Region, authoritative root configuration, and matching
2762 AWS partition. Reconciliation must reuse that same proof rather than
2763 trusting a bridge-shaped suffix independently.
2764 """
2765 try:
2766 region = self._get_deploy_region(stack_name)
2767 except Exception:
2768 logger.debug(
2769 "Configured deploy Region lookup failed for %s; checking orphan shape",
2770 stack_name,
2771 exc_info=True,
2772 )
2773 else:
2774 if region:
2775 return region
2777 orphan_region = self._get_orphan_regional_api_region(stack_name)
2778 return orphan_region or self.config.api_gateway_region
2780 def _ensure_analytics_enabled_for_destroy(self) -> bool:
2781 """Temporarily enable analytics so CDK includes the stack for destroy."""
2782 try:
2783 current = get_analytics_config()
2784 if not current.get("enabled"):
2785 update_analytics_config({"enabled": True})
2786 return True
2787 except Exception as exc:
2788 logger.debug(
2789 "Failed to enable analytics toggle for destroy: %s",
2790 exc,
2791 exc_info=True,
2792 )
2793 return False
2795 def _restore_analytics_disabled(self) -> None:
2796 """Restore analytics toggle to disabled after destroy."""
2797 try:
2798 update_analytics_config({"enabled": False})
2799 except Exception as exc:
2800 logger.warning(
2801 "Failed to restore analytics toggle to disabled after destroy: %s",
2802 exc,
2803 exc_info=True,
2804 )
2806 def _remove_api_gateway_analytics_dependency(
2807 self,
2808 *,
2809 allow_bootstrap: bool = True,
2810 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
2811 expected_stack_ids: Mapping[str, str | None] | None = None,
2812 prepared_change_sets: PreparedChangeSetAuthority | None = None,
2813 authorize_stack: StackAuthorizationCallback | None = None,
2814 strict_deployment_token: str | None = None,
2815 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
2816 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
2817 ) -> bool:
2818 """Redeploy gco-api-gateway with analytics disabled to drop cross-stack imports.
2820 The analytics stack exports values (Cognito pool ARN, presigned-URL
2821 Lambda ARN) that gco-api-gateway imports for the /studio/* routes.
2822 CloudFormation blocks deletion of stacks with consumed exports. By
2823 disabling analytics and redeploying the API gateway, the /studio/*
2824 routes are removed and the imports are dropped, unblocking the
2825 analytics stack deletion.
2827 Returns:
2828 True if the analytics stack is safe to destroy (either because
2829 no consumer remains or because the redeploy successfully
2830 dropped the imports). False if a consumer of the analytics
2831 exports still exists and the analytics destroy will fail.
2832 """
2833 api_gateway_stack = f"{self.config.project_name}-api-gateway"
2834 analytics_stack = f"{self.config.project_name}-analytics"
2836 strict_identity = expected_stack_ids is not None
2837 if strict_identity:
2838 assert expected_stack_ids is not None
2839 if api_gateway_stack not in expected_stack_ids: 2839 ↛ 2840line 2839 didn't jump to line 2840 because the condition on line 2839 was never true
2840 raise RuntimeError(
2841 f"Strict teardown lacks authoritative target state for {api_gateway_stack}"
2842 )
2843 api_gateway_expected_id = expected_stack_ids[api_gateway_stack]
2844 else:
2845 api_gateway_expected_id = None
2847 # Fast path: if the api-gateway stack doesn't exist (or has already
2848 # been deleted/rolled-back into a non-consuming state), there's
2849 # nothing importing the analytics exports. Skip the redeploy entirely.
2850 if not self._stack_exists_in_cloudformation(
2851 api_gateway_stack,
2852 expected_stack_id=api_gateway_expected_id,
2853 require_expected_identity=strict_identity,
2854 ):
2855 logger.info(
2856 "%s does not exist in CloudFormation; skipping redeploy before analytics destroy.",
2857 api_gateway_stack,
2858 )
2859 return True
2861 # Second fast path: if the deployed api-gateway isn't actually
2862 # importing anything from the analytics stack, we don't need to
2863 # touch it. This happens when analytics was never fully wired up.
2864 if not self._api_gateway_imports_from_analytics():
2865 logger.info(
2866 "%s does not import any %s exports; skipping redeploy before analytics destroy.",
2867 api_gateway_stack,
2868 analytics_stack,
2869 )
2870 return True
2872 if strict_identity and ( 2872 ↛ 2875line 2872 didn't jump to line 2875 because the condition on line 2872 was never true
2873 not strict_deployment_token or on_change_set_prepared is None or authorize_stack is None
2874 ):
2875 raise RuntimeError(
2876 "Strict analytics teardown cannot remove API imports without "
2877 "prepared-change-set authority"
2878 )
2880 try:
2881 # Temporarily disable analytics so CDK drops the /studio/* routes.
2882 current = get_analytics_config()
2883 was_enabled = current.get("enabled", False)
2884 if was_enabled:
2885 update_analytics_config({"enabled": False})
2887 print(f" Updating {api_gateway_stack} to remove analytics routes...")
2888 import tempfile
2890 with tempfile.TemporaryDirectory() as tmp_out:
2891 success = self.deploy(
2892 stack_name=api_gateway_stack,
2893 require_approval=False,
2894 exclusively=True,
2895 output_dir=tmp_out,
2896 allow_bootstrap=allow_bootstrap,
2897 bootstrap_stacks=bootstrap_stacks,
2898 expected_stack_ids=expected_stack_ids,
2899 prepared_change_sets=prepared_change_sets,
2900 authorize_stack=authorize_stack,
2901 strict_deployment_token=strict_deployment_token,
2902 on_change_set_prepared=on_change_set_prepared,
2903 on_ecr_repository_created=on_ecr_repository_created,
2904 )
2906 # Re-enable analytics so CDK can synthesize the analytics stack
2907 # for the destroy operation (custom resources need to fire).
2908 if was_enabled:
2909 update_analytics_config({"enabled": True})
2911 if not success:
2912 # The redeploy failed. That's only a real problem if the
2913 # api-gateway still imports analytics exports. Recheck:
2914 # the auto-cleanup of ROLLBACK_COMPLETE stacks may have
2915 # deleted the consumer entirely, in which case the destroy
2916 # can still proceed.
2917 if not self._api_gateway_imports_from_analytics():
2918 logger.info(
2919 "%s redeploy failed, but the stack no longer imports "
2920 "analytics exports (likely deleted during cleanup). "
2921 "Analytics destroy can proceed.",
2922 api_gateway_stack,
2923 )
2924 return True
2925 logger.error(
2926 "Failed to redeploy %s to drop analytics imports, and the "
2927 "stack still consumes analytics exports. Destroying %s will "
2928 "fail with 'Export ... cannot be deleted as it is in use'. "
2929 "Fix %s first (see events above) and retry.",
2930 api_gateway_stack,
2931 analytics_stack,
2932 api_gateway_stack,
2933 )
2934 return False
2936 return True
2937 except Exception as exc:
2938 logger.warning(
2939 "Failed to remove API gateway analytics dependency: %s",
2940 exc,
2941 exc_info=True,
2942 )
2943 # On unexpected exceptions, recheck whether imports remain.
2944 # Be permissive only if we can confirm the destroy is safe.
2945 try:
2946 return not self._api_gateway_imports_from_analytics()
2947 except Exception:
2948 return False
2950 def _api_gateway_imports_from_analytics(self) -> bool:
2951 """Return True if gco-api-gateway imports any exports from gco-analytics.
2953 Uses CloudFormation's ``list_exports`` + ``list_imports`` to detect
2954 cross-stack references at runtime. This is more reliable than
2955 inspecting the CDK app because it reflects what's actually
2956 deployed.
2957 """
2958 import boto3
2960 analytics_stack = f"{self.config.project_name}-analytics"
2961 api_gateway_stack = f"{self.config.project_name}-api-gateway"
2963 region = self._get_deploy_region(analytics_stack)
2964 if not region:
2965 return False
2967 try:
2968 cfn = boto3.client("cloudformation", region_name=region)
2969 # Collect every export whose owning stack is the analytics stack.
2970 analytics_exports: list[str] = []
2971 paginator = cfn.get_paginator("list_exports")
2972 for page in paginator.paginate():
2973 for export in page.get("Exports", []):
2974 owner = export.get("ExportingStackId", "")
2975 # ExportingStackId is a full ARN; match by stack name.
2976 if f":stack/{analytics_stack}/" in owner:
2977 analytics_exports.append(export["Name"])
2979 if not analytics_exports:
2980 return False
2982 # For each export, check whether the api-gateway stack is
2983 # listed as an importer. ``list_imports`` returns the stack
2984 # names that currently import the given export.
2985 import_paginator = cfn.get_paginator("list_imports")
2986 for export_name in analytics_exports:
2987 try:
2988 for page in import_paginator.paginate(ExportName=export_name):
2989 for importer in page.get("Imports", []):
2990 if importer == api_gateway_stack: 2990 ↛ 2989line 2990 didn't jump to line 2989 because the condition on line 2990 was always true
2991 return True
2992 except Exception as exc:
2993 # ``list_imports`` raises when an export has zero
2994 # consumers — treat that as "not imported" and move on.
2995 logger.debug(
2996 "list_imports(%s) failed (likely no consumers): %s",
2997 export_name,
2998 exc,
2999 )
3000 return False
3001 except Exception as exc:
3002 logger.debug(
3003 "Failed to check analytics imports for %s: %s",
3004 api_gateway_stack,
3005 exc,
3006 exc_info=True,
3007 )
3008 # On failure to check, err on the side of attempting the
3009 # redeploy so we don't skip necessary cleanup.
3010 return True
3012 def bootstrap(
3013 self,
3014 account: str | None = None,
3015 region: str | None = None,
3016 ) -> bool:
3017 """Bootstrap CDK in an AWS account/region."""
3018 cmd = ["bootstrap"]
3020 if account and region:
3021 cmd.append(f"aws://{account}/{region}")
3022 elif region: 3022 ↛ 3025line 3022 didn't jump to line 3025 because the condition on line 3022 was always true
3023 cmd.append(f"aws://unknown-account/{region}")
3025 result = self._run_cdk(cmd)
3026 return result.returncode == 0
3028 def is_bootstrapped(self, region: str) -> bool:
3029 """Check if CDK has been bootstrapped in a region.
3031 Looks for the CDKToolkit CloudFormation stack which is created
3032 by ``cdk bootstrap``. Result is cached per region for the lifetime
3033 of this StackManager instance.
3034 """
3035 if not hasattr(self, "_bootstrap_cache"): 3035 ↛ 3038line 3035 didn't jump to line 3038 because the condition on line 3035 was always true
3036 self._bootstrap_cache: dict[str, bool] = {}
3038 if region in self._bootstrap_cache: 3038 ↛ 3039line 3038 didn't jump to line 3039 because the condition on line 3038 was never true
3039 return self._bootstrap_cache[region]
3041 import boto3
3043 cf = boto3.client("cloudformation", region_name=region)
3044 try:
3045 response = cf.describe_stacks(StackName="CDKToolkit")
3046 stacks = response.get("Stacks", [])
3047 if stacks:
3048 status = stacks[0].get("StackStatus", "")
3049 # Any non-deleted state counts as bootstrapped
3050 result = "DELETE" not in status
3051 self._bootstrap_cache[region] = result
3052 return result
3053 except ClientError:
3054 pass # Stack doesn't exist — not bootstrapped
3055 except Exception as e:
3056 logger.debug("Failed to check CDK bootstrap in %s: %s", region, e)
3058 self._bootstrap_cache[region] = False
3059 return False
3061 def _validate_bootstrap_stack(
3062 self,
3063 region: str,
3064 expected: Mapping[str, str],
3065 ) -> None:
3066 """Require the exact preflighted CDKToolkit ARN and healthy status."""
3067 import boto3
3069 expected_id = str(expected.get("stack_id") or "")
3070 expected_status = str(expected.get("status") or "")
3071 if not expected_id or expected_status not in _BOOTSTRAP_HEALTHY_STATUSES: 3071 ↛ 3072line 3071 didn't jump to line 3072 because the condition on line 3071 was never true
3072 raise RuntimeError(f"Invalid checkpointed CDKToolkit identity for {region}")
3073 cfn = boto3.client("cloudformation", region_name=region)
3074 try:
3075 response = cfn.describe_stacks(StackName=expected_id)
3076 except Exception as exc:
3077 raise RuntimeError(
3078 f"Could not revalidate checkpointed CDKToolkit {expected_id} in {region}"
3079 ) from exc
3080 stacks = response.get("Stacks", [])
3081 if len(stacks) != 1: 3081 ↛ 3082line 3081 didn't jump to line 3082 because the condition on line 3081 was never true
3082 raise RuntimeError(f"CDKToolkit {expected_id} returned an invalid identity")
3083 stack = stacks[0]
3084 actual_id = str(stack.get("StackId") or "")
3085 actual_status = str(stack.get("StackStatus") or "")
3086 if stack.get("StackName") != "CDKToolkit" or actual_id != expected_id:
3087 raise RuntimeError(f"CDKToolkit identity changed in {region}")
3088 if actual_status != expected_status or actual_status not in _BOOTSTRAP_HEALTHY_STATUSES:
3089 raise RuntimeError(
3090 f"CDKToolkit {expected_id} status changed from {expected_status} "
3091 f"to {actual_status or 'unknown'}"
3092 )
3094 @staticmethod
3095 def _strict_change_set_name(stack_name: str, token: str) -> str:
3096 """Return one deterministic, run-scoped CloudFormation change-set name."""
3097 safe_token = "".join(
3098 character if character.isascii() and character.isalnum() else "-" for character in token
3099 )
3100 safe_token = "-".join(part for part in safe_token.split("-") if part)
3101 digest = hashlib.sha256(f"{token}:{stack_name}".encode()).hexdigest()[:16]
3102 namespace = "gco"
3103 max_token_length = 128 - len(namespace) - len(digest) - 2
3104 safe_token = (safe_token or "live-validation")[:max_token_length]
3105 return f"{namespace}-{safe_token}-{digest}"
3107 def _preflight_strict_change_set(
3108 self,
3109 *,
3110 stack_name: str,
3111 change_set_name: str,
3112 expected_stack_id: str | None,
3113 prepared_change_sets: Mapping[str, Mapping[str, str]],
3114 ) -> None:
3115 """Reject an existing deterministic change set without checkpoint authority."""
3116 import boto3
3118 region = self._get_deploy_region(stack_name)
3119 if not region: 3119 ↛ 3120line 3119 didn't jump to line 3120 because the condition on line 3119 was never true
3120 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
3121 cfn = boto3.client("cloudformation", region_name=region)
3122 try:
3123 change_set = cfn.describe_change_set(
3124 ChangeSetName=change_set_name,
3125 StackName=stack_name,
3126 )
3127 except ClientError as exc:
3128 if self._change_set_missing(exc):
3129 return
3130 # DescribeChangeSet reports a stack-style ValidationError when both
3131 # the deterministic change set and its fresh target stack are absent.
3132 # The target was authoritatively checked immediately above; only an
3133 # empty create history can safely interpret this as "not prepared".
3134 if expected_stack_id is None and not prepared_change_sets and self._stack_missing(exc):
3135 return
3136 raise RuntimeError(
3137 f"Could not preflight strict change set {change_set_name} for {stack_name}"
3138 ) from exc
3140 change_set_id = str(change_set.get("ChangeSetId") or "")
3141 observed_change_set_name = str(change_set.get("ChangeSetName") or "")
3142 stack_id = str(change_set.get("StackId") or "")
3143 if not change_set_id or not stack_id or observed_change_set_name != change_set_name: 3143 ↛ 3144line 3143 didn't jump to line 3144 because the condition on line 3143 was never true
3144 raise RuntimeError(
3145 f"Existing strict change set {change_set_name} omitted immutable identities"
3146 )
3147 self._validate_strict_change_set_arns(
3148 stack_name=stack_name,
3149 change_set_name=change_set_name,
3150 stack_id=stack_id,
3151 change_set_id=change_set_id,
3152 region=region,
3153 )
3154 prepared_record = prepared_change_sets.get(change_set_id)
3155 if prepared_record is None:
3156 raise RuntimeError(
3157 f"Existing strict change set {change_set_id} lacks checkpoint authority"
3158 )
3159 recorded_change_set_id = str(prepared_record.get("change_set_id") or "")
3160 recorded_stack_id = str(prepared_record.get("stack_id") or "")
3161 recorded_type = str(prepared_record.get("change_set_type") or "")
3162 if ( 3162 ↛ 3169line 3162 didn't jump to line 3169 because the condition on line 3162 was never true
3163 expected_stack_id is None
3164 or stack_id != expected_stack_id
3165 or recorded_change_set_id != change_set_id
3166 or recorded_stack_id != stack_id
3167 or recorded_type not in {"CREATE", "UPDATE"}
3168 ):
3169 raise RuntimeError(f"Existing strict change-set authority changed for {stack_name}")
3171 @staticmethod
3172 def _validate_strict_change_set_arns(
3173 *,
3174 stack_name: str,
3175 change_set_name: str,
3176 stack_id: str,
3177 change_set_id: str,
3178 region: str,
3179 ) -> None:
3180 """Require both prepared identities to be exact, related CloudFormation ARNs."""
3182 def split_arn(identifier: str, label: str) -> tuple[str, str, str, str]:
3183 parts = identifier.split(":", 5)
3184 if (
3185 len(parts) != 6
3186 or parts[0] != "arn"
3187 or not (parts[1] == "aws" or parts[1].startswith("aws-"))
3188 or parts[2] != "cloudformation"
3189 or parts[3] != region
3190 or not parts[4]
3191 or not parts[5]
3192 ):
3193 raise RuntimeError(f"Strict {label} has an invalid CloudFormation ARN")
3194 return parts[1], parts[3], parts[4], parts[5]
3196 stack_partition, _stack_region, stack_account, stack_resource = split_arn(
3197 stack_id,
3198 "stack identity",
3199 )
3200 stack_prefix = f"stack/{stack_name}/"
3201 if not stack_resource.startswith(stack_prefix) or not stack_resource.removeprefix( 3201 ↛ 3204line 3201 didn't jump to line 3204 because the condition on line 3201 was never true
3202 stack_prefix
3203 ):
3204 raise RuntimeError(
3205 f"Strict stack identity {stack_id} does not name expected stack {stack_name}"
3206 )
3208 change_partition, _change_region, change_account, change_resource = split_arn(
3209 change_set_id,
3210 "change-set identity",
3211 )
3212 change_prefix = f"changeSet/{change_set_name}/"
3213 if not change_resource.startswith(change_prefix) or not change_resource.removeprefix( 3213 ↛ 3216line 3213 didn't jump to line 3216 because the condition on line 3213 was never true
3214 change_prefix
3215 ):
3216 raise RuntimeError(
3217 f"Strict change-set identity {change_set_id} does not name {change_set_name}"
3218 )
3219 if change_partition != stack_partition or change_account != stack_account:
3220 raise RuntimeError(
3221 "Strict stack and change-set identities belong to different AWS authorities"
3222 )
3224 def _execute_prepared_change_set(
3225 self,
3226 *,
3227 stack_name: str,
3228 change_set_name: str,
3229 expected_stack_id: str | None,
3230 expected_tags: Mapping[str, str] | None,
3231 prepared_change_sets: Mapping[str, Mapping[str, str]],
3232 preparation_succeeded: bool,
3233 authorize_stack: StackAuthorizationCallback | None,
3234 on_change_set_prepared: ChangeSetPreparedCallback,
3235 allow_noop: bool,
3236 timeout: float,
3237 ) -> bool:
3238 """Validate, checkpoint, and execute only the deterministic CDK change set."""
3239 import boto3
3241 region = self._get_deploy_region(stack_name)
3242 if not region: 3242 ↛ 3243line 3242 didn't jump to line 3243 because the condition on line 3242 was never true
3243 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
3244 cfn = boto3.client("cloudformation", region_name=region)
3245 try:
3246 change_set = cfn.describe_change_set(
3247 ChangeSetName=change_set_name,
3248 StackName=stack_name,
3249 )
3250 except ClientError as exc:
3251 if not self._change_set_missing(exc): 3251 ↛ 3252line 3251 didn't jump to line 3252 because the condition on line 3251 was never true
3252 raise RuntimeError(
3253 f"Could not inspect strict change set {change_set_name} for {stack_name}"
3254 ) from exc
3255 if allow_noop and expected_stack_id: 3255 ↛ 3271line 3255 didn't jump to line 3271 because the condition on line 3255 was always true
3256 target = self._describe_stack_target(
3257 stack_name,
3258 expected_stack_id=expected_stack_id,
3259 require_expected_identity=True,
3260 )
3261 if target is not None: 3261 ↛ 3271line 3261 didn't jump to line 3271 because the condition on line 3261 was always true
3262 stack = target[2]
3263 status = str(stack.get("StackStatus") or "")
3264 if status in _BOOTSTRAP_HEALTHY_STATUSES: 3264 ↛ 3271line 3264 didn't jump to line 3271 because the condition on line 3264 was always true
3265 if authorize_stack is None: 3265 ↛ 3266line 3265 didn't jump to line 3266 because the condition on line 3265 was never true
3266 raise RuntimeError(
3267 f"Strict no-op for {stack_name} lacks exact authorization"
3268 ) from exc
3269 authorize_stack(stack_name, region, expected_stack_id)
3270 return True
3271 raise RuntimeError(
3272 f"CDK did not create the strict change set {change_set_name} for {stack_name}"
3273 ) from exc
3275 change_set_id = str(change_set.get("ChangeSetId") or "")
3276 observed_change_set_name = str(change_set.get("ChangeSetName") or "")
3277 stack_id = str(change_set.get("StackId") or "")
3278 status = str(change_set.get("Status") or "")
3279 execution_status = str(change_set.get("ExecutionStatus") or "")
3280 if not change_set_id or not stack_id: 3280 ↛ 3281line 3280 didn't jump to line 3281 because the condition on line 3280 was never true
3281 raise RuntimeError(f"Strict change set {change_set_name} omitted immutable identities")
3282 if observed_change_set_name != change_set_name: 3282 ↛ 3283line 3282 didn't jump to line 3283 because the condition on line 3282 was never true
3283 raise RuntimeError(
3284 f"Strict change set identity changed from {change_set_name} "
3285 f"to {observed_change_set_name or 'unknown'}"
3286 )
3287 self._validate_strict_change_set_arns(
3288 stack_name=stack_name,
3289 change_set_name=change_set_name,
3290 stack_id=stack_id,
3291 change_set_id=change_set_id,
3292 region=region,
3293 )
3294 prepared_record = prepared_change_sets.get(change_set_id)
3295 if prepared_record is None:
3296 # DescribeChangeSet does not expose ChangeSetType. For a newly
3297 # prepared change set, the pre-CDK exact target state is the only
3298 # authoritative source: absence means CREATE; an exact stack means
3299 # UPDATE. Resumes use the persisted per-change-set record below.
3300 change_set_type = "CREATE" if expected_stack_id is None else "UPDATE"
3301 else:
3302 recorded_change_set_id = str(prepared_record.get("change_set_id") or "")
3303 recorded_stack_id = str(prepared_record.get("stack_id") or "")
3304 change_set_type = str(prepared_record.get("change_set_type") or "")
3305 if recorded_change_set_id != change_set_id or recorded_stack_id != stack_id: 3305 ↛ 3306line 3305 didn't jump to line 3306 because the condition on line 3305 was never true
3306 raise RuntimeError(
3307 f"Persisted strict change-set authority changed for {stack_name}"
3308 )
3309 if change_set_type not in {"CREATE", "UPDATE"}: 3309 ↛ 3310line 3309 didn't jump to line 3310 because the condition on line 3309 was never true
3310 raise RuntimeError(
3311 f"Persisted strict change set for {stack_name} has invalid type "
3312 f"{change_set_type or 'unknown'}"
3313 )
3314 if change_set_type == "UPDATE" and expected_stack_id is None: 3314 ↛ 3315line 3314 didn't jump to line 3315 because the condition on line 3314 was never true
3315 raise RuntimeError(
3316 f"Strict change set for absent {stack_name} unexpectedly performs UPDATE"
3317 )
3318 if expected_stack_id is not None and stack_id != expected_stack_id: 3318 ↛ 3319line 3318 didn't jump to line 3319 because the condition on line 3318 was never true
3319 raise RuntimeError(
3320 f"Strict change set targets replacement {stack_id}; expected {expected_stack_id}"
3321 )
3322 observed_tags = {
3323 str(tag.get("Key")): str(tag.get("Value"))
3324 for tag in change_set.get("Tags", [])
3325 if tag.get("Key") is not None
3326 }
3327 for key, value in (expected_tags or {}).items():
3328 if observed_tags.get(str(key)) != str(value):
3329 raise RuntimeError(f"Strict change set {change_set_id} omitted required tag {key}")
3331 status_reason = " ".join(str(change_set.get("StatusReason") or "").split()).lower()
3332 empty_change_set = (
3333 "submitted information didn't contain changes" in status_reason
3334 or "no updates are to be performed" in status_reason
3335 )
3336 if (
3337 status == "FAILED"
3338 and empty_change_set
3339 and (allow_noop or prepared_record is not None)
3340 and expected_stack_id
3341 ):
3342 if stack_id != expected_stack_id: 3342 ↛ 3343line 3342 didn't jump to line 3343 because the condition on line 3342 was never true
3343 raise RuntimeError(
3344 f"Empty strict change set targets {stack_id}; expected {expected_stack_id}"
3345 )
3346 target = self._describe_stack_target(
3347 stack_name,
3348 expected_stack_id=expected_stack_id,
3349 require_expected_identity=True,
3350 )
3351 if target is None or str(target[2].get("StackStatus") or "") not in ( 3351 ↛ 3354line 3351 didn't jump to line 3354 because the condition on line 3351 was never true
3352 _BOOTSTRAP_HEALTHY_STATUSES
3353 ):
3354 raise RuntimeError(
3355 f"Empty strict change set {change_set_id} has no healthy exact stack"
3356 )
3357 if authorize_stack is None: 3357 ↛ 3358line 3357 didn't jump to line 3358 because the condition on line 3357 was never true
3358 raise RuntimeError(f"Strict no-op for {stack_name} lacks exact authorization")
3359 authorize_stack(stack_name, region, expected_stack_id)
3360 on_change_set_prepared(
3361 stack_name,
3362 region,
3363 stack_id,
3364 change_set_id,
3365 change_set_type,
3366 )
3367 return True
3368 if status != "CREATE_COMPLETE" or execution_status not in { 3368 ↛ 3372line 3368 didn't jump to line 3372 because the condition on line 3368 was never true
3369 "AVAILABLE",
3370 "EXECUTE_COMPLETE",
3371 }:
3372 raise RuntimeError(
3373 f"Strict change set {change_set_id} is {status}/{execution_status}, not usable"
3374 )
3376 if execution_status == "AVAILABLE" and (
3377 prepared_record is None and not preparation_succeeded
3378 ):
3379 raise RuntimeError(
3380 f"Strict change set {change_set_id} was not produced by this preparation"
3381 )
3382 if execution_status == "EXECUTE_COMPLETE" and (
3383 prepared_record is None or expected_stack_id is None
3384 ):
3385 raise RuntimeError(
3386 f"Executed strict change set {change_set_id} lacks prior checkpoint authority"
3387 )
3389 if expected_stack_id is not None:
3390 if authorize_stack is None: 3390 ↛ 3391line 3390 didn't jump to line 3391 because the condition on line 3390 was never true
3391 raise RuntimeError(f"Strict change set for {stack_name} lacks exact authorization")
3392 authorize_stack(stack_name, region, stack_id)
3394 if execution_status == "EXECUTE_COMPLETE":
3395 target = self._describe_stack_target(
3396 stack_name,
3397 expected_stack_id=stack_id,
3398 require_expected_identity=True,
3399 )
3400 if target is None or str(target[2].get("StackStatus") or "") not in ( 3400 ↛ 3403line 3400 didn't jump to line 3403 because the condition on line 3400 was never true
3401 _BOOTSTRAP_HEALTHY_STATUSES
3402 ):
3403 raise RuntimeError(
3404 f"Executed strict change set {change_set_id} has no healthy exact stack"
3405 )
3406 elif change_set_type == "CREATE":
3407 target = self._describe_stack_target(
3408 stack_name,
3409 expected_stack_id=stack_id,
3410 require_expected_identity=True,
3411 )
3412 if target is None or str(target[2].get("StackStatus") or "") != "REVIEW_IN_PROGRESS":
3413 raise RuntimeError(
3414 f"Prepared CREATE change set {change_set_id} has no exact review stack"
3415 )
3417 on_change_set_prepared(
3418 stack_name,
3419 region,
3420 stack_id,
3421 change_set_id,
3422 change_set_type,
3423 )
3424 if execution_status == "EXECUTE_COMPLETE":
3425 return True
3426 if self._cdk_cancel_event.is_set():
3427 raise RuntimeError(
3428 f"Strict change set {change_set_id} was checkpointed but execution was cancelled"
3429 )
3431 cfn.execute_change_set(ChangeSetName=change_set_id)
3432 settled = self._wait_for_stack_settle(
3433 stack_name,
3434 timeout=timeout,
3435 stack_identifier=stack_id,
3436 )
3437 if settled not in _BOOTSTRAP_HEALTHY_STATUSES: 3437 ↛ 3438line 3437 didn't jump to line 3438 because the condition on line 3437 was never true
3438 logger.error(
3439 "Strict change set %s for %s settled as %s",
3440 change_set_id,
3441 stack_name,
3442 settled or "unknown",
3443 )
3444 return False
3445 return True
3447 def ensure_bootstrapped(self, region: str) -> bool:
3448 """Ensure a region is CDK-bootstrapped, auto-bootstrapping if needed.
3450 Returns True if the region is (or was successfully) bootstrapped.
3451 """
3452 if self.is_bootstrapped(region):
3453 return True
3455 print(f"ℹ Region {region} is not CDK-bootstrapped. Bootstrapping now...")
3456 success = self.bootstrap(region=region)
3457 if success:
3458 # Update cache so we don't re-check this region
3459 if not hasattr(self, "_bootstrap_cache"): 3459 ↛ 3461line 3459 didn't jump to line 3461 because the condition on line 3459 was always true
3460 self._bootstrap_cache = {}
3461 self._bootstrap_cache[region] = True
3462 print(f"✓ CDK bootstrapped in {region}")
3463 else:
3464 print(f"✗ Failed to bootstrap CDK in {region}")
3465 return success
3467 def _get_deploy_region(self, stack_name: str) -> str | None:
3468 """Determine the target AWS region for a given stack name."""
3469 from .config import _load_cdk_json
3471 cdk_regions = _load_cdk_json()
3473 # Named stacks are classified by suffix and regional stacks by the
3474 # ``<project>-`` prefix (#139) so a non-``gco`` deployment resolves
3475 # regions for its own ``<project>-*`` stacks — otherwise the image
3476 # mirror (which calls this to pick a regional stack's region) would
3477 # silently no-op. For the default ``gco`` behaviour is unchanged.
3478 region: str | None
3479 if stack_name.endswith("-global"):
3480 region = cdk_regions.get("global") or self.config.global_region
3481 return region
3482 if stack_name.endswith("-api-gateway"):
3483 region = cdk_regions.get("api_gateway") or self.config.api_gateway_region
3484 return region
3485 if stack_name.endswith("-monitoring"):
3486 region = cdk_regions.get("monitoring") or self.config.monitoring_region
3487 return region
3488 if stack_name.endswith("-analytics"):
3489 # The analytics stack shares the API gateway region so the
3490 # presigned-URL Lambda can hook into the existing /studio/*
3491 # routes on the same API Gateway.
3492 region = cdk_regions.get("api_gateway") or self.config.api_gateway_region
3493 return region
3495 # Regional API bridges use ``<project>-regional-api-<region>``. Resolve
3496 # this exact shape before generic regional stacks; otherwise the generic
3497 # project-prefix branch returns the malformed ``regional-api-<region>``.
3498 # Requiring a configured deployment region also prevents bridge-shaped
3499 # typos from being treated as valid AWS regions.
3500 bridge_prefix = f"{self.config.project_name}-regional-api-"
3501 if stack_name.startswith(bridge_prefix):
3502 region = stack_name[len(bridge_prefix) :]
3503 configured_regions = {str(item) for item in (cdk_regions.get("regional") or [])}
3504 return region if region in configured_regions else None
3506 # Base regional stacks: {project}-{region}. The region is whatever
3507 # follows the project prefix (regions contain hyphens, so we strip
3508 # the known prefix rather than guess a split point).
3509 prefix = f"{self.config.project_name}-"
3510 if stack_name.startswith(prefix):
3511 return stack_name[len(prefix) :]
3513 return None
3515 def _mirror_target_regions(self, stack_name: str | None, all_stacks: bool) -> list[str]:
3516 """Regional regions to auto-mirror images for on this deploy.
3518 Only regional stacks (``gco-<region>``) run a Helm install that needs the
3519 mirror; the named global / api-gateway / monitoring / analytics stacks do
3520 not. For ``--all`` the regional regions come straight from cdk.json
3521 (``deployment_regions.regional``) so no synth is required. Returns a
3522 de-duplicated, order-stable list.
3523 """
3524 # Derive the prefix from project_name (#139) so a non-``gco``
3525 # deployment's regional stacks (``<project>-<region>``) are still
3526 # recognised — otherwise the mirror would silently no-op and the
3527 # regional Volcano Helm install would have no images to pull.
3528 prefix = f"{self.config.project_name}-"
3529 named = {
3530 f"{prefix}global",
3531 f"{prefix}api-gateway",
3532 f"{prefix}monitoring",
3533 f"{prefix}analytics",
3534 }
3535 if all_stacks:
3536 from .config import _load_cdk_json
3538 regional = _load_cdk_json().get("regional") or []
3539 return list(dict.fromkeys(str(r) for r in regional))
3541 # Bridge stacks contain no regional Helm consumers and must not trigger
3542 # image mirroring. Match the exact project-scoped prefix so a project
3543 # name containing ``regional-api`` remains unambiguous.
3544 bridge_prefix = f"{self.config.project_name}-regional-api-"
3545 if stack_name and stack_name.startswith(bridge_prefix):
3546 return []
3548 if stack_name and stack_name.startswith(prefix) and stack_name not in named:
3549 region = self._get_deploy_region(stack_name)
3550 return [region] if region else []
3551 return []
3553 def _mirror_images_if_enabled(
3554 self,
3555 stack_name: str | None,
3556 all_stacks: bool,
3557 repository_tags: Mapping[str, str] | None = None,
3558 on_repository_created: EcrRepositoryCreatedCallback | None = None,
3559 ) -> None:
3560 """Mirror third-party images into ECR before a regional stack deploys.
3562 No-op unless ``volcano_image_mirror.enabled`` is set in cdk.json. Mirrors
3563 every relevant regional region (see :meth:`_mirror_target_regions`); the
3564 copy is idempotent and skips images already present, so a fresh deploy
3565 seeds the mirror automatically and repeat deploys cost only a few ECR
3566 describe calls. Raises **before** any CDK call if an enabled mirror fails,
3567 so a deploy never points a consumer (e.g. Volcano's ``image_registry``)
3568 at images that aren't in ECR yet.
3569 """
3570 from . import _image_mirror as image_mirror
3572 cfg = image_mirror.read_mirror_config()
3573 if not cfg["enabled"]:
3574 return
3576 regions = self._mirror_target_regions(stack_name, all_stacks)
3577 for region in regions:
3578 print(f"Mirroring third-party images into ECR for {region} ...")
3579 try:
3580 image_mirror.mirror_images(
3581 region,
3582 ecr_namespace=cfg["ecr_namespace"],
3583 skip_existing=True,
3584 repository_tags=repository_tags,
3585 on_repository_created=on_repository_created,
3586 )
3587 except Exception as exc: # noqa: BLE001 - surface a clear, actionable failure
3588 raise RuntimeError(
3589 f"Image mirror failed for region {region}: {exc}\n"
3590 "volcano_image_mirror is enabled but the images could not be "
3591 "mirrored into ECR. Fix the cause (container runtime / network / "
3592 "credentials) or run "
3593 f"'gco images mirror --region {region}' manually, "
3594 "then retry. Aborting before CDK so the deploy never points a "
3595 "consumer at images that aren't in ECR."
3596 ) from exc
3598 def get_outputs(self, stack_name: str, region: str) -> dict[str, str]:
3599 """Get stack outputs from CloudFormation."""
3600 import boto3
3602 cf = boto3.client("cloudformation", region_name=region)
3603 try:
3604 response = cf.describe_stacks(StackName=stack_name)
3605 if response["Stacks"]:
3606 stack = response["Stacks"][0]
3607 outputs: dict[str, str] = {}
3608 for output in stack.get("Outputs", []):
3609 outputs[str(output["OutputKey"])] = str(output["OutputValue"])
3610 return outputs
3611 except Exception as e:
3612 logger.debug("Failed to get outputs for %s in %s: %s", stack_name, region, e)
3613 return {}
3615 def get_stack_status(self, stack_name: str, region: str) -> StackInfo | None:
3616 """Get detailed stack status from CloudFormation."""
3617 import boto3
3619 cf = boto3.client("cloudformation", region_name=region)
3620 try:
3621 response = cf.describe_stacks(StackName=stack_name)
3622 if response["Stacks"]:
3623 stack = response["Stacks"][0]
3624 return StackInfo(
3625 name=stack["StackName"],
3626 status=stack["StackStatus"],
3627 region=region,
3628 created_time=stack.get("CreationTime"),
3629 updated_time=stack.get("LastUpdatedTime"),
3630 outputs={o["OutputKey"]: o["OutputValue"] for o in stack.get("Outputs", [])},
3631 tags={t["Key"]: t["Value"] for t in stack.get("Tags", [])},
3632 )
3633 except Exception as e:
3634 logger.debug("Failed to get stack status for %s in %s: %s", stack_name, region, e)
3635 return None
3637 def deploy_orchestrated(
3638 self,
3639 require_approval: bool = True,
3640 outputs_file: str | None = None,
3641 parameters: dict[str, str] | None = None,
3642 tags: dict[str, str] | None = None,
3643 progress: str = "events",
3644 on_stack_start: Callable[[str], None] | None = None,
3645 on_stack_complete: Callable[[str, bool], None] | None = None,
3646 parallel: bool = False,
3647 max_workers: int = 4,
3648 allow_bootstrap: bool = True,
3649 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
3650 expected_stack_ids: Mapping[str, str | None] | None = None,
3651 prepared_change_sets: PreparedChangeSetAuthority | None = None,
3652 authorize_stack: StackAuthorizationCallback | None = None,
3653 strict_deployment_token: str | None = None,
3654 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
3655 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
3656 ) -> tuple[bool, list[str], list[str]]:
3657 """
3658 Deploy all stacks in the correct order.
3660 Deploys global stacks first, then base regional stacks, regional API
3661 bridges, and finally monitoring. Parallelism never crosses a dependency
3662 level.
3664 Args:
3665 require_approval: Whether to require approval for changes
3666 outputs_file: File to write outputs to
3667 parameters: CDK parameters
3668 tags: Tags to apply to stacks
3669 progress: Progress display type
3670 on_stack_start: Callback(stack_name) called when starting a stack
3671 on_stack_complete: Callback(stack_name, success) called when stack completes
3672 parallel: Deploy regional stacks in parallel
3673 max_workers: Maximum number of parallel deployments (default: 4)
3675 Returns:
3676 Tuple of (overall_success, successful_stacks, failed_stacks)
3677 """
3678 stacks = self.list_stacks()
3679 stack_names = set(stacks)
3680 project_name = self.config.project_name
3681 ordered_stacks = get_stack_deployment_order(stacks, project_name=project_name)
3683 strict_deployment = (
3684 strict_deployment_token is not None or on_change_set_prepared is not None
3685 )
3686 if strict_deployment:
3687 if not strict_deployment_token or on_change_set_prepared is None: 3687 ↛ 3688line 3687 didn't jump to line 3688 because the condition on line 3687 was never true
3688 raise RuntimeError(
3689 "Strict deployment requires both a run token and a prepared-change-set callback"
3690 )
3691 if allow_bootstrap: 3691 ↛ 3692line 3691 didn't jump to line 3692 because the condition on line 3691 was never true
3692 raise RuntimeError("Strict orchestrated deployment cannot auto-bootstrap")
3693 if authorize_stack is None: 3693 ↛ 3694line 3693 didn't jump to line 3694 because the condition on line 3693 was never true
3694 raise RuntimeError("Strict orchestrated deployment requires an exact authorizer")
3695 if expected_stack_ids is None: 3695 ↛ 3696line 3695 didn't jump to line 3696 because the condition on line 3695 was never true
3696 raise RuntimeError("Strict orchestrated deployment lacks target identities")
3697 if prepared_change_sets is None: 3697 ↛ 3698line 3697 didn't jump to line 3698 because the condition on line 3697 was never true
3698 raise RuntimeError("Strict orchestrated deployment lacks change-set history")
3699 missing = sorted(set(stacks) - set(expected_stack_ids))
3700 unexpected = sorted(set(expected_stack_ids) - set(stacks))
3701 if missing or unexpected: 3701 ↛ 3702line 3701 didn't jump to line 3702 because the condition on line 3701 was never true
3702 raise RuntimeError(
3703 "Strict deployment target map does not match the CDK graph; "
3704 f"missing={missing}, unexpected={unexpected}"
3705 )
3706 missing_history = sorted(set(stacks) - set(prepared_change_sets))
3707 unexpected_history = sorted(set(prepared_change_sets) - set(stacks))
3708 if missing_history or unexpected_history: 3708 ↛ 3709line 3708 didn't jump to line 3709 because the condition on line 3708 was never true
3709 raise RuntimeError(
3710 "Strict change-set history does not match the CDK graph; "
3711 f"missing={missing_history}, unexpected={unexpected_history}"
3712 )
3714 # Validate every toolkit and every expected stack before the first
3715 # repository copy, stuck-stack recovery, or CloudFormation mutation.
3716 validated_regions: set[str] = set()
3717 for target_name in ordered_stacks: 3717 ↛ 3749line 3717 didn't jump to line 3749 because the loop on line 3717 didn't complete
3718 region = self._get_deploy_region(target_name)
3719 if not region: 3719 ↛ 3720line 3719 didn't jump to line 3720 because the condition on line 3719 was never true
3720 raise RuntimeError(f"Could not resolve deploy Region for {target_name}")
3721 if region not in validated_regions:
3722 expected_bootstrap = (bootstrap_stacks or {}).get(region)
3723 if expected_bootstrap is None: 3723 ↛ 3724line 3723 didn't jump to line 3724 because the condition on line 3723 was never true
3724 raise RuntimeError(
3725 f"Strict deployment lacks a checkpointed CDKToolkit identity for {region}"
3726 )
3727 self._validate_bootstrap_stack(region, expected_bootstrap)
3728 validated_regions.add(region)
3730 expected_id = expected_stack_ids[target_name]
3731 target = self._describe_stack_target(
3732 target_name,
3733 expected_stack_id=expected_id,
3734 require_expected_identity=True,
3735 )
3736 if expected_id is not None and target is None: 3736 ↛ 3737line 3736 didn't jump to line 3737 because the condition on line 3736 was never true
3737 raise RuntimeError(
3738 f"Checkpointed stack {expected_id} is absent; refusing recreation"
3739 )
3740 if target is not None: 3740 ↛ 3717line 3740 didn't jump to line 3717 because the condition on line 3740 was always true
3741 authorize_stack(target_name, region, str(target[2]["StackId"]))
3743 # Separate stacks into four dependency levels by suffix/marker so
3744 # ordering is independent of project_name (#139):
3745 # 1. Pre-regional global stacks (<project>-global, <project>-api-gateway)
3746 # 2. Base regional stacks (<project>-<region>, parallel-safe)
3747 # 3. Regional API bridges (<project>-regional-api-<region>, depend on base)
3748 # 4. Monitoring (depends on regional stacks)
3749 pre_regional_stacks = [s for s in ordered_stacks if s.endswith(("-global", "-api-gateway"))]
3750 regional_api_stacks = [
3751 s
3752 for s in ordered_stacks
3753 if _is_regional_api_bridge_stack(
3754 s,
3755 project_name=project_name,
3756 stack_names=stack_names,
3757 )
3758 ]
3759 regional_stacks = [
3760 s
3761 for s in ordered_stacks
3762 if not s.endswith(("-global", "-api-gateway", "-monitoring"))
3763 and not _is_regional_api_bridge_stack(
3764 s,
3765 project_name=project_name,
3766 stack_names=stack_names,
3767 )
3768 ]
3769 post_regional_stacks = [s for s in ordered_stacks if s.endswith("-monitoring")]
3771 successful: list[str] = []
3772 failed: list[str] = []
3773 deployment_safety: _StackOperationSafetyKwargs = {
3774 "allow_bootstrap": allow_bootstrap,
3775 "bootstrap_stacks": bootstrap_stacks,
3776 "expected_stack_ids": expected_stack_ids,
3777 "prepared_change_sets": prepared_change_sets,
3778 "authorize_stack": authorize_stack,
3779 "strict_deployment_token": strict_deployment_token,
3780 "on_change_set_prepared": on_change_set_prepared,
3781 "on_ecr_repository_created": on_ecr_repository_created,
3782 }
3784 # Phase 1: Deploy pre-regional global stacks sequentially
3785 for stack_name in pre_regional_stacks:
3786 if on_stack_start:
3787 on_stack_start(stack_name)
3789 success = self.deploy(
3790 stack_name=stack_name,
3791 require_approval=require_approval,
3792 outputs_file=outputs_file,
3793 parameters=parameters,
3794 tags=tags,
3795 progress=progress,
3796 **deployment_safety,
3797 )
3799 if success:
3800 successful.append(stack_name)
3801 else:
3802 failed.append(stack_name)
3804 if on_stack_complete:
3805 on_stack_complete(stack_name, success)
3807 # Stop on failure to prevent cascading issues
3808 if not success:
3809 return False, successful, failed
3811 # Phase 2: Deploy regional stacks (parallel or sequential)
3812 # All regional stacks pass --exclusively: globals are already deployed
3813 # in Phase 1, so CDK doesn't need to re-evaluate them. Skipping that
3814 # re-evaluation avoids re-running custom resources (notably
3815 # KubectlApplyManifests) on the global stacks every time a regional
3816 # stack is deployed — that would otherwise re-apply manifests and
3817 # rollout-restart controllers for no actual change.
3818 if regional_stacks: 3818 ↛ 3877line 3818 didn't jump to line 3877 because the condition on line 3818 was always true
3819 if parallel and len(regional_stacks) > 1:
3820 # Parallel deployment of regional stacks
3821 successful_regional, failed_regional = self._deploy_stacks_parallel(
3822 stacks=regional_stacks,
3823 require_approval=require_approval,
3824 outputs_file=outputs_file,
3825 parameters=parameters,
3826 tags=tags,
3827 progress=progress,
3828 on_stack_start=on_stack_start,
3829 on_stack_complete=on_stack_complete,
3830 max_workers=max_workers,
3831 allow_bootstrap=allow_bootstrap,
3832 bootstrap_stacks=bootstrap_stacks,
3833 expected_stack_ids=expected_stack_ids,
3834 prepared_change_sets=prepared_change_sets,
3835 authorize_stack=authorize_stack,
3836 strict_deployment_token=strict_deployment_token,
3837 on_change_set_prepared=on_change_set_prepared,
3838 on_ecr_repository_created=on_ecr_repository_created,
3839 )
3840 successful.extend(successful_regional)
3841 failed.extend(failed_regional)
3843 # Stop if any regional stack failed
3844 if failed_regional:
3845 return False, successful, failed
3846 else:
3847 # Sequential deployment
3848 for stack_name in regional_stacks:
3849 if on_stack_start:
3850 on_stack_start(stack_name)
3852 success = self.deploy(
3853 stack_name=stack_name,
3854 require_approval=require_approval,
3855 outputs_file=outputs_file,
3856 parameters=parameters,
3857 tags=tags,
3858 progress=progress,
3859 exclusively=True,
3860 **deployment_safety,
3861 )
3863 if success: 3863 ↛ 3866line 3863 didn't jump to line 3866 because the condition on line 3863 was always true
3864 successful.append(stack_name)
3865 else:
3866 failed.append(stack_name)
3868 if on_stack_complete:
3869 on_stack_complete(stack_name, success)
3871 # Stop on failure
3872 if not success: 3872 ↛ 3873line 3872 didn't jump to line 3873 because the condition on line 3872 was never true
3873 return False, successful, failed
3875 # Phase 3: Deploy regional API bridges only after every base regional
3876 # stack is complete. Bridges within this level remain parallel-safe.
3877 if regional_api_stacks:
3878 if parallel and len(regional_api_stacks) > 1:
3879 successful_api, failed_api = self._deploy_stacks_parallel(
3880 stacks=regional_api_stacks,
3881 require_approval=require_approval,
3882 outputs_file=outputs_file,
3883 parameters=parameters,
3884 tags=tags,
3885 progress=progress,
3886 on_stack_start=on_stack_start,
3887 on_stack_complete=on_stack_complete,
3888 max_workers=max_workers,
3889 allow_bootstrap=allow_bootstrap,
3890 bootstrap_stacks=bootstrap_stacks,
3891 expected_stack_ids=expected_stack_ids,
3892 prepared_change_sets=prepared_change_sets,
3893 authorize_stack=authorize_stack,
3894 strict_deployment_token=strict_deployment_token,
3895 on_change_set_prepared=on_change_set_prepared,
3896 on_ecr_repository_created=on_ecr_repository_created,
3897 )
3898 successful.extend(successful_api)
3899 failed.extend(failed_api)
3900 if failed_api: 3900 ↛ 3901line 3900 didn't jump to line 3901 because the condition on line 3900 was never true
3901 return False, successful, failed
3902 else:
3903 for stack_name in regional_api_stacks:
3904 if on_stack_start: 3904 ↛ 3905line 3904 didn't jump to line 3905 because the condition on line 3904 was never true
3905 on_stack_start(stack_name)
3907 success = self.deploy(
3908 stack_name=stack_name,
3909 require_approval=require_approval,
3910 outputs_file=outputs_file,
3911 parameters=parameters,
3912 tags=tags,
3913 progress=progress,
3914 exclusively=True,
3915 **deployment_safety,
3916 )
3917 if success: 3917 ↛ 3920line 3917 didn't jump to line 3920 because the condition on line 3917 was always true
3918 successful.append(stack_name)
3919 else:
3920 failed.append(stack_name)
3921 if on_stack_complete: 3921 ↛ 3922line 3921 didn't jump to line 3922 because the condition on line 3921 was never true
3922 on_stack_complete(stack_name, success)
3923 if not success: 3923 ↛ 3924line 3923 didn't jump to line 3924 because the condition on line 3923 was never true
3924 return False, successful, failed
3926 # Phase 4: Deploy post-regional stacks (monitoring) sequentially.
3927 # Same rationale as Phase 2: every upstream stack is already
3928 # deployed, so --exclusively prevents a redundant pass over
3929 # global/api-gateway/regional.
3930 for stack_name in post_regional_stacks:
3931 if on_stack_start: 3931 ↛ 3932line 3931 didn't jump to line 3932 because the condition on line 3931 was never true
3932 on_stack_start(stack_name)
3934 success = self.deploy(
3935 stack_name=stack_name,
3936 require_approval=require_approval,
3937 outputs_file=outputs_file,
3938 parameters=parameters,
3939 tags=tags,
3940 progress=progress,
3941 exclusively=True,
3942 **deployment_safety,
3943 )
3945 if success: 3945 ↛ 3948line 3945 didn't jump to line 3948 because the condition on line 3945 was always true
3946 successful.append(stack_name)
3947 else:
3948 failed.append(stack_name)
3950 if on_stack_complete: 3950 ↛ 3951line 3950 didn't jump to line 3951 because the condition on line 3950 was never true
3951 on_stack_complete(stack_name, success)
3953 if not success: 3953 ↛ 3954line 3953 didn't jump to line 3954 because the condition on line 3953 was never true
3954 return False, successful, failed
3956 return len(failed) == 0, successful, failed
3958 def _deploy_stacks_parallel(
3959 self,
3960 stacks: list[str],
3961 require_approval: bool,
3962 outputs_file: str | None,
3963 parameters: dict[str, str] | None,
3964 tags: dict[str, str] | None,
3965 progress: str,
3966 on_stack_start: Callable[[str], None] | None,
3967 on_stack_complete: Callable[[str, bool], None] | None,
3968 max_workers: int,
3969 allow_bootstrap: bool,
3970 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None,
3971 expected_stack_ids: Mapping[str, str | None] | None,
3972 prepared_change_sets: PreparedChangeSetAuthority | None,
3973 authorize_stack: StackAuthorizationCallback | None,
3974 strict_deployment_token: str | None = None,
3975 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
3976 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
3977 ) -> tuple[list[str], list[str]]:
3978 """Deploy multiple stacks in parallel using separate CDK output directories."""
3979 import tempfile
3981 successful: list[str] = []
3982 failed: list[str] = []
3983 lock = Lock()
3985 def deploy_single(stack_name: str) -> tuple[str, bool]:
3986 # Use a unique output directory in /tmp for each parallel deployment
3987 # This avoids CDK copying cdk.out.* directories into assets
3988 output_dir = tempfile.mkdtemp(prefix=f"cdk-{stack_name}-")
3989 try:
3990 if on_stack_start: 3990 ↛ 3991line 3990 didn't jump to line 3991 because the condition on line 3990 was never true
3991 with lock:
3992 on_stack_start(stack_name)
3994 success = self.deploy(
3995 stack_name=stack_name,
3996 require_approval=require_approval,
3997 outputs_file=outputs_file,
3998 parameters=parameters,
3999 tags=tags,
4000 progress=progress,
4001 output_dir=output_dir,
4002 exclusively=True,
4003 allow_bootstrap=allow_bootstrap,
4004 bootstrap_stacks=bootstrap_stacks,
4005 expected_stack_ids=expected_stack_ids,
4006 prepared_change_sets=prepared_change_sets,
4007 authorize_stack=authorize_stack,
4008 strict_deployment_token=strict_deployment_token,
4009 on_change_set_prepared=on_change_set_prepared,
4010 on_ecr_repository_created=on_ecr_repository_created,
4011 )
4012 return stack_name, success
4013 finally:
4014 try:
4015 import shutil
4017 if os.path.exists(output_dir):
4018 shutil.rmtree(output_dir)
4019 except Exception as e:
4020 logger.debug("Cleanup of %s failed: %s", output_dir, e)
4022 self._cdk_cancel_event.clear()
4023 futures: dict[Any, str] = {}
4024 executor = ThreadPoolExecutor(max_workers=max_workers)
4025 try:
4026 futures = {executor.submit(deploy_single, stack): stack for stack in stacks}
4028 for future in as_completed(futures):
4029 stack_name, success = future.result()
4031 with lock:
4032 if success:
4033 successful.append(stack_name)
4034 else:
4035 failed.append(stack_name)
4037 if on_stack_complete: 4037 ↛ 4038line 4037 didn't jump to line 4038 because the condition on line 4037 was never true
4038 on_stack_complete(stack_name, success)
4039 except BaseException:
4040 # Terminate registered process groups before waiting for executor
4041 # shutdown; the context-manager form waits first and can deadlock an
4042 # interrupted orchestration behind a still-running CDK worker.
4043 self.cancel_active_cdk_processes()
4044 for future in futures:
4045 future.cancel()
4046 executor.shutdown(wait=True, cancel_futures=True)
4047 raise
4048 else:
4049 executor.shutdown(wait=True)
4050 finally:
4051 self._cdk_cancel_event.clear()
4053 return successful, failed
4055 def _resolve_strict_teardown_resources(
4056 self,
4057 *,
4058 stacks: Collection[str],
4059 regional_stacks: Collection[str],
4060 expected_stack_ids: Mapping[str, str | None],
4061 authorize_stack: StackAuthorizationCallback,
4062 ) -> dict[str, dict[str, str]]:
4063 """Authorize every live stack, then resolve helper IDs from exact stack ARNs."""
4064 import boto3
4066 live_targets: dict[str, tuple[str, Any, dict[str, Any]]] = {}
4067 for stack_name in stacks:
4068 expected_stack_id = expected_stack_ids[stack_name]
4069 target = self._describe_stack_target(
4070 stack_name,
4071 expected_stack_id=expected_stack_id,
4072 require_expected_identity=True,
4073 )
4074 if target is None: 4074 ↛ 4075line 4074 didn't jump to line 4075 because the condition on line 4074 was never true
4075 continue
4076 region, _cloudformation, stack = target
4077 stack_id = str(stack["StackId"])
4078 authorize_stack(stack_name, region, stack_id)
4079 live_targets[stack_name] = target
4081 project_name = self.config.project_name
4082 base_regional_stacks: list[str] = []
4083 for stack_name in regional_stacks:
4084 deploy_region = self._get_deploy_region(stack_name)
4085 if deploy_region and stack_name == f"{project_name}-{deploy_region}": 4085 ↛ 4083line 4085 didn't jump to line 4083 because the condition on line 4085 was always true
4086 base_regional_stacks.append(stack_name)
4088 resolved: dict[str, dict[str, str]] = {}
4089 for stack_name in base_regional_stacks:
4090 target = live_targets.get(stack_name)
4091 if target is None: 4091 ↛ 4092line 4091 didn't jump to line 4092 because the condition on line 4091 was never true
4092 continue
4093 region, cloudformation, stack = target
4094 stack_id = str(stack["StackId"])
4095 summaries: list[dict[str, Any]] = []
4096 paginator = cloudformation.get_paginator("list_stack_resources")
4097 for page in paginator.paginate(StackName=stack_id):
4098 summaries.extend(page.get("StackResourceSummaries", []))
4100 vpc_ids = {
4101 str(item["PhysicalResourceId"])
4102 for item in summaries
4103 if item.get("ResourceType") == "AWS::EC2::VPC" and item.get("PhysicalResourceId")
4104 }
4105 cluster_names = {
4106 str(item["PhysicalResourceId"])
4107 for item in summaries
4108 if item.get("ResourceType") == "AWS::EKS::Cluster"
4109 and item.get("PhysicalResourceId")
4110 }
4111 if len(vpc_ids) > 1 or len(cluster_names) > 1: 4111 ↛ 4112line 4111 didn't jump to line 4112 because the condition on line 4111 was never true
4112 raise RuntimeError(f"Exact stack {stack_id} returned ambiguous VPC/EKS resources")
4114 details = {
4115 "stack_name": stack_name,
4116 "stack_id": stack_id,
4117 "region": region,
4118 }
4119 vpc_id = next(iter(vpc_ids), "")
4120 cluster_name = next(iter(cluster_names), "")
4121 if vpc_id: 4121 ↛ 4123line 4121 didn't jump to line 4123 because the condition on line 4121 was always true
4122 details["vpc_id"] = vpc_id
4123 if cluster_name: 4123 ↛ 4174line 4123 didn't jump to line 4174 because the condition on line 4123 was always true
4124 details["cluster_name"] = cluster_name
4125 cluster: dict[str, Any] | None
4126 try:
4127 cluster = boto3.client("eks", region_name=region).describe_cluster(
4128 name=cluster_name
4129 )["cluster"]
4130 except ClientError as exc:
4131 if exc.response.get("Error", {}).get("Code") == "ResourceNotFoundException":
4132 cluster = None
4133 else:
4134 raise
4135 if cluster is not None: 4135 ↛ 4152line 4135 didn't jump to line 4152 because the condition on line 4135 was always true
4136 if str(cluster.get("name") or "") != cluster_name: 4136 ↛ 4137line 4136 didn't jump to line 4137 because the condition on line 4136 was never true
4137 raise RuntimeError(
4138 f"EKS returned a changed identity for {region}:{cluster_name}"
4139 )
4140 networking = cluster.get("resourcesVpcConfig") or {}
4141 cluster_vpc_id = str(networking.get("vpcId") or "")
4142 security_group_id = str(networking.get("clusterSecurityGroupId") or "")
4143 if vpc_id and cluster_vpc_id != vpc_id: 4143 ↛ 4144line 4143 didn't jump to line 4144 because the condition on line 4143 was never true
4144 raise RuntimeError(
4145 f"EKS cluster {cluster_name} no longer belongs to exact VPC {vpc_id}"
4146 )
4147 if not security_group_id: 4147 ↛ 4148line 4147 didn't jump to line 4148 because the condition on line 4147 was never true
4148 raise RuntimeError(
4149 f"EKS cluster {cluster_name} omitted its security-group identity"
4150 )
4151 details["cluster_security_group_id"] = security_group_id
4152 elif vpc_id:
4153 # On teardown resume the cluster may already be gone while
4154 # its managed SG remains. Resolve the SG ID inside the exact
4155 # stack VPC using the exact cluster physical ID.
4156 ec2 = boto3.client("ec2", region_name=region)
4157 groups = ec2.describe_security_groups(
4158 Filters=[
4159 {"Name": "vpc-id", "Values": [vpc_id]},
4160 {
4161 "Name": "tag:aws:eks:cluster-name",
4162 "Values": [cluster_name],
4163 },
4164 ]
4165 ).get("SecurityGroups", [])
4166 group_ids = {str(group["GroupId"]) for group in groups if group.get("GroupId")}
4167 if len(group_ids) > 1:
4168 raise RuntimeError(
4169 f"Exact VPC {vpc_id} has ambiguous EKS security groups for "
4170 f"{cluster_name}"
4171 )
4172 if group_ids:
4173 details["cluster_security_group_id"] = next(iter(group_ids))
4174 resolved[stack_name] = details
4175 return resolved
4177 def _destroy_phase_remaining_stacks(
4178 self,
4179 phase_name: str,
4180 stacks: Collection[str],
4181 expected_stack_ids: Mapping[str, str | None] | None = None,
4182 ) -> list[str]:
4183 """Return stacks still present after a dependency phase.
4185 A lookup error is treated as present: advancing when absence cannot be
4186 proven is less safe than stopping for an operator retry.
4187 """
4188 remaining: list[str] = []
4189 for stack_name in stacks:
4190 try:
4191 present = self._stack_exists_in_cloudformation(
4192 stack_name,
4193 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4194 require_expected_identity=expected_stack_ids is not None,
4195 )
4196 except Exception:
4197 logger.exception(
4198 "Could not verify %s absence after %s",
4199 stack_name,
4200 phase_name,
4201 )
4202 present = True
4203 if present:
4204 remaining.append(stack_name)
4205 if remaining:
4206 print(
4207 f" {phase_name} barrier blocked: stack absence was not confirmed for "
4208 + ", ".join(remaining)
4209 )
4210 return remaining
4212 def destroy_orchestrated(
4213 self,
4214 force: bool = False,
4215 on_stack_start: Callable[[str], None] | None = None,
4216 on_stack_complete: Callable[[str, bool], None] | None = None,
4217 parallel: bool = False,
4218 max_workers: int = 4,
4219 expected_stack_ids: Mapping[str, str | None] | None = None,
4220 prepared_change_sets: PreparedChangeSetAuthority | None = None,
4221 authorize_stack: StackAuthorizationCallback | None = None,
4222 allow_bootstrap: bool = True,
4223 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
4224 on_cleanup_complete: CleanupOutcomeCallback | None = None,
4225 strict_deployment_token: str | None = None,
4226 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
4227 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
4228 ) -> tuple[bool, list[str], list[str]]:
4229 """Destroy stacks in dependency order with optional exact-ARN authority."""
4230 app_stacks = self.list_stacks()
4231 strict_identity = expected_stack_ids is not None
4232 if strict_identity:
4233 assert expected_stack_ids is not None
4234 missing = sorted(set(app_stacks) - set(expected_stack_ids))
4235 if missing:
4236 raise RuntimeError(
4237 f"Strict teardown target map is incomplete before cleanup; missing={missing}"
4238 )
4239 if authorize_stack is None: 4239 ↛ 4240line 4239 didn't jump to line 4240 because the condition on line 4239 was never true
4240 raise RuntimeError("Strict teardown requires an exact stack authorizer")
4241 invalid = sorted(
4242 name
4243 for name, stack_id in expected_stack_ids.items()
4244 if stack_id is not None and not str(stack_id).startswith("arn:")
4245 )
4246 if invalid: 4246 ↛ 4247line 4246 didn't jump to line 4247 because the condition on line 4246 was never true
4247 raise RuntimeError(
4248 f"Strict teardown has invalid stack identities for: {', '.join(invalid)}"
4249 )
4251 strict_prepared_deployment = (
4252 strict_deployment_token is not None or on_change_set_prepared is not None
4253 )
4254 if strict_prepared_deployment:
4255 if not strict_deployment_token or on_change_set_prepared is None: 4255 ↛ 4256line 4255 didn't jump to line 4256 because the condition on line 4255 was never true
4256 raise RuntimeError(
4257 "Strict teardown dependency deployment requires both a run token "
4258 "and a prepared-change-set callback"
4259 )
4260 if prepared_change_sets is None: 4260 ↛ 4261line 4260 didn't jump to line 4261 because the condition on line 4260 was never true
4261 raise RuntimeError("Strict teardown lacks prepared change-set history")
4262 expected_history_keys = set(expected_stack_ids or {})
4263 if set(prepared_change_sets) != expected_history_keys: 4263 ↛ 4264line 4263 didn't jump to line 4264 because the condition on line 4263 was never true
4264 raise RuntimeError(
4265 "Strict teardown change-set history does not match target identities"
4266 )
4268 stacks = list(app_stacks)
4269 if expected_stack_ids is not None:
4270 for stack_name in expected_stack_ids:
4271 if stack_name not in stacks: 4271 ↛ 4272line 4271 didn't jump to line 4272 because the condition on line 4271 was never true
4272 stacks.append(stack_name)
4273 project_name = self.config.project_name
4274 (
4275 post_regional_stacks,
4276 regional_api_stacks,
4277 regional_stacks,
4278 pre_regional_stacks,
4279 ) = _get_stack_destroy_phases(stacks, project_name=project_name)
4281 strict_resources: dict[str, dict[str, str]] = {}
4282 if strict_identity:
4283 assert expected_stack_ids is not None
4284 assert authorize_stack is not None
4285 strict_resources = self._resolve_strict_teardown_resources(
4286 stacks=stacks,
4287 regional_stacks=regional_stacks,
4288 expected_stack_ids=expected_stack_ids,
4289 authorize_stack=authorize_stack,
4290 )
4292 destroy_safety: _StackOperationSafetyKwargs = {
4293 "expected_stack_ids": expected_stack_ids,
4294 "prepared_change_sets": prepared_change_sets,
4295 "authorize_stack": authorize_stack,
4296 "allow_bootstrap": allow_bootstrap,
4297 "bootstrap_stacks": bootstrap_stacks,
4298 "strict_deployment_token": strict_deployment_token,
4299 "on_change_set_prepared": on_change_set_prepared,
4300 "on_ecr_repository_created": on_ecr_repository_created,
4301 }
4303 def record_cleanup(name: str, details: dict[str, Any]) -> None:
4304 if on_cleanup_complete is not None:
4305 on_cleanup_complete(name, details)
4307 if not self._image_registry_destroy_preflight(force=force):
4308 return False, [], list(stacks)
4310 bastion_targets = {
4311 name: details for name, details in strict_resources.items() if details.get("vpc_id")
4312 }
4313 bastions = self.cleanup_orphaned_bastions(
4314 stacks,
4315 parallel=parallel,
4316 resource_targets=bastion_targets if strict_identity else None,
4317 )
4318 record_cleanup("bastions", {"terminated_instances": bastions})
4320 # Non-strict teardowns also retire the bastion's standing IAM
4321 # role/profile and, below, the implicit log groups CloudFormation
4322 # never modeled. Strict (live-validation) teardowns skip both: the
4323 # harness owns fenced log-group deletion and audits IAM itself.
4324 if not strict_identity:
4325 record_cleanup("bastion-iam", self._cleanup_bastion_iam())
4327 global_stack_name = f"{project_name}-global"
4328 backup = self._cleanup_backup_vault(
4329 expected_stack_id=(expected_stack_ids or {}).get(global_stack_name),
4330 authorize_stack=authorize_stack,
4331 require_expected_identity=strict_identity,
4332 )
4333 record_cleanup("backup-vault", backup)
4334 if strict_identity and backup.get("errors"): 4334 ↛ 4335line 4334 didn't jump to line 4335 because the condition on line 4334 was never true
4335 raise RuntimeError(
4336 "Strict backup-vault cleanup failed before stack deletion: "
4337 + json.dumps(backup["errors"], sort_keys=True)
4338 )
4340 successful: list[str] = []
4341 failed: list[str] = []
4343 # Capture implicit log-group names while the source stacks still
4344 # exist; the exact derived names are deleted by ``finish`` below
4345 # once their stacks are gone. Strict teardowns collect nothing —
4346 # the live-validation harness owns fenced log-group deletion.
4347 implicit_log_groups: dict[str, dict[str, Any]] = {}
4348 if not strict_identity:
4349 implicit_log_groups = self._collect_implicit_log_groups(stacks)
4351 def finish(overall: bool) -> tuple[bool, list[str], list[str]]:
4352 """Funnel every exit through the implicit log-group sweep.
4354 Called at each return point so a partially failed teardown
4355 still cleans up the stacks that DID delete. New exit paths
4356 must return through here as well.
4357 """
4358 if implicit_log_groups:
4359 record_cleanup(
4360 "implicit-log-groups",
4361 self._cleanup_implicit_log_groups(implicit_log_groups, successful),
4362 )
4363 return overall, successful, failed
4365 for stack_name in post_regional_stacks:
4366 if on_stack_start: 4366 ↛ 4367line 4366 didn't jump to line 4367 because the condition on line 4366 was never true
4367 on_stack_start(stack_name)
4368 success = self.destroy(
4369 stack_name=stack_name,
4370 force=force,
4371 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4372 **destroy_safety,
4373 )
4374 (successful if success else failed).append(stack_name)
4375 if on_stack_complete: 4375 ↛ 4376line 4375 didn't jump to line 4376 because the condition on line 4375 was never true
4376 on_stack_complete(stack_name, success)
4378 phase_remaining = self._destroy_phase_remaining_stacks(
4379 "post-regional",
4380 post_regional_stacks,
4381 expected_stack_ids,
4382 )
4383 for stack_name in phase_remaining: 4383 ↛ 4384line 4383 didn't jump to line 4384 because the loop on line 4383 never started
4384 if stack_name not in failed:
4385 failed.append(stack_name)
4386 if any(stack in failed for stack in post_regional_stacks) or phase_remaining: 4386 ↛ 4387line 4386 didn't jump to line 4387 because the condition on line 4386 was never true
4387 return finish(False)
4389 if regional_api_stacks:
4390 if parallel and len(regional_api_stacks) > 1:
4391 successful_api, failed_api = self._destroy_stacks_parallel(
4392 stacks=regional_api_stacks,
4393 force=force,
4394 on_stack_start=on_stack_start,
4395 on_stack_complete=on_stack_complete,
4396 max_workers=max_workers,
4397 expected_stack_ids=expected_stack_ids,
4398 authorize_stack=authorize_stack,
4399 allow_bootstrap=allow_bootstrap,
4400 bootstrap_stacks=bootstrap_stacks,
4401 prepared_change_sets=prepared_change_sets,
4402 strict_deployment_token=strict_deployment_token,
4403 on_change_set_prepared=on_change_set_prepared,
4404 on_ecr_repository_created=on_ecr_repository_created,
4405 )
4406 successful.extend(successful_api)
4407 failed.extend(failed_api)
4408 else:
4409 for stack_name in regional_api_stacks:
4410 if on_stack_start: 4410 ↛ 4411line 4410 didn't jump to line 4411 because the condition on line 4410 was never true
4411 on_stack_start(stack_name)
4412 success = self.destroy(
4413 stack_name=stack_name,
4414 force=force,
4415 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4416 **destroy_safety,
4417 )
4418 (successful if success else failed).append(stack_name)
4419 if on_stack_complete: 4419 ↛ 4420line 4419 didn't jump to line 4420 because the condition on line 4419 was never true
4420 on_stack_complete(stack_name, success)
4421 phase_remaining = self._destroy_phase_remaining_stacks(
4422 "regional API bridge",
4423 regional_api_stacks,
4424 expected_stack_ids,
4425 )
4426 for stack_name in phase_remaining: 4426 ↛ 4427line 4426 didn't jump to line 4427 because the loop on line 4426 never started
4427 if stack_name not in failed:
4428 failed.append(stack_name)
4429 if any(stack in failed for stack in regional_api_stacks) or phase_remaining: 4429 ↛ 4430line 4429 didn't jump to line 4430 because the condition on line 4429 was never true
4430 return finish(False)
4432 watchdog_stops: dict[str, Event] = {}
4433 watchdog_threads: dict[str, Thread] = {}
4434 watchdog_targets = (
4435 [
4436 name
4437 for name in regional_stacks
4438 if strict_resources.get(name, {}).get("cluster_security_group_id")
4439 ]
4440 if strict_identity
4441 else list(regional_stacks)
4442 )
4443 try:
4444 for stack_name in watchdog_targets:
4445 details = strict_resources.get(stack_name, {})
4446 stop_event = Event()
4447 watchdog_stops[stack_name] = stop_event
4448 watchdog_threads[stack_name] = self._start_eks_sg_watchdog(
4449 stack_name,
4450 stop_event,
4451 region=details.get("region"),
4452 security_group_id=details.get("cluster_security_group_id"),
4453 vpc_id=details.get("vpc_id"),
4454 )
4456 if regional_stacks: 4456 ↛ 4489line 4456 didn't jump to line 4489 because the condition on line 4456 was always true
4457 if parallel and len(regional_stacks) > 1:
4458 successful_regional, failed_regional = self._destroy_stacks_parallel(
4459 stacks=regional_stacks,
4460 force=force,
4461 on_stack_start=on_stack_start,
4462 on_stack_complete=on_stack_complete,
4463 max_workers=max_workers,
4464 expected_stack_ids=expected_stack_ids,
4465 authorize_stack=authorize_stack,
4466 allow_bootstrap=allow_bootstrap,
4467 bootstrap_stacks=bootstrap_stacks,
4468 prepared_change_sets=prepared_change_sets,
4469 strict_deployment_token=strict_deployment_token,
4470 on_change_set_prepared=on_change_set_prepared,
4471 on_ecr_repository_created=on_ecr_repository_created,
4472 )
4473 successful.extend(successful_regional)
4474 failed.extend(failed_regional)
4475 else:
4476 for stack_name in regional_stacks:
4477 if on_stack_start:
4478 on_stack_start(stack_name)
4479 success = self.destroy(
4480 stack_name=stack_name,
4481 force=force,
4482 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4483 **destroy_safety,
4484 )
4485 (successful if success else failed).append(stack_name)
4486 if on_stack_complete:
4487 on_stack_complete(stack_name, success)
4488 finally:
4489 for stop_event in watchdog_stops.values():
4490 stop_event.set()
4491 for stack_name, thread in watchdog_threads.items():
4492 try:
4493 thread.join(timeout=5)
4494 except Exception as exc:
4495 logger.exception("Could not join teardown watchdog for %s", stack_name)
4496 record_cleanup(
4497 "eks-security-group",
4498 {"stack": stack_name, "errors": [f"{type(exc).__name__}: {exc}"]},
4499 )
4500 if strict_identity and stack_name not in failed:
4501 failed.append(stack_name)
4502 continue
4503 details = strict_resources.get(stack_name, {})
4504 if strict_identity and thread.is_alive(): 4504 ↛ 4505line 4504 didn't jump to line 4505 because the condition on line 4504 was never true
4505 outcome = {
4506 "stack": stack_name,
4507 "errors": ["watchdog thread did not stop"],
4508 }
4509 else:
4510 outcome = self._cleanup_eks_security_groups(
4511 stack_name,
4512 region=details.get("region"),
4513 security_group_id=details.get("cluster_security_group_id"),
4514 vpc_id=details.get("vpc_id"),
4515 )
4516 record_cleanup("eks-security-group", outcome)
4517 if (
4518 strict_identity
4519 and (outcome.get("errors") or outcome.get("blocked_by_enis"))
4520 and stack_name not in failed
4521 ):
4522 failed.append(stack_name)
4524 phase_remaining = self._destroy_phase_remaining_stacks(
4525 "regional",
4526 regional_stacks,
4527 expected_stack_ids,
4528 )
4529 for stack_name in phase_remaining:
4530 if stack_name not in failed:
4531 failed.append(stack_name)
4532 if any(stack in failed for stack in regional_stacks) or phase_remaining:
4533 return finish(False)
4535 for stack_name in pre_regional_stacks:
4536 if on_stack_start:
4537 on_stack_start(stack_name)
4538 success = self.destroy(
4539 stack_name=stack_name,
4540 force=force,
4541 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4542 **destroy_safety,
4543 )
4544 (successful if success else failed).append(stack_name)
4545 if on_stack_complete:
4546 on_stack_complete(stack_name, success)
4548 phase_remaining = self._destroy_phase_remaining_stacks(
4549 "pre-regional global",
4550 [stack_name],
4551 expected_stack_ids,
4552 )
4553 for remaining_stack in phase_remaining: 4553 ↛ 4554line 4553 didn't jump to line 4554 because the loop on line 4553 never started
4554 if remaining_stack not in failed:
4555 failed.append(remaining_stack)
4556 if not success or phase_remaining:
4557 return finish(False)
4559 return finish(len(failed) == 0)
4561 def _destroy_stacks_parallel(
4562 self,
4563 stacks: list[str],
4564 force: bool,
4565 on_stack_start: Callable[[str], None] | None,
4566 on_stack_complete: Callable[[str, bool], None] | None,
4567 max_workers: int,
4568 expected_stack_ids: Mapping[str, str | None] | None,
4569 authorize_stack: StackAuthorizationCallback | None,
4570 allow_bootstrap: bool,
4571 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None,
4572 prepared_change_sets: PreparedChangeSetAuthority | None,
4573 strict_deployment_token: str | None = None,
4574 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
4575 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
4576 ) -> tuple[list[str], list[str]]:
4577 """Destroy multiple stacks in parallel using separate CDK output directories."""
4578 import tempfile
4580 successful: list[str] = []
4581 failed: list[str] = []
4582 lock = Lock()
4584 def destroy_single(stack_name: str) -> tuple[str, bool]:
4585 # Use a unique output directory in /tmp for each parallel destruction
4586 output_dir = tempfile.mkdtemp(prefix=f"cdk-{stack_name}-")
4587 try:
4588 if on_stack_start: 4588 ↛ 4589line 4588 didn't jump to line 4589 because the condition on line 4588 was never true
4589 with lock:
4590 on_stack_start(stack_name)
4592 success = self.destroy(
4593 stack_name=stack_name,
4594 force=force,
4595 output_dir=output_dir,
4596 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4597 expected_stack_ids=expected_stack_ids,
4598 authorize_stack=authorize_stack,
4599 allow_bootstrap=allow_bootstrap,
4600 bootstrap_stacks=bootstrap_stacks,
4601 prepared_change_sets=prepared_change_sets,
4602 strict_deployment_token=strict_deployment_token,
4603 on_change_set_prepared=on_change_set_prepared,
4604 on_ecr_repository_created=on_ecr_repository_created,
4605 )
4606 return stack_name, success
4607 finally:
4608 try:
4609 import shutil
4611 if os.path.exists(output_dir):
4612 shutil.rmtree(output_dir)
4613 except Exception as e:
4614 logger.debug("Cleanup of %s failed: %s", output_dir, e)
4616 self._cdk_cancel_event.clear()
4617 futures: dict[Any, str] = {}
4618 executor = ThreadPoolExecutor(max_workers=max_workers)
4619 try:
4620 futures = {executor.submit(destroy_single, stack): stack for stack in stacks}
4622 for future in as_completed(futures):
4623 stack_name, success = future.result()
4625 with lock:
4626 if success:
4627 successful.append(stack_name)
4628 else:
4629 failed.append(stack_name)
4631 if on_stack_complete: 4631 ↛ 4632line 4631 didn't jump to line 4632 because the condition on line 4631 was never true
4632 on_stack_complete(stack_name, success)
4633 except BaseException:
4634 self.cancel_active_cdk_processes()
4635 for future in futures:
4636 future.cancel()
4637 executor.shutdown(wait=True, cancel_futures=True)
4638 raise
4639 else:
4640 executor.shutdown(wait=True)
4641 finally:
4642 self._cdk_cancel_event.clear()
4644 return successful, failed
4646 def _cleanup_backup_vault(
4647 self,
4648 *,
4649 expected_stack_id: str | None = None,
4650 authorize_stack: StackAuthorizationCallback | None = None,
4651 require_expected_identity: bool = False,
4652 ) -> dict[str, Any]:
4653 """Delete points only from the exact stack resource's physical vault."""
4654 import boto3
4656 global_region = self.config.global_region
4657 global_stack_name = f"{self.config.project_name}-global"
4658 result: dict[str, Any] = {
4659 "stack_name": global_stack_name,
4660 "stack_id": expected_stack_id,
4661 "status": "not-needed",
4662 "deleted_recovery_points": 0,
4663 "errors": [],
4664 }
4666 try:
4667 target = self._describe_stack_target(
4668 global_stack_name,
4669 expected_stack_id=expected_stack_id,
4670 require_expected_identity=require_expected_identity,
4671 )
4672 if target is None: 4672 ↛ 4673line 4672 didn't jump to line 4673 because the condition on line 4672 was never true
4673 result["status"] = "stack-absent"
4674 return result
4675 region, cloudformation, stack = target
4676 stack_id = str(stack["StackId"])
4677 result["stack_id"] = stack_id
4678 if region != global_region: 4678 ↛ 4679line 4678 didn't jump to line 4679 because the condition on line 4678 was never true
4679 raise RuntimeError(f"Global stack resolved to {region}, expected {global_region}")
4680 if authorize_stack is not None: 4680 ↛ 4681line 4680 didn't jump to line 4681 because the condition on line 4680 was never true
4681 authorize_stack(global_stack_name, region, stack_id)
4683 resources: list[dict[str, Any]] = []
4684 for page in cloudformation.get_paginator("list_stack_resources").paginate(
4685 StackName=stack_id
4686 ):
4687 resources.extend(
4688 resource
4689 for resource in page.get("StackResourceSummaries", [])
4690 if resource.get("ResourceType") == "AWS::Backup::BackupVault"
4691 and resource.get("PhysicalResourceId")
4692 )
4693 if not resources:
4694 result["status"] = "vault-resource-absent"
4695 return result
4696 if len(resources) != 1: 4696 ↛ 4697line 4696 didn't jump to line 4697 because the condition on line 4696 was never true
4697 raise RuntimeError(
4698 f"Expected one AWS::Backup::BackupVault in {stack_id}; found {len(resources)}"
4699 )
4701 resource = resources[0]
4702 physical_id = str(resource["PhysicalResourceId"])
4703 if physical_id.startswith("arn:"): 4703 ↛ 4709line 4703 didn't jump to line 4709 because the condition on line 4703 was always true
4704 parts = physical_id.split(":", 5)
4705 if len(parts) != 6 or not parts[5].startswith("backup-vault:"): 4705 ↛ 4706line 4705 didn't jump to line 4706 because the condition on line 4705 was never true
4706 raise RuntimeError(f"Invalid backup vault physical ARN: {physical_id}")
4707 vault_name = parts[5].removeprefix("backup-vault:")
4708 else:
4709 vault_name = physical_id
4710 if not vault_name: 4710 ↛ 4711line 4710 didn't jump to line 4711 because the condition on line 4710 was never true
4711 raise RuntimeError("CloudFormation returned an empty backup vault physical ID")
4713 backup_client = boto3.client("backup", region_name=global_region)
4714 described_vault = backup_client.describe_backup_vault(BackupVaultName=vault_name)
4715 vault_arn = str(described_vault.get("BackupVaultArn") or "")
4716 arn_parts = vault_arn.split(":", 5)
4717 if ( 4717 ↛ 4723line 4717 didn't jump to line 4723 because the condition on line 4717 was never true
4718 len(arn_parts) != 6
4719 or arn_parts[2] != "backup"
4720 or arn_parts[3] != global_region
4721 or arn_parts[5] != f"backup-vault:{vault_name}"
4722 ):
4723 raise RuntimeError(
4724 "AWS Backup identity does not match the CloudFormation physical resource"
4725 )
4726 if physical_id.startswith("arn:") and physical_id != vault_arn: 4726 ↛ 4727line 4726 didn't jump to line 4727 because the condition on line 4726 was never true
4727 raise RuntimeError("Backup vault ARN changed after CloudFormation resolution")
4729 result.update(
4730 {
4731 "status": "inspected",
4732 "logical_id": str(resource.get("LogicalResourceId") or ""),
4733 "physical_id": physical_id,
4734 "vault_name": vault_name,
4735 "vault_arn": vault_arn,
4736 }
4737 )
4738 paginator = backup_client.get_paginator("list_recovery_points_by_backup_vault")
4739 for page in paginator.paginate(BackupVaultName=vault_name):
4740 for recovery_point in page.get("RecoveryPoints", []):
4741 recovery_point_arn = recovery_point.get("RecoveryPointArn")
4742 if not recovery_point_arn: 4742 ↛ 4743line 4742 didn't jump to line 4743 because the condition on line 4742 was never true
4743 continue
4744 try:
4745 backup_client.delete_recovery_point(
4746 BackupVaultName=vault_name,
4747 RecoveryPointArn=recovery_point_arn,
4748 )
4749 result["deleted_recovery_points"] += 1
4750 except Exception as exc:
4751 result["errors"].append(
4752 {
4753 "recovery_point_arn": str(recovery_point_arn),
4754 "error": f"{type(exc).__name__}: {exc}",
4755 }
4756 )
4757 if result["deleted_recovery_points"]:
4758 print(
4759 f" Cleaned up {result['deleted_recovery_points']} backup recovery "
4760 f"points from {vault_name}"
4761 )
4762 result["status"] = "completed" if not result["errors"] else "partial"
4763 except Exception as exc:
4764 result["status"] = "failed"
4765 result["errors"].append({"error": f"{type(exc).__name__}: {exc}"})
4766 print(f" Warning: Backup vault cleanup failed (non-fatal): {exc}")
4767 return result
4769 def cleanup_orphaned_bastions(
4770 self,
4771 stacks: list[str] | None = None,
4772 *,
4773 parallel: bool = True,
4774 resource_targets: Mapping[str, Mapping[str, str]] | None = None,
4775 ) -> int:
4776 """Terminate CLI bastions, using exact stack VPC IDs in strict mode."""
4777 if stacks is None: 4777 ↛ 4778line 4777 didn't jump to line 4778 because the condition on line 4777 was never true
4778 stacks = self.list_stacks()
4779 if resource_targets is not None: 4779 ↛ 4780line 4779 didn't jump to line 4780 because the condition on line 4779 was never true
4780 regional_stacks = [name for name in stacks if name in resource_targets]
4781 else:
4782 regional_stacks = [
4783 stack
4784 for stack in stacks
4785 if not stack.endswith(("-global", "-api-gateway", "-monitoring", "-analytics"))
4786 ]
4788 def cleanup_one(stack_name: str) -> int:
4789 details = (resource_targets or {}).get(stack_name, {})
4790 return self._cleanup_orphaned_bastions(
4791 stack_name,
4792 region=details.get("region"),
4793 vpc_id=details.get("vpc_id"),
4794 fail_closed=resource_targets is not None,
4795 )
4797 if not regional_stacks: 4797 ↛ 4798line 4797 didn't jump to line 4798 because the condition on line 4797 was never true
4798 return 0
4799 if len(regional_stacks) == 1 or not parallel:
4800 terminated = sum(cleanup_one(stack_name) for stack_name in regional_stacks)
4801 else:
4802 # Each region has independent EC2 waiters. Run them concurrently so
4803 # one slow termination does not add its full timeout to every other
4804 # region before CloudFormation can start deleting stacks.
4805 with ThreadPoolExecutor(max_workers=min(4, len(regional_stacks))) as executor:
4806 terminated = sum(executor.map(cleanup_one, regional_stacks))
4807 if terminated:
4808 print(
4809 f" Requested termination for {terminated} orphaned ephemeral "
4810 "SSM bastion(s) before stack deletion."
4811 )
4812 return terminated
4814 def _cleanup_orphaned_bastions(
4815 self,
4816 stack_name: str,
4817 *,
4818 region: str | None = None,
4819 vpc_id: str | None = None,
4820 fail_closed: bool = False,
4821 ) -> int:
4822 """Terminate tagged bastions only inside a resolved stack VPC."""
4823 import boto3
4825 from .ephemeral_bastion import (
4826 BASTION_PURPOSE,
4827 TAG_EPHEMERAL_KEY,
4828 TAG_PROJECT_KEY,
4829 TAG_PURPOSE_KEY,
4830 bastion_instance_name,
4831 )
4833 region = region or self._get_deploy_region(stack_name)
4834 if not region:
4835 if fail_closed: 4835 ↛ 4836line 4835 didn't jump to line 4836 because the condition on line 4835 was never true
4836 raise RuntimeError(f"Strict bastion cleanup lacks a Region for {stack_name}")
4837 return 0
4839 project_name = str(self.config.project_name)
4840 expected_name = bastion_instance_name(project_name)
4841 try:
4842 ec2 = boto3.client("ec2", region_name=region)
4843 if vpc_id: 4843 ↛ 4844line 4843 didn't jump to line 4844 because the condition on line 4843 was never true
4844 vpcs = [{"VpcId": vpc_id}]
4845 else:
4846 vpcs = ec2.describe_vpcs(
4847 Filters=[
4848 {
4849 "Name": "tag:aws:cloudformation:stack-name",
4850 "Values": [stack_name],
4851 }
4852 ]
4853 ).get("Vpcs", [])
4854 except Exception as exc:
4855 if fail_closed: 4855 ↛ 4856line 4855 didn't jump to line 4856 because the condition on line 4855 was never true
4856 raise RuntimeError(
4857 f"Strict bastion cleanup could not inspect {stack_name}"
4858 ) from exc
4859 print(f" Warning: Bastion cleanup could not inspect {stack_name}: {exc}")
4860 return 0
4862 instance_ids: list[str] = []
4863 eni_ids: list[str] = []
4864 for vpc in vpcs:
4865 candidate_vpc_id = str(vpc.get("VpcId") or "")
4866 if not candidate_vpc_id: 4866 ↛ 4867line 4866 didn't jump to line 4867 because the condition on line 4866 was never true
4867 if fail_closed:
4868 raise RuntimeError(f"Strict bastion cleanup has no VPC ID for {stack_name}")
4869 continue
4870 try:
4871 reservations = ec2.describe_instances(
4872 Filters=[
4873 {"Name": "vpc-id", "Values": [candidate_vpc_id]},
4874 {"Name": f"tag:{TAG_EPHEMERAL_KEY}", "Values": ["true"]},
4875 {"Name": f"tag:{TAG_PURPOSE_KEY}", "Values": [BASTION_PURPOSE]},
4876 {
4877 "Name": "instance-state-name",
4878 "Values": [
4879 "pending",
4880 "running",
4881 "stopping",
4882 "stopped",
4883 "shutting-down",
4884 ],
4885 },
4886 ]
4887 ).get("Reservations", [])
4888 except Exception as exc:
4889 if fail_closed:
4890 raise RuntimeError(
4891 f"Strict bastion lookup failed in {stack_name} ({candidate_vpc_id})"
4892 ) from exc
4893 logger.warning(
4894 "Bastion lookup failed in %s (%s): %s",
4895 stack_name,
4896 candidate_vpc_id,
4897 exc,
4898 )
4899 continue
4901 for reservation in reservations:
4902 for instance in reservation.get("Instances", []):
4903 tags = {
4904 str(tag.get("Key")): str(tag.get("Value"))
4905 for tag in instance.get("Tags", [])
4906 if tag.get("Key") is not None
4907 }
4908 tagged_project = tags.get(TAG_PROJECT_KEY)
4909 if tagged_project != project_name and not (
4910 tagged_project is None and tags.get("Name") == expected_name
4911 ):
4912 continue
4913 instance_id = instance.get("InstanceId")
4914 if instance_id: 4914 ↛ 4916line 4914 didn't jump to line 4916 because the condition on line 4914 was always true
4915 instance_ids.append(str(instance_id))
4916 for interface in instance.get("NetworkInterfaces", []):
4917 attachment = interface.get("Attachment") or {}
4918 if not (
4919 attachment.get("DeviceIndex") == 0
4920 and attachment.get("DeleteOnTermination") is True
4921 ):
4922 continue
4923 eni_id = interface.get("NetworkInterfaceId")
4924 if eni_id: 4924 ↛ 4916line 4924 didn't jump to line 4916 because the condition on line 4924 was always true
4925 eni_ids.append(str(eni_id))
4927 instance_ids = list(dict.fromkeys(instance_ids))
4928 eni_ids = list(dict.fromkeys(eni_ids))
4929 if not instance_ids:
4930 return 0
4932 try:
4933 ec2.terminate_instances(InstanceIds=instance_ids)
4934 except Exception as exc:
4935 if fail_closed:
4936 raise RuntimeError(f"Strict bastion termination failed in {stack_name}") from exc
4937 print(f" Warning: Failed to terminate ephemeral bastion(s) in {stack_name}: {exc}")
4938 return 0
4940 print(
4941 f" Terminating {len(instance_ids)} ephemeral SSM bastion(s) in "
4942 f"{stack_name}: {', '.join(instance_ids)}"
4943 )
4944 try:
4945 ec2.get_waiter("instance_terminated").wait(
4946 InstanceIds=instance_ids,
4947 WaiterConfig={"Delay": 5, "MaxAttempts": 60},
4948 )
4949 except Exception as exc:
4950 if fail_closed:
4951 raise RuntimeError(
4952 f"Strict bastion termination did not converge in {stack_name}"
4953 ) from exc
4954 logger.warning("Timed out waiting for bastion termination in %s: %s", stack_name, exc)
4956 remaining_enis = self._wait_for_bastion_network_interfaces(ec2, eni_ids)
4957 if remaining_enis: 4957 ↛ 4958line 4957 didn't jump to line 4958 because the condition on line 4957 was never true
4958 message = (
4959 f"{len(remaining_enis)} bastion network interface(s) in {stack_name} "
4960 f"have not released: {', '.join(sorted(remaining_enis))}"
4961 )
4962 if fail_closed:
4963 raise RuntimeError(message)
4964 print(f" Warning: {message}. The destroy retry will check again.")
4965 return len(instance_ids)
4967 @staticmethod
4968 def _wait_for_bastion_network_interfaces(
4969 ec2: Any,
4970 eni_ids: list[str],
4971 *,
4972 timeout_seconds: float = 120.0,
4973 poll_interval_seconds: float = 2.0,
4974 ) -> set[str]:
4975 """Wait for terminated bastion ENIs, deleting detached leftovers.
4977 EC2 normally deletes a primary ENI with its instance. If it becomes
4978 detached instead, it is safe to delete here because its owning instance
4979 was selected by the project/VPC bastion filters and termination has
4980 already been requested.
4981 """
4982 import time as _time
4984 remaining = set(eni_ids)
4985 deadline = _time.monotonic() + timeout_seconds
4986 while remaining: 4986 ↛ 5021line 4986 didn't jump to line 5021 because the condition on line 4986 was always true
4987 for eni_id in tuple(remaining):
4988 try:
4989 response = ec2.describe_network_interfaces(NetworkInterfaceIds=[eni_id])
4990 except ClientError as exc:
4991 code = exc.response.get("Error", {}).get("Code")
4992 if code == "InvalidNetworkInterfaceID.NotFound": 4992 ↛ 4995line 4992 didn't jump to line 4995 because the condition on line 4992 was always true
4993 remaining.discard(eni_id)
4994 continue
4995 logger.warning("Could not inspect bastion ENI %s: %s", eni_id, exc)
4996 return remaining
4997 except Exception as exc: # noqa: BLE001 - cleanup is best-effort
4998 logger.warning("Could not inspect bastion ENI %s: %s", eni_id, exc)
4999 return remaining
5001 interfaces = response.get("NetworkInterfaces", [])
5002 if not interfaces: 5002 ↛ 5003line 5002 didn't jump to line 5003 because the condition on line 5002 was never true
5003 remaining.discard(eni_id)
5004 continue
5005 if interfaces[0].get("Status") == "available":
5006 try:
5007 ec2.delete_network_interface(NetworkInterfaceId=eni_id)
5008 remaining.discard(eni_id)
5009 except ClientError as exc:
5010 code = exc.response.get("Error", {}).get("Code")
5011 if code == "InvalidNetworkInterfaceID.NotFound":
5012 remaining.discard(eni_id)
5013 else:
5014 logger.debug("Delete of bastion ENI %s failed: %s", eni_id, exc)
5015 except Exception as exc: # noqa: BLE001 - retry until timeout
5016 logger.debug("Delete of bastion ENI %s failed: %s", eni_id, exc)
5018 if not remaining or _time.monotonic() >= deadline: 5018 ↛ 5020line 5018 didn't jump to line 5020 because the condition on line 5018 was always true
5019 break
5020 _time.sleep(poll_interval_seconds)
5021 return remaining
5023 # ------------------------------------------------------------------
5024 # Implicit log-group + bastion IAM cleanup (non-strict destroy only)
5025 # ------------------------------------------------------------------
5026 #
5027 # CloudFormation only deletes the log groups it modeled. Lambda default
5028 # groups (``/aws/lambda/<function>``), the EKS control-plane group
5029 # (``/aws/eks/<cluster>/cluster``), and the Container Insights groups
5030 # (``/aws/containerinsights/<cluster>/…``) are created out-of-band by
5031 # the services themselves, so ``destroy-all`` used to report success
5032 # while leaving them behind — a real teardown orphaned 22 of them plus
5033 # the ephemeral-bastion IAM role/profile, which then failed the live
5034 # release validation's clean-account baseline gate.
5035 #
5036 # The cleanup below deletes ONLY exact names derived from the project's
5037 # own stack resources, captured while the stacks still exist, and only
5038 # for stacks whose deletion actually succeeded. It never runs in strict
5039 # (live-validation) teardowns: the harness checkpoints, tags, and
5040 # fences its own log-group generations and must remain the single
5041 # owner of that deletion authority.
5043 # The service-side patterns implicit log groups follow. An explicit
5044 # ``AWS::Logs::LogGroup`` resource is deliberately absent here —
5045 # CloudFormation owns those directly.
5046 _EKS_CONTAINER_INSIGHTS_SUFFIXES = ("application", "dataplane", "host", "performance")
5048 @staticmethod
5049 def _implicit_log_group_names(resource_type: str, physical_id: str) -> tuple[str, ...]:
5050 """Exact implicit log-group names a stack resource creates out-of-band."""
5051 if resource_type == "AWS::Lambda::Function":
5052 return (f"/aws/lambda/{physical_id}",)
5053 if resource_type == "AWS::EKS::Cluster":
5054 return (
5055 f"/aws/eks/{physical_id}/cluster",
5056 *(
5057 f"/aws/containerinsights/{physical_id}/{suffix}"
5058 for suffix in StackManager._EKS_CONTAINER_INSIGHTS_SUFFIXES
5059 ),
5060 )
5061 return ()
5063 def _collect_implicit_log_groups(self, stacks: Collection[str]) -> dict[str, dict[str, Any]]:
5064 """Derive per-stack implicit log-group names while the stacks are live.
5066 Best-effort: a stack that cannot be described or listed is skipped
5067 with a warning — collection must never block the destroy itself.
5068 """
5069 collected: dict[str, dict[str, Any]] = {}
5070 for stack_name in stacks:
5071 try:
5072 target = self._describe_stack_target(stack_name)
5073 if target is None:
5074 continue
5075 region, cloudformation, stack = target
5076 names: list[str] = []
5077 paginator = cloudformation.get_paginator("list_stack_resources")
5078 for page in paginator.paginate(StackName=str(stack["StackId"])):
5079 for item in page.get("StackResourceSummaries", []):
5080 resource_type = str(item.get("ResourceType") or "")
5081 physical_id = str(item.get("PhysicalResourceId") or "")
5082 if not physical_id:
5083 continue
5084 names.extend(self._implicit_log_group_names(resource_type, physical_id))
5085 if names: 5085 ↛ 5070line 5085 didn't jump to line 5070 because the condition on line 5085 was always true
5086 collected[stack_name] = {"region": region, "log_groups": sorted(set(names))}
5087 except Exception as exc: # noqa: BLE001 - best-effort collection
5088 logger.warning("Could not derive implicit log groups for %s: %s", stack_name, exc)
5089 return collected
5091 def _cleanup_implicit_log_groups(
5092 self,
5093 collected: Mapping[str, Mapping[str, Any]],
5094 successful_stacks: Collection[str],
5095 ) -> dict[str, Any]:
5096 """Delete the exact derived log groups of successfully destroyed stacks.
5098 A missing group is normal (a Lambda that never logged, or a custom
5099 ``LoggingConfig`` pointing elsewhere) and is recorded, not retried.
5100 Every error is recorded and swallowed: cleanup never converts a
5101 successful destroy into a failure.
5102 """
5103 import boto3
5105 outcome: dict[str, Any] = {"deleted": [], "missing": [], "errors": []}
5106 clients: dict[str, Any] = {}
5107 for stack_name in sorted(successful_stacks):
5108 details = collected.get(stack_name)
5109 if not details: 5109 ↛ 5110line 5109 didn't jump to line 5110 because the condition on line 5109 was never true
5110 continue
5111 region = str(details.get("region") or "")
5112 for name in details.get("log_groups", []):
5113 try:
5114 client = clients.get(region)
5115 if client is None:
5116 client = boto3.client("logs", region_name=region)
5117 clients[region] = client
5118 client.delete_log_group(logGroupName=name)
5119 outcome["deleted"].append(f"{region}:{name}")
5120 except ClientError as exc:
5121 code = str(exc.response.get("Error", {}).get("Code") or "")
5122 if code == "ResourceNotFoundException":
5123 outcome["missing"].append(f"{region}:{name}")
5124 else:
5125 outcome["errors"].append(f"{region}:{name}: {code}")
5126 except Exception as exc: # noqa: BLE001 - best-effort cleanup
5127 outcome["errors"].append(f"{region}:{name}: {type(exc).__name__}: {exc}")
5128 if outcome["deleted"]:
5129 print(
5130 f" Deleted {len(outcome['deleted'])} implicit CloudWatch log group(s) "
5131 "left behind by Lambda/EKS/Container Insights."
5132 )
5133 for failure in outcome["errors"]:
5134 logger.warning("Implicit log-group cleanup failed for %s", failure)
5135 return outcome
5137 def _cleanup_bastion_iam(self) -> dict[str, Any]:
5138 """Best-effort teardown of the ephemeral-bastion IAM role + profile.
5140 ``destroy_ephemeral_bastion`` already attempts this when a tunnel
5141 closes normally, but a killed process leaves the pair behind (they
5142 cost nothing, yet fail any clean-account audit). Deletion is by the
5143 exact project-scoped names from the bastion naming contract; a
5144 ``NoSuchEntity`` response simply means there was nothing to clean.
5145 """
5146 from .ephemeral_bastion import (
5147 _run_aws,
5148 bastion_profile_name,
5149 bastion_role_name,
5150 build_iam_teardown_commands,
5151 )
5153 outcome: dict[str, Any] = {
5154 "completed_steps": 0,
5155 "absent_steps": 0,
5156 "errors": [],
5157 }
5158 try:
5159 role_name = bastion_role_name(self.config.project_name)
5160 profile_name = bastion_profile_name(self.config.project_name)
5161 outcome["role"] = role_name
5162 outcome["profile"] = profile_name
5163 steps = build_iam_teardown_commands(
5164 role_name,
5165 profile_name,
5166 self.config.global_region,
5167 )
5168 except Exception as exc: # noqa: BLE001 - best-effort cleanup
5169 outcome["errors"].append(f"{type(exc).__name__}: {exc}")
5170 return outcome
5171 for step in steps:
5172 try:
5173 _run_aws(step)
5174 outcome["completed_steps"] += 1
5175 except RuntimeError as exc:
5176 if "NoSuchEntity" in str(exc):
5177 outcome["absent_steps"] += 1
5178 continue
5179 outcome["errors"].append(f"{' '.join(step[:3])}: {exc}")
5180 if outcome["completed_steps"] and not outcome["errors"]:
5181 print(" Removed the ephemeral-bastion IAM role and instance profile.")
5182 for failure in outcome["errors"]:
5183 logger.warning("Bastion IAM teardown step failed: %s", failure)
5184 return outcome
5186 def cleanup_eks_security_groups(self) -> None:
5187 """Clean up EKS-managed security groups across all regional stacks.
5189 Called between destroy retries to remove orphaned security groups
5190 that block VPC deletion.
5191 """
5192 stacks = self.list_stacks()
5193 # Regional stacks are everything that isn't a named global stack;
5194 # classify by suffix so this works for any project_name (#139).
5195 regional_stacks = [
5196 s for s in stacks if not s.endswith(("-global", "-api-gateway", "-monitoring"))
5197 ]
5198 for stack_name in regional_stacks:
5199 self._cleanup_eks_security_groups(stack_name)
5201 def cleanup_orphaned_network_interfaces(self) -> None:
5202 """Report and clear resources that can block VPC deletion, across all
5203 regional stacks. Run between destroy retries.
5205 Generalizes ``cleanup_eks_security_groups`` (which force-deletes the
5206 ``eks-cluster-sg-*`` security group + its ENIs that EKS leaves behind)
5207 with a broader sweep: for each regional stack's VPC it enumerates every
5208 remaining network interface, categorizes them (Global Accelerator / ELB
5209 / EKS / other), deletes the ones that are safe to remove (detached and
5210 not service-managed), and prints a friendly summary of what it found and
5211 what the next retry is waiting on. Service-managed ENIs (Global
5212 Accelerator, ELB) are released asynchronously by AWS once the endpoint /
5213 load balancer is gone, so we report them rather than fight them.
5214 """
5215 stacks = self.list_stacks()
5216 regional_stacks = [
5217 s for s in stacks if not s.endswith(("-global", "-api-gateway", "-monitoring"))
5218 ]
5219 for stack_name in regional_stacks:
5220 # Existing behaviour first: clear the EKS cluster SG + its ENIs.
5221 self._cleanup_eks_security_groups(stack_name)
5222 # Then report (and safely clear) anything else lingering in the VPC.
5223 summary = self._summarize_orphaned_enis(stack_name)
5224 self._print_orphaned_eni_summary(stack_name, summary)
5226 @staticmethod
5227 def _classify_orphaned_eni(eni: dict[str, Any]) -> str:
5228 """Bucket a network interface by which AWS service owns it.
5230 Uses ``InterfaceType`` first (authoritative for Global Accelerator and
5231 the load-balancer types) and falls back to the human ``Description``
5232 string for the EKS / ELB cases that present as a plain ``interface``.
5233 Returns one of ``global_accelerator`` / ``elb`` / ``eks`` / ``other``.
5234 """
5235 itype = str(eni.get("InterfaceType") or "").lower()
5236 desc = str(eni.get("Description") or "").lower()
5237 if (
5238 itype == "global_accelerator_managed"
5239 or "global_accelerator" in desc
5240 or "global accelerator" in desc
5241 ):
5242 return "global_accelerator"
5243 if itype in ("load_balancer", "network_load_balancer") or desc.startswith("elb "):
5244 return "elb"
5245 if "eks" in desc or "k8s" in desc or "kubernetes" in desc:
5246 return "eks"
5247 return "other"
5249 def _summarize_orphaned_enis(self, stack_name: str) -> dict[str, int]:
5250 """Inspect the stack's VPC(s) for lingering ENIs, categorize them, and
5251 best-effort delete the ones that are safe to remove.
5253 "Safe to remove" means ``Status == "available"`` (detached) and not
5254 ``RequesterManaged`` (i.e. not owned by a service like GA / ELB, which
5255 rejects manual deletion and releases the ENI on its own schedule).
5257 Returns a dict of counts: per-category totals plus ``deleted`` and
5258 ``vpcs``. Wholly best-effort — any AWS error degrades to the counts
5259 gathered so far rather than raising into the destroy flow.
5260 """
5261 import boto3
5263 region = stack_name.replace(f"{self.config.project_name}-", "", 1)
5264 summary: dict[str, int] = {
5265 "global_accelerator": 0,
5266 "elb": 0,
5267 "eks": 0,
5268 "other": 0,
5269 "deleted": 0,
5270 "vpcs": 0,
5271 }
5272 try:
5273 ec2 = boto3.client("ec2", region_name=region)
5274 vpcs = ec2.describe_vpcs(
5275 Filters=[{"Name": "tag:aws:cloudformation:stack-name", "Values": [stack_name]}]
5276 ).get("Vpcs", [])
5277 except Exception as e: # noqa: BLE001
5278 logger.debug("ENI sweep: VPC lookup failed for %s: %s", stack_name, e)
5279 return summary
5281 for vpc in vpcs:
5282 summary["vpcs"] += 1
5283 vpc_id = vpc.get("VpcId")
5284 try:
5285 enis = ec2.describe_network_interfaces(
5286 Filters=[{"Name": "vpc-id", "Values": [vpc_id]}]
5287 ).get("NetworkInterfaces", [])
5288 except Exception as e: # noqa: BLE001
5289 logger.debug("ENI sweep: describe ENIs failed for %s: %s", vpc_id, e)
5290 continue
5292 for eni in enis:
5293 summary[self._classify_orphaned_eni(eni)] += 1
5294 detached = eni.get("Status") == "available"
5295 service_managed = bool(eni.get("RequesterManaged", False))
5296 if detached and not service_managed:
5297 eni_id = eni.get("NetworkInterfaceId")
5298 try:
5299 ec2.delete_network_interface(NetworkInterfaceId=eni_id)
5300 summary["deleted"] += 1
5301 logger.debug("ENI sweep: deleted detached ENI %s in %s", eni_id, vpc_id)
5302 except Exception as e: # noqa: BLE001
5303 logger.debug("ENI sweep: delete of %s failed: %s", eni_id, e)
5304 return summary
5306 @staticmethod
5307 def _print_orphaned_eni_summary(stack_name: str, summary: dict[str, int]) -> None:
5308 """Print a friendly summary of what the ENI sweep found and handled."""
5309 categories = (
5310 ("global_accelerator", "Global Accelerator-managed"),
5311 ("elb", "ELB-managed"),
5312 ("eks", "EKS-managed"),
5313 ("other", "other"),
5314 )
5315 total = sum(summary.get(key, 0) for key, _ in categories)
5316 if total == 0:
5317 return
5318 breakdown = ", ".join(
5319 f"{summary[key]} {label}" for key, label in categories if summary.get(key)
5320 )
5321 print(f" {stack_name}: {total} network interface(s) still in the VPC ({breakdown}).")
5322 if summary.get("deleted"): 5322 ↛ 5324line 5322 didn't jump to line 5324 because the condition on line 5322 was always true
5323 print(f" Removed {summary['deleted']} detached interface(s).")
5324 remaining = total - summary.get("deleted", 0)
5325 if remaining > 0:
5326 print(
5327 f" {remaining} still held by AWS — Global Accelerator / ELB release these "
5328 "asynchronously once the endpoint and load balancer are gone; the next retry "
5329 "proceeds once they drain."
5330 )
5332 def _cleanup_eks_security_groups(
5333 self,
5334 stack_name: str,
5335 *,
5336 region: str | None = None,
5337 security_group_id: str | None = None,
5338 vpc_id: str | None = None,
5339 ) -> dict[str, Any]:
5340 """Delete empty EKS SGs, optionally by one exact preauthorized ID."""
5341 import boto3
5343 project_name = self.config.project_name
5344 region = region or stack_name.replace(f"{project_name}-", "", 1)
5345 cluster_name = stack_name
5346 outcome: dict[str, Any] = {
5347 "stack": stack_name,
5348 "region": region,
5349 "security_group_id": security_group_id,
5350 "inspected": 0,
5351 "deleted": [],
5352 "blocked_by_enis": [],
5353 "errors": [],
5354 }
5356 try:
5357 ec2 = boto3.client("ec2", region_name=region)
5358 try:
5359 if security_group_id: 5359 ↛ 5360line 5359 didn't jump to line 5360 because the condition on line 5359 was never true
5360 response = ec2.describe_security_groups(GroupIds=[security_group_id])
5361 else:
5362 response = ec2.describe_security_groups(
5363 Filters=[
5364 {
5365 "Name": "group-name",
5366 "Values": [f"eks-cluster-sg-{cluster_name}-*"],
5367 }
5368 ]
5369 )
5370 except ClientError as exc:
5371 if (
5372 security_group_id
5373 and exc.response.get("Error", {}).get("Code") == "InvalidGroup.NotFound"
5374 ):
5375 outcome["absent"] = True
5376 return outcome
5377 raise
5379 for security_group in response.get("SecurityGroups", []):
5380 outcome["inspected"] += 1
5381 group_id = str(security_group["GroupId"])
5382 group_name = str(security_group.get("GroupName", ""))
5383 if security_group_id and group_id != security_group_id: 5383 ↛ 5384line 5383 didn't jump to line 5384 because the condition on line 5383 was never true
5384 raise RuntimeError(
5385 f"EC2 returned changed security-group identity for {security_group_id}"
5386 )
5387 if vpc_id and str(security_group.get("VpcId") or "") != vpc_id: 5387 ↛ 5388line 5387 didn't jump to line 5388 because the condition on line 5387 was never true
5388 raise RuntimeError(
5389 f"Security group {group_id} no longer belongs to exact VPC {vpc_id}"
5390 )
5391 interfaces = ec2.describe_network_interfaces(
5392 Filters=[{"Name": "group-id", "Values": [group_id]}]
5393 ).get("NetworkInterfaces", [])
5394 if interfaces:
5395 outcome["blocked_by_enis"].append(
5396 {
5397 "group_id": group_id,
5398 "group_name": group_name,
5399 "network_interface_ids": sorted(
5400 str(interface.get("NetworkInterfaceId") or "")
5401 for interface in interfaces
5402 if interface.get("NetworkInterfaceId")
5403 ),
5404 }
5405 )
5406 logger.debug(
5407 "Waiting for AWS to release %d EKS-managed ENI(s) from %s",
5408 len(interfaces),
5409 group_name,
5410 )
5411 continue
5412 try:
5413 ec2.delete_security_group(GroupId=group_id)
5414 outcome["deleted"].append({"group_id": group_id, "group_name": group_name})
5415 print(f" Cleaned up empty EKS security group: {group_name} ({group_id})")
5416 except ClientError as exc:
5417 if exc.response.get("Error", {}).get("Code") == "InvalidGroup.NotFound":
5418 outcome["absent"] = True
5419 continue
5420 outcome["errors"].append(
5421 {"group_id": group_id, "error": f"{type(exc).__name__}: {exc}"}
5422 )
5423 except Exception as exc:
5424 outcome["errors"].append(
5425 {"group_id": group_id, "error": f"{type(exc).__name__}: {exc}"}
5426 )
5427 except Exception as exc:
5428 outcome["errors"].append({"error": f"{type(exc).__name__}: {exc}"})
5429 logger.debug("EKS security group cleanup for %s failed: %s", stack_name, exc)
5430 return outcome
5432 def _start_eks_sg_watchdog(
5433 self,
5434 stack_name: str,
5435 stop_event: Event,
5436 *,
5437 region: str | None = None,
5438 security_group_id: str | None = None,
5439 vpc_id: str | None = None,
5440 ) -> Thread:
5441 """Start a background thread that polls for orphaned EKS security groups.
5443 EKS creates an ``eks-cluster-sg-<cluster-name>-*`` security group that
5444 is owned by the EKS service (not CloudFormation). The watchdog observes
5445 it throughout regional teardown and removes it only after AWS has
5446 released every attached ENI. Service-managed interfaces are never
5447 detached or deleted by the CLI.
5449 The thread exits when ``stop_event`` is set by the orchestrator at
5450 the end of the regional phase.
5451 """
5453 def _watchdog() -> None:
5454 while not stop_event.is_set():
5455 try:
5456 self._cleanup_eks_security_groups(
5457 stack_name,
5458 region=region,
5459 security_group_id=security_group_id,
5460 vpc_id=vpc_id,
5461 )
5462 except Exception as e:
5463 logger.debug(
5464 "EKS SG watchdog tick for %s failed (non-fatal): %s",
5465 stack_name,
5466 e,
5467 )
5468 # ``wait`` returns immediately when the event is set, so this
5469 # doubles as the sleep-and-shutdown-check in one call.
5470 stop_event.wait(timeout=30)
5472 thread = Thread(
5473 target=_watchdog,
5474 name=f"eks-sg-watchdog-{stack_name}",
5475 daemon=True,
5476 )
5477 thread.start()
5478 return thread
5481def get_stack_manager(config: GCOConfig) -> StackManager:
5482 """Factory function to get a StackManager instance."""
5483 return StackManager(config)
5486def _is_regional_api_bridge_stack(
5487 stack: str,
5488 *,
5489 project_name: str,
5490 stack_names: Collection[str],
5491) -> bool:
5492 """Return whether ``stack`` is a configured per-Region API bridge.
5494 A bare ``"-regional-api-"`` substring is ambiguous because it is valid
5495 inside ``project_name``. Match the exact project-scoped bridge prefix and
5496 require the corresponding configured ``<project>-<region>`` base stack.
5497 """
5498 bridge_prefix = f"{project_name}-regional-api-"
5499 if not stack.startswith(bridge_prefix):
5500 return False
5501 region = stack.removeprefix(bridge_prefix)
5502 return bool(region) and f"{project_name}-{region}" in stack_names
5505def _get_stack_destroy_phases(
5506 stacks: list[str],
5507 *,
5508 project_name: str,
5509) -> tuple[list[str], list[str], list[str], list[str]]:
5510 """Classify and order the exact phases used by orchestrated destroy.
5512 Returns monitoring, regional API bridge, base regional, and pre-regional
5513 global phases. The public preview helper and the execution path both flatten
5514 this result, so custom project names and bridge dependencies cannot drift.
5515 """
5516 stack_names = set(stacks)
5517 monitoring_stacks = sorted(
5518 (stack for stack in stacks if stack.endswith("-monitoring")),
5519 reverse=True,
5520 )
5521 regional_api_stacks = sorted(
5522 (
5523 stack
5524 for stack in stacks
5525 if _is_regional_api_bridge_stack(
5526 stack,
5527 project_name=project_name,
5528 stack_names=stack_names,
5529 )
5530 ),
5531 reverse=True,
5532 )
5533 regional_stacks = sorted(
5534 (
5535 stack
5536 for stack in stacks
5537 if not stack.endswith(("-global", "-api-gateway", "-monitoring"))
5538 and not _is_regional_api_bridge_stack(
5539 stack,
5540 project_name=project_name,
5541 stack_names=stack_names,
5542 )
5543 ),
5544 reverse=True,
5545 )
5546 pre_regional_stacks = sorted(
5547 (stack for stack in stacks if stack.endswith(("-global", "-api-gateway"))),
5548 key=lambda stack: (
5549 1 if stack.endswith("-api-gateway") else (2 if stack.endswith("-global") else 0)
5550 ),
5551 )
5552 return (
5553 monitoring_stacks,
5554 regional_api_stacks,
5555 regional_stacks,
5556 pre_regional_stacks,
5557 )
5560def get_stack_deployment_order(
5561 stacks: list[str],
5562 *,
5563 project_name: str = "gco",
5564) -> list[str]:
5565 """
5566 Get the correct deployment order for stacks.
5568 Order: global stacks first, then regional stacks.
5569 Global stacks: <project>-global, <project>-api-gateway,
5570 <project>-analytics, <project>-monitoring
5571 Regional stacks: <project>-{region} (e.g., gco-us-east-1)
5573 Named stacks are classified by suffix so ordering is independent of
5574 ``project_name`` (#139): a non-``gco`` deployment (``acme-global`` …)
5575 orders identically. Regional stacks are ``<project>-<region>`` and match
5576 no named suffix, so they fall through to the regional bucket.
5577 """
5578 stack_names = set(stacks)
5579 global_stacks = []
5580 regional_stacks = []
5581 regional_api_stacks = []
5583 # Named (non-regional) stack priority by suffix (lower = deploy first).
5584 suffix_priority = {
5585 "-global": 1,
5586 "-api-gateway": 2,
5587 "-analytics": 2.5,
5588 "-monitoring": 3,
5589 }
5591 def _named_priority(stack: str) -> float | None:
5592 for suffix, prio in suffix_priority.items():
5593 if stack.endswith(suffix):
5594 return prio
5595 return None
5597 for stack in stacks:
5598 priority = _named_priority(stack)
5599 if priority is not None:
5600 global_stacks.append((priority, stack))
5601 elif _is_regional_api_bridge_stack(
5602 stack,
5603 project_name=project_name,
5604 stack_names=stack_names,
5605 ):
5606 regional_api_stacks.append(stack)
5607 else:
5608 regional_stacks.append(stack)
5610 # Keep bridge dependencies after every base regional stack. The
5611 # orchestrated lifecycle further separates monitoring into its own phase.
5612 global_stacks.sort(key=lambda x: x[0])
5613 regional_stacks.sort()
5614 regional_api_stacks.sort()
5616 return [s[1] for s in global_stacks] + regional_stacks + regional_api_stacks
5619def get_stack_destroy_order(
5620 stacks: list[str],
5621 *,
5622 project_name: str = "gco",
5623) -> list[str]:
5624 """Return the exact project-aware order used by orchestrated destroy."""
5625 phases = _get_stack_destroy_phases(stacks, project_name=project_name)
5626 return [stack for phase in phases for stack in phase]
5629# =============================================================================
5630# Feature toggle helpers
5631# =============================================================================
5633_FSX_DEFAULTS: dict[str, Any] = {
5634 "enabled": False,
5635 "storage_capacity_gib": 1200,
5636 "deployment_type": "SCRATCH_2",
5637 "per_unit_storage_throughput": 200,
5638 "data_compression_type": "LZ4",
5639 "import_path": None,
5640 "export_path": None,
5641 "auto_import_policy": "NEW_CHANGED_DELETED",
5642}
5645def _find_cdk_json() -> Path | None:
5646 """Find cdk.json in current or parent directories."""
5647 current = Path.cwd()
5648 for parent in [current] + list(current.parents):
5649 cdk_path = parent / "cdk.json"
5650 if cdk_path.exists():
5651 return cdk_path
5652 return None
5655def get_fsx_config(region: str | None = None) -> dict[str, Any]:
5656 """Get current FSx for Lustre configuration from cdk.json.
5658 Args:
5659 region: Optional region to get config for. If provided, checks for
5660 region-specific overrides first.
5662 Returns:
5663 FSx configuration dictionary
5664 """
5665 return _get_feature_config("fsx_lustre", _FSX_DEFAULTS, region)
5668def update_fsx_config(settings: dict[str, Any], region: str | None = None) -> None:
5669 """Update FSx for Lustre configuration in cdk.json.
5671 Args:
5672 settings: FSx settings to update
5673 region: Optional region for region-specific config. If None, updates global config.
5674 """
5675 _update_feature_config("fsx_lustre", settings, _FSX_DEFAULTS, region)
5678# =============================================================================
5679# Generic feature toggle helpers (used by FSx, Valkey, Aurora, and future features)
5680# =============================================================================
5683def _get_feature_config(
5684 feature_key: str,
5685 default_config: dict[str, Any],
5686 region: str | None = None,
5687) -> dict[str, Any]:
5688 """Get configuration for a toggleable feature from cdk.json.
5690 Args:
5691 feature_key: The cdk.json context key (e.g. "valkey", "aurora_pgvector").
5692 default_config: Default configuration values when the key is missing.
5693 region: Optional region for region-specific overrides.
5695 Returns:
5696 Merged configuration dictionary.
5697 """
5698 cdk_json_path = _find_cdk_json()
5699 if not cdk_json_path:
5700 raise RuntimeError("cdk.json not found")
5702 import json
5704 with open(cdk_json_path, encoding="utf-8") as f:
5705 cdk_config = json.load(f)
5707 global_config = cdk_config.get("context", {}).get(feature_key, default_config)
5709 if region:
5710 region_key = f"{feature_key}_regions"
5711 region_overrides = cdk_config.get("context", {}).get(region_key, {})
5712 if region in region_overrides:
5713 merged = {**global_config, **region_overrides[region]}
5714 merged["region"] = region
5715 merged["is_region_specific"] = True
5716 return merged
5718 result = {**default_config, **global_config}
5719 result["is_region_specific"] = False
5720 return result
5723def _update_feature_config(
5724 feature_key: str,
5725 settings: dict[str, Any],
5726 default_config: dict[str, Any],
5727 region: str | None = None,
5728) -> None:
5729 """Update configuration for a toggleable feature in cdk.json.
5731 Args:
5732 feature_key: The cdk.json context key (e.g. "valkey", "aurora_pgvector").
5733 settings: Settings to update.
5734 default_config: Default configuration values when the key is missing.
5735 region: Optional region for region-specific config.
5736 """
5737 cdk_json_path = _find_cdk_json()
5738 if not cdk_json_path:
5739 raise RuntimeError("cdk.json not found")
5741 import json
5743 with open(cdk_json_path, encoding="utf-8") as f:
5744 cdk_config = json.load(f)
5746 if "context" not in cdk_config:
5747 cdk_config["context"] = {}
5749 if region:
5750 region_key = f"{feature_key}_regions"
5751 if region_key not in cdk_config["context"]: 5751 ↛ 5753line 5751 didn't jump to line 5753 because the condition on line 5751 was always true
5752 cdk_config["context"][region_key] = {}
5753 if region not in cdk_config["context"][region_key]: 5753 ↛ 5755line 5753 didn't jump to line 5755 because the condition on line 5753 was always true
5754 cdk_config["context"][region_key][region] = {}
5755 for key, value in settings.items():
5756 if value is not None or key == "enabled": 5756 ↛ 5755line 5756 didn't jump to line 5755 because the condition on line 5756 was always true
5757 cdk_config["context"][region_key][region][key] = value
5758 else:
5759 if feature_key not in cdk_config["context"]:
5760 cdk_config["context"][feature_key] = {**default_config}
5761 for key, value in settings.items():
5762 if value is not None or key == "enabled":
5763 cdk_config["context"][feature_key][key] = value
5765 serialized = json.dumps(cdk_config, indent=2).encode("utf-8")
5766 _atomic_write_bytes(
5767 cdk_json_path,
5768 serialized,
5769 mode=stat.S_IMODE(cdk_json_path.stat().st_mode),
5770 )
5773# =============================================================================
5774# Valkey configuration
5775# =============================================================================
5777_VALKEY_DEFAULTS: dict[str, Any] = {
5778 "enabled": False,
5779 "max_data_storage_gb": 5,
5780 "max_ecpu_per_second": 5000,
5781 "snapshot_retention_limit": 1,
5782}
5785def get_valkey_config(region: str | None = None) -> dict[str, Any]:
5786 """Get current Valkey Serverless configuration from cdk.json."""
5787 return _get_feature_config("valkey", _VALKEY_DEFAULTS, region)
5790def update_valkey_config(settings: dict[str, Any], region: str | None = None) -> None:
5791 """Update Valkey Serverless configuration in cdk.json."""
5792 _update_feature_config("valkey", settings, _VALKEY_DEFAULTS, region)
5795# =============================================================================
5796# Aurora pgvector configuration
5797# =============================================================================
5799_AURORA_DEFAULTS: dict[str, Any] = {
5800 "enabled": False,
5801 "min_acu": 0,
5802 "max_acu": 16,
5803 "backup_retention_days": 7,
5804 "deletion_protection": False,
5805}
5808def get_aurora_config(region: str | None = None) -> dict[str, Any]:
5809 """Get current Aurora pgvector configuration from cdk.json."""
5810 return _get_feature_config("aurora_pgvector", _AURORA_DEFAULTS, region)
5813def update_aurora_config(settings: dict[str, Any], region: str | None = None) -> None:
5814 """Update Aurora pgvector configuration in cdk.json."""
5815 _update_feature_config("aurora_pgvector", settings, _AURORA_DEFAULTS, region)
5818# =============================================================================
5819# Analytics environment configuration
5820# =============================================================================
5822_ANALYTICS_DEFAULTS: dict[str, Any] = {
5823 "enabled": False,
5824 "hyperpod": {"enabled": False},
5825 "canvas": {"enabled": False},
5826 "cognito": {"domain_prefix": None, "removal_policy": "destroy"},
5827 "efs": {"removal_policy": "destroy"},
5828 "studio": {"user_profile_name_prefix": None},
5829}
5832def get_analytics_config() -> dict[str, Any]:
5833 """Get the analytics environment configuration from cdk.json.
5835 The analytics stack is single-region by construction (lives in the
5836 api-gateway region), so this helper does not accept a region argument.
5837 Returned dict is the defaults merged with any operator overrides from
5838 the ``context.analytics_environment`` block.
5839 """
5840 return _get_feature_config("analytics_environment", _ANALYTICS_DEFAULTS)
5843def update_analytics_config(settings: dict[str, Any]) -> None:
5844 """Update the analytics environment configuration in cdk.json.
5846 Mirrors ``update_valkey_config`` / ``update_aurora_config``. Nested
5847 keys under ``analytics_environment`` (``hyperpod``, ``canvas``,
5848 ``cognito``, ``efs``, ``studio``) are merged one level deep rather
5849 than replaced wholesale — ``enable --hyperpod`` must not clobber
5850 ``cognito.removal_policy``.
5851 """
5852 _update_feature_config("analytics_environment", settings, _ANALYTICS_DEFAULTS)
5855# =============================================================================
5856# Cluster observability configuration
5857# =============================================================================
5859# Mirrors the on-by-default cdk.json cluster_observability block. Unlike the
5860# other feature toggles this one defaults to enabled=True: a stock deploy
5861# installs kube-prometheus-stack on every regional cluster and operators opt
5862# out. The CDK side reads/validates the same block via
5863# ConfigLoader.get_cluster_observability_config.
5864_CLUSTER_OBSERVABILITY_DEFAULTS: dict[str, Any] = {
5865 "enabled": True,
5866 "grafana": {
5867 "persistence_size": "10Gi",
5868 "admin_user": "admin",
5869 "admin_password_rotation_schedule": "0 4 1 * *",
5870 },
5871 "prometheus": {"persistence_size": "50Gi", "retention": "15d"},
5872 "alertmanager": {"enabled": True, "persistence_size": "5Gi"},
5873}
5876def get_cluster_observability_config() -> dict[str, Any]:
5877 """Get the cluster observability configuration from cdk.json.
5879 Observability is per-region (installed on every regional cluster) but the
5880 toggle itself is global, so this takes no region argument. Returns the
5881 defaults merged with any operator overrides from the
5882 ``context.cluster_observability`` block.
5883 """
5884 return _get_feature_config("cluster_observability", _CLUSTER_OBSERVABILITY_DEFAULTS)
5887def update_cluster_observability_config(settings: dict[str, Any]) -> None:
5888 """Update the cluster observability toggle in cdk.json.
5890 ``gco monitoring enable`` / ``disable`` pass ``{"enabled": True/False}``;
5891 the grafana/prometheus/alertmanager sub-blocks are left untouched so an
5892 operator's sizing/retention/rotation overrides survive a disable/enable
5893 cycle.
5894 """
5895 _update_feature_config("cluster_observability", settings, _CLUSTER_OBSERVABILITY_DEFAULTS)