Codex
678 lines
31 KiB
Python
678 lines
31 KiB
Python
def active_hwlab_ci_writes():
|
|
result = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get pipelinerun,taskrun -n hwlab-ci --no-headers 2>/dev/null | awk '$2 != \"True\" && $2 != \"False\" {print}' | head -40"], 15)
|
|
lines = [line for line in (result.get("stdout") or "").splitlines() if line.strip()]
|
|
return {"ok": result["exitCode"] == 0, "activeCount": len(lines), "activePreview": lines, "command": bounded(result)}
|
|
|
|
def active_hwlab_ci_jobs():
|
|
result = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get jobs -n hwlab-ci --no-headers 2>/dev/null | awk '$2 != \"Complete\" && $2 != \"Failed\" {print}' | head -40"], 15)
|
|
lines = [line for line in (result.get("stdout") or "").splitlines() if line.strip()]
|
|
return {"ok": result["exitCode"] == 0, "activeCount": len(lines), "activePreview": lines, "command": bounded(result)}
|
|
|
|
def wait_no_active_hwlab_ci(timeout=180):
|
|
deadline = time.time() + timeout
|
|
last = None
|
|
while time.time() < deadline:
|
|
writes = active_hwlab_ci_writes()
|
|
jobs = active_hwlab_ci_jobs()
|
|
last = {"writes": writes, "jobs": jobs}
|
|
if writes.get("ok") and jobs.get("ok") and int(writes.get("activeCount") or 0) == 0 and int(jobs.get("activeCount") or 0) == 0:
|
|
return {"ok": True, "last": last}
|
|
time.sleep(5)
|
|
return {"ok": False, "last": last}
|
|
|
|
def kubectl_json(args, timeout=20):
|
|
result = command(["env", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml", "kubectl"] + args + ["-o", "json"], timeout)
|
|
if result["exitCode"] != 0:
|
|
return None
|
|
try:
|
|
return json.loads(result["stdout"] or "{}")
|
|
except Exception:
|
|
return None
|
|
|
|
def kctl(args, timeout=30):
|
|
return command(["env", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml", "kubectl"] + args, timeout)
|
|
|
|
def workload_image_refs():
|
|
result = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get deploy,sts,ds,pod -A -o jsonpath='{range .items[*]}{range .spec.containers[*]}{.image}{\"\\n\"}{end}{range .spec.initContainers[*]}{.image}{\"\\n\"}{end}{range .spec.template.spec.containers[*]}{.image}{\"\\n\"}{end}{range .spec.template.spec.initContainers[*]}{.image}{\"\\n\"}{end}{end}' 2>/dev/null | sort -u"], 30)
|
|
refs = set()
|
|
digests = set()
|
|
for image in (result.get("stdout") or "").splitlines():
|
|
image = image.strip()
|
|
if not image.startswith("127.0.0.1:5000/"):
|
|
continue
|
|
ref = image.split("127.0.0.1:5000/", 1)[1]
|
|
if "@sha256:" in ref:
|
|
repo, digest = ref.split("@", 1)
|
|
refs.add((repo, "@" + digest))
|
|
digests.add("sha256:" + digest.split(":", 1)[1])
|
|
elif ":" in ref:
|
|
repo, tag = ref.rsplit(":", 1)
|
|
refs.add((repo, tag))
|
|
return refs, digests, bounded(result)
|
|
|
|
def registry_request(method, path, headers=None, timeout=20):
|
|
url = "http://127.0.0.1:5000" + path
|
|
req = urllib.request.Request(url, method=method, headers=headers or {})
|
|
with urllib.request.urlopen(req, timeout=timeout) as response:
|
|
body = response.read()
|
|
return {"status": response.status, "headers": dict(response.headers), "body": body.decode("utf-8", errors="replace")}
|
|
|
|
def registry_tag_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) != "tags":
|
|
continue
|
|
rel = os.path.relpath(repo_root, root)
|
|
suffix = "/_manifests/tags"
|
|
if not rel.endswith(suffix):
|
|
continue
|
|
repo = rel[:-len(suffix)]
|
|
try:
|
|
tags = os.listdir(repo_root)
|
|
except OSError:
|
|
continue
|
|
for tag in sorted(tags):
|
|
link = os.path.join(repo_root, tag, "current", "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,
|
|
"tag": tag,
|
|
"digest": digest,
|
|
"mtime": stat.st_mtime,
|
|
"mtimeIso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(stat.st_mtime)),
|
|
"path": os.path.join(repo_root, tag),
|
|
})
|
|
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") 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)
|
|
keep = set()
|
|
keep_reasons = {}
|
|
for repo, items in by_repo.items():
|
|
items.sort(key=lambda item: item["mtime"], reverse=True)
|
|
for row in items[:keep_per_repo]:
|
|
key = (row["repo"], row["tag"])
|
|
keep.add(key)
|
|
keep_reasons[key] = "latest-per-repo"
|
|
for row in items:
|
|
key = (row["repo"], row["tag"])
|
|
if row["tag"] in REGISTRY_PROTECTED_TAGS:
|
|
keep.add(key)
|
|
keep_reasons[key] = "protected-tag"
|
|
if key in refs:
|
|
keep.add(key)
|
|
keep_reasons[key] = "workload-tag-ref"
|
|
if row["digest"] in digests:
|
|
keep.add(key)
|
|
keep_reasons[key] = "workload-digest-ref"
|
|
if row["repo"].startswith("hwlab/cache/"):
|
|
keep.add(key)
|
|
keep_reasons[key] = "cache-repo"
|
|
if row["mtime"] >= cutoff:
|
|
keep.add(key)
|
|
keep_reasons[key] = "recent-tag"
|
|
delete_rows = []
|
|
kept_count = 0
|
|
delete_by_repo = {}
|
|
keep_by_repo = {}
|
|
kept_digests = set()
|
|
for row in rows:
|
|
key = (row["repo"], row["tag"])
|
|
should_delete = (
|
|
key not in keep
|
|
and row["repo"].startswith("hwlab/hwlab-")
|
|
and re.match(r"^[0-9a-f]{7,40}$", row["tag"]) is not None
|
|
)
|
|
if should_delete:
|
|
delete_rows.append(row)
|
|
delete_by_repo[row["repo"]] = delete_by_repo.get(row["repo"], 0) + 1
|
|
else:
|
|
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())
|
|
registry_size = du_size(REGISTRY_ROOT, 30) or 0
|
|
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,
|
|
},
|
|
"deleteManifestsByRepo": {repo: sorted(list(digests)) for repo, digests in deletable_manifests.items()},
|
|
"refsCommand": refs_command,
|
|
}
|
|
|
|
def registry_deployment_preflight():
|
|
dep = kubectl_json(["-n", "hwlab-ci", "get", "deploy", "hwlab-registry"], 20)
|
|
if not dep:
|
|
return {"ok": False, "reason": "registry-deployment-missing"}
|
|
spec = ((dep.get("spec") or {}).get("template") or {}).get("spec") or {}
|
|
containers = spec.get("containers") or []
|
|
volumes = spec.get("volumes") or []
|
|
registry_container = next((item for item in containers if item.get("name") == "registry"), containers[0] if containers else {})
|
|
mounts = registry_container.get("volumeMounts") or []
|
|
has_host_path = any(((vol.get("hostPath") or {}).get("path") == REGISTRY_ROOT and vol.get("name") == "storage") for vol in volumes)
|
|
has_mount = any((mount.get("name") == "storage" and mount.get("mountPath") == "/var/lib/registry") for mount in mounts)
|
|
image = str(registry_container.get("image") or "")
|
|
ok = bool(has_host_path and has_mount and image.startswith("registry:") and spec.get("hostNetwork") is True)
|
|
return {
|
|
"ok": ok,
|
|
"reason": "ok" if ok else "unexpected-registry-deployment-shape",
|
|
"image": image,
|
|
"hostNetwork": spec.get("hostNetwork"),
|
|
"hasExpectedHostPath": has_host_path,
|
|
"hasExpectedMount": has_mount,
|
|
"replicas": (dep.get("spec") or {}).get("replicas"),
|
|
"readyReplicas": (dep.get("status") or {}).get("readyReplicas"),
|
|
}
|
|
|
|
def cronjob_suspend_states(names):
|
|
states = {}
|
|
for name in names:
|
|
data = kubectl_json(["-n", "hwlab-ci", "get", "cronjob", name], 15)
|
|
if data:
|
|
states[name] = bool(((data.get("spec") or {}).get("suspend")) is True)
|
|
return states
|
|
|
|
def patch_cronjob_suspend(name, suspend):
|
|
payload = json.dumps({"spec": {"suspend": bool(suspend)}})
|
|
return kctl(["-n", "hwlab-ci", "patch", "cronjob", name, "--type=merge", "-p", payload], 30)
|
|
|
|
def wait_registry_pod_count(target, timeout=90):
|
|
deadline = time.time() + timeout
|
|
last = None
|
|
while time.time() < deadline:
|
|
result = kctl(["-n", "hwlab-ci", "get", "pods", "-l", "app.kubernetes.io/name=hwlab-registry", "--no-headers"], 20)
|
|
last = bounded(result)
|
|
lines = [line for line in (result.get("stdout") or "").splitlines() if line.strip()]
|
|
active = []
|
|
for line in lines:
|
|
parts = line.split()
|
|
status = parts[2] if len(parts) >= 3 else ""
|
|
if status in set(["Completed", "Error", "Failed", "Succeeded"]):
|
|
continue
|
|
active.append(line)
|
|
if len(active) == target:
|
|
return {"ok": True, "lines": active, "allLines": lines, "last": last}
|
|
time.sleep(2)
|
|
return {"ok": False, "lines": [], "last": last}
|
|
|
|
def wait_pod_terminal(name, timeout=900):
|
|
deadline = time.time() + timeout
|
|
last = None
|
|
while time.time() < deadline:
|
|
data = kubectl_json(["-n", "hwlab-ci", "get", "pod", name], 20)
|
|
if data:
|
|
phase = ((data.get("status") or {}).get("phase")) or ""
|
|
last = {"phase": phase}
|
|
if phase == "Succeeded":
|
|
return {"ok": True, "phase": phase}
|
|
if phase == "Failed":
|
|
return {"ok": False, "phase": phase}
|
|
time.sleep(3)
|
|
return {"ok": False, "phase": "Timeout", "last": last}
|
|
|
|
def execute_registry_retention():
|
|
if PROVIDER_ID.upper() != "G14":
|
|
raise RuntimeError("HWLAB registry retention is only supported on G14")
|
|
deployment = registry_deployment_preflight()
|
|
if not deployment.get("ok"):
|
|
raise RuntimeError("registry deployment preflight failed: %s" % deployment.get("reason"))
|
|
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 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 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
|
|
gc_name = "hwlab-registry-gc-%s" % int(time.time())
|
|
steps = []
|
|
try:
|
|
for name in original_crons:
|
|
result = patch_cronjob_suspend(name, True)
|
|
steps.append({"step": "suspend-cronjob", "name": name, "result": bounded(result)})
|
|
if result["exitCode"] != 0:
|
|
raise RuntimeError("failed to suspend cronjob %s" % name)
|
|
idle_after_suspend = wait_no_active_hwlab_ci(180)
|
|
steps.append({"step": "idle-after-suspend", "result": idle_after_suspend})
|
|
if not idle_after_suspend.get("ok"):
|
|
raise RuntimeError("refusing registry maintenance because hwlab-ci did not become idle after suspend")
|
|
|
|
deleted_manifests = []
|
|
for repo, digests in delete_manifests.items():
|
|
encoded_repo = "/".join(urllib.parse.quote(part, safe="") for part in repo.split("/"))
|
|
for digest in digests:
|
|
try:
|
|
result = registry_request("DELETE", "/v2/%s/manifests/%s" % (encoded_repo, urllib.parse.quote(digest, safe=":")), {"Accept": "application/vnd.docker.distribution.manifest.v2+json, application/vnd.oci.image.manifest.v1+json"})
|
|
deleted_manifests.append({"repo": repo, "digest": digest, "status": result.get("status")})
|
|
except urllib.error.HTTPError as exc:
|
|
if exc.code == 404:
|
|
deleted_manifests.append({"repo": repo, "digest": digest, "status": 404})
|
|
else:
|
|
raise
|
|
steps.append({"step": "registry-api-delete-manifests", "count": len(deleted_manifests), "preview": deleted_manifests[:20]})
|
|
|
|
scale_down = kctl(["-n", "hwlab-ci", "scale", "deploy", "hwlab-registry", "--replicas=0"], 60)
|
|
steps.append({"step": "scale-registry-down", "result": bounded(scale_down)})
|
|
if scale_down["exitCode"] != 0:
|
|
raise RuntimeError("failed to scale registry down")
|
|
waited_down = wait_registry_pod_count(0, 120)
|
|
steps.append({"step": "wait-registry-down", "result": waited_down})
|
|
if not waited_down.get("ok"):
|
|
raise RuntimeError("registry pod did not scale down")
|
|
|
|
deleted = []
|
|
for row in delete_rows:
|
|
path = os.path.abspath(str(row.get("path") or ""))
|
|
if not path.startswith(REGISTRY_REPOSITORY_ROOT + "/") or "/_manifests/tags/" not in path:
|
|
raise RuntimeError("refusing unexpected registry tag path: %s" % path)
|
|
if not re.match(r"^[0-9a-f]{7,40}$", str(row.get("tag") or "")):
|
|
raise RuntimeError("refusing unexpected registry tag name: %s" % row.get("tag"))
|
|
if os.path.isdir(path) and not os.path.islink(path):
|
|
shutil.rmtree(path)
|
|
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": {
|
|
"restartPolicy": "Never",
|
|
"containers": [{
|
|
"name": "registry-gc",
|
|
"image": "registry:2.8.3",
|
|
"command": ["registry", "garbage-collect", "/etc/docker/registry/config.yml"],
|
|
"volumeMounts": [{"name": "storage", "mountPath": "/var/lib/registry"}],
|
|
}],
|
|
"volumes": [{"name": "storage", "hostPath": {"path": REGISTRY_ROOT, "type": "DirectoryOrCreate"}}],
|
|
},
|
|
}
|
|
run_gc = kctl(["-n", "hwlab-ci", "run", gc_name, "--restart=Never", "--image=registry:2.8.3", "--overrides=%s" % json.dumps(overrides)], 60)
|
|
steps.append({"step": "start-registry-gc-pod", "result": bounded(run_gc), "pod": gc_name})
|
|
if run_gc["exitCode"] != 0:
|
|
raise RuntimeError("failed to start registry GC pod")
|
|
waited_gc = wait_pod_terminal(gc_name, 900)
|
|
steps.append({"step": "wait-registry-gc", "result": waited_gc})
|
|
logs = kctl(["-n", "hwlab-ci", "logs", gc_name], 120)
|
|
steps.append({"step": "registry-gc-logs", "result": bounded(logs)})
|
|
if not waited_gc.get("ok"):
|
|
raise RuntimeError("registry GC pod did not complete successfully")
|
|
finally:
|
|
cleanup_gc = kctl(["-n", "hwlab-ci", "delete", "pod", gc_name, "--ignore-not-found=true"], 60)
|
|
steps.append({"step": "delete-registry-gc-pod", "result": bounded(cleanup_gc)})
|
|
scale_up = kctl(["-n", "hwlab-ci", "scale", "deploy", "hwlab-registry", "--replicas=%s" % int(deployment.get("replicas") or 1)], 60)
|
|
steps.append({"step": "scale-registry-up", "result": bounded(scale_up)})
|
|
rollout = kctl(["-n", "hwlab-ci", "rollout", "status", "deploy/hwlab-registry", "--timeout=180s"], 200)
|
|
steps.append({"step": "wait-registry-rollout", "result": bounded(rollout)})
|
|
for name, was_suspended in original_crons.items():
|
|
restore = patch_cronjob_suspend(name, was_suspended)
|
|
steps.append({"step": "restore-cronjob", "name": name, "suspend": was_suspended, "result": bounded(restore)})
|
|
after = du_size(REGISTRY_ROOT, 60) or 0
|
|
return {
|
|
"reclaimedBytes": max(0, before - after),
|
|
"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,
|
|
"steps": steps[-12:],
|
|
},
|
|
}
|
|
|
|
def execute_registry_garbage_collect_only():
|
|
if PROVIDER_ID.upper() != "G14":
|
|
raise RuntimeError("HWLAB registry garbage-collect is only supported on G14")
|
|
deployment = registry_deployment_preflight()
|
|
if not deployment.get("ok"):
|
|
raise RuntimeError("registry deployment preflight failed: %s" % deployment.get("reason"))
|
|
cronjobs = ["hwlab-g14-branch-poller", "hwlab-v02-branch-poller"]
|
|
original_crons = cronjob_suspend_states(cronjobs)
|
|
before = du_size(REGISTRY_ROOT, 60) or 0
|
|
gc_name = "hwlab-registry-gc-%s" % int(time.time())
|
|
steps = []
|
|
try:
|
|
for name in original_crons:
|
|
result = patch_cronjob_suspend(name, True)
|
|
steps.append({"step": "suspend-cronjob", "name": name, "result": bounded(result)})
|
|
if result["exitCode"] != 0:
|
|
raise RuntimeError("failed to suspend cronjob %s" % name)
|
|
idle_after_suspend = wait_no_active_hwlab_ci(180)
|
|
steps.append({"step": "idle-after-suspend", "result": idle_after_suspend})
|
|
if not idle_after_suspend.get("ok"):
|
|
raise RuntimeError("refusing registry maintenance because hwlab-ci did not become idle after suspend")
|
|
|
|
scale_down = kctl(["-n", "hwlab-ci", "scale", "deploy", "hwlab-registry", "--replicas=0"], 60)
|
|
steps.append({"step": "scale-registry-down", "result": bounded(scale_down)})
|
|
if scale_down["exitCode"] != 0:
|
|
raise RuntimeError("failed to scale registry down")
|
|
waited_down = wait_registry_pod_count(0, 120)
|
|
steps.append({"step": "wait-registry-down", "result": waited_down})
|
|
if not waited_down.get("ok"):
|
|
raise RuntimeError("registry pod did not scale down")
|
|
|
|
overrides = {
|
|
"apiVersion": "v1",
|
|
"spec": {
|
|
"restartPolicy": "Never",
|
|
"containers": [{
|
|
"name": "registry-gc",
|
|
"image": "registry:2.8.3",
|
|
"command": ["registry", "garbage-collect", "/etc/docker/registry/config.yml"],
|
|
"volumeMounts": [{"name": "storage", "mountPath": "/var/lib/registry"}],
|
|
}],
|
|
"volumes": [{"name": "storage", "hostPath": {"path": REGISTRY_ROOT, "type": "DirectoryOrCreate"}}],
|
|
},
|
|
}
|
|
run_gc = kctl(["-n", "hwlab-ci", "run", gc_name, "--restart=Never", "--image=registry:2.8.3", "--overrides=%s" % json.dumps(overrides)], 60)
|
|
steps.append({"step": "start-registry-gc-pod", "result": bounded(run_gc), "pod": gc_name})
|
|
if run_gc["exitCode"] != 0:
|
|
raise RuntimeError("failed to start registry GC pod")
|
|
waited_gc = wait_pod_terminal(gc_name, 900)
|
|
steps.append({"step": "wait-registry-gc", "result": waited_gc})
|
|
logs = kctl(["-n", "hwlab-ci", "logs", gc_name], 120)
|
|
steps.append({"step": "registry-gc-logs", "result": bounded(logs)})
|
|
if not waited_gc.get("ok"):
|
|
raise RuntimeError("registry GC pod did not complete successfully")
|
|
finally:
|
|
cleanup_gc = kctl(["-n", "hwlab-ci", "delete", "pod", gc_name, "--ignore-not-found=true"], 60)
|
|
steps.append({"step": "delete-registry-gc-pod", "result": bounded(cleanup_gc)})
|
|
scale_up = kctl(["-n", "hwlab-ci", "scale", "deploy", "hwlab-registry", "--replicas=%s" % int(deployment.get("replicas") or 1)], 60)
|
|
steps.append({"step": "scale-registry-up", "result": bounded(scale_up)})
|
|
rollout = kctl(["-n", "hwlab-ci", "rollout", "status", "deploy/hwlab-registry", "--timeout=180s"], 200)
|
|
steps.append({"step": "wait-registry-rollout", "result": bounded(rollout)})
|
|
for name, was_suspended in original_crons.items():
|
|
restore = patch_cronjob_suspend(name, was_suspended)
|
|
steps.append({"step": "restore-cronjob", "name": name, "suspend": was_suspended, "result": bounded(restore)})
|
|
after = du_size(REGISTRY_ROOT, 60) or 0
|
|
return {
|
|
"reclaimedBytes": max(0, before - after),
|
|
"commandOutput": {
|
|
"message": "official registry garbage-collect only; no additional tag deletion",
|
|
"diskBeforeBytes": before,
|
|
"diskAfterBytes": after,
|
|
"steps": steps[-12:],
|
|
},
|
|
}
|
|
|
|
def start_registry_retention_job(mode):
|
|
job_id = "g14-registry-%s-%s" % (int(time.time()), os.getpid())
|
|
paths = job_paths(job_id)
|
|
started_at = now_iso()
|
|
initial = {
|
|
"ok": True,
|
|
"action": "gc remote status",
|
|
"providerId": PROVIDER_ID,
|
|
"jobId": job_id,
|
|
"status": "running",
|
|
"kind": "hwlab-registry-retention-gc" if mode == "retention" else "hwlab-registry-garbage-collect",
|
|
"mode": mode,
|
|
"startedAt": started_at,
|
|
"statePath": paths["state"],
|
|
"logPath": paths["log"],
|
|
"options": OPTIONS,
|
|
}
|
|
write_json_atomic(paths["state"], initial)
|
|
pid = os.fork()
|
|
if pid != 0:
|
|
return {
|
|
"status": "started",
|
|
"reclaimedBytes": None,
|
|
"commandOutput": {
|
|
"jobId": job_id,
|
|
"pid": pid,
|
|
"statePath": paths["state"],
|
|
"logPath": paths["log"],
|
|
"statusCommand": "bun scripts/cli.ts gc remote %s status --job-id %s" % (PROVIDER_ID, job_id),
|
|
"message": "registry retention GC is running as a detached remote job",
|
|
},
|
|
}
|
|
|
|
try:
|
|
os.setsid()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
devnull = os.open(os.devnull, os.O_RDONLY)
|
|
os.dup2(devnull, 0)
|
|
os.close(devnull)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
log_handle = open(paths["log"], "a", encoding="utf-8", buffering=1)
|
|
os.dup2(log_handle.fileno(), 1)
|
|
os.dup2(log_handle.fileno(), 2)
|
|
except Exception:
|
|
log_handle = None
|
|
try:
|
|
print("[%s] starting HWLAB registry %s job %s" % (now_iso(), mode, job_id), flush=True)
|
|
result = execute_registry_retention() if mode == "retention" else execute_registry_garbage_collect_only()
|
|
payload = dict(initial)
|
|
payload.update({
|
|
"status": "succeeded",
|
|
"finishedAt": now_iso(),
|
|
"result": result,
|
|
"diskAfter": df_snapshot(),
|
|
"clusterAfter": cluster_preflight(),
|
|
})
|
|
write_json_atomic(paths["state"], payload)
|
|
print("[%s] completed HWLAB registry %s job %s" % (now_iso(), mode, job_id), flush=True)
|
|
os._exit(0)
|
|
except Exception as exc:
|
|
payload = dict(initial)
|
|
payload.update({
|
|
"ok": False,
|
|
"status": "failed",
|
|
"finishedAt": now_iso(),
|
|
"error": str(exc),
|
|
"diskAfter": df_snapshot(),
|
|
"clusterAfter": cluster_preflight(),
|
|
})
|
|
try:
|
|
write_json_atomic(paths["state"], payload)
|
|
except Exception:
|
|
pass
|
|
print("[%s] failed HWLAB registry %s job %s: %s" % (now_iso(), mode, job_id, exc), flush=True)
|
|
os._exit(1)
|
|
finally:
|
|
try:
|
|
if log_handle:
|
|
log_handle.close()
|
|
except Exception:
|
|
pass
|