Coverage for cli/_image_mirror.py: 100.00%

1"""

2Mirror third-party container images into the project's own ECR — shared core.

4Some upstream registries (chiefly Docker Hub, ``docker.io``) rate-limit

5anonymous pulls and have **no credential-free** ECR pull-through cache. On a

6cold cluster that can stall image pulls — most visibly Volcano, whose images

7(``volcanosh/vc-*``) live only on Docker Hub and gate its Helm install. The

8credential-free fix is to **mirror** those images into a ``gco/*`` ECR namespace

9and point the consumer (a Helm values override, a manifest, …) at the mirror, so

10the cluster pulls from same-account ECR with the pull-only node role it already

11has. See ``docs/CUSTOMIZATION.md``.

13This module is the reusable, **general** mirror core — it copies an arbitrary

14list of images, not just Volcano's. It is shared by two callers:

16- ``gco images mirror`` (in ``cli/commands/images_cmd.py``) — the operator CLI.

17- ``cli/stacks.py`` (``StackManager.deploy``) — the auto-mirror that runs before

18 a regional stack's Helm install, so a fresh ``gco stacks deploy`` with

19 ``volcano_image_mirror.enabled`` just works (no separate manual step).

21═══════════════════════════════════════════════════════════════════════════

22HOW TO ADD AN IMAGE TO THE MIRROR

23═══════════════════════════════════════════════════════════════════════════

24The set of images to mirror is produced by :func:`collect_source_refs`, which

25returns a flat list of fully-qualified upstream refs

26(``"<registry>/<repo>:<tag>"``). To add one, extend that function — two flavors:

281. **Static ref** — append a literal, e.g.::

30 refs.append("docker.io/bitnami/redis:7.4.1")

322. **Chart-derived ref** — derive the name/tag from

33 ``lambda/helm-installer/charts.yaml`` so the mirror never drifts from the

34 deployed Helm chart version. Use :func:`_volcano_source_refs` as the template

35 (read the chart's ``values`` block, build ``docker.io/<image>:<tag>``).

37Then wire up the **consumer** so the cluster actually pulls the mirrored copy

38instead of the upstream — typically a Helm ``image_registry``/``image`` override

39in ``gco/stacks/regional_stack.py`` (see ``_helm_chart_value_overrides`` /

40``_configure_volcano_image_mirror`` for the Volcano example) or a manifest image

41reference. The mirror copies ``<registry>/<repo>:<tag>`` to

42``<account>.dkr.ecr.<region>.<url-suffix>/<ecr_namespace>/<repo>:<tag>``, so the

43consumer must point at ``<…>/<ecr_namespace>/<repo>``.

45WHY mirror rather than pull-through cache: ECR pull-through cache for Docker Hub

46*requires* stored credentials (anonymous is unsupported), and on EKS Auto Mode

47the pull-only, service-managed node role complicates cache-miss imports.

48Mirroring needs no credential — the images become plain ``gco/*`` ECR repos.

49═══════════════════════════════════════════════════════════════════════════

51The copy preserves the **full manifest list** (every architecture) so an arm64

52(Graviton) node and an amd64 node both find a matching image; a naive

53``pull``/``tag``/``push`` (which drops all but the host architecture) is never

54used. The concrete mechanism is chosen at runtime — Docker Buildx

55(``buildx imagetools create``), Finch/nerdctl (``--all-platforms``), or skopeo

56(``copy --all``) — see :func:`resolve_copy_strategy`.

57"""

59from __future__ import annotations

61import base64

62import json

63import shutil

64import subprocess # invokes container CLI / skopeo with fixed, non-shell argv

65from collections.abc import Callable, Mapping

66from dataclasses import dataclass

67from pathlib import Path

68from typing import Any

70import boto3

71import yaml

73from ._image_uri import ecr_registry_host

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

76# Generated at (UTC): 2026-09-18T02:11:36Z

77# Generated from Git commit: b8faa9689385cea16155a285a7f70cf6d488e512

78# Flowchart(s) generated from this file:

79# * ``read_mirror_config`` -> ``diagrams/code_diagrams/cli/_image_mirror.read_mirror_config.html``

80# (PNG: ``diagrams/code_diagrams/cli/_image_mirror.read_mirror_config.png``)

81# * ``mirror_images`` -> ``diagrams/code_diagrams/cli/_image_mirror.mirror_images.html``

82# (PNG: ``diagrams/code_diagrams/cli/_image_mirror.mirror_images.png``)

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

84# <pyflowchart-code-diagram> END

87# Repo root is the parent of cli/.

88_REPO_ROOT = Path(__file__).resolve().parent.parent

89_CHARTS_YAML = _REPO_ROOT / "lambda" / "helm-installer" / "charts.yaml"

90_CDK_JSON = _REPO_ROOT / "cdk.json"

92# Fallback mirror namespace used only when cdk.json can't be read to resolve

93# the project prefix. Normal paths derive ``<project_name>/dockerhub`` from

94# cdk.json's ``project_name`` (see ``read_mirror_config``) so two deployments

95# mirror into isolated ECR namespaces (#139).

96_DEFAULT_NAMESPACE = "gco/dockerhub"

98# The Volcano components enabled by default (controller, scheduler, admission),

99# keyed by the ``basic.<key>`` image-name field in charts.yaml. The

100# agent/agent-scheduler images are not pulled in the default configuration and

101# are intentionally excluded.

102_VOLCANO_IMAGE_NAME_KEYS = (

103 "controller_image_name",

104 "scheduler_image_name",

105 "admission_image_name",

106)

107# Upstream registry Volcano's chart pulls from by default (chart value

108# ``basic.image_registry``).

109_VOLCANO_UPSTREAM_REGISTRY = "docker.io"

110

111# Default logger — callers may pass their own ``log`` callable (e.g. to route

112# through a deploy progress stream).

113LogFn = Callable[[str], None]

114RepositoryCreatedCallback = Callable[[str, Mapping[str, Any]], None]

115

116

117def _bind_repository_created_callback(

118 callback: RepositoryCreatedCallback,

119 region: str,

120) -> Callable[[Mapping[str, Any]], None]:

121 """Bind one target Region without obscuring the callback's payload type."""

122

123 def notify(repository: Mapping[str, Any]) -> None:

124 callback(region, repository)

125

126 return notify

127

128

129@dataclass(frozen=True)

130class MirrorItem:

131 """One image to copy: ``source_ref`` -> ``dest_ref`` (repo ``dest_repo``)."""

132

133 source_ref: str

134 dest_repo: str

135 dest_ref: str

136

137 @property

138 def tag(self) -> str:

139 """The image tag (the segment after the final ``:`` of ``dest_ref``)."""

140 return self.dest_ref.rsplit(":", 1)[1]

141

142

143def parse_source_ref(ref: str) -> tuple[str, str]:

144 """Split ``"<registry>/<repo>:<tag>"`` into ``(repo_path, tag)``.

145

146 The registry host (first slash-delimited segment) is dropped so the image

147 can be re-homed under the ECR namespace while preserving its repo path, e.g.

148 ``"docker.io/volcanosh/vc-scheduler:v1.15.0"`` -> ``("volcanosh/vc-scheduler",

149 "v1.15.0")``.

150 """

151 if "/" not in ref:

152 raise ValueError(f"source ref must include a registry host: {ref!r}")

153 _registry, rest = ref.split("/", 1)

154 if ":" not in rest:

155 raise ValueError(f"source ref must include a tag: {ref!r}")

156 repo_path, tag = rest.rsplit(":", 1)

157 if not repo_path or not tag:

158 raise ValueError(f"could not parse source ref {ref!r}")

159 return repo_path, tag

160

161

162def load_charts_config(charts_path: Path = _CHARTS_YAML) -> dict[str, Any]:

163 """Load and return the parsed ``charts.yaml`` mapping."""

164 with open(charts_path, encoding="utf-8") as f:

165 data = yaml.safe_load(f) or {}

166 if not isinstance(data, dict):

167 raise ValueError(f"{charts_path} did not parse to a mapping")

168 return data

169

170

171def _volcano_source_refs(charts_config: dict[str, Any]) -> list[str]:

172 """Return Volcano's upstream image refs derived from ``charts.yaml``.

173

174 Reads ``charts.charts.volcano.values.basic`` — the per-component

175 ``*_image_name`` fields and the shared ``image_tag_version`` (each

176 component's own ``*_image_tag_version`` takes precedence when set, matching

177 the chart's templating) — and returns ``["docker.io/<image>:<tag>", ...]``.

178 Deriving from the chart keeps the mirror tag identical to what Helm requests.

179 """

180 volcano = (charts_config.get("charts", {}) or {}).get("volcano", {}) or {}

181 basic = (volcano.get("values", {}) or {}).get("basic", {}) or {}

182 shared_tag = str(basic.get("image_tag_version") or "").strip()

183 if not shared_tag:

184 raise ValueError(

185 "volcano.values.basic.image_tag_version is missing from charts.yaml; "

186 "cannot determine which Volcano image tag to mirror."

187 )

188

189 refs: list[str] = []

190 for name_key in _VOLCANO_IMAGE_NAME_KEYS:

191 image_name = str(basic.get(name_key) or "").strip()

192 if not image_name:

193 continue

194 component = name_key.removesuffix("_image_name")

195 per_component_tag = str(basic.get(f"{component}_image_tag_version") or "").strip()

196 tag = per_component_tag or shared_tag

197 refs.append(f"{_VOLCANO_UPSTREAM_REGISTRY}/{image_name}:{tag}")

198 if not refs:

199 raise ValueError(

200 "No Volcano component image names found under volcano.values.basic in charts.yaml."

201 )

202 return refs

203

204

205def collect_source_refs(charts_config: dict[str, Any] | None = None) -> list[str]:

206 """Return every upstream image ref to mirror (``"<registry>/<repo>:<tag>"``).

207

208 This is the single extension point for the mirror. To add an image, append

209 to ``refs`` below — either a static literal or a chart-derived ref (see the

210 module docstring, "HOW TO ADD AN IMAGE TO THE MIRROR"). Remember to also

211 point the image's *consumer* at the mirrored copy.

212 """

213 charts_config = charts_config if charts_config is not None else load_charts_config()

214 refs: list[str] = []

215

216 # Volcano (docker.io/volcanosh/vc-*) — the reason this mirror exists.

217 refs.extend(_volcano_source_refs(charts_config))

218

219 # ── ADD MORE IMAGES HERE ──────────────────────────────────────────────

220 # e.g. refs.append("docker.io/bitnami/redis:7.4.1") # static, or

221 # refs.extend(_my_chart_source_refs(charts_config)) # chart-derived

222 # Then wire the consumer to <ecr_namespace>/<repo> (see module docstring).

223

224 return refs

225

226

227def plan_from_sources(

228 source_refs: list[str], registry_host: str, ecr_namespace: str

229) -> list[MirrorItem]:

230 """Compute the copy plan: one :class:`MirrorItem` per source ref.

231

232 ``registry_host`` is ``<account>.dkr.ecr.<region>.<url-suffix>`` and

233 ``ecr_namespace`` is the destination prefix (e.g. ``gco/dockerhub``). The

234 destination preserves the upstream repo path so it lines up with whatever

235 ``image_registry``/``image`` override the consumer points at

236 (``<registry_host>/<ecr_namespace>`` + ``/<repo_path>``).

237 """

238 ecr_namespace = ecr_namespace.strip("/")

239 items: list[MirrorItem] = []

240 for ref in source_refs:

241 repo_path, tag = parse_source_ref(ref)

242 dest_repo = f"{ecr_namespace}/{repo_path}"

243 dest_ref = f"{registry_host}/{dest_repo}:{tag}"

244 items.append(MirrorItem(source_ref=ref, dest_repo=dest_repo, dest_ref=dest_ref))

245 return items

246

247

248def read_mirror_config(cdk_json_path: Path = _CDK_JSON) -> dict[str, Any]:

249 """Return ``{enabled, ecr_namespace}`` from cdk.json ``volcano_image_mirror``.

250

251 Defaults to disabled / ``<project_name>/dockerhub`` (``gco/dockerhub`` for

252 the stock project) when the block is absent, so a second deployment mirrors

253 into its own ECR namespace (#139).

254 Used by the deploy path to decide whether to auto-mirror. (The cdk.json key

255 is still named ``volcano_image_mirror`` — Volcano is the only consumer today

256 — but the mirror itself is general; see :func:`collect_source_refs`.)

257 """

258 try:

259 with open(cdk_json_path, encoding="utf-8") as f:

260 ctx = json.load(f).get("context", {}) or {}

261 except OSError, json.JSONDecodeError:

262 return {"enabled": False, "ecr_namespace": _DEFAULT_NAMESPACE}

263 # Default the mirror namespace to the deployment's own project prefix

264 # (``<project_name>/dockerhub``) so a second deployment mirrors into its own

265 # ECR namespace (#139); resolves to ``gco/dockerhub`` for the stock project.

266 default_namespace = f"{ctx.get('project_name') or 'gco'}/dockerhub"

267 block = ctx.get("volcano_image_mirror") or {}

268 namespace = str(block.get("ecr_namespace", default_namespace)).strip("/") or default_namespace

269 return {"enabled": bool(block.get("enabled", False)), "ecr_namespace": namespace}

270

271

272def cdk_default_namespace(cdk_json_path: Path = _CDK_JSON) -> str:

273 """Return ``volcano_image_mirror.ecr_namespace`` from cdk.json (default gco/dockerhub)."""

274 return str(read_mirror_config(cdk_json_path)["ecr_namespace"])

275

276

277def _account_id() -> str:

278 return str(boto3.client("sts").get_caller_identity()["Account"])

279

280

281def _registry_host(account_id: str, region: str) -> str:

282 """Return the ECR host using botocore's partition URL suffix metadata."""

283 return ecr_registry_host(account_id, region)

284

285

286def detect_runtime() -> str:

287 """Return the container CLI to drive (``docker``, ``finch``, or ``podman``).

288

289 Mirrors the project's runtime preference (docker > finch > podman). Note

290 that on a Finch-based setup the ``docker`` shim may itself be Finch — the

291 copy strategy is resolved separately by probing for capabilities rather

292 than trusting the command name.

293 """

294 for cmd in ("docker", "finch", "podman"):

295 if shutil.which(cmd):

296 return cmd

297 raise RuntimeError(

298 "No container CLI found (looked for docker, finch, podman). Install one, "

299 "or install skopeo for a daemon-less copy."

300 )

301

302

303def _runtime_has_buildx(runtime: str) -> bool:

304 """True if ``<runtime> buildx version`` succeeds (Docker Buildx present)."""

305 try:

306 return (

307 subprocess.run( # fixed argv, no shell

308 [runtime, "buildx", "version"],

309 capture_output=True,

310 timeout=15,

311 check=False,

312 ).returncode

313 == 0

314 )

315 except OSError, subprocess.SubprocessError:

316 return False

317

318

319def _runtime_supports_all_platforms(runtime: str) -> bool:

320 """True if ``<runtime> pull`` advertises ``--all-platforms`` (Finch/nerdctl)."""

321 try:

322 out = subprocess.run( # fixed argv, no shell

323 [runtime, "pull", "--help"],

324 capture_output=True,

325 text=True,

326 timeout=15,

327 check=False,

328 )

329 except OSError, subprocess.SubprocessError:

330 return False

331 return "--all-platforms" in (out.stdout + out.stderr)

332

333

334def resolve_copy_strategy(runtime: str) -> str:

335 """Pick a multi-arch-preserving copy strategy from what's available.

336

337 Priority:

338 1. ``buildx`` — ``<runtime> buildx imagetools create`` (registry-to-registry,

339 no local pull). Best when Docker Buildx is present.

340 2. ``all-platforms`` — ``<runtime> pull/tag/push --all-platforms``

341 (Finch / nerdctl), preserves the manifest list via containerd.

342 3. ``skopeo`` — ``skopeo copy --all`` (daemon-less), if skopeo is on PATH.

343

344 Every strategy preserves all architectures; a plain ``pull``/``tag``/``push``

345 (which would drop every arch except the host's) is never used. Raises with

346 guidance if none is available.

347 """

