diff --git a/.forgejo/scripts/registry_prune.py b/.forgejo/scripts/registry_prune.py index 12c40ec..18d62f5 100644 --- a/.forgejo/scripts/registry_prune.py +++ b/.forgejo/scripts/registry_prune.py @@ -1,19 +1,22 @@ #!/usr/bin/env python3 """ -Registry Image Tag Pruner - Keeps only the latest 1 SHA-tag per repository. +Registry Image Tag Pruner - Keeps only the latest N SHA-tag(s) per repository. Usage: python3 registry_prune.py \ --registry registry.nxtgauge.com \ --repo nxtgauge-rust-gateway \ --username "$REGISTRY_USERNAME" \ - --password "$REGISTRY_PASSWORD" + --password "$REGISTRY_PASSWORD" \ + --protect "$SHA" Environment variables can also be used: REGISTRY_HOST, REGISTRY_REPO, REGISTRY_USERNAME, REGISTRY_PASSWORD SHA-like tags are identified by pattern: ^[a-f0-9]{40}$ Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are NEVER deleted. +--protect tags (e.g. the SHA this very CI run just built) are NEVER deleted, +regardless of how they sort - this run's own image must survive its own prune. Exit code: 0 on success (or if prune fails gracefully), non-zero only on critical error. """ @@ -27,16 +30,29 @@ import time from urllib.request import Request, urlopen from urllib.error import URLError, HTTPError +MANIFEST_ACCEPT = ", ".join([ + "application/vnd.docker.distribution.manifest.list.v2+json", + "application/vnd.docker.distribution.manifest.v2+json", + "application/vnd.oci.image.index.v1+json", + "application/vnd.oci.image.manifest.v1+json", +]) + def parse_args(): parser = argparse.ArgumentParser( - description="Prune Docker registry tags, keeping only the latest SHA tag." + description="Prune Docker registry tags, keeping only the latest SHA tag(s)." ) parser.add_argument("--registry", default=os.environ.get("REGISTRY_HOST")) parser.add_argument("--repo", default=os.environ.get("REGISTRY_REPO")) parser.add_argument("--username", default=os.environ.get("REGISTRY_USERNAME")) parser.add_argument("--password", default=os.environ.get("REGISTRY_PASSWORD")) parser.add_argument("--keep", type=int, default=1, help="Number of SHA tags to keep (default: 1)") + parser.add_argument( + "--protect", + action="append", + default=[], + help="Tag that must never be deleted (e.g. the SHA this CI run just built). Repeatable.", + ) return parser.parse_args() @@ -76,23 +92,68 @@ def api_request(url: str, method: str, username: str, password: str, data=None, return None -def get_tag_digest(registry: str, repo: str, tag: str, username: str, password: str) -> tuple[str, str] | None: - """Get the digest (sha256:...) and created time for a tag.""" - url = f"https://{registry}/v2/{repo}/manifests/{tag}" +def _get_manifest(registry: str, repo: str, ref: str, username: str, password: str): + """GET a manifest by tag or digest. Returns (digest, media_type, body_json) or None.""" + url = f"https://{registry}/v2/{repo}/manifests/{ref}" auth = base64.b64encode(f"{username}:{password}".encode()).decode() for attempt in range(1, 4): try: req = Request(url, method="GET", headers={ "Authorization": f"Basic {auth}", - "Accept": "application/vnd.docker.distribution.manifest.v2+json", + "Accept": MANIFEST_ACCEPT, }) with urlopen(req, timeout=30) as response: digest = response.headers.get("Docker-Content-Digest", "") - created = response.headers.get("Date", "") - return digest, created + media_type = response.headers.get("Content-Type", "") + body = json.loads(response.read()) + return digest, media_type, body except Exception as e: - print(f" [RETRY {attempt}/3] Getting digest for {tag}: {e}") + print(f" [RETRY {attempt}/3] Fetching manifest {ref}: {e}") + time.sleep(attempt) + + return None + + +def get_tag_created(registry: str, repo: str, tag: str, username: str, password: str) -> tuple[str, str] | None: + """ + Get the digest and true build timestamp for a tag, by reading the + "created" field baked into the image's config blob at build time - + NOT the HTTP response's Date header, which only reflects the moment + this GET request happened (every tag queried in the same prune run + ends up with a near-identical Date, so sorting by it is close to + random and can rank a tag pushed seconds ago as the oldest one). + """ + result = _get_manifest(registry, repo, tag, username, password) + if result is None: + return None + tag_digest, media_type, body = result + + # A tag can point at a multi-platform manifest list/index rather than + # a single image manifest - descend into the first platform entry to + # reach an actual image manifest with a "config" pointer. + if "manifests" in body and body["manifests"]: + child_digest = body["manifests"][0]["digest"] + child = _get_manifest(registry, repo, child_digest, username, password) + if child is None: + return None + _, _, body = child + + config = body.get("config") + if not config or "digest" not in config: + return None + + blob_url = f"https://{registry}/v2/{repo}/blobs/{config['digest']}" + auth = base64.b64encode(f"{username}:{password}".encode()).decode() + for attempt in range(1, 4): + try: + req = Request(blob_url, method="GET", headers={"Authorization": f"Basic {auth}"}) + with urlopen(req, timeout=30) as response: + config_json = json.loads(response.read()) + created = config_json.get("created", "") + return tag_digest, created + except Exception as e: + print(f" [RETRY {attempt}/3] Fetching config blob for {tag}: {e}") time.sleep(attempt) return None @@ -129,19 +190,22 @@ def is_sha_tag(tag: str) -> bool: return bool(re.match(r"^[a-f0-9]{40}$", tag)) -def prune_tags(registry: str, repo: str, username: str, password: str, keep: int = 1) -> bool: +def prune_tags(registry: str, repo: str, username: str, password: str, keep: int = 1, protect: list[str] = ()) -> bool: """ Main prune logic: - List all tags for the repo - Filter SHA-like tags - - Sort by created date (newest first) - - Keep newest `keep` tags + - Sort by the image's actual build timestamp (newest first) + - Keep newest `keep` tags, plus anything in `protect` - Delete older SHA tags by digest - - Never delete non-SHA tags + - Never delete non-SHA tags or protected tags """ print(f"\n=== Pruning {registry}/{repo} ===") print(f"Strategy: Keep {keep} newest SHA tag(s), delete older SHA tags") - print(f"Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are preserved\n") + print(f"Non-SHA tags (e.g., high-performance-latest, main-latest, latest) are preserved") + if protect: + print(f"Protected SHA tags (never deleted): {', '.join(protect)}") + print() # Get catalog (list of repos) catalog_url = f"https://{registry}/v2/_catalog" @@ -180,17 +244,17 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int print("\n[INFO] No SHA tags to prune") return True - # Get digest and created time for each SHA tag + # Get digest and true build time for each SHA tag tag_info = [] for tag in sha_tags: - result = get_tag_digest(registry, repo, tag, username, password) + result = get_tag_created(registry, repo, tag, username, password) if result: digest, created = result tag_info.append({ "tag": tag, "digest": digest, "created": created, - "timestamp": parse_http_date(created) if created else 0, + "timestamp": parse_image_date(created) if created else 0, }) time.sleep(0.1) # Be nice to the registry @@ -201,17 +265,19 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int # Sort by timestamp (newest first) tag_info.sort(key=lambda x: x["timestamp"], reverse=True) - print(f"\nSHA tags sorted by age (newest first):") + print(f"\nSHA tags sorted by build time (newest first):") for i, info in enumerate(tag_info): - marker = " [KEEP]" if i < keep else " [DELETE]" + is_protected = info["tag"] in protect + will_keep = i < keep or is_protected + marker = " [KEEP]" + (" (protected)" if is_protected else "") if will_keep else " [DELETE]" print(f" {i+1}. {info['tag']} ({info['created'] or 'unknown date'}){marker}") - # Delete older SHA tags + # Delete older, unprotected SHA tags deleted_count = 0 kept_count = 0 for i, info in enumerate(tag_info): - if i < keep: + if i < keep or info["tag"] in protect: print(f"\n[KEEP] {info['tag']}") kept_count += 1 continue @@ -233,11 +299,11 @@ def prune_tags(registry: str, repo: str, username: str, password: str, keep: int return True -def parse_http_date(date_str: str) -> float: - """Parse HTTP Date header to timestamp.""" - from email.utils import parsedate_to_datetime +def parse_image_date(date_str: str) -> float: + """Parse an OCI image config's ISO-8601 "created" field to a timestamp.""" + from datetime import datetime try: - return parsedate_to_datetime(date_str).timestamp() + return datetime.fromisoformat(date_str.replace("Z", "+00:00")).timestamp() except Exception: return 0 @@ -261,7 +327,7 @@ def main(): print(f"Username: {username}") try: - success = prune_tags(registry, repo, username, password, args.keep) + success = prune_tags(registry, repo, username, password, args.keep, args.protect) if success: print("\n[OK] Prune completed successfully") sys.exit(0) diff --git a/.forgejo/workflows/build.yaml b/.forgejo/workflows/build.yaml index b64f290..2fde0b7 100644 --- a/.forgejo/workflows/build.yaml +++ b/.forgejo/workflows/build.yaml @@ -87,6 +87,7 @@ jobs: REGISTRY_NAMESPACE: ${{ secrets.REGISTRY_NAMESPACE || 'ashwin' }} REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }} REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }} + SHA: ${{ github.sha }} run: | set -euo pipefail python3 .forgejo/scripts/registry_prune.py \ @@ -94,6 +95,7 @@ jobs: --repo "$REGISTRY_NAMESPACE/nxtgauge-frontend-solid" \ --username "$REGISTRY_USERNAME" \ --password "$REGISTRY_PASSWORD" \ + --protect "$SHA" \ --keep 2 - name: Update GitOps release