大规模镜像仓库治理项目的全流程复盘：Harbor中200TB存量镜像的清理、压缩与生命周期管理

大规模镜像仓库治理项目的全流程复盘：Harbor中200TB存量镜像的清理、压缩与生命周期管理

一、项目背景与问题提出

随着企业容器化进程的快速推进，Harbor镜像仓库逐渐成为DevOps流程中的核心基础设施。然而，在缺乏有效治理策略的情况下，镜像仓库的容量和性能问题日益凸显。在本次治理项目启动时，我们面对的是一个存储量达到200TB、包含超过15万个镜像仓库、累计镜像层数超过1200万的超大规模Harbor集群。

该项目源于一次生产事故：2025年6月，由于Harbor后端存储（S3兼容对象存储）的容量达到98%，导致所有新的镜像推送操作失败，直接影响当天37个微服务的版本发布。此次事故造成了约4小时的发布窗口丢失，涉及电商平台的核心交易服务更新推迟，预估影响交易额约280万元。

在深入分析问题根因后，我们识别出以下核心挑战：

1. 存储浪费严重：通过初步采样分析，发现约42%的镜像层是冗余的（相同digest的多份拷贝），约28%的镜像是超过180天未被拉取的历史版本
2. 清理风险高：缺乏准确的镜像依赖关系图谱，误删正在被使用的镜像会导致生产事故
3. 性能瓶颈显现：Harbor的API响应时间在峰值时段超过15秒，镜像清单查询（manifest list）的P99延迟达到8.7秒
4. 合规压力增大：安全团队要求所有镜像必须满足"最小必要原则"，即删除所有包含已知漏洞的中间构建产物

为系统性解决这些问题，我们启动了为期3个月的镜像仓库治理专项项目，目标是在保证零生产事故的前提下，将存储占用降低50%以上，并将API平均响应时间降低至3秒以内。

二、现状调研与技术分析

2.1 镜像存储结构分析

Harbor底层使用Docker Registry V2协议，镜像存储采用内容寻址（content-addressable storage）机制。每个镜像由以下部分组成：

• Manifest文件：记录镜像的配置和层索引信息（JSON格式）
• Config文件：记录镜像的详细配置（环境变量、入口命令等）
• Layer文件：实际的文件系统层（.tar.gz或.gz格式）

通过分析Harbor的存储后端，我们发现存储占用的主要分布为：

2.2 镜像依赖关系图谱构建

安全清理镜像的前提是准确理解镜像之间的引用关系。我们开发了基于图数据库的镜像依赖分析工具，用于构建完整的引用链。

依赖关系主要包含以下几个维度：

1. 直接引用：Kubernetes Deployment/StatefulSet等资源中显式指定的镜像Tag
2. 间接引用：Helm Chart中通过Values.yaml参数化的镜像引用
3. 构建依赖：Dockerfile中FROM指令引用的基础镜像
4. 多阶段构建产物：构建阶段产生但最终镜像不需要的中间层

以下是镜像依赖关系分析的Mermaid图：

2.3 性能瓶颈定位

通过Harbor的性能剖析（profile），我们定位到以下性能瓶颈点：

1. 数据库查询效率低：PostgreSQL中的artifact表已积累超过200万条记录，且缺乏有效的索引覆盖
2. 存储后端请求放大：每次API调用会触发多次S3的HEAD请求（用于检查blob是否存在），在高并发场景下造成大量延迟
3. 标签列表查询未优化：GET /v2/<project>/<repo>/tags/list接口在仓库包含大量标签时（某些仓库标签数超过5000），响应时间呈线性增长

三、治理方案设计与实施

3.1 整体技术架构

我们设计了一套分阶段的镜像仓库治理方案，包含以下核心组件：

1. 镜像元数据扫描器：周期性扫描Harbor中所有项目的元数据，构建镜像关系图谱
2. 智能保留策略引擎：基于多维度规则（时间、引用关系、漏洞状态、业务标签）计算镜像的保留权重
3. 安全删除执行器：采用"标记-验证-删除"三步策略，确保删除操作的安全性
4. 存储压缩优化器：对保留的镜像进行层去重和压缩算法优化
5. 生命周期管理器：基于Webhook的自动化生命周期管理，防止问题复发

3.2 镜像元数据扫描与依赖分析实现

以下是镜像元数据扫描核心代码的实现：

