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