fix: expand g14 registry gc retention

2026-05-30 05:40:16 +00:00
parent ce519dbc0c
commit 231d17ebab
6 changed files with 221 additions and 27 deletions
@@ -104,11 +104,10 @@ function parseRemoteGcOptions(args: string[]): RemoteGcOptions {
      options.coreDumpMinAgeHours = parseNonNegativeNumber(arg, args[++index]);
    } else if (arg === "--registry-keep-per-repo") {
      const value = parseNonNegativeNumber(arg, args[++index]);
-      if (!Number.isInteger(value) || value < 10) throw new Error("--registry-keep-per-repo must be an integer >= 10");
+      if (!Number.isInteger(value) || value < 1) throw new Error("--registry-keep-per-repo must be an integer >= 1");
      options.registryKeepPerRepo = Math.min(value, 50);
    } else if (arg === "--registry-min-age-hours") {
      const value = parseNonNegativeNumber(arg, args[++index]);
-      if (value < 24) throw new Error("--registry-min-age-hours must be >= 24");
      options.registryMinAgeHours = value;
    } else if (arg === "--job-id") {
      const value = args[++index];
@@ -613,12 +612,131 @@ def registry_tag_rows():
            })
    return rows

+def registry_revision_rows():
+    rows = []
+    root = REGISTRY_REPOSITORY_ROOT
+    if not os.path.isdir(root):
+        return rows
+    for repo_root, dirs, files in os.walk(root):
+        if os.path.basename(repo_root) != "sha256":
+            continue
+        rel = os.path.relpath(repo_root, root)
+        suffix = "/_manifests/revisions/sha256"
+        if not rel.endswith(suffix):
+            continue
+        repo = rel[:-len(suffix)]
+        try:
+            revisions = os.listdir(repo_root)
+        except OSError:
+            continue
+        for digest_hex in sorted(revisions):
+            path = os.path.join(repo_root, digest_hex)
+            link = os.path.join(path, "link")
+            if not os.path.isfile(link):
+                continue
+            try:
+                with open(link, "r", encoding="utf-8") as handle:
+                    digest = handle.read().strip()
+                stat = os.stat(link)
+            except OSError:
+                continue
+            rows.append({
+                "repo": repo,
+                "digest": digest,
+                "mtime": stat.st_mtime,
+                "mtimeIso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)),
+                "path": path,
+            })
+    return rows
+
+def registry_retention_repo(repo):
+    return repo.startswith("hwlab/hwlab-") or repo.startswith("hwlab/cache/hwlab-")
+
+def registry_digest_hex(digest):
+    if not isinstance(digest, str) or not digest.startswith("sha256:"):
+        return None
+    value = digest.split(":", 1)[1]
+    if re.match(r"^[0-9a-f]{64}$", value) is None:
+        return None
+    return value
+
+def registry_blob_data_path(digest):
+    value = registry_digest_hex(digest)
+    if value is None:
+        return None
+    return os.path.join(REGISTRY_ROOT, "docker/registry/v2/blobs/sha256", value[:2], value, "data")
+
+_manifest_cache = {}
+def registry_manifest_json(digest):
+    if digest in _manifest_cache:
+        return _manifest_cache[digest]
+    path = registry_blob_data_path(digest)
+    if path is None or not os.path.isfile(path):
+        _manifest_cache[digest] = None
+        return None
+    try:
+        with open(path, "rb") as handle:
+            data = handle.read(8 * 1024 * 1024)
+        value = json.loads(data.decode("utf-8"))
+    except Exception:
+        value = None
+    _manifest_cache[digest] = value
+    return value
+
+def registry_manifest_refs(digest):
+    manifest = registry_manifest_json(digest)
+    if not isinstance(manifest, dict):
+        return set()
+    refs = set()
+    config = manifest.get("config") or {}
+    config_digest = config.get("digest")
+    if isinstance(config_digest, str) and registry_digest_hex(config_digest) is not None:
+        refs.add(config_digest)
+    for item in manifest.get("layers") or []:
+        item_digest = (item or {}).get("digest")
+        if isinstance(item_digest, str) and registry_digest_hex(item_digest) is not None:
+            refs.add(item_digest)
+    for item in manifest.get("manifests") or []:
+        item_digest = (item or {}).get("digest")
+        if isinstance(item_digest, str) and registry_digest_hex(item_digest) is not None:
+            refs.add(item_digest)
+    return refs
+
+def registry_digest_closure(seed):
+    seen = set()
+    stack = list(seed)
+    while stack:
+        digest = stack.pop()
+        if digest in seen or registry_digest_hex(digest) is None:
+            continue
+        seen.add(digest)
+        for child in registry_manifest_refs(digest):
+            if child not in seen:
+                stack.append(child)
+    return seen
+
+def registry_blob_size(digest):
+    path = registry_blob_data_path(digest)
+    if path is None or not os.path.isfile(path):
+        return 0
+    try:
+        return int(os.lstat(path).st_blocks) * 512
+    except OSError:
+        return 0
+
+def estimate_registry_reclaim(delete_manifest_digests, kept_manifest_digests):
+    deleted = registry_digest_closure(delete_manifest_digests)
+    kept = registry_digest_closure(kept_manifest_digests)
+    reclaim = deleted - kept
+    return sum(registry_blob_size(digest) for digest in reclaim)
+
 def plan_registry_retention():
