homelab/kubernetes/apps/gitea/package-retention.py

"""Prune old CI image builds from Gitea's container registry (REN-1203).

See package-retention.yaml for what is pruned, what is kept, and why.
Standard library only.
"""
import json
import os
import re
import ssl
import sys
import urllib.parse
import urllib.request
from datetime import datetime

GITEA = os.environ["GITEA_URL"].rstrip("/")
TOKEN = os.environ["GITEA_TOKEN"]
REGISTRY_HOST = os.environ["REGISTRY_HOST"]
OWNERS = os.environ["OWNERS"].split(",")
KEEP = int(os.environ["KEEP_IMAGES"])
GROUP_SECONDS = int(os.environ["GROUP_SECONDS"])
MAX_DELETES = int(os.environ["MAX_DELETES"])
EXCLUDE = re.compile(os.environ["EXCLUDE_PACKAGES"]) if os.environ.get("EXCLUDE_PACKAGES") else None
DRY_RUN = os.environ.get("DRY_RUN", "false") == "true"

# A package is a CI build package if at least one of its tags looks like this.
BUILD_TAG = re.compile(r"^(build\.[0-9]+|main-[0-9a-f]{7,40})$")
ALWAYS_KEEP = {"main", "latest"}

SA = "/var/run/secrets/kubernetes.io/serviceaccount"


def log(msg):
    print(msg, flush=True)


def gitea(method, path):
    req = urllib.request.Request(
        GITEA + path, method=method, headers={"Authorization": "token " + TOKEN}
    )
    with urllib.request.urlopen(req, timeout=60) as resp:
        body = resp.read()
        return json.loads(body) if body else None


def ts(s):
    # Gitea returns RFC 3339, e.g. 2026-10-07T18:07:32Z or +02:00.
    return datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp()


def running_images():
    """Tags in use by any pod, per package. Packages referenced by digest only
    cannot be matched to a tag, so they are returned in `unsafe` and skipped."""
    ctx = ssl.create_default_context(cafile=SA + "/ca.crt")
    with open(SA + "/token") as f:
        sa_token = f.read().strip()
    req = urllib.request.Request(
        "https://kubernetes.default.svc/api/v1/pods",
        headers={"Authorization": "Bearer " + sa_token},
    )
    with urllib.request.urlopen(req, timeout=60, context=ctx) as resp:
        pods = json.load(resp)["items"]
    in_use, unsafe = {}, set()
    prefix = REGISTRY_HOST + "/"
    for pod in pods:
        spec = pod["spec"]
        for c in (
            spec.get("containers", [])
            + spec.get("initContainers", [])
            + spec.get("ephemeralContainers", [])
        ):
            image = c["image"]
            if not image.startswith(prefix):
                continue
            ref = image[len(prefix):].split("@", 1)[0]
            repo, _, tag = ref.rpartition(":")
            if not repo or "/" in tag:
                unsafe.add(ref)
                continue
            in_use.setdefault(repo, set()).add(tag)
    return in_use, unsafe


def list_packages(owner):
    out, page = [], 1
    while True:
        batch = gitea(
            "GET", f"/api/v1/packages/{owner}?type=container&limit=50&page={page}"
        )
        if not batch:
            return out
        out.extend(batch)
        page += 1


def plan(versions, protected):
    """Group tags into images (one push writes all tags of an image within a
    few seconds) and return the tags of the images to delete."""
    tags = sorted(
        (v for v in versions if not v["version"].startswith("sha256:")),
        key=lambda v: ts(v["created_at"]),
        reverse=True,
    )
    images, last = [], None
    for v in tags:
        t = ts(v["created_at"])
        if last is None or last - t > GROUP_SECONDS:
            images.append([])
        images[-1].append(v["version"])
        last = t
    delete = []
    for i, image in enumerate(images):
        if i < KEEP or set(image) & (ALWAYS_KEEP | protected):
            continue
        delete.extend(image)
    return images, delete


def main():
    in_use, unsafe = running_images()
    log(f"running tags from {REGISTRY_HOST}: {sum(len(t) for t in in_use.values())}")
    work = []
    for owner in OWNERS:
        by_name = {}
        for v in list_packages(owner):
            by_name.setdefault(v["name"], []).append(v)
        for name, versions in sorted(by_name.items()):
            full = f"{owner}/{name}"
            if EXCLUDE and EXCLUDE.search(full):
                log(f"skip {full}: excluded")
                continue
            if not any(BUILD_TAG.match(v["version"]) for v in versions):
                continue
            if full in unsafe:
                log(f"skip {full}: a pod pins it by digest only")
                continue
            protected = in_use.get(full, set())
            images, delete = plan(versions, protected)
            log(f"{full}: {len(images)} images, {len(delete)} tags to delete, running {sorted(protected)}")
            work += [(owner, name, tag) for tag in delete]
    if len(work) > MAX_DELETES:
        log(f"abort: {len(work)} deletes exceed MAX_DELETES={MAX_DELETES}; nothing deleted")
        return 1
    failed = 0
    for owner, name, tag in work:
        if DRY_RUN:
            log(f"dry-run: would delete {owner}/{name}:{tag}")
            continue
        path = "/api/v1/packages/{}/container/{}/{}".format(
            owner, urllib.parse.quote(name, safe=""), urllib.parse.quote(tag, safe="")
        )
        try:
            gitea("DELETE", path)
            log(f"deleted {owner}/{name}:{tag}")
        except Exception as e:  # keep going; report at the end
            failed += 1
            log(f"FAILED {owner}/{name}:{tag}: {e}")
    log(f"done: {len(work) - failed} deleted, {failed} failed, dry_run={DRY_RUN}")
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())