Codex
1795 lines
75 KiB
TypeScript
1795 lines
75 KiB
TypeScript
import { Buffer } from "node:buffer";
|
|
|
|
import { type UniDeskConfig } from "./config";
|
|
import { runSshCommandCapture } from "./ssh";
|
|
|
|
interface RemoteGcOptions {
|
|
confirm: boolean;
|
|
journal: boolean;
|
|
journalTargetBytes: number;
|
|
dockerLogs: boolean;
|
|
dockerLogMaxBytes: number;
|
|
buildCache: boolean;
|
|
buildCacheUntil: string;
|
|
tmp: boolean;
|
|
tmpMinAgeHours: number;
|
|
aptCache: boolean;
|
|
coreDumps: boolean;
|
|
coreDumpMinAgeHours: number;
|
|
hwlabRegistry: boolean;
|
|
registryGcOnly: boolean;
|
|
registryKeepPerRepo: number;
|
|
registryMinAgeHours: number;
|
|
targetUsePercent?: number;
|
|
jobId?: string;
|
|
limit: number;
|
|
resultLimit: number;
|
|
full: boolean;
|
|
}
|
|
|
|
const DEFAULT_REMOTE_OPTIONS: RemoteGcOptions = {
|
|
confirm: false,
|
|
journal: true,
|
|
journalTargetBytes: 512 * 1024 * 1024,
|
|
dockerLogs: true,
|
|
dockerLogMaxBytes: 50 * 1024 * 1024,
|
|
buildCache: true,
|
|
buildCacheUntil: "24h",
|
|
tmp: true,
|
|
tmpMinAgeHours: 24,
|
|
aptCache: true,
|
|
coreDumps: true,
|
|
coreDumpMinAgeHours: 1,
|
|
hwlabRegistry: false,
|
|
registryGcOnly: false,
|
|
registryKeepPerRepo: 20,
|
|
registryMinAgeHours: 48,
|
|
limit: 50,
|
|
resultLimit: 50,
|
|
full: false,
|
|
};
|
|
|
|
export async function runRemoteGcCommand(config: UniDeskConfig, providerId: string | undefined, action: string | undefined, args: string[]): Promise<unknown> {
|
|
if (providerId === undefined || providerId.length === 0) {
|
|
return {
|
|
ok: false,
|
|
error: "gc-remote-provider-required",
|
|
usage: "bun scripts/cli.ts gc remote <providerId> plan|run|status [--confirm]",
|
|
};
|
|
}
|
|
const subaction = action ?? "plan";
|
|
const options = parseRemoteGcOptions(args);
|
|
if (subaction === "plan" || subaction === "dry-run") return await runRemoteGc(config, providerId, "plan", options);
|
|
if (subaction === "status") return await runRemoteGc(config, providerId, "status", options);
|
|
if (subaction === "run") {
|
|
if (!options.confirm) {
|
|
return {
|
|
ok: false,
|
|
error: "gc-remote-run-requires-confirm",
|
|
dryRun: true,
|
|
mutation: false,
|
|
requiredFlag: "--confirm",
|
|
planCommand: `bun scripts/cli.ts gc remote ${providerId} plan`,
|
|
runCommand: `bun scripts/cli.ts gc remote ${providerId} run --confirm`,
|
|
};
|
|
}
|
|
return await runRemoteGc(config, providerId, "run", options);
|
|
}
|
|
return {
|
|
ok: false,
|
|
error: "unsupported-gc-remote-action",
|
|
action: subaction,
|
|
supportedActions: ["plan", "run", "status"],
|
|
};
|
|
}
|
|
|
|
function parseRemoteGcOptions(args: string[]): RemoteGcOptions {
|
|
const options = { ...DEFAULT_REMOTE_OPTIONS };
|
|
for (let index = 0; index < args.length; index += 1) {
|
|
const arg = args[index] ?? "";
|
|
if (arg === "--confirm") {
|
|
options.confirm = true;
|
|
} else if (arg === "--dry-run") {
|
|
// accepted for plan compatibility
|
|
} else if (arg === "--journal-target-size") {
|
|
options.journalTargetBytes = parseSizeOption(arg, args[++index]);
|
|
} else if (arg === "--docker-log-max-bytes") {
|
|
options.dockerLogMaxBytes = parseSizeOption(arg, args[++index]);
|
|
} else if (arg === "--build-cache-until") {
|
|
const value = args[++index];
|
|
if (!value || !/^\d+(s|m|h|d)$/u.test(value)) throw new Error(`${arg} must look like 24h, 7d, 30m or 60s`);
|
|
options.buildCacheUntil = value;
|
|
} else if (arg === "--tmp-min-age-hours") {
|
|
options.tmpMinAgeHours = parseNonNegativeNumber(arg, args[++index]);
|
|
} else if (arg === "--core-dump-min-age-hours") {
|
|
options.coreDumpMinAgeHours = parseNonNegativeNumber(arg, args[++index]);
|
|
} else if (arg === "--registry-keep-per-repo") {
|
|
const value = parseNonNegativeNumber(arg, args[++index]);
|
|
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]);
|
|
options.registryMinAgeHours = value;
|
|
} else if (arg === "--target-use-percent") {
|
|
const value = parseNonNegativeNumber(arg, args[++index]);
|
|
if (!Number.isInteger(value) || value < 1 || value > 99) throw new Error("--target-use-percent must be an integer from 1 to 99");
|
|
options.targetUsePercent = value;
|
|
} else if (arg === "--job-id") {
|
|
const value = args[++index];
|
|
if (!value || !/^[A-Za-z0-9._-]{1,128}$/u.test(value)) throw new Error("--job-id must be a safe job id");
|
|
options.jobId = value;
|
|
} else if (arg === "--limit") {
|
|
const value = parseNonNegativeNumber(arg, args[++index]);
|
|
if (!Number.isInteger(value) || value <= 0) throw new Error("--limit must be a positive integer");
|
|
options.limit = Math.min(value, 5000);
|
|
} else if (arg === "--result-limit") {
|
|
const value = parseNonNegativeNumber(arg, args[++index]);
|
|
if (!Number.isInteger(value) || value <= 0) throw new Error("--result-limit must be a positive integer");
|
|
options.resultLimit = Math.min(value, 5000);
|
|
} else if (arg === "--no-journal") {
|
|
options.journal = false;
|
|
} else if (arg === "--no-docker-logs") {
|
|
options.dockerLogs = false;
|
|
} else if (arg === "--no-build-cache") {
|
|
options.buildCache = false;
|
|
} else if (arg === "--no-tmp") {
|
|
options.tmp = false;
|
|
} else if (arg === "--no-apt-cache") {
|
|
options.aptCache = false;
|
|
} else if (arg === "--no-core-dumps") {
|
|
options.coreDumps = false;
|
|
} else if (arg === "--include-hwlab-registry") {
|
|
options.hwlabRegistry = true;
|
|
} else if (arg === "--registry-gc-only") {
|
|
options.hwlabRegistry = true;
|
|
options.registryGcOnly = true;
|
|
} else if (arg === "--full" || arg === "--raw") {
|
|
options.full = true;
|
|
} else {
|
|
throw new Error(`unknown gc remote option: ${arg}`);
|
|
}
|
|
}
|
|
return options;
|
|
}
|
|
|
|
function parseNonNegativeNumber(name: string, raw: string | undefined): number {
|
|
const value = Number(raw);
|
|
if (!Number.isFinite(value) || value < 0) throw new Error(`${name} must be a non-negative number`);
|
|
return value;
|
|
}
|
|
|
|
function parseSizeOption(name: string, raw: string | undefined): number {
|
|
const value = parseSize(raw ?? "");
|
|
if (value === null || value <= 0) throw new Error(`${name} must be a positive size such as 512M, 1GiB or 50000000`);
|
|
return value;
|
|
}
|
|
|
|
function parseSize(raw: string): number | null {
|
|
const match = raw.trim().match(/^(\d+(?:\.\d+)?)\s*(b|k|kb|kib|m|mb|mib|g|gb|gib)?$/iu);
|
|
if (!match) return null;
|
|
const value = Number(match[1]);
|
|
const unit = (match[2] ?? "b").toLowerCase();
|
|
const multiplier =
|
|
unit === "g" || unit === "gb" || unit === "gib" ? 1024 ** 3
|
|
: unit === "m" || unit === "mb" || unit === "mib" ? 1024 ** 2
|
|
: unit === "k" || unit === "kb" || unit === "kib" ? 1024
|
|
: 1;
|
|
const bytes = Math.floor(value * multiplier);
|
|
return Number.isFinite(bytes) ? bytes : null;
|
|
}
|
|
|
|
async function runRemoteGc(config: UniDeskConfig, providerId: string, action: "plan" | "run" | "status", options: RemoteGcOptions): Promise<unknown> {
|
|
const scriptConfig = Buffer.from(JSON.stringify({ providerId, action, options }), "utf8").toString("base64");
|
|
const result = await runSshCommandCapture(config, providerId, ["py"], remoteGcPython(scriptConfig));
|
|
if (result.exitCode !== 0) {
|
|
return {
|
|
ok: false,
|
|
error: "gc-remote-command-failed",
|
|
providerId,
|
|
action: `gc remote ${action}`,
|
|
exitCode: result.exitCode,
|
|
stdoutTail: result.stdout.slice(-4000),
|
|
stderrTail: result.stderr.slice(-4000),
|
|
};
|
|
}
|
|
try {
|
|
return JSON.parse(result.stdout) as unknown;
|
|
} catch (error) {
|
|
return {
|
|
ok: false,
|
|
error: "gc-remote-invalid-json",
|
|
providerId,
|
|
action: `gc remote ${action}`,
|
|
message: error instanceof Error ? error.message : String(error),
|
|
stdoutTail: result.stdout.slice(-4000),
|
|
stderrTail: result.stderr.slice(-4000),
|
|
};
|
|
}
|
|
}
|
|
|
|
function remoteGcPython(configBase64: string): string {
|
|
return String.raw`
|
|
import base64
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
CONFIG = json.loads(base64.b64decode("${configBase64}").decode("utf-8"))
|
|
PROVIDER_ID = str(CONFIG.get("providerId") or "")
|
|
ACTION = str(CONFIG.get("action") or "plan")
|
|
OPTIONS = CONFIG.get("options") or {}
|
|
|
|
TMP_PREFIX_ALLOWLIST = [
|
|
"hwlab-agent-",
|
|
"hwlab-cd-",
|
|
"hwlab-cli-cicd-",
|
|
"hwlab-codeagent-trace",
|
|
"hwlab-desired-state-",
|
|
"hwlab-g14-",
|
|
"hwlab-main-",
|
|
"hwlab-merge-",
|
|
"hwlab-pr",
|
|
"hwlab-refresh-",
|
|
"hwlab-remote-",
|
|
"hwlab-ts-check",
|
|
"hwlab-bun-runtime-check-",
|
|
"hwlab-v02-",
|
|
"playwright-artifacts-",
|
|
"playwright_chromiumdev_profile-",
|
|
"unidesk-apply-patch-v2-perf-",
|
|
"unidesk-clean-",
|
|
"unidesk-code-queue",
|
|
"unidesk-hwlab-cd-",
|
|
"unidesk-pr",
|
|
"unidesk-tran-runner",
|
|
"bunx-",
|
|
"codex-app-schema",
|
|
"codex-app-ts",
|
|
"marked-",
|
|
"node-compile-cache",
|
|
]
|
|
|
|
TMP_EXACT_PROTECT = set([
|
|
"/tmp/codex-apply-patch",
|
|
"/tmp/codex-ipc",
|
|
"/tmp/tmux-0",
|
|
"/tmp/snap-private-tmp",
|
|
])
|
|
|
|
CORE_DUMP_DIR_ALLOWLIST = set([
|
|
"/root/unidesk",
|
|
])
|
|
|
|
REGISTRY_REPOSITORY_ROOT = "/var/lib/hwlab/registry/docker/registry/v2/repositories"
|
|
REGISTRY_ROOT = "/var/lib/hwlab/registry"
|
|
REGISTRY_PROTECTED_TAGS = set([
|
|
"latest",
|
|
"16-alpine",
|
|
"20-bookworm-slim",
|
|
"node22-alpine-v1",
|
|
"node22-alpine-bun-v1",
|
|
"sidecar",
|
|
"1b99888d3dae",
|
|
])
|
|
|
|
EXPECTED_G14_NODE = "ubuntu-rog-zephyrus-g14-ga401iv-ga401iv"
|
|
REMOTE_GC_JOB_DIR = "/tmp/unidesk-gc-remote/jobs"
|
|
REMOTE_STDOUT_JSON_LIMIT = 256 * 1024
|
|
|
|
def now_iso():
|
|
return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
|
|
|
|
def command(cmd, timeout=10):
|
|
try:
|
|
p = subprocess.run(cmd, text=True, capture_output=True, timeout=timeout)
|
|
return {"exitCode": p.returncode, "stdout": p.stdout, "stderr": p.stderr, "timedOut": False}
|
|
except subprocess.TimeoutExpired as exc:
|
|
return {
|
|
"exitCode": None,
|
|
"stdout": exc.stdout or "",
|
|
"stderr": exc.stderr or ("timed out after %ss" % timeout),
|
|
"timedOut": True,
|
|
}
|
|
except Exception as exc:
|
|
return {"exitCode": None, "stdout": "", "stderr": str(exc), "timedOut": False}
|
|
|
|
def bounded(result):
|
|
return {
|
|
"exitCode": result.get("exitCode"),
|
|
"timedOut": bool(result.get("timedOut")),
|
|
"stdoutTail": str(result.get("stdout") or "")[-2000:],
|
|
"stderrTail": str(result.get("stderr") or "")[-2000:],
|
|
}
|
|
|
|
def job_id_or_none():
|
|
raw = str(OPTIONS.get("jobId") or "")
|
|
if raw and re.match(r"^[A-Za-z0-9._-]{1,128}$", raw):
|
|
return raw
|
|
return None
|
|
|
|
def job_paths(job_id):
|
|
os.makedirs(REMOTE_GC_JOB_DIR, exist_ok=True)
|
|
return {
|
|
"state": os.path.join(REMOTE_GC_JOB_DIR, "%s.json" % job_id),
|
|
"log": os.path.join(REMOTE_GC_JOB_DIR, "%s.log" % job_id),
|
|
}
|
|
|
|
def status_command(job_id):
|
|
return "bun scripts/cli.ts gc remote %s status --job-id %s" % (PROVIDER_ID, job_id)
|
|
|
|
def write_json_atomic(path, payload):
|
|
tmp = "%s.tmp.%s" % (path, os.getpid())
|
|
with open(tmp, "w", encoding="utf-8") as handle:
|
|
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
os.replace(tmp, path)
|
|
|
|
def read_file_tail(path, limit=12000):
|
|
try:
|
|
size = os.path.getsize(path)
|
|
with open(path, "rb") as handle:
|
|
if size > limit:
|
|
handle.seek(size - limit)
|
|
data = handle.read()
|
|
return data.decode("utf-8", errors="replace")
|
|
except OSError:
|
|
return ""
|
|
|
|
def stdout_page(items):
|
|
if not isinstance(items, list):
|
|
return items
|
|
raw_limit = OPTIONS.get("resultLimit") or OPTIONS.get("limit") or 50
|
|
try:
|
|
limit = int(raw_limit)
|
|
except Exception:
|
|
limit = 50
|
|
limit = max(1, min(limit, 100))
|
|
return items[:limit]
|
|
|
|
def compact_payload_for_stdout(payload, full_size_bytes, job_id=None, paths=None):
|
|
compact = {
|
|
"ok": payload.get("ok", True),
|
|
"action": payload.get("action") or "gc remote",
|
|
"providerId": payload.get("providerId") or PROVIDER_ID,
|
|
"output": {
|
|
"truncated": True,
|
|
"reason": "stdout-size-guard",
|
|
"fullResultBytes": full_size_bytes,
|
|
},
|
|
}
|
|
for key in [
|
|
"dryRun", "mutation", "observedAt", "status", "kind", "mode",
|
|
"startedAt", "finishedAt", "error", "message", "options",
|
|
"diskBefore", "diskAfter", "clusterPreflight", "clusterAfter",
|
|
"summary", "policy",
|
|
]:
|
|
if key in payload:
|
|
compact[key] = payload[key]
|
|
if job_id:
|
|
state_path = paths["state"] if paths else payload.get("statePath")
|
|
compact["jobId"] = job_id
|
|
compact["statePath"] = state_path
|
|
compact["statusCommand"] = status_command(job_id)
|
|
compact["fullResult"] = {
|
|
"jobId": job_id,
|
|
"statePath": state_path,
|
|
"statusCommand": status_command(job_id),
|
|
}
|
|
compact["output"]["fullResultJobId"] = job_id
|
|
if "results" in payload:
|
|
results = payload.get("results") or []
|
|
compact["results"] = stdout_page(results)
|
|
compact["returnedResultCount"] = len(compact["results"])
|
|
compact["omittedResultCount"] = max(0, len(results) - len(compact["results"])) if isinstance(results, list) else 0
|
|
if "candidates" in payload:
|
|
candidates = payload.get("candidates") or []
|
|
compact["candidates"] = stdout_page(candidates)
|
|
compact["returnedCandidateCount"] = len(compact["candidates"])
|
|
compact["omittedCandidateCount"] = max(0, len(candidates) - len(compact["candidates"])) if isinstance(candidates, list) else 0
|
|
if "protected" in payload:
|
|
compact["protected"] = payload["protected"]
|
|
if "logTail" in payload:
|
|
compact["logTail"] = str(payload.get("logTail") or "")[-12000:]
|
|
return compact
|
|
|
|
def emit_json(payload, persist_large=True):
|
|
raw = json.dumps(payload, ensure_ascii=False, indent=2)
|
|
full_size = len(raw.encode("utf-8"))
|
|
if full_size <= REMOTE_STDOUT_JSON_LIMIT:
|
|
print(raw)
|
|
return
|
|
job_id = str(payload.get("jobId") or "")
|
|
paths = None
|
|
if persist_large:
|
|
if not job_id:
|
|
provider_slug = re.sub(r"[^A-Za-z0-9._-]+", "-", PROVIDER_ID.lower()).strip("-") or "provider"
|
|
job_id = "%s-gc-output-%s-%s" % (provider_slug, int(time.time()), os.getpid())
|
|
paths = job_paths(job_id)
|
|
payload = dict(payload)
|
|
payload.update({
|
|
"jobId": job_id,
|
|
"statePath": paths["state"],
|
|
"statusCommand": status_command(job_id),
|
|
"outputPersistedAt": now_iso(),
|
|
})
|
|
write_json_atomic(paths["state"], payload)
|
|
elif job_id:
|
|
paths = job_paths(job_id)
|
|
compact = compact_payload_for_stdout(payload, full_size, job_id or None, paths)
|
|
print(json.dumps(compact, ensure_ascii=False, indent=2))
|
|
|
|
def remote_gc_job_status():
|
|
job_id = job_id_or_none()
|
|
if not job_id:
|
|
return {
|
|
"ok": False,
|
|
"error": "gc-remote-status-requires-job-id",
|
|
"requiredFlag": "--job-id",
|
|
}
|
|
paths = job_paths(job_id)
|
|
if not os.path.isfile(paths["state"]):
|
|
return {
|
|
"ok": False,
|
|
"error": "gc-remote-job-not-found",
|
|
"jobId": job_id,
|
|
"statePath": paths["state"],
|
|
"logTail": read_file_tail(paths["log"]),
|
|
}
|
|
try:
|
|
with open(paths["state"], "r", encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
except Exception as exc:
|
|
return {
|
|
"ok": False,
|
|
"error": "gc-remote-job-state-invalid",
|
|
"jobId": job_id,
|
|
"message": str(exc),
|
|
"statePath": paths["state"],
|
|
"logTail": read_file_tail(paths["log"]),
|
|
}
|
|
payload["logTail"] = read_file_tail(paths["log"])
|
|
return payload
|
|
|
|
def path_size(path):
|
|
try:
|
|
if os.path.islink(path) or os.path.isfile(path):
|
|
return os.lstat(path).st_size
|
|
if not os.path.isdir(path):
|
|
return 0
|
|
total = 0
|
|
for root, dirs, files in os.walk(path):
|
|
for name in files:
|
|
child = os.path.join(root, name)
|
|
try:
|
|
total += os.lstat(child).st_size
|
|
except OSError:
|
|
pass
|
|
for name in dirs:
|
|
child = os.path.join(root, name)
|
|
try:
|
|
if os.path.islink(child):
|
|
total += os.lstat(child).st_size
|
|
except OSError:
|
|
pass
|
|
return total
|
|
except OSError:
|
|
return 0
|
|
|
|
def du_size(path, timeout=20):
|
|
if not os.path.exists(path):
|
|
return None
|
|
result = command(["du", "-sxB1", path], timeout)
|
|
if result["exitCode"] != 0:
|
|
return path_size(path)
|
|
text = result["stdout"].strip()
|
|
if not text:
|
|
return 0
|
|
try:
|
|
return int(text.split()[0])
|
|
except Exception:
|
|
return path_size(path)
|
|
|
|
def allocated_file_size(path):
|
|
try:
|
|
stat = os.stat(path)
|
|
blocks = getattr(stat, "st_blocks", 0)
|
|
if blocks:
|
|
return int(blocks) * 512
|
|
return int(stat.st_size)
|
|
except OSError:
|
|
return 0
|
|
|
|
def df_snapshot():
|
|
result = command(["df", "-B1", "-P", "/"], 5)
|
|
if result["exitCode"] != 0:
|
|
return None
|
|
lines = result["stdout"].strip().splitlines()
|
|
if len(lines) < 2:
|
|
return None
|
|
parts = lines[1].split()
|
|
if len(parts) < 6:
|
|
return None
|
|
return {
|
|
"filesystem": parts[0],
|
|
"sizeBytes": int(parts[1]),
|
|
"usedBytes": int(parts[2]),
|
|
"availableBytes": int(parts[3]),
|
|
"usePercent": int(parts[4].replace("%", "")),
|
|
"mount": parts[5],
|
|
}
|
|
|
|
def fmt_bytes(value):
|
|
units = ["B", "KiB", "MiB", "GiB", "TiB"]
|
|
size = float(max(0, int(value or 0)))
|
|
idx = 0
|
|
while size >= 1024 and idx < len(units) - 1:
|
|
size /= 1024.0
|
|
idx += 1
|
|
return ("%0.0f %s" if size >= 10 or idx == 0 else "%0.1f %s") % (size, units[idx])
|
|
|
|
def disk_use_percent(size_bytes, used_bytes):
|
|
try:
|
|
size = int(size_bytes or 0)
|
|
used = int(used_bytes or 0)
|
|
except Exception:
|
|
return None
|
|
if size <= 0:
|
|
return None
|
|
return int((max(0, used) * 100 + size - 1) // size)
|
|
|
|
def parse_journal_usage(text):
|
|
m = re.search(r"take up\s+([0-9.]+)\s*([KMGT]?)(?:i?B|B)?", text, re.I)
|
|
if not m:
|
|
return None
|
|
mult = {"": 1, "K": 1024, "M": 1024**2, "G": 1024**3, "T": 1024**4}.get(m.group(2).upper(), 1)
|
|
return int(float(m.group(1)) * mult)
|
|
|
|
def parse_docker_human_size(raw):
|
|
raw = str(raw).split("(")[0].strip()
|
|
m = re.match(r"^([0-9.]+)\s*([KMGT]?B)$", raw, re.I)
|
|
if not m:
|
|
return None
|
|
mult = {"B": 1, "KB": 1000, "MB": 1000**2, "GB": 1000**3, "TB": 1000**4}.get(m.group(2).upper(), 1)
|
|
return int(float(m.group(1)) * mult)
|
|
|
|
def parse_docker_build_cache(text):
|
|
for line in text.splitlines():
|
|
if not line.startswith("Build Cache"):
|
|
continue
|
|
match = re.match(r"^Build Cache\s+\S+\s+\S+\s+(\S+)\s+(\S+)", line.strip())
|
|
if not match:
|
|
continue
|
|
size = parse_docker_human_size(match.group(1))
|
|
reclaim = parse_docker_human_size(match.group(2))
|
|
if size is None or reclaim is None:
|
|
return None
|
|
return {"sizeBytes": size, "reclaimableBytes": reclaim}
|
|
return None
|
|
|
|
def docker_containers():
|
|
ps = command(["docker", "ps", "-qa", "--no-trunc"], 5)
|
|
if ps["exitCode"] != 0 or not ps["stdout"].strip():
|
|
return []
|
|
ids = ps["stdout"].split()
|
|
inspect = command(["docker", "inspect"] + ids, 10)
|
|
if inspect["exitCode"] != 0 or not inspect["stdout"].strip():
|
|
return []
|
|
try:
|
|
data = json.loads(inspect["stdout"])
|
|
except Exception:
|
|
return []
|
|
rows = []
|
|
for item in data:
|
|
cfg = item.get("Config") or {}
|
|
rows.append({
|
|
"id": str(item.get("Id") or ""),
|
|
"name": str(item.get("Name") or "").lstrip("/"),
|
|
"image": str(cfg.get("Image") or item.get("Image") or ""),
|
|
"logPath": str(item.get("LogPath") or ""),
|
|
})
|
|
return [row for row in rows if row["id"]]
|
|
|
|
def cluster_preflight():
|
|
node_cmd = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get nodes -o jsonpath='{range .items[*]}{.metadata.name}{\"\\n\"}{end}' 2>/dev/null"], 10)
|
|
pods_cmd = command(["sh", "-lc", "KUBECONFIG=/etc/rancher/k3s/k3s.yaml kubectl get pods -n hwlab-dev --no-headers 2>/dev/null | wc -l"], 10)
|
|
nodes = [line.strip() for line in node_cmd["stdout"].splitlines() if line.strip()]
|
|
expected = EXPECTED_G14_NODE if PROVIDER_ID.upper() == "G14" else None
|
|
ok = True
|
|
reason = "ok"
|
|
if expected is not None and expected not in nodes:
|
|
ok = False
|
|
reason = "expected-g14-node-missing"
|
|
return {
|
|
"ok": ok,
|
|
"reason": reason,
|
|
"providerId": PROVIDER_ID,
|
|
"hostname": command(["hostname"], 5)["stdout"].strip(),
|
|
"expectedNode": expected,
|
|
"nodes": nodes,
|
|
"nodeCommand": bounded(node_cmd),
|
|
"hwlabDevPodCount": int(pods_cmd["stdout"].strip() or "0") if pods_cmd["exitCode"] == 0 else None,
|
|
"hwlabDevPodCommand": bounded(pods_cmd),
|
|
}
|
|
|
|
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
|
|
|
|
def collect_protected():
|
|
protected_paths = [
|
|
("hwlab-k3s-runtime", "/var/lib/rancher/k3s", "Native k3s runtime, containerd state, local-path storage and control-plane data are protected."),
|
|
("hwlab-k3s-storage", "/var/lib/rancher/k3s/storage", "Local-path PVC data is protected; cleanup must go through HWLAB/Tekton retention commands."),
|
|
("hwlab-kubelet", "/var/lib/kubelet", "Kubelet pod/runtime state is protected."),
|
|
("host-containerd", "/var/lib/containerd", "Host containerd state is protected; generic gc does not prune containerd images."),
|
|
("hwlab-host-data", "/var/lib/hwlab", "HWLAB host data/cache is protected from generic remote gc until a HWLAB-specific retention rule classifies it."),
|
|
("hwlab-source", "/root/hwlab", "G14 HWLAB fixed source workspace is protected."),
|
|
("hwlab-v02-source", "/root/hwlab-v02", "HWLAB v0.2 fixed source workspace is protected."),
|
|
("agentrun-source", "/root/agentrun", "AgentRun fixed source workspace is protected."),
|
|
("docker-images-and-volumes", "docker-images-volumes", "Remote gc does not remove Docker images, containers, volumes or Compose projects."),
|
|
("k8s-api-objects", "deployments-statefulsets-secrets-pvcs", "Remote gc does not mutate Kubernetes workloads, Secrets, PVCs, PVs, Argo CD or Tekton objects."),
|
|
]
|
|
result = []
|
|
for kind, ref, reason in protected_paths:
|
|
item = {"kind": kind, "risk": "blocked", "ref": ref, "reason": reason}
|
|
if ref.startswith("/") and os.path.exists(ref):
|
|
item["sizeBytes"] = du_size(ref)
|
|
result.append(item)
|
|
return result
|
|
|
|
def collect_candidates(observed_at):
|
|
candidates = []
|
|
if OPTIONS.get("journal", True):
|
|
usage = command(["journalctl", "--disk-usage"], 5)
|
|
current = parse_journal_usage((usage["stdout"] or "") + (usage["stderr"] or ""))
|
|
target = int(OPTIONS.get("journalTargetBytes") or 536870912)
|
|
if current is not None and current > target:
|
|
candidates.append({
|
|
"id": "journalctl:vacuum",
|
|
"kind": "journal-vacuum",
|
|
"risk": "medium",
|
|
"description": "Vacuum systemd journal to %s" % fmt_bytes(target),
|
|
"sizeBytes": current,
|
|
"estimatedReclaimBytes": max(0, current - target),
|
|
"action": {"command": ["journalctl", "--vacuum-size=%s" % target]},
|
|
})
|
|
|
|
if OPTIONS.get("dockerLogs", True):
|
|
max_bytes = int(OPTIONS.get("dockerLogMaxBytes") or 52428800)
|
|
for container in docker_containers():
|
|
path = container.get("logPath") or ""
|
|
if not path or not os.path.exists(path):
|
|
continue
|
|
try:
|
|
size = os.path.getsize(path)
|
|
except OSError:
|
|
continue
|
|
if size <= max_bytes:
|
|
continue
|
|
candidates.append({
|
|
"id": "docker-json-log:%s" % container["id"],
|
|
"kind": "docker-json-log-truncate",
|
|
"risk": "medium",
|
|
"description": "Truncate Docker json-file log larger than %s" % fmt_bytes(max_bytes),
|
|
"path": path,
|
|
"container": {"id": container["id"][:12], "name": container["name"], "image": container["image"]},
|
|
"sizeBytes": size,
|
|
"estimatedReclaimBytes": size,
|
|
"action": {"op": "truncate", "targetBytes": 0},
|
|
})
|
|
|
|
if OPTIONS.get("buildCache", True):
|
|
system_df = command(["docker", "system", "df"], 8)
|
|
if system_df["exitCode"] == 0:
|
|
cache = parse_docker_build_cache(system_df["stdout"])
|
|
if cache is not None and cache["reclaimableBytes"] > 0:
|
|
until = str(OPTIONS.get("buildCacheUntil") or "24h")
|
|
candidates.append({
|
|
"id": "docker-builder:prune",
|
|
"kind": "docker-build-cache-prune",
|
|
"risk": "low",
|
|
"description": "Prune Docker BuildKit cache unused for %s" % until,
|
|
"sizeBytes": cache["sizeBytes"],
|
|
"estimatedReclaimBytes": cache["reclaimableBytes"],
|
|
"action": {"command": ["docker", "builder", "prune", "--all", "--force", "--filter", "until=%s" % until], "estimate": "docker-system-df-reclaimable-upper-bound"},
|
|
})
|
|
|
|
if OPTIONS.get("aptCache", True):
|
|
apt_path = "/var/cache/apt/archives"
|
|
size = du_size(apt_path) or 0
|
|
if size > 10 * 1024 * 1024:
|
|
candidates.append({
|
|
"id": "apt-cache:clean",
|
|
"kind": "apt-cache-clean",
|
|
"risk": "low",
|
|
"description": "Clean downloaded apt package archives",
|
|
"path": apt_path,
|
|
"sizeBytes": size,
|
|
"estimatedReclaimBytes": size,
|
|
"action": {"command": ["apt-get", "clean"]},
|
|
})
|
|
|
|
if OPTIONS.get("coreDumps", True):
|
|
cutoff = time.time() - float(OPTIONS.get("coreDumpMinAgeHours") or 1) * 3600
|
|
for root in sorted(CORE_DUMP_DIR_ALLOWLIST):
|
|
if not os.path.isdir(root):
|
|
continue
|
|
for name in os.listdir(root):
|
|
if not re.match(r"^core\.\d+$", name):
|
|
continue
|
|
path = os.path.join(root, name)
|
|
try:
|
|
stat = os.lstat(path)
|
|
except OSError:
|
|
continue
|
|
if not os.path.isfile(path) or os.path.islink(path):
|
|
continue
|
|
if stat.st_mtime >= cutoff:
|
|
continue
|
|
disk_size = allocated_file_size(path)
|
|
candidates.append({
|
|
"id": "core-dump:%s" % path,
|
|
"kind": "core-dump-delete",
|
|
"risk": "low",
|
|
"description": "Delete untracked process core dump older than %s hours" % OPTIONS.get("coreDumpMinAgeHours"),
|
|
"path": path,
|
|
"sizeBytes": int(disk_size),
|
|
"estimatedReclaimBytes": int(disk_size),
|
|
"apparentSizeBytes": int(stat.st_size),
|
|
"action": {"op": "unlink", "allowlist": "root-unidesk-core-dot-pid"},
|
|
})
|
|
|
|
if OPTIONS.get("tmp", True) and os.path.isdir("/tmp"):
|
|
cutoff = time.time() - float(OPTIONS.get("tmpMinAgeHours") or 24) * 3600
|
|
for name in os.listdir("/tmp"):
|
|
path = os.path.join("/tmp", name)
|
|
if path in TMP_EXACT_PROTECT:
|
|
continue
|
|
if not any(name.startswith(prefix) for prefix in TMP_PREFIX_ALLOWLIST):
|
|
continue
|
|
try:
|
|
stat = os.lstat(path)
|
|
except OSError:
|
|
continue
|
|
if stat.st_mtime >= cutoff:
|
|
continue
|
|
size = du_size(path, 8) or path_size(path)
|
|
if size <= 0:
|
|
continue
|
|
candidates.append({
|
|
"id": "tmp:%s" % path,
|
|
"kind": "tmp-path-delete",
|
|
"risk": "low",
|
|
"description": "Delete allowlisted /tmp path older than %s hours" % OPTIONS.get("tmpMinAgeHours"),
|
|
"path": path,
|
|
"sizeBytes": size,
|
|
"estimatedReclaimBytes": size,
|
|
"action": {"op": "rm-recursive", "allowlist": "tmp-prefix"},
|
|
})
|
|
if OPTIONS.get("hwlabRegistry", False):
|
|
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 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, 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,
|
|
"action": {
|
|
"op": "registry-retention-gc",
|
|
"requiresMaintenanceWindow": True,
|
|
"keepPerRepo": summary.get("keepPerRepo"),
|
|
"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:
|
|
candidates.append({
|
|
"id": "hwlab-registry:garbage-collect-only",
|
|
"kind": "hwlab-registry-garbage-collect",
|
|
"risk": "medium",
|
|
"description": "Run official HWLAB registry garbage-collect without deleting additional tags; useful after a previously interrupted retention run",
|
|
"path": REGISTRY_ROOT,
|
|
"sizeBytes": int(summary.get("registrySizeBytes") or 0),
|
|
"estimatedReclaimBytes": 0,
|
|
"action": {
|
|
"op": "registry-garbage-collect-only",
|
|
"requiresMaintenanceWindow": True,
|
|
"deleteTags": 0,
|
|
"registryPlan": summary,
|
|
},
|
|
})
|
|
return sorted(candidates, key=lambda item: item.get("estimatedReclaimBytes") or 0, reverse=True)
|
|
|
|
def target_assessment(disk, estimated_reclaim):
|
|
raw = OPTIONS.get("targetUsePercent")
|
|
if raw is None:
|
|
return None
|
|
if not disk:
|
|
return {
|
|
"targetUsePercent": raw,
|
|
"ok": False,
|
|
"state": "unavailable",
|
|
"reason": "disk-snapshot-unavailable",
|
|
}
|
|
try:
|
|
target = int(raw)
|
|
size = int(disk.get("sizeBytes") or 0)
|
|
used = int(disk.get("usedBytes") or 0)
|
|
reclaim = max(0, int(estimated_reclaim or 0))
|
|
except Exception:
|
|
return {
|
|
"targetUsePercent": raw,
|
|
"ok": False,
|
|
"state": "unavailable",
|
|
"reason": "invalid-disk-snapshot",
|
|
}
|
|
target_used_bytes = (size * target) // 100
|
|
required = max(0, used - target_used_bytes)
|
|
projected_used = max(0, used - reclaim)
|
|
projected_use_percent = disk_use_percent(size, projected_used)
|
|
enough = reclaim >= required
|
|
if required == 0:
|
|
state = "already-below-target"
|
|
elif enough:
|
|
state = "candidate-estimate-meets-target"
|
|
elif reclaim <= 0:
|
|
state = "safe-stop-no-meaningful-candidates"
|
|
else:
|
|
state = "shortfall"
|
|
shortfall = max(0, required - reclaim)
|
|
return {
|
|
"targetUsePercent": target,
|
|
"ok": required == 0 or enough,
|
|
"state": state,
|
|
"currentUsePercent": disk.get("usePercent"),
|
|
"currentUsedBytes": used,
|
|
"currentUsed": fmt_bytes(used),
|
|
"targetUsedBytes": target_used_bytes,
|
|
"targetUsed": fmt_bytes(target_used_bytes),
|
|
"requiredReclaimBytes": required,
|
|
"requiredReclaim": fmt_bytes(required),
|
|
"estimatedReclaimBytes": reclaim,
|
|
"estimatedReclaim": fmt_bytes(reclaim),
|
|
"shortfallBytes": shortfall,
|
|
"shortfall": fmt_bytes(shortfall),
|
|
"projectedUsedBytes": projected_used,
|
|
"projectedUsed": fmt_bytes(projected_used),
|
|
"projectedUsePercent": projected_use_percent,
|
|
"safeStop": required > 0 and not enough,
|
|
"decision": "stop-and-escalate-retention-or-capacity" if required > 0 and not enough else "target-covered-by-safe-candidates",
|
|
}
|
|
|
|
def summarize(candidates, returned, disk=None):
|
|
by_kind = {}
|
|
total = 0
|
|
for item in candidates:
|
|
size = int(item.get("estimatedReclaimBytes") or 0)
|
|
total += size
|
|
kind = item.get("kind") or "unknown"
|
|
current = by_kind.setdefault(kind, {"count": 0, "estimatedReclaimBytes": 0, "estimatedReclaim": "0 B"})
|
|
current["count"] += 1
|
|
current["estimatedReclaimBytes"] += size
|
|
current["estimatedReclaim"] = fmt_bytes(current["estimatedReclaimBytes"])
|
|
returned_total = sum(int(item.get("estimatedReclaimBytes") or 0) for item in returned)
|
|
return {
|
|
"candidateCount": len(candidates),
|
|
"returnedCandidateCount": len(returned),
|
|
"estimatedReclaimBytes": total,
|
|
"estimatedReclaim": fmt_bytes(total),
|
|
"returnedEstimatedReclaimBytes": returned_total,
|
|
"returnedEstimatedReclaim": fmt_bytes(returned_total),
|
|
"byKind": by_kind,
|
|
"target": target_assessment(disk, total),
|
|
}
|
|
|
|
def assert_tmp_candidate(path):
|
|
resolved = os.path.abspath(path)
|
|
if not resolved.startswith("/tmp/"):
|
|
raise RuntimeError("refusing to remove non-/tmp path: %s" % path)
|
|
if resolved in TMP_EXACT_PROTECT:
|
|
raise RuntimeError("refusing to remove protected tmp path: %s" % path)
|
|
name = os.path.basename(resolved)
|
|
if not any(name.startswith(prefix) for prefix in TMP_PREFIX_ALLOWLIST):
|
|
raise RuntimeError("refusing to remove tmp path outside allowlist: %s" % path)
|
|
|
|
def assert_core_dump_candidate(path):
|
|
resolved = os.path.abspath(path)
|
|
parent = os.path.dirname(resolved)
|
|
name = os.path.basename(resolved)
|
|
if parent not in CORE_DUMP_DIR_ALLOWLIST:
|
|
raise RuntimeError("refusing to remove core dump outside allowlisted directory: %s" % path)
|
|
if not re.match(r"^core\.\d+$", name):
|
|
raise RuntimeError("refusing to remove non core.<pid> file: %s" % path)
|
|
if not os.path.isfile(resolved) or os.path.islink(resolved):
|
|
raise RuntimeError("refusing to remove non-regular core dump: %s" % path)
|
|
git = command(["git", "-C", parent, "ls-files", "--error-unmatch", name], 10)
|
|
if git["exitCode"] == 0:
|
|
raise RuntimeError("refusing to remove git-tracked file: %s" % path)
|
|
if git["exitCode"] != 1:
|
|
raise RuntimeError("refusing to remove core dump because git tracking check failed: %s" % path)
|
|
fuser = command(["fuser", resolved], 5)
|
|
if fuser["exitCode"] is None:
|
|
raise RuntimeError("refusing to remove core dump because fuser check was unavailable: %s" % path)
|
|
if fuser["exitCode"] == 0:
|
|
raise RuntimeError("refusing to remove core dump with active process reference: %s" % path)
|
|
|
|
def execute(candidate):
|
|
kind = candidate.get("kind")
|
|
if kind == "journal-vacuum":
|
|
result = command(["journalctl", "--vacuum-size=%s" % int(OPTIONS.get("journalTargetBytes") or 536870912)], 30)
|
|
if result["exitCode"] != 0:
|
|
raise RuntimeError((result["stderr"] or "journalctl vacuum failed").strip())
|
|
return {"reclaimedBytes": None, "commandOutput": bounded(result)}
|
|
if kind == "docker-json-log-truncate":
|
|
path = candidate.get("path") or ""
|
|
if not path.startswith("/var/lib/docker/containers/"):
|
|
raise RuntimeError("refusing to truncate Docker log outside /var/lib/docker/containers")
|
|
before = os.path.getsize(path) if os.path.exists(path) else 0
|
|
with open(path, "r+b") as handle:
|
|
handle.truncate(0)
|
|
return {"reclaimedBytes": before}
|
|
if kind == "docker-build-cache-prune":
|
|
until = str(OPTIONS.get("buildCacheUntil") or "24h")
|
|
result = command(["docker", "builder", "prune", "--all", "--force", "--filter", "until=%s" % until], 45)
|
|
if result["exitCode"] != 0:
|
|
raise RuntimeError((result["stderr"] or "docker builder prune failed").strip())
|
|
return {"reclaimedBytes": None, "commandOutput": bounded(result)}
|
|
if kind == "apt-cache-clean":
|
|
before = du_size("/var/cache/apt/archives") or 0
|
|
result = command(["apt-get", "clean"], 30)
|
|
if result["exitCode"] != 0:
|
|
raise RuntimeError((result["stderr"] or "apt-get clean failed").strip())
|
|
after = du_size("/var/cache/apt/archives") or 0
|
|
return {"reclaimedBytes": max(0, before - after), "commandOutput": bounded(result)}
|
|
if kind == "tmp-path-delete":
|
|
path = candidate.get("path") or ""
|
|
assert_tmp_candidate(path)
|
|
before = du_size(path, 8) or path_size(path)
|
|
if os.path.isdir(path) and not os.path.islink(path):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
else:
|
|
try:
|
|
os.unlink(path)
|
|
except FileNotFoundError:
|
|
pass
|
|
return {"reclaimedBytes": before}
|
|
if kind == "core-dump-delete":
|
|
path = candidate.get("path") or ""
|
|
assert_core_dump_candidate(path)
|
|
before = allocated_file_size(path)
|
|
os.unlink(path)
|
|
return {"reclaimedBytes": before}
|
|
if kind == "hwlab-registry-retention-gc":
|
|
return start_registry_retention_job("retention")
|
|
if kind == "hwlab-registry-garbage-collect":
|
|
return start_registry_retention_job("garbage-collect")
|
|
raise RuntimeError("unsupported remote gc candidate kind: %s" % kind)
|
|
|
|
def visible_items(items):
|
|
if bool(OPTIONS.get("full")):
|
|
return items
|
|
return items[:int(OPTIONS.get("limit") or 50)]
|
|
|
|
def returned_results(results):
|
|
if bool(OPTIONS.get("full")):
|
|
return results
|
|
failed = [item for item in results if item.get("status") == "failed"]
|
|
started = [item for item in results if item.get("status") == "started"]
|
|
succeeded = [item for item in results if item.get("status") == "succeeded"]
|
|
return (failed + started + succeeded)[:int(OPTIONS.get("resultLimit") or 50)]
|
|
|
|
def plan_payload(observed_at, preflight, protected, candidates, visible):
|
|
disk = df_snapshot()
|
|
return {
|
|
"ok": True,
|
|
"action": "gc remote plan",
|
|
"providerId": PROVIDER_ID,
|
|
"dryRun": True,
|
|
"mutation": False,
|
|
"observedAt": observed_at,
|
|
"options": OPTIONS,
|
|
"diskBefore": disk,
|
|
"clusterPreflight": preflight,
|
|
"summary": summarize(candidates, visible, disk),
|
|
"candidates": visible,
|
|
"protected": protected,
|
|
"policy": {
|
|
"requiresRunConfirm": True,
|
|
"runCommand": "bun scripts/cli.ts gc remote %s run --confirm" % PROVIDER_ID,
|
|
"neverTouches": [
|
|
"/var/lib/rancher/k3s",
|
|
"/var/lib/rancher/k3s/storage",
|
|
"/var/lib/kubelet",
|
|
"/var/lib/containerd",
|
|
"/var/lib/hwlab unless --include-hwlab-registry is explicitly supplied",
|
|
"Kubernetes Deployments/StatefulSets/Secrets/PVCs/PVs",
|
|
"HWLAB fixed source workspaces",
|
|
"Docker images, containers and volumes",
|
|
],
|
|
"notes": [
|
|
"Remote gc only executes the returned candidate page unless --full or a larger --limit is supplied.",
|
|
"G14 run requires the expected native k3s node preflight before mutation.",
|
|
"HWLAB DEV runtime and local-path PVC data are protected and require HWLAB-specific retention commands.",
|
|
"Core dump cleanup only removes untracked /root/unidesk/core.<pid> regular files with no active fuser reference.",
|
|
"HWLAB registry retention is opt-in: it keeps workload tag/digest refs, all tags newer than the retention age and the newest N tags per repo before official registry garbage-collect.",
|
|
"When summary.target.safeStop is true, do not broaden deletion scope; choose registry retention, k3s/containerd image cache maintenance, PVC/runtime retention or capacity expansion explicitly.",
|
|
],
|
|
},
|
|
}
|
|
|
|
def main():
|
|
observed_at = now_iso()
|
|
preflight = cluster_preflight()
|
|
protected = collect_protected()
|
|
candidates = collect_candidates(observed_at)
|
|
visible = visible_items(candidates)
|
|
if ACTION == "plan":
|
|
emit_json(plan_payload(observed_at, preflight, protected, candidates, visible), persist_large=True)
|
|
return 0
|
|
if ACTION == "status":
|
|
emit_json(remote_gc_job_status(), persist_large=False)
|
|
return 0
|
|
if ACTION != "run":
|
|
emit_json({"ok": False, "error": "unsupported-remote-gc-action", "action": ACTION}, persist_large=False)
|
|
return 0
|
|
if PROVIDER_ID.upper() == "G14" and not preflight.get("ok"):
|
|
emit_json({
|
|
"ok": False,
|
|
"error": "gc-remote-g14-preflight-failed",
|
|
"action": "gc remote run",
|
|
"providerId": PROVIDER_ID,
|
|
"dryRun": True,
|
|
"mutation": False,
|
|
"clusterPreflight": preflight,
|
|
"plan": plan_payload(observed_at, preflight, protected, candidates, visible),
|
|
}, persist_large=True)
|
|
return 0
|
|
disk_before = df_snapshot()
|
|
results = []
|
|
for candidate in visible:
|
|
try:
|
|
execution = execute(candidate)
|
|
item = dict(candidate)
|
|
item.update({"status": execution.get("status") or "succeeded", "reclaimedBytes": execution.get("reclaimedBytes")})
|
|
if "commandOutput" in execution:
|
|
item["commandOutput"] = execution["commandOutput"]
|
|
results.append(item)
|
|
except Exception as exc:
|
|
item = dict(candidate)
|
|
item.update({"status": "failed", "reclaimedBytes": None, "error": str(exc)})
|
|
results.append(item)
|
|
disk_after = df_snapshot()
|
|
failed = [item for item in results if item.get("status") == "failed"]
|
|
returned = returned_results(results)
|
|
run_summary = summarize(visible, returned, disk_before)
|
|
run_summary.update({
|
|
"plannedCandidateCount": len(visible),
|
|
"attemptedCount": len(results),
|
|
"startedCount": len([item for item in results if item.get("status") == "started"]),
|
|
"succeededCount": len([item for item in results if item.get("status") == "succeeded"]),
|
|
"failedCount": len(failed),
|
|
"actualDiskReclaimBytes": (disk_after["availableBytes"] - disk_before["availableBytes"]) if disk_before and disk_after else None,
|
|
"actualDiskReclaim": fmt_bytes(disk_after["availableBytes"] - disk_before["availableBytes"]) if disk_before and disk_after else None,
|
|
"targetAfter": target_assessment(disk_after, 0),
|
|
"resultCount": len(results),
|
|
"returnedResultCount": len(returned),
|
|
"omittedResultCount": max(0, len(results) - len(returned)),
|
|
})
|
|
payload = {
|
|
"ok": len(failed) == 0,
|
|
"action": "gc remote run",
|
|
"providerId": PROVIDER_ID,
|
|
"dryRun": False,
|
|
"mutation": True,
|
|
"observedAt": now_iso(),
|
|
"options": OPTIONS,
|
|
"diskBefore": disk_before,
|
|
"diskAfter": disk_after,
|
|
"clusterPreflight": preflight,
|
|
"clusterAfter": cluster_preflight(),
|
|
"summary": run_summary,
|
|
"results": returned,
|
|
"protected": protected,
|
|
}
|
|
emit_json(payload, persist_large=True)
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
`;
|
|
}
|