# -*- coding: utf-8 -*- """ Harbor镜像元数据扫描与依赖分析模块 负责扫描Harbor中所有镜像的元数据，构建镜像依赖关系图谱 """ import json import logging import time from typing import Dict, List, Set, Optional, Tuple from dataclasses import dataclass, field from datetime import datetime, timedelta import requests from neo4j import GraphDatabase logger = logging.getLogger(__name__) @dataclass class ImageManifest: """镜像Manifest数据结构""" digest: str # 镜像摘要（sha256:...） media_type: str # Manifest媒体类型 config_digest: str # 配置层摘要 layer_digests: List[str] # 层摘要列表 size: int # 总大小（字节） push_time: datetime # 推送时间 project_name: str # 所属项目 repository_name: str # 所属仓库 tags: List[str] # 关联的标签列表 @dataclass class ImageReference: """镜像引用关系""" source_type: str # 引用源类型（k8s, helm, dockerfile） source_id: str # 引用源标识 target_image: str # 目标镜像（repo:tag格式） target_digest: Optional[str] # 目标镜像摘要 last_access_time: datetime # 最后访问时间 reference_count: int = 1 # 引用次数 class HarborMetadataScanner: """ Harbor元数据扫描器 负责从Harbor API获取镜像元数据，并构建依赖关系图谱 """ def __init__(self, harbor_url: str, username: str, password: str, neo4j_uri: str, neo4j_user: str, neo4j_password: str): """ 初始化扫描器 Args: harbor_url: Harbor服务地址 username: Harbor用户名 password: Harbor密码 neo4j_uri: Neo4j数据库连接URI neo4j_user: Neo4j用户名 neo4j_password: Neo4j密码 """ self.harbor_url = harbor_url.rstrip('/') self.session = requests.Session() self.session.auth = (username, password) self.session.headers.update({ "Accept": "application/json", "Content-Type": "application/json" }) self.session.timeout = 30 # 初始化Neo4j图数据库连接 self.neo4j_driver = GraphDatabase.driver( neo4j_uri, auth=(neo4j_user, neo4j_password) ) # 扫描统计信息 self.stats = { "projects_scanned": 0, "repositories_scanned": 0, "artifacts_scanned": 0, "references_found": 0, "start_time": None, "end_time": None } def scan_all_projects(self) -> Dict: """ 扫描所有Harbor项目 Returns: 扫描统计信息字典 """ self.stats["start_time"] = datetime.now() logger.info("开始扫描Harbor元数据...") try: # 获取所有项目列表 projects = self._get_all_projects() logger.info(f"发现{len(projects)}个项目，开始逐个扫描") for project in projects: project_name = project["name"] logger.info(f"正在扫描项目: {project_name}") self._scan_project(project_name) self.stats["projects_scanned"] += 1 # 每扫描完成一个项目，打印进度信息 logger.info(f"进度: {self.stats['projects_scanned']}/{len(projects)} " f"项目, {self.stats['repositories_scanned']} 仓库, " f"{self.stats['artifacts_scanned']} 镜像") self.stats["end_time"] = datetime.now() duration = (self.stats["end_time"] - self.stats["start_time"]).total_seconds() logger.info(f"扫描完成! 总耗时: {duration:.1f}秒") logger.info(f"统计信息: {json.dumps(self.stats, indent=2, default=str)}") return self.stats except Exception as e: logger.error(f"扫描过程中发生错误: {e}", exc_info=True) raise def _get_all_projects(self) -> List[Dict]: """ 获取Harbor中的所有项目（处理分页） Returns: 项目信息列表 """ projects = [] page = 1 page_size = 100 while True: try: response = self.session.get( f"{self.harbor_url}/api/v2.0/projects", params={"page": page, "page_size": page_size} ) response.raise_for_status() batch = response.json() if not batch: break projects.extend(batch) # 检查是否还有更多数据 if len(batch) < page_size: break page += 1 except requests.exceptions.RequestException as req_err: logger.error(f"获取项目列表失败: {req_err}") raise return projects def _scan_project(self, project_name: str): """ 扫描指定项目下的所有仓库和镜像 Args: project_name: 项目名称 """ repositories = self._get_project_repositories(project_name) for repo in repositories: repo_name = repo["name"] logger.debug(f"扫描仓库: {repo_name}") self._scan_repository(project_name, repo_name) self.stats["repositories_scanned"] += 1 def _get_project_repositories(self, project_name: str) -> List[Dict]: """ 获取项目下的所有镜像仓库 Args: project_name: 项目名称 Returns: 仓库信息列表 """ repositories = [] page = 1 page_size = 100 while True: response = self.session.get( f"{self.harbor_url}/api/v2.0/projects/{project_name}/repositories", params={"page": page, "page_size": page_size} ) response.raise_for_status() batch = response.json() if not batch: break repositories.extend(batch) if len(batch) < page_size: break page += 1 return repositories def _scan_repository(self, project_name: str, repository_name: str): """ 扫描指定仓库中的所有镜像制品 Args: project_name: 项目名称 repository_name: 仓库名称 """ artifacts = self._get_repository_artifacts(project_name, repository_name) for artifact in artifacts: try: manifest = self._parse_artifact_to_manifest(project_name, repository_name, artifact) self._store_manifest_in_graph(manifest) self.stats["artifacts_scanned"] += 1 except Exception as e: logger.warning(f"解析镜像失败: {repository_name}:{artifact.get('digest', 'unknown')}, " f"错误: {e}") continue def _get_repository_artifacts(self, project_name: str, repository_name: str) -> List[Dict]: """获取仓库中的所有制品（镜像）""" artifacts = [] page = 1 page_size = 100 while True: response = self.session.get( f"{self.harbor_url}/api/v2.0/projects/{project_name}/repositories/" f"{requests.utils.quote(repository_name, safe='')}/artifacts", params={"page": page, "page_size": page_size, "with_tag": True} ) response.raise_for_status() batch = response.json() if not batch: break artifacts.extend(batch) if len(batch) < page_size: break page += 1 return artifacts def _parse_artifact_to_manifest(self, project_name: str, repository_name: str, artifact: Dict) -> ImageManifest: """ 将Harbor API返回的artifact数据解析为统一的ImageManifest结构 Args: project_name: 项目名称 repository_name: 仓库名称 artifact: Harbor API返回的artifact数据 Returns: ImageManifest对象 """ # 提取层摘要列表 layer_digests = [] if "manifest_media_type" in artifact and "application/vnd.docker.distribution.manifest.v2+json" in artifact["manifest_media_type"]: # V2 Manifest格式 for layer in artifact.get("layers", []): layer_digests.append(layer.get("digest", "")) elif "manifest_media_type" in artifact and "application/vnd.oci.image.manifest.v1+json" in artifact["manifest_media_type"]: # OCI Manifest格式 for layer in artifact.get("layers", []): layer_digests.append(layer.get("digest", "")) # 提取标签列表 tags = [tag["name"] for tag in artifact.get("tags", [])] # 解析推送时间 push_time_str = artifact.get("push_time", "") push_time = datetime.now() if push_time_str: try: push_time = datetime.fromisoformat(push_time_str.replace("Z", "+00:00")) except ValueError: logger.warning(f"推送时间解析失败: {push_time_str}") return ImageManifest( digest=artifact.get("digest", ""), media_type=artifact.get("manifest_media_type", ""), config_digest=artifact.get("config", {}).get("digest", ""), layer_digests=layer_digests, size=artifact.get("size", 0), push_time=push_time, project_name=project_name, repository_name=repository_name, tags=tags ) def _store_manifest_in_graph(self, manifest: ImageManifest): """ 将镜像Manifest信息存储到Neo4j图数据库 Args: manifest: ImageManifest对象 """ with self.neo4j_driver.session() as session: # 创建或更新镜像节点 cypher_query = """ MERGE (img:Image {digest: $digest}) SET img.media_type = $media_type, img.config_digest = $config_digest, img.size = $size, img.push_time = $push_time, img.project_name = $project_name, img.repository_name = $repository_name, img.tags = $tags, img.last_updated = timestamp() """ session.run(cypher_query, digest=manifest.digest, media_type=manifest.media_type, config_digest=manifest.config_digest, size=manifest.size, push_time=manifest.push_time.isoformat(), project_name=manifest.project_name, repository_name=manifest.repository_name, tags=manifest.tags) # 创建镜像与层之间的"HAS_LAYER"关系 for layer_digest in manifest.layer_digests: session.run(""" MATCH (img:Image {digest: $img_digest}) MERGE (layer:Layer {digest: $layer_digest}) MERGE (img)-[:HAS_LAYER]->(layer) """, img_digest=manifest.digest, layer_digest=layer_digest) def analyze_kubernetes_references(self, kubeconfig_path: str): """ 分析Kubernetes集群中的镜像引用关系 Args: kubeconfig_path: Kubeconfig文件路径 """ logger.info("开始分析Kubernetes集群中的镜像引用...") try: from kubernetes import client, config # 加载Kubernetes配置 config.load_kube_config(config_file=kubeconfig_path) v1 = client.CoreV1Api() apps_v1 = client.AppsV1Api() # 获取所有Namespace namespaces = v1.list_namespace() for ns in namespaces.items: namespace_name = ns.metadata.name logger.debug(f"分析Namespace: {namespace_name}") # 分析Deployment deployments = apps_v1.list_namespaced_deployment(namespace_name) for deploy in deployments.items: self._extract_deployment_images(deploy, namespace_name) # 分析StatefulSet statefulsets = apps_v1.list_namespaced_stateful_set(namespace_name) for sts in statefulsets.items: self._extract_statefulset_images(sts, namespace_name) # 分析DaemonSet daemonsets = apps_v1.list_namespaced_daemon_set(namespace_name) for ds in daemonsets.items: self._extract_daemonset_images(ds, namespace_name) logger.info(f"Kubernetes引用分析完成，发现{self.stats['references_found']}个引用关系") except Exception as e: logger.error(f"Kubernetes引用分析失败: {e}", exc_info=True) raise def _extract_deployment_images(self, deployment, namespace: str): """从Deployment中提取镜像引用""" deploy_name = deployment.metadata.name for container in deployment.spec.template.spec.containers: image_ref = container.image self._store_image_reference( source_type="kubernetes_deployment", source_id=f"{namespace}/{deploy_name}", target_image=image_ref, namespace=namespace ) def _extract_statefulset_images(self, statefulset, namespace: str): """从StatefulSet中提取镜像引用""" sts_name = statefulset.metadata.name for container in statefulset.spec.template.spec.containers: image_ref = container.image self._store_image_reference( source_type="kubernetes_statefulset", source_id=f"{namespace}/{sts_name}", target_image=image_ref, namespace=namespace ) def _extract_daemonset_images(self, daemonset, namespace: str): """从DaemonSet中提取镜像引用""" ds_name = daemonset.metadata.name for container in daemonset.spec.template.spec.containers: image_ref = container.image self._store_image_reference( source_type="kubernetes_daemonset", source_id=f"{namespace}/{ds_name}", target_image=image_ref, namespace=namespace ) def _store_image_reference(self, source_type: str, source_id: str, target_image: str, namespace: str): """ 将镜像引用关系存储到Neo4j图数据库 Args: source_type: 引用源类型 source_id: 引用源标识 target_image: 目标镜像引用 namespace: Kubernetes命名空间 """ with self.neo4j_driver.session() as session: cypher_query = """ MATCH (img:Image) WHERE $target_image IN [img.repository_name + ':' + tag | tag IN img.tags] OR img.digest = $target_image MERGE (src:ReferenceSource {id: $source_id, type: $source_type}) SET src.namespace = $namespace, src.last_seen = timestamp() MERGE (src)-[:REFERENCES {last_seen: timestamp()}]->(img) """ try: session.run(cypher_query, target_image=target_image, source_id=source_id, source_type=source_type, namespace=namespace) self.stats["references_found"] += 1 except Exception as e: logger.warning(f"存储引用关系失败: {source_id} -> {target_image}, 错误: {e}") def close(self): """关闭所有连接""" if self.neo4j_driver: self.neo4j_driver.close() if self.session: self.session.close() # 主执行流程 def main(): """主函数：执行完整的元数据扫描流程""" # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/harbor-scanner/metadata-scan.log'), logging.StreamHandler() ] ) # 初始化扫描器（配置应从环境变量或配置文件读取） scanner = HarborMetadataScanner( harbor_url="https://harbor.example.com", username="admin", password="your-harbor-password", neo4j_uri="bolt://neo4j:7687", neo4j_user="neo4j", neo4j_password="your-neo4j-password" ) try: # 步骤1：扫描Harbor元数据 logger.info("=" * 60) logger.info("步骤1: 扫描Harbor元数据") logger.info("=" * 60) scan_stats = scanner.scan_all_projects() # 步骤2：分析Kubernetes引用 logger.info("=" * 60) logger.info("步骤2: 分析Kubernetes集群引用") logger.info("=" * 60) scanner.analyze_kubernetes_references(kubeconfig_path="/root/.kube/config") # 步骤3：生成清理建议报告 logger.info("=" * 60) logger.info("步骤3: 生成清理建议报告") logger.info("=" * 60) # TODO: 实现清理建议生成逻辑 logger.info("元数据扫描和依赖分析完成!") except Exception as e: logger.critical(f"扫描流程异常终止: {e}", exc_info=True) raise finally: scanner.close() if __name__ == "__main__": main()

