Coverage for cli / images.py: 100.00%
546 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""
2Container image registry management for GCO CLI.
4Provides ``ImageManager`` for building, pushing, and managing user
5container images stored in per-project ECR repositories under the
6``gco/`` prefix. Builds run through the same container runtime
7(Docker, Finch, or Podman) used by CDK asset bundling, detected via
8``cli._container_runtime``.
10The ECR repository layout mirrors the project naming convention:
11``<account>.dkr.ecr.<region>.<url-suffix>/gco/<name>:<tag>``.
13Read-only methods (``list_repos``, ``list_tags``, ``describe``,
14``get_uri``, ``replication_get``, ``replication_status``) hit ECR
15directly via boto3 and do not invoke any container runtime.
17Administrative methods (``init``, ``lifecycle_get``, ``lifecycle_set``,
18``replication_sync``) configure the repository surface and are
19idempotent — re-running them is safe.
21Destructive methods (``delete_tag``, ``delete_repo``, ``cleanup``,
22``prune``, ``orphans``) require explicit caller intent and never run
23implicitly.
24"""
26from __future__ import annotations
28import base64
29import json
30import logging
31import os
32import re
33import subprocess
34from datetime import UTC, datetime, timedelta
35from pathlib import Path
36from typing import Any
38import boto3
39from botocore.exceptions import ClientError
41from gco.service_images import discover_service_dockerfiles
43from ._container_runtime import detect_container_runtime
44from ._image_uri import (
45 aws_partition,
46 ecr_registry_host,
47)
48from ._image_uri import (
49 rewrite_image_uri_for_region as _rewrite_image_uri_for_region, # noqa: F401
50)
51from .config import GCOConfig, _load_cdk_json, get_config
53# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
54# Generated at (UTC): 2026-10-25T13:00:05Z
55# Generated from Git commit: de508a874e7ac267180e26a4c6fc56c1cd0878c2
56# Flowchart(s) generated from this file:
57# * ``ImageManager.build`` -> ``diagrams/code_diagrams/cli/images.ImageManager_build.html``
58# (PNG: ``diagrams/code_diagrams/cli/images.ImageManager_build.png``)
59# * ``ImageManager.push`` -> ``diagrams/code_diagrams/cli/images.ImageManager_push.html``
60# (PNG: ``diagrams/code_diagrams/cli/images.ImageManager_push.png``)
61# * ``ImageManager.cleanup`` -> ``diagrams/code_diagrams/cli/images.ImageManager_cleanup.html``
62# (PNG: ``diagrams/code_diagrams/cli/images.ImageManager_cleanup.png``)
63# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
64# <pyflowchart-code-diagram> END
67logger = logging.getLogger(__name__)
69# Image name and tag validation regexes.
70#
71# Names: short, dns-friendly. Lowercase letter start, lowercase
72# alphanumerics and dashes after, max 63 characters total. The regex
73# also accepts a single character (``^[a-z]$``) — any longer name
74# requires a closing alphanumeric so dangling dashes are rejected.
75_NAME_RE = re.compile(r"^[a-z][a-z0-9-]{0,62}$")
77# Tags: docker reference grammar. First character must be alnum or
78# underscore; subsequent characters allow dot, dash, underscore.
79# 128 chars max.
80_TAG_RE = re.compile(r"^[a-zA-Z0-9_][a-zA-Z0-9_.\-]{0,127}$")
82# Project repository prefix. Every repo this manager creates lives under
83# ``<project_name>/`` (default ``gco/``) so a single replication / lifecycle /
84# removal-policy rule can target the whole deployment, and two deployments in
85# one account+region get isolated ECR namespaces (#139). Resolved per-instance
86# from ``config.project_name`` into ``self._repo_prefix`` in ``__init__``.
88# First-party images that GCO builds and ships itself, as opposed to the user
89# images pushed through ``build``/``push``. Each entry pairs the logical image
90# name (which becomes the ``gco/<name>`` ECR repository suffix) with the
91# Dockerfile that produces it, and the mapping is *discovered* from
92# ``dockerfiles/Dockerfile.<name>`` rather than typed out: the previous
93# hand-kept list had already drifted (it was missing cost-monitor), so the
94# filenames are the single source of truth shared with the regional stack and
95# the CI build/scan matrix. See ``gco.service_images``.
96_MAINTAINED_IMAGES: dict[str, str] = discover_service_dockerfiles()
98# Default image served by disaggregated prefill/decode deployments when the
99# operator does not supply one. As of this tag the upstream vLLM OpenAI server
100# image bundles the Mooncake transfer engine as a first-class KV-connector
101# dependency, so GCO no longer builds or maintains its own image — deploys pull
102# this upstream image directly from Docker Hub. Pinned to an explicit version
103# for reproducibility; bump intentionally when validating a new vLLM release
104# and never use a mutable/rolling tag such as ``latest``.
105_DISAGGREGATED_DEFAULT_IMAGE = "vllm/vllm-openai:v0.32.0"
107# Default lifecycle policy parameters.
108_DEFAULT_KEEP_TAGGED = 20
109_DEFAULT_EXPIRE_UNTAGGED_DAYS = 7
111# Digest extraction from ``docker push`` stdout/stderr. The runtime
112# emits a line of the form ``... digest: sha256:... size: ...``.
113_DIGEST_RE = re.compile(r"sha256:[a-f0-9]{64}")
116class ImageManager:
117 """Manages user container images in ECR.
119 Construction is cheap — no AWS calls happen until a method is
120 invoked. The account ID and target region are resolved lazily.
121 """
123 def __init__(self, config: GCOConfig | None = None, region: str | None = None):
124 self.config = config or get_config()
125 # ECR repo namespace for this deployment (#139): repos live under
126 # ``<project_name>/`` so two deployments in one account+region don't
127 # share an ECR namespace. Defaults to ``gco`` — byte-identical to the
128 # pre-#139 hardcoded prefix for the stock deployment.
129 self._repo_prefix = self.config.project_name
130 self.region = self._resolve_region(region)
131 self._account_id_cache: str | None = None
133 # ------------------------------------------------------------------
134 # Region / account helpers
135 # ------------------------------------------------------------------
136 def _resolve_region(self, region: str | None) -> str:
137 """Pick a region for ECR API calls.
139 Priority: explicit argument, ``AWS_DEFAULT_REGION``, then the global
140 region where the shared ECR registry is deployed. ``GCOConfig`` has no
141 ``regions`` attribute; deployment-region discovery is handled
142 separately by :meth:`_replication_regions`.
143 """
144 if region:
145 return region
146 env_region = os.environ.get("AWS_DEFAULT_REGION")
147 if env_region:
148 return env_region
149 return str(self.config.global_region)
151 def _account_id(self) -> str:
152 """Return the AWS account ID via STS GetCallerIdentity (cached)."""
153 if self._account_id_cache is None:
154 sts = boto3.client("sts")
155 self._account_id_cache = sts.get_caller_identity()["Account"]
156 return self._account_id_cache
158 def _registry_host(self) -> str:
159 """Return the partition-correct ECR registry host for this region."""
160 return ecr_registry_host(self._account_id(), self.region)
162 def _repo_arn(self, name: str) -> str:
163 """Return the full ARN of the repository under the project prefix."""
164 return (
165 f"arn:{aws_partition(self.region)}:ecr:{self.region}:{self._account_id()}:"
166 f"repository/{self._repo_prefix}/{name}"
167 )
169 def _ecr_client(self) -> Any:
170 """Return a boto3 ECR client targeting the manager's region."""
171 return boto3.client("ecr", region_name=self.region)
173 # ------------------------------------------------------------------
174 # Validation helpers
175 # ------------------------------------------------------------------
176 def _validate_context(self, context: str) -> Path:
177 """Validate the build context path.
179 The path must exist on disk and resolve to a directory. Raw
180 ``..`` segments in the supplied string are rejected outright
181 so the caller can't trick the manager into reaching outside an
182 intended workspace; the resolved path is then returned for use
183 as ``cwd`` of the build.
184 """
185 # Reject string-level traversal segments BEFORE resolving the
186 # path so callers receive a clear error rather than a silent
187 # rewrite up the tree.
188 parts = Path(context).parts
189 if ".." in parts:
190 raise ValueError(f"Invalid build context: path traversal not allowed: {context}")
191 resolved = Path(context).resolve()
192 if not resolved.exists():
193 raise FileNotFoundError(f"Build context not found: {context}")
194 if not resolved.is_dir():
195 raise ValueError(f"Build context is not a directory: {context}")
196 return resolved
198 def _validate_name(self, name: str) -> str:
199 """Validate an image name against ``_NAME_RE``."""
200 if not _NAME_RE.match(name):
201 raise ValueError(
202 f"Invalid image name: {name!r}. Expected lowercase letters, "
203 "digits, and dashes; must start with a letter; max 63 chars."
204 )
205 return name
207 def _validate_tag(self, tag: str) -> str:
208 """Validate an image tag against ``_TAG_RE``."""
209 if not _TAG_RE.match(tag):
210 raise ValueError(
211 f"Invalid image tag: {tag!r}. Expected alphanumerics, dots, "
212 "dashes, and underscores; max 128 chars."
213 )
214 return tag
216 # ------------------------------------------------------------------
217 # Default-value helpers
218 # ------------------------------------------------------------------
219 def _git_short_sha(self) -> str | None:
220 """Return the current short git SHA, or ``None`` when unavailable."""
221 try:
222 result = subprocess.run(
223 ["git", "rev-parse", "--short", "HEAD"],
224 capture_output=True,
225 text=True,
226 timeout=5,
227 check=False,
228 )
229 if result.returncode == 0 and result.stdout.strip():
230 return result.stdout.strip()
231 except (FileNotFoundError, subprocess.TimeoutExpired, OSError) as e:
232 logger.debug("git rev-parse failed: %s", e)
233 return None
235 def _default_tag(self) -> str:
236 """Return ``_git_short_sha()`` when available, else ``"latest"``."""
237 sha = self._git_short_sha()
238 return sha if sha else "latest"
240 def _default_lifecycle_policy(self) -> dict[str, Any]:
241 """Return the default ECR lifecycle policy as a dict.
243 The policy keeps the most recent ``_DEFAULT_KEEP_TAGGED`` tagged
244 images and expires untagged images after
245 ``_DEFAULT_EXPIRE_UNTAGGED_DAYS`` days. The structure matches
246 the JSON shape that ``ecr.put_lifecycle_policy`` accepts after
247 being JSON-stringified at the call site.
248 """
249 return {
250 "rules": [
251 {
252 "rulePriority": 1,
253 "description": (f"Keep last {_DEFAULT_KEEP_TAGGED} tagged images"),
254 "selection": {
255 "tagStatus": "tagged",
256 "countType": "imageCountMoreThan",
257 "countNumber": _DEFAULT_KEEP_TAGGED,
258 "tagPatternList": ["*"],
259 },
260 "action": {"type": "expire"},
261 },
262 {
263 "rulePriority": 2,
264 "description": (f"Expire untagged after {_DEFAULT_EXPIRE_UNTAGGED_DAYS} days"),
265 "selection": {
266 "tagStatus": "untagged",
267 "countType": "sinceImagePushed",
268 "countUnit": "days",
269 "countNumber": _DEFAULT_EXPIRE_UNTAGGED_DAYS,
270 },
271 "action": {"type": "expire"},
272 },
273 ],
274 }
276 # ------------------------------------------------------------------
277 # Output helpers
278 # ------------------------------------------------------------------
279 def _extract_digest(self, push_output: str) -> str | None:
280 """Pull the first ``sha256:...`` digest out of push stdout/stderr."""
281 match = _DIGEST_RE.search(push_output)
282 return match.group(0) if match else None
284 # ------------------------------------------------------------------
285 # ECR repository helpers (used by build/push)
286 # ------------------------------------------------------------------
287 def _runtime_or_error(self) -> str:
288 """Return the detected container runtime, or raise a friendly error."""
289 runtime = detect_container_runtime()
290 if not runtime:
291 from ._container_runtime import container_runtime_error_message
293 raise RuntimeError(container_runtime_error_message(allow_cdk_docker=True))
294 return runtime
296 def _ecr_login(self, runtime: str) -> None:
297 """Authenticate the runtime against the ECR registry."""
298 ecr = self._ecr_client()
299 token = ecr.get_authorization_token()["authorizationData"][0]["authorizationToken"]
300 username, password = base64.b64decode(token).decode().split(":", 1)
301 registry = self._registry_host()
302 result = subprocess.run(
303 [runtime, "login", "-u", username, "--password-stdin", registry],
304 input=password.encode(),
305 capture_output=True,
306 check=False,
307 )
308 if result.returncode != 0:
309 raise RuntimeError(
310 f"{runtime} login to {registry} failed: "
311 f"{result.stderr.decode(errors='replace').strip()}"
312 )
314 def _check_tag_immutable_collision(self, name: str, tag: str) -> None:
315 """Block re-pushing a tag when the repo is immutable.
317 ECR repos can be configured with ``imageTagMutability=IMMUTABLE``,
318 in which case attempting to overwrite an existing tag silently
319 succeeds at build time but fails at push time with a confusing
320 error. Catch this earlier and surface a helpful message.
321 """
322 ecr = self._ecr_client()
323 repo_name = f"{self._repo_prefix}/{name}"
324 try:
325 repo_resp = ecr.describe_repositories(repositoryNames=[repo_name])
326 except ecr.exceptions.RepositoryNotFoundException:
327 return
328 except ClientError as e:
329 code = e.response.get("Error", {}).get("Code", "")
330 if code == "RepositoryNotFoundException":
331 return
332 raise
334 repos = repo_resp.get("repositories", [])
335 if not repos:
336 return
337 mutability = repos[0].get("imageTagMutability", "MUTABLE")
338 if mutability != "IMMUTABLE":
339 return
341 try:
342 existing = ecr.describe_images(
343 repositoryName=repo_name,
344 imageIds=[{"imageTag": tag}],
345 )
346 except ecr.exceptions.ImageNotFoundException:
347 return
348 except ClientError as e:
349 code = e.response.get("Error", {}).get("Code", "")
350 if code == "ImageNotFoundException":
351 return
352 raise
354 if existing.get("imageDetails"):
355 raise RuntimeError(
356 f"Tag {tag!r} already exists on immutable repo "
357 f"{repo_name!r}. Re-run with a different tag, e.g. "
358 f"--tag <new_tag>."
359 )
361 def _apply_retain_tag(self, name: str) -> None:
362 """Apply the ``gco:retain=true`` resource tag to the repository."""
363 ecr = self._ecr_client()
364 ecr.tag_resource(
365 resourceArn=self._repo_arn(name),
366 tags=[{"Key": "gco:retain", "Value": "true"}],
367 )
369 # ------------------------------------------------------------------
370 # build / push
371 # ------------------------------------------------------------------
372 def build(
373 self,
374 context: str,
375 name: str,
376 tag: str | None = None,
377 dockerfile: str = "Dockerfile",
378 build_args: dict[str, str] | None = None,
379 platform: str = "linux/amd64",
380 retain: bool = False,
381 quiet: bool = False,
382 ) -> dict[str, Any]:
383 """Build a container image and push it to the project's ECR repo.
385 Args:
386 context: Build context directory.
387 name: Image name (validated; lowercase letters, digits, dashes).
388 tag: Image tag (defaults to git short SHA, else ``latest``).
389 dockerfile: Path to the Dockerfile, relative to ``context``.
390 build_args: Optional ``KEY=value`` build args.
391 platform: ``--platform`` argument for the build (default
392 ``linux/amd64``).
393 retain: When True, mark the repository with ``gco:retain=true``
394 so it survives stack destroys.
395 quiet: Capture container build output instead of writing it to the
396 command's output stream. Used for machine-readable CLI output.
398 Returns:
399 ``{"image_uri", "digest", "size_bytes", ...}``.
400 """
401 ctx = self._validate_context(context)
402 validated_name = self._validate_name(name)
403 validated_tag = self._validate_tag(tag if tag is not None else self._default_tag())
405 df_path = (ctx / dockerfile).resolve()
406 if not df_path.exists() or not df_path.is_file():
407 raise FileNotFoundError(f"Dockerfile not found: {df_path} (relative to {ctx})")
408 # Use path components: /tmp/app-evil has the string prefix /tmp/app.
409 if not df_path.is_relative_to(ctx):
410 raise ValueError(f"Dockerfile must live inside the build context: {df_path}")
412 runtime = self._runtime_or_error()
413 self.init(name, retain=retain)
414 self._check_tag_immutable_collision(validated_name, validated_tag)
415 self._ecr_login(runtime)
417 full_uri = f"{self._registry_host()}/{self._repo_prefix}/{validated_name}:{validated_tag}"
419 build_cmd: list[str] = [
420 runtime,
421 "build",
422 "-t",
423 full_uri,
424 "--platform",
425 platform,
426 "-f",
427 str(df_path),
428 ]
429 for key, value in (build_args or {}).items():
430 build_cmd.extend(["--build-arg", f"{key}={value}"])
431 build_cmd.append(str(ctx))
433 logger.info("Building image: %s", " ".join(build_cmd))
434 build_run_kwargs: dict[str, Any] = {"cwd": str(ctx)}
435 if quiet:
436 build_run_kwargs.update(capture_output=True, text=True)
437 # check=True: a failed build must not fall through to the push below.
438 subprocess.run(build_cmd, check=True, **build_run_kwargs)
440 push_result = subprocess.run(
441 [runtime, "push", full_uri],
442 capture_output=True,
443 text=True,
444 check=True,
445 cwd=str(ctx),
446 )
447 digest = self._extract_digest((push_result.stdout or "") + (push_result.stderr or ""))
449 if retain:
450 self._apply_retain_tag(validated_name)
452 size_bytes = self._image_size_bytes(validated_name, validated_tag)
454 return {
455 "image_uri": full_uri,
456 "digest": digest,
457 "size_bytes": size_bytes,
458 "runtime": runtime,
459 "repository": f"{self._repo_prefix}/{validated_name}",
460 "tag": validated_tag,
461 "region": self.region,
462 "retain": retain,
463 }
465 def push(
466 self,
467 name: str,
468 tag: str,
469 local_image: str,
470 retain: bool = False,
471 quiet: bool = False,
472 ) -> dict[str, Any]:
473 """Push an already-built local image to the project's ECR repo.
475 Tags ``local_image`` as the project URI before invoking
476 ``<runtime> push``. Skips the build step but otherwise mirrors
477 ``build`` (init repo, login, push, optional retain tag). When ``quiet``
478 is true, the local tag command is captured for machine-readable output.
479 """
480 validated_name = self._validate_name(name)
481 validated_tag = self._validate_tag(tag)
482 if not local_image:
483 raise ValueError("local_image must be a non-empty image reference")
485 runtime = self._runtime_or_error()
486 self.init(name, retain=retain)
487 self._check_tag_immutable_collision(validated_name, validated_tag)
488 self._ecr_login(runtime)
490 full_uri = f"{self._registry_host()}/{self._repo_prefix}/{validated_name}:{validated_tag}"
492 tag_run_kwargs: dict[str, Any] = {}
493 if quiet:
494 tag_run_kwargs.update(capture_output=True, text=True)
495 # check=True: a failed tag must not fall through to the push below.
496 subprocess.run([runtime, "tag", local_image, full_uri], check=True, **tag_run_kwargs)
497 push_result = subprocess.run(
498 [runtime, "push", full_uri],
499 capture_output=True,
500 text=True,
501 check=True,
502 )
503 digest = self._extract_digest((push_result.stdout or "") + (push_result.stderr or ""))
505 if retain:
506 self._apply_retain_tag(validated_name)
508 size_bytes = self._image_size_bytes(validated_name, validated_tag)
510 return {
511 "image_uri": full_uri,
512 "digest": digest,
513 "size_bytes": size_bytes,
514 "runtime": runtime,
515 "repository": f"{self._repo_prefix}/{validated_name}",
516 "tag": validated_tag,
517 "region": self.region,
518 "retain": retain,
519 }
521 def _image_size_bytes(self, name: str, tag: str) -> int | None:
522 """Best-effort ECR lookup for the pushed image size."""
523 ecr = self._ecr_client()
524 try:
525 resp = ecr.describe_images(
526 repositoryName=f"{self._repo_prefix}/{name}",
527 imageIds=[{"imageTag": tag}],
528 )
529 details = resp.get("imageDetails", [])
530 if details:
531 size = details[0].get("imageSizeInBytes")
532 if isinstance(size, int):
533 return size
534 except Exception as e:
535 logger.debug("describe_images for size lookup failed: %s", e)
536 return None
538 # ------------------------------------------------------------------
539 # Read-only methods
540 # ------------------------------------------------------------------
541 def list_repos(self) -> list[dict[str, Any]]:
542 """List every repository under the project's ``gco/`` prefix."""
543 ecr = self._ecr_client()
544 repos: list[dict[str, Any]] = []
545 paginator = ecr.get_paginator("describe_repositories")
546 for page in paginator.paginate():
547 for repo in page.get("repositories", []):
548 repo_name = repo.get("repositoryName", "")
549 if not repo_name.startswith(f"{self._repo_prefix}/"):
550 continue
551 image_count = self._image_count(repo_name)
552 repos.append(
553 {
554 "name": repo_name,
555 "arn": repo.get("repositoryArn"),
556 "uri": repo.get("repositoryUri"),
557 "created_at": _isoformat(repo.get("createdAt")),
558 "image_count": image_count,
559 "tag_mutability": repo.get("imageTagMutability"),
560 }
561 )
562 return repos
564 def _image_count(self, repository_name: str) -> int:
565 """Best-effort count of images in a repository."""
566 ecr = self._ecr_client()
567 try:
568 count = 0
569 paginator = ecr.get_paginator("describe_images")
570 for page in paginator.paginate(repositoryName=repository_name):
571 count += len(page.get("imageDetails", []))
572 return count
573 except Exception as e:
574 logger.debug("describe_images count for %s failed: %s", repository_name, e)
575 return 0
577 def list_tags(self, name: str) -> list[dict[str, Any]]:
578 """List every tag (with digest, pushed date, size) on a repository."""
579 validated = self._validate_name(name)
580 ecr = self._ecr_client()
581 rows: list[dict[str, Any]] = []
582 paginator = ecr.get_paginator("describe_images")
583 for page in paginator.paginate(
584 repositoryName=f"{self._repo_prefix}/{validated}",
585 ):
586 for detail in page.get("imageDetails", []):
587 for tag in detail.get("imageTags", []) or [None]:
588 rows.append(
589 {
590 "tag": tag,
591 "digest": detail.get("imageDigest"),
592 "pushed_at": _isoformat(detail.get("imagePushedAt")),
593 "size_bytes": detail.get("imageSizeInBytes"),
594 }
595 )
596 return rows
598 def describe(self, name: str, tag: str) -> dict[str, Any]:
599 """Return the full ECR image details for a single tag."""
600 validated_name = self._validate_name(name)
601 validated_tag = self._validate_tag(tag)
602 ecr = self._ecr_client()
603 resp = ecr.describe_images(
604 repositoryName=f"{self._repo_prefix}/{validated_name}",
605 imageIds=[{"imageTag": validated_tag}],
606 )
607 details = resp.get("imageDetails", [])
608 if not details:
609 return {}
610 detail = details[0]
611 return {
612 "name": f"{self._repo_prefix}/{validated_name}",
613 "tag": validated_tag,
614 "digest": detail.get("imageDigest"),
615 "pushed_at": _isoformat(detail.get("imagePushedAt")),
616 "size_bytes": detail.get("imageSizeInBytes"),
617 "tags": detail.get("imageTags", []),
618 "scan_findings_summary": detail.get("imageScanFindingsSummary"),
619 }
621 def get_uri(self, name: str, tag: str = "latest") -> str:
622 """Return the full registry URI for ``name:tag``. No API call."""
623 validated_name = self._validate_name(name)
624 validated_tag = self._validate_tag(tag)
625 return f"{self._registry_host()}/{self._repo_prefix}/{validated_name}:{validated_tag}"
627 def list_maintained_images(self, tag: str = "latest") -> list[dict[str, Any]]:
628 """List the first-party images GCO builds and ships.
630 Returns one row per shipped image with its logical name, the
631 ``gco/<name>`` repository, the Dockerfile that produces it, and
632 the registry URI for ``tag``. No API call — the catalog is
633 resolved locally from the shipped Dockerfile set.
634 """
635 rows: list[dict[str, Any]] = []
636 for name, dockerfile in _MAINTAINED_IMAGES.items():
637 rows.append(
638 {
639 "name": name,
640 "repository": f"{self._repo_prefix}/{name}",
641 "dockerfile": dockerfile,
642 "uri": self.get_uri(name, tag),
643 }
644 )
645 return rows
647 def get_maintained_image(self, name: str, tag: str = "latest") -> dict[str, Any]:
648 """Resolve a single shipped image by name.
650 Raises ``ValueError`` for a name that is not one of the shipped
651 images, listing the known names so the caller can correct the
652 lookup.
653 """
654 dockerfile = _MAINTAINED_IMAGES.get(name)
655 if dockerfile is None:
656 known = ", ".join(sorted(_MAINTAINED_IMAGES))
657 raise ValueError(f"Unknown maintained image: {name!r}. Known images: {known}.")
658 return {
659 "name": name,
660 "repository": f"{self._repo_prefix}/{name}",
661 "dockerfile": dockerfile,
662 "uri": self.get_uri(name, tag),
663 }
665 def default_disaggregated_image_uri(self, tag: str | None = None) -> str:
666 """Return the image reference disaggregated prefill/decode deploys serve from.
668 As of the pinned tag, the upstream ``vllm/vllm-openai`` image bundles
669 the Mooncake transfer engine, so GCO no longer builds its own image —
670 deploys pull this upstream image from Docker Hub directly. Returns the
671 pinned reference; ``tag``, when given, overrides only the version.
672 """
673 if tag:
674 repo = _DISAGGREGATED_DEFAULT_IMAGE.rsplit(":", 1)[0]
675 return f"{repo}:{tag}"
676 return _DISAGGREGATED_DEFAULT_IMAGE
678 def _current_replication_configuration(self, ecr: Any) -> tuple[str | None, dict[str, Any]]:
679 """Return the registry ID and current ECR replication configuration."""
680 try:
681 response = ecr.get_replication_configuration()
682 except ClientError as e:
683 code = e.response.get("Error", {}).get("Code", "")
684 if code == "ReplicationConfigurationNotFoundException":
685 return None, {"rules": []}
686 raise
688 configuration = response.get("replicationConfiguration") or {"rules": []}
689 if not isinstance(configuration, dict):
690 configuration = {"rules": []}
691 if not isinstance(configuration.get("rules"), list):
692 configuration = {**configuration, "rules": []}
693 return response.get("registryId"), configuration
695 def replication_get(self) -> dict[str, Any]:
696 """Return the current ECR replication configuration, or ``{}``.
698 The historical ``policy`` response key is retained for CLI/API
699 compatibility, but its value now comes from the replication API rather
700 than the unrelated registry-permissions policy API.
701 """
702 registry_id, configuration = self._current_replication_configuration(self._ecr_client())
703 if not configuration.get("rules"):
704 return {}
705 return {
706 "registryId": registry_id,
707 "policy": configuration,
708 }
710 def _replication_regions(self) -> list[str]:
711 """Resolve deployed regional destinations from supported config data."""
712 deployment_regions = _load_cdk_json().get("regional", [])
713 candidates = deployment_regions if isinstance(deployment_regions, list) else []
714 if not candidates:
715 default_region = getattr(self.config, "default_region", None)
716 if isinstance(default_region, str) and default_region:
717 candidates = [default_region]
719 # Preserve declaration order while removing invalid values/duplicates.
720 return list(
721 dict.fromkeys(region for region in candidates if isinstance(region, str) and region)
722 )
724 def replication_status(self) -> list[dict[str, Any]]:
725 """Per-repo replication status across the project repos."""
726 ecr = self._ecr_client()
727 rows: list[dict[str, Any]] = []
728 for repo in self.list_repos():
729 repo_name = repo["name"]
730 paginator = ecr.get_paginator("describe_images")
731 try:
732 for page in paginator.paginate(repositoryName=repo_name):
733 for detail in page.get("imageDetails", []):
734 digest = detail.get("imageDigest")
735 try:
736 status = ecr.describe_image_replication_status(
737 repositoryName=repo_name,
738 imageId={"imageDigest": digest},
739 )
740 for entry in status.get("replicationStatuses", []):
741 rows.append(
742 {
743 "repository": repo_name,
744 "digest": digest,
745 "region": entry.get("region"),
746 "status": entry.get("status"),
747 "registry_id": entry.get("registryId"),
748 }
749 )
750 except (ClientError, AttributeError) as e:
751 logger.debug(
752 "describe_image_replication_status failed for %s %s: %s",
753 repo_name,
754 digest,
755 e,
756 )
757 except ClientError as e:
758 logger.debug("describe_images failed for %s: %s", repo_name, e)
759 return rows
761 # ------------------------------------------------------------------
762 # Administrative methods
763 # ------------------------------------------------------------------
764 def init(self, name: str, retain: bool = False) -> dict[str, Any]:
765 """Create the project repository idempotently with default lifecycle.
767 ``CreateRepository`` is invoked with ``imageTagMutability=MUTABLE``
768 and ``scanOnPush=True``. If the repository already exists, the
769 method becomes a no-op for repository creation but still applies
770 the default lifecycle policy and the optional ``gco:retain`` tag.
771 """
772 validated = self._validate_name(name)
773 repo_name = f"{self._repo_prefix}/{validated}"
774 ecr = self._ecr_client()
776 created = False
777 try:
778 ecr.create_repository(
779 repositoryName=repo_name,
780 imageTagMutability="MUTABLE",
781 imageScanningConfiguration={"scanOnPush": True},
782 tags=[
783 {"Key": "Project", "Value": self.config.project_name},
784 ],
785 )
786 created = True
787 except ecr.exceptions.RepositoryAlreadyExistsException:
788 # Idempotent init — re-running ``gco images init`` against an
789 # already-provisioned repo is a no-op for create_repository.
790 # We still flow through the lifecycle/retain blocks below so
791 # any drift in policy is healed on every call.
792 logger.debug("repository %s already exists; skipping create", repo_name)
793 except ClientError as e:
794 code = e.response.get("Error", {}).get("Code", "")
795 if code != "RepositoryAlreadyExistsException":
796 raise
798 try:
799 ecr.put_lifecycle_policy(
800 repositoryName=repo_name,
801 lifecyclePolicyText=json.dumps(self._default_lifecycle_policy()),
802 )
803 except ClientError as e:
804 logger.debug("put_lifecycle_policy on %s failed: %s", repo_name, e)
806 if retain:
807 try:
808 self._apply_retain_tag(validated)
809 except ClientError as e:
810 logger.debug("apply retain tag on %s failed: %s", repo_name, e)
812 return {
813 "name": repo_name,
814 "created": created,
815 "retain": retain,
816 }
818 def lifecycle_get(self, name: str) -> dict[str, Any]:
819 """Return the lifecycle policy on a repository, or ``{}``."""
820 validated = self._validate_name(name)
821 ecr = self._ecr_client()
822 try:
823 resp = ecr.get_lifecycle_policy(
824 repositoryName=f"{self._repo_prefix}/{validated}",
825 )
826 policy_text = resp.get("lifecyclePolicyText")
827 if policy_text:
828 return {
829 "name": f"{self._repo_prefix}/{validated}",
830 "policy": json.loads(policy_text),
831 }
832 except ecr.exceptions.LifecyclePolicyNotFoundException:
833 return {}
834 except ClientError as e:
835 code = e.response.get("Error", {}).get("Code", "")
836 if code == "LifecyclePolicyNotFoundException":
837 return {}
838 raise
839 return {}
841 def lifecycle_set(self, name: str, policy: dict[str, Any]) -> dict[str, Any]:
842 """Replace the lifecycle policy on a repository."""
843 validated = self._validate_name(name)
844 ecr = self._ecr_client()
845 resp = ecr.put_lifecycle_policy(
846 repositoryName=f"{self._repo_prefix}/{validated}",
847 lifecyclePolicyText=json.dumps(policy),
848 )
849 return {
850 "name": f"{self._repo_prefix}/{validated}",
851 "registry_id": resp.get("registryId"),
852 "policy": policy,
853 }
855 def replication_sync(self) -> dict[str, Any]:
856 """Apply this project's replication rule without clobbering others.
858 Existing rules for unrelated repository prefixes are retained. If no
859 non-source destination can be resolved, no write is made; this avoids
860 replacing a valid registry configuration with an empty rule set.
861 """
862 ecr = self._ecr_client()
863 registry_id, current = self._current_replication_configuration(ecr)
864 destinations = [region for region in self._replication_regions() if region != self.region]
866 if not destinations:
867 return {
868 "configuration": current,
869 "destinations": [],
870 "registry_id": registry_id,
871 "updated": False,
872 }
874 account = self._account_id()
875 managed_filter = {
876 "filter": f"{self._repo_prefix}/",
877 "filterType": "PREFIX_MATCH",
878 }
879 managed_rule = {
880 "destinations": [{"region": region, "registryId": account} for region in destinations],
881 "repositoryFilters": [managed_filter],
882 }
884 preserved_rules: list[dict[str, Any]] = []
885 for existing_rule in current.get("rules", []):
886 if not isinstance(existing_rule, dict):
887 continue
888 filters = existing_rule.get("repositoryFilters") or []
889 managed_filters = [item for item in filters if item == managed_filter]
890 if not managed_filters:
891 preserved_rules.append(existing_rule)
892 continue
894 # A rule can contain filters for multiple prefixes. Retain the
895 # unrelated filters with their original destinations while replacing
896 # only this project's managed filter.
897 unrelated_filters = [item for item in filters if item != managed_filter]
898 if unrelated_filters:
899 preserved_rules.append({**existing_rule, "repositoryFilters": unrelated_filters})
901 configuration = {
902 **current,
903 "rules": [*preserved_rules, managed_rule],
904 }
905 response = ecr.put_replication_configuration(replicationConfiguration=configuration)
906 return {
907 "configuration": configuration,
908 "destinations": destinations,
909 "registry_id": response.get("registryId") or registry_id or account,
910 "updated": True,
911 }
913 # ------------------------------------------------------------------
914 # Destructive methods
915 # ------------------------------------------------------------------
916 def delete_tag(self, name: str, tag: str) -> dict[str, Any]:
917 """Delete a single tag from a repository."""
918 validated_name = self._validate_name(name)
919 validated_tag = self._validate_tag(tag)
920 ecr = self._ecr_client()
921 resp = ecr.batch_delete_image(
922 repositoryName=f"{self._repo_prefix}/{validated_name}",
923 imageIds=[{"imageTag": validated_tag}],
924 )
925 return {
926 "name": f"{self._repo_prefix}/{validated_name}",
927 "tag": validated_tag,
928 "deleted": [
929 {"digest": d.get("imageDigest"), "tag": d.get("imageTag")}
930 for d in resp.get("imageIds", [])
931 ],
932 "failures": resp.get("failures", []),
933 }
935 def delete_repo(self, name: str, force: bool = False) -> dict[str, Any]:
936 """Delete a repository (optionally including its images)."""
937 validated = self._validate_name(name)
938 ecr = self._ecr_client()
939 resp = ecr.delete_repository(
940 repositoryName=f"{self._repo_prefix}/{validated}",
941 force=force,
942 )
943 return {
944 "name": f"{self._repo_prefix}/{validated}",
945 "deleted": True,
946 "registry_id": resp.get("repository", {}).get("registryId"),
947 }
949 def cleanup(
950 self,
951 name: str | None = None,
952 all: bool = False,
953 ) -> dict[str, Any]:
954 """Delete every untagged image across one or all project repos."""
955 if not name and not all:
956 raise ValueError("cleanup() requires either a name or all=True")
958 repos: list[str]
959 if name:
960 validated = self._validate_name(name)
961 repos = [f"{self._repo_prefix}/{validated}"]
962 else:
963 repos = [r["name"] for r in self.list_repos()]
965 ecr = self._ecr_client()
966 repos_touched = 0
967 tags_deleted = 0
968 bytes_freed = 0
970 for repo_name in repos:
971 untagged_ids: list[dict[str, str]] = []
972 untagged_size = 0
973 try:
974 paginator = ecr.get_paginator("describe_images")
975 for page in paginator.paginate(
976 repositoryName=repo_name,
977 filter={"tagStatus": "UNTAGGED"},
978 ):
979 for detail in page.get("imageDetails", []):
980 digest = detail.get("imageDigest")
981 if not digest:
982 continue
983 untagged_ids.append({"imageDigest": digest})
984 size = detail.get("imageSizeInBytes") or 0
985 if isinstance(size, int):
986 untagged_size += size
987 except ClientError as e:
988 logger.debug("describe_images for cleanup of %s failed: %s", repo_name, e)
989 continue
991 if not untagged_ids:
992 continue
993 repos_touched += 1
994 # batch_delete_image accepts up to 100 ids per call.
995 for chunk_start in range(0, len(untagged_ids), 100):
996 chunk = untagged_ids[chunk_start : chunk_start + 100]
997 resp = ecr.batch_delete_image(
998 repositoryName=repo_name,
999 imageIds=chunk,
1000 )
1001 tags_deleted += len(resp.get("imageIds", []))
1002 bytes_freed += untagged_size
1004 return {
1005 "repos_touched": repos_touched,
1006 "tags_deleted": tags_deleted,
1007 "bytes_freed": bytes_freed,
1008 }
1010 def prune(self, dry_run: bool = True) -> dict[str, Any]:
1011 """Remove untagged images older than 30 days.
1013 Returns the same shape as ``cleanup``; when ``dry_run`` is True
1014 (the default), no images are deleted.
1015 """
1016 cutoff = datetime.now(UTC) - timedelta(days=30)
1017 ecr = self._ecr_client()
1018 repos_touched = 0
1019 tags_deleted = 0
1020 bytes_freed = 0
1022 for repo in self.list_repos():
1023 repo_name = repo["name"]
1024 stale_ids: list[dict[str, str]] = []
1025 stale_size = 0
1026 try:
1027 paginator = ecr.get_paginator("describe_images")
1028 for page in paginator.paginate(
1029 repositoryName=repo_name,
1030 filter={"tagStatus": "UNTAGGED"},
1031 ):
1032 for detail in page.get("imageDetails", []):
1033 pushed = detail.get("imagePushedAt")
1034 if pushed and pushed >= cutoff:
1035 continue
1036 digest = detail.get("imageDigest")
1037 if not digest:
1038 continue
1039 stale_ids.append({"imageDigest": digest})
1040 size = detail.get("imageSizeInBytes") or 0
1041 if isinstance(size, int):
1042 stale_size += size
1043 except ClientError as e:
1044 logger.debug("describe_images for prune of %s failed: %s", repo_name, e)
1045 continue
1047 if not stale_ids:
1048 continue
1049 repos_touched += 1
1050 tags_deleted += len(stale_ids)
1051 bytes_freed += stale_size
1052 if dry_run:
1053 continue
1054 for chunk_start in range(0, len(stale_ids), 100):
1055 chunk = stale_ids[chunk_start : chunk_start + 100]
1056 ecr.batch_delete_image(
1057 repositoryName=repo_name,
1058 imageIds=chunk,
1059 )
1061 return {
1062 "dry_run": dry_run,
1063 "repos_touched": repos_touched,
1064 "tags_deleted": tags_deleted,
1065 "bytes_freed": bytes_freed,
1066 }
1068 def orphans(self, threshold_days: int = 30) -> list[dict[str, Any]]:
1069 """List ``gco/*`` tags older than ``threshold_days`` with no references.
1071 Cross-references against:
1072 * inference endpoint specs (via :class:`cli.inference.InferenceManager`),
1073 * recent jobs (best-effort; returns empty for the jobs side when
1074 the queue table schema is unavailable).
1075 """
1076 cutoff = datetime.now(UTC) - timedelta(days=threshold_days)
1077 referenced: set[str] = set()
1078 referenced.update(self._collect_inference_image_refs())
1079 referenced.update(self._collect_recent_job_image_refs(threshold_days))
1081 rows: list[dict[str, Any]] = []
1082 for repo in self.list_repos():
1083 repo_name = repo["name"]
1084 for tag_row in self.list_tags(repo_name.removeprefix(f"{self._repo_prefix}/")):
1085 tag = tag_row.get("tag")
1086 if not tag:
1087 continue
1088 pushed = self._parse_iso(tag_row.get("pushed_at"))
1089 if pushed and pushed >= cutoff:
1090 continue
1091 uri = f"{self._registry_host()}/{repo_name}:{tag}"
1092 if uri in referenced:
1093 continue
1094 rows.append(
1095 {
1096 "repository": repo_name,
1097 "tag": tag,
1098 "digest": tag_row.get("digest"),
1099 "pushed_at": tag_row.get("pushed_at"),
1100 "uri": uri,
1101 }
1102 )
1103 return rows
1105 def _collect_inference_image_refs(self) -> set[str]:
1106 """Return every image URI referenced by a registered inference endpoint."""
1107 try:
1108 from .inference import InferenceManager
1109 except Exception as e:
1110 logger.debug("InferenceManager unavailable: %s", e)
1111 return set()
1112 try:
1113 manager = InferenceManager(self.config)
1114 endpoints = manager.list_endpoints()
1115 except Exception as e:
1116 logger.debug("list_endpoints failed: %s", e)
1117 return set()
1118 refs: set[str] = set()
1119 for ep in endpoints or []:
1120 spec = ep.get("spec") or {}
1121 image = spec.get("image") if isinstance(spec, dict) else None
1122 if image:
1123 refs.add(image)
1124 canary = spec.get("canary") if isinstance(spec, dict) else None
1125 if isinstance(canary, dict) and canary.get("image"):
1126 refs.add(canary["image"])
1127 return refs
1129 def _collect_recent_job_image_refs(self, threshold_days: int = 30) -> set[str]:
1130 """Return image URIs referenced by jobs newer than ``threshold_days``.
1132 Walks every deployed region via :class:`cli.jobs.JobManager` and
1133 unions the ``image_refs`` field on each ``JobInfo`` whose
1134 ``created_time`` is within the cutoff. Treats jobs without a
1135 ``created_time`` as in-window so a freshly-submitted job that
1136 hasn't yet been picked up by the cluster's status loop isn't
1137 accidentally considered orphaned.
1139 Best-effort: any per-region failure is logged at debug and
1140 skipped so the orphan scan still completes against the
1141 regions that did respond. The deferred import breaks an
1142 otherwise-circular ``cli.images`` ↔ ``cli.jobs`` dependency.
1143 """
1144 try:
1145 from .jobs import JobManager
1146 except Exception as e:
1147 logger.debug("JobManager unavailable: %s", e)
1148 return set()
1150 try:
1151 manager = JobManager(self.config)
1152 except Exception as e:
1153 logger.debug("JobManager init failed: %s", e)
1154 return set()
1156 try:
1157 jobs = manager.list_jobs(all_regions=True)
1158 except Exception as e:
1159 logger.debug("list_jobs(all_regions=True) failed: %s", e)
1160 return set()
1162 cutoff = datetime.now(UTC) - timedelta(days=threshold_days)
1163 refs: set[str] = set()
1164 for job in jobs or []:
1165 created = getattr(job, "created_time", None)
1166 if isinstance(created, datetime):
1167 created_aware = created if created.tzinfo else created.replace(tzinfo=UTC)
1168 if created_aware < cutoff:
1169 continue
1170 image_refs = getattr(job, "image_refs", None) or []
1171 for ref in image_refs:
1172 if isinstance(ref, str) and ref:
1173 refs.add(ref)
1174 return refs
1176 @staticmethod
1177 def _parse_iso(value: Any) -> datetime | None:
1178 """Parse an ISO-8601 string into a tz-aware datetime, else None."""
1179 if isinstance(value, datetime):
1180 return value if value.tzinfo else value.replace(tzinfo=UTC)
1181 if not isinstance(value, str):
1182 return None
1183 try:
1184 parsed = datetime.fromisoformat(value)
1185 except ValueError:
1186 return None
1187 return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
1190def _isoformat(value: Any) -> str | None:
1191 """Return ISO-8601 form of a datetime, or pass-through for strings."""
1192 if value is None:
1193 return None
1194 if isinstance(value, datetime):
1195 return value.isoformat()
1196 return str(value)
1199def get_image_manager(config: GCOConfig | None = None, region: str | None = None) -> ImageManager:
1200 """Factory function for ``ImageManager``."""
1201 return ImageManager(config=config, region=region)
1204def default_disaggregated_image(
1205 config: GCOConfig | None = None,
1206 region: str | None = None,
1207 tag: str | None = None,
1208) -> str:
1209 """Resolve the default image reference for disaggregated prefill/decode deploys.
1211 Convenience wrapper around
1212 :meth:`ImageManager.default_disaggregated_image_uri` for callers
1213 that only need the reference and do not otherwise hold a manager.
1214 """
1215 return get_image_manager(config=config, region=region).default_disaggregated_image_uri(tag)