当前位置：首页 > news >正文

python自动测试 crictl 可以从哪些国内镜像源成功拉取镜像

news 2025/8/27 11:19:36

自动测试多个国内主流镜像源，并报告哪些可用。

Python 检测脚本

#!/usr/bin/env python3
"""
crictl_mirror_tester.py
功能：自动测试 crictl 可以从哪些国内镜像源成功拉取镜像。
使用方法：节点上运行 python3 crictl_mirror_tester.py
"""import subprocess
import time
import json
from concurrent.futures import ThreadPoolExecutor, as_completedclass CrictlMirrorTester:def __init__(self):self.test_image = "nginx:alpine"  # 用于测试的镜像self.timeout = 120  # 单个镜像拉取超时时间（秒）self.results = []# 国内主流镜像源列表self.mirrors = [{"name": "阿里云","url": "registry.cn-hangzhou.aliyuncs.com/library/nginx:alpine","status": "待测试","time": 0,"output": ""},{"name": "华为云","url": "swr.cn-east-2.myhuaweicloud.com/library/nginx:alpine", "status": "待测试","time": 0,"output": ""},{"name": "腾讯云","url": "ccr.ccs.tencentyun.com/library/nginx:alpine","status": "待测试", "time": 0,"output": ""},{"name": "网易云","url": "hub-mirror.c.163.com/library/nginx:latest","status": "待测试","time": 0,"output": ""},{"name": "Docker官方源(直接)","url": "nginx:alpine","status": "待测试","time": 0,"output": ""},{"name": "中科大镜像","url": "docker.mirrors.ustc.edu.cn/library/nginx:alpine","status": "待测试","time": 0,"output": ""}]def test_single_mirror(self, mirror):"""测试单个镜像源"""print(f"正在测试 {mirror['name']} ({mirror['url']})...")start_time = time.time()try:# 执行 crictl pull 命令cmd = ["sudo", "crictl", "pull", mirror["url"]]result = subprocess.run(cmd, capture_output=True, text=True, timeout=self.timeout)end_time = time.time()mirror["time"] = round(end_time - start_time, 2)mirror["output"] = result.stderr if result.stderr else result.stdoutif result.returncode == 0:mirror["status"] = "✅ 成功"print(f"  {mirror['name']}: ✅ 成功 ({mirror['time']}s)")else:mirror["status"] = "❌ 失败"print(f"  {mirror['name']}: ❌ 失败")except subprocess.TimeoutExpired:end_time = time.time()mirror["time"] = round(end_time - start_time, 2)mirror["status"] = "⏰ 超时"mirror["output"] = f"拉取操作超时（{self.timeout}秒）"print(f"  {mirror['name']}: ⏰ 超时")except Exception as e:end_time = time.time()mirror["time"] = round(end_time - start_time, 2)mirror["status"] = "⚠️ 异常"mirror["output"] = str(e)print(f"  {mirror['name']}: ⚠️ 异常")return mirrordef run_tests(self):"""运行所有测试"""print("=" * 60)print("开始测试 crictl 镜像源可用性")print("=" * 60)print(f"测试镜像: {self.test_image}")print(f"超时时间: {self.timeout}秒")print("=" * 60)# 使用线程池并行测试（最多3个并发）with ThreadPoolExecutor(max_workers=3) as executor:future_to_mirror = {executor.submit(self.test_single_mirror, mirror): mirror for mirror in self.mirrors}for future in as_completed(future_to_mirror):try:result = future.result()self.results.append(result)except Exception as e:print(f"测试过程中发生错误: {e}")def generate_report(self):"""生成测试报告"""print("\n" + "=" * 60)print("测试报告")print("=" * 60)# 按成功率排序：成功 -> 失败 -> 超时 -> 异常success_first = sorted(self.results, key=lambda x: (0 if "成功" in x["status"] else 1 if "失败" in x["status"] else2 if "超时" in x["status"] else 3))for mirror in success_first:color = "32m" if "成功" in mirror["status"] else "31m"print(f"{mirror['status']} - {mirror['name']} - {mirror['time']}s")print(f"  镜像地址: {mirror['url']}")if mirror['output'] and "成功" not in mirror['status']:print(f"  错误信息: {mirror['output'][:100]}...")print()# 统计结果successful = [m for m in self.results if "成功" in m["status"]]failed = [m for m in self.results if "失败" in m["status"]]timeout = [m for m in self.results if "超时" in m["status"]]error = [m for m in self.results if "异常" in m["status"]]print("=" * 60)print(f"测试完成！")print(f"✅ 成功的源: {len(successful)} 个")print(f"❌ 失败的源: {len(failed)} 个") print(f"⏰ 超时的源: {len(timeout)} 个")print(f"⚠️ 异常的源: {len(error)} 个")print("=" * 60)# 推荐最佳镜像源if successful:# 选择耗时最短的成功源best_mirror = min(successful, key=lambda x: x["time"])print(f"🎯 推荐使用: {best_mirror['name']}")print(f"   镜像地址: {best_mirror['url']}")print(f"   拉取时间: {best_mirror['time']} 秒")else:print("❌ 没有可用的镜像源，请检查网络连接或节点配置")print("=" * 60)return self.resultsdef cleanup(self):"""清理测试生成的镜像"""print("\n正在清理测试镜像...")for mirror in self.results:if "成功" in mirror["status"]:try:cmd = ["sudo", "crictl", "rmi", mirror["url"]]subprocess.run(cmd, capture_output=True, timeout=30)except:passprint("清理完成！")def main():"""主函数"""try:tester = CrictlMirrorTester()tester.run_tests()results = tester.generate_report()tester.cleanup()# 返回退出码：如果有成功的源返回0，否则返回1successful = any("成功" in m["status"] for m in results)exit(0 if successful else 1)except KeyboardInterrupt:print("\n用户中断测试")exit(1)except Exception as e:print(f"程序执行错误: {e}")exit(1)if __name__ == "__main__":main()