3.3 智能保留策略引擎

保留策略引擎基于多维度规则计算镜像的保留权重。核心规则包括：

1. 时间维度：最近30天内被拉取过的镜像自动保留；30-180天未被拉取的镜像进入待审核状态；超过180天未被拉取的镜像标记为可删除
2. 引用关系维度：仍被Kubernetes资源引用的镜像强制保留；被Helm Chart引用的镜像强制保留
3. 业务标签维度：带有retain=true标签的镜像强制保留；带有auto-cleanup=true标签的镜像优先清理
4. 漏洞状态维度：包含严重漏洞（CVSS评分≥9.0）且无可修复版本的镜像，在通知安全团队后删除

为实现灵活的策略配置，我们设计了基于DSL（领域特定语言）的策略定义语法：

# 镜像保留策略配置示例 retention_policies: - name: "生产环境关键服务保护策略" priority: 100 conditions: - field: "tags" operator: "contains_any" values: ["prod", "stable", "latest"] - field: "project_name" operator: "in" values: ["production", "core-services"] action: "retain" retention_days: 365 - name: "开发环境临时镜像清理策略" priority: 50 conditions: - field: "project_name" operator: "in" values: ["dev", "test", "staging"] - field: "last_pull_time" operator: "older_than" value: "7d" action: "delete" grace_period_days: 3 # 宽限期：标记后3天才执行删除 - name: "构建缓存镜像清理策略" priority: 80 conditions: - field: "tags" operator: "contains_any" values: ["build-cache", "ci-temp"] - field: "push_time" operator: "older_than" value: "1d" action: "delete" grace_period_days: 0 # 立即删除

3.4 安全删除执行器

安全删除采用"标记-验证-删除"三步策略：

阶段1：标记（Mark）

• 基于保留策略引擎的输出，生成待删除镜像清单
• 对待删除镜像添加deletion-candidate=true标签
• 发送删除预告通知给相关团队（通过企业微信/邮件）

阶段2：验证（Verify）

• 在宽限期内，持续监控待删除镜像的引用状态变化
• 如果镜像在宽限期内被重新引用，自动取消删除标记
• 执行删除前的最后确认：再次检查Kubernetes集群中的引用状态

阶段3：删除（Delete）

• 调用Harbor API执行软删除（将镜像标记为已删除，但实际数据保留）
• 等待7天后，如果无异常反馈，执行硬删除（从存储后端物理删除）
• 记录删除操作的完整审计日志

四、实施过程与关键里程碑

4.1 分阶段实施计划

为降低风险，我们将治理项目分为4个阶段执行：

阶段1：分析与规划（第1-2周）

• 完成镜像元数据的全量扫描
• 构建镜像依赖关系图谱
• 制定保留策略规则
• 输出《镜像仓库现状分析报告》

阶段2：小规模试点（第3-4周）

• 选择2个非核心项目的镜像仓库进行试点清理
• 验证保留策略的准确性
• 优化删除执行流程
• 输出《试点清理总结报告》