348 if _runtime_has_buildx(runtime):

349 return "buildx"

350 if _runtime_supports_all_platforms(runtime):

351 return "all-platforms"

352 if shutil.which("skopeo"):

353 return "skopeo"

354 raise RuntimeError(

355 f"No multi-arch image-copy method available. Need one of: "

356 f"'{runtime} buildx' (Docker Buildx), '{runtime} pull --all-platforms' "

357 f"(Finch/nerdctl), or skopeo on PATH."

358 )

359

360

361def ensure_repository(

362 ecr_client: Any,

363 repo_name: str,

364 log: LogFn = print,

365 repository_tags: Mapping[str, str] | None = None,

366 on_created: Callable[[Mapping[str, Any]], None] | None = None,

367) -> bool:

368 """Create a repository and synchronously publish its causal acknowledgement."""

369 kwargs: dict[str, Any] = {"repositoryName": repo_name}

370 if repository_tags:

371 kwargs["tags"] = [

372 {"Key": str(key), "Value": str(value)} for key, value in sorted(repository_tags.items())

373 ]

374 try:

375 response = ecr_client.create_repository(**kwargs)

376 repository = response.get("repository") if isinstance(response, dict) else None

377 if on_created is not None:

378 if not isinstance(repository, dict):

379 raise RuntimeError(

380 f"ECR create_repository omitted its acknowledgement for {repo_name}"

381 )

382 if str(repository.get("repositoryName") or "") != repo_name:

383 raise RuntimeError(

384 f"ECR create_repository acknowledged a different repository for {repo_name}"

385 )

386 on_created(repository)

387 log(f" created ECR repository {repo_name}")

388 return True

389 except ecr_client.exceptions.RepositoryAlreadyExistsException:

390 log(f" ECR repository {repo_name} already exists")

391 return False

392

393

394def tag_exists(ecr_client: Any, repo_name: str, tag: str) -> bool:

395 """True if ``tag`` already exists in the ECR repo (drives skip-if-mirrored).

396

397 Returns False when the repository or tag does not exist, so the caller

398 mirrors it. Any other error propagates.

399 """

400 try:

401 resp = ecr_client.describe_images(repositoryName=repo_name, imageIds=[{"imageTag": tag}])

402 return bool(resp.get("imageDetails"))

403 except ecr_client.exceptions.ImageNotFoundException:

404 return False

405 except ecr_client.exceptions.RepositoryNotFoundException:

406 return False

407

408

409def ecr_auth(region: str) -> tuple[str, str]:

410 """Return ``(username, password)`` for the region's ECR registry."""

411 ecr = boto3.client("ecr", region_name=region)

412 token = ecr.get_authorization_token()["authorizationData"][0]["authorizationToken"]

413 username, password = base64.b64decode(token).decode().split(":", 1)

414 return username, password

415

416

417def runtime_login(

418 runtime: str, registry_host: str, username: str, password: str, log: LogFn = print

419) -> None:

420 """Authenticate the container runtime against the ECR registry."""

421 result = subprocess.run( # fixed argv, no shell

422 [runtime, "login", "--username", username, "--password-stdin", registry_host],

423 input=password.encode(),

424 capture_output=True,

425 check=False,

426 )

427 if result.returncode != 0:

428 raise RuntimeError(

429 f"{runtime} login to {registry_host} failed: "

430 f"{result.stderr.decode(errors='replace').strip()}"

431 )

432 log(f" authenticated {runtime} to {registry_host}")

433

434

435def _copy_commands(item: MirrorItem, runtime: str, strategy: str, password: str) -> list[list[str]]:

436 """Build the argv list(s) for one image copy under the chosen strategy.

437

438 Factored out (pure) so the command shape is unit-testable without invoking

439 any runtime. All strategies preserve the full multi-arch manifest list.

440 """

441 if strategy == "buildx":

442 return [

443 [runtime, "buildx", "imagetools", "create", "--tag", item.dest_ref, item.source_ref]

444 ]

445 if strategy == "all-platforms":

446 return [

447 [runtime, "pull", "--all-platforms", item.source_ref],

448 [runtime, "tag", item.source_ref, item.dest_ref],

449 [runtime, "push", "--all-platforms", item.dest_ref],

450 ]

451 if strategy == "skopeo":

452 return [

453 [

454 "skopeo",

455 "copy",

456 "--all",

457 "--dest-creds",

458 f"AWS:{password}",

459 f"docker://{item.source_ref}",

460 f"docker://{item.dest_ref}",

461 ]

462 ]

463 raise ValueError(f"unknown copy strategy: {strategy!r}")

464

465

466def copy_image(

467 item: MirrorItem,

468 runtime: str = "docker",

469 strategy: str = "buildx",

470 password: str = "", # nosec B107 # empty means "no registry login"; callers pass the real token

471 log: LogFn = print,

472) -> None:

473 """Copy one image registry-to-registry, preserving the full manifest list.

474

475 Dispatches to the resolved ``strategy`` (``buildx`` / ``all-platforms`` /

476 ``skopeo``) — see :func:`resolve_copy_strategy`. Every strategy carries all

477 architectures so both amd64 and arm64 (Graviton) nodes find a match.

478 """

479 log(f" copying {item.source_ref} -> {item.dest_ref} [{strategy}]")

480 for cmd in _copy_commands(item, runtime, strategy, password):

481 result = subprocess.run( # fixed argv, no shell

482 cmd, capture_output=True, text=True, check=False

483 )

484 if result.returncode != 0:

485 raise RuntimeError(

486 f"image copy failed for {item.source_ref} "

487 f"(strategy {strategy}, step {cmd[:3]}): "

488 f"{(result.stderr or result.stdout).strip()}"

489 )

490

491

492def plan_mirror(

493 region: str,

494 ecr_namespace: str | None = None,

495 source_refs: list[str] | None = None,

496 charts_path: Path | None = None,

497) -> dict[str, Any]:

498 """Resolve the mirror plan as plain data — no ECR writes, no image copies.

499

500 Read-only counterpart to :func:`mirror_images`: it resolves the account and

501 destination registry (a single STS ``GetCallerIdentity`` call) and computes

502 where each upstream image *would* be mirrored, but creates no repositories

503 and copies nothing. Backs the ``images_mirror_plan`` MCP tool and is reused

504 by :func:`mirror_status`.

505

506 ``source_refs`` defaults to :func:`collect_source_refs`; ``ecr_namespace``

507 defaults to :func:`cdk_default_namespace`. Returns ``{region, registry,

508 ecr_namespace, images: [{source_ref, dest_repo, dest_ref, tag}, ...]}`` where

509 ``registry`` is ``<account>.dkr.ecr.<region>.<url-suffix>/<ecr_namespace>``.

510 """

511 ecr_namespace = (ecr_namespace or cdk_default_namespace()).strip("/")

512 if source_refs is None:

513 source_refs = collect_source_refs(load_charts_config(charts_path) if charts_path else None)

514 registry_host = _registry_host(_account_id(), region)

515 plan = plan_from_sources(source_refs, registry_host, ecr_namespace)

516 return {

517 "region": region,

518 "registry": f"{registry_host}/{ecr_namespace}",

519 "ecr_namespace": ecr_namespace,

520 "images": [

521 {

522 "source_ref": item.source_ref,

523 "dest_repo": item.dest_repo,

524 "dest_ref": item.dest_ref,

525 "tag": item.tag,

526 }

527 for item in plan

528 ],

529 }

530

531

532def mirror_status(

533 region: str,

534 ecr_namespace: str | None = None,

535 source_refs: list[str] | None = None,

536 charts_path: Path | None = None,

537) -> dict[str, Any]:

538 """Report, per planned image, whether it is already mirrored in ECR.

539

540 Read-only: builds the plan via :func:`plan_mirror`, then probes each

541 destination tag with :func:`tag_exists` (ECR ``DescribeImages``; no writes).

542 Returns the plan augmented with a ``mirrored`` bool per image plus top-level

543 ``all_mirrored`` and ``missing`` (the destination refs not yet present), so

544 an operator can tell at a glance whether a deploy's auto-mirror still has

545 anything to copy before the consuming Helm install runs.

546 """

547 plan = plan_mirror(region, ecr_namespace, source_refs, charts_path)

548 ecr_client = boto3.client("ecr", region_name=region)

549 images: list[dict[str, Any]] = []

550 missing: list[str] = []

551 for img in plan["images"]:

552 present = tag_exists(ecr_client, img["dest_repo"], img["tag"])

553 images.append({**img, "mirrored": present})

554 if not present:

555 missing.append(img["dest_ref"])

556 return {

557 "region": plan["region"],

558 "registry": plan["registry"],

559 "ecr_namespace": plan["ecr_namespace"],

560 "images": images,

561 "all_mirrored": not missing,

562 "missing": missing,

563 }

564

565

566def mirror_images(

567 region: str,

568 ecr_namespace: str | None = None,

569 source_refs: list[str] | None = None,

570 charts_path: Path | None = None,

571 skip_existing: bool = True,

572 log: LogFn = print,

573 repository_tags: Mapping[str, str] | None = None,

574 on_repository_created: RepositoryCreatedCallback | None = None,

575) -> dict[str, Any]:

576 """Mirror the configured upstream images into ECR for ``region`` (full flow).

577

578 ``source_refs`` defaults to :func:`collect_source_refs` (every registered

579 image). Resolves the account/registry, builds the plan, picks a multi-arch

580 copy strategy, authenticates, then for each image ensures the repo exists and

581 copies it — skipping any tag already present when ``skip_existing`` is True

582 (so repeat deploys are a fast no-op). Returns a summary dict with the

583 destination ``registry``, the ``strategy`` used, and the ``mirrored`` /

584 ``skipped`` destination refs.

585 """

586 ecr_namespace = (ecr_namespace or cdk_default_namespace()).strip("/")

587 if source_refs is None:

588 source_refs = collect_source_refs(load_charts_config(charts_path) if charts_path else None)

589

590 account_id = _account_id()

591 registry_host = _registry_host(account_id, region)

592 plan = plan_from_sources(source_refs, registry_host, ecr_namespace)

593

594 runtime = detect_runtime()

595 strategy = resolve_copy_strategy(runtime)

596

597 log(

598 f"Mirroring {len(plan)} image(s) into "

599 f"{registry_host}/{ecr_namespace} (region {region}) "

600 f"via {strategy} (runtime: {runtime}):"

601 )

602

603 # Auth: every strategy needs ECR push credentials. buildx / all-platforms

604 # use the runtime's credential store (so we `<runtime> login`); skopeo takes

605 # the password inline via --dest-creds.

606 username, password = ecr_auth(region)

607 if strategy != "skopeo":

608 runtime_login(runtime, registry_host, username, password, log=log)

609

610 ecr_client = boto3.client("ecr", region_name=region)

611 mirrored: list[str] = []

612 skipped: list[str] = []

613 created_repositories: list[str] = []

614 on_created = (

615 _bind_repository_created_callback(on_repository_created, region)

616 if on_repository_created is not None

617 else None

618 )

619 for item in plan:

620 if ensure_repository(

621 ecr_client,

622 item.dest_repo,

623 log=log,

624 repository_tags=repository_tags,

625 on_created=on_created,

626 ):

627 created_repositories.append(item.dest_repo)

628 if skip_existing and tag_exists(ecr_client, item.dest_repo, item.tag):

629 log(f" skip (already mirrored): {item.dest_ref}")

630 skipped.append(item.dest_ref)

631 continue

632 copy_image(item, runtime=runtime, strategy=strategy, password=password, log=log)

633 mirrored.append(item.dest_ref)

634

635 return {

636 "registry": f"{registry_host}/{ecr_namespace}",

637 "strategy": strategy,

638 "mirrored": mirrored,

639 "skipped": skipped,

640 "created_repositories": sorted(set(created_repositories)),

641 }

Coverage for cli / _image_mirror.py: 100.00%

216 statements