公司的Harbor镜像仓是个无比古早的版本：v1.7.1-c3806525，算是无比笨重
现在的话，肯定用 zot 来搭建了
Harbor 由于用了近 10 年了，有很多无效的镜像存在，0字节，删不掉，单个仓都有1400多个了
想了很久如何把这个破事给了结，终极大法就是扫描出这些损坏的镜像，然后干脆用 busybox 的最小镜像推上去顶替掉，然后再删掉。
说老实话，过程一点也不简单！！！！
首先要安装 skopeo，用这个进行推送，因为新版本Docker的格式已经变了，现在直接拉busybox的镜像，docker push 推不上去，格式不兼容！！！先拿一个TAG来测试：
apt-get install -y skopeo
docker pull busybox:1.37-musl
docker save busybox:1.37-musl -o /tmp/busybox.tar
export HARBOR_USER=admin
export HARBOR_PASSWORD=xxxxxxx
export BAD_TAG=tag-release-202511261018.20251126102001
# 转换格式，推一个试试
skopeo copy --format v2s2 docker-archive:/tmp/busybox.tar "docker://harbor.rendoumi.com/svc/order-service:${BAD_TAG}"
# 然后删除一个试试
curl -sS -i -X DELETE -u "$HARBOR_USER:$HARBOR_PASSWORD" "https://harbor.rendoumi.com/api/repositories/svc/order-service/tags/${BAD_TAG}"
# 回复是200就是没问题了
HTTP/1.1 200 OK
Server: nginx
Date: Mon, 26 Oct 2026 12:22:23 GMT
Content-Type: text/plain; charset=utf-8
Content-Length: 0
Connection: keep-alive
Set-Cookie: sid=xxxxxxxx; Path=/; HttpOnly
然后写个Python3的程序来自动清理吧：
注意参数，需要从harbor里获得：
# Harbor的参数
HARBOR_URL = os.getenv("HARBOR_URL", "https://harbor.rendoumi.com").rstrip("/")
HARBOR_USER = "admin"
HARBOR_PASSWORD = "xxxxxxx"
# 仅清理指定项目和仓库
PROJECT_ID = 2
REPOSITORY = "svc/order-service"
运行方法：
# 先预跑一遍，看看
./clean_harbor.py
# 来吧，真的动手删除
./clean_harbor.py --delete --confirm
clear_harbor.py的内容如下：
#!/usr/bin/env python3
import os
import sys
import argparse
import subprocess
import shutil
from urllib.parse import quote
import requests
from requests.auth import HTTPBasicAuth
HARBOR_URL = os.getenv("HARBOR_URL", "https://harbor.rendoumi.com").rstrip("/")
HARBOR_USER = "admin"
HARBOR_PASSWORD = "xxxxxxx"
# 仅清理指定项目和仓库
PROJECT_ID = 2
REPOSITORY = "svc/order-service"
# 用于修复异常 Tag 的本地镜像归档
BUSYBOX_ARCHIVE = "/tmp/busybox.tar"
# 异常 Tag 的三个特征必须同时满足
ZERO_CREATED_TIME = "0001-01-01T00:00:00Z"
def create_session():
if not HARBOR_USER or not HARBOR_PASSWORD:
sys.exit(
"错误：请设置 HARBOR_USER 和 HARBOR_PASSWORD 环境变量"
)
session = requests.Session()
session.auth = HTTPBasicAuth(HARBOR_USER, HARBOR_PASSWORD)
session.verify = True
session.headers.update({"Accept": "application/json"})
return session
def get_tags(session):
"""获取仓库全部标签；超时或临时错误时自动重试。"""
repo_path = quote(REPOSITORY, safe="")
url = f"{HARBOR_URL}/api/repositories/{repo_path}/tags"
max_retries = 5
timeout = (10, 180) # 连接超时 10 秒，读取超时 180 秒
for attempt in range(1, max_retries + 1):
try:
print(
f"正在读取标签（第 {attempt}/{max_retries} 次尝试）……"
)
response = session.get(url, timeout=timeout)
if response.status_code != 200:
if response.status_code in (429, 500, 502, 503, 504):
print(
f"Harbor 暂时返回 HTTP {response.status_code}"
)
else:
raise RuntimeError(
f"获取标签失败：HTTP {response.status_code}\n"
f"{response.text[:1000]}"
)
else:
data = response.json()
if not isinstance(data, list):
raise RuntimeError(
f"标签 API 返回了非列表数据：{data!r}"
)
print(f"成功读取 {len(data)} 个标签")
return data
except (requests.Timeout, requests.ConnectionError) as exc:
print(f"请求暂时失败：{exc}")
if attempt < max_retries:
wait_seconds = min(attempt * 5, 20)
print(f"{wait_seconds} 秒后重试……")
time.sleep(wait_seconds)
raise RuntimeError(
f"连续 {max_retries} 次尝试后仍无法获取标签，请检查 Harbor 响应情况"
)
def is_malformed(tag):
"""严格筛选符合全部三个异常条件的标签。"""
return (
tag.get("size") == 0
and tag.get("digest") == ""
and tag.get("created") == ZERO_CREATED_TIME
)
def repair_tag_with_skopeo(tag_name):
"""
先将 Docker Schema 2 格式的 BusyBox 镜像推送到目标 Tag。
只有 skopeo 返回成功，才允许继续执行 DELETE。
"""
destination = (
f"docker://harbor.bybon.cn/"
f"{REPOSITORY}:{tag_name}"
)
command = [
"skopeo",
"copy",
"--format",
"v2s2",
f"docker-archive:{BUSYBOX_ARCHIVE}",
destination,
]
print(f"[修复] 正在向 Harbor 推送有效 manifest：{tag_name}")
try:
result = subprocess.run(
command,
check=False,
text=True,
)
except OSError as exc:
raise RuntimeError(
f"无法执行 skopeo：{exc}"
) from exc
if result.returncode != 0:
raise RuntimeError(
f"skopeo 推送失败，退出码：{result.returncode}"
)
print(f"[修复成功] {tag_name}")
def delete_tag(session, tag_name):
"""调用 Harbor 1.7.1 旧版 API 删除单个 Tag。"""
repo_path = quote(REPOSITORY, safe="")
encoded_tag = quote(tag_name, safe="")
url = (
f"{HARBOR_URL}/api/repositories/{repo_path}"
f"/tags/{encoded_tag}"
)
return session.delete(url, timeout=30)
def verify_tag_deleted(session, tag_name):
"""查询 Tag，确认删除结果。"""
repo_path = quote(REPOSITORY, safe="")
encoded_tag = quote(tag_name, safe="")
url = (
f"{HARBOR_URL}/api/repositories/{repo_path}"
f"/tags/{encoded_tag}"
)
response = session.get(url, timeout=30)
if response.status_code == 404:
return True
if response.status_code == 200:
return False
raise RuntimeError(
f"验证删除结果失败：HTTP {response.status_code}\n"
f"{response.text[:300]}"
)
def main():
parser = argparse.ArgumentParser(
description=(
"清理 Harbor 1.7.1 异常标签："
"先用 skopeo 修复 manifest，再调用 API 删除"
)
)
parser.add_argument(
"--delete",
action="store_true",
help="执行实际修复和删除；默认仅扫描预览",
)
parser.add_argument(
"--confirm",
action="store_true",
help="执行前确认项目和仓库范围",
)
args = parser.parse_args()
if args.confirm and not args.delete:
parser.error("--confirm 必须与 --delete 一起使用")
if not shutil.which("skopeo"):
sys.exit("错误：系统中未找到 skopeo，请先安装")
session = create_session()
print(f"Harbor: {HARBOR_URL}")
print(f"Project ID: {PROJECT_ID}")
print(f"Repository: {REPOSITORY}")
print(f"镜像归档: {BUSYBOX_ARCHIVE}")
print()
print("正在读取标签……")
tags = get_tags(session)
malformed = [
tag for tag in tags
if is_malformed(tag)
]
print(f"\n标签总数：{len(tags)}")
print(f"符合异常条件的标签数：{len(malformed)}")
if not malformed:
print("没有发现符合全部三个条件的异常标签。")
return
print("\n待处理标签：")
for tag in malformed:
print(
f" {tag.get('name', '<unknown>')} | "
f"size={tag.get('size')} | "
f"digest={tag.get('digest')!r} | "
f"created={tag.get('created')}"
)
if not args.delete:
print("\n当前为预览模式，没有修改或删除任何标签。")
print("确认列表无误后，使用 --delete --confirm 执行。")
return
# 实际操作前检查本地镜像归档
if not os.path.isfile(BUSYBOX_ARCHIVE):
sys.exit(
f"错误：找不到镜像归档文件 {BUSYBOX_ARCHIVE}"
)
if args.confirm:
expected = f"{PROJECT_ID}/{REPOSITORY}"
answer = input(
f"\n输入 {expected} 确认修复并删除上述标签："
).strip()
if answer != expected:
print("确认内容不匹配，取消操作。")
return
success = 0
failed = 0
for index, tag in enumerate(malformed, 1):
name = tag.get("name")
if not isinstance(name, str) or not name:
print(f"[跳过] 第 {index} 个标签没有有效名称")
failed += 1
continue
print(f"\n[{index}/{len(malformed)}] 处理 Tag：{name}")
# 第一步：用 skopeo 修复 manifest
try:
repair_tag_with_skopeo(name)
except RuntimeError as exc:
print(f"[失败] {name}：{exc}")
print("[安全措施] 推送失败，不执行 DELETE")
failed += 1
continue
# 第二步：调用 Harbor API 删除
try:
response = delete_tag(session, name)
if response.status_code not in (200, 202, 204):
print(
f"[删除失败] {name} "
f"(HTTP {response.status_code}): "
f"{response.text[:300]}"
)
failed += 1
continue
print(
f"[DELETE 已接受] {name} "
f"(HTTP {response.status_code})"
)
except requests.RequestException as exc:
print(f"[删除异常] {name}: {exc}")
failed += 1
continue
# 第三步：重新查询，验证 Tag 确实消失
try:
if verify_tag_deleted(session, name):
print(f"[成功] {name} 已确认删除")
success += 1
else:
print(
f"[验证失败] {name}："
"DELETE 已返回成功，但 Tag 仍然存在"
)
failed += 1
except (requests.RequestException, RuntimeError) as exc:
print(f"[验证异常] {name}: {exc}")
failed += 1
print("\n清理结束")
print(f"确认删除成功：{success}")
print(f"失败或未能确认：{failed}")
if __name__ == "__main__":
main()