Coverage for cli / stacks.py: 100.00%
2859 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +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, nullcontext
56from dataclasses import dataclass, field
57from datetime import UTC, datetime
58from functools import lru_cache
59from pathlib import Path
60from threading import Event, Lock, RLock, Thread, local
61from typing import TYPE_CHECKING, Any, BinaryIO, Literal, TypedDict
63from botocore.exceptions import ClientError
64from click import Abort
66from gco.lambda_shared_sources import LAMBDA_SHARED_SOURCE_TARGETS
67from gco.stacks.constants import (
68 known_cloudformation_regions,
69 validated_deployment_partition,
70 validated_regional_deployment_regions,
71)
73from .output import confirm, interactive_echo
75# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
76# Generated at (UTC): 2026-09-12T06:04:03Z
77# Generated from Git commit: e96e2c39c3626a5088651f43873dfade6a346850
78# Flowchart(s) generated from this file:
79# * ``StackManager.deploy_orchestrated`` -> ``diagrams/code_diagrams/cli/stacks.StackManager_deploy_orchestrated.html``
80# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager_deploy_orchestrated.png``)
81# * ``StackManager.destroy_orchestrated`` -> ``diagrams/code_diagrams/cli/stacks.StackManager_destroy_orchestrated.html``
82# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager_destroy_orchestrated.png``)
83# * ``StackManager._mirror_images_if_enabled`` -> ``diagrams/code_diagrams/cli/stacks.StackManager__mirror_images_if_enabled.html``
84# (PNG: ``diagrams/code_diagrams/cli/stacks.StackManager__mirror_images_if_enabled.png``)
85# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
86# <pyflowchart-code-diagram> END
89if TYPE_CHECKING:
90 from .config import GCOConfig
92logger = logging.getLogger(__name__)
94# Every writer that replaces cdk.json must participate in the same transaction
95# lock. The process-local RLock handles threads and nested feature updates; the
96# advisory process lock uses a stable directory descriptor on POSIX and a
97# persistent sidecar file on Windows, so it survives ``os.replace`` of the
98# configuration inode.
99_CONFIG_LOCK_FILENAME = ".gco-config.lock"
100_CONFIG_THREAD_LOCKS: dict[Path, Any] = {}
101_CONFIG_THREAD_LOCKS_GUARD = Lock()
102_CONFIG_LOCK_STATE = local()
104# Python packages ``app.py`` imports at CDK synth time. They ship in the
105# optional ``[cdk]`` extra (see pyproject.toml), NOT the base install, so a
106# lightweight ``uvx`` / ``pip install`` of ``gco-cli`` that skips the extra
107# cannot synthesize or deploy. ``StackManager._ensure_cdk_toolchain`` checks
108# for these before invoking ``cdk`` so a missing toolchain is actionable.
109_CDK_TOOLCHAIN_MODULES = ("aws_cdk", "cdk_nag")
110_INFERENCE_STREAMING_PACKAGE_FILES = ("index.mjs", "package.json", "package-lock.json")
111_KUBECTL_PACKAGE_INPUTS = ("handler.py", "requirements.txt", "manifests")
112_LAMBDA_BUILD_MANIFEST = ".gco-build-manifest.json"
113_LAMBDA_BUILD_MANIFEST_VERSION = 1
114_LAMBDA_SOURCE_IGNORED_DIRECTORIES = frozenset({"__pycache__", ".mypy_cache", ".pytest_cache"})
115_LAMBDA_SOURCE_IGNORED_FILES = frozenset({".DS_Store"})
116_LAMBDA_SOURCE_COPY_IGNORE_PATTERNS = (
117 "__pycache__",
118 ".mypy_cache",
119 ".pytest_cache",
120 ".DS_Store",
121 "*.pyc",
122 "*.pyo",
123)
124_FILE_LOCK_RETRY_SECONDS = 0.05
125# 15 minutes: comfortably above the longest legitimate hold (a cold publisher
126# rebuild, minutes) while bounding the pathological one (an abandoned pytest
127# session's session-long shared locks, indefinite).
128_ASSET_LOCK_TIMEOUT_SECONDS_DEFAULT = 900.0
129_CONFIG_LOCK_TIMEOUT_SECONDS_DEFAULT = 900.0
130_FileLockPurpose = Literal["asset", "configuration"]
131_CDK_ASSET_CONSUMER_MAX_ATTEMPTS = 3
132_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS = 7200.0
133_CLOUDFORMATION_DELETE_POLL_SECONDS = 15.0
134_CLOUDFORMATION_DELETE_HEARTBEAT_SECONDS = 60.0
135_CLOUDFORMATION_SETTLE_UNKNOWN_TIMEOUT_SECONDS = 60.0
136_CLOUDFORMATION_SETTLE_UNKNOWN_POLL_SECONDS = 5.0
137_BOOTSTRAP_HEALTHY_STATUSES = frozenset({"CREATE_COMPLETE", "UPDATE_COMPLETE"})
138# CDK's ``prepare-change-set`` mode can return before a fresh CREATE change set
139# is visible through CloudFormation's read path. Poll only that authoritative
140# fresh-create absence window; every access, identity, tag, and ownership error
141# still fails immediately. Sixteen reads at two-second intervals bound the
142# eventual-consistency allowance to 30 seconds after the first attempt.
143_STRICT_CHANGE_SET_INSPECTION_ATTEMPTS = 16
144_STRICT_CHANGE_SET_INSPECTION_RETRY_SECONDS = 2.0
145_LIVE_VALIDATION_PROVIDER_LOG_CONTEXT = "gco_live_validation_retain_provider_log_groups"
147# LAMBDA_SHARED_SOURCE_TARGETS is imported from the dependency-light inventory
148# shared by deploy packaging, diagram reconciliation, and commit-time guards.
149StackAuthorizationCallback = Callable[[str, str, str], None]
150CleanupOutcomeCallback = Callable[[str, dict[str, Any]], None]
151ChangeSetPreparedCallback = Callable[[str, str, str, str, str], None]
152PreparedChangeSetAuthority = Mapping[str, Mapping[str, Mapping[str, str]]]
153EcrRepositoryCreatedCallback = Callable[[str, Mapping[str, Any]], None]
156class _StackOperationSafetyKwargs(TypedDict):
157 """Type-preserving keyword bundle shared by strict deploy and destroy calls."""
159 allow_bootstrap: bool
160 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None
161 expected_stack_ids: Mapping[str, str | None] | None
162 prepared_change_sets: PreparedChangeSetAuthority | None
163 authorize_stack: StackAuthorizationCallback | None
164 strict_deployment_token: str | None
165 on_change_set_prepared: ChangeSetPreparedCallback | None
166 on_ecr_repository_created: EcrRepositoryCreatedCallback | None
169@dataclass(frozen=True)
170class _CdkAssetSpec:
171 """One canonical generated asset consumed by the CDK application."""
173 name: str
174 source_directory: str
175 build_directory: str
176 source_inputs: tuple[str, ...] | None
178 def paths(self, project_root: Path) -> tuple[Path, Path]:
179 lambda_dir = project_root / "lambda"
180 return lambda_dir / self.source_directory, lambda_dir / self.build_directory
183_KUBECTL_CDK_ASSET = _CdkAssetSpec(
184 name="kubectl-applier-simple",
185 source_directory="kubectl-applier-simple",
186 build_directory="kubectl-applier-simple-build",
187 source_inputs=_KUBECTL_PACKAGE_INPUTS,
188)
189_HELM_CDK_ASSET = _CdkAssetSpec(
190 name="helm-installer",
191 source_directory="helm-installer",
192 build_directory="helm-installer-build",
193 source_inputs=None,
194)
195_INFERENCE_STREAMING_CDK_ASSET = _CdkAssetSpec(
196 name="inference-streaming-proxy",
197 source_directory="inference-streaming-proxy",
198 build_directory="inference-streaming-proxy-build",
199 source_inputs=_INFERENCE_STREAMING_PACKAGE_FILES,
200)
201_CDK_ASSET_SPECS = (
202 _KUBECTL_CDK_ASSET,
203 _HELM_CDK_ASSET,
204 _INFERENCE_STREAMING_CDK_ASSET,
205)
208class _AssetThreadState(local):
209 """Per-thread nesting state; each OS lock still spans the full process."""
211 def __init__(self) -> None:
212 self.held: dict[str, tuple[bool, int]] = {}
213 self.active_consumers: dict[str, int] = {}
216_asset_thread_state = _AssetThreadState()
219def _asset_tree_paths(root: Path, source_inputs: tuple[str, ...] | None) -> Iterator[Path]:
220 """Yield deterministic source or build-tree entries below ``root``."""
221 selected: set[Path] = set()
222 if source_inputs is None:
223 selected.update(root.rglob("*"))
224 else:
225 for relative_name in source_inputs:
226 path = root / relative_name
227 if not path.exists() and not path.is_symlink():
228 raise FileNotFoundError(path)
229 selected.add(path)
230 if path.is_dir() and not path.is_symlink():
231 selected.update(path.rglob("*"))
232 yield from sorted(selected, key=lambda path: path.relative_to(root).as_posix())
235def _asset_tree_digest(
236 root: Path,
237 *,
238 source_inputs: tuple[str, ...] | None = None,
239) -> str | None:
240 """Hash every deployable entry in a source selection or complete build tree.
242 The completion manifest and local cache files are excluded. Regular-file
243 content, paths, modes, directory entries, and symlink targets are included
244 so removing any installed transitive dependency invalidates the build.
245 """
246 if not root.is_dir():
247 return None
248 digest = hashlib.sha256()
249 try:
250 for path in _asset_tree_paths(root, source_inputs):
251 relative = path.relative_to(root)
252 if any(part in _LAMBDA_SOURCE_IGNORED_DIRECTORIES for part in relative.parts):
253 continue
254 if (
255 path.name in _LAMBDA_SOURCE_IGNORED_FILES
256 or path.name == _LAMBDA_BUILD_MANIFEST
257 or path.suffix in {".pyc", ".pyo"}
258 ):
259 continue
261 metadata = path.lstat()
262 relative_bytes = relative.as_posix().encode("utf-8")
263 digest.update(len(relative_bytes).to_bytes(8, "big"))
264 digest.update(relative_bytes)
265 digest.update(stat.S_IMODE(metadata.st_mode).to_bytes(4, "big"))
267 if path.is_symlink():
268 target = os.readlink(path).encode("utf-8")
269 digest.update(b"L")
270 digest.update(len(target).to_bytes(8, "big"))
271 digest.update(target)
272 elif path.is_dir():
273 digest.update(b"D")
274 elif path.is_file():
275 file_digest = hashlib.sha256()
276 with path.open("rb") as handle:
277 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
278 file_digest.update(chunk)
279 digest.update(b"F")
280 digest.update(file_digest.digest())
281 else:
282 return None
283 except OSError, UnicodeError:
284 return None
285 return digest.hexdigest()
288def _read_build_manifest(build_dir: Path) -> dict[str, Any] | None:
289 try:
290 value = json.loads((build_dir / _LAMBDA_BUILD_MANIFEST).read_text(encoding="utf-8"))
291 except OSError, UnicodeError, json.JSONDecodeError:
292 return None
293 return value if isinstance(value, dict) else None
296def _write_build_manifest(build_dir: Path, source_digest: str) -> None:
297 """Write the completion marker only after the staged build is complete."""
298 build_digest = _asset_tree_digest(build_dir)
299 if build_digest is None:
300 raise RuntimeError(f"Unable to hash completed Lambda asset {build_dir.name}")
301 manifest = {
302 "schema_version": _LAMBDA_BUILD_MANIFEST_VERSION,
303 "source_digest": source_digest,
304 "build_digest": build_digest,
305 }
306 manifest_path = build_dir / _LAMBDA_BUILD_MANIFEST
307 with manifest_path.open("x", encoding="utf-8") as handle:
308 json.dump(manifest, handle, sort_keys=True, separators=(",", ":"))
309 handle.write("\n")
310 handle.flush()
311 os.fsync(handle.fileno())
314def _asset_build_is_fresh_unlocked(
315 source_dir: Path,
316 build_dir: Path,
317 *,
318 source_inputs: tuple[str, ...] | None,
319) -> bool:
320 manifest = _read_build_manifest(build_dir)
321 if manifest is None or manifest.get("schema_version") != _LAMBDA_BUILD_MANIFEST_VERSION:
322 return False
323 source_digest = _asset_tree_digest(source_dir, source_inputs=source_inputs)
324 build_digest = _asset_tree_digest(build_dir)
325 return (
326 source_digest is not None
327 and build_digest is not None
328 and manifest.get("source_digest") == source_digest
329 and manifest.get("build_digest") == build_digest
330 )
333def _thread_asset_locks() -> dict[str, tuple[bool, int]]:
334 """Return locks held by the current thread for safe nested consumers."""
335 return _asset_thread_state.held
338def _ensure_windows_lock_byte(lock_file: BinaryIO) -> None:
339 """Ensure msvcrt has a real byte range to lock."""
340 lock_file.seek(0, os.SEEK_END)
341 if lock_file.tell() == 0:
342 lock_file.write(b"\0")
343 lock_file.flush()
344 lock_file.seek(0)
347def _windows_lock_is_contended(exc: OSError) -> bool:
348 return exc.errno in {errno.EACCES, errno.EAGAIN, errno.EDEADLK} or getattr(
349 exc,
350 "winerror",
351 None,
352 ) in {32, 33, 36}
355def _file_lock_timeout_seconds(purpose: _FileLockPurpose) -> float:
356 """Return the bounded wait for one class of interprocess lock."""
357 if purpose == "asset":
358 env_name = "GCO_ASSET_LOCK_TIMEOUT_SECONDS"
359 default = _ASSET_LOCK_TIMEOUT_SECONDS_DEFAULT
360 else:
361 env_name = "GCO_CONFIG_LOCK_TIMEOUT_SECONDS"
362 default = _CONFIG_LOCK_TIMEOUT_SECONDS_DEFAULT
364 raw = os.environ.get(env_name, "")
365 try:
366 value = float(raw)
367 except ValueError:
368 return default
369 if not math.isfinite(value) or value <= 0:
370 return default
371 return value
374def _warn_file_lock_contended(
375 lock_name: object,
376 *,
377 exclusive: bool,
378 timeout: float,
379 purpose: _FileLockPurpose,
380) -> None:
381 if purpose == "configuration":
382 logger.warning(
383 "Waiting up to %.0fs for the configuration lock on %s — another CLI "
384 "or MCP process is updating cdk.json. Wait for it to finish; tune via "
385 "GCO_CONFIG_LOCK_TIMEOUT_SECONDS.",
386 timeout,
387 lock_name,
388 )
389 return
391 mode = "exclusive" if exclusive else "shared"
392 logger.warning(
393 "Waiting up to %.0fs for the %s asset lock on %s — another process holds "
394 "it (a pytest session holds shared locks for its whole run; a "
395 "deploy/synth/destroy holds the exclusive lock while rebuilding). "
396 "Find the holder with `lsof %s`; tune via GCO_ASSET_LOCK_TIMEOUT_SECONDS.",
397 timeout,
398 mode,
399 lock_name,
400 lock_name,
401 )
404def _raise_file_lock_timeout(
405 lock_name: object,
406 *,
407 timeout: float,
408 purpose: _FileLockPurpose,
409) -> None:
410 if purpose == "configuration":
411 raise TimeoutError(
412 f"Timed out after {timeout:.0f}s waiting for the configuration lock on "
413 f"{lock_name}. Another CLI or MCP process is updating cdk.json. Wait "
414 "for it to finish and retry; raise GCO_CONFIG_LOCK_TIMEOUT_SECONDS "
415 "to wait longer."
416 )
418 raise TimeoutError(
419 f"Timed out after {timeout:.0f}s waiting for the asset lock on {lock_name}. "
420 "Another process still holds it — often an abandoned pytest session, which "
421 f"keeps shared locks until it exits. Find it with `lsof {lock_name}`, stop "
422 "it, and retry; raise GCO_ASSET_LOCK_TIMEOUT_SECONDS to wait longer."
423 )
426def _posix_lock_is_contended(exc: OSError) -> bool:
427 return isinstance(exc, BlockingIOError) or exc.errno in {errno.EACCES, errno.EAGAIN}
430def _acquire_posix_flock(
431 lock_fd: int,
432 *,
433 lock_name: object,
434 exclusive: bool,
435 purpose: _FileLockPurpose,
436) -> None:
437 """Acquire one POSIX flock with the shared warning and timeout contract."""
438 import fcntl
440 operation = fcntl.LOCK_EX if exclusive else fcntl.LOCK_SH
441 try:
442 fcntl.flock(lock_fd, operation | fcntl.LOCK_NB)
443 return
444 except OSError as exc:
445 if not _posix_lock_is_contended(exc):
446 raise
448 timeout = _file_lock_timeout_seconds(purpose)
449 deadline = time.monotonic() + timeout
450 _warn_file_lock_contended(
451 lock_name,
452 exclusive=exclusive,
453 timeout=timeout,
454 purpose=purpose,
455 )
456 while True:
457 try:
458 fcntl.flock(lock_fd, operation | fcntl.LOCK_NB)
459 return
460 except OSError as exc:
461 if not _posix_lock_is_contended(exc):
462 raise
463 if time.monotonic() >= deadline:
464 _raise_file_lock_timeout(
465 lock_name,
466 timeout=timeout,
467 purpose=purpose,
468 )
469 time.sleep(_FILE_LOCK_RETRY_SECONDS)
472def _acquire_file_lock(
473 lock_file: BinaryIO,
474 *,
475 exclusive: bool,
476 purpose: _FileLockPurpose,
477) -> None:
478 """Acquire a platform-native interprocess lock, loudly and boundedly.
480 The first attempt is non-blocking. On contention a purpose-specific warning
481 names the lock file, then acquisition polls until the env-tunable deadline
482 so a stuck holder produces an actionable error instead of an indefinite
483 silent hang.
484 """
485 if os.name == "nt":
486 import msvcrt
488 msvcrt_api: Any = msvcrt
489 _ensure_windows_lock_byte(lock_file)
490 lock_name = getattr(lock_file, "name", "<unknown>")
491 warned = False
492 deadline: float | None = None
493 timeout: float | None = None
494 while True:
495 lock_file.seek(0)
496 try:
497 # msvcrt exposes only exclusive byte-range locks. Serializing
498 # Windows readers and writers preserves correctness while POSIX
499 # keeps true shared-reader concurrency through flock below.
500 msvcrt_api.locking(lock_file.fileno(), msvcrt_api.LK_NBLCK, 1)
501 return
502 except OSError as exc:
503 if not _windows_lock_is_contended(exc):
504 raise
505 if not warned:
506 timeout = _file_lock_timeout_seconds(purpose)
507 deadline = time.monotonic() + timeout
508 _warn_file_lock_contended(
509 lock_name,
510 exclusive=exclusive,
511 timeout=timeout,
512 purpose=purpose,
513 )
514 warned = True
515 assert deadline is not None and timeout is not None
516 if time.monotonic() >= deadline:
517 _raise_file_lock_timeout(
518 lock_name,
519 timeout=timeout,
520 purpose=purpose,
521 )
522 time.sleep(_FILE_LOCK_RETRY_SECONDS)
524 _acquire_posix_flock(
525 lock_file.fileno(),
526 lock_name=getattr(lock_file, "name", "<unknown>"),
527 exclusive=exclusive,
528 purpose=purpose,
529 )
532def _release_file_lock(lock_file: BinaryIO) -> None:
533 """Release the matching platform-native interprocess lock."""
534 if os.name == "nt":
535 import msvcrt
537 msvcrt_api: Any = msvcrt
538 lock_file.seek(0)
539 msvcrt_api.locking(lock_file.fileno(), msvcrt_api.LK_UNLCK, 1)
540 return
542 import fcntl
544 fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
547@contextmanager
548def _lambda_asset_lock(build_dir: Path, *, exclusive: bool) -> Iterator[None]:
549 """Serialize publishers and keep freshness reads off rename windows."""
550 build_dir.parent.mkdir(parents=True, exist_ok=True)
551 lock_path = build_dir.with_name(f".{build_dir.name}.lock")
552 lock_key = os.path.normcase(os.path.abspath(lock_path))
553 held = _thread_asset_locks()
554 existing = held.get(lock_key)
555 if existing is not None:
556 held_exclusive, depth = existing
557 if exclusive and not held_exclusive:
558 raise RuntimeError(f"Cannot upgrade shared asset lock to exclusive: {lock_path}")
559 held[lock_key] = (held_exclusive, depth + 1)
560 try:
561 yield
562 finally:
563 held[lock_key] = (held_exclusive, depth)
564 return
566 with lock_path.open("a+b") as lock_file:
567 _acquire_file_lock(lock_file, exclusive=exclusive, purpose="asset")
568 held[lock_key] = (exclusive, 1)
569 try:
570 yield
571 finally:
572 held.pop(lock_key, None)
573 _release_file_lock(lock_file)
576def _asset_build_is_fresh(
577 source_dir: Path,
578 build_dir: Path,
579 *,
580 source_inputs: tuple[str, ...] | None,
581) -> bool:
582 with _lambda_asset_lock(build_dir, exclusive=False):
583 return _asset_build_is_fresh_unlocked(
584 source_dir,
585 build_dir,
586 source_inputs=source_inputs,
587 )
590def _remove_asset_tree(path: Path) -> None:
591 if path.exists() or path.is_symlink():
592 _safe_rmtree(path)
595def _recover_interrupted_asset_publish(build_dir: Path) -> None:
596 """Restore a prior final tree and discard abandoned staging directories."""
597 staging_dirs = list(build_dir.parent.glob(f".{build_dir.name}.staging-*"))
598 backup_dirs = list(build_dir.parent.glob(f".{build_dir.name}.backup-*"))
600 if not build_dir.exists() and backup_dirs:
601 try:
602 newest_backup = max(backup_dirs, key=lambda path: path.stat().st_mtime_ns)
603 except OSError:
604 newest_backup = backup_dirs[0]
605 os.replace(newest_backup, build_dir)
607 for path in [*staging_dirs, *backup_dirs]:
608 _remove_asset_tree(path)
611def _publish_staged_asset(staging_dir: Path, build_dir: Path) -> None:
612 """Publish one complete staged tree with rollback to the previous final."""
613 backup_dir = build_dir.with_name(f".{build_dir.name}.backup-{uuid.uuid4().hex}")
614 had_previous = build_dir.exists()
615 if had_previous:
616 os.replace(build_dir, backup_dir)
617 try:
618 os.replace(staging_dir, build_dir)
619 except Exception:
620 if had_previous and backup_dir.exists() and not build_dir.exists():
621 os.replace(backup_dir, build_dir)
622 raise
623 if backup_dir.exists():
624 _remove_asset_tree(backup_dir)
627def _prepare_lambda_asset(
628 source_dir: Path,
629 build_dir: Path,
630 *,
631 source_inputs: tuple[str, ...] | None,
632 display_name: str,
633 builder: Callable[[Path], None],
634) -> bool:
635 """Build and atomically publish an asset when its completion proof is stale.
637 Freshness is checked under a *shared* lock first, so the common case —
638 the asset is already source-current — never contends: concurrent pytest
639 workers validate in parallel instead of serialising behind one writer,
640 and a deploy/destroy against fresh assets never blocks on a pytest
641 session's session-long shared locks. Only a genuinely stale asset
642 escalates to the exclusive publisher lock, which re-checks freshness
643 after acquisition (another publisher may have finished the same rebuild
644 while this one waited).
645 """
646 if _asset_build_is_fresh(source_dir, build_dir, source_inputs=source_inputs):
647 return False
648 with _lambda_asset_lock(build_dir, exclusive=True):
649 _recover_interrupted_asset_publish(build_dir)
650 source_digest = _asset_tree_digest(source_dir, source_inputs=source_inputs)
651 if source_digest is None:
652 raise RuntimeError(f"{display_name} source inputs are incomplete or unreadable")
653 if _asset_build_is_fresh_unlocked(
654 source_dir,
655 build_dir,
656 source_inputs=source_inputs,
657 ):
658 return False
660 print(f" Building {display_name}...")
661 staging_dir = Path(
662 tempfile.mkdtemp(prefix=f".{build_dir.name}.staging-", dir=build_dir.parent)
663 )
664 try:
665 builder(staging_dir)
666 if _asset_tree_digest(source_dir, source_inputs=source_inputs) != source_digest:
667 raise RuntimeError(f"{display_name} sources changed while packaging")
668 _write_build_manifest(staging_dir, source_digest)
669 if not _asset_build_is_fresh_unlocked(
670 source_dir,
671 staging_dir,
672 source_inputs=source_inputs,
673 ):
674 raise RuntimeError(f"{display_name} completion manifest verification failed")
675 _publish_staged_asset(staging_dir, build_dir)
676 finally:
677 _remove_asset_tree(staging_dir)
678 print(f" {display_name} built successfully")
679 return True
682def _atomic_copy_file(source: Path, target: Path) -> None:
683 """Replace one checked-in Lambda source copy without exposing partial bytes."""
684 temporary = target.with_name(f".{target.name}.tmp-{uuid.uuid4().hex}")
685 try:
686 shutil.copy2(source, temporary)
687 os.replace(temporary, target)
688 finally:
689 temporary.unlink(missing_ok=True)
692def _atomic_write_bytes(target: Path, content: bytes, *, mode: int | None = None) -> None:
693 """Atomically restore exact bytes without exposing a partial configuration."""
694 temporary = target.with_name(f".{target.name}.tmp-{uuid.uuid4().hex}")
695 try:
696 temporary.write_bytes(content)
697 if mode is not None:
698 os.chmod(temporary, mode)
699 os.replace(temporary, target)
700 finally:
701 temporary.unlink(missing_ok=True)
704class ConfigMutationLockError(RuntimeError):
705 """The shared cdk.json transaction lock could not be acquired."""
708@contextmanager
709def _config_process_lock(lock_key: Path) -> Iterator[None]:
710 """Hold the platform-native process lock for one config directory."""
711 if os.name == "nt":
712 # Windows cannot open a directory for ``msvcrt.locking``. A persistent
713 # sidecar in that directory gives every CLI/MCP process the same stable
714 # inode even while cdk.json itself is atomically replaced.
715 lock_path = lock_key / _CONFIG_LOCK_FILENAME
716 lock_file: BinaryIO | None = None
717 try:
718 lock_file = lock_path.open("a+b")
719 _acquire_file_lock(lock_file, exclusive=True, purpose="configuration")
720 except OSError as exc:
721 if lock_file is not None:
722 lock_file.close()
723 raise ConfigMutationLockError(
724 f"could not lock configuration directory {lock_key}: {exc}"
725 ) from exc
727 try:
728 yield
729 finally:
730 assert lock_file is not None
731 try:
732 _release_file_lock(lock_file)
733 finally:
734 lock_file.close()
735 return
737 # Keep the POSIX directory lock: unlike a lock on cdk.json, the descriptor
738 # continues to identify the same object when an atomic writer replaces the
739 # configuration file.
740 import fcntl
742 flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0)
743 flags |= getattr(os, "O_DIRECTORY", 0)
744 lock_fd: int | None = None
745 try:
746 lock_fd = os.open(lock_key, flags)
747 _acquire_posix_flock(
748 lock_fd,
749 lock_name=str(lock_key),
750 exclusive=True,
751 purpose="configuration",
752 )
753 except OSError as exc:
754 if lock_fd is not None:
755 os.close(lock_fd)
756 raise ConfigMutationLockError(
757 f"could not lock configuration directory {lock_key}: {exc}"
758 ) from exc
760 try:
761 yield
762 finally:
763 assert lock_fd is not None
764 try:
765 fcntl.flock(lock_fd, fcntl.LOCK_UN)
766 finally:
767 os.close(lock_fd)
770@contextmanager
771def _config_mutation_lock(path: Path) -> Iterator[None]:
772 """Serialize a complete read/modify/replace transaction for ``path``.
774 POSIX locks the stable directory descriptor; Windows locks a persistent
775 sidecar in that directory. A thread-local held-set makes this context
776 reentrant, which is required by analytics teardown: it holds the
777 transaction across its temporary mutation and nested feature-toggle writes.
778 """
779 lock_key = path.parent.resolve()
780 with _CONFIG_THREAD_LOCKS_GUARD:
781 thread_lock = _CONFIG_THREAD_LOCKS.setdefault(lock_key, RLock())
783 with thread_lock:
784 held_directories = getattr(_CONFIG_LOCK_STATE, "held_directories", None)
785 if held_directories is None:
786 held_directories = set()
787 _CONFIG_LOCK_STATE.held_directories = held_directories
788 if lock_key in held_directories:
789 yield
790 return
792 with _config_process_lock(lock_key):
793 held_directories.add(lock_key)
794 try:
795 yield
796 finally:
797 held_directories.discard(lock_key)
800@lru_cache(maxsize=1)
801def _known_cloudformation_regions() -> frozenset[str]:
802 """Return every AWS SDK-known Region that exposes CloudFormation."""
803 return known_cloudformation_regions()
806class CdkToolchainError(RuntimeError):
807 """The CDK Python toolchain (``aws-cdk-lib`` / ``cdk-nag``) is not
808 importable in the environment that will run ``cdk``.
810 Raised before shelling out to ``cdk`` so operators get a clear install
811 hint instead of the cryptic ``ImportError: cannot import name 'App' from
812 'aws_cdk'`` that the ``python3 app.py`` synth subprocess would otherwise
813 emit from a base (extra-less) install.
814 """
817@dataclass
818class StackInfo:
819 """Information about a CDK stack."""
821 name: str
822 status: str
823 region: str
824 created_time: datetime | None = None
825 updated_time: datetime | None = None
826 outputs: dict[str, str] = field(default_factory=dict)
827 tags: dict[str, str] = field(default_factory=dict)
829 def to_dict(self) -> dict[str, Any]:
830 return {
831 "name": self.name,
832 "status": self.status,
833 "region": self.region,
834 "created_time": self.created_time.isoformat() if self.created_time else None,
835 "updated_time": self.updated_time.isoformat() if self.updated_time else None,
836 "outputs": self.outputs,
837 "tags": self.tags,
838 }
841def _safe_rmtree(path: Path) -> None:
842 """Remove a directory tree, handling broken symlinks on macOS.
844 shutil.rmtree can fail with ``OSError: [Errno 66] Directory not empty``
845 on macOS when pip-installed packages (e.g. botocore) contain broken
846 symlinks or extended-attribute resource forks.
848 Falls back to ``rm -rf`` via subprocess, but only after validating the
849 path is a real directory under the project tree to avoid accidents.
850 """
851 resolved = path.resolve()
853 # Safety: refuse to remove anything that isn't clearly a final, staging,
854 # or rollback Lambda build artifact inside the project tree.
855 artifact_name = resolved.name
856 is_final = artifact_name.endswith("-build")
857 is_ephemeral = artifact_name.startswith(".") and (
858 ".staging-" in artifact_name or ".backup-" in artifact_name
859 )
860 if "lambda" not in resolved.parts or not (is_final or is_ephemeral):
861 raise ValueError(f"Refusing to remove unexpected path: {resolved}")
863 try:
864 shutil.rmtree(str(resolved))
865 except OSError:
866 subprocess.run(["rm", "-rf", "--", str(resolved)], check=True)
869# Container runtime detection lives in cli/_container_runtime.py so it can
870# be shared between StackManager (CDK asset bundling) and ImageManager
871# (gco images build/push). The uncached probe is imported from there;
872# this module keeps its own small cache so existing tests that reset
873# ``cli.stacks._container_runtime_cache`` continue to work without
874# touching the new module's cache.
875from cli._container_runtime import ( # noqa: E402
876 _detect_container_runtime_uncached,
877)
879# Cached result for container runtime detection (None = not yet checked)
880_container_runtime_cache: str | None = None
881_container_runtime_checked: bool = False
884def _detect_container_runtime() -> str | None:
885 """
886 Detect available container runtime for CDK asset bundling.
888 Thin caching wrapper around the shared
889 ``cli._container_runtime._detect_container_runtime_uncached`` probe.
890 The cache state is held on this module so tests that patch or reset
891 ``cli.stacks._container_runtime_cache`` keep working unchanged.
892 """
893 global _container_runtime_cache, _container_runtime_checked
894 if _container_runtime_checked:
895 return _container_runtime_cache
897 _container_runtime_cache = _detect_container_runtime_uncached()
898 _container_runtime_checked = True
899 return _container_runtime_cache
902def prepare_cdk_assets(project_root: str | Path) -> None:
903 """Prepare every ignored Lambda asset consumed by the CDK application.
905 This is the shared entry point for build-only callers. CDK consumers must
906 use :func:`cdk_asset_consumer` so the resulting paths remain immutable
907 until app construction and synthesis finish.
908 """
909 manager = StackManager.__new__(StackManager)
910 manager.project_root = Path(project_root)
911 manager._ensure_lambda_build()
914def _thread_asset_consumers() -> dict[str, int]:
915 return _asset_thread_state.active_consumers
918@contextmanager
919def cdk_asset_consumer(project_root: str | Path) -> Iterator[None]:
920 """Hold source-current generated assets stable through CDK synthesis.
922 Preparation runs before any shared locks are acquired. The complete set of
923 canonical paths is then locked in deterministic order and every completion
924 manifest is revalidated while publishers are excluded. A stale observation
925 releases all locks and retries preparation; repeated source churn fails
926 closed instead of exposing CDK to a missing or mixed-version tree.
927 """
928 root = Path(project_root)
929 root_key = os.path.normcase(os.path.abspath(root))
930 active = _thread_asset_consumers()
931 if root_key in active:
932 active[root_key] += 1
933 try:
934 yield
935 finally:
936 active[root_key] -= 1
937 return
939 stale_assets: list[str] = []
940 # Attempt 0 validates under shared locks without preparing anything: when
941 # every asset is already source-current (always true in CI, where the
942 # composite build action runs first, and true locally on any second run)
943 # the consumer takes no exclusive lock and does one hash pass. Concurrent
944 # consumers — xdist workers — therefore proceed in parallel instead of
945 # serialising behind the publisher lock. Later attempts keep the original
946 # prepare-then-revalidate budget for genuinely stale trees.
947 for attempt in range(_CDK_ASSET_CONSUMER_MAX_ATTEMPTS + 1):
948 if attempt:
949 prepare_cdk_assets(root)
950 resolved_assets = []
951 for spec in _CDK_ASSET_SPECS:
952 source_dir, build_dir = spec.paths(root)
953 # Include a source-backed path even during the publisher's
954 # final-to-backup rename gap, when the canonical build is absent.
955 if source_dir.exists() or build_dir.exists():
956 resolved_assets.append((spec, source_dir, build_dir))
958 with ExitStack() as locks:
959 for _spec, _source_dir, build_dir in sorted(
960 resolved_assets,
961 key=lambda item: str(item[2]),
962 ):
963 locks.enter_context(_lambda_asset_lock(build_dir, exclusive=False))
965 stale_assets = [
966 spec.name
967 for spec, source_dir, build_dir in resolved_assets
968 if not _asset_build_is_fresh_unlocked(
969 source_dir,
970 build_dir,
971 source_inputs=spec.source_inputs,
972 )
973 ]
974 if stale_assets:
975 continue
977 active[root_key] = 1
978 try:
979 yield
980 finally:
981 active.pop(root_key, None)
982 return
984 names = ", ".join(stale_assets) or "unknown assets"
985 raise RuntimeError(
986 "Generated CDK assets changed repeatedly while acquiring consumer locks: "
987 f"{names}. Stop concurrent source edits and retry."
988 )
991class StackManager:
992 """Manages CDK stack operations."""
994 def __init__(self, config: GCOConfig, project_root: Path | None = None):
995 self.config = config
996 self.project_root = project_root or self._find_project_root()
997 # Resolve CDK only when a CDK-backed operation runs. CloudFormation-only
998 # status/output commands must not require a local Node/CDK installation.
999 self._cdk_path: str | None = None
1000 self._active_cdk_processes: dict[int, Any] = {}
1001 self._active_cdk_lock = Lock()
1002 self._cdk_cancel_event = Event()
1003 # Extra `--context key=value` pairs appended to every app-evaluating
1004 # CDK invocation (deploy/destroy/diff/list/synth). Set once via
1005 # set_extra_cdk_context; used by the live release validation harness
1006 # to force-enable optional Helm charts (helm_enabled_overrides) for a
1007 # run without mutating the checked-out cdk.json.
1008 self._extra_cdk_context: dict[str, str] = {}
1010 def set_extra_cdk_context(self, context: Mapping[str, str]) -> None:
1011 """Register `--context` pairs for every subsequent CDK invocation.
1013 Keys and values must be plain strings without shell metacharacters'
1014 risk (argv is passed as a list, never a shell string); a key that is
1015 empty or contains ``=`` is refused because it could not round-trip
1016 through the CDK CLI's ``key=value`` form unambiguously.
1017 """
1018 validated: dict[str, str] = {}
1019 for key, value in context.items():
1020 if not key or "=" in key:
1021 raise ValueError(f"Invalid CDK context key: {key!r}")
1022 validated[str(key)] = str(value)
1023 self._extra_cdk_context = validated
1025 def _find_project_root(self) -> Path:
1026 """Find the project root by looking for cdk.json."""
1027 current = Path.cwd()
1028 for parent in [current] + list(current.parents):
1029 if (parent / "cdk.json").exists():
1030 return parent
1031 return current
1033 def _find_cdk(self) -> str:
1034 """Find the dependency-locked CDK executable when available."""
1035 # Prefer the repository's locked tool when ``npm ci`` has populated it.
1036 local_cdk = self.project_root / "node_modules" / ".bin" / "cdk"
1037 if local_cdk.is_file():
1038 return str(local_cdk)
1040 # Fall back to PATH for installed distributions that do not include
1041 # the repository's root npm graph.
1042 try:
1043 result = subprocess.run(["which", "cdk"], capture_output=True, text=True, check=True)
1044 return result.stdout.strip()
1045 except subprocess.CalledProcessError:
1046 pass
1048 # Check common global-install locations.
1049 for path in ["/usr/local/bin/cdk", "~/.npm-global/bin/cdk"]:
1050 expanded = os.path.expanduser(path)
1051 if os.path.exists(expanded):
1052 return expanded
1054 raise CdkToolchainError(
1055 "AWS CDK CLI is not installed. Run "
1056 "'npm ci --ignore-scripts --no-audit --no-fund' at the project root "
1057 "to install the dependency-locked CLI."
1058 )
1060 @staticmethod
1061 def _kubectl_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
1062 """Return whether the kubectl build has a valid full-tree completion proof."""
1063 return _asset_build_is_fresh(
1064 source_dir,
1065 build_dir,
1066 source_inputs=_KUBECTL_CDK_ASSET.source_inputs,
1067 )
1069 @staticmethod
1070 def _helm_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
1071 """Return whether the Helm build has a valid full-tree completion proof."""
1072 return _asset_build_is_fresh(
1073 source_dir,
1074 build_dir,
1075 source_inputs=_HELM_CDK_ASSET.source_inputs,
1076 )
1078 @staticmethod
1079 def _inference_streaming_build_is_fresh(source_dir: Path, build_dir: Path) -> bool:
1080 """Return whether the Node build has a valid full-tree completion proof."""
1081 return _asset_build_is_fresh(
1082 source_dir,
1083 build_dir,
1084 source_inputs=_INFERENCE_STREAMING_CDK_ASSET.source_inputs,
1085 )
1087 def _ensure_lambda_build(self) -> None:
1088 """Atomically prepare every generated Lambda asset when source-stale.
1090 Every builder takes its per-asset interprocess lock, repairs an
1091 interrupted publish, and rechecks freshness before doing installation
1092 work. Concurrent app evaluations therefore either reuse one complete
1093 final tree or publish another complete tree; they never share a
1094 directory while pip/npm/copy operations are mutating it.
1095 """
1096 for spec, builder in (
1097 (_KUBECTL_CDK_ASSET, self._build_kubectl_lambda),
1098 (_HELM_CDK_ASSET, self._build_helm_installer_lambda),
1099 (_INFERENCE_STREAMING_CDK_ASSET, self._build_inference_streaming_proxy_lambda),
1100 ):
1101 source_dir, _build_dir = spec.paths(self.project_root)
1102 if source_dir.exists():
1103 builder()
1105 def _check_and_fix_stuck_stack(
1106 self,
1107 stack_name: str,
1108 *,
1109 expected_stack_id: str | None = None,
1110 authorize_stack: StackAuthorizationCallback | None = None,
1111 strict_ownership: bool = False,
1112 ) -> None:
1113 """Delete a stuck stack only after revalidating its immutable identity."""
1114 import boto3
1116 region = self._get_deploy_region(stack_name)
1117 if not region:
1118 if strict_ownership:
1119 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
1120 return
1122 cfn = boto3.client("cloudformation", region_name=region)
1123 try:
1124 response = cfn.describe_stacks(StackName=stack_name)
1125 except ClientError as exc:
1126 error = exc.response.get("Error", {})
1127 if (
1128 error.get("Code") == "ValidationError"
1129 and "does not exist" in str(error.get("Message", "")).lower()
1130 ):
1131 return
1132 if strict_ownership:
1133 raise
1134 logger.debug("Stack pre-check for %s failed: %s", stack_name, exc)
1135 return
1136 except Exception as exc:
1137 if strict_ownership:
1138 raise
1139 logger.debug("Stack pre-check for %s failed: %s", stack_name, exc)
1140 return
1142 stacks = response.get("Stacks", [])
1143 if len(stacks) != 1:
1144 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
1145 stack = stacks[0]
1146 stack_id = str(stack.get("StackId") or "")
1147 if stack.get("StackName") != stack_name or not stack_id:
1148 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
1149 if strict_ownership and expected_stack_id is None:
1150 raise RuntimeError(
1151 f"Refusing to adopt uncheckpointed stack {region}:{stack_name} ({stack_id})"
1152 )
1153 if expected_stack_id is not None and stack_id != expected_stack_id:
1154 raise RuntimeError(
1155 f"Stack identity changed for {region}:{stack_name}; expected {expected_stack_id}, "
1156 f"found {stack_id}"
1157 )
1159 stuck_states = {
1160 "REVIEW_IN_PROGRESS",
1161 "ROLLBACK_COMPLETE",
1162 "ROLLBACK_FAILED",
1163 "CREATE_FAILED",
1164 "DELETE_FAILED",
1165 }
1166 status = str(stack.get("StackStatus") or "")
1167 if status not in stuck_states:
1168 return
1169 if authorize_stack is not None:
1170 authorize_stack(stack_name, region, stack_id)
1172 print(f" Stack {stack_name} is in {status} state, cleaning up...")
1173 cfn.delete_stack(StackName=stack_id)
1174 waiter = cfn.get_waiter("stack_delete_complete")
1175 waiter.wait(StackName=stack_id, WaiterConfig={"Delay": 10, "MaxAttempts": 60})
1176 print(f" Stack {stack_name} cleaned up, will recreate on deploy")
1178 #: Resource statuses that carry the *cause* of a failed stack operation.
1179 _ROOT_CAUSE_STATUSES = frozenset(
1180 {"CREATE_FAILED", "UPDATE_FAILED", "DELETE_FAILED", "IMPORT_FAILED"}
1181 )
1182 #: Reasons CloudFormation attaches to resources it merely abandoned because
1183 #: a sibling failed first; they are noise next to the real failure.
1184 _CASCADE_REASON_MARKERS = ("Resource creation cancelled", "Resource update cancelled")
1185 #: Pagination ceiling for the operation-event walk (100 events per page).
1186 _MAX_EVENT_PAGES = 10
1187 #: How many root-cause events to print before truncating.
1188 _MAX_DIAGNOSED_EVENTS = 5
1190 @classmethod
1191 def _collect_operation_events(cls, cfn: Any, stack_name: str) -> list[dict[str, Any]]:
1192 """Return the newest-first events of the stack's most recent operation.
1194 ``describe_stack_events`` pages newest first and a rolled-back create
1195 of a large stack buries the one ``CREATE_FAILED`` that explains
1196 everything under dozens of ``DELETE_COMPLETE`` rollback rows — the
1197 first page alone is not enough. Walk pages until the stack-level
1198 ``*_IN_PROGRESS`` event CloudFormation records as "User Initiated"
1199 (the operation's start), bounded by ``_MAX_EVENT_PAGES``.
1200 """
1201 collected: list[dict[str, Any]] = []
1202 token: str | None = None
1203 for _ in range(cls._MAX_EVENT_PAGES):
1204 kwargs: dict[str, Any] = {"StackName": stack_name}
1205 if token:
1206 kwargs["NextToken"] = token
1207 response = cfn.describe_stack_events(**kwargs)
1208 for event in response.get("StackEvents", []):
1209 collected.append(event)
1210 status = str(event.get("ResourceStatus") or "")
1211 if (
1212 event.get("LogicalResourceId") == stack_name
1213 and status.endswith("_IN_PROGRESS")
1214 and not status.startswith("ROLLBACK")
1215 and not status.startswith("UPDATE_ROLLBACK")
1216 and "User Initiated" in str(event.get("ResourceStatusReason") or "")
1217 ):
1218 return collected
1219 token_value = response.get("NextToken")
1220 token = token_value if isinstance(token_value, str) and token_value else None
1221 if not token:
1222 break
1223 return collected
1225 @classmethod
1226 def _summarize_failure_events(
1227 cls, events: list[dict[str, Any]], stack_name: str
1228 ) -> list[dict[str, Any]]:
1229 """Pick the events worth showing: resource root causes, then the stack verdict.
1231 Root causes are resource-level ``*_FAILED`` events whose reason is not
1232 a cascade marker; they are printed oldest first so the first failure —
1233 the one that triggered the rollback — leads. The stack-level
1234 ``ROLLBACK_IN_PROGRESS``/``*_FAILED`` event follows because its reason
1235 lists every failed logical id. When no root cause survives the filter
1236 (the event window was exhausted), fall back to any failed/rollback
1237 events so the operator still sees CloudFormation's own words.
1238 """
1239 root_causes = [
1240 event
1241 for event in reversed(events)
1242 if str(event.get("ResourceStatus") or "") in cls._ROOT_CAUSE_STATUSES
1243 and event.get("LogicalResourceId") != stack_name
1244 and not any(
1245 marker in str(event.get("ResourceStatusReason") or "")
1246 for marker in cls._CASCADE_REASON_MARKERS
1247 )
1248 ]
1249 stack_verdicts = [
1250 event
1251 for event in events
1252 if event.get("LogicalResourceId") == stack_name
1253 and (
1254 "ROLLBACK" in str(event.get("ResourceStatus") or "")
1255 or "FAILED" in str(event.get("ResourceStatus") or "")
1256 )
1257 and event.get("ResourceStatusReason")
1258 ]
1259 selected = root_causes[: cls._MAX_DIAGNOSED_EVENTS] + stack_verdicts[:1]
1260 if selected:
1261 return selected
1262 return [
1263 event
1264 for event in events
1265 if "FAILED" in str(event.get("ResourceStatus") or "")
1266 or "ROLLBACK" in str(event.get("ResourceStatus") or "")
1267 ][: cls._MAX_DIAGNOSED_EVENTS]
1269 def _diagnose_deploy_failure(self, stack_name: str) -> None:
1270 """Fetch CloudFormation events after a failed deploy and print diagnostics.
1272 Gives users actionable information instead of just the CDK error
1273 message: the resource-level reason that actually failed (an S3
1274 ``BucketAlreadyExists``, an IAM propagation error, ...) rather than
1275 the stack's bare ``ROLLBACK_COMPLETE``.
1276 """
1277 import boto3
1279 region = self._get_deploy_region(stack_name)
1280 if not region:
1281 return
1283 try:
1284 cfn = boto3.client("cloudformation", region_name=region)
1286 events = self._collect_operation_events(cfn, stack_name)
1287 failed = self._summarize_failure_events(events, stack_name)
1289 if failed:
1290 print(f"\n CloudFormation failure details for {stack_name}:")
1291 for event in failed:
1292 resource = event.get("LogicalResourceId", "unknown")
1293 resource_type = event.get("ResourceType")
1294 status = event.get("ResourceStatus", "unknown")
1295 reason = event.get("ResourceStatusReason", "no reason given")
1296 label = f"{resource} ({resource_type})" if resource_type else str(resource)
1297 print(f" {label}: {status}")
1298 print(f" {reason}")
1300 # Check stack status for actionable advice
1301 try:
1302 stack_resp = cfn.describe_stacks(StackName=stack_name)
1303 status = stack_resp["Stacks"][0]["StackStatus"]
1305 advice = {
1306 "REVIEW_IN_PROGRESS": (
1307 "Stack is stuck in REVIEW_IN_PROGRESS. "
1308 "Run: aws cloudformation delete-stack "
1309 f"--stack-name {stack_name} --region {region}"
1310 ),
1311 "ROLLBACK_COMPLETE": (
1312 "Stack rolled back. Delete it and retry: "
1313 f"aws cloudformation delete-stack "
1314 f"--stack-name {stack_name} --region {region}"
1315 ),
1316 "ROLLBACK_FAILED": (
1317 "Stack rollback failed. Delete with --retain: "
1318 f"aws cloudformation delete-stack "
1319 f"--stack-name {stack_name} --region {region}"
1320 ),
1321 "UPDATE_ROLLBACK_COMPLETE": (
1322 "Update rolled back but stack is stable. "
1323 "Check the events above and retry the deploy."
1324 ),
1325 }
1327 if status in advice:
1328 print(f"\n Suggested fix: {advice[status]}")
1330 except Exception as e:
1331 logger.debug("Failed to parse stack events: %s", e)
1333 except Exception as e:
1334 logger.debug("Failed to diagnose deploy failure for %s: %s", stack_name, e)
1335 # Best effort — don't fail the deploy further
1337 def _sync_lambda_sources(self) -> None:
1338 """Atomically synchronize canonical shared files before asset ensures.
1340 Checked-in copies keep raw CDK evaluation deterministic. Deploy updates
1341 those copies before generated assets are checked, and never mutates a
1342 generated final build tree in place. The source->targets mapping lives
1343 in ``gco.lambda_shared_sources`` so deploy packaging, diagram
1344 reconciliation, and commit-time identity tests consume one
1345 dependency-light inventory.
1346 """
1347 if getattr(self, "_lambda_sources_synced", False):
1348 return
1350 for source_rel, target_rels in LAMBDA_SHARED_SOURCE_TARGETS.items():
1351 shared_source = self.project_root / source_rel
1352 if not shared_source.exists():
1353 continue
1354 for target_rel in target_rels:
1355 target = self.project_root / target_rel
1356 if target.parent.exists():
1357 _atomic_copy_file(shared_source, target)
1358 self._lambda_sources_synced = True
1360 def _rebuild_lambda_packages(self) -> None:
1361 """Compatibility wrapper for a source-current atomic asset ensure."""
1362 if getattr(self, "_lambda_packages_rebuilt", False):
1363 return
1364 self._ensure_lambda_build()
1365 self._lambda_packages_rebuilt = True
1367 def _build_lambda_packages(self) -> None:
1368 """Source-check and atomically publish all generated Lambda packages."""
1369 self._build_kubectl_lambda()
1370 self._build_helm_installer_lambda()
1371 self._build_inference_streaming_proxy_lambda()
1373 def _build_kubectl_lambda(self) -> None:
1374 """Build the kubectl-applier-simple Lambda package."""
1375 source_dir, build_dir = _KUBECTL_CDK_ASSET.paths(self.project_root)
1376 requirements = source_dir / "requirements.txt"
1377 if not source_dir.is_dir() or not requirements.is_file():
1378 return
1380 def build(staging_dir: Path) -> None:
1381 shutil.copy2(source_dir / "handler.py", staging_dir / "handler.py")
1382 shutil.copy2(requirements, staging_dir / "requirements.txt")
1383 shutil.copytree(source_dir / "manifests", staging_dir / "manifests")
1384 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - static pip arguments and project-owned paths
1385 [
1386 sys.executable,
1387 "-m",
1388 "pip",
1389 "install",
1390 "-r",
1391 str(requirements),
1392 "-t",
1393 str(staging_dir),
1394 "--upgrade",
1395 "--platform",
1396 "manylinux2014_x86_64",
1397 "--only-binary=:all:",
1398 "--quiet",
1399 ],
1400 capture_output=True,
1401 text=True,
1402 )
1403 if result.returncode != 0:
1404 raise RuntimeError(
1405 "kubectl Lambda dependency installation failed: " + result.stderr[:200]
1406 )
1408 _prepare_lambda_asset(
1409 source_dir,
1410 build_dir,
1411 source_inputs=_KUBECTL_CDK_ASSET.source_inputs,
1412 display_name="kubectl-applier-simple Lambda package",
1413 builder=build,
1414 )
1416 def _build_helm_installer_lambda(self) -> None:
1417 """Build the complete helm-installer Lambda Docker context."""
1418 source_dir, build_dir = _HELM_CDK_ASSET.paths(self.project_root)
1419 if not source_dir.is_dir():
1420 return
1422 def build(staging_dir: Path) -> None:
1423 shutil.copytree(
1424 source_dir,
1425 staging_dir,
1426 ignore=shutil.ignore_patterns(*_LAMBDA_SOURCE_COPY_IGNORE_PATTERNS),
1427 dirs_exist_ok=True,
1428 )
1430 _prepare_lambda_asset(
1431 source_dir,
1432 build_dir,
1433 source_inputs=_HELM_CDK_ASSET.source_inputs,
1434 display_name="helm-installer Lambda package",
1435 builder=build,
1436 )
1438 def _build_inference_streaming_proxy_lambda(self) -> None:
1439 """Build the Node.js streaming Lambda with its pinned AWS SDK clients."""
1440 source_dir, build_dir = _INFERENCE_STREAMING_CDK_ASSET.paths(self.project_root)
1441 if not source_dir.is_dir():
1442 return
1444 package_files = _INFERENCE_STREAMING_CDK_ASSET.source_inputs
1445 assert package_files is not None
1446 missing = [name for name in package_files if not (source_dir / name).is_file()]
1447 if missing:
1448 raise RuntimeError(
1449 "Inference streaming Lambda package is incomplete; missing: " + ", ".join(missing)
1450 )
1451 try:
1452 package_manager = str(
1453 json.loads((source_dir / "package.json").read_text(encoding="utf-8")).get(
1454 "packageManager", ""
1455 )
1456 )
1457 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
1458 raise RuntimeError("Unable to read the inference streaming Lambda npm pin") from exc
1459 required_npm = package_manager.removeprefix("npm@")
1460 version_parts = required_npm.split(".")
1461 if (
1462 not package_manager.startswith("npm@")
1463 or len(version_parts) != 3
1464 or any(not part.isdigit() for part in version_parts)
1465 ):
1466 raise RuntimeError(
1467 "Inference streaming Lambda packageManager must pin an exact npm version"
1468 )
1470 def build(staging_dir: Path) -> None:
1471 npm = shutil.which("npm")
1472 if npm is None:
1473 raise RuntimeError(
1474 f"npm {required_npm} is required to package the inference streaming Lambda; "
1475 "install the Node.js version pinned in .nvmrc"
1476 )
1477 try:
1478 version_result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved executable and project-owned cwd
1479 [npm, "--version"],
1480 cwd=source_dir,
1481 capture_output=True,
1482 text=True,
1483 timeout=30,
1484 )
1485 except (OSError, subprocess.TimeoutExpired) as exc:
1486 raise RuntimeError("Unable to verify the npm packaging version") from exc
1487 actual_npm = version_result.stdout.strip()
1488 if version_result.returncode != 0 or actual_npm != required_npm:
1489 found = actual_npm or "unavailable"
1490 raise RuntimeError(
1491 f"npm {required_npm} is required to package the inference streaming Lambda; "
1492 f"found {found}. Run: npm install --global npm@{required_npm}"
1493 )
1495 for name in package_files:
1496 shutil.copy2(source_dir / name, staging_dir / name)
1497 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved npm path, static arguments, and project-owned cwd
1498 [
1499 npm,
1500 "ci",
1501 "--omit=dev",
1502 "--ignore-scripts",
1503 "--no-audit",
1504 "--no-fund",
1505 ],
1506 cwd=staging_dir,
1507 capture_output=True,
1508 text=True,
1509 )
1510 if result.returncode != 0:
1511 raise RuntimeError(
1512 "Failed to install pinned inference streaming Lambda dependencies: "
1513 + result.stderr[:500]
1514 )
1516 _prepare_lambda_asset(
1517 source_dir,
1518 build_dir,
1519 source_inputs=package_files,
1520 display_name="inference-streaming-proxy Lambda package",
1521 builder=build,
1522 )
1524 def _get_python_path(self) -> str:
1525 """
1526 Get PYTHONPATH that includes the current Python's site-packages.
1528 This is critical for pipx installations where CDK runs `python3 app.py`
1529 using the system Python, which doesn't have aws_cdk installed.
1530 By setting PYTHONPATH, we ensure CDK's subprocess can find our modules.
1531 """
1532 # Get all site-packages directories from the current Python
1533 site_packages = site.getsitepackages()
1535 # Also include user site-packages if available
1536 user_site = site.getusersitepackages()
1537 if user_site and os.path.isdir(user_site):
1538 site_packages.append(user_site)
1540 # Include the directory containing the current module (for editable installs)
1541 current_module_dir = Path(__file__).parent.parent
1542 if current_module_dir.exists():
1543 site_packages.append(str(current_module_dir))
1545 # Combine with existing PYTHONPATH if any
1546 existing_path = os.environ.get("PYTHONPATH", "")
1547 all_paths = site_packages + ([existing_path] if existing_path else [])
1549 return os.pathsep.join(all_paths)
1551 def _ensure_cdk_toolchain(self) -> None:
1552 """Preflight the CDK Python toolchain before invoking ``cdk``.
1554 Infra operations run ``python3 app.py`` (via the Node ``cdk`` CLI),
1555 which imports ``aws_cdk`` and ``cdk_nag``. Those ship in the optional
1556 ``[cdk]`` extra — a base ``uvx`` / ``pip`` install of ``gco-cli`` does
1557 not include them, so the synth subprocess fails with a cryptic
1558 ``ImportError: cannot import name 'App' from 'aws_cdk'``. Detect the
1559 missing toolchain up front and raise :class:`CdkToolchainError` with an
1560 actionable install hint instead.
1561 """
1562 missing = [m for m in _CDK_TOOLCHAIN_MODULES if importlib.util.find_spec(m) is None]
1563 if not missing:
1564 return
1565 raise CdkToolchainError(
1566 "CDK toolchain not available: cannot import "
1567 + ", ".join(missing)
1568 + ".\nInfrastructure operations (deploy / synth / diff / list / destroy / "
1569 "bootstrap) need the CDK Python packages installed in the SAME "
1570 "environment as the `gco` CLI, plus a repository checkout providing "
1571 "`app.py` and `cdk.json`.\n"
1572 "Install the `[cdk]` extra one of these ways:\n"
1573 ' - uv: uv tool install "gco-cli[cdk] @ '
1574 'git+https://github.com/aws-solutions-library-samples/global-capacity-orchestrator-on-aws.git@<tag>"\n'
1575 ' - pip: pip install -e ".[cdk,mcp]" (from a clone)\n'
1576 " - or use the dev container (see QUICKSTART.md), which bundles the "
1577 "full toolchain.\n"
1578 "See gco_mcp/README.md (Setup) for the deploy-capable configuration."
1579 )
1581 @staticmethod
1582 def _terminate_cdk_process(process: Any) -> None:
1583 """Terminate one complete CDK process tree with a bounded grace period."""
1584 if process.poll() is not None:
1585 return
1587 if os.name == "nt":
1588 taskkill = shutil.which("taskkill.exe") or shutil.which("taskkill")
1590 def terminate_tree(*, force: bool) -> bool:
1591 if taskkill is None:
1592 return False
1593 command = [taskkill, "/PID", str(process.pid), "/T"]
1594 if force:
1595 command.append("/F")
1596 try:
1597 result = subprocess.run( # nosemgrep: dangerous-subprocess-use-audit - resolved Windows system utility and numeric child PID
1598 command,
1599 capture_output=True,
1600 text=True,
1601 check=False,
1602 timeout=30,
1603 creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
1604 )
1605 except OSError, subprocess.TimeoutExpired:
1606 return False
1607 return result.returncode == 0
1609 terminate_tree(force=False)
1610 try:
1611 process.wait(timeout=30)
1612 except OSError, subprocess.TimeoutExpired:
1613 terminate_tree(force=True)
1614 if process.poll() is None:
1615 process.kill()
1616 process.wait()
1617 return
1619 try:
1620 os.killpg(process.pid, signal.SIGTERM)
1621 process.wait(timeout=30)
1622 except OSError, subprocess.TimeoutExpired:
1623 if process.poll() is None:
1624 try:
1625 os.killpg(process.pid, signal.SIGKILL)
1626 finally:
1627 process.wait()
1629 def cancel_active_cdk_processes(self) -> None:
1630 """Prevent new CDK work and terminate every process group currently registered."""
1631 self._cdk_cancel_event.set()
1632 with self._active_cdk_lock:
1633 processes = list(self._active_cdk_processes.values())
1634 for process in processes:
1635 self._terminate_cdk_process(process)
1637 def _run_cdk(
1638 self,
1639 command: list[str],
1640 capture_output: bool = False,
1641 env: dict[str, str] | None = None,
1642 timeout: float | None = None,
1643 ) -> subprocess.CompletedProcess[str]:
1644 """Run a CDK command.
1646 Args:
1647 command: CDK subcommand argv (e.g. ``["destroy", "gco-us-east-1", "--force"]``).
1648 capture_output: Capture stdout / stderr instead of streaming.
1649 env: Extra env vars merged onto the parent process environment.
1650 timeout: Wall-clock timeout in seconds. ``None`` (default) waits
1651 forever — preserving the old behaviour for ``synth`` / ``list``.
1652 When set, on timeout we send SIGTERM, give the CDK process up
1653 to 30 seconds to exit cleanly, then SIGKILL, and finally
1654 re-raise ``subprocess.TimeoutExpired`` so callers can decide
1655 how to handle a hung subprocess. ``deploy()`` and ``destroy()``
1656 pass a per-stack budget so a wedged ``cdk destroy`` (e.g. its
1657 post-delete polling loop hanging after CloudFormation has
1658 already finished) can't block the orchestrator forever.
1659 """
1660 # Fail fast with an actionable message when the CDK Python toolchain
1661 # isn't importable (e.g. a base uvx/pip install without the [cdk]
1662 # extra), instead of letting the ``python3 app.py`` subprocess surface
1663 # a cryptic ImportError.
1664 self._ensure_cdk_toolchain()
1666 # These commands all evaluate app.py, including list and destroy. The
1667 # stack graph references ignored generated Lambda assets, so prepare
1668 # them centrally rather than relying on individual command wrappers.
1669 if command and command[0] in {"deploy", "destroy", "diff", "list", "synth"}:
1670 self._ensure_lambda_build()
1671 # Apply registered context overrides uniformly to every
1672 # app-evaluating command so deploy, destroy, and the stack listing
1673 # all synthesize the same graph (see set_extra_cdk_context).
1674 for key, value in sorted(self._extra_cdk_context.items()):
1675 command = [*command, "--context", f"{key}={value}"]
1677 full_env = os.environ.copy()
1679 # Inject PYTHONPATH so CDK's python3 subprocess can find aws_cdk
1680 # This is essential for pipx installations
1681 full_env["PYTHONPATH"] = self._get_python_path()
1683 if env:
1684 full_env.update(env)
1686 cdk_path = self._cdk_path
1687 if cdk_path is None:
1688 cdk_path = self._find_cdk()
1689 self._cdk_path = cdk_path
1690 cdk_cmd = [cdk_path, *command]
1692 if self._cdk_cancel_event.is_set():
1693 raise RuntimeError("CDK operation cancelled before process start")
1694 popen_kwargs: dict[str, Any] = {
1695 "cwd": self.project_root,
1696 "stdout": subprocess.PIPE if capture_output else None,
1697 "stderr": subprocess.PIPE if capture_output else None,
1698 "text": True,
1699 "env": full_env,
1700 "start_new_session": os.name == "posix",
1701 }
1702 if os.name == "nt":
1703 popen_kwargs["creationflags"] = getattr(
1704 subprocess,
1705 "CREATE_NEW_PROCESS_GROUP",
1706 0,
1707 )
1708 process = subprocess.Popen( # nosemgrep: dangerous-subprocess-use-audit - static CDK argv, no shell
1709 cdk_cmd,
1710 **popen_kwargs,
1711 )
1712 with self._active_cdk_lock:
1713 self._active_cdk_processes[process.pid] = process
1714 if self._cdk_cancel_event.is_set():
1715 self._terminate_cdk_process(process)
1716 with self._active_cdk_lock:
1717 self._active_cdk_processes.pop(process.pid, None)
1718 raise RuntimeError("CDK operation cancelled during process start")
1720 try:
1721 stdout, stderr = process.communicate(timeout=timeout)
1722 except subprocess.TimeoutExpired as exc:
1723 self._terminate_cdk_process(process)
1724 logger.warning(
1725 "cdk command timed out after %ss: %s",
1726 timeout,
1727 " ".join(cdk_cmd),
1728 )
1729 raise subprocess.TimeoutExpired(
1730 cdk_cmd,
1731 exc.timeout,
1732 output=exc.output,
1733 stderr=exc.stderr,
1734 ) from exc
1735 except BaseException:
1736 self._terminate_cdk_process(process)
1737 raise
1738 finally:
1739 with self._active_cdk_lock:
1740 self._active_cdk_processes.pop(process.pid, None)
1741 return subprocess.CompletedProcess(
1742 cdk_cmd,
1743 process.returncode,
1744 stdout=stdout or "",
1745 stderr=stderr or "",
1746 )
1748 def list_stacks(self) -> list[str]:
1749 """List all available CDK stacks."""
1750 result = self._run_cdk(["list"], capture_output=True)
1751 if result.returncode != 0:
1752 raise RuntimeError(f"Failed to list stacks: {result.stderr}")
1753 return [s.strip() for s in result.stdout.strip().split("\n") if s.strip()]
1755 def synth(self, stack_name: str | None = None, quiet: bool = True) -> str:
1756 """Synthesize CloudFormation templates from source-current assets."""
1757 self._ensure_lambda_build()
1758 cmd = ["synth"]
1759 if stack_name:
1760 cmd.append(stack_name)
1761 if quiet:
1762 cmd.append("--quiet")
1764 result = self._run_cdk(cmd, capture_output=True)
1765 if result.returncode != 0:
1766 raise RuntimeError(f"CDK synth failed: {result.stderr}")
1767 return str(result.stdout)
1769 def diff(self, stack_name: str | None = None) -> str:
1770 """Show diff between deployed and source-current local stacks."""
1771 self._ensure_lambda_build()
1772 cmd = ["diff", "--no-color"]
1773 if stack_name:
1774 cmd.append(stack_name)
1776 result = self._run_cdk(cmd, capture_output=True)
1777 # diff returns non-zero if there are differences, which is expected
1778 return str(result.stdout or result.stderr)
1780 def deploy(
1781 self,
1782 stack_name: str | None = None,
1783 require_approval: bool = True,
1784 all_stacks: bool = False,
1785 outputs_file: str | None = None,
1786 parameters: dict[str, str] | None = None,
1787 tags: dict[str, str] | None = None,
1788 progress: str = "events",
1789 output_dir: str | None = None,
1790 exclusively: bool = False,
1791 allow_bootstrap: bool = True,
1792 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
1793 expected_stack_ids: Mapping[str, str | None] | None = None,
1794 prepared_change_sets: PreparedChangeSetAuthority | None = None,
1795 authorize_stack: StackAuthorizationCallback | None = None,
1796 strict_deployment_token: str | None = None,
1797 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
1798 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
1799 ) -> bool:
1800 """Deploy CDK stacks.
1802 Args:
1803 stack_name: Name of the stack to deploy
1804 require_approval: Whether to require approval for changes
1805 all_stacks: Deploy all stacks
1806 outputs_file: File to write outputs to
1807 parameters: CDK parameters
1808 tags: Tags to apply to stacks
1809 progress: Progress display type
1810 output_dir: Custom CDK output directory (for parallel deployments)
1811 exclusively: Pass ``--exclusively`` to CDK so only the named
1812 stack is evaluated, not its transitive dependencies. Used by
1813 ``deploy_orchestrated`` once earlier phases have already
1814 deployed the globals — re-synthesizing them every phase
1815 forces custom resources (notably KubectlApplyManifests)
1816 to re-run each time, adding minutes per phase for no
1817 actual change.
1818 """
1819 # Synchronize canonical checked-in copies first, then source-check and
1820 # atomically publish only stale generated assets. A deploy must never
1821 # destructively rebuild a fresh tree while another CDK process may be
1822 # fingerprinting it.
1823 self._sync_lambda_sources()
1824 self._ensure_lambda_build()
1826 strict_deployment = (
1827 strict_deployment_token is not None or on_change_set_prepared is not None
1828 )
1829 expected_stack_id: str | None = None
1830 prepared_change_set_records: Mapping[str, Mapping[str, str]] = {}
1831 change_set_name: str | None = None
1832 if strict_deployment:
1833 if not stack_name or all_stacks:
1834 raise RuntimeError("Strict deployment requires exactly one named stack")
1835 if not strict_deployment_token or on_change_set_prepared is None:
1836 raise RuntimeError(
1837 "Strict deployment requires both a run token and a prepared-change-set callback"
1838 )
1839 if allow_bootstrap:
1840 raise RuntimeError("Strict deployment cannot auto-bootstrap a Region")
1841 if authorize_stack is None:
1842 raise RuntimeError("Strict deployment requires an exact stack authorizer")
1843 if expected_stack_ids is None or stack_name not in expected_stack_ids:
1844 raise RuntimeError(
1845 f"Strict deployment lacks authoritative target state for {stack_name}"
1846 )
1847 expected_stack_id = expected_stack_ids[stack_name]
1848 if prepared_change_sets is None or stack_name not in prepared_change_sets:
1849 raise RuntimeError(
1850 f"Strict deployment lacks prepared change-set history for {stack_name}"
1851 )
1852 prepared_change_set_records = prepared_change_sets[stack_name]
1853 change_set_name = self._strict_change_set_name(
1854 stack_name,
1855 strict_deployment_token,
1856 )
1858 # Validate bootstrap identity before any AWS mutation. In strict mode
1859 # this also revalidates the expected stack ARN (or authoritative
1860 # absence) before image mirroring or change-set preparation.
1861 if stack_name:
1862 region = self._get_deploy_region(stack_name)
1863 if not region:
1864 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
1865 if allow_bootstrap:
1866 if not self.ensure_bootstrapped(region):
1867 raise RuntimeError(
1868 f"Region {region} could not be bootstrapped. "
1869 "Run 'gco stacks bootstrap --region "
1870 f"{region}' manually to diagnose."
1871 )
1872 else:
1873 expected_bootstrap = (bootstrap_stacks or {}).get(region)
1874 if expected_bootstrap is None:
1875 raise RuntimeError(
1876 f"Strict deployment lacks a checkpointed CDKToolkit identity for {region}"
1877 )
1878 self._validate_bootstrap_stack(region, expected_bootstrap)
1879 if strict_deployment:
1880 target = self._describe_stack_target(
1881 stack_name,
1882 expected_stack_id=expected_stack_id,
1883 require_expected_identity=True,
1884 )
1885 if expected_stack_id is not None and target is None:
1886 raise RuntimeError(
1887 f"Checkpointed stack {expected_stack_id} is absent; refusing recreation"
1888 )
1889 assert change_set_name is not None
1890 self._preflight_strict_change_set(
1891 stack_name=stack_name,
1892 change_set_name=change_set_name,
1893 expected_stack_id=expected_stack_id,
1894 prepared_change_sets=prepared_change_set_records,
1895 )
1897 # Name-based stuck-stack recovery is intentionally disabled for strict
1898 # deployments. A prepared change set must establish CREATE-vs-UPDATE
1899 # authority without deleting or adopting anything by name.
1900 if stack_name and not strict_deployment:
1901 self._check_and_fix_stuck_stack(
1902 stack_name,
1903 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
1904 authorize_stack=authorize_stack,
1905 strict_ownership=not allow_bootstrap,
1906 )
1908 # Ensure container runtime is available for building images
1909 runtime = _detect_container_runtime()
1910 if not runtime:
1911 from cli._container_runtime import container_runtime_error_message
1913 raise RuntimeError(container_runtime_error_message())
1915 # Mirror third-party images into ECR only after strict bootstrap and
1916 # target checks. Repository creation acknowledgements are persisted
1917 # synchronously by the live-validation callback before any image copy.
1918 self._mirror_images_if_enabled(
1919 stack_name=stack_name,
1920 all_stacks=all_stacks,
1921 repository_tags=tags,
1922 on_repository_created=on_ecr_repository_created,
1923 )
1925 cmd = ["deploy"]
1927 if all_stacks:
1928 cmd.append("--all")
1929 elif stack_name:
1930 cmd.append(stack_name)
1932 # --exclusively tells CDK to deploy *only* the named stack, not its
1933 # transitive dependencies. deploy_orchestrated sets this once the
1934 # earlier phases (global, api-gateway) are already in place so that
1935 # the regional and monitoring phases don't re-synthesize and
1936 # re-evaluate globals on every pass.
1937 if exclusively and stack_name and not all_stacks:
1938 cmd.append("--exclusively")
1940 if strict_deployment:
1941 assert change_set_name is not None
1942 cmd.extend(
1943 [
1944 "--method",
1945 "prepare-change-set",
1946 "--change-set-name",
1947 change_set_name,
1948 "--context",
1949 f"{_LIVE_VALIDATION_PROVIDER_LOG_CONTEXT}=true",
1950 ]
1951 )
1953 if not require_approval:
1954 cmd.extend(["--require-approval", "never"])
1956 if outputs_file:
1957 cmd.extend(["--outputs-file", outputs_file])
1959 if parameters:
1960 for key, value in parameters.items():
1961 cmd.extend(["--parameters", f"{key}={value}"])
1963 if tags:
1964 for key, value in tags.items():
1965 cmd.extend(["--tags", f"{key}={value}"])
1967 cmd.extend(["--progress", progress])
1969 # Use custom output directory for parallel deployments
1970 if output_dir:
1971 cmd.extend(["--output", output_dir])
1973 # Set CDK_DOCKER env var if not already set
1974 env = {"CDK_DOCKER": runtime} if not os.environ.get("CDK_DOCKER") else None
1976 # Per-stack wall-clock cap so a wedged ``cdk deploy`` (e.g. an
1977 # IAM eventual-consistency wait that never completes) can't block
1978 # the orchestrator forever. Default 60 minutes — long enough for
1979 # a fresh EKS cluster cold start. Override via
1980 # GCO_CDK_DEPLOY_TIMEOUT_SECONDS.
1981 timeout_s = float(os.environ.get("GCO_CDK_DEPLOY_TIMEOUT_SECONDS", "3600"))
1983 # Timestamp (UTC) marking the start of this deploy attempt. The failure
1984 # reconciliation below uses it to tell a *fresh* CloudFormation
1985 # completion (cdk's client-side polling gave up just after CFN finished
1986 # — a real success) apart from a *stale* terminal state left by a
1987 # previous deploy (cdk failed before touching CloudFormation — a real
1988 # failure that must not be masked).
1989 deploy_start = datetime.now(UTC)
1991 try:
1992 result = self._run_cdk(cmd, env=env, timeout=timeout_s)
1993 success = result.returncode == 0
1994 except subprocess.TimeoutExpired:
1995 print(
1996 f" cdk deploy timed out after {timeout_s}s for "
1997 f"{stack_name or 'all stacks'}; verifying CloudFormation state..."
1998 )
1999 success = False
2001 if self._cdk_cancel_event.is_set():
2002 raise RuntimeError("CDK deployment cancelled before AWS-side reconciliation")
2004 if strict_deployment:
2005 assert stack_name is not None
2006 assert change_set_name is not None
2007 assert on_change_set_prepared is not None
2008 try:
2009 success = self._execute_prepared_change_set(
2010 stack_name=stack_name,
2011 change_set_name=change_set_name,
2012 expected_stack_id=expected_stack_id,
2013 expected_tags=tags,
2014 prepared_change_sets=prepared_change_set_records,
2015 preparation_succeeded=success,
2016 authorize_stack=authorize_stack,
2017 on_change_set_prepared=on_change_set_prepared,
2018 allow_noop=success,
2019 timeout=timeout_s,
2020 )
2021 except Exception:
2022 self._diagnose_deploy_failure(stack_name)
2023 raise
2024 if not success:
2025 self._diagnose_deploy_failure(stack_name)
2027 if success and "analytics" in stack_name:
2028 api_gateway_stack = f"{self.config.project_name}-api-gateway"
2029 print(f" Updating {api_gateway_stack} with analytics routes...")
2030 success = self.deploy(
2031 stack_name=api_gateway_stack,
2032 require_approval=require_approval,
2033 outputs_file=outputs_file,
2034 parameters=parameters,
2035 tags=tags,
2036 progress=progress,
2037 exclusively=True,
2038 allow_bootstrap=allow_bootstrap,
2039 bootstrap_stacks=bootstrap_stacks,
2040 expected_stack_ids=expected_stack_ids,
2041 prepared_change_sets=prepared_change_sets,
2042 authorize_stack=authorize_stack,
2043 strict_deployment_token=(f"{strict_deployment_token}-analytics-routes"),
2044 on_change_set_prepared=on_change_set_prepared,
2045 on_ecr_repository_created=on_ecr_repository_created,
2046 )
2047 return success
2049 # Reconcile a cdk failure/timeout against CloudFormation. cdk's
2050 # client-side polling can give up (a transient ``read EADDRNOTAVAIL``
2051 # socket error, or our wall-clock timeout) while CloudFormation keeps
2052 # working server-side, so a non-zero exit does not always mean the
2053 # deploy failed. The trick is to reconcile without masking a *real*
2054 # failure by mistaking a stale terminal state for a fresh success.
2055 if stack_name and not all_stacks and not success:
2056 cfn_status = self._get_stack_status(stack_name)
2057 if cfn_status is not None and cfn_status.endswith("_IN_PROGRESS"):
2058 # CloudFormation is still mid-operation — observing that is
2059 # itself proof it ran an operation for this attempt. Wait for it
2060 # to settle and accept a terminal COMPLETE as a genuine success.
2061 print(
2062 f" cdk exited non-zero but {stack_name} is {cfn_status} in "
2063 "CloudFormation; waiting for the operation to settle..."
2064 )
2065 settled_status = self._wait_for_stack_settle(stack_name)
2066 if settled_status in ("CREATE_COMPLETE", "UPDATE_COMPLETE"):
2067 print(
2068 f" cdk reported a non-zero exit but {stack_name} settled "
2069 f"to {settled_status} in CloudFormation — treating as "
2070 "success."
2071 )
2072 success = True
2073 elif cfn_status in ("CREATE_COMPLETE", "UPDATE_COMPLETE"):
2074 # The stack is already terminal and CloudFormation is not
2075 # mid-flight. Two very different situations look identical on
2076 # status alone; only the stack's last-operation time tells them
2077 # apart:
2078 # * cdk's polling gave up just *after* CloudFormation finished
2079 # this attempt's operation — a genuine success whose
2080 # last-update time is newer than when we started.
2081 # * cdk failed *before* it ever touched CloudFormation (a
2082 # synth error, a cloud-assembly schema mismatch, an
2083 # asset/image build failure); the stack is merely sitting in
2084 # a *previous* deploy's COMPLETE state, whose last-update
2085 # time predates this attempt. Masking this is the
2086 # false-success bug this guards against.
2087 last_op = self._get_stack_last_update_time(stack_name)
2088 if last_op is not None and last_op >= deploy_start:
2089 print(
2090 f" cdk reported a non-zero exit but {stack_name} shows a "
2091 f"fresh {cfn_status} in CloudFormation — treating as "
2092 "success."
2093 )
2094 success = True
2095 else:
2096 print(
2097 f" cdk failed and {stack_name} is {cfn_status}, but no "
2098 "new CloudFormation operation ran for this attempt — cdk "
2099 "failed before touching CloudFormation. Treating as a "
2100 "failed deploy."
2101 )
2103 # Conversely, when cdk reports success, confirm CloudFormation actually
2104 # landed in a terminal success state. A zero cdk exit can still mask a
2105 # stack that silently rolled back (e.g. UPDATE_ROLLBACK_COMPLETE) or is
2106 # otherwise not in a healthy COMPLETE state — verifying the AWS-side
2107 # truth keeps deploy() from reporting a rolled-back stack as deployed.
2108 # A None status (lookup failed / transient) leaves cdk's verdict intact;
2109 # we only override on a *known* non-success state. No-op deploys stay in
2110 # CREATE_COMPLETE/UPDATE_COMPLETE, so this never false-fails them — we
2111 # deliberately don't require LastUpdatedTime to advance.
2112 if success and stack_name and not all_stacks:
2113 cfn_status = self._get_stack_status(stack_name)
2114 if cfn_status is not None and cfn_status not in (
2115 "CREATE_COMPLETE",
2116 "UPDATE_COMPLETE",
2117 ):
2118 print(
2119 f" cdk reported success but {stack_name} is in {cfn_status} "
2120 f"in CloudFormation — treating as a failed deploy."
2121 )
2122 success = False
2124 if not success and stack_name:
2125 self._diagnose_deploy_failure(stack_name)
2127 # After deploying gco-analytics, automatically redeploy
2128 # gco-api-gateway to wire in the /studio/* routes (the API gateway
2129 # imports the Cognito pool ARN and presigned-URL Lambda ARN from
2130 # the analytics stack).
2131 if success and stack_name and "analytics" in stack_name and not all_stacks:
2132 api_gateway_stack = f"{self.config.project_name}-api-gateway"
2133 print(f" Updating {api_gateway_stack} with analytics routes...")
2134 success = self.deploy(
2135 stack_name=api_gateway_stack,
2136 require_approval=require_approval,
2137 outputs_file=outputs_file,
2138 parameters=parameters,
2139 tags=tags,
2140 progress=progress,
2141 exclusively=True,
2142 allow_bootstrap=allow_bootstrap,
2143 bootstrap_stacks=bootstrap_stacks,
2144 expected_stack_ids=expected_stack_ids,
2145 authorize_stack=authorize_stack,
2146 on_ecr_repository_created=on_ecr_repository_created,
2147 )
2149 return success
2151 def destroy(
2152 self,
2153 stack_name: str | None = None,
2154 all_stacks: bool = False,
2155 force: bool = False,
2156 output_dir: str | None = None,
2157 expected_stack_id: str | None = None,
2158 expected_stack_ids: Mapping[str, str | None] | None = None,
2159 prepared_change_sets: PreparedChangeSetAuthority | None = None,
2160 authorize_stack: StackAuthorizationCallback | None = None,
2161 allow_bootstrap: bool = True,
2162 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
2163 strict_deployment_token: str | None = None,
2164 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
2165 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
2166 ) -> bool:
2167 """Destroy stacks while restoring any temporary config mutation exactly."""
2168 config_path: Path | None = None
2169 if stack_name and not all_stacks and "analytics" in stack_name:
2170 config_path = _find_cdk_json()
2171 if config_path is None:
2172 raise RuntimeError("cdk.json not found before analytics destroy")
2174 # Analytics teardown may temporarily enable a disabled stack in the CDK
2175 # app. Hold the shared configuration transaction through restore so a
2176 # concurrent CLI/MCP edit cannot be silently overwritten by the exact-
2177 # bytes rollback in ``finally``.
2178 lock_context = (
2179 _config_mutation_lock(config_path) if config_path is not None else nullcontext()
2180 )
2181 with lock_context:
2182 original_bytes: bytes | None = None
2183 original_mode: int | None = None
2184 if config_path is not None:
2185 original_bytes = config_path.read_bytes()
2186 original_mode = stat.S_IMODE(config_path.stat().st_mode)
2187 try:
2188 return self._destroy(
2189 stack_name=stack_name,
2190 all_stacks=all_stacks,
2191 force=force,
2192 output_dir=output_dir,
2193 expected_stack_id=expected_stack_id,
2194 expected_stack_ids=expected_stack_ids,
2195 prepared_change_sets=prepared_change_sets,
2196 authorize_stack=authorize_stack,
2197 allow_bootstrap=allow_bootstrap,
2198 bootstrap_stacks=bootstrap_stacks,
2199 strict_deployment_token=strict_deployment_token,
2200 on_change_set_prepared=on_change_set_prepared,
2201 on_ecr_repository_created=on_ecr_repository_created,
2202 )
2203 finally:
2204 if config_path is not None and original_bytes is not None:
2205 _atomic_write_bytes(config_path, original_bytes, mode=original_mode)
2207 def _destroy(
2208 self,
2209 stack_name: str | None = None,
2210 all_stacks: bool = False,
2211 force: bool = False,
2212 output_dir: str | None = None,
2213 expected_stack_id: str | None = None,
2214 expected_stack_ids: Mapping[str, str | None] | None = None,
2215 prepared_change_sets: PreparedChangeSetAuthority | None = None,
2216 authorize_stack: StackAuthorizationCallback | None = None,
2217 allow_bootstrap: bool = True,
2218 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
2219 strict_deployment_token: str | None = None,
2220 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
2221 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
2222 ) -> bool:
2223 """Destroy CDK stacks.
2225 If the target stack exists in CloudFormation but isn't in the CDK
2226 app (e.g. because a toggle was disabled), temporarily enables the
2227 toggle so CDK can synthesize and destroy the stack properly. This
2228 ensures custom resource cleanup handlers (like the analytics
2229 cleanup Lambda) fire during deletion.
2231 Args:
2232 stack_name: Name of the stack to destroy
2233 all_stacks: Destroy all stacks
2234 force: Skip confirmation prompts
2235 output_dir: Custom CDK output directory (for parallel deployments)
2236 """
2237 if all_stacks and (expected_stack_id is not None or expected_stack_ids is not None):
2238 raise RuntimeError("Identity-fenced teardown cannot use all_stacks=True")
2239 if stack_name is None and (expected_stack_id is not None or expected_stack_ids is not None):
2240 raise RuntimeError("Identity-fenced teardown requires exactly one named stack")
2242 strict_identity = expected_stack_id is not None or expected_stack_ids is not None
2243 if stack_name is not None and expected_stack_ids is not None:
2244 if stack_name not in expected_stack_ids:
2245 raise RuntimeError(
2246 f"Strict teardown lacks authoritative target state for {stack_name}"
2247 )
2248 mapped_stack_id = expected_stack_ids[stack_name]
2249 if expected_stack_id is not None and expected_stack_id != mapped_stack_id:
2250 raise RuntimeError(f"Conflicting expected stack identities for {stack_name}")
2251 expected_stack_id = mapped_stack_id
2252 if strict_identity and authorize_stack is None:
2253 raise RuntimeError("Identity-fenced teardown requires an exact stack authorizer")
2255 # Image-registry pre-destroy guards. Only fires for the global
2256 # stack (where the registry lives) and only when the operator
2257 # has explicitly chosen ``removal_policy: "destroy"``. The
2258 # default ``retain`` posture is a no-op here. See
2259 # ``_image_registry_destroy_preflight`` for the exact rules.
2260 if (
2261 stack_name is not None
2262 and stack_name.endswith("-global")
2263 and not all_stacks
2264 and not self._image_registry_destroy_preflight(force=force)
2265 ):
2266 return False
2268 # Strict callers never enter CDK's name-based destroy or toggle-based
2269 # recovery paths. Analytics may first require one strict prepared
2270 # change set on the exact API stack to remove cross-stack imports.
2271 if strict_identity and stack_name and not all_stacks:
2272 if self._cdk_cancel_event.is_set():
2273 raise RuntimeError(f"Strict teardown cancelled before deleting {stack_name}")
2274 if "analytics" in stack_name:
2275 if expected_stack_ids is None:
2276 raise RuntimeError(
2277 "Identity-fenced analytics teardown requires the complete expected "
2278 "stack identity map"
2279 )
2280 safe_to_destroy = self._remove_api_gateway_analytics_dependency(
2281 allow_bootstrap=allow_bootstrap,
2282 bootstrap_stacks=bootstrap_stacks,
2283 expected_stack_ids=expected_stack_ids,
2284 prepared_change_sets=prepared_change_sets,
2285 authorize_stack=authorize_stack,
2286 strict_deployment_token=(
2287 f"{strict_deployment_token}-drop-analytics-routes"
2288 if strict_deployment_token is not None
2289 else None
2290 ),
2291 on_change_set_prepared=on_change_set_prepared,
2292 on_ecr_repository_created=on_ecr_repository_created,
2293 )
2294 if not safe_to_destroy:
2295 return False
2296 return self._cloudformation_delete_stack(
2297 stack_name,
2298 expected_stack_id=expected_stack_id,
2299 authorize_stack=authorize_stack,
2300 require_expected_identity=True,
2301 )
2303 # A regional API bridge disappears from the CDK app when its Region is
2304 # removed from configuration. Only this exact project-scoped shape with
2305 # an SDK-known CloudFormation Region may bypass CDK; configured bridges,
2306 # arbitrary suffixes, and every other stack keep the normal CDK path.
2307 if stack_name and not all_stacks:
2308 orphan_region = self._get_orphan_regional_api_region(stack_name)
2309 if orphan_region is not None:
2310 if not self._stack_exists_in_cloudformation(stack_name):
2311 return True
2312 print(
2313 f" {stack_name} is absent from the configured CDK app; "
2314 f"deleting it directly in {orphan_region}..."
2315 )
2316 return self._cloudformation_delete_stack(
2317 stack_name,
2318 expected_stack_id=expected_stack_id,
2319 authorize_stack=authorize_stack,
2320 )
2322 # If destroying a specific stack that exists in CloudFormation but
2323 # might not be in the CDK app, temporarily enable its toggle.
2324 toggle_restored = False
2325 if (
2326 stack_name
2327 and not all_stacks
2328 and "analytics" in stack_name
2329 and self._stack_exists_in_cloudformation(stack_name)
2330 ):
2331 toggle_restored = self._ensure_analytics_enabled_for_destroy()
2333 # The analytics stack exports values (e.g. Cognito pool ARN) that
2334 # gco-api-gateway imports. CloudFormation blocks deletion of stacks
2335 # with consumed exports. To break the dependency, redeploy the API
2336 # gateway with analytics disabled first, then destroy analytics.
2337 if stack_name and not all_stacks and "analytics" in stack_name:
2338 safe_to_destroy = self._remove_api_gateway_analytics_dependency(
2339 allow_bootstrap=allow_bootstrap,
2340 bootstrap_stacks=bootstrap_stacks,
2341 expected_stack_ids=expected_stack_ids,
2342 prepared_change_sets=prepared_change_sets,
2343 authorize_stack=authorize_stack,
2344 strict_deployment_token=strict_deployment_token,
2345 on_change_set_prepared=on_change_set_prepared,
2346 on_ecr_repository_created=on_ecr_repository_created,
2347 )
2348 if not safe_to_destroy:
2349 # Restore analytics toggle before bailing out.
2350 if toggle_restored:
2351 self._restore_analytics_disabled()
2352 project = self.config.project_name
2353 print(
2354 f" Aborting {project}-analytics destroy: {project}-api-gateway "
2355 "still imports analytics exports. Fix the API gateway and retry."
2356 )
2357 return False
2359 # Non-strict callers may still use CDK's name-based path. Strict calls
2360 # returned above after exact-ARN deletion.
2361 cmd = ["destroy"]
2363 if all_stacks:
2364 cmd.append("--all")
2365 elif stack_name:
2366 cmd.append(stack_name)
2367 # --exclusively prevents CDK from cascading the destroy to
2368 # dependent stacks (e.g. destroying gco-analytics should not
2369 # also destroy gco-api-gateway just because it references the
2370 # presigned-URL Lambda ARN).
2371 cmd.append("--exclusively")
2373 if force:
2374 cmd.append("--force")
2376 if output_dir:
2377 cmd.extend(["--output", output_dir])
2379 # Per-stack wall-clock cap so a wedged ``cdk destroy`` (its
2380 # post-delete polling loop hanging after CloudFormation has
2381 # already finished) can't block the orchestrator forever. Default
2382 # 90 minutes: a healthy EKS regional teardown has been observed
2383 # needing ~60 (the VPC Lambda ENI detach alone can serialise for
2384 # 20+ while CloudFormation keeps making progress), and the prior
2385 # 45-minute cap killed the poller mid-delete — the AWS-side
2386 # reconciliation below recovered, but the timeout should mark a
2387 # wedged CDK, not a normal teardown. Override via
2388 # GCO_CDK_DESTROY_TIMEOUT_SECONDS.
2389 timeout_s = float(os.environ.get("GCO_CDK_DESTROY_TIMEOUT_SECONDS", "5400"))
2391 try:
2392 result = self._run_cdk(cmd, timeout=timeout_s)
2393 cdk_succeeded = result.returncode == 0
2394 except subprocess.TimeoutExpired:
2395 # CDK hung. Verify the AWS-side state below — if the stack
2396 # is gone in CloudFormation, the destroy actually succeeded
2397 # and the timeout was just CDK's polling loop wedged.
2398 print(
2399 f" cdk destroy timed out after {timeout_s}s; verifying "
2400 f"CloudFormation state for {stack_name}..."
2401 )
2402 cdk_succeeded = False
2404 if self._cdk_cancel_event.is_set():
2405 raise RuntimeError("CDK teardown cancelled before AWS-side reconciliation")
2407 # Restore the toggle if we changed it
2408 if toggle_restored:
2409 self._restore_analytics_disabled()
2411 # Reconcile against CloudFormation. A local CDK timeout/failure is not
2412 # an AWS failure when the delete operation is still healthy. Once AWS
2413 # reports DELETE_IN_PROGRESS, wait for bounded server-side convergence
2414 # instead of letting the orchestrator advance into dependent stacks.
2415 if stack_name and not all_stacks:
2416 still_present = self._stack_exists_in_cloudformation(stack_name)
2417 if not still_present:
2418 if not cdk_succeeded:
2419 print(
2420 f" cdk reported a non-zero exit but {stack_name} is "
2421 "already deleted in CloudFormation — treating as success."
2422 )
2423 return True
2425 status = self._get_stack_status(stack_name, expected_stack_id)
2426 if status == "DELETE_IN_PROGRESS":
2427 return self._wait_for_stack_delete_convergence(
2428 stack_name,
2429 initial_status=status,
2430 )
2431 if status == "DELETE_FAILED":
2432 self._print_stack_delete_heartbeat(
2433 stack_name,
2434 status,
2435 expected_stack_id,
2436 )
2437 print(f" {stack_name} reached DELETE_FAILED; refusing to continue teardown.")
2438 return False
2440 # A zero CDK exit with a still-present, non-deleting stack is a rare
2441 # client-side false success. Start deletion directly and then use
2442 # the same bounded convergence loop. A non-zero exit in any other
2443 # state means CDK failed before it started a delete operation.
2444 if cdk_succeeded:
2445 return self._cloudformation_delete_stack(stack_name)
2446 print(
2447 f" cdk failed and {stack_name} is still {status or 'in an unknown state'}; "
2448 "no active CloudFormation delete operation was confirmed."
2449 )
2450 return False
2452 return cdk_succeeded
2454 # ------------------------------------------------------------------
2455 # Image registry pre-destroy guards
2456 # ------------------------------------------------------------------
2457 def _read_images_config(self) -> dict[str, Any]:
2458 """Read the ``images`` block from cdk.json with defaults applied.
2460 Mirrors the parser in ``gco/stacks/global_stack.py`` so the CLI
2461 can reason about the same fields without importing the CDK
2462 module (which pulls aws_cdk and the full constructs surface).
2463 Defaults stay aligned with the global-stack parser; any value
2464 that fails validation (e.g. an unexpected ``removal_policy``)
2465 is silently coerced to ``"retain"`` here so the CLI never blocks
2466 on a typo — the actual deploy-time validation is the global
2467 stack's responsibility.
2468 """
2469 import json
2471 cdk_json_path = _find_cdk_json()
2472 if not cdk_json_path:
2473 return {
2474 "removal_policy": "retain",
2475 "empty_on_delete": False,
2476 }
2477 try:
2478 with open(cdk_json_path, encoding="utf-8") as f:
2479 ctx = json.load(f).get("context", {}) or {}
2480 except (OSError, json.JSONDecodeError) as exc:
2481 logger.debug("Failed to read cdk.json for images config: %s", exc)
2482 return {"removal_policy": "retain", "empty_on_delete": False}
2484 raw = ctx.get("images") or {}
2485 removal_policy = str(raw.get("removal_policy", "retain")).strip().lower()
2486 if removal_policy not in ("retain", "destroy"):
2487 removal_policy = "retain"
2488 return {
2489 "removal_policy": removal_policy,
2490 "empty_on_delete": bool(raw.get("empty_on_delete", False)),
2491 }
2493 def _build_image_registry_inventory(self) -> dict[str, Any]:
2494 """Aggregate repo / tag / size / reference counts for the registry.
2496 Returns a dict shape suitable for printing to the operator. Best
2497 effort: a missing ImageManager dependency or an AWS error
2498 produces a partially-populated dict rather than raising.
2499 """
2500 inventory: dict[str, Any] = {
2501 "repo_count": 0,
2502 "tag_count": 0,
2503 "total_bytes": 0,
2504 "endpoint_refs": 0,
2505 "job_refs": 0,
2506 }
2507 try:
2508 from cli.images import ImageManager
2509 except Exception as exc: # noqa: BLE001
2510 logger.debug("ImageManager import failed during preflight: %s", exc)
2511 return inventory
2513 try:
2514 manager = ImageManager(config=self.config)
2515 repos = manager.list_repos()
2516 inventory["repo_count"] = len(repos)
2517 # Repos this deployment owns live under ``<project_name>/`` (#139).
2518 repo_prefix = f"{self.config.project_name}/"
2519 for repo in repos:
2520 repo_name = repo.get("name", "")
2521 if not repo_name.startswith(repo_prefix):
2522 continue
2523 short = repo_name.removeprefix(repo_prefix)
2524 try:
2525 tags = manager.list_tags(short)
2526 except Exception as exc: # noqa: BLE001
2527 logger.debug("list_tags failed for %s: %s", repo_name, exc)
2528 continue
2529 inventory["tag_count"] += len(tags)
2530 for row in tags:
2531 size = row.get("size_bytes")
2532 if isinstance(size, int):
2533 inventory["total_bytes"] += size
2534 try:
2535 inventory["endpoint_refs"] = len(manager._collect_inference_image_refs())
2536 except Exception as exc: # noqa: BLE001
2537 logger.debug("inference ref collection failed: %s", exc)
2538 try:
2539 inventory["job_refs"] = len(manager._collect_recent_job_image_refs())
2540 except Exception as exc: # noqa: BLE001
2541 logger.debug("job ref collection failed: %s", exc)
2542 except Exception as exc: # noqa: BLE001
2543 logger.debug("Image registry inventory failed: %s", exc)
2544 return inventory
2546 def _image_registry_destroy_preflight(self, *, force: bool) -> bool:
2547 """Validate the image-registry destroy posture before invoking CFN.
2549 Two rules:
2551 1. ``removal_policy: "destroy"`` AND ``empty_on_delete: false``
2552 → refuse with the literal helpful-error message pointing
2553 the operator at ``gco images cleanup --all`` or at flipping
2554 ``empty_on_delete: true``.
2556 2. ``removal_policy: "destroy"`` AND ``empty_on_delete: true``
2557 → print the inventory summary first. On a TTY the operator
2558 is also prompted for confirmation; non-TTY runs proceed
2559 (the operator presumably passed ``-y`` or is automating).
2561 Returns True when the destroy may proceed, False when it has
2562 been refused or declined.
2563 """
2564 cfg = self._read_images_config()
2565 if cfg["removal_policy"] != "destroy":
2566 return True
2568 if not cfg["empty_on_delete"]:
2569 print(
2570 f"Repos under {self.config.project_name}/* are not empty and "
2571 "empty_on_delete is false. Run 'gco images cleanup --all' "
2572 "first, or set images.empty_on_delete: true in cdk.json."
2573 )
2574 return False
2576 inventory = self._build_image_registry_inventory()
2577 gib = inventory["total_bytes"] / (1024**3) if inventory["total_bytes"] else 0.0
2578 interactive_echo("Image registry inventory before destroy:")
2579 interactive_echo(f" repos: {inventory['repo_count']}")
2580 interactive_echo(f" tags: {inventory['tag_count']}")
2581 interactive_echo(f" total size: {gib:.2f} GiB")
2582 interactive_echo(f" referencing endpoints: {inventory['endpoint_refs']}")
2583 interactive_echo(f" recent job refs: {inventory['job_refs']}")
2585 # Already confirmed via -y, or non-interactive — proceed.
2586 if force or not sys.stdin.isatty():
2587 return True
2589 try:
2590 if confirm(
2591 f"Destroy {self.config.project_name}-global and delete every "
2592 f"{self.config.project_name}/* repo?",
2593 default=False,
2594 ):
2595 return True
2596 except Abort:
2597 interactive_echo("Aborted.")
2598 return False
2599 interactive_echo("Aborted.")
2600 return False
2602 @staticmethod
2603 def _stack_missing(exc: ClientError) -> bool:
2604 error = exc.response.get("Error", {})
2605 return bool(
2606 error.get("Code") == "ValidationError"
2607 and "does not exist" in str(error.get("Message", "")).lower()
2608 )
2610 @staticmethod
2611 def _change_set_missing(exc: ClientError) -> bool:
2612 """Return whether CloudFormation authoritatively reports an absent change set."""
2613 return bool(exc.response.get("Error", {}).get("Code") == "ChangeSetNotFound")
2615 def _describe_stack_target(
2616 self,
2617 stack_name: str,
2618 *,
2619 expected_stack_id: str | None = None,
2620 require_expected_identity: bool = False,
2621 ) -> tuple[str, Any, dict[str, Any]] | None:
2622 """Resolve live/absent/tombstone/replacement state without name adoption."""
2623 import boto3
2625 region = self._get_destroy_region(stack_name)
2626 cfn = boto3.client("cloudformation", region_name=region)
2628 def describe(identifier: str) -> dict[str, Any] | None:
2629 try:
2630 response = cfn.describe_stacks(StackName=identifier)
2631 except ClientError as exc:
2632 if self._stack_missing(exc):
2633 return None
2634 raise
2635 stacks = response.get("Stacks", [])
2636 if len(stacks) != 1:
2637 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2638 stack = stacks[0]
2639 if not isinstance(stack, dict):
2640 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2641 stack_id = str(stack.get("StackId") or "")
2642 if stack.get("StackName") != stack_name or not stack_id:
2643 raise RuntimeError(f"CloudFormation returned an invalid identity for {stack_name}")
2644 return stack
2646 exact = describe(expected_stack_id) if expected_stack_id else None
2647 if exact is not None and str(exact.get("StackStatus") or "") != "DELETE_COMPLETE":
2648 if str(exact.get("StackId") or "") != expected_stack_id:
2649 raise RuntimeError(f"Stack identity changed for {region}:{stack_name}")
2650 return region, cfn, exact
2652 by_name = describe(stack_name)
2653 if by_name is None or str(by_name.get("StackStatus") or "") == "DELETE_COMPLETE":
2654 return None
2655 actual_id = str(by_name.get("StackId") or "")
2656 if expected_stack_id is not None and actual_id != expected_stack_id:
2657 raise RuntimeError(
2658 f"Checkpointed stack {expected_stack_id} is absent or deleted but same-name "
2659 f"replacement {actual_id} exists; refusing adoption"
2660 )
2661 if expected_stack_id is None and require_expected_identity:
2662 raise RuntimeError(
2663 f"Refusing name-authorized access to uncheckpointed stack "
2664 f"{region}:{stack_name} ({actual_id})"
2665 )
2666 return region, cfn, by_name
2668 def _stack_exists_in_cloudformation(
2669 self,
2670 stack_name: str,
2671 expected_stack_id: str | None = None,
2672 *,
2673 require_expected_identity: bool = False,
2674 ) -> bool:
2675 """Return whether the exact live target exists, rejecting replacements."""
2676 target = self._describe_stack_target(
2677 stack_name,
2678 expected_stack_id=expected_stack_id,
2679 require_expected_identity=require_expected_identity,
2680 )
2681 return target is not None
2683 def _get_stack_status(
2684 self,
2685 stack_name: str,
2686 stack_identifier: str | None = None,
2687 ) -> str | None:
2688 """Return the live CloudFormation status of ``stack_name`` or None.
2690 Used by ``deploy()`` to reconcile against AWS-side state when ``cdk
2691 deploy`` returns a non-zero exit code or times out — if the stack
2692 actually finished CREATE_COMPLETE or UPDATE_COMPLETE on the AWS
2693 side, the deploy succeeded regardless of what cdk reported.
2694 Returns None when the stack does not exist or the lookup itself
2695 fails (network blip, perms, etc.) so callers can treat the
2696 unknown case as 'cdk's verdict stands'.
2697 """
2698 import boto3
2700 try:
2701 region = self._get_destroy_region(stack_name)
2702 cfn = boto3.client("cloudformation", region_name=region)
2703 resp = cfn.describe_stacks(StackName=stack_identifier or stack_name)
2704 return str(resp["Stacks"][0]["StackStatus"])
2705 except Exception:
2706 return None
2708 def _get_stack_last_update_time(self, stack_name: str) -> datetime | None:
2709 """Return the UTC time of ``stack_name``'s most recent CloudFormation
2710 operation, or None if the stack is absent or the lookup fails.
2712 Uses ``LastUpdatedTime`` when the stack has been updated at least once,
2713 falling back to ``CreationTime`` for a stack that has only ever been
2714 created. ``deploy()`` compares this against the moment the deploy
2715 attempt started to decide whether a cdk failure/timeout that leaves the
2716 stack ``*_COMPLETE`` reflects a *fresh* operation (cdk's polling merely
2717 gave up early — success) or a *stale* one left by a previous deploy
2718 (cdk failed before touching CloudFormation — a real failure). A None
2719 return keeps the conservative 'cdk's failure stands' verdict.
2720 """
2721 import boto3
2723 try:
2724 region = self._get_destroy_region(stack_name)
2725 cfn = boto3.client("cloudformation", region_name=region)
2726 resp = cfn.describe_stacks(StackName=stack_name)
2727 stack = resp["Stacks"][0]
2728 last_op = stack.get("LastUpdatedTime") or stack.get("CreationTime")
2729 return last_op if isinstance(last_op, datetime) else None
2730 except Exception:
2731 return None
2733 def _wait_for_stack_settle(
2734 self,
2735 stack_name: str,
2736 timeout: float | None = None,
2737 stack_identifier: str | None = None,
2738 ) -> str | None:
2739 """Poll CloudFormation until a stack settles, retrying transient unknown reads."""
2740 import time
2742 if timeout is None:
2743 timeout = float(os.environ.get("GCO_CDK_SETTLE_TIMEOUT_SECONDS", "1200"))
2744 deadline = time.monotonic() + timeout
2745 unknown_started: float | None = None
2746 status = self._get_stack_status(stack_name, stack_identifier)
2747 while True:
2748 now = time.monotonic()
2749 if self._cdk_cancel_event.is_set():
2750 return status
2751 if status is None:
2752 if unknown_started is None:
2753 unknown_started = now
2754 unknown_deadline = min(
2755 deadline,
2756 unknown_started + _CLOUDFORMATION_SETTLE_UNKNOWN_TIMEOUT_SECONDS,
2757 )
2758 if now >= unknown_deadline:
2759 return None
2760 time.sleep(
2761 min(
2762 _CLOUDFORMATION_SETTLE_UNKNOWN_POLL_SECONDS,
2763 unknown_deadline - now,
2764 )
2765 )
2766 status = self._get_stack_status(stack_name, stack_identifier)
2767 continue
2768 unknown_started = None
2769 if not status.endswith("_IN_PROGRESS") or now >= deadline:
2770 return status
2771 time.sleep(min(15.0, deadline - now))
2772 status = self._get_stack_status(stack_name, stack_identifier)
2774 def _get_latest_stack_event(
2775 self,
2776 stack_name: str,
2777 stack_identifier: str | None = None,
2778 ) -> dict[str, Any] | None:
2779 """Return the newest CloudFormation event for delete heartbeats."""
2780 import boto3
2782 try:
2783 region = self._get_destroy_region(stack_name)
2784 cfn = boto3.client("cloudformation", region_name=region)
2785 events = cfn.describe_stack_events(StackName=stack_identifier or stack_name).get(
2786 "StackEvents", []
2787 )
2788 return events[0] if events else None
2789 except Exception:
2790 logger.debug("Could not read delete events for %s", stack_name, exc_info=True)
2791 return None
2793 def _print_stack_delete_heartbeat(
2794 self,
2795 stack_name: str,
2796 status: str | None,
2797 stack_identifier: str | None = None,
2798 ) -> None:
2799 """Print the latest AWS-side state while a long delete converges."""
2800 event = self._get_latest_stack_event(stack_name, stack_identifier)
2801 if not event:
2802 print(f" {stack_name}: CloudFormation status {status or 'unknown'}")
2803 return
2805 timestamp = event.get("Timestamp")
2806 timestamp_text = (
2807 timestamp.isoformat() if isinstance(timestamp, datetime) else str(timestamp or "")
2808 )
2809 logical_id = str(event.get("LogicalResourceId") or stack_name)
2810 resource_status = str(event.get("ResourceStatus") or status or "unknown")
2811 reason = " ".join(str(event.get("ResourceStatusReason") or "").split())
2812 if len(reason) > 400:
2813 reason = reason[:397] + "..."
2814 suffix = f" — {reason}" if reason else ""
2815 print(
2816 f" {stack_name}: {status or 'unknown'}; latest event "
2817 f"{timestamp_text} {logical_id} {resource_status}{suffix}"
2818 )
2820 def _wait_for_stack_delete_convergence(
2821 self,
2822 stack_name: str,
2823 *,
2824 timeout: float | None = None,
2825 poll_interval: float = _CLOUDFORMATION_DELETE_POLL_SECONDS,
2826 heartbeat_interval: float = _CLOUDFORMATION_DELETE_HEARTBEAT_SECONDS,
2827 initial_status: str = "DELETE_IN_PROGRESS",
2828 expected_stack_id: str | None = None,
2829 require_expected_identity: bool = False,
2830 ) -> bool:
2831 """Wait for an AWS-side stack delete to finish without trusting CDK polling.
2833 The caller must already have evidence that a delete operation started.
2834 Transient status-read failures are tolerated after that proof, but a
2835 terminal ``DELETE_FAILED`` or the overall deadline fails closed.
2836 """
2837 if timeout is None:
2838 try:
2839 timeout = float(
2840 os.environ.get(
2841 "GCO_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS",
2842 str(_CLOUDFORMATION_DELETE_TIMEOUT_SECONDS),
2843 )
2844 )
2845 except ValueError:
2846 timeout = _CLOUDFORMATION_DELETE_TIMEOUT_SECONDS
2847 if not math.isfinite(timeout) or timeout <= 0:
2848 raise ValueError("CloudFormation delete timeout must be positive and finite")
2849 if (
2850 not math.isfinite(poll_interval)
2851 or not math.isfinite(heartbeat_interval)
2852 or poll_interval <= 0
2853 or heartbeat_interval <= 0
2854 ):
2855 raise ValueError("CloudFormation delete polling intervals must be positive and finite")
2857 deadline = time.monotonic() + timeout
2858 next_heartbeat = time.monotonic()
2859 status: str | None = initial_status
2860 last_printed_status: str | None = None
2862 while True:
2863 if self._cdk_cancel_event.is_set():
2864 logger.warning("CloudFormation delete wait cancelled for %s", stack_name)
2865 return False
2866 try:
2867 if not self._stack_exists_in_cloudformation(
2868 stack_name,
2869 expected_stack_id=expected_stack_id,
2870 require_expected_identity=require_expected_identity,
2871 ):
2872 print(f" {stack_name} is absent from CloudFormation.")
2873 return True
2874 except RuntimeError:
2875 raise
2876 except Exception:
2877 logger.debug(
2878 "CloudFormation presence check failed for %s",
2879 stack_name,
2880 exc_info=True,
2881 )
2883 now = time.monotonic()
2884 if status == "DELETE_COMPLETE":
2885 return True
2886 if status == "DELETE_FAILED":
2887 self._print_stack_delete_heartbeat(
2888 stack_name,
2889 status,
2890 expected_stack_id,
2891 )
2892 return False
2893 if status not in (None, "DELETE_IN_PROGRESS"):
2894 self._print_stack_delete_heartbeat(
2895 stack_name,
2896 status,
2897 expected_stack_id,
2898 )
2899 print(
2900 f" {stack_name} left DELETE_IN_PROGRESS without being deleted; "
2901 "refusing to continue teardown."
2902 )
2903 return False
2904 if now >= deadline:
2905 self._print_stack_delete_heartbeat(
2906 stack_name,
2907 status,
2908 expected_stack_id,
2909 )
2910 print(
2911 f" Timed out after {timeout:.0f}s waiting for {stack_name} "
2912 "to disappear from CloudFormation."
2913 )
2914 return False
2915 if status != last_printed_status or now >= next_heartbeat:
2916 self._print_stack_delete_heartbeat(
2917 stack_name,
2918 status,
2919 expected_stack_id,
2920 )
2921 last_printed_status = status
2922 next_heartbeat = now + heartbeat_interval
2924 time.sleep(min(poll_interval, max(0.0, deadline - now)))
2925 status = self._get_stack_status(stack_name, expected_stack_id)
2927 def _cloudformation_delete_stack(
2928 self,
2929 stack_name: str,
2930 *,
2931 expected_stack_id: str | None = None,
2932 authorize_stack: StackAuthorizationCallback | None = None,
2933 require_expected_identity: bool = False,
2934 ) -> bool:
2935 """Delete an immediately revalidated stack by immutable ARN."""
2936 if self._cdk_cancel_event.is_set():
2937 raise RuntimeError(f"CloudFormation deletion cancelled before {stack_name}")
2938 target = self._describe_stack_target(
2939 stack_name,
2940 expected_stack_id=expected_stack_id,
2941 require_expected_identity=require_expected_identity,
2942 )
2943 if target is None:
2944 return True
2945 region, cfn, stack = target
2946 stack_id = str(stack["StackId"])
2947 status = str(stack.get("StackStatus") or "")
2948 if authorize_stack is not None:
2949 authorize_stack(stack_name, region, stack_id)
2950 if status == "DELETE_IN_PROGRESS":
2951 return self._wait_for_stack_delete_convergence(
2952 stack_name,
2953 initial_status=status,
2954 expected_stack_id=stack_id,
2955 require_expected_identity=require_expected_identity,
2956 )
2957 try:
2958 cfn.delete_stack(StackName=stack_id)
2959 except Exception:
2960 logger.debug("Direct CloudFormation delete failed for %s", stack_id, exc_info=True)
2961 return False
2962 return self._wait_for_stack_delete_convergence(
2963 stack_name,
2964 expected_stack_id=stack_id,
2965 require_expected_identity=require_expected_identity,
2966 )
2968 def _validated_regional_api_region(self, stack_name: str) -> str | None:
2969 """Return an exact project bridge's SDK-known CloudFormation Region."""
2970 bridge_prefix = f"{self.config.project_name}-regional-api-"
2971 if not stack_name.startswith(bridge_prefix):
2972 return None
2974 region = stack_name[len(bridge_prefix) :]
2975 if not region:
2976 return None
2977 try:
2978 return region if region in _known_cloudformation_regions() else None
2979 except Exception:
2980 logger.debug(
2981 "Could not validate regional API bridge Region for %s",
2982 stack_name,
2983 exc_info=True,
2984 )
2985 return None
2987 def _configured_regional_api_regions(
2988 self,
2989 ) -> tuple[frozenset[str], str] | None:
2990 """Read valid root regions and their partition for orphan deletion.
2992 Returning ``None`` means the configuration could not prove anything:
2993 missing, unreadable, malformed, wrong-project, incomplete, empty, and
2994 duplicate Region configurations all fail closed under the same contract
2995 used by :class:`ConfigLoader`.
2996 """
2997 path = self.project_root / "cdk.json"
2998 try:
2999 data = json.loads(path.read_text(encoding="utf-8"))
3000 if not isinstance(data, dict):
3001 return None
3002 context = data.get("context")
3003 if not isinstance(context, dict):
3004 return None
3005 configured_project = context.get("project_name")
3006 if (
3007 not isinstance(configured_project, str)
3008 or not configured_project
3009 or configured_project != self.config.project_name
3010 ):
3011 return None
3012 deployment_regions = context.get("deployment_regions")
3013 if not isinstance(deployment_regions, dict):
3014 return None
3016 known_regions = _known_cloudformation_regions()
3017 for key in ("global", "api_gateway", "monitoring"):
3018 region = deployment_regions.get(key)
3019 if not isinstance(region, str) or region not in known_regions:
3020 return None
3022 try:
3023 regional = validated_regional_deployment_regions(
3024 deployment_regions.get("regional"),
3025 known_regions=known_regions,
3026 )
3027 deployment_partition = validated_deployment_partition(
3028 (
3029 deployment_regions["global"],
3030 deployment_regions["api_gateway"],
3031 deployment_regions["monitoring"],
3032 *regional,
3033 )
3034 )
3035 except RuntimeError, ValueError:
3036 return None
3037 return frozenset(regional), deployment_partition
3038 except OSError, UnicodeError, json.JSONDecodeError, TypeError:
3039 logger.debug(
3040 "Could not read authoritative regional configuration from %s",
3041 path,
3042 exc_info=True,
3043 )
3044 return None
3046 def _get_orphan_regional_api_region(self, stack_name: str) -> str | None:
3047 """Return a bridge Region only when valid root config proves it absent.
3049 This result authorizes bypassing CDK and deleting a stack directly via
3050 CloudFormation. Merely failing a normal configuration lookup can never
3051 be interpreted as proof that the stack is orphaned.
3052 """
3053 region = self._validated_regional_api_region(stack_name)
3054 if region is None:
3055 return None
3056 configured = self._configured_regional_api_regions()
3057 if configured is None:
3058 return None
3059 configured_regions, deployment_partition = configured
3060 if region in configured_regions:
3061 return None
3062 try:
3063 candidate_partition = validated_deployment_partition((region,))
3064 except RuntimeError, ValueError:
3065 return None
3066 if candidate_partition != deployment_partition:
3067 return None
3068 return region
3070 def _get_destroy_region(self, stack_name: str) -> str:
3071 """Determine a configured or cryptographically bounded destroy Region.
3073 Deploy resolution intentionally requires bridge Regions to remain in
3074 ``cdk.json``. A removed bridge may still resolve through the orphan
3075 path, but that path validates the exact project-scoped name, SDK-known
3076 CloudFormation Region, authoritative root configuration, and matching
3077 AWS partition. Reconciliation must reuse that same proof rather than
3078 trusting a bridge-shaped suffix independently.
3079 """
3080 try:
3081 region = self._get_deploy_region(stack_name)
3082 except Exception:
3083 logger.debug(
3084 "Configured deploy Region lookup failed for %s; checking orphan shape",
3085 stack_name,
3086 exc_info=True,
3087 )
3088 else:
3089 if region:
3090 return region
3092 orphan_region = self._get_orphan_regional_api_region(stack_name)
3093 return orphan_region or self.config.api_gateway_region
3095 def _ensure_analytics_enabled_for_destroy(self) -> bool:
3096 """Temporarily enable analytics so CDK includes the stack for destroy."""
3097 try:
3098 current = get_analytics_config()
3099 if not current.get("enabled"):
3100 update_analytics_config({"enabled": True})
3101 return True
3102 except Exception as exc:
3103 logger.debug(
3104 "Failed to enable analytics toggle for destroy: %s",
3105 exc,
3106 exc_info=True,
3107 )
3108 return False
3110 def _restore_analytics_disabled(self) -> None:
3111 """Restore analytics toggle to disabled after destroy."""
3112 try:
3113 update_analytics_config({"enabled": False})
3114 except Exception as exc:
3115 logger.warning(
3116 "Failed to restore analytics toggle to disabled after destroy: %s",
3117 exc,
3118 exc_info=True,
3119 )
3121 def _remove_api_gateway_analytics_dependency(
3122 self,
3123 *,
3124 allow_bootstrap: bool = True,
3125 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
3126 expected_stack_ids: Mapping[str, str | None] | None = None,
3127 prepared_change_sets: PreparedChangeSetAuthority | None = None,
3128 authorize_stack: StackAuthorizationCallback | None = None,
3129 strict_deployment_token: str | None = None,
3130 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
3131 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
3132 ) -> bool:
3133 """Redeploy gco-api-gateway with analytics disabled to drop cross-stack imports.
3135 The analytics stack exports values (Cognito pool ARN, presigned-URL
3136 Lambda ARN) that gco-api-gateway imports for the /studio/* routes.
3137 CloudFormation blocks deletion of stacks with consumed exports. By
3138 disabling analytics and redeploying the API gateway, the /studio/*
3139 routes are removed and the imports are dropped, unblocking the
3140 analytics stack deletion.
3142 Returns:
3143 True if the analytics stack is safe to destroy (either because
3144 no consumer remains or because the redeploy successfully
3145 dropped the imports). False if a consumer of the analytics
3146 exports still exists and the analytics destroy will fail.
3147 """
3148 api_gateway_stack = f"{self.config.project_name}-api-gateway"
3149 analytics_stack = f"{self.config.project_name}-analytics"
3151 strict_identity = expected_stack_ids is not None
3152 if strict_identity:
3153 assert expected_stack_ids is not None
3154 if api_gateway_stack not in expected_stack_ids:
3155 raise RuntimeError(
3156 f"Strict teardown lacks authoritative target state for {api_gateway_stack}"
3157 )
3158 api_gateway_expected_id = expected_stack_ids[api_gateway_stack]
3159 else:
3160 api_gateway_expected_id = None
3162 # Fast path: if the api-gateway stack doesn't exist (or has already
3163 # been deleted/rolled-back into a non-consuming state), there's
3164 # nothing importing the analytics exports. Skip the redeploy entirely.
3165 if not self._stack_exists_in_cloudformation(
3166 api_gateway_stack,
3167 expected_stack_id=api_gateway_expected_id,
3168 require_expected_identity=strict_identity,
3169 ):
3170 logger.info(
3171 "%s does not exist in CloudFormation; skipping redeploy before analytics destroy.",
3172 api_gateway_stack,
3173 )
3174 return True
3176 # Second fast path: if the deployed api-gateway isn't actually
3177 # importing anything from the analytics stack, we don't need to
3178 # touch it. This happens when analytics was never fully wired up.
3179 if not self._api_gateway_imports_from_analytics():
3180 logger.info(
3181 "%s does not import any %s exports; skipping redeploy before analytics destroy.",
3182 api_gateway_stack,
3183 analytics_stack,
3184 )
3185 return True
3187 if strict_identity and (
3188 not strict_deployment_token or on_change_set_prepared is None or authorize_stack is None
3189 ):
3190 raise RuntimeError(
3191 "Strict analytics teardown cannot remove API imports without "
3192 "prepared-change-set authority"
3193 )
3195 try:
3196 # Temporarily disable analytics so CDK drops the /studio/* routes.
3197 current = get_analytics_config()
3198 was_enabled = current.get("enabled", False)
3199 if was_enabled:
3200 update_analytics_config({"enabled": False})
3202 print(f" Updating {api_gateway_stack} to remove analytics routes...")
3203 import tempfile
3205 with tempfile.TemporaryDirectory() as tmp_out:
3206 success = self.deploy(
3207 stack_name=api_gateway_stack,
3208 require_approval=False,
3209 exclusively=True,
3210 output_dir=tmp_out,
3211 allow_bootstrap=allow_bootstrap,
3212 bootstrap_stacks=bootstrap_stacks,
3213 expected_stack_ids=expected_stack_ids,
3214 prepared_change_sets=prepared_change_sets,
3215 authorize_stack=authorize_stack,
3216 strict_deployment_token=strict_deployment_token,
3217 on_change_set_prepared=on_change_set_prepared,
3218 on_ecr_repository_created=on_ecr_repository_created,
3219 )
3221 # Re-enable analytics so CDK can synthesize the analytics stack
3222 # for the destroy operation (custom resources need to fire).
3223 if was_enabled:
3224 update_analytics_config({"enabled": True})
3226 if not success:
3227 # The redeploy failed. That's only a real problem if the
3228 # api-gateway still imports analytics exports. Recheck:
3229 # the auto-cleanup of ROLLBACK_COMPLETE stacks may have
3230 # deleted the consumer entirely, in which case the destroy
3231 # can still proceed.
3232 if not self._api_gateway_imports_from_analytics():
3233 logger.info(
3234 "%s redeploy failed, but the stack no longer imports "
3235 "analytics exports (likely deleted during cleanup). "
3236 "Analytics destroy can proceed.",
3237 api_gateway_stack,
3238 )
3239 return True
3240 logger.error(
3241 "Failed to redeploy %s to drop analytics imports, and the "
3242 "stack still consumes analytics exports. Destroying %s will "
3243 "fail with 'Export ... cannot be deleted as it is in use'. "
3244 "Fix %s first (see events above) and retry.",
3245 api_gateway_stack,
3246 analytics_stack,
3247 api_gateway_stack,
3248 )
3249 return False
3251 return True
3252 except Exception as exc:
3253 logger.warning(
3254 "Failed to remove API gateway analytics dependency: %s",
3255 exc,
3256 exc_info=True,
3257 )
3258 # On unexpected exceptions, recheck whether imports remain.
3259 # Be permissive only if we can confirm the destroy is safe.
3260 try:
3261 return not self._api_gateway_imports_from_analytics()
3262 except Exception:
3263 return False
3265 def _api_gateway_imports_from_analytics(self) -> bool:
3266 """Return True if gco-api-gateway imports any exports from gco-analytics.
3268 Uses CloudFormation's ``list_exports`` + ``list_imports`` to detect
3269 cross-stack references at runtime. This is more reliable than
3270 inspecting the CDK app because it reflects what's actually
3271 deployed.
3272 """
3273 import boto3
3275 analytics_stack = f"{self.config.project_name}-analytics"
3276 api_gateway_stack = f"{self.config.project_name}-api-gateway"
3278 region = self._get_deploy_region(analytics_stack)
3279 if not region:
3280 return False
3282 try:
3283 cfn = boto3.client("cloudformation", region_name=region)
3284 # Collect every export whose owning stack is the analytics stack.
3285 analytics_exports: list[str] = []
3286 paginator = cfn.get_paginator("list_exports")
3287 for page in paginator.paginate():
3288 for export in page.get("Exports", []):
3289 owner = export.get("ExportingStackId", "")
3290 # ExportingStackId is a full ARN; match by stack name.
3291 if f":stack/{analytics_stack}/" in owner:
3292 analytics_exports.append(export["Name"])
3294 if not analytics_exports:
3295 return False
3297 # For each export, check whether the api-gateway stack is
3298 # listed as an importer. ``list_imports`` returns the stack
3299 # names that currently import the given export.
3300 import_paginator = cfn.get_paginator("list_imports")
3301 for export_name in analytics_exports:
3302 try:
3303 for page in import_paginator.paginate(ExportName=export_name):
3304 for importer in page.get("Imports", []):
3305 if importer == api_gateway_stack:
3306 return True
3307 except Exception as exc:
3308 # ``list_imports`` raises when an export has zero
3309 # consumers — treat that as "not imported" and move on.
3310 logger.debug(
3311 "list_imports(%s) failed (likely no consumers): %s",
3312 export_name,
3313 exc,
3314 )
3315 return False
3316 except Exception as exc:
3317 logger.debug(
3318 "Failed to check analytics imports for %s: %s",
3319 api_gateway_stack,
3320 exc,
3321 exc_info=True,
3322 )
3323 # On failure to check, err on the side of attempting the
3324 # redeploy so we don't skip necessary cleanup.
3325 return True
3327 def bootstrap(
3328 self,
3329 account: str | None = None,
3330 region: str | None = None,
3331 ) -> bool:
3332 """Bootstrap CDK in an AWS account/region."""
3333 cmd = ["bootstrap"]
3335 if account and region:
3336 cmd.append(f"aws://{account}/{region}")
3337 elif region:
3338 cmd.append(f"aws://unknown-account/{region}")
3340 result = self._run_cdk(cmd)
3341 return result.returncode == 0
3343 def is_bootstrapped(self, region: str) -> bool:
3344 """Check if CDK has been bootstrapped in a region.
3346 Looks for the CDKToolkit CloudFormation stack which is created
3347 by ``cdk bootstrap``. Result is cached per region for the lifetime
3348 of this StackManager instance.
3349 """
3350 if not hasattr(self, "_bootstrap_cache"):
3351 self._bootstrap_cache: dict[str, bool] = {}
3353 if region in self._bootstrap_cache:
3354 return self._bootstrap_cache[region]
3356 import boto3
3358 cf = boto3.client("cloudformation", region_name=region)
3359 try:
3360 response = cf.describe_stacks(StackName="CDKToolkit")
3361 stacks = response.get("Stacks", [])
3362 if stacks:
3363 status = stacks[0].get("StackStatus", "")
3364 # Any non-deleted state counts as bootstrapped
3365 result = "DELETE" not in status
3366 self._bootstrap_cache[region] = result
3367 return result
3368 except ClientError:
3369 pass # Stack doesn't exist — not bootstrapped
3370 except Exception as e:
3371 logger.debug("Failed to check CDK bootstrap in %s: %s", region, e)
3373 self._bootstrap_cache[region] = False
3374 return False
3376 def _validate_bootstrap_stack(
3377 self,
3378 region: str,
3379 expected: Mapping[str, str],
3380 ) -> None:
3381 """Require the exact preflighted CDKToolkit ARN and healthy status."""
3382 import boto3
3384 expected_id = str(expected.get("stack_id") or "")
3385 expected_status = str(expected.get("status") or "")
3386 if not expected_id or expected_status not in _BOOTSTRAP_HEALTHY_STATUSES:
3387 raise RuntimeError(f"Invalid checkpointed CDKToolkit identity for {region}")
3388 cfn = boto3.client("cloudformation", region_name=region)
3389 try:
3390 response = cfn.describe_stacks(StackName=expected_id)
3391 except Exception as exc:
3392 raise RuntimeError(
3393 f"Could not revalidate checkpointed CDKToolkit {expected_id} in {region}"
3394 ) from exc
3395 stacks = response.get("Stacks", [])
3396 if len(stacks) != 1:
3397 raise RuntimeError(f"CDKToolkit {expected_id} returned an invalid identity")
3398 stack = stacks[0]
3399 actual_id = str(stack.get("StackId") or "")
3400 actual_status = str(stack.get("StackStatus") or "")
3401 if stack.get("StackName") != "CDKToolkit" or actual_id != expected_id:
3402 raise RuntimeError(f"CDKToolkit identity changed in {region}")
3403 if actual_status != expected_status or actual_status not in _BOOTSTRAP_HEALTHY_STATUSES:
3404 raise RuntimeError(
3405 f"CDKToolkit {expected_id} status changed from {expected_status} "
3406 f"to {actual_status or 'unknown'}"
3407 )
3409 @staticmethod
3410 def _strict_change_set_name(stack_name: str, token: str) -> str:
3411 """Return one deterministic, run-scoped CloudFormation change-set name."""
3412 safe_token = "".join(
3413 character if character.isascii() and character.isalnum() else "-" for character in token
3414 )
3415 safe_token = "-".join(part for part in safe_token.split("-") if part)
3416 digest = hashlib.sha256(f"{token}:{stack_name}".encode()).hexdigest()[:16]
3417 namespace = "gco"
3418 max_token_length = 128 - len(namespace) - len(digest) - 2
3419 safe_token = (safe_token or "live-validation")[:max_token_length]
3420 return f"{namespace}-{safe_token}-{digest}"
3422 def _preflight_strict_change_set(
3423 self,
3424 *,
3425 stack_name: str,
3426 change_set_name: str,
3427 expected_stack_id: str | None,
3428 prepared_change_sets: Mapping[str, Mapping[str, str]],
3429 ) -> None:
3430 """Reject an existing deterministic change set without checkpoint authority."""
3431 import boto3
3433 region = self._get_deploy_region(stack_name)
3434 if not region:
3435 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
3436 cfn = boto3.client("cloudformation", region_name=region)
3437 try:
3438 change_set = cfn.describe_change_set(
3439 ChangeSetName=change_set_name,
3440 StackName=stack_name,
3441 )
3442 except ClientError as exc:
3443 if self._change_set_missing(exc):
3444 return
3445 # DescribeChangeSet reports a stack-style ValidationError when both
3446 # the deterministic change set and its fresh target stack are absent.
3447 # The target was authoritatively checked immediately above; only an
3448 # empty create history can safely interpret this as "not prepared".
3449 if expected_stack_id is None and not prepared_change_sets and self._stack_missing(exc):
3450 return
3451 raise RuntimeError(
3452 f"Could not preflight strict change set {change_set_name} for {stack_name}"
3453 ) from exc
3455 change_set_id = str(change_set.get("ChangeSetId") or "")
3456 observed_change_set_name = str(change_set.get("ChangeSetName") or "")
3457 stack_id = str(change_set.get("StackId") or "")
3458 if not change_set_id or not stack_id or observed_change_set_name != change_set_name:
3459 raise RuntimeError(
3460 f"Existing strict change set {change_set_name} omitted immutable identities"
3461 )
3462 self._validate_strict_change_set_arns(
3463 stack_name=stack_name,
3464 change_set_name=change_set_name,
3465 stack_id=stack_id,
3466 change_set_id=change_set_id,
3467 region=region,
3468 )
3469 prepared_record = prepared_change_sets.get(change_set_id)
3470 if prepared_record is None:
3471 raise RuntimeError(
3472 f"Existing strict change set {change_set_id} lacks checkpoint authority"
3473 )
3474 recorded_change_set_id = str(prepared_record.get("change_set_id") or "")
3475 recorded_stack_id = str(prepared_record.get("stack_id") or "")
3476 recorded_type = str(prepared_record.get("change_set_type") or "")
3477 if (
3478 expected_stack_id is None
3479 or stack_id != expected_stack_id
3480 or recorded_change_set_id != change_set_id
3481 or recorded_stack_id != stack_id
3482 or recorded_type not in {"CREATE", "UPDATE"}
3483 ):
3484 raise RuntimeError(f"Existing strict change-set authority changed for {stack_name}")
3486 @staticmethod
3487 def _validate_strict_change_set_arns(
3488 *,
3489 stack_name: str,
3490 change_set_name: str,
3491 stack_id: str,
3492 change_set_id: str,
3493 region: str,
3494 ) -> None:
3495 """Require both prepared identities to be exact, related CloudFormation ARNs."""
3497 def split_arn(identifier: str, label: str) -> tuple[str, str, str, str]:
3498 parts = identifier.split(":", 5)
3499 if (
3500 len(parts) != 6
3501 or parts[0] != "arn"
3502 or not (parts[1] == "aws" or parts[1].startswith("aws-"))
3503 or parts[2] != "cloudformation"
3504 or parts[3] != region
3505 or not parts[4]
3506 or not parts[5]
3507 ):
3508 raise RuntimeError(f"Strict {label} has an invalid CloudFormation ARN")
3509 return parts[1], parts[3], parts[4], parts[5]
3511 stack_partition, _stack_region, stack_account, stack_resource = split_arn(
3512 stack_id,
3513 "stack identity",
3514 )
3515 stack_prefix = f"stack/{stack_name}/"
3516 if not stack_resource.startswith(stack_prefix) or not stack_resource.removeprefix(
3517 stack_prefix
3518 ):
3519 raise RuntimeError(
3520 f"Strict stack identity {stack_id} does not name expected stack {stack_name}"
3521 )
3523 change_partition, _change_region, change_account, change_resource = split_arn(
3524 change_set_id,
3525 "change-set identity",
3526 )
3527 change_prefix = f"changeSet/{change_set_name}/"
3528 if not change_resource.startswith(change_prefix) or not change_resource.removeprefix(
3529 change_prefix
3530 ):
3531 raise RuntimeError(
3532 f"Strict change-set identity {change_set_id} does not name {change_set_name}"
3533 )
3534 if change_partition != stack_partition or change_account != stack_account:
3535 raise RuntimeError(
3536 "Strict stack and change-set identities belong to different AWS authorities"
3537 )
3539 def _execute_prepared_change_set(
3540 self,
3541 *,
3542 stack_name: str,
3543 change_set_name: str,
3544 expected_stack_id: str | None,
3545 expected_tags: Mapping[str, str] | None,
3546 prepared_change_sets: Mapping[str, Mapping[str, str]],
3547 preparation_succeeded: bool,
3548 authorize_stack: StackAuthorizationCallback | None,
3549 on_change_set_prepared: ChangeSetPreparedCallback,
3550 allow_noop: bool,
3551 timeout: float,
3552 ) -> bool:
3553 """Validate, checkpoint, and execute only the deterministic CDK change set."""
3554 import boto3
3556 region = self._get_deploy_region(stack_name)
3557 if not region:
3558 raise RuntimeError(f"Could not resolve deploy Region for {stack_name}")
3559 cfn = boto3.client("cloudformation", region_name=region)
3560 change_set: dict[str, Any] = {}
3561 inspection_attempts = (
3562 _STRICT_CHANGE_SET_INSPECTION_ATTEMPTS
3563 if expected_stack_id is None and not prepared_change_sets
3564 else 1
3565 )
3566 inspection_attempt = 0
3567 while True:
3568 try:
3569 change_set = cfn.describe_change_set(
3570 ChangeSetName=change_set_name,
3571 StackName=stack_name,
3572 )
3573 break
3574 except ClientError as exc:
3575 fresh_create_not_visible = bool(
3576 expected_stack_id is None
3577 and not prepared_change_sets
3578 and (self._change_set_missing(exc) or self._stack_missing(exc))
3579 )
3580 if fresh_create_not_visible and inspection_attempt + 1 < inspection_attempts:
3581 if self._cdk_cancel_event.is_set():
3582 raise RuntimeError(
3583 "Strict change-set inspection cancelled before ownership checkpoint"
3584 ) from exc
3585 time.sleep(_STRICT_CHANGE_SET_INSPECTION_RETRY_SECONDS)
3586 inspection_attempt += 1
3587 continue
3588 if not self._change_set_missing(exc) and not fresh_create_not_visible:
3589 raise RuntimeError(
3590 f"Could not inspect strict change set {change_set_name} for {stack_name}"
3591 ) from exc
3592 if allow_noop and expected_stack_id:
3593 target = self._describe_stack_target(
3594 stack_name,
3595 expected_stack_id=expected_stack_id,
3596 require_expected_identity=True,
3597 )
3598 if target is not None:
3599 stack = target[2]
3600 status = str(stack.get("StackStatus") or "")
3601 if status in _BOOTSTRAP_HEALTHY_STATUSES:
3602 if authorize_stack is None:
3603 raise RuntimeError(
3604 f"Strict no-op for {stack_name} lacks exact authorization"
3605 ) from exc
3606 authorize_stack(stack_name, region, expected_stack_id)
3607 return True
3608 raise RuntimeError(
3609 f"CDK did not create the strict change set {change_set_name} for {stack_name}"
3610 ) from exc
3612 change_set_id = str(change_set.get("ChangeSetId") or "")
3613 observed_change_set_name = str(change_set.get("ChangeSetName") or "")
3614 stack_id = str(change_set.get("StackId") or "")
3615 status = str(change_set.get("Status") or "")
3616 execution_status = str(change_set.get("ExecutionStatus") or "")
3617 if not change_set_id or not stack_id:
3618 raise RuntimeError(f"Strict change set {change_set_name} omitted immutable identities")
3619 if observed_change_set_name != change_set_name:
3620 raise RuntimeError(
3621 f"Strict change set identity changed from {change_set_name} "
3622 f"to {observed_change_set_name or 'unknown'}"
3623 )
3624 self._validate_strict_change_set_arns(
3625 stack_name=stack_name,
3626 change_set_name=change_set_name,
3627 stack_id=stack_id,
3628 change_set_id=change_set_id,
3629 region=region,
3630 )
3631 prepared_record = prepared_change_sets.get(change_set_id)
3632 if prepared_record is None:
3633 # DescribeChangeSet does not expose ChangeSetType. For a newly
3634 # prepared change set, the pre-CDK exact target state is the only
3635 # authoritative source: absence means CREATE; an exact stack means
3636 # UPDATE. Resumes use the persisted per-change-set record below.
3637 change_set_type = "CREATE" if expected_stack_id is None else "UPDATE"
3638 else:
3639 recorded_change_set_id = str(prepared_record.get("change_set_id") or "")
3640 recorded_stack_id = str(prepared_record.get("stack_id") or "")
3641 change_set_type = str(prepared_record.get("change_set_type") or "")
3642 if recorded_change_set_id != change_set_id or recorded_stack_id != stack_id:
3643 raise RuntimeError(
3644 f"Persisted strict change-set authority changed for {stack_name}"
3645 )
3646 if change_set_type not in {"CREATE", "UPDATE"}:
3647 raise RuntimeError(
3648 f"Persisted strict change set for {stack_name} has invalid type "
3649 f"{change_set_type or 'unknown'}"
3650 )
3651 if change_set_type == "UPDATE" and expected_stack_id is None:
3652 raise RuntimeError(
3653 f"Strict change set for absent {stack_name} unexpectedly performs UPDATE"
3654 )
3655 if expected_stack_id is not None and stack_id != expected_stack_id:
3656 raise RuntimeError(
3657 f"Strict change set targets replacement {stack_id}; expected {expected_stack_id}"
3658 )
3659 observed_tags = {
3660 str(tag.get("Key")): str(tag.get("Value"))
3661 for tag in change_set.get("Tags", [])
3662 if tag.get("Key") is not None
3663 }
3664 for key, value in (expected_tags or {}).items():
3665 if observed_tags.get(str(key)) != str(value):
3666 raise RuntimeError(f"Strict change set {change_set_id} omitted required tag {key}")
3668 status_reason = " ".join(str(change_set.get("StatusReason") or "").split()).lower()
3669 empty_change_set = (
3670 "submitted information didn't contain changes" in status_reason
3671 or "no updates are to be performed" in status_reason
3672 )
3673 if (
3674 status == "FAILED"
3675 and empty_change_set
3676 and (allow_noop or prepared_record is not None)
3677 and expected_stack_id
3678 ):
3679 # stack_id == expected_stack_id is already guaranteed here: the
3680 # identity check above raises for any mismatch whenever
3681 # expected_stack_id is not None, and this branch requires a
3682 # truthy expected_stack_id.
3683 target = self._describe_stack_target(
3684 stack_name,
3685 expected_stack_id=expected_stack_id,
3686 require_expected_identity=True,
3687 )
3688 if target is None or str(target[2].get("StackStatus") or "") not in (
3689 _BOOTSTRAP_HEALTHY_STATUSES
3690 ):
3691 raise RuntimeError(
3692 f"Empty strict change set {change_set_id} has no healthy exact stack"
3693 )
3694 if authorize_stack is None:
3695 raise RuntimeError(f"Strict no-op for {stack_name} lacks exact authorization")
3696 authorize_stack(stack_name, region, expected_stack_id)
3697 on_change_set_prepared(
3698 stack_name,
3699 region,
3700 stack_id,
3701 change_set_id,
3702 change_set_type,
3703 )
3704 return True
3705 if status != "CREATE_COMPLETE" or execution_status not in {
3706 "AVAILABLE",
3707 "EXECUTE_COMPLETE",
3708 }:
3709 raise RuntimeError(
3710 f"Strict change set {change_set_id} is {status}/{execution_status}, not usable"
3711 )
3713 if execution_status == "AVAILABLE" and (
3714 prepared_record is None and not preparation_succeeded
3715 ):
3716 raise RuntimeError(
3717 f"Strict change set {change_set_id} was not produced by this preparation"
3718 )
3719 if execution_status == "EXECUTE_COMPLETE" and (
3720 prepared_record is None or expected_stack_id is None
3721 ):
3722 raise RuntimeError(
3723 f"Executed strict change set {change_set_id} lacks prior checkpoint authority"
3724 )
3726 if expected_stack_id is not None:
3727 if authorize_stack is None:
3728 raise RuntimeError(f"Strict change set for {stack_name} lacks exact authorization")
3729 authorize_stack(stack_name, region, stack_id)
3731 if execution_status == "EXECUTE_COMPLETE":
3732 target = self._describe_stack_target(
3733 stack_name,
3734 expected_stack_id=stack_id,
3735 require_expected_identity=True,
3736 )
3737 if target is None or str(target[2].get("StackStatus") or "") not in (
3738 _BOOTSTRAP_HEALTHY_STATUSES
3739 ):
3740 raise RuntimeError(
3741 f"Executed strict change set {change_set_id} has no healthy exact stack"
3742 )
3743 elif change_set_type == "CREATE":
3744 target = self._describe_stack_target(
3745 stack_name,
3746 expected_stack_id=stack_id,
3747 require_expected_identity=True,
3748 )
3749 if target is None or str(target[2].get("StackStatus") or "") != "REVIEW_IN_PROGRESS":
3750 raise RuntimeError(
3751 f"Prepared CREATE change set {change_set_id} has no exact review stack"
3752 )
3754 on_change_set_prepared(
3755 stack_name,
3756 region,
3757 stack_id,
3758 change_set_id,
3759 change_set_type,
3760 )
3761 if execution_status == "EXECUTE_COMPLETE":
3762 return True
3763 if self._cdk_cancel_event.is_set():
3764 raise RuntimeError(
3765 f"Strict change set {change_set_id} was checkpointed but execution was cancelled"
3766 )
3768 cfn.execute_change_set(ChangeSetName=change_set_id)
3769 settled = self._wait_for_stack_settle(
3770 stack_name,
3771 timeout=timeout,
3772 stack_identifier=stack_id,
3773 )
3774 if settled not in _BOOTSTRAP_HEALTHY_STATUSES:
3775 logger.error(
3776 "Strict change set %s for %s settled as %s",
3777 change_set_id,
3778 stack_name,
3779 settled or "unknown",
3780 )
3781 return False
3782 return True
3784 def ensure_bootstrapped(self, region: str) -> bool:
3785 """Ensure a region is CDK-bootstrapped, auto-bootstrapping if needed.
3787 Returns True if the region is (or was successfully) bootstrapped.
3788 """
3789 if self.is_bootstrapped(region):
3790 return True
3792 print(f"ℹ Region {region} is not CDK-bootstrapped. Bootstrapping now...")
3793 success = self.bootstrap(region=region)
3794 if success:
3795 # Update cache so we don't re-check this region
3796 if not hasattr(self, "_bootstrap_cache"):
3797 self._bootstrap_cache = {}
3798 self._bootstrap_cache[region] = True
3799 print(f"✓ CDK bootstrapped in {region}")
3800 else:
3801 print(f"✗ Failed to bootstrap CDK in {region}")
3802 return success
3804 def _get_deploy_region(self, stack_name: str) -> str | None:
3805 """Determine the target AWS region for a given stack name."""
3806 from .config import _load_cdk_json
3808 cdk_regions = _load_cdk_json()
3810 # Named stacks are classified by suffix and regional stacks by the
3811 # ``<project>-`` prefix (#139) so a non-``gco`` deployment resolves
3812 # regions for its own ``<project>-*`` stacks — otherwise the image
3813 # mirror (which calls this to pick a regional stack's region) would
3814 # silently no-op. For the default ``gco`` behaviour is unchanged.
3815 region: str | None
3816 if stack_name.endswith("-global"):
3817 region = cdk_regions.get("global") or self.config.global_region
3818 return region
3819 if stack_name.endswith("-api-gateway"):
3820 region = cdk_regions.get("api_gateway") or self.config.api_gateway_region
3821 return region
3822 if stack_name.endswith("-monitoring"):
3823 region = cdk_regions.get("monitoring") or self.config.monitoring_region
3824 return region
3825 if stack_name.endswith("-analytics"):
3826 # The analytics stack shares the API gateway region so the
3827 # presigned-URL Lambda can hook into the existing /studio/*
3828 # routes on the same API Gateway.
3829 region = cdk_regions.get("api_gateway") or self.config.api_gateway_region
3830 return region
3832 # Regional API bridges use ``<project>-regional-api-<region>``. Resolve
3833 # this exact shape before generic regional stacks; otherwise the generic
3834 # project-prefix branch returns the malformed ``regional-api-<region>``.
3835 # Requiring a configured deployment region also prevents bridge-shaped
3836 # typos from being treated as valid AWS regions.
3837 bridge_prefix = f"{self.config.project_name}-regional-api-"
3838 if stack_name.startswith(bridge_prefix):
3839 region = stack_name[len(bridge_prefix) :]
3840 configured_regions = {str(item) for item in (cdk_regions.get("regional") or [])}
3841 return region if region in configured_regions else None
3843 # Base regional stacks: {project}-{region}. The region is whatever
3844 # follows the project prefix (regions contain hyphens, so we strip
3845 # the known prefix rather than guess a split point).
3846 prefix = f"{self.config.project_name}-"
3847 if stack_name.startswith(prefix):
3848 return stack_name[len(prefix) :]
3850 return None
3852 def _mirror_target_regions(self, stack_name: str | None, all_stacks: bool) -> list[str]:
3853 """Regional regions to auto-mirror images for on this deploy.
3855 Only regional stacks (``gco-<region>``) run a Helm install that needs the
3856 mirror; the named global / api-gateway / monitoring / analytics stacks do
3857 not. For ``--all`` the regional regions come straight from cdk.json
3858 (``deployment_regions.regional``) so no synth is required. Returns a
3859 de-duplicated, order-stable list.
3860 """
3861 # Derive the prefix from project_name (#139) so a non-``gco``
3862 # deployment's regional stacks (``<project>-<region>``) are still
3863 # recognised — otherwise the mirror would silently no-op and the
3864 # regional Volcano Helm install would have no images to pull.
3865 prefix = f"{self.config.project_name}-"
3866 named = {
3867 f"{prefix}global",
3868 f"{prefix}api-gateway",
3869 f"{prefix}monitoring",
3870 f"{prefix}analytics",
3871 }
3872 if all_stacks:
3873 from .config import _load_cdk_json
3875 regional = _load_cdk_json().get("regional") or []
3876 return list(dict.fromkeys(str(r) for r in regional))
3878 # Bridge stacks contain no regional Helm consumers and must not trigger
3879 # image mirroring. Match the exact project-scoped prefix so a project
3880 # name containing ``regional-api`` remains unambiguous.
3881 bridge_prefix = f"{self.config.project_name}-regional-api-"
3882 if stack_name and stack_name.startswith(bridge_prefix):
3883 return []
3885 if stack_name and stack_name.startswith(prefix) and stack_name not in named:
3886 region = self._get_deploy_region(stack_name)
3887 return [region] if region else []
3888 return []
3890 def _mirror_images_if_enabled(
3891 self,
3892 stack_name: str | None,
3893 all_stacks: bool,
3894 repository_tags: Mapping[str, str] | None = None,
3895 on_repository_created: EcrRepositoryCreatedCallback | None = None,
3896 ) -> None:
3897 """Mirror third-party images into ECR before a regional stack deploys.
3899 No-op unless ``volcano_image_mirror.enabled`` is set in cdk.json. Mirrors
3900 every relevant regional region (see :meth:`_mirror_target_regions`); the
3901 copy is idempotent and skips images already present, so a fresh deploy
3902 seeds the mirror automatically and repeat deploys cost only a few ECR
3903 describe calls. Raises **before** any CDK call if an enabled mirror fails,
3904 so a deploy never points a consumer (e.g. Volcano's ``image_registry``)
3905 at images that aren't in ECR yet.
3906 """
3907 from . import _image_mirror as image_mirror
3909 cfg = image_mirror.read_mirror_config()
3910 if not cfg["enabled"]:
3911 return
3913 regions = self._mirror_target_regions(stack_name, all_stacks)
3914 for region in regions:
3915 print(f"Mirroring third-party images into ECR for {region} ...")
3916 try:
3917 image_mirror.mirror_images(
3918 region,
3919 ecr_namespace=cfg["ecr_namespace"],
3920 skip_existing=True,
3921 repository_tags=repository_tags,
3922 on_repository_created=on_repository_created,
3923 )
3924 except Exception as exc: # noqa: BLE001 - surface a clear, actionable failure
3925 raise RuntimeError(
3926 f"Image mirror failed for region {region}: {exc}\n"
3927 "volcano_image_mirror is enabled but the images could not be "
3928 "mirrored into ECR. Fix the cause (container runtime / network / "
3929 "credentials) or run "
3930 f"'gco images mirror --region {region}' manually, "
3931 "then retry. Aborting before CDK so the deploy never points a "
3932 "consumer at images that aren't in ECR."
3933 ) from exc
3935 def get_outputs(self, stack_name: str, region: str) -> dict[str, str]:
3936 """Get stack outputs from CloudFormation."""
3937 import boto3
3939 cf = boto3.client("cloudformation", region_name=region)
3940 try:
3941 response = cf.describe_stacks(StackName=stack_name)
3942 if response["Stacks"]:
3943 stack = response["Stacks"][0]
3944 outputs: dict[str, str] = {}
3945 for output in stack.get("Outputs", []):
3946 outputs[str(output["OutputKey"])] = str(output["OutputValue"])
3947 return outputs
3948 except Exception as e:
3949 logger.debug("Failed to get outputs for %s in %s: %s", stack_name, region, e)
3950 return {}
3952 def get_stack_status(self, stack_name: str, region: str) -> StackInfo | None:
3953 """Get detailed stack status from CloudFormation."""
3954 import boto3
3956 cf = boto3.client("cloudformation", region_name=region)
3957 try:
3958 response = cf.describe_stacks(StackName=stack_name)
3959 if response["Stacks"]:
3960 stack = response["Stacks"][0]
3961 return StackInfo(
3962 name=stack["StackName"],
3963 status=stack["StackStatus"],
3964 region=region,
3965 created_time=stack.get("CreationTime"),
3966 updated_time=stack.get("LastUpdatedTime"),
3967 outputs={o["OutputKey"]: o["OutputValue"] for o in stack.get("Outputs", [])},
3968 tags={t["Key"]: t["Value"] for t in stack.get("Tags", [])},
3969 )
3970 except Exception as e:
3971 logger.debug("Failed to get stack status for %s in %s: %s", stack_name, region, e)
3972 return None
3974 def deploy_orchestrated(
3975 self,
3976 require_approval: bool = True,
3977 outputs_file: str | None = None,
3978 parameters: dict[str, str] | None = None,
3979 tags: dict[str, str] | None = None,
3980 progress: str = "events",
3981 on_stack_start: Callable[[str], None] | None = None,
3982 on_stack_complete: Callable[[str, bool], None] | None = None,
3983 parallel: bool = False,
3984 max_workers: int = 4,
3985 allow_bootstrap: bool = True,
3986 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
3987 expected_stack_ids: Mapping[str, str | None] | None = None,
3988 prepared_change_sets: PreparedChangeSetAuthority | None = None,
3989 authorize_stack: StackAuthorizationCallback | None = None,
3990 strict_deployment_token: str | None = None,
3991 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
3992 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
3993 ) -> tuple[bool, list[str], list[str]]:
3994 """
3995 Deploy all stacks in the correct order.
3997 Deploys global stacks first, then base regional stacks, regional API
3998 bridges, and finally monitoring. Parallelism never crosses a dependency
3999 level.
4001 Args:
4002 require_approval: Whether to require approval for changes
4003 outputs_file: File to write outputs to
4004 parameters: CDK parameters
4005 tags: Tags to apply to stacks
4006 progress: Progress display type
4007 on_stack_start: Callback(stack_name) called when starting a stack
4008 on_stack_complete: Callback(stack_name, success) called when stack completes
4009 parallel: Deploy regional stacks in parallel
4010 max_workers: Maximum number of parallel deployments (default: 4)
4012 Returns:
4013 Tuple of (overall_success, successful_stacks, failed_stacks)
4014 """
4015 stacks = self.list_stacks()
4016 stack_names = set(stacks)
4017 project_name = self.config.project_name
4018 ordered_stacks = get_stack_deployment_order(stacks, project_name=project_name)
4020 strict_deployment = (
4021 strict_deployment_token is not None or on_change_set_prepared is not None
4022 )
4023 if strict_deployment:
4024 if not strict_deployment_token or on_change_set_prepared is None:
4025 raise RuntimeError(
4026 "Strict deployment requires both a run token and a prepared-change-set callback"
4027 )
4028 if allow_bootstrap:
4029 raise RuntimeError("Strict orchestrated deployment cannot auto-bootstrap")
4030 if authorize_stack is None:
4031 raise RuntimeError("Strict orchestrated deployment requires an exact authorizer")
4032 if expected_stack_ids is None:
4033 raise RuntimeError("Strict orchestrated deployment lacks target identities")
4034 if prepared_change_sets is None:
4035 raise RuntimeError("Strict orchestrated deployment lacks change-set history")
4036 missing = sorted(set(stacks) - set(expected_stack_ids))
4037 unexpected = sorted(set(expected_stack_ids) - set(stacks))
4038 if missing or unexpected:
4039 raise RuntimeError(
4040 "Strict deployment target map does not match the CDK graph; "
4041 f"missing={missing}, unexpected={unexpected}"
4042 )
4043 missing_history = sorted(set(stacks) - set(prepared_change_sets))
4044 unexpected_history = sorted(set(prepared_change_sets) - set(stacks))
4045 if missing_history or unexpected_history:
4046 raise RuntimeError(
4047 "Strict change-set history does not match the CDK graph; "
4048 f"missing={missing_history}, unexpected={unexpected_history}"
4049 )
4051 # Validate every toolkit and every expected stack before the first
4052 # repository copy, stuck-stack recovery, or CloudFormation mutation.
4053 validated_regions: set[str] = set()
4054 for target_name in ordered_stacks:
4055 region = self._get_deploy_region(target_name)
4056 if not region:
4057 raise RuntimeError(f"Could not resolve deploy Region for {target_name}")
4058 if region not in validated_regions:
4059 expected_bootstrap = (bootstrap_stacks or {}).get(region)
4060 if expected_bootstrap is None:
4061 raise RuntimeError(
4062 f"Strict deployment lacks a checkpointed CDKToolkit identity for {region}"
4063 )
4064 self._validate_bootstrap_stack(region, expected_bootstrap)
4065 validated_regions.add(region)
4067 expected_id = expected_stack_ids[target_name]
4068 target = self._describe_stack_target(
4069 target_name,
4070 expected_stack_id=expected_id,
4071 require_expected_identity=True,
4072 )
4073 if expected_id is not None and target is None:
4074 raise RuntimeError(
4075 f"Checkpointed stack {expected_id} is absent; refusing recreation"
4076 )
4077 if target is not None:
4078 authorize_stack(target_name, region, str(target[2]["StackId"]))
4080 # Separate stacks into four dependency levels by suffix/marker so
4081 # ordering is independent of project_name (#139):
4082 # 1. Pre-regional global stacks (<project>-global, <project>-api-gateway)
4083 # 2. Base regional stacks (<project>-<region>, parallel-safe)
4084 # 3. Regional API bridges (<project>-regional-api-<region>, depend on base)
4085 # 4. Monitoring (depends on regional stacks)
4086 pre_regional_stacks = [s for s in ordered_stacks if s.endswith(("-global", "-api-gateway"))]
4087 regional_api_stacks = [
4088 s
4089 for s in ordered_stacks
4090 if _is_regional_api_bridge_stack(
4091 s,
4092 project_name=project_name,
4093 stack_names=stack_names,
4094 )
4095 ]
4096 regional_stacks = [
4097 s
4098 for s in ordered_stacks
4099 if not s.endswith(("-global", "-api-gateway", "-monitoring"))
4100 and not _is_regional_api_bridge_stack(
4101 s,
4102 project_name=project_name,
4103 stack_names=stack_names,
4104 )
4105 ]
4106 post_regional_stacks = [s for s in ordered_stacks if s.endswith("-monitoring")]
4108 successful: list[str] = []
4109 failed: list[str] = []
4110 deployment_safety: _StackOperationSafetyKwargs = {
4111 "allow_bootstrap": allow_bootstrap,
4112 "bootstrap_stacks": bootstrap_stacks,
4113 "expected_stack_ids": expected_stack_ids,
4114 "prepared_change_sets": prepared_change_sets,
4115 "authorize_stack": authorize_stack,
4116 "strict_deployment_token": strict_deployment_token,
4117 "on_change_set_prepared": on_change_set_prepared,
4118 "on_ecr_repository_created": on_ecr_repository_created,
4119 }
4121 # Phase 1: Deploy pre-regional global stacks sequentially
4122 for stack_name in pre_regional_stacks:
4123 if on_stack_start:
4124 on_stack_start(stack_name)
4126 success = self.deploy(
4127 stack_name=stack_name,
4128 require_approval=require_approval,
4129 outputs_file=outputs_file,
4130 parameters=parameters,
4131 tags=tags,
4132 progress=progress,
4133 **deployment_safety,
4134 )
4136 if success:
4137 successful.append(stack_name)
4138 else:
4139 failed.append(stack_name)
4141 if on_stack_complete:
4142 on_stack_complete(stack_name, success)
4144 # Stop on failure to prevent cascading issues
4145 if not success:
4146 return False, successful, failed
4148 # Phase 2: Deploy regional stacks (parallel or sequential)
4149 # All regional stacks pass --exclusively: globals are already deployed
4150 # in Phase 1, so CDK doesn't need to re-evaluate them. Skipping that
4151 # re-evaluation avoids re-running custom resources (notably
4152 # KubectlApplyManifests) on the global stacks every time a regional
4153 # stack is deployed — that would otherwise re-apply manifests and
4154 # rollout-restart controllers for no actual change.
4155 if regional_stacks:
4156 if parallel and len(regional_stacks) > 1:
4157 # Parallel deployment of regional stacks
4158 successful_regional, failed_regional = self._deploy_stacks_parallel(
4159 stacks=regional_stacks,
4160 require_approval=require_approval,
4161 outputs_file=outputs_file,
4162 parameters=parameters,
4163 tags=tags,
4164 progress=progress,
4165 on_stack_start=on_stack_start,
4166 on_stack_complete=on_stack_complete,
4167 max_workers=max_workers,
4168 allow_bootstrap=allow_bootstrap,
4169 bootstrap_stacks=bootstrap_stacks,
4170 expected_stack_ids=expected_stack_ids,
4171 prepared_change_sets=prepared_change_sets,
4172 authorize_stack=authorize_stack,
4173 strict_deployment_token=strict_deployment_token,
4174 on_change_set_prepared=on_change_set_prepared,
4175 on_ecr_repository_created=on_ecr_repository_created,
4176 )
4177 successful.extend(successful_regional)
4178 failed.extend(failed_regional)
4180 # Stop if any regional stack failed
4181 if failed_regional:
4182 return False, successful, failed
4183 else:
4184 # Sequential deployment
4185 for stack_name in regional_stacks:
4186 if on_stack_start:
4187 on_stack_start(stack_name)
4189 success = self.deploy(
4190 stack_name=stack_name,
4191 require_approval=require_approval,
4192 outputs_file=outputs_file,
4193 parameters=parameters,
4194 tags=tags,
4195 progress=progress,
4196 exclusively=True,
4197 **deployment_safety,
4198 )
4200 if success:
4201 successful.append(stack_name)
4202 else:
4203 failed.append(stack_name)
4205 if on_stack_complete:
4206 on_stack_complete(stack_name, success)
4208 # Stop on failure
4209 if not success:
4210 return False, successful, failed
4212 # Phase 3: Deploy regional API bridges only after every base regional
4213 # stack is complete. Bridges within this level remain parallel-safe.
4214 if regional_api_stacks:
4215 if parallel and len(regional_api_stacks) > 1:
4216 successful_api, failed_api = self._deploy_stacks_parallel(
4217 stacks=regional_api_stacks,
4218 require_approval=require_approval,
4219 outputs_file=outputs_file,
4220 parameters=parameters,
4221 tags=tags,
4222 progress=progress,
4223 on_stack_start=on_stack_start,
4224 on_stack_complete=on_stack_complete,
4225 max_workers=max_workers,
4226 allow_bootstrap=allow_bootstrap,
4227 bootstrap_stacks=bootstrap_stacks,
4228 expected_stack_ids=expected_stack_ids,
4229 prepared_change_sets=prepared_change_sets,
4230 authorize_stack=authorize_stack,
4231 strict_deployment_token=strict_deployment_token,
4232 on_change_set_prepared=on_change_set_prepared,
4233 on_ecr_repository_created=on_ecr_repository_created,
4234 )
4235 successful.extend(successful_api)
4236 failed.extend(failed_api)
4237 if failed_api:
4238 return False, successful, failed
4239 else:
4240 for stack_name in regional_api_stacks:
4241 if on_stack_start:
4242 on_stack_start(stack_name)
4244 success = self.deploy(
4245 stack_name=stack_name,
4246 require_approval=require_approval,
4247 outputs_file=outputs_file,
4248 parameters=parameters,
4249 tags=tags,
4250 progress=progress,
4251 exclusively=True,
4252 **deployment_safety,
4253 )
4254 if success:
4255 successful.append(stack_name)
4256 else:
4257 failed.append(stack_name)
4258 if on_stack_complete:
4259 on_stack_complete(stack_name, success)
4260 if not success:
4261 return False, successful, failed
4263 # Phase 4: Deploy post-regional stacks (monitoring) sequentially.
4264 # Same rationale as Phase 2: every upstream stack is already
4265 # deployed, so --exclusively prevents a redundant pass over
4266 # global/api-gateway/regional.
4267 for stack_name in post_regional_stacks:
4268 if on_stack_start:
4269 on_stack_start(stack_name)
4271 success = self.deploy(
4272 stack_name=stack_name,
4273 require_approval=require_approval,
4274 outputs_file=outputs_file,
4275 parameters=parameters,
4276 tags=tags,
4277 progress=progress,
4278 exclusively=True,
4279 **deployment_safety,
4280 )
4282 if success:
4283 successful.append(stack_name)
4284 else:
4285 failed.append(stack_name)
4287 if on_stack_complete:
4288 on_stack_complete(stack_name, success)
4290 if not success:
4291 return False, successful, failed
4293 return len(failed) == 0, successful, failed
4295 def _deploy_stacks_parallel(
4296 self,
4297 stacks: list[str],
4298 require_approval: bool,
4299 outputs_file: str | None,
4300 parameters: dict[str, str] | None,
4301 tags: dict[str, str] | None,
4302 progress: str,
4303 on_stack_start: Callable[[str], None] | None,
4304 on_stack_complete: Callable[[str, bool], None] | None,
4305 max_workers: int,
4306 allow_bootstrap: bool,
4307 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None,
4308 expected_stack_ids: Mapping[str, str | None] | None,
4309 prepared_change_sets: PreparedChangeSetAuthority | None,
4310 authorize_stack: StackAuthorizationCallback | None,
4311 strict_deployment_token: str | None = None,
4312 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
4313 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
4314 ) -> tuple[list[str], list[str]]:
4315 """Deploy multiple stacks in parallel using separate CDK output directories."""
4316 import tempfile
4318 successful: list[str] = []
4319 failed: list[str] = []
4320 lock = Lock()
4322 def deploy_single(stack_name: str) -> tuple[str, bool]:
4323 # Use a unique output directory in /tmp for each parallel deployment
4324 # This avoids CDK copying cdk.out.* directories into assets
4325 output_dir = tempfile.mkdtemp(prefix=f"cdk-{stack_name}-")
4326 try:
4327 if on_stack_start:
4328 with lock:
4329 on_stack_start(stack_name)
4331 success = self.deploy(
4332 stack_name=stack_name,
4333 require_approval=require_approval,
4334 outputs_file=outputs_file,
4335 parameters=parameters,
4336 tags=tags,
4337 progress=progress,
4338 output_dir=output_dir,
4339 exclusively=True,
4340 allow_bootstrap=allow_bootstrap,
4341 bootstrap_stacks=bootstrap_stacks,
4342 expected_stack_ids=expected_stack_ids,
4343 prepared_change_sets=prepared_change_sets,
4344 authorize_stack=authorize_stack,
4345 strict_deployment_token=strict_deployment_token,
4346 on_change_set_prepared=on_change_set_prepared,
4347 on_ecr_repository_created=on_ecr_repository_created,
4348 )
4349 return stack_name, success
4350 finally:
4351 try:
4352 import shutil
4354 if os.path.exists(output_dir):
4355 shutil.rmtree(output_dir)
4356 except Exception as e:
4357 logger.debug("Cleanup of %s failed: %s", output_dir, e)
4359 self._cdk_cancel_event.clear()
4360 futures: dict[Any, str] = {}
4361 executor = ThreadPoolExecutor(max_workers=max_workers)
4362 try:
4363 futures = {executor.submit(deploy_single, stack): stack for stack in stacks}
4365 for future in as_completed(futures):
4366 stack_name, success = future.result()
4368 with lock:
4369 if success:
4370 successful.append(stack_name)
4371 else:
4372 failed.append(stack_name)
4374 if on_stack_complete:
4375 on_stack_complete(stack_name, success)
4376 except BaseException:
4377 # Terminate registered process groups before waiting for executor
4378 # shutdown; the context-manager form waits first and can deadlock an
4379 # interrupted orchestration behind a still-running CDK worker.
4380 self.cancel_active_cdk_processes()
4381 for future in futures:
4382 future.cancel()
4383 executor.shutdown(wait=True, cancel_futures=True)
4384 raise
4385 else:
4386 executor.shutdown(wait=True)
4387 finally:
4388 self._cdk_cancel_event.clear()
4390 return successful, failed
4392 def _resolve_strict_teardown_resources(
4393 self,
4394 *,
4395 stacks: Collection[str],
4396 regional_stacks: Collection[str],
4397 expected_stack_ids: Mapping[str, str | None],
4398 authorize_stack: StackAuthorizationCallback,
4399 ) -> dict[str, dict[str, str]]:
4400 """Authorize every live stack, then resolve helper IDs from exact stack ARNs."""
4401 import boto3
4403 live_targets: dict[str, tuple[str, Any, dict[str, Any]]] = {}
4404 for stack_name in stacks:
4405 expected_stack_id = expected_stack_ids[stack_name]
4406 target = self._describe_stack_target(
4407 stack_name,
4408 expected_stack_id=expected_stack_id,
4409 require_expected_identity=True,
4410 )
4411 if target is None:
4412 continue
4413 region, _cloudformation, stack = target
4414 stack_id = str(stack["StackId"])
4415 authorize_stack(stack_name, region, stack_id)
4416 live_targets[stack_name] = target
4418 project_name = self.config.project_name
4419 base_regional_stacks: list[str] = []
4420 for stack_name in regional_stacks:
4421 deploy_region = self._get_deploy_region(stack_name)
4422 if deploy_region and stack_name == f"{project_name}-{deploy_region}":
4423 base_regional_stacks.append(stack_name)
4425 resolved: dict[str, dict[str, str]] = {}
4426 for stack_name in base_regional_stacks:
4427 target = live_targets.get(stack_name)
4428 if target is None:
4429 continue
4430 region, cloudformation, stack = target
4431 stack_id = str(stack["StackId"])
4432 summaries: list[dict[str, Any]] = []
4433 paginator = cloudformation.get_paginator("list_stack_resources")
4434 for page in paginator.paginate(StackName=stack_id):
4435 summaries.extend(page.get("StackResourceSummaries", []))
4437 vpc_ids = {
4438 str(item["PhysicalResourceId"])
4439 for item in summaries
4440 if item.get("ResourceType") == "AWS::EC2::VPC" and item.get("PhysicalResourceId")
4441 }
4442 cluster_names = {
4443 str(item["PhysicalResourceId"])
4444 for item in summaries
4445 if item.get("ResourceType") == "AWS::EKS::Cluster"
4446 and item.get("PhysicalResourceId")
4447 }
4448 if len(vpc_ids) > 1 or len(cluster_names) > 1:
4449 raise RuntimeError(f"Exact stack {stack_id} returned ambiguous VPC/EKS resources")
4451 details = {
4452 "stack_name": stack_name,
4453 "stack_id": stack_id,
4454 "region": region,
4455 }
4456 vpc_id = next(iter(vpc_ids), "")
4457 cluster_name = next(iter(cluster_names), "")
4458 if vpc_id:
4459 details["vpc_id"] = vpc_id
4460 if cluster_name:
4461 details["cluster_name"] = cluster_name
4462 cluster: dict[str, Any] | None
4463 try:
4464 cluster = boto3.client("eks", region_name=region).describe_cluster(
4465 name=cluster_name
4466 )["cluster"]
4467 except ClientError as exc:
4468 if exc.response.get("Error", {}).get("Code") == "ResourceNotFoundException":
4469 cluster = None
4470 else:
4471 raise
4472 if cluster is not None:
4473 if str(cluster.get("name") or "") != cluster_name:
4474 raise RuntimeError(
4475 f"EKS returned a changed identity for {region}:{cluster_name}"
4476 )
4477 networking = cluster.get("resourcesVpcConfig") or {}
4478 cluster_vpc_id = str(networking.get("vpcId") or "")
4479 security_group_id = str(networking.get("clusterSecurityGroupId") or "")
4480 if vpc_id and cluster_vpc_id != vpc_id:
4481 raise RuntimeError(
4482 f"EKS cluster {cluster_name} no longer belongs to exact VPC {vpc_id}"
4483 )
4484 if not security_group_id:
4485 raise RuntimeError(
4486 f"EKS cluster {cluster_name} omitted its security-group identity"
4487 )
4488 details["cluster_security_group_id"] = security_group_id
4489 elif vpc_id:
4490 # On teardown resume the cluster may already be gone while
4491 # its managed SG remains. Resolve the SG ID inside the exact
4492 # stack VPC using the exact cluster physical ID.
4493 ec2 = boto3.client("ec2", region_name=region)
4494 groups = ec2.describe_security_groups(
4495 Filters=[
4496 {"Name": "vpc-id", "Values": [vpc_id]},
4497 {
4498 "Name": "tag:aws:eks:cluster-name",
4499 "Values": [cluster_name],
4500 },
4501 ]
4502 ).get("SecurityGroups", [])
4503 group_ids = {str(group["GroupId"]) for group in groups if group.get("GroupId")}
4504 if len(group_ids) > 1:
4505 raise RuntimeError(
4506 f"Exact VPC {vpc_id} has ambiguous EKS security groups for "
4507 f"{cluster_name}"
4508 )
4509 if group_ids:
4510 details["cluster_security_group_id"] = next(iter(group_ids))
4511 resolved[stack_name] = details
4512 return resolved
4514 def _destroy_phase_remaining_stacks(
4515 self,
4516 phase_name: str,
4517 stacks: Collection[str],
4518 expected_stack_ids: Mapping[str, str | None] | None = None,
4519 ) -> list[str]:
4520 """Return stacks still present after a dependency phase.
4522 A lookup error is treated as present: advancing when absence cannot be
4523 proven is less safe than stopping for an operator retry.
4524 """
4525 remaining: list[str] = []
4526 for stack_name in stacks:
4527 try:
4528 present = self._stack_exists_in_cloudformation(
4529 stack_name,
4530 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4531 require_expected_identity=expected_stack_ids is not None,
4532 )
4533 except Exception:
4534 logger.exception(
4535 "Could not verify %s absence after %s",
4536 stack_name,
4537 phase_name,
4538 )
4539 present = True
4540 if present:
4541 remaining.append(stack_name)
4542 if remaining:
4543 print(
4544 f" {phase_name} barrier blocked: stack absence was not confirmed for "
4545 + ", ".join(remaining)
4546 )
4547 return remaining
4549 def destroy_orchestrated(
4550 self,
4551 force: bool = False,
4552 on_stack_start: Callable[[str], None] | None = None,
4553 on_stack_complete: Callable[[str, bool], None] | None = None,
4554 parallel: bool = False,
4555 max_workers: int = 4,
4556 expected_stack_ids: Mapping[str, str | None] | None = None,
4557 prepared_change_sets: PreparedChangeSetAuthority | None = None,
4558 authorize_stack: StackAuthorizationCallback | None = None,
4559 allow_bootstrap: bool = True,
4560 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None = None,
4561 on_cleanup_complete: CleanupOutcomeCallback | None = None,
4562 strict_deployment_token: str | None = None,
4563 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
4564 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
4565 retain_volumes: bool = False,
4566 ) -> tuple[bool, list[str], list[str]]:
4567 """Destroy stacks in dependency order with optional exact-ARN authority.
4569 ``retain_volumes`` reports the destroyed clusters' orphaned CSI volumes
4570 instead of deleting them; see ``_cleanup_cluster_volumes``.
4571 """
4572 app_stacks = self.list_stacks()
4573 strict_identity = expected_stack_ids is not None
4574 if strict_identity:
4575 assert expected_stack_ids is not None
4576 missing = sorted(set(app_stacks) - set(expected_stack_ids))
4577 if missing:
4578 raise RuntimeError(
4579 f"Strict teardown target map is incomplete before cleanup; missing={missing}"
4580 )
4581 if authorize_stack is None:
4582 raise RuntimeError("Strict teardown requires an exact stack authorizer")
4583 invalid = sorted(
4584 name
4585 for name, stack_id in expected_stack_ids.items()
4586 if stack_id is not None and not str(stack_id).startswith("arn:")
4587 )
4588 if invalid:
4589 raise RuntimeError(
4590 f"Strict teardown has invalid stack identities for: {', '.join(invalid)}"
4591 )
4593 strict_prepared_deployment = (
4594 strict_deployment_token is not None or on_change_set_prepared is not None
4595 )
4596 if strict_prepared_deployment:
4597 if not strict_deployment_token or on_change_set_prepared is None:
4598 raise RuntimeError(
4599 "Strict teardown dependency deployment requires both a run token "
4600 "and a prepared-change-set callback"
4601 )
4602 if prepared_change_sets is None:
4603 raise RuntimeError("Strict teardown lacks prepared change-set history")
4604 expected_history_keys = set(expected_stack_ids or {})
4605 if set(prepared_change_sets) != expected_history_keys:
4606 raise RuntimeError(
4607 "Strict teardown change-set history does not match target identities"
4608 )
4610 stacks = list(app_stacks)
4611 if expected_stack_ids is not None:
4612 for stack_name in expected_stack_ids:
4613 if stack_name not in stacks:
4614 stacks.append(stack_name)
4615 project_name = self.config.project_name
4616 (
4617 post_regional_stacks,
4618 regional_api_stacks,
4619 regional_stacks,
4620 pre_regional_stacks,
4621 ) = _get_stack_destroy_phases(stacks, project_name=project_name)
4623 strict_resources: dict[str, dict[str, str]] = {}
4624 if strict_identity:
4625 assert expected_stack_ids is not None
4626 assert authorize_stack is not None
4627 strict_resources = self._resolve_strict_teardown_resources(
4628 stacks=stacks,
4629 regional_stacks=regional_stacks,
4630 expected_stack_ids=expected_stack_ids,
4631 authorize_stack=authorize_stack,
4632 )
4634 destroy_safety: _StackOperationSafetyKwargs = {
4635 "expected_stack_ids": expected_stack_ids,
4636 "prepared_change_sets": prepared_change_sets,
4637 "authorize_stack": authorize_stack,
4638 "allow_bootstrap": allow_bootstrap,
4639 "bootstrap_stacks": bootstrap_stacks,
4640 "strict_deployment_token": strict_deployment_token,
4641 "on_change_set_prepared": on_change_set_prepared,
4642 "on_ecr_repository_created": on_ecr_repository_created,
4643 }
4645 def record_cleanup(name: str, details: dict[str, Any]) -> None:
4646 if on_cleanup_complete is not None:
4647 on_cleanup_complete(name, details)
4649 if not self._image_registry_destroy_preflight(force=force):
4650 return False, [], list(stacks)
4652 bastion_targets = {
4653 name: details for name, details in strict_resources.items() if details.get("vpc_id")
4654 }
4655 bastions = self.cleanup_orphaned_bastions(
4656 stacks,
4657 parallel=parallel,
4658 resource_targets=bastion_targets if strict_identity else None,
4659 )
4660 record_cleanup("bastions", {"terminated_instances": bastions})
4662 # Non-strict teardowns also retire the bastion's standing IAM
4663 # role/profile and, below, the implicit log groups CloudFormation
4664 # never modeled. Strict (live-validation) teardowns skip both: the
4665 # harness owns fenced log-group deletion and audits IAM itself.
4666 if not strict_identity:
4667 record_cleanup("bastion-iam", self._cleanup_bastion_iam())
4669 global_stack_name = f"{project_name}-global"
4670 backup = self._cleanup_backup_vault(
4671 expected_stack_id=(expected_stack_ids or {}).get(global_stack_name),
4672 authorize_stack=authorize_stack,
4673 require_expected_identity=strict_identity,
4674 )
4675 record_cleanup("backup-vault", backup)
4676 if strict_identity and backup.get("errors"):
4677 raise RuntimeError(
4678 "Strict backup-vault cleanup failed before stack deletion: "
4679 + json.dumps(backup["errors"], sort_keys=True)
4680 )
4682 successful: list[str] = []
4683 failed: list[str] = []
4685 # Capture implicit log-group names while the source stacks still
4686 # exist; the exact derived names are deleted by ``finish`` below
4687 # once their stacks are gone. Strict teardowns collect nothing —
4688 # the live-validation harness owns fenced log-group deletion.
4689 implicit_log_groups: dict[str, dict[str, Any]] = {}
4690 if not strict_identity:
4691 implicit_log_groups = self._collect_implicit_log_groups(stacks)
4693 def finish(overall: bool) -> tuple[bool, list[str], list[str]]:
4694 """Funnel every exit through the teardown sweeps.
4696 Called at each return point so a partially failed teardown
4697 still cleans up the stacks that DID delete (implicit log
4698 groups), while the success-only sweeps (runtime traffic-dial
4699 parameters) run exactly when everything is gone. New exit
4700 paths must return through here as well.
4701 """
4702 if implicit_log_groups:
4703 record_cleanup(
4704 "implicit-log-groups",
4705 self._cleanup_implicit_log_groups(implicit_log_groups, successful),
4706 )
4707 if overall:
4708 # Only after a complete teardown: while any stack survives,
4709 # the accelerator may still be live and a manual override on
4710 # it is standing operator intent the purge must not erase.
4711 # Strict teardowns run this too — the runtime dial parameters
4712 # are untagged, so the harness's tagging-index audit cannot
4713 # see them and no one else owns their removal.
4714 record_cleanup(
4715 "traffic-dial-parameters",
4716 self._cleanup_traffic_dial_parameters(),
4717 )
4718 return overall, successful, failed
4720 for stack_name in post_regional_stacks:
4721 if on_stack_start:
4722 on_stack_start(stack_name)
4723 success = self.destroy(
4724 stack_name=stack_name,
4725 force=force,
4726 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4727 **destroy_safety,
4728 )
4729 (successful if success else failed).append(stack_name)
4730 if on_stack_complete:
4731 on_stack_complete(stack_name, success)
4733 phase_remaining = self._destroy_phase_remaining_stacks(
4734 "post-regional",
4735 post_regional_stacks,
4736 expected_stack_ids,
4737 )
4738 for stack_name in phase_remaining:
4739 if stack_name not in failed:
4740 failed.append(stack_name)
4741 if any(stack in failed for stack in post_regional_stacks) or phase_remaining:
4742 return finish(False)
4744 if regional_api_stacks:
4745 if parallel and len(regional_api_stacks) > 1:
4746 successful_api, failed_api = self._destroy_stacks_parallel(
4747 stacks=regional_api_stacks,
4748 force=force,
4749 on_stack_start=on_stack_start,
4750 on_stack_complete=on_stack_complete,
4751 max_workers=max_workers,
4752 expected_stack_ids=expected_stack_ids,
4753 authorize_stack=authorize_stack,
4754 allow_bootstrap=allow_bootstrap,
4755 bootstrap_stacks=bootstrap_stacks,
4756 prepared_change_sets=prepared_change_sets,
4757 strict_deployment_token=strict_deployment_token,
4758 on_change_set_prepared=on_change_set_prepared,
4759 on_ecr_repository_created=on_ecr_repository_created,
4760 )
4761 successful.extend(successful_api)
4762 failed.extend(failed_api)
4763 else:
4764 for stack_name in regional_api_stacks:
4765 if on_stack_start:
4766 on_stack_start(stack_name)
4767 success = self.destroy(
4768 stack_name=stack_name,
4769 force=force,
4770 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4771 **destroy_safety,
4772 )
4773 (successful if success else failed).append(stack_name)
4774 if on_stack_complete:
4775 on_stack_complete(stack_name, success)
4776 phase_remaining = self._destroy_phase_remaining_stacks(
4777 "regional API bridge",
4778 regional_api_stacks,
4779 expected_stack_ids,
4780 )
4781 for stack_name in phase_remaining:
4782 if stack_name not in failed:
4783 failed.append(stack_name)
4784 if any(stack in failed for stack in regional_api_stacks) or phase_remaining:
4785 return finish(False)
4787 watchdog_stops: dict[str, Event] = {}
4788 watchdog_threads: dict[str, Thread] = {}
4789 watchdog_targets = (
4790 [
4791 name
4792 for name in regional_stacks
4793 if strict_resources.get(name, {}).get("cluster_security_group_id")
4794 ]
4795 if strict_identity
4796 else list(regional_stacks)
4797 )
4798 try:
4799 for stack_name in watchdog_targets:
4800 details = strict_resources.get(stack_name, {})
4801 stop_event = Event()
4802 watchdog_stops[stack_name] = stop_event
4803 watchdog_threads[stack_name] = self._start_eks_sg_watchdog(
4804 stack_name,
4805 stop_event,
4806 region=details.get("region"),
4807 security_group_id=details.get("cluster_security_group_id"),
4808 vpc_id=details.get("vpc_id"),
4809 )
4811 if regional_stacks:
4812 if parallel and len(regional_stacks) > 1:
4813 successful_regional, failed_regional = self._destroy_stacks_parallel(
4814 stacks=regional_stacks,
4815 force=force,
4816 on_stack_start=on_stack_start,
4817 on_stack_complete=on_stack_complete,
4818 max_workers=max_workers,
4819 expected_stack_ids=expected_stack_ids,
4820 authorize_stack=authorize_stack,
4821 allow_bootstrap=allow_bootstrap,
4822 bootstrap_stacks=bootstrap_stacks,
4823 prepared_change_sets=prepared_change_sets,
4824 strict_deployment_token=strict_deployment_token,
4825 on_change_set_prepared=on_change_set_prepared,
4826 on_ecr_repository_created=on_ecr_repository_created,
4827 )
4828 successful.extend(successful_regional)
4829 failed.extend(failed_regional)
4830 else:
4831 for stack_name in regional_stacks:
4832 if on_stack_start:
4833 on_stack_start(stack_name)
4834 success = self.destroy(
4835 stack_name=stack_name,
4836 force=force,
4837 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4838 **destroy_safety,
4839 )
4840 (successful if success else failed).append(stack_name)
4841 if on_stack_complete:
4842 on_stack_complete(stack_name, success)
4843 finally:
4844 for stop_event in watchdog_stops.values():
4845 stop_event.set()
4846 for stack_name, thread in watchdog_threads.items():
4847 try:
4848 thread.join(timeout=5)
4849 except Exception as exc:
4850 logger.exception("Could not join teardown watchdog for %s", stack_name)
4851 record_cleanup(
4852 "eks-security-group",
4853 {"stack": stack_name, "errors": [f"{type(exc).__name__}: {exc}"]},
4854 )
4855 if strict_identity and stack_name not in failed:
4856 failed.append(stack_name)
4857 continue
4858 details = strict_resources.get(stack_name, {})
4859 if strict_identity and thread.is_alive():
4860 outcome = {
4861 "stack": stack_name,
4862 "errors": ["watchdog thread did not stop"],
4863 }
4864 else:
4865 outcome = self._cleanup_eks_security_groups(
4866 stack_name,
4867 region=details.get("region"),
4868 security_group_id=details.get("cluster_security_group_id"),
4869 vpc_id=details.get("vpc_id"),
4870 )
4871 record_cleanup("eks-security-group", outcome)
4872 if (
4873 strict_identity
4874 and (outcome.get("errors") or outcome.get("blocked_by_enis"))
4875 and stack_name not in failed
4876 ):
4877 failed.append(stack_name)
4879 phase_remaining = self._destroy_phase_remaining_stacks(
4880 "regional",
4881 regional_stacks,
4882 expected_stack_ids,
4883 )
4884 for stack_name in phase_remaining:
4885 if stack_name not in failed:
4886 failed.append(stack_name)
4887 if any(stack in failed for stack in regional_stacks) or phase_remaining:
4888 return finish(False)
4890 # Every regional stack is verifiably absent here, so each EKS cluster and
4891 # its CSI driver are gone and the volumes it provisioned can never
4892 # reattach. Running after the barrier (rather than inside the per-stack
4893 # loop above) means parallel and sequential teardowns publish the same
4894 # outcomes in the same order, with no concurrent access to the report.
4895 # Cleanup results deliberately do not feed ``failed``: these stacks are
4896 # already deleted, and relabeling one as failed would send the CLI's
4897 # retry loop back to CDK for a stack that no longer exists.
4898 for stack_name in regional_stacks:
4899 record_cleanup(
4900 "dynamic-pvs",
4901 self._cleanup_cluster_volumes(
4902 stack_name,
4903 region=strict_resources.get(stack_name, {}).get("region"),
4904 retain=retain_volumes,
4905 ),
4906 )
4908 for stack_name in pre_regional_stacks:
4909 if on_stack_start:
4910 on_stack_start(stack_name)
4911 success = self.destroy(
4912 stack_name=stack_name,
4913 force=force,
4914 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4915 **destroy_safety,
4916 )
4917 (successful if success else failed).append(stack_name)
4918 if on_stack_complete:
4919 on_stack_complete(stack_name, success)
4921 phase_remaining = self._destroy_phase_remaining_stacks(
4922 "pre-regional global",
4923 [stack_name],
4924 expected_stack_ids,
4925 )
4926 for remaining_stack in phase_remaining:
4927 if remaining_stack not in failed:
4928 failed.append(remaining_stack)
4929 if not success or phase_remaining:
4930 return finish(False)
4932 return finish(len(failed) == 0)
4934 def _destroy_stacks_parallel(
4935 self,
4936 stacks: list[str],
4937 force: bool,
4938 on_stack_start: Callable[[str], None] | None,
4939 on_stack_complete: Callable[[str, bool], None] | None,
4940 max_workers: int,
4941 expected_stack_ids: Mapping[str, str | None] | None,
4942 authorize_stack: StackAuthorizationCallback | None,
4943 allow_bootstrap: bool,
4944 bootstrap_stacks: Mapping[str, Mapping[str, str]] | None,
4945 prepared_change_sets: PreparedChangeSetAuthority | None,
4946 strict_deployment_token: str | None = None,
4947 on_change_set_prepared: ChangeSetPreparedCallback | None = None,
4948 on_ecr_repository_created: EcrRepositoryCreatedCallback | None = None,
4949 ) -> tuple[list[str], list[str]]:
4950 """Destroy multiple stacks in parallel using separate CDK output directories."""
4951 import tempfile
4953 successful: list[str] = []
4954 failed: list[str] = []
4955 lock = Lock()
4957 def destroy_single(stack_name: str) -> tuple[str, bool]:
4958 # Use a unique output directory in /tmp for each parallel destruction
4959 output_dir = tempfile.mkdtemp(prefix=f"cdk-{stack_name}-")
4960 try:
4961 if on_stack_start:
4962 with lock:
4963 on_stack_start(stack_name)
4965 success = self.destroy(
4966 stack_name=stack_name,
4967 force=force,
4968 output_dir=output_dir,
4969 expected_stack_id=(expected_stack_ids or {}).get(stack_name),
4970 expected_stack_ids=expected_stack_ids,
4971 authorize_stack=authorize_stack,
4972 allow_bootstrap=allow_bootstrap,
4973 bootstrap_stacks=bootstrap_stacks,
4974 prepared_change_sets=prepared_change_sets,
4975 strict_deployment_token=strict_deployment_token,
4976 on_change_set_prepared=on_change_set_prepared,
4977 on_ecr_repository_created=on_ecr_repository_created,
4978 )
4979 return stack_name, success
4980 finally:
4981 try:
4982 import shutil
4984 if os.path.exists(output_dir):
4985 shutil.rmtree(output_dir)
4986 except Exception as e:
4987 logger.debug("Cleanup of %s failed: %s", output_dir, e)
4989 self._cdk_cancel_event.clear()
4990 futures: dict[Any, str] = {}
4991 executor = ThreadPoolExecutor(max_workers=max_workers)
4992 try:
4993 futures = {executor.submit(destroy_single, stack): stack for stack in stacks}
4995 for future in as_completed(futures):
4996 stack_name, success = future.result()
4998 with lock:
4999 if success:
5000 successful.append(stack_name)
5001 else:
5002 failed.append(stack_name)
5004 if on_stack_complete:
5005 on_stack_complete(stack_name, success)
5006 except BaseException:
5007 self.cancel_active_cdk_processes()
5008 for future in futures:
5009 future.cancel()
5010 executor.shutdown(wait=True, cancel_futures=True)
5011 raise
5012 else:
5013 executor.shutdown(wait=True)
5014 finally:
5015 self._cdk_cancel_event.clear()
5017 return successful, failed
5019 def _cleanup_backup_vault(
5020 self,
5021 *,
5022 expected_stack_id: str | None = None,
5023 authorize_stack: StackAuthorizationCallback | None = None,
5024 require_expected_identity: bool = False,
5025 ) -> dict[str, Any]:
5026 """Delete points only from the exact stack resource's physical vault."""
5027 import boto3
5029 global_region = self.config.global_region
5030 global_stack_name = f"{self.config.project_name}-global"
5031 result: dict[str, Any] = {
5032 "stack_name": global_stack_name,
5033 "stack_id": expected_stack_id,
5034 "status": "not-needed",
5035 "deleted_recovery_points": 0,
5036 "errors": [],
5037 }
5039 try:
5040 target = self._describe_stack_target(
5041 global_stack_name,
5042 expected_stack_id=expected_stack_id,
5043 require_expected_identity=require_expected_identity,
5044 )
5045 if target is None:
5046 result["status"] = "stack-absent"
5047 return result
5048 region, cloudformation, stack = target
5049 stack_id = str(stack["StackId"])
5050 result["stack_id"] = stack_id
5051 if region != global_region:
5052 raise RuntimeError(f"Global stack resolved to {region}, expected {global_region}")
5053 if authorize_stack is not None:
5054 authorize_stack(global_stack_name, region, stack_id)
5056 resources: list[dict[str, Any]] = []
5057 for page in cloudformation.get_paginator("list_stack_resources").paginate(
5058 StackName=stack_id
5059 ):
5060 resources.extend(
5061 resource
5062 for resource in page.get("StackResourceSummaries", [])
5063 if resource.get("ResourceType") == "AWS::Backup::BackupVault"
5064 and resource.get("PhysicalResourceId")
5065 )
5066 if not resources:
5067 result["status"] = "vault-resource-absent"
5068 return result
5069 if len(resources) != 1:
5070 raise RuntimeError(
5071 f"Expected one AWS::Backup::BackupVault in {stack_id}; found {len(resources)}"
5072 )
5074 resource = resources[0]
5075 physical_id = str(resource["PhysicalResourceId"])
5076 if physical_id.startswith("arn:"):
5077 parts = physical_id.split(":", 5)
5078 if len(parts) != 6 or not parts[5].startswith("backup-vault:"):
5079 raise RuntimeError(f"Invalid backup vault physical ARN: {physical_id}")
5080 vault_name = parts[5].removeprefix("backup-vault:")
5081 else:
5082 vault_name = physical_id
5083 if not vault_name:
5084 raise RuntimeError("CloudFormation returned an empty backup vault physical ID")
5086 backup_client = boto3.client("backup", region_name=global_region)
5087 described_vault = backup_client.describe_backup_vault(BackupVaultName=vault_name)
5088 vault_arn = str(described_vault.get("BackupVaultArn") or "")
5089 arn_parts = vault_arn.split(":", 5)
5090 if (
5091 len(arn_parts) != 6
5092 or arn_parts[2] != "backup"
5093 or arn_parts[3] != global_region
5094 or arn_parts[5] != f"backup-vault:{vault_name}"
5095 ):
5096 raise RuntimeError(
5097 "AWS Backup identity does not match the CloudFormation physical resource"
5098 )
5099 if physical_id.startswith("arn:") and physical_id != vault_arn:
5100 raise RuntimeError("Backup vault ARN changed after CloudFormation resolution")
5102 result.update(
5103 {
5104 "status": "inspected",
5105 "logical_id": str(resource.get("LogicalResourceId") or ""),
5106 "physical_id": physical_id,
5107 "vault_name": vault_name,
5108 "vault_arn": vault_arn,
5109 }
5110 )
5111 paginator = backup_client.get_paginator("list_recovery_points_by_backup_vault")
5112 for page in paginator.paginate(BackupVaultName=vault_name):
5113 for recovery_point in page.get("RecoveryPoints", []):
5114 recovery_point_arn = recovery_point.get("RecoveryPointArn")
5115 if not recovery_point_arn:
5116 continue
5117 try:
5118 backup_client.delete_recovery_point(
5119 BackupVaultName=vault_name,
5120 RecoveryPointArn=recovery_point_arn,
5121 )
5122 result["deleted_recovery_points"] += 1
5123 except Exception as exc:
5124 result["errors"].append(
5125 {
5126 "recovery_point_arn": str(recovery_point_arn),
5127 "error": f"{type(exc).__name__}: {exc}",
5128 }
5129 )
5130 if result["deleted_recovery_points"]:
5131 print(
5132 f" Cleaned up {result['deleted_recovery_points']} backup recovery "
5133 f"points from {vault_name}"
5134 )
5135 result["status"] = "completed" if not result["errors"] else "partial"
5136 except Exception as exc:
5137 result["status"] = "failed"
5138 result["errors"].append({"error": f"{type(exc).__name__}: {exc}"})
5139 print(f" Warning: Backup vault cleanup failed (non-fatal): {exc}")
5140 return result
5142 def cleanup_orphaned_bastions(
5143 self,
5144 stacks: list[str] | None = None,
5145 *,
5146 parallel: bool = True,
5147 resource_targets: Mapping[str, Mapping[str, str]] | None = None,
5148 ) -> int:
5149 """Terminate CLI bastions, using exact stack VPC IDs in strict mode."""
5150 if stacks is None:
5151 stacks = self.list_stacks()
5152 if resource_targets is not None:
5153 regional_stacks = [name for name in stacks if name in resource_targets]
5154 else:
5155 regional_stacks = [
5156 stack
5157 for stack in stacks
5158 if not stack.endswith(("-global", "-api-gateway", "-monitoring", "-analytics"))
5159 ]
5161 def cleanup_one(stack_name: str) -> int:
5162 details = (resource_targets or {}).get(stack_name, {})
5163 return self._cleanup_orphaned_bastions(
5164 stack_name,
5165 region=details.get("region"),
5166 vpc_id=details.get("vpc_id"),
5167 fail_closed=resource_targets is not None,
5168 )
5170 if not regional_stacks:
5171 return 0
5172 if len(regional_stacks) == 1 or not parallel:
5173 terminated = sum(cleanup_one(stack_name) for stack_name in regional_stacks)
5174 else:
5175 # Each region has independent EC2 waiters. Run them concurrently so
5176 # one slow termination does not add its full timeout to every other
5177 # region before CloudFormation can start deleting stacks.
5178 with ThreadPoolExecutor(max_workers=min(4, len(regional_stacks))) as executor:
5179 terminated = sum(executor.map(cleanup_one, regional_stacks))
5180 if terminated:
5181 print(
5182 f" Requested termination for {terminated} orphaned ephemeral "
5183 "SSM bastion(s) before stack deletion."
5184 )
5185 return terminated
5187 def _cleanup_orphaned_bastions(
5188 self,
5189 stack_name: str,
5190 *,
5191 region: str | None = None,
5192 vpc_id: str | None = None,
5193 fail_closed: bool = False,
5194 ) -> int:
5195 """Terminate tagged bastions only inside a resolved stack VPC."""
5196 import boto3
5198 from .ephemeral_bastion import (
5199 BASTION_PURPOSE,
5200 TAG_EPHEMERAL_KEY,
5201 TAG_PROJECT_KEY,
5202 TAG_PURPOSE_KEY,
5203 bastion_instance_name,
5204 )
5206 region = region or self._get_deploy_region(stack_name)
5207 if not region:
5208 if fail_closed:
5209 raise RuntimeError(f"Strict bastion cleanup lacks a Region for {stack_name}")
5210 return 0
5212 project_name = str(self.config.project_name)
5213 expected_name = bastion_instance_name(project_name)
5214 try:
5215 ec2 = boto3.client("ec2", region_name=region)
5216 if vpc_id:
5217 vpcs = [{"VpcId": vpc_id}]
5218 else:
5219 vpcs = ec2.describe_vpcs(
5220 Filters=[
5221 {
5222 "Name": "tag:aws:cloudformation:stack-name",
5223 "Values": [stack_name],
5224 }
5225 ]
5226 ).get("Vpcs", [])
5227 except Exception as exc:
5228 if fail_closed:
5229 raise RuntimeError(
5230 f"Strict bastion cleanup could not inspect {stack_name}"
5231 ) from exc
5232 print(f" Warning: Bastion cleanup could not inspect {stack_name}: {exc}")
5233 return 0
5235 instance_ids: list[str] = []
5236 eni_ids: list[str] = []
5237 for vpc in vpcs:
5238 candidate_vpc_id = str(vpc.get("VpcId") or "")
5239 if not candidate_vpc_id:
5240 if fail_closed:
5241 raise RuntimeError(f"Strict bastion cleanup has no VPC ID for {stack_name}")
5242 continue
5243 try:
5244 reservations = ec2.describe_instances(
5245 Filters=[
5246 {"Name": "vpc-id", "Values": [candidate_vpc_id]},
5247 {"Name": f"tag:{TAG_EPHEMERAL_KEY}", "Values": ["true"]},
5248 {"Name": f"tag:{TAG_PURPOSE_KEY}", "Values": [BASTION_PURPOSE]},
5249 {
5250 "Name": "instance-state-name",
5251 "Values": [
5252 "pending",
5253 "running",
5254 "stopping",
5255 "stopped",
5256 "shutting-down",
5257 ],
5258 },
5259 ]
5260 ).get("Reservations", [])
5261 except Exception as exc:
5262 if fail_closed:
5263 raise RuntimeError(
5264 f"Strict bastion lookup failed in {stack_name} ({candidate_vpc_id})"
5265 ) from exc
5266 logger.warning(
5267 "Bastion lookup failed in %s (%s): %s",
5268 stack_name,
5269 candidate_vpc_id,
5270 exc,
5271 )
5272 continue
5274 for reservation in reservations:
5275 for instance in reservation.get("Instances", []):
5276 tags = {
5277 str(tag.get("Key")): str(tag.get("Value"))
5278 for tag in instance.get("Tags", [])
5279 if tag.get("Key") is not None
5280 }
5281 tagged_project = tags.get(TAG_PROJECT_KEY)
5282 if tagged_project != project_name and not (
5283 tagged_project is None and tags.get("Name") == expected_name
5284 ):
5285 continue
5286 instance_id = instance.get("InstanceId")
5287 if instance_id:
5288 instance_ids.append(str(instance_id))
5289 for interface in instance.get("NetworkInterfaces", []):
5290 attachment = interface.get("Attachment") or {}
5291 if not (
5292 attachment.get("DeviceIndex") == 0
5293 and attachment.get("DeleteOnTermination") is True
5294 ):
5295 continue
5296 eni_id = interface.get("NetworkInterfaceId")
5297 if eni_id:
5298 eni_ids.append(str(eni_id))
5300 instance_ids = list(dict.fromkeys(instance_ids))
5301 eni_ids = list(dict.fromkeys(eni_ids))
5302 if not instance_ids:
5303 return 0
5305 try:
5306 ec2.terminate_instances(InstanceIds=instance_ids)
5307 except Exception as exc:
5308 if fail_closed:
5309 raise RuntimeError(f"Strict bastion termination failed in {stack_name}") from exc
5310 print(f" Warning: Failed to terminate ephemeral bastion(s) in {stack_name}: {exc}")
5311 return 0
5313 print(
5314 f" Terminating {len(instance_ids)} ephemeral SSM bastion(s) in "
5315 f"{stack_name}: {', '.join(instance_ids)}"
5316 )
5317 try:
5318 ec2.get_waiter("instance_terminated").wait(
5319 InstanceIds=instance_ids,
5320 WaiterConfig={"Delay": 5, "MaxAttempts": 60},
5321 )
5322 except Exception as exc:
5323 if fail_closed:
5324 raise RuntimeError(
5325 f"Strict bastion termination did not converge in {stack_name}"
5326 ) from exc
5327 logger.warning("Timed out waiting for bastion termination in %s: %s", stack_name, exc)
5329 remaining_enis = self._wait_for_bastion_network_interfaces(ec2, eni_ids)
5330 if remaining_enis:
5331 message = (
5332 f"{len(remaining_enis)} bastion network interface(s) in {stack_name} "
5333 f"have not released: {', '.join(sorted(remaining_enis))}"
5334 )
5335 if fail_closed:
5336 raise RuntimeError(message)
5337 print(f" Warning: {message}. The destroy retry will check again.")
5338 return len(instance_ids)
5340 @staticmethod
5341 def _wait_for_bastion_network_interfaces(
5342 ec2: Any,
5343 eni_ids: list[str],
5344 *,
5345 timeout_seconds: float = 120.0,
5346 poll_interval_seconds: float = 2.0,
5347 ) -> set[str]:
5348 """Wait for terminated bastion ENIs, deleting detached leftovers.
5350 EC2 normally deletes a primary ENI with its instance. If it becomes
5351 detached instead, it is safe to delete here because its owning instance
5352 was selected by the project/VPC bastion filters and termination has
5353 already been requested.
5354 """
5355 import time as _time
5357 remaining = set(eni_ids)
5358 deadline = _time.monotonic() + timeout_seconds
5359 while remaining:
5360 for eni_id in tuple(remaining):
5361 try:
5362 response = ec2.describe_network_interfaces(NetworkInterfaceIds=[eni_id])
5363 except ClientError as exc:
5364 code = exc.response.get("Error", {}).get("Code")
5365 if code == "InvalidNetworkInterfaceID.NotFound":
5366 remaining.discard(eni_id)
5367 continue
5368 logger.warning("Could not inspect bastion ENI %s: %s", eni_id, exc)
5369 return remaining
5370 except Exception as exc: # noqa: BLE001 - cleanup is best-effort
5371 logger.warning("Could not inspect bastion ENI %s: %s", eni_id, exc)
5372 return remaining
5374 interfaces = response.get("NetworkInterfaces", [])
5375 if not interfaces:
5376 remaining.discard(eni_id)
5377 continue
5378 if interfaces[0].get("Status") == "available":
5379 try:
5380 ec2.delete_network_interface(NetworkInterfaceId=eni_id)
5381 remaining.discard(eni_id)
5382 except ClientError as exc:
5383 code = exc.response.get("Error", {}).get("Code")
5384 if code == "InvalidNetworkInterfaceID.NotFound":
5385 remaining.discard(eni_id)
5386 else:
5387 logger.debug("Delete of bastion ENI %s failed: %s", eni_id, exc)
5388 except Exception as exc: # noqa: BLE001 - retry until timeout
5389 logger.debug("Delete of bastion ENI %s failed: %s", eni_id, exc)
5391 if not remaining or _time.monotonic() >= deadline:
5392 break
5393 _time.sleep(poll_interval_seconds)
5394 return remaining
5396 # ------------------------------------------------------------------
5397 # Implicit log-group + bastion IAM cleanup (non-strict destroy only)
5398 # ------------------------------------------------------------------
5399 #
5400 # CloudFormation only deletes the log groups it modeled. Lambda default
5401 # groups (``/aws/lambda/<function>``), the EKS control-plane group
5402 # (``/aws/eks/<cluster>/cluster``), and the Container Insights groups
5403 # (``/aws/containerinsights/<cluster>/…``) are created out-of-band by
5404 # the services themselves, so ``destroy-all`` used to report success
5405 # while leaving them behind — a real teardown orphaned 22 of them plus
5406 # the ephemeral-bastion IAM role/profile, which then failed the live
5407 # release validation's clean-account baseline gate.
5408 #
5409 # The cleanup below deletes ONLY exact names derived from the project's
5410 # own stack resources, captured while the stacks still exist, and only
5411 # for stacks whose deletion actually succeeded. It never runs in strict
5412 # (live-validation) teardowns: the harness checkpoints, tags, and
5413 # fences its own log-group generations and must remain the single
5414 # owner of that deletion authority.
5416 # The service-side patterns implicit log groups follow. An explicit
5417 # ``AWS::Logs::LogGroup`` resource is deliberately absent here —
5418 # CloudFormation owns those directly.
5419 _EKS_CONTAINER_INSIGHTS_SUFFIXES = ("application", "dataplane", "host", "performance")
5421 @staticmethod
5422 def _implicit_log_group_names(resource_type: str, physical_id: str) -> tuple[str, ...]:
5423 """Exact implicit log-group names a stack resource creates out-of-band."""
5424 if resource_type == "AWS::Lambda::Function":
5425 return (f"/aws/lambda/{physical_id}",)
5426 if resource_type == "AWS::EKS::Cluster":
5427 return (
5428 f"/aws/eks/{physical_id}/cluster",
5429 *(
5430 f"/aws/containerinsights/{physical_id}/{suffix}"
5431 for suffix in StackManager._EKS_CONTAINER_INSIGHTS_SUFFIXES
5432 ),
5433 )
5434 return ()
5436 def _collect_implicit_log_groups(self, stacks: Collection[str]) -> dict[str, dict[str, Any]]:
5437 """Derive per-stack implicit log-group names while the stacks are live.
5439 Best-effort: a stack that cannot be described or listed is skipped
5440 with a warning — collection must never block the destroy itself.
5441 """
5442 collected: dict[str, dict[str, Any]] = {}
5443 for stack_name in stacks:
5444 try:
5445 target = self._describe_stack_target(stack_name)
5446 if target is None:
5447 continue
5448 region, cloudformation, stack = target
5449 names: list[str] = []
5450 paginator = cloudformation.get_paginator("list_stack_resources")
5451 for page in paginator.paginate(StackName=str(stack["StackId"])):
5452 for item in page.get("StackResourceSummaries", []):
5453 resource_type = str(item.get("ResourceType") or "")
5454 physical_id = str(item.get("PhysicalResourceId") or "")
5455 if not physical_id:
5456 continue
5457 names.extend(self._implicit_log_group_names(resource_type, physical_id))
5458 if names:
5459 collected[stack_name] = {"region": region, "log_groups": sorted(set(names))}
5460 except Exception as exc: # noqa: BLE001 - best-effort collection
5461 logger.warning("Could not derive implicit log groups for %s: %s", stack_name, exc)
5462 return collected
5464 def _cleanup_implicit_log_groups(
5465 self,
5466 collected: Mapping[str, Mapping[str, Any]],
5467 successful_stacks: Collection[str],
5468 ) -> dict[str, Any]:
5469 """Delete the exact derived log groups of successfully destroyed stacks.
5471 A missing group is normal (a Lambda that never logged, or a custom
5472 ``LoggingConfig`` pointing elsewhere) and is recorded, not retried.
5473 Every error is recorded and swallowed: cleanup never converts a
5474 successful destroy into a failure.
5475 """
5476 import boto3
5478 outcome: dict[str, Any] = {"deleted": [], "missing": [], "errors": []}
5479 clients: dict[str, Any] = {}
5480 for stack_name in sorted(successful_stacks):
5481 details = collected.get(stack_name)
5482 if not details:
5483 continue
5484 region = str(details.get("region") or "")
5485 for name in details.get("log_groups", []):
5486 try:
5487 client = clients.get(region)
5488 if client is None:
5489 client = boto3.client("logs", region_name=region)
5490 clients[region] = client
5491 client.delete_log_group(logGroupName=name)
5492 outcome["deleted"].append(f"{region}:{name}")
5493 except ClientError as exc:
5494 code = str(exc.response.get("Error", {}).get("Code") or "")
5495 if code == "ResourceNotFoundException":
5496 outcome["missing"].append(f"{region}:{name}")
5497 else:
5498 outcome["errors"].append(f"{region}:{name}: {code}")
5499 except Exception as exc: # noqa: BLE001 - best-effort cleanup
5500 outcome["errors"].append(f"{region}:{name}: {type(exc).__name__}: {exc}")
5501 if outcome["deleted"]:
5502 print(
5503 f" Deleted {len(outcome['deleted'])} implicit CloudWatch log group(s) "
5504 "left behind by Lambda/EKS/Container Insights."
5505 )
5506 for failure in outcome["errors"]:
5507 logger.warning("Implicit log-group cleanup failed for %s", failure)
5508 return outcome
5510 def _cleanup_traffic_dial_parameters(self) -> dict[str, Any]:
5511 """Best-effort purge of the runtime traffic-dial SSM parameter tree.
5513 The traffic-dial controller Lambda writes ``/{project}/traffic-dial/
5514 state`` and ``gco capacity traffic-dial set`` writes ``override-*``
5515 siblings at runtime; CloudFormation never owns them, so stack
5516 deletion leaves them behind. A surviving override is the real
5517 hazard: the scheduled controller honors overrides indefinitely, so
5518 the next deployment in this account would silently pin that region's
5519 dial until someone noticed. Runs only after a *fully* successful
5520 teardown — while any stack remains, the accelerator may still be
5521 live and an override on it is standing operator intent.
5522 """
5523 from .capacity.traffic_dial import TrafficDialManager
5525 outcome: dict[str, Any] = {"deleted": [], "errors": []}
5526 try:
5527 outcome["deleted"] = TrafficDialManager(self.config).purge_runtime_parameters()
5528 except Exception as exc: # noqa: BLE001 - best-effort cleanup
5529 outcome["errors"].append(f"{type(exc).__name__}: {exc}")
5530 if outcome["deleted"]:
5531 print(
5532 f" Deleted {len(outcome['deleted'])} runtime traffic-dial SSM "
5533 "parameter(s) (controller state / manual overrides)."
5534 )
5535 for failure in outcome["errors"]:
5536 logger.warning("Traffic-dial parameter cleanup failed: %s", failure)
5537 return outcome
5539 def _cleanup_bastion_iam(self) -> dict[str, Any]:
5540 """Best-effort teardown of the ephemeral-bastion IAM role + profile.
5542 ``destroy_ephemeral_bastion`` already attempts this when a tunnel
5543 closes normally, but a killed process leaves the pair behind (they
5544 cost nothing, yet fail any clean-account audit). Deletion is by the
5545 exact project-scoped names from the bastion naming contract; a
5546 ``NoSuchEntity`` response simply means there was nothing to clean.
5547 """
5548 from .ephemeral_bastion import (
5549 _run_aws,
5550 bastion_profile_name,
5551 bastion_role_name,
5552 build_iam_teardown_commands,
5553 )
5555 outcome: dict[str, Any] = {
5556 "completed_steps": 0,
5557 "absent_steps": 0,
5558 "errors": [],
5559 }
5560 try:
5561 role_name = bastion_role_name(self.config.project_name)
5562 profile_name = bastion_profile_name(self.config.project_name)
5563 outcome["role"] = role_name
5564 outcome["profile"] = profile_name
5565 steps = build_iam_teardown_commands(
5566 role_name,
5567 profile_name,
5568 self.config.global_region,
5569 )
5570 except Exception as exc: # noqa: BLE001 - best-effort cleanup
5571 outcome["errors"].append(f"{type(exc).__name__}: {exc}")
5572 return outcome
5573 for step in steps:
5574 try:
5575 _run_aws(step)
5576 outcome["completed_steps"] += 1
5577 except RuntimeError as exc:
5578 if "NoSuchEntity" in str(exc):
5579 outcome["absent_steps"] += 1
5580 continue
5581 outcome["errors"].append(f"{' '.join(step[:3])}: {exc}")
5582 if outcome["completed_steps"] and not outcome["errors"]:
5583 print(" Removed the ephemeral-bastion IAM role and instance profile.")
5584 for failure in outcome["errors"]:
5585 logger.warning("Bastion IAM teardown step failed: %s", failure)
5586 return outcome
5588 def cleanup_eks_security_groups(self) -> None:
5589 """Clean up EKS-managed security groups across all regional stacks.
5591 Called between destroy retries to remove orphaned security groups
5592 that block VPC deletion.
5593 """
5594 stacks = self.list_stacks()
5595 # Regional stacks are everything that isn't a named global stack;
5596 # classify by suffix so this works for any project_name (#139).
5597 regional_stacks = [
5598 s for s in stacks if not s.endswith(("-global", "-api-gateway", "-monitoring"))
5599 ]
5600 for stack_name in regional_stacks:
5601 self._cleanup_eks_security_groups(stack_name)
5603 def cleanup_orphaned_network_interfaces(self) -> None:
5604 """Report and clear resources that can block VPC deletion, across all
5605 regional stacks. Run between destroy retries.
5607 Generalizes ``cleanup_eks_security_groups`` (which force-deletes the
5608 ``eks-cluster-sg-*`` security group + its ENIs that EKS leaves behind)
5609 with a broader sweep: for each regional stack's VPC it enumerates every
5610 remaining network interface, categorizes them (Global Accelerator / ELB
5611 / EKS / other), deletes the ones that are safe to remove (detached and
5612 not service-managed), and prints a friendly summary of what it found and
5613 what the next retry is waiting on. Service-managed ENIs (Global
5614 Accelerator, ELB) are released asynchronously by AWS once the endpoint /
5615 load balancer is gone, so we report them rather than fight them.
5616 """
5617 stacks = self.list_stacks()
5618 regional_stacks = [
5619 s for s in stacks if not s.endswith(("-global", "-api-gateway", "-monitoring"))
5620 ]
5621 for stack_name in regional_stacks:
5622 # Existing behaviour first: clear the EKS cluster SG + its ENIs.
5623 self._cleanup_eks_security_groups(stack_name)
5624 # Then report (and safely clear) anything else lingering in the VPC.
5625 summary = self._summarize_orphaned_enis(stack_name)
5626 self._print_orphaned_eni_summary(stack_name, summary)
5628 @staticmethod
5629 def _classify_orphaned_eni(eni: dict[str, Any]) -> str:
5630 """Bucket a network interface by which AWS service owns it.
5632 Uses ``InterfaceType`` first (authoritative for Global Accelerator and
5633 the load-balancer types) and falls back to the human ``Description``
5634 string for the EKS / ELB cases that present as a plain ``interface``.
5635 Returns one of ``global_accelerator`` / ``elb`` / ``eks`` / ``other``.
5636 """
5637 itype = str(eni.get("InterfaceType") or "").lower()
5638 desc = str(eni.get("Description") or "").lower()
5639 if (
5640 itype == "global_accelerator_managed"
5641 or "global_accelerator" in desc
5642 or "global accelerator" in desc
5643 ):
5644 return "global_accelerator"
5645 if itype in ("load_balancer", "network_load_balancer") or desc.startswith("elb "):
5646 return "elb"
5647 if "eks" in desc or "k8s" in desc or "kubernetes" in desc:
5648 return "eks"
5649 return "other"
5651 def _summarize_orphaned_enis(self, stack_name: str) -> dict[str, int]:
5652 """Inspect the stack's VPC(s) for lingering ENIs, categorize them, and
5653 best-effort delete the ones that are safe to remove.
5655 "Safe to remove" means ``Status == "available"`` (detached) and not
5656 ``RequesterManaged`` (i.e. not owned by a service like GA / ELB, which
5657 rejects manual deletion and releases the ENI on its own schedule).
5659 Returns a dict of counts: per-category totals plus ``deleted`` and
5660 ``vpcs``. Wholly best-effort — any AWS error degrades to the counts
5661 gathered so far rather than raising into the destroy flow.
5662 """
5663 import boto3
5665 region = stack_name.replace(f"{self.config.project_name}-", "", 1)
5666 summary: dict[str, int] = {
5667 "global_accelerator": 0,
5668 "elb": 0,
5669 "eks": 0,
5670 "other": 0,
5671 "deleted": 0,
5672 "vpcs": 0,
5673 }
5674 try:
5675 ec2 = boto3.client("ec2", region_name=region)
5676 vpcs = ec2.describe_vpcs(
5677 Filters=[{"Name": "tag:aws:cloudformation:stack-name", "Values": [stack_name]}]
5678 ).get("Vpcs", [])
5679 except Exception as e: # noqa: BLE001
5680 logger.debug("ENI sweep: VPC lookup failed for %s: %s", stack_name, e)
5681 return summary
5683 for vpc in vpcs:
5684 summary["vpcs"] += 1
5685 vpc_id = vpc.get("VpcId")
5686 try:
5687 enis = ec2.describe_network_interfaces(
5688 Filters=[{"Name": "vpc-id", "Values": [vpc_id]}]
5689 ).get("NetworkInterfaces", [])
5690 except Exception as e: # noqa: BLE001
5691 logger.debug("ENI sweep: describe ENIs failed for %s: %s", vpc_id, e)
5692 continue
5694 for eni in enis:
5695 summary[self._classify_orphaned_eni(eni)] += 1
5696 detached = eni.get("Status") == "available"
5697 service_managed = bool(eni.get("RequesterManaged", False))
5698 if detached and not service_managed:
5699 eni_id = eni.get("NetworkInterfaceId")
5700 try:
5701 ec2.delete_network_interface(NetworkInterfaceId=eni_id)
5702 summary["deleted"] += 1
5703 logger.debug("ENI sweep: deleted detached ENI %s in %s", eni_id, vpc_id)
5704 except Exception as e: # noqa: BLE001
5705 logger.debug("ENI sweep: delete of %s failed: %s", eni_id, e)
5706 return summary
5708 @staticmethod
5709 def _print_orphaned_eni_summary(stack_name: str, summary: dict[str, int]) -> None:
5710 """Print a friendly summary of what the ENI sweep found and handled."""
5711 categories = (
5712 ("global_accelerator", "Global Accelerator-managed"),
5713 ("elb", "ELB-managed"),
5714 ("eks", "EKS-managed"),
5715 ("other", "other"),
5716 )
5717 total = sum(summary.get(key, 0) for key, _ in categories)
5718 if total == 0:
5719 return
5720 breakdown = ", ".join(
5721 f"{summary[key]} {label}" for key, label in categories if summary.get(key)
5722 )
5723 print(f" {stack_name}: {total} network interface(s) still in the VPC ({breakdown}).")
5724 if summary.get("deleted"):
5725 print(f" Removed {summary['deleted']} detached interface(s).")
5726 remaining = total - summary.get("deleted", 0)
5727 if remaining > 0:
5728 print(
5729 f" {remaining} still held by AWS — Global Accelerator / ELB release these "
5730 "asynchronously once the endpoint and load balancer are gone; the next retry "
5731 "proceeds once they drain."
5732 )
5734 def _cleanup_eks_security_groups(
5735 self,
5736 stack_name: str,
5737 *,
5738 region: str | None = None,
5739 security_group_id: str | None = None,
5740 vpc_id: str | None = None,
5741 ) -> dict[str, Any]:
5742 """Delete empty EKS SGs, optionally by one exact preauthorized ID."""
5743 import boto3
5745 project_name = self.config.project_name
5746 region = region or stack_name.replace(f"{project_name}-", "", 1)
5747 cluster_name = stack_name
5748 outcome: dict[str, Any] = {
5749 "stack": stack_name,
5750 "region": region,
5751 "security_group_id": security_group_id,
5752 "inspected": 0,
5753 "deleted": [],
5754 "blocked_by_enis": [],
5755 "errors": [],
5756 }
5758 try:
5759 ec2 = boto3.client("ec2", region_name=region)
5760 try:
5761 if security_group_id:
5762 response = ec2.describe_security_groups(GroupIds=[security_group_id])
5763 else:
5764 response = ec2.describe_security_groups(
5765 Filters=[
5766 {
5767 "Name": "group-name",
5768 "Values": [f"eks-cluster-sg-{cluster_name}-*"],
5769 }
5770 ]
5771 )
5772 except ClientError as exc:
5773 if (
5774 security_group_id
5775 and exc.response.get("Error", {}).get("Code") == "InvalidGroup.NotFound"
5776 ):
5777 outcome["absent"] = True
5778 return outcome
5779 raise
5781 for security_group in response.get("SecurityGroups", []):
5782 outcome["inspected"] += 1
5783 group_id = str(security_group["GroupId"])
5784 group_name = str(security_group.get("GroupName", ""))
5785 if security_group_id and group_id != security_group_id:
5786 raise RuntimeError(
5787 f"EC2 returned changed security-group identity for {security_group_id}"
5788 )
5789 if vpc_id and str(security_group.get("VpcId") or "") != vpc_id:
5790 raise RuntimeError(
5791 f"Security group {group_id} no longer belongs to exact VPC {vpc_id}"
5792 )
5793 interfaces = ec2.describe_network_interfaces(
5794 Filters=[{"Name": "group-id", "Values": [group_id]}]
5795 ).get("NetworkInterfaces", [])
5796 if interfaces:
5797 outcome["blocked_by_enis"].append(
5798 {
5799 "group_id": group_id,
5800 "group_name": group_name,
5801 "network_interface_ids": sorted(
5802 str(interface.get("NetworkInterfaceId") or "")
5803 for interface in interfaces
5804 if interface.get("NetworkInterfaceId")
5805 ),
5806 }
5807 )
5808 logger.debug(
5809 "Waiting for AWS to release %d EKS-managed ENI(s) from %s",
5810 len(interfaces),
5811 group_name,
5812 )
5813 continue
5814 try:
5815 ec2.delete_security_group(GroupId=group_id)
5816 outcome["deleted"].append({"group_id": group_id, "group_name": group_name})
5817 print(f" Cleaned up empty EKS security group: {group_name} ({group_id})")
5818 except ClientError as exc:
5819 if exc.response.get("Error", {}).get("Code") == "InvalidGroup.NotFound":
5820 outcome["absent"] = True
5821 continue
5822 outcome["errors"].append(
5823 {"group_id": group_id, "error": f"{type(exc).__name__}: {exc}"}
5824 )
5825 except Exception as exc:
5826 outcome["errors"].append(
5827 {"group_id": group_id, "error": f"{type(exc).__name__}: {exc}"}
5828 )
5829 except Exception as exc:
5830 outcome["errors"].append({"error": f"{type(exc).__name__}: {exc}"})
5831 logger.debug("EKS security group cleanup for %s failed: %s", stack_name, exc)
5832 return outcome
5834 def cleanup_cluster_volumes(
5835 self,
5836 stack_name: str,
5837 *,
5838 region: str | None = None,
5839 retain: bool = False,
5840 ) -> dict[str, Any]:
5841 """Sweep one regional stack's orphaned CSI volumes; no-op for global stacks.
5843 Entry point for the single-stack ``gco stacks destroy`` path, which has no
5844 orchestrated cleanup barrier of its own. Global stacks host no cluster, so
5845 they resolve to no work rather than a derived pseudo-Region.
5846 """
5847 if stack_name.endswith(("-global", "-api-gateway", "-monitoring")):
5848 return {"stack": stack_name, "skipped": "not-a-regional-stack"}
5849 return self._cleanup_cluster_volumes(stack_name, region=region, retain=retain)
5851 def _cleanup_cluster_volumes(
5852 self,
5853 stack_name: str,
5854 *,
5855 region: str | None = None,
5856 retain: bool = False,
5857 ) -> dict[str, Any]:
5858 """Delete the EBS volumes a destroyed cluster's CSI driver left behind.
5860 Deleting an EKS cluster does not delete the PersistentVolumes its EBS CSI
5861 driver provisioned, so every deploy/destroy cycle strands ``available``
5862 volumes tagged ``kubernetes.io/cluster/<cluster>`` for a cluster that no
5863 longer exists. Nothing can reattach them and they bill indefinitely (#268).
5864 They carry the CSI driver's tags rather than the CDK ``Project`` tag, so no
5865 project-scoped sweep or cost query can see them.
5867 Deletion is the default because it honors intent already declared
5868 elsewhere: the ``gco-observability-gp3`` StorageClass sets
5869 ``reclaimPolicy: Delete``, and these volumes survive only because the
5870 cluster is torn down before its PVCs are, so the CSI driver never receives
5871 the delete event. ``retain=True`` reports them instead, and either way
5872 every volume is named in the outcome — a silent leak is the actual bug.
5874 Fail-closed on ordering: the sweep first proves the cluster is absent, so a
5875 still-reconciling CSI driver is never raced. Ownership, ``available``
5876 state, and zero attachments are then rechecked immediately before each
5877 delete rather than trusted from the discovery snapshot. One volume's
5878 failure never stops the others, and nothing raises into the destroy flow.
5879 """
5880 import boto3
5882 project_name = self.config.project_name
5883 region = region or stack_name.replace(f"{project_name}-", "", 1)
5884 cluster_name = stack_name
5885 cluster_tag = f"kubernetes.io/cluster/{cluster_name}"
5886 outcome: dict[str, Any] = {
5887 "stack": stack_name,
5888 "region": region,
5889 "cluster": cluster_name,
5890 "retained": retain,
5891 "inspected": 0,
5892 "deleted": [],
5893 "surviving": [],
5894 "errors": [],
5895 }
5897 try:
5898 # Ordering gate: a live cluster means its CSI driver may still be
5899 # reconciling, so a detached volume can simply be between pod
5900 # restarts. Only a proven-absent cluster makes these volumes garbage.
5901 try:
5902 boto3.client("eks", region_name=region).describe_cluster(name=cluster_name)
5903 except ClientError as exc:
5904 if exc.response.get("Error", {}).get("Code") != "ResourceNotFoundException":
5905 raise
5906 else:
5907 outcome["cluster_present"] = True
5908 logger.debug(
5909 "Skipping volume cleanup for %s: cluster is still present",
5910 cluster_name,
5911 )
5912 return outcome
5914 ec2 = boto3.client("ec2", region_name=region)
5915 volumes: list[dict[str, Any]] = []
5916 for page in ec2.get_paginator("describe_volumes").paginate(
5917 Filters=[
5918 {"Name": "tag-key", "Values": [cluster_tag]},
5919 {"Name": "status", "Values": ["available"]},
5920 ]
5921 ):
5922 volumes.extend(page.get("Volumes", []))
5924 for volume in volumes:
5925 outcome["inspected"] += 1
5926 volume_id = str(volume["VolumeId"])
5927 record = {
5928 "volume_id": volume_id,
5929 "size_gib": volume.get("Size"),
5930 "volume_type": volume.get("VolumeType"),
5931 "availability_zone": volume.get("AvailabilityZone"),
5932 "pvc": _volume_pvc_name(volume),
5933 }
5934 if retain:
5935 outcome["surviving"].append({**record, "reason": "retained-by-request"})
5936 continue
5937 blocked = self._volume_delete_blocked(ec2, volume_id, cluster_tag=cluster_tag)
5938 if blocked is not None:
5939 outcome["surviving"].append({**record, "reason": blocked})
5940 logger.debug("Leaving volume %s in place: %s", volume_id, blocked)
5941 continue
5942 try:
5943 ec2.delete_volume(VolumeId=volume_id)
5944 outcome["deleted"].append(record)
5945 except ClientError as exc:
5946 if exc.response.get("Error", {}).get("Code") == "InvalidVolume.NotFound":
5947 outcome["absent"] = True
5948 continue
5949 outcome["errors"].append(
5950 {"volume_id": volume_id, "error": f"{type(exc).__name__}: {exc}"}
5951 )
5952 except Exception as exc:
5953 outcome["errors"].append(
5954 {"volume_id": volume_id, "error": f"{type(exc).__name__}: {exc}"}
5955 )
5956 except Exception as exc:
5957 outcome["errors"].append({"error": f"{type(exc).__name__}: {exc}"})
5958 logger.debug("Cluster volume cleanup for %s failed: %s", stack_name, exc)
5959 self._price_surviving_volumes(outcome)
5960 _print_cluster_volume_outcome(outcome)
5961 return outcome
5963 @staticmethod
5964 def _volume_storage_price_per_gib_month(region: str, volume_type: str) -> float | None:
5965 """Return the current on-demand $/GiB-month for a volume type in a Region.
5967 Priced at teardown time against the target Region rather than from a
5968 constant in this file: EBS rates differ per Region and change over time, so
5969 a checked-in number would quietly drift into misinforming the operator.
5971 Returns ``None`` whenever the real rate cannot be established — no
5972 credentials for ``pricing:GetProducts``, an unroutable endpoint, an
5973 emulator that does not implement the Price List API, or an unrecognized
5974 response shape. Callers must say the cost is unknown rather than
5975 substitute a guess. Timeouts are short and retries few because this is a
5976 cosmetic annotation on the teardown path and must never hold it up.
5977 """
5978 try:
5979 import boto3
5980 from botocore.config import Config
5982 # The Price List API is only offered in a few Regions; us-east-1 is
5983 # the canonical endpoint and is what cli/capacity/checker.py uses.
5984 # The Region being priced is a filter, not the endpoint.
5985 pricing = boto3.client(
5986 "pricing",
5987 region_name="us-east-1",
5988 config=Config(
5989 connect_timeout=3,
5990 read_timeout=5,
5991 retries={"max_attempts": 2},
5992 ),
5993 )
5994 response = pricing.get_products(
5995 ServiceCode="AmazonEC2",
5996 Filters=[
5997 {"Type": "TERM_MATCH", "Field": "productFamily", "Value": "Storage"},
5998 {"Type": "TERM_MATCH", "Field": "volumeApiName", "Value": volume_type},
5999 {"Type": "TERM_MATCH", "Field": "regionCode", "Value": region},
6000 ],
6001 MaxResults=1,
6002 )
6003 for entry in response.get("PriceList") or []:
6004 product = json.loads(entry)
6005 for term in (product.get("terms") or {}).get("OnDemand", {}).values():
6006 for dimension in (term.get("priceDimensions") or {}).values():
6007 usd = (dimension.get("pricePerUnit") or {}).get("USD")
6008 if usd is not None:
6009 return float(usd)
6010 except Exception as exc:
6011 logger.debug(
6012 "Could not price %s storage in %s: %s",
6013 volume_type,
6014 region,
6015 exc,
6016 )
6017 return None
6019 def _price_surviving_volumes(self, outcome: dict[str, Any]) -> None:
6020 """Annotate an outcome with the monthly cost of the volumes left behind.
6022 Sets ``monthly_cost_usd`` when every surviving volume's type could be
6023 priced, and ``monthly_cost_unavailable`` with the reason otherwise. Only
6024 runs when volumes actually survived, so a teardown that cleaned up
6025 completely makes no pricing call at all.
6026 """
6027 surviving = list(outcome.get("surviving") or [])
6028 if not surviving:
6029 return
6030 region = str(outcome.get("region") or "")
6031 rates: dict[str, float | None] = {}
6032 for record in surviving:
6033 volume_type = str(record.get("volume_type") or "")
6034 if volume_type and volume_type not in rates:
6035 rates[volume_type] = self._volume_storage_price_per_gib_month(region, volume_type)
6037 unpriced = sorted({name for name, rate in rates.items() if rate is None})
6038 if not rates or unpriced:
6039 outcome["monthly_cost_unavailable"] = (
6040 "could not retrieve current EBS pricing for "
6041 + (", ".join(unpriced) if unpriced else "these volumes")
6042 + f" in {region}"
6043 )
6044 return
6045 outcome["monthly_cost_usd"] = round(
6046 sum(
6047 int(record.get("size_gib") or 0)
6048 * (rates.get(str(record.get("volume_type"))) or 0.0)
6049 for record in surviving
6050 ),
6051 2,
6052 )
6054 @staticmethod
6055 def _volume_delete_blocked(
6056 ec2: Any,
6057 volume_id: str,
6058 *,
6059 cluster_tag: str,
6060 ) -> str | None:
6061 """Return why ``volume_id`` must not be deleted, or None when it may be.
6063 Re-reads the volume immediately before deletion so a volume that was
6064 reattached, or whose ownership tag changed, since discovery is left alone.
6065 An unreadable volume is reported as blocked: acting without confirmation
6066 is worse than leaving one volume behind for the operator to see.
6067 """
6068 try:
6069 described = ec2.describe_volumes(VolumeIds=[volume_id]).get("Volumes", [])
6070 except ClientError as exc:
6071 if exc.response.get("Error", {}).get("Code") == "InvalidVolume.NotFound":
6072 return "already-absent"
6073 return f"recheck-failed: {exc.response.get('Error', {}).get('Code') or 'ClientError'}"
6074 except Exception as exc:
6075 return f"recheck-failed: {type(exc).__name__}"
6076 if len(described) != 1:
6077 return "recheck-returned-ambiguous-identity"
6078 current = described[0]
6079 if str(current.get("VolumeId") or "") != volume_id:
6080 return "recheck-returned-changed-identity"
6081 if str(current.get("State") or "") != "available":
6082 return f"state-is-{current.get('State') or 'unknown'}"
6083 if current.get("Attachments"):
6084 return "volume-has-attachments"
6085 if not any(str(tag.get("Key") or "") == cluster_tag for tag in current.get("Tags") or []):
6086 return "cluster-ownership-tag-absent"
6087 return None
6089 def _start_eks_sg_watchdog(
6090 self,
6091 stack_name: str,
6092 stop_event: Event,
6093 *,
6094 region: str | None = None,
6095 security_group_id: str | None = None,
6096 vpc_id: str | None = None,
6097 ) -> Thread:
6098 """Start a background thread that polls for orphaned EKS security groups.
6100 EKS creates an ``eks-cluster-sg-<cluster-name>-*`` security group that
6101 is owned by the EKS service (not CloudFormation). The watchdog observes
6102 it throughout regional teardown and removes it only after AWS has
6103 released every attached ENI. Service-managed interfaces are never
6104 detached or deleted by the CLI.
6106 The thread exits when ``stop_event`` is set by the orchestrator at
6107 the end of the regional phase.
6108 """
6110 def _watchdog() -> None:
6111 while not stop_event.is_set():
6112 try:
6113 self._cleanup_eks_security_groups(
6114 stack_name,
6115 region=region,
6116 security_group_id=security_group_id,
6117 vpc_id=vpc_id,
6118 )
6119 except Exception as e:
6120 logger.debug(
6121 "EKS SG watchdog tick for %s failed (non-fatal): %s",
6122 stack_name,
6123 e,
6124 )
6125 # ``wait`` returns immediately when the event is set, so this
6126 # doubles as the sleep-and-shutdown-check in one call.
6127 stop_event.wait(timeout=30)
6129 thread = Thread(
6130 target=_watchdog,
6131 name=f"eks-sg-watchdog-{stack_name}",
6132 daemon=True,
6133 )
6134 thread.start()
6135 return thread
6138def get_stack_manager(config: GCOConfig) -> StackManager:
6139 """Factory function to get a StackManager instance."""
6140 return StackManager(config)
6143def _volume_pvc_name(volume: Mapping[str, Any]) -> str | None:
6144 """Return the PVC name the CSI driver recorded on a volume, when present."""
6145 for tag in volume.get("Tags") or []:
6146 if str(tag.get("Key") or "") == "kubernetes.io/created-for/pvc/name":
6147 return str(tag.get("Value") or "") or None
6148 return None
6151def _describe_volume_record(record: Mapping[str, Any]) -> str:
6152 """Render one volume as ``vol-x (50 GiB gp3, us-west-2a, pvc=prometheus-db)``."""
6153 size = f"{record['size_gib']} GiB" if record.get("size_gib") else "unknown size"
6154 if record.get("volume_type"):
6155 size = f"{size} {record['volume_type']}"
6156 parts = [size]
6157 if record.get("availability_zone"):
6158 parts.append(str(record["availability_zone"]))
6159 if record.get("pvc"):
6160 parts.append(f"pvc={record['pvc']}")
6161 return f"{record['volume_id']} ({', '.join(parts)})"
6164def _print_cluster_volume_outcome(outcome: Mapping[str, Any]) -> None:
6165 """Report what a cluster-volume sweep deleted, left behind, or could not do.
6167 Silent retention is the defect this feature exists to fix (#268), so every
6168 surviving volume is named on stdout — not only the deleted ones.
6169 """
6170 deleted = list(outcome.get("deleted") or [])
6171 surviving = list(outcome.get("surviving") or [])
6172 errors = list(outcome.get("errors") or [])
6173 cluster = outcome.get("cluster")
6175 if deleted:
6176 total_gib = sum(int(record.get("size_gib") or 0) for record in deleted)
6177 print(
6178 f" Cleaned up {len(deleted)} orphaned EBS volume(s) "
6179 f"({total_gib} GiB) left by cluster {cluster}:"
6180 )
6181 for record in deleted:
6182 print(f" - {_describe_volume_record(record)}")
6184 if surviving:
6185 total_gib = sum(int(record.get("size_gib") or 0) for record in surviving)
6186 cost = outcome.get("monthly_cost_usd")
6187 if cost is not None:
6188 billing = f", ${cost:.2f}/month at current {outcome.get('region')} rates"
6189 else:
6190 billing = ""
6191 print(
6192 f" {len(surviving)} EBS volume(s) ({total_gib} GiB{billing}) from "
6193 f"cluster {cluster} were left in place:"
6194 )
6195 for record in surviving:
6196 print(f" - {_describe_volume_record(record)} [{record.get('reason')}]")
6197 if cost is None:
6198 print(f" Ongoing cost: {outcome.get('monthly_cost_unavailable')}.")
6199 print(
6200 " Nothing can reattach these once the cluster is gone. Delete them "
6201 "with: aws ec2 delete-volume --region "
6202 f"{outcome.get('region')} --volume-id <vol-id>"
6203 )
6205 for failure in errors:
6206 target = failure.get("volume_id") or cluster
6207 logger.warning("Could not dispose of EBS volume %s: %s", target, failure.get("error"))
6208 print(f" Could not dispose of EBS volume {target}: {failure.get('error')}")
6211def _is_regional_api_bridge_stack(
6212 stack: str,
6213 *,
6214 project_name: str,
6215 stack_names: Collection[str],
6216) -> bool:
6217 """Return whether ``stack`` is a configured per-Region API bridge.
6219 A bare ``"-regional-api-"`` substring is ambiguous because it is valid
6220 inside ``project_name``. Match the exact project-scoped bridge prefix and
6221 require the corresponding configured ``<project>-<region>`` base stack.
6222 """
6223 bridge_prefix = f"{project_name}-regional-api-"
6224 if not stack.startswith(bridge_prefix):
6225 return False
6226 region = stack.removeprefix(bridge_prefix)
6227 return bool(region) and f"{project_name}-{region}" in stack_names
6230def _get_stack_destroy_phases(
6231 stacks: list[str],
6232 *,
6233 project_name: str,
6234) -> tuple[list[str], list[str], list[str], list[str]]:
6235 """Classify and order the exact phases used by orchestrated destroy.
6237 Returns monitoring, regional API bridge, base regional, and pre-regional
6238 global phases. The public preview helper and the execution path both flatten
6239 this result, so custom project names and bridge dependencies cannot drift.
6240 """
6241 stack_names = set(stacks)
6242 monitoring_stacks = sorted(
6243 (stack for stack in stacks if stack.endswith("-monitoring")),
6244 reverse=True,
6245 )
6246 regional_api_stacks = sorted(
6247 (
6248 stack
6249 for stack in stacks
6250 if _is_regional_api_bridge_stack(
6251 stack,
6252 project_name=project_name,
6253 stack_names=stack_names,
6254 )
6255 ),
6256 reverse=True,
6257 )
6258 regional_stacks = sorted(
6259 (
6260 stack
6261 for stack in stacks
6262 if not stack.endswith(("-global", "-api-gateway", "-monitoring"))
6263 and not _is_regional_api_bridge_stack(
6264 stack,
6265 project_name=project_name,
6266 stack_names=stack_names,
6267 )
6268 ),
6269 reverse=True,
6270 )
6271 pre_regional_stacks = sorted(
6272 (stack for stack in stacks if stack.endswith(("-global", "-api-gateway"))),
6273 key=lambda stack: (
6274 1 if stack.endswith("-api-gateway") else (2 if stack.endswith("-global") else 0)
6275 ),
6276 )
6277 return (
6278 monitoring_stacks,
6279 regional_api_stacks,
6280 regional_stacks,
6281 pre_regional_stacks,
6282 )
6285def get_stack_deployment_order(
6286 stacks: list[str],
6287 *,
6288 project_name: str = "gco",
6289) -> list[str]:
6290 """
6291 Get the correct deployment order for stacks.
6293 Order: global stacks first, then regional stacks.
6294 Global stacks: <project>-global, <project>-api-gateway,
6295 <project>-analytics, <project>-monitoring
6296 Regional stacks: <project>-{region} (e.g., gco-us-east-1)
6298 Named stacks are classified by suffix so ordering is independent of
6299 ``project_name`` (#139): a non-``gco`` deployment (``acme-global`` …)
6300 orders identically. Regional stacks are ``<project>-<region>`` and match
6301 no named suffix, so they fall through to the regional bucket.
6302 """
6303 stack_names = set(stacks)
6304 global_stacks = []
6305 regional_stacks = []
6306 regional_api_stacks = []
6308 # Named (non-regional) stack priority by suffix (lower = deploy first).
6309 suffix_priority = {
6310 "-global": 1,
6311 "-api-gateway": 2,
6312 "-analytics": 2.5,
6313 "-monitoring": 3,
6314 }
6316 def _named_priority(stack: str) -> float | None:
6317 for suffix, prio in suffix_priority.items():
6318 if stack.endswith(suffix):
6319 return prio
6320 return None
6322 for stack in stacks:
6323 priority = _named_priority(stack)
6324 if priority is not None:
6325 global_stacks.append((priority, stack))
6326 elif _is_regional_api_bridge_stack(
6327 stack,
6328 project_name=project_name,
6329 stack_names=stack_names,
6330 ):
6331 regional_api_stacks.append(stack)
6332 else:
6333 regional_stacks.append(stack)
6335 # Keep bridge dependencies after every base regional stack. The
6336 # orchestrated lifecycle further separates monitoring into its own phase.
6337 global_stacks.sort(key=lambda x: x[0])
6338 regional_stacks.sort()
6339 regional_api_stacks.sort()
6341 return [s[1] for s in global_stacks] + regional_stacks + regional_api_stacks
6344def get_stack_destroy_order(
6345 stacks: list[str],
6346 *,
6347 project_name: str = "gco",
6348) -> list[str]:
6349 """Return the exact project-aware order used by orchestrated destroy."""
6350 phases = _get_stack_destroy_phases(stacks, project_name=project_name)
6351 return [stack for phase in phases for stack in phase]
6354# =============================================================================
6355# Feature toggle helpers
6356# =============================================================================
6358_FSX_DEFAULTS: dict[str, Any] = {
6359 "enabled": False,
6360 "storage_capacity_gib": 1200,
6361 "deployment_type": "SCRATCH_2",
6362 "per_unit_storage_throughput": 200,
6363 "data_compression_type": "LZ4",
6364 "import_path": None,
6365 "export_path": None,
6366 "auto_import_policy": "NEW_CHANGED_DELETED",
6367}
6370def _find_cdk_json() -> Path | None:
6371 """Find cdk.json in current or parent directories."""
6372 current = Path.cwd()
6373 for parent in [current] + list(current.parents):
6374 cdk_path = parent / "cdk.json"
6375 if cdk_path.exists():
6376 return cdk_path
6377 return None
6380def get_fsx_config(region: str | None = None) -> dict[str, Any]:
6381 """Get current FSx for Lustre configuration from cdk.json.
6383 Args:
6384 region: Optional region to get config for. If provided, checks for
6385 region-specific overrides first.
6387 Returns:
6388 FSx configuration dictionary
6389 """
6390 return _get_feature_config("fsx_lustre", _FSX_DEFAULTS, region)
6393def update_fsx_config(settings: dict[str, Any], region: str | None = None) -> None:
6394 """Update FSx for Lustre configuration in cdk.json.
6396 Args:
6397 settings: FSx settings to update
6398 region: Optional region for region-specific config. If None, updates global config.
6399 """
6400 _update_feature_config("fsx_lustre", settings, _FSX_DEFAULTS, region)
6403# =============================================================================
6404# EKS cluster access configuration (endpoint mode + CIDR allowlist)
6405# =============================================================================
6407_EKS_CLUSTER_DEFAULTS: dict[str, Any] = {
6408 "endpoint_access": "PRIVATE",
6409 "public_access_cidrs": [],
6410 "developer_access": [],
6411}
6414def get_eks_cluster_config() -> dict[str, Any]:
6415 """Get the current eks_cluster configuration from cdk.json.
6417 Synth-time only: the values are read by ``gco/stacks/regional_stack.py``
6418 at the next deploy. There is no per-region override — the block applies
6419 to every regional cluster.
6420 """
6421 return _get_feature_config("eks_cluster", _EKS_CLUSTER_DEFAULTS, None)
6424def update_eks_cluster_config(settings: dict[str, Any]) -> None:
6425 """Update the eks_cluster configuration in cdk.json (config only)."""
6426 _update_feature_config("eks_cluster", settings, _EKS_CLUSTER_DEFAULTS, None)
6429# =============================================================================
6430# Generic feature toggle helpers (used by FSx, Valkey, Aurora, and future features)
6431# =============================================================================
6434def _get_feature_config(
6435 feature_key: str,
6436 default_config: dict[str, Any],
6437 region: str | None = None,
6438) -> dict[str, Any]:
6439 """Get configuration for a toggleable feature from cdk.json.
6441 Args:
6442 feature_key: The cdk.json context key (e.g. "valkey", "aurora_pgvector").
6443 default_config: Default configuration values when the key is missing.
6444 region: Optional region for region-specific overrides.
6446 Returns:
6447 Merged configuration dictionary.
6448 """
6449 cdk_json_path = _find_cdk_json()
6450 if not cdk_json_path:
6451 raise RuntimeError("cdk.json not found")
6453 import json
6455 with open(cdk_json_path, encoding="utf-8") as f:
6456 cdk_config = json.load(f)
6458 global_config = cdk_config.get("context", {}).get(feature_key, default_config)
6460 if region:
6461 region_key = f"{feature_key}_regions"
6462 region_overrides = cdk_config.get("context", {}).get(region_key, {})
6463 if region in region_overrides:
6464 merged = {**global_config, **region_overrides[region]}
6465 merged["region"] = region
6466 merged["is_region_specific"] = True
6467 return merged
6469 result = {**default_config, **global_config}
6470 result["is_region_specific"] = False
6471 return result
6474def _update_feature_config(
6475 feature_key: str,
6476 settings: dict[str, Any],
6477 default_config: dict[str, Any],
6478 region: str | None = None,
6479) -> None:
6480 """Update configuration for a toggleable feature in cdk.json.
6482 Args:
6483 feature_key: The cdk.json context key (e.g. "valkey", "aurora_pgvector").
6484 settings: Settings to update.
6485 default_config: Default configuration values when the key is missing.
6486 region: Optional region for region-specific config.
6487 """
6488 cdk_json_path = _find_cdk_json()
6489 if not cdk_json_path:
6490 raise RuntimeError("cdk.json not found")
6492 import json
6494 with _config_mutation_lock(cdk_json_path):
6495 with open(cdk_json_path, encoding="utf-8") as f:
6496 cdk_config = json.load(f)
6498 if "context" not in cdk_config:
6499 cdk_config["context"] = {}
6501 if region:
6502 region_key = f"{feature_key}_regions"
6503 if region_key not in cdk_config["context"]:
6504 cdk_config["context"][region_key] = {}
6505 if region not in cdk_config["context"][region_key]:
6506 cdk_config["context"][region_key][region] = {}
6507 for key, value in settings.items():
6508 if value is not None or key == "enabled":
6509 cdk_config["context"][region_key][region][key] = value
6510 else:
6511 if feature_key not in cdk_config["context"]:
6512 cdk_config["context"][feature_key] = {**default_config}
6513 for key, value in settings.items():
6514 if value is not None or key == "enabled":
6515 cdk_config["context"][feature_key][key] = value
6517 serialized = json.dumps(cdk_config, indent=2).encode("utf-8")
6518 _atomic_write_bytes(
6519 cdk_json_path,
6520 serialized,
6521 mode=stat.S_IMODE(cdk_json_path.stat().st_mode),
6522 )
6525# =============================================================================
6526# Valkey configuration
6527# =============================================================================
6529_VALKEY_DEFAULTS: dict[str, Any] = {
6530 "enabled": False,
6531 "max_data_storage_gb": 5,
6532 "max_ecpu_per_second": 5000,
6533 "snapshot_retention_limit": 1,
6534}
6537def get_valkey_config(region: str | None = None) -> dict[str, Any]:
6538 """Get current Valkey Serverless configuration from cdk.json."""
6539 return _get_feature_config("valkey", _VALKEY_DEFAULTS, region)
6542def update_valkey_config(settings: dict[str, Any], region: str | None = None) -> None:
6543 """Update Valkey Serverless configuration in cdk.json."""
6544 _update_feature_config("valkey", settings, _VALKEY_DEFAULTS, region)
6547# =============================================================================
6548# Aurora pgvector configuration
6549# =============================================================================
6551_AURORA_DEFAULTS: dict[str, Any] = {
6552 "enabled": False,
6553 "min_acu": 0,
6554 "max_acu": 16,
6555 "backup_retention_days": 7,
6556 "deletion_protection": False,
6557}
6560def get_aurora_config(region: str | None = None) -> dict[str, Any]:
6561 """Get current Aurora pgvector configuration from cdk.json."""
6562 return _get_feature_config("aurora_pgvector", _AURORA_DEFAULTS, region)
6565def update_aurora_config(settings: dict[str, Any], region: str | None = None) -> None:
6566 """Update Aurora pgvector configuration in cdk.json."""
6567 _update_feature_config("aurora_pgvector", settings, _AURORA_DEFAULTS, region)
6570# =============================================================================
6571# Analytics environment configuration
6572# =============================================================================
6574_ANALYTICS_DEFAULTS: dict[str, Any] = {
6575 "enabled": False,
6576 "hyperpod": {"enabled": False},
6577 "canvas": {"enabled": False},
6578 "cognito": {"domain_prefix": None, "removal_policy": "destroy"},
6579 "efs": {"removal_policy": "destroy"},
6580 "studio": {"user_profile_name_prefix": None},
6581}
6584def get_analytics_config() -> dict[str, Any]:
6585 """Get the analytics environment configuration from cdk.json.
6587 The analytics stack is single-region by construction (lives in the
6588 api-gateway region), so this helper does not accept a region argument.
6589 Returned dict is the defaults merged with any operator overrides from
6590 the ``context.analytics_environment`` block.
6591 """
6592 return _get_feature_config("analytics_environment", _ANALYTICS_DEFAULTS)
6595def update_analytics_config(settings: dict[str, Any]) -> None:
6596 """Update the analytics environment configuration in cdk.json.
6598 Mirrors ``update_valkey_config`` / ``update_aurora_config``. Nested
6599 keys under ``analytics_environment`` (``hyperpod``, ``canvas``,
6600 ``cognito``, ``efs``, ``studio``) are merged one level deep rather
6601 than replaced wholesale — ``enable --hyperpod`` must not clobber
6602 ``cognito.removal_policy``.
6603 """
6604 _update_feature_config("analytics_environment", settings, _ANALYTICS_DEFAULTS)
6607# =============================================================================
6608# Cluster observability configuration
6609# =============================================================================
6611# Mirrors the on-by-default cdk.json cluster_observability block. Unlike the
6612# other feature toggles this one defaults to enabled=True: a stock deploy
6613# installs kube-prometheus-stack on every regional cluster and operators opt
6614# out. The CDK side reads/validates the same block via
6615# ConfigLoader.get_cluster_observability_config.
6616_CLUSTER_OBSERVABILITY_DEFAULTS: dict[str, Any] = {
6617 "enabled": True,
6618 "grafana": {
6619 "persistence_size": "10Gi",
6620 "admin_user": "admin",
6621 "admin_password_rotation_schedule": "0 4 1 * *",
6622 },
6623 "prometheus": {"persistence_size": "50Gi", "retention": "15d"},
6624 "alertmanager": {"enabled": True, "persistence_size": "5Gi"},
6625}
6628def get_cluster_observability_config() -> dict[str, Any]:
6629 """Get the cluster observability configuration from cdk.json.
6631 Observability is per-region (installed on every regional cluster) but the
6632 toggle itself is global, so this takes no region argument. Returns the
6633 defaults merged with any operator overrides from the
6634 ``context.cluster_observability`` block.
6635 """
6636 return _get_feature_config("cluster_observability", _CLUSTER_OBSERVABILITY_DEFAULTS)
6639def update_cluster_observability_config(settings: dict[str, Any]) -> None:
6640 """Update the cluster observability toggle in cdk.json.
6642 ``gco monitoring enable`` / ``disable`` pass ``{"enabled": True/False}``;
6643 the grafana/prometheus/alertmanager sub-blocks are left untouched so an
6644 operator's sizing/retention/rotation overrides survive a disable/enable
6645 cycle.
6646 """
6647 _update_feature_config("cluster_observability", settings, _CLUSTER_OBSERVABILITY_DEFAULTS)