阶段3：大规模清理（第5-8周）

• 分批次执行全量镜像清理
• 每批次清理后观察48小时，确认无异常
• 同步进行性能优化（数据库索引、API缓存等）
• 输出每周清理进度报告

阶段4：生命周期管理体系建设（第9-12周）

• 部署自动化的镜像生命周期管理Webhook
• 建立定期清理的CronJob任务
• 完善监控告警体系
• 输出《镜像仓库运维手册》

4.2 关键技术难点与解决方案

难点1：大规模删除操作导致的性能抖动

问题描述：在执行批量删除操作时，Harbor的API响应时间出现剧烈抖动（P99延迟从3秒飙升至45秒），影响正常的CI/CD流程。

根因分析：删除操作会触发PostgreSQL中的级联删除（cascade delete），当单次删除包含大量tag的镜像时，数据库事务执行时间超过30秒，导致连接池耗尽。

解决方案：

1. 将批量删除操作改为分批串行执行，每批次删除不超过50个tag
2. 在PostgreSQL中增加artifact表和tag表的复合索引，加速删除查询
3. 引入删除操作的速率限制（rate limiting），将删除API的QPS限制在10以内

难点2：跨项目镜像共享导致的误删风险

问题描述：多个项目可能引用同一个基础镜像（如base-images/java:11），如果仅基于项目维度判断引用关系，可能误删其他项目仍在使用的镜像。