使用方法

将脚本保存到节点上：
```
vim crictl_mirror_tester.py
```
复制上面的代码并保存。
给予执行权限：
```
chmod +x crictl_mirror_tester.py
```

运行测试：

python3 crictl_mirror_tester.py

或者

./crictl_mirror_tester.py

脚本特点

并发测试：使用线程池并行测试多个镜像源，大大缩短测试时间
超时控制：单个镜像拉取超时自动终止，避免长时间等待
详细报告：彩色输出，清晰显示每个源的测试结果和耗时
自动清理：测试完成后自动删除下载的镜像，节省磁盘空间
智能推荐：自动推荐拉取速度最快的可用镜像源
错误处理：完善的异常处理，确保程序稳定运行

预期输出示例

============================================================
开始测试 crictl 镜像源可用性
============================================================
测试镜像: nginx:alpine
超时时间: 120秒
============================================================
正在测试 阿里云 (registry.cn-hangzhou.aliyuncs.com/library/nginx:alpine)...阿里云: ✅ 成功 (12.34s)
正在测试 华为云 (swr.cn-east-2.myhuaweicloud.com/library/nginx:alpine)...华为云: ✅ 成功 (8.21s)
...============================================================
测试报告
============================================================
✅ 成功 - 华为云 - 8.21s镜像地址: swr.cn-east-2.myhuaweicloud.com/library/nginx:alpine✅ 成功 - 阿里云 - 12.34s  镜像地址: registry.cn-hangzhou.aliyuncs.com/library/nginx:alpine❌ 失败 - 腾讯云 - 15.67s镜像地址: ccr.ccs.tencentyun.com/library/nginx:alpine错误信息: failed to pull image: rpc error: code = Unknown desc = failed to pull...============================================================
测试完成！
✅ 成功的源: 2 个
❌ 失败的源: 1 个
⏰ 超时的源: 0 个
⚠️ 异常的源: 0 个
============================================================
🎯 推荐使用: 华为云镜像地址: swr.cn-east-2.myhuaweicloud.com/library/nginx:alpine  拉取时间: 8.21 秒
============================================================

脚本可以帮你快速找出哪个国内镜像源最快最稳定！

查看全文

http://www.dtcms.com/a/352289.html