镜像同步管理平台

多环境（IDC ↔ 云、生产 ↔ 灾备）常需镜像复制：开发推送到 Harbor A，生产集群只信任 Harbor B。手工 docker pull/tag/push 易出错且难审计。本平台提供：同步任务配置 → 队列执行 → 状态追踪 → Webhook 触发。

---

架构思路

┌──────────────┐    POST /sync     ┌─────────────────┐
│ Harbor A     │◀─── webhook ─────│   FastAPI       │
│ (source)     │                  │   任务 API       │
└──────┬───────┘                  └────────┬────────┘
       │                                   │
       │  skopeo / crane copy              │ enqueue
       ▼                                   ▼
┌──────────────┐                  ┌─────────────────┐
│  Sync Worker │─────────────────▶│  Task Store     │
│  子进程执行    │   更新状态        │  SQLite/Redis   │
└──────┬───────┘                  └─────────────────┘
       │
       ▼
┌──────────────┐
│ Harbor B     │
│ (destination)│
└──────────────┘

组件	说明
Registry 适配	Harbor API 查 tag；认证用 Robot Account
同步引擎	优先 skopeo copy（无需本地 docker daemon）
任务模型	pending → running → success / failed，支持重试
触发方式	API 手动、Harbor Webhook、定时 Cron

---

模块划分

image-sync-platform/
├── config/          # registries.yaml, sync_rules.yaml
├── registry/        # harbor_client, auth
├── sync/            # skopeo_runner, planner, worker
├── models/task.py
├── api/             # routes_sync, routes_webhook
└── scheduler/cron_sync.py

---

核心代码骨架

sync_rules.yaml

rules:
  - name: app-frontend-prod
    source:
      registry: harbor-a.internal
      project: frontend
      repo: web-ui
    destination:
      registry: harbor-b.internal
      project: frontend
      repo: web-ui
    tag_pattern: "^(main-|v\\d+\\.\\d+\\.\\d+)$"  # 只同步 main-* 与 semver
    enabled: true

harbor_client.py

"""Harbor v2 API：列出仓库 tag（简化示例）。"""
import httpx
import os

class HarborClient:
    def __init__(self, base_url: str, project: str, user_env: str, pass_env: str):
        self.base_url = base_url.rstrip("/")
        self.project = project
        self.auth = (os.environ[user_env], os.environ[pass_env])

    def list_tags(self, repo: str) -> list[str]:
        url = f"{self.base_url}/api/v2.0/projects/{self.project}/repositories/{repo}/artifacts"
        tags = []
        with httpx.Client(auth=self.auth, timeout=30) as client:
            r = client.get(url, params={"page_size": 100})
            r.raise_for_status()
            for art in r.json():
                for t in art.get("tags") or []:
                    tags.append(t["name"])
        return tags

skopeo_runner.py

def copy_image(src_ref: str, dst_ref: str, src_creds: str, dst_creds: str, timeout: int = 600) -> None:
    cmd = ["skopeo", "copy", "--src-creds", src_creds, "--dest-creds", dst_creds,
           f"docker://{src_ref}", f"docker://{dst_ref}"]
    r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if r.returncode != 0:
        raise RuntimeError(f"skopeo copy 失败: {r.stderr[-2000:]}")

planner.py + worker.py

import re
import uuid
from sync.skopeo_runner import copy_image
from models.task import SyncTask, TaskStatus

def plan_tasks(rule: dict, tags: list[str]) -> list[SyncTask]:
    pat = re.compile(rule["tag_pattern"])
    tasks = []
    for tag in tags:
        if not pat.match(tag):
            continue
        tasks.append(SyncTask(
            id=str(uuid.uuid4()),
            rule_name=rule["name"],
            tag=tag,
            status=TaskStatus.pending,
        ))
    return tasks

def execute_task(task: SyncTask, rule: dict) -> None:
    task.status = TaskStatus.running
    src = f"{rule['source']['registry']}/{rule['source']['project']}/{rule['source']['repo']}:{task.tag}"
    dst = f"{rule['destination']['registry']}/{rule['destination']['project']}/{rule['destination']['repo']}:{task.tag}"
    copy_image(src, dst, src_creds="...", dst_creds="...")
    task.status = TaskStatus.success

api/routes（Webhook + 任务查询）

@app.post("/webhook/harbor")
async def harbor_push(request: Request, background_tasks: BackgroundTasks):
    body = await request.json()
    if body.get("type") != "PUSH_ARTIFACT":
        return {"ignored": True}
    repo = body["event_data"]["repository"]["name"]
    tag = body["event_data"]["resources"][0]["tag"]
    background_tasks.add_task(enqueue_sync_for_tag, repo, tag)
    return {"accepted": True, "repo": repo, "tag": tag}

@app.post("/api/v1/sync/rules/{rule_name}/run")
def run_rule(rule_name: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(run_rule_async, rule_name)
    return {"message": f"规则 {rule_name} 已提交"}

@app.get("/api/v1/sync/tasks/{task_id}")
def get_task(task_id: str):
    return task_repository.get(task_id)

---

扩展点

扩展	做法
增量 vs 全量	Webhook 做增量；Cron 夜间 plan_tasks 补漏
多架构 manifest	skopeo --all 复制 multi-arch index
签名与合规	同步后 Notation/Cosign verify；失败则标记 blocked
带宽控制	worker 限并发 + 分时段；大镜像单独队列
UI / 替代	第一期可导出 CSV；复杂场景也可用 Harbor 原生复制，本平台侧重自定义过滤与审计

注意

Worker 需预装 skopeo；Robot Account 仅授予源 pull、目标 push。凭证走环境变量或 Vault，日志勿打印 token。