解决方案：

1. 在依赖关系图谱中，将镜像层的共享关系显式建模（SHARED_BY关系类型）
2. 删除决策时，不仅检查直接引用，还递归检查层共享图中的依赖节点
3. 对于被多个项目共享的镜像，只有当所有引用方都同意删除时才执行删除

难点3：GC（垃圾回收）执行时间窗口难以把控

问题描述：Harbor的GC操作需要停止新镜像的推送（设置只读模式），而在业务高峰期难以找到合适的执行窗口。

解决方案：

1. 将Harbor升级至v2.5+版本，利用该版本引入的"在线GC"特性（不需要设置只读模式）
2. 对于必须使用传统GC模式的场景，采用"滚动GC"策略：在多个Harbor实例之间滚动执行GC，确保服务不中断
3. 将GC操作自动化，设置在每周日凌晨2:00自动执行

4.3 清理效果数据

经过3个月的治理，我们取得了显著的成效：

此外，我们还建立了如下的长期运行机制：

• 每周自动清理超过90天未被拉取的镜像（预计每周释放约2TB空间）
• 每日检查镜像漏洞状态，自动标记包含严重漏洞的镜像
• 每月生成镜像仓库健康度报告，发送给相关团队

五、总结

本次Harbor镜像仓库治理项目是一次大规模的基础设施优化实践，涵盖了元数据管理、依赖关系分析、智能决策引擎和自动化执行等多个技术层面。通过系统性的治理，我们不仅解决了迫在眉睫的存储和性能问题，还建立了可持续的镜像生命周期管理体系。

核心经验总结如下：

技术架构层面：

1. 图数据库是依赖关系分析的最佳选择：传统的关系型数据库难以高效表达镜像之间复杂的引用和共享关系，Neo4j的图遍历能力极大提升了分析效率
2. 分批渐进式清理是降低风险的关键：面对200TB的海量数据，必须采用"小步快跑"的策略，通过试点验证、分批执行来控制系统性风险
3. 自动化生命周期管理防止问题复发：一次性清理只能解决存量问题，必须建立自动化的增量管理机制才能确保长期效果

工程实践层面：

1. 保留策略需要业务深度参与：纯技术指标（如最后拉取时间）不足以做出准确的删除决策，必须结合业务场景（如合规性要求、审计需求）来制定策略
2. 删除操作必须可回溯：所有删除操作都要保留完整的审计日志，并支持快速恢复（我们设计了"软删除+延迟硬删除"的双重保护机制）
3. 性能优化要系统性思考：API响应慢不仅是应用层的问题，还涉及数据库索引设计、存储后端请求模式、缓存策略等多个层面，必须系统性优化

团队协作层面：

1. 跨团队沟通至关重要：镜像治理涉及安全团队（漏洞管理）、开发团队（镜像构建规范）、SRE团队（基础设施维护）等多个角色，必须建立高效的沟通机制
2. 文档和培训要同步推进：新的生命周期管理流程需要所有相关人员理解并遵守，我们编写的《镜像仓库运维手册》和组织的3场培训起到了关键作用

未来优化方向包括：引入基于机器学习的镜像重要性预测模型（基于历史拉取模式、构建频率等特征），进一步智能化的保留决策；探索镜像层的跨仓库全局去重机制，在分布式Harbor实例之间共享存储；研究基于eStargz等延迟拉取技术的镜像分发优化，降低对镜像仓库的读取压力。

镜像仓库治理不是一次性的项目，而是需要持续投入和优化的长期工作。只有通过技术手段和管理流程的双重保障，才能确保容器化基础设施的健康和可持续发展。