Docker 工具类/通用函数封装实践

## 一、为什么需要封装 Docker 工具类？

在实际项目中，我们经常会遇到以下场景：

- 需要批量启动/停止多个容器
- 需要检查容器健康状态
- 需要清理无用的镜像和容器
- 需要获取容器日志或执行命令
- 需要构建镜像并推送到仓库

如果每次都写一遍完整的 Docker 命令，不仅代码冗余，而且容易出错。通过封装通用函数，我们可以：

1. **提高代码复用性**：一次封装，多处使用
2. **统一错误处理**：集中处理异常情况
3. **简化调用方式**：提供简洁的 API 接口
4. **便于测试和维护**：逻辑集中，易于调试

## 二、技术选型

根据不同的开发语言，我们可以选择不同的 Docker SDK：

| 语言 | 推荐库 | 说明 |
|------|--------|------|
| Python | docker-py | 官方维护，功能完善 |
| Go | docker/docker/client | 官方 SDK，性能好 |
| Node.js | dockerode | 社区活跃，API 友好 |
| Java | docker-java | 功能全面，支持异步 |

本文以 **Python** 为例进行演示，因为 docker-py 使用简单且功能强大。

## 三、核心工具类实现

### 1. 基础连接管理

```python
import docker
from typing import Optional

class DockerClient:
    """Docker 客户端管理类"""
    
    _instance = None
    
    def __new__(cls, base_url: Optional[str] = None):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._init_client(base_url)
        return cls._instance
    
    def _init_client(self, base_url: Optional[str] = None):
        """初始化 Docker 客户端"""
        try:
            if base_url:
                self.client = docker.DockerClient(base_url=base_url)
            else:
                self.client = docker.from_env()
            self.client.ping()  # 验证连接
        except Exception as e:
            raise ConnectionError(f"无法连接 Docker: {e}")
    
    def get_client(self) -> docker.DockerClient:
        """获取客户端实例"""
        return self.client
```

### 2. 容器管理通用函数

```python
class ContainerManager:
    """容器管理工具类"""
    
    def __init__(self, client: docker.DockerClient):
        self.client = client
    
    def list_containers(self, all: bool = False, filters: dict = None) -> list:
        """列出容器
        
        Args:
            all: 是否显示所有容器（包括停止的）
            filters: 过滤条件，如 {'status': 'running'}
        
        Returns:
            容器列表
        """
        try:
            return self.client.containers.list(all=all, filters=filters)
        except docker.errors.APIError as e:
            raise RuntimeError(f"获取容器列表失败: {e}")
    
    def start_container(self, container_id: str) -> bool:
        """启动容器"""
        try:
            container = self.client.containers.get(container_id)
            container.start()
            return True
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"启动容器失败: {e}")
    
    def stop_container(self, container_id: str, timeout: int = 10) -> bool:
        """停止容器"""
        try:
            container = self.client.containers.get(container_id)
            container.stop(timeout=timeout)
            return True
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"停止容器失败: {e}")
    
    def restart_container(self, container_id: str, timeout: int = 10) -> bool:
        """重启容器"""
        try:
            container = self.client.containers.get(container_id)
            container.restart(timeout=timeout)
            return True
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"重启容器失败: {e}")
    
    def remove_container(self, container_id: str, force: bool = False) -> bool:
        """删除容器"""
        try:
            container = self.client.containers.get(container_id)
            container.remove(force=force)
            return True
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"删除容器失败: {e}")
```

### 3. 镜像管理通用函数

```python
class ImageManager:
    """镜像管理工具类"""
    
    def __init__(self, client: docker.DockerClient):
        self.client = client
    
    def pull_image(self, image_name: str, tag: str = "latest") -> dict:
        """拉取镜像"""
        try:
            image = self.client.images.pull(image_name, tag=tag)
            return {
                "id": image.id,
                "tags": image.tags,
                "size": image.attrs.get("Size", 0)
            }
        except docker.errors.ImageNotFound:
            raise ValueError(f"镜像 {image_name}:{tag} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"拉取镜像失败: {e}")
    
    def build_image(self, path: str, tag: str, dockerfile: str = "Dockerfile") -> str:
        """构建镜像"""
        try:
            image, logs = self.client.images.build(
                path=path,
                tag=tag,
                dockerfile=dockerfile,
                rm=True
            )
            return image.id
        except docker.errors.BuildError as e:
            raise RuntimeError(f"构建镜像失败: {e}")
    
    def remove_image(self, image_id: str, force: bool = False) -> bool:
        """删除镜像"""
        try:
            self.client.images.remove(image_id, force=force)
            return True
        except docker.errors.ImageNotFound:
            raise ValueError(f"镜像 {image_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"删除镜像失败: {e}")
    
    def list_images(self, filters: dict = None) -> list:
        """列出本地镜像"""
        try:
            return self.client.images.list(filters=filters)
        except docker.errors.APIError as e:
            raise RuntimeError(f"获取镜像列表失败: {e}")
```

### 4. 容器日志与执行命令

```python
class ContainerOps:
    """容器操作工具类"""
    
    def __init__(self, client: docker.DockerClient):
        self.client = client
    
    def get_logs(self, container_id: str, tail: int = 100, timestamps: bool = False) -> str:
        """获取容器日志"""
        try:
            container = self.client.containers.get(container_id)
            return container.logs(tail=tail, timestamps=timestamps).decode("utf-8")
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"获取日志失败: {e}")
    
    def exec_command(self, container_id: str, command: str) -> tuple:
        """在容器中执行命令
        
        Returns:
            (exit_code, output)
        """
        try:
            container = self.client.containers.get(container_id)
            exec_result = container.exec_run(command)
            return exec_result.exit_code, exec_result.output.decode("utf-8")
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"执行命令失败: {e}")
    
    def get_container_stats(self, container_id: str) -> dict:
        """获取容器资源使用情况"""
        try:
            container = self.client.containers.get(container_id)
            stats = container.stats(stream=False)
            return {
                "cpu_usage": stats["cpu_stats"]["cpu_usage"]["total_usage"],
                "memory_usage": stats["memory_stats"]["usage"],
                "memory_limit": stats["memory_stats"]["limit"],
                "network_rx": stats["networks"].get("eth0", {}).get("rx_bytes", 0),
                "network_tx": stats["networks"].get("eth0", {}).get("tx_bytes", 0)
            }
        except docker.errors.NotFound:
            raise ValueError(f"容器 {container_id} 不存在")
        except docker.errors.APIError as e:
            raise RuntimeError(f"获取容器状态失败: {e}")
```

### 5. 批量操作与清理工具

```python
class DockerUtils:
    """Docker 通用工具函数"""
    
    def __init__(self, client: docker.DockerClient):
        self.client = client
        self.container_mgr = ContainerManager(client)
        self.image_mgr = ImageManager(client)
        self.container_ops = ContainerOps(client)
    
    def cleanup_unused(self, include_volumes: bool = False) -> dict:
        """清理未使用的资源
        
        Returns:
            清理结果统计
        """
        result = {
            "containers_removed": 0,
            "images_removed": 0,
            "volumes_removed": 0
        }
        
        # 清理停止的容器
        stopped_containers = self.client.containers.list(
            all=True, 
            filters={"status": "exited"}
        )
        for container in stopped_containers:
            try:
                container.remove()
                result["containers_removed"] += 1
            except docker.errors.APIError:
                continue
        
        # 清理悬空镜像
        dangling_images = self.client.images.list(filters={"dangling": True})
        for image in dangling_images:
            try:
                self.client.images.remove(image.id)
                result["images_removed"] += 1
            except docker.errors.APIError:
                continue
        
        # 清理未使用的卷
        if include_volumes:
            volumes = self.client.volumes.list()
            for volume in volumes:
                if not volume.attrs.get("Containers"):
                    try:
                        volume.remove()
                        result["volumes_removed"] += 1
                    except docker.errors.APIError:
                        continue
        
        return result
    
    def get_system_info(self) -> dict:
        """获取 Docker 系统信息"""
        info = self.client.info()
        return {
            "containers": info["Containers"],
            "running_containers": info["ContainersRunning"],
            "images": info["Images"],
            "version": info["ServerVersion"],
            "os_type": info["OSType"],
            "architecture": info["Architecture"],
            "memory_total": info["MemTotal"],
            "cpu_count": info["NCPU"]
        }
    
    def wait_for_container_ready(self, container_id: str, timeout: int = 30) -> bool:
        """等待容器就绪"""
        import time
        
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:
                container = self.client.containers.get(container_id)
                if container.status == "running":
                    # 检查健康状态
                    health = container.attrs.get("State", {}).get("Health", {})
                    if not health or health.get("Status") == "healthy":
                        return True
            except docker.errors.NotFound:
                pass
            time.sleep(1)
        
        return False
```

## 四、使用示例

```python
# 初始化客户端
docker_client = DockerClient()
client = docker_client.get_client()

# 创建工具实例
utils = DockerUtils(client)

# 查看系统信息
info = utils.get_system_info()
print(f"Docker 版本: {info['version']}")
print(f"运行中容器: {info['running_containers']}/{info['containers']}")

# 列出所有运行中的容器
running_containers = utils.container_mgr.list_containers(
    filters={"status": "running"}
)
for container in running_containers:
    print(f"容器: {container.name} - {container.status}")

# 构建并启动容器
image_id = utils.image_mgr.build_image(
    path="./myapp",
    tag="myapp:latest"
)
print(f"镜像构建成功: {image_id}")

# 启动容器
container = client.containers.run(
    "myapp:latest",
    detach=True,
    ports={"8080/tcp": 8080}
)
print(f"容器启动成功: {container.id}")

# 等待容器就绪
if utils.wait_for_container_ready(container.id, timeout=60):
    print("容器已就绪")
    
    # 获取日志
    logs = utils.container_ops.get_logs(container.id, tail=50)
    print(f"最近日志:\n{logs}")
    
    # 执行命令
    exit_code, output = utils.container_ops.exec_command(
        container.id, 
        "curl -s http://localhost:8080/health"
    )
    print(f"健康检查结果: {output}")

# 清理资源
result = utils.cleanup_unused(include_volumes=True)
print(f"清理完成: {result}")
```

## 五、最佳实践建议

### 1. 错误处理
- 统一使用自定义异常类型，便于上层捕获和处理
- 对 Docker API 错误进行包装，提供更友好的错误信息

### 2. 连接管理
- 使用单例模式管理客户端连接，避免重复创建
- 支持自定义 Docker 主机地址，方便连接远程 Docker

### 3. 性能优化
- 批量操作时使用 Docker API 的过滤功能，减少数据传输
- 获取日志时使用 `tail` 参数限制返回数量

### 4. 安全性
- 避免在代码中硬编码敏感信息
- 使用环境变量或配置文件管理 Docker 连接参数

### 5. 测试
- 使用 `docker-mock` 等库进行单元测试
- 在 CI/CD 中集成 Docker 集成测试

## 六、总结

通过封装 Docker 工具类，我们可以：

- **减少重复代码**：将常用的 Docker 操作抽象为通用函数
- **提高代码质量**：统一的错误处理和日志记录
- **便于维护**：修改 Docker 相关逻辑只需改动一处
- **提升开发效率**：提供简洁的 API 接口，快速集成