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