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