-    keep_per_repo = int(OPTIONS.get("registryKeepPerRepo") or 5)
-    min_age_hours = float(OPTIONS.get("registryMinAgeHours") or 48)
+    keep_per_repo = int(OPTIONS.get("registryKeepPerRepo") if OPTIONS.get("registryKeepPerRepo") is not None else 5)
+    min_age_hours = float(OPTIONS.get("registryMinAgeHours") if OPTIONS.get("registryMinAgeHours") is not None else 48)
    cutoff = time.time() - min_age_hours * 3600
    refs, digests, refs_command = workload_image_refs()
    rows = registry_tag_rows()
+    revision_rows = registry_revision_rows()
    by_repo = {}
    for row in rows:
        by_repo.setdefault(row["repo"], []).append(row)
@@ -666,30 +784,50 @@ def plan_registry_retention():
            kept_count += 1
            kept_digests.add(row["digest"])
            keep_by_repo[row["repo"]] = keep_by_repo.get(row["repo"], 0) + 1
+    protected_digests = kept_digests | digests
+    protected_digests.update(row["digest"] for row in revision_rows if not registry_retention_repo(row["repo"]))
+    protected_digests = registry_digest_closure(protected_digests)
+    delete_revision_rows = []
+    revision_delete_by_repo = {}
+    for row in revision_rows:
+        if not registry_retention_repo(row["repo"]):
+            continue
+        if row["digest"] in protected_digests:
+            continue
+        delete_revision_rows.append(row)
+        revision_delete_by_repo[row["repo"]] = revision_delete_by_repo.get(row["repo"], 0) + 1
+    kept_revision_digests = set(row["digest"] for row in revision_rows if row not in delete_revision_rows)
+    delete_revision_digests = set(row["digest"] for row in delete_revision_rows)
    deletable_manifests = {}
    for row in delete_rows:
        if row["digest"] in kept_digests:
            continue
        deletable_manifests.setdefault(row["repo"], set()).add(row["digest"])
+    for row in delete_revision_rows:
+        deletable_manifests.setdefault(row["repo"], set()).add(row["digest"])
    deletable_manifest_count = sum(len(items) for items in deletable_manifests.values())
-    estimate = 0
    registry_size = du_size(REGISTRY_ROOT, 30) or 0
-    if rows:
-        estimate = int(registry_size * min(0.75, deletable_manifest_count / float(len(rows)) * 0.55))
+    estimate = estimate_registry_reclaim(delete_revision_digests, kept_revision_digests)
    return {
        "tagRows": rows,
+        "revisionRows": revision_rows,
        "deleteRows": delete_rows,
+        "deleteRevisionRows": delete_revision_rows,
        "summary": {
            "totalTags": len(rows),
+            "totalRevisions": len(revision_rows),
            "repoCount": len(by_repo),
            "keepPerRepo": keep_per_repo,
            "minAgeHours": min_age_hours,
            "protectedWorkloadRefs": len(refs),
            "protectedDigestRefs": len(digests),
+            "protectedDigestClosure": len(protected_digests),
            "keptTags": kept_count,
            "deleteTags": len(delete_rows),
            "deleteManifests": deletable_manifest_count,
+            "deleteRevisions": len(delete_revision_rows),
            "deleteByRepo": delete_by_repo,
+            "revisionDeleteByRepo": revision_delete_by_repo,
            "keepByRepo": keep_by_repo,
            "registrySizeBytes": registry_size,
            "estimatedReclaimBytes": estimate,
@@ -779,11 +917,12 @@ def execute_registry_retention():
        raise RuntimeError("refusing registry maintenance while hwlab-ci PipelineRun/TaskRun is active")
    plan = plan_registry_retention()
    delete_rows = plan.get("deleteRows") or []
+    delete_revision_rows = plan.get("deleteRevisionRows") or []
    delete_manifests = plan.get("deleteManifestsByRepo") or {}
-    if not delete_rows:
-        return {"reclaimedBytes": 0, "commandOutput": {"message": "no registry tags matched conservative retention", "registryPlan": plan.get("summary")}}
+    if not delete_rows and not delete_revision_rows:
+        return {"reclaimedBytes": 0, "commandOutput": {"message": "no registry tags or revisions matched conservative retention", "registryPlan": plan.get("summary")}}
    if not delete_manifests:
-        return {"reclaimedBytes": 0, "commandOutput": {"message": "matched tags are still referenced by retained tags; registry GC would not reclaim blobs", "registryPlan": plan.get("summary")}}
+        return {"reclaimedBytes": 0, "commandOutput": {"message": "matched manifests are still referenced by retained manifests; registry GC would not reclaim blobs", "registryPlan": plan.get("summary")}}
    cronjobs = ["hwlab-g14-branch-poller", "hwlab-v02-branch-poller"]
    original_crons = cronjob_suspend_states(cronjobs)
    before = du_size(REGISTRY_ROOT, 60) or 0
@@ -835,6 +974,21 @@ def execute_registry_retention():
                deleted.append({"repo": row.get("repo"), "tag": row.get("tag"), "digest": row.get("digest")})
        steps.append({"step": "delete-tag-directories", "count": len(deleted)})

+        deleted_revisions = []
+        for row in delete_revision_rows:
+            path = os.path.abspath(str(row.get("path") or ""))
+            digest_hex = registry_digest_hex(str(row.get("digest") or ""))
+            if digest_hex is None:
+                raise RuntimeError("refusing unexpected registry revision digest: %s" % row.get("digest"))
+            if not path.startswith(REGISTRY_REPOSITORY_ROOT + "/") or "/_manifests/revisions/sha256/" not in path:
+                raise RuntimeError("refusing unexpected registry revision path: %s" % path)
+            if os.path.basename(path) != digest_hex:
+                raise RuntimeError("refusing registry revision path/digest mismatch: %s" % path)
+            if os.path.isdir(path) and not os.path.islink(path):
+                shutil.rmtree(path)
+                deleted_revisions.append({"repo": row.get("repo"), "digest": row.get("digest")})
+        steps.append({"step": "delete-revision-directories", "count": len(deleted_revisions)})
+
        overrides = {
            "apiVersion": "v1",
            "spec": {
@@ -874,6 +1028,7 @@ def execute_registry_retention():
        "commandOutput": {
            "registryPlan": plan.get("summary"),
            "deletedTagCount": len(delete_rows),
+            "deletedRevisionCount": len(delete_revision_rows),
            "deletedManifestCount": sum(len(items) for items in delete_manifests.values()),
            "diskBeforeBytes": before,
            "diskAfterBytes": after,
@@ -1199,13 +1354,14 @@ def collect_candidates(observed_at):
        registry = plan_registry_retention()
        summary = registry.get("summary") or {}
        delete_rows = registry.get("deleteRows") or []
+        delete_revision_rows = registry.get("deleteRevisionRows") or []
        estimate = int(summary.get("estimatedReclaimBytes") or 0)
-        if delete_rows:
+        if delete_rows or delete_revision_rows:
            candidates.append({
                "id": "hwlab-registry:retention-gc",
                "kind": "hwlab-registry-retention-gc",
                "risk": "medium",
-                "description": "Conservative HWLAB registry retention: keep current workload refs, recent tags and latest tags per repo, then run official registry garbage-collect",
+                "description": "Conservative HWLAB registry retention: keep current workload refs, retained tags and protected repos, delete stale manifest revisions, then run official registry garbage-collect",
                "path": REGISTRY_ROOT,
                "sizeBytes": int(summary.get("registrySizeBytes") or 0),
                "estimatedReclaimBytes": estimate,
@@ -1216,9 +1372,12 @@ def collect_candidates(observed_at):
                    "minAgeHours": summary.get("minAgeHours"),
                    "deleteTags": len(delete_rows),
                    "deleteManifests": summary.get("deleteManifests"),
+                    "deleteRevisions": summary.get("deleteRevisions"),
                    "deleteByRepo": summary.get("deleteByRepo"),
+                    "revisionDeleteByRepo": summary.get("revisionDeleteByRepo"),
                    "protectedWorkloadRefs": summary.get("protectedWorkloadRefs"),
                    "protectedDigestRefs": summary.get("protectedDigestRefs"),
+                    "protectedDigestClosure": summary.get("protectedDigestClosure"),
                },
            })
        elif bool(OPTIONS.get("registryGcOnly")) and int(summary.get("totalTags") or 0) > 0 and int(summary.get("deleteTags") or 0) == 0: