Docker-容器诊断工具：我是如何构建这样一个能够监控并修复我的容器的AI代理程序的

也许你会觉得这种情况很熟悉：你的生产环境中的容器在凌晨3点崩溃了。等到你醒来的时候，它已经连续出现了同样的错误两个小时了。你需要通过SSH登录到容器中，提取日志文件，解读那些晦涩难懂的错误信息，在网上搜索相关解决方案，最后才重新启动容器。这样，你早上本来可以用来做其他事情的20分钟就被浪费掉了。而最糟糕的是，下周这种情况还会再次发生。

我受够了这种循环。当时我在一台Linode服务器上运行了5个基于容器的服务：一个Flask API、一个Postgres数据库、一个Nginx反向代理服务器、一个Redis缓存系统，以及一个后台工作进程。每隔一周，其中就会有一个服务出现故障。日志文件杂乱无章，错误信息也不明显，而我却要浪费时间去调试那些本可以在几秒钟内就被自动检测并解决的问题。

因此，我开发了一个更好的解决方案：一个Python脚本，它可以实时监控这些容器，一旦发现错误，就会利用Claude算法分析出问题所在，并在不需要打扰你的情况下自动修复这些问题。我将这个工具称为“Container Doctor”。这并不是什么神奇的东西，它只不过是Docker API、大语言模型以及一些自动化脚本的结合而已。下面我会详细说明我是如何开发这个工具的，在开发过程中遇到了哪些问题，以及如果再做一次的话，我会做出哪些不同的调整。

目录

1. 为什么不直接使用Prometheus呢？

2. 系统架构

3. 项目设置

4. 监控脚本——详细步骤

5. Claude诊断提示语（以及结构的重要性）

6. 自动修复逻辑——为何要采取保守策略

7. 添加Slack通知功能

8. 健康检查接口

9. 限制Claude调用的频率

10. Docker Compose配置——完整步骤

11. 在生产环境中发现的真实问题

12. 成本分析——这项工作的实际花费

13. 安全方面的考虑因素

14. 如果再做一次，我会做出哪些不同之处

15. 下一步计划是什么？

为什么不直接使用Prometheus呢？

这确实是个好问题。Prometheus、Grafana、DataDog这些工具都非常优秀，但对于我的需求来说，它们反而有些过度了。我在一台每月只需20美元的Linode服务器上运行了5个容器，而要使用Prometheus的话，就需要部署一个指标收集服务器，为每个服务配置相应的数据导出机制，搭建Grafana监控界面，还要编写警报规则。仅仅为了监控这5个容器，就要耗费这么多时间和资源，实在有些得不偿失。

即便如此，这些工具也只能告诉你“发生了什么”。它们会显示内存使用量的突然增加或500错误率之类的数据，但却无法说明“为什么会发生这种情况”。你仍然需要人类来查看日志，找出根本原因，然后决定该采取什么措施。

这就是我想填补的空白。我并不需要另一个控制面板，而是需要一种能够解读堆栈跟踪信息、理解具体情境，并且要么自动解决问题，要么明确告诉我该做什么的工具。事实证明，Claude在这方面表现得非常出色。它能够快速解读Python堆栈跟踪信息，从而找出问题所在——其效率甚至超过了大多数初级开发人员（说实话，有些资深开发者也不及它的水平）。

架构设计

各部分之间的连接关系如下：

┌─────────────────────────────────────────────┐
│              Docker主机                      │
│                                               │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐   │
│  │   Web服务    │  │   API服务    │  │    数据库    │   │
│  │ (使用nginx)  │  │ (使用flask)  │  │(使用postgres)│   │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘   │
│       │              │              │         │
│       └──────────────┼──────────────┘         │
│                      │                         │
│              Docker套接字                     │
│                      │                         │
│            ┌─────────┴─────────┐              │
│            │ Container Doctor  │              │
│            │  (Python代理)   │              │
│            └─────────┬─────────┘              │
│                      │                         │
└──────────────────────┼─────────────────────────┘
                       │
              ┌────────┴────────┐
              │   Claude API    │
              │  (用于诊断)    │
              └────────┬────────┘
                       │
              ┌────────┴────────┐
              │  Slack通知机制  │
              │  (用于发送警报)       │
              └─────────────────┘

整个流程的工作原理如下：

1. Container Doctor会在自己的容器中运行，并连接Docker套接字。

2. 每隔10秒，它会从每个目标容器中获取最后50行的日志信息。

3. 它会扫描这些日志中的错误线索（如“error”、“exception”等关键词）。

4. 一旦发现异常，它就会将相关日志连同结构化的提示信息一起发送给Claude。

5. Claude会返回一份JSON格式的诊断报告，其中包含根本原因、问题严重程度、建议的解决方法以及是否可以自动重启容器。

6. 如果问题严重程度较高且自动重启是安全的，脚本就会重新启动相应的容器。

7. 无论结果如何，系统都会通过Slack发送包含完整诊断信息的通知。

   另外，还有一个简单的健康检查接口，可以用来查看Container Doctor本身的运行状态。

关键在于：这个脚本本身并不尝试进行复杂的诊断分析，而是将所有工作都交给了Claude来处理。脚本的任务仅仅是收集日志、将其转发给Claude，并执行相应的响应操作而已。

项目设置

创建您的项目目录：

mkdir container-doctor && cd container-doctor

这是您的requirements.txt文件内容：

docker==7.0.0
anthropic>>=0.28.0
python-dotenv==1.0.0
flask==3.0.0
requests==2.31.0

为测试目的在本地安装所需依赖项：pip install -r requirements.txt

创建一个.env文件：

ANTHROPIC_API_KEY=sk-ant-...
TARGET_contAINERS=web,api,db
CHECK_INTERVAL=10
LOG_lines=50
AUTO_fix=true
SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL
POSTGRES_USER=user
POSTGRES_PASSWORD=changeme
POSTGRES_DB=mydb
MAX_DIAGNOSES_PER_HOUR=20

关于CHECK_INTERVAL的说明：10秒的间隔时间已经相当快了。在生产环境中，我会将其调整为30到60秒。在开发阶段，我保持这个较短的间隔时间以便能更快地看到测试结果，说实话，后来我都忘了调整它……直到我的API费用账单提醒了我才想起来。

监控脚本——逐行解析

以下是完整的container_doctor.py文件。之后我会逐一解释其中的重要部分：

import docker
import json
import time
import logging
import os
import requests
from datetime import datetime, timedelta
from collections import defaultdict
from threading import Thread
from flask import Flask, jsonify
from anthropic import Anthropic

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

client = Anthropic()
docker_client = None

# --- 配置参数 ---
TARGET_contAINERS = os.getenv("TARGET_CONTAINERS", "").split(",")
CHECK_INTERVAL = int(os.getenv("CHECK INTERVAL", "10"))
LOG_lines = int(os.getenv("LOG_LINES", "50"))
AUTO_fix = os.getenv("AUTO_FIX", "true").lower() == "true"
SLACK_WEBHOOK = os.getenv("SLACK/WebHOOK_URL", "")
MAX_DIAGNOSES = int(os.getenv("MAX_DIAGNOSES_PER_HOUR", "20"))

# --- 状态跟踪 ---
diagnosis_history = []
fix_history = defaultdict(list)
last_error_seen = {}
rate_limit_counter = defaultdict(int)
rate_limit_reset = datetime.now() + timedelta(hours=1)

app = Flask(__name__)

def get_docker_client():
    """延迟初始化Docker客户端。"""
    global docker_client
    if docker_client is None:
        docker_client = docker.from_env()
    return docker_client


def get_container_logs(container_name):
    """获取容器的最后N行日志。"""
    try:
        container = get_docker_client().containers.get(container_name)
        logs = container.logs(
            tail=LOG_lines,
            timestamps=True
        ).decode("utf-8")
        return logs
    except docker.errors.NotFound:
        logger.warning(f"未找到容器{container_name}。跳过此操作。")
        return None
    except docker.errors.APIError as e:
        logger.error(f"获取容器{container_name}的日志时出现Docker API错误：{e}")
        return None
    except Exception as e:
        logger.error(f"在获取容器{container_name}的日志时发生意外错误：{e}")
        return None


def detect_errors(logs):
    """检查日志中是否包含错误信息。"""
    error_patterns = [
        "error", "exception", "traceback", "failed", "crash",
        "fatal", "panic", "segmentation fault", "out of memory",
        "killed", "oomkiller", "connection refused", "timeout",
        "permission denied", "no such file", "errno"
    ]
    logs_lower = logs.lower()
    found = []
    for pattern in error_patterns:
        if pattern in logs_lower:
            found.append(pattern)
    return found


def is_new_error(container_name, logs):
    """判断这是否是一个新的错误，还是我们之前已经诊断过的错误。"""
    log_hash = hash(logs[-200:])  # 计算最后200个字符的哈希值
    if last_error_seen.get(container_name) == log_hash:
        return False
    last_error_seen[container_name] = log_hash
    return True


def check_rate_limit():
    """确保不会因为发送过多请求而干扰Claude的工作。"""
    global rate_limit_counter, rate_limit_reset

    now = datetime.now()
    if now > rate_limit_reset:
        rate_limit_counter.clear()
        rate_limit_reset = now + timedelta(hours=1)

    total = sum(rate_limit_counter.values())
    if total >= MAX_DIAGNOSES:
        logger.warning(f"已达到每小时{total}/{MAX_DIAGNOSES}次的请求限制。跳过此次诊断操作。")
        return False
    return True


def diagnose_with_claude(container_name, logs, error_patterns):
    """将日志发送给Claude进行诊断。"""
    if not check_rate_limit():
        return None

    rate_limit_counter[container_name] += 1

    prompt = f"""您是一名DevOps专家，正在分析容器的日志。

容器：{container_name}
时间戳：{datetime.now().isoformat()}
检测到的错误模式：{', '.join(error_patterns)}

最近的日志：
---
{logs}
---

请分析这些日志，并仅回复有效的JSON格式内容（不要使用markdown或解释性文字）：
{{
    "root_cause": "用一句话准确说明出了问题所在",
    "severity": "低|中|高",
    "suggested_fix": "操作员应按照以下步骤进行修复",
    "auto_restart_safe": true或false,
    "config_suggestions": ["ENV_VAR=value", "..."],
    "likely_recurring": true或false,
    "estimated_impact": "如果不进行修复，将会产生什么影响"
}}
"""

    try:
        message = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=600,
            messages=[
                {"role": "user", "content": prompt}
            ]
        )
        return message.content[0].text
    except Exception as e:
        logger.error(f"Claude API出现错误：{e}")
        return None


def parse_diagnosis(diagnosis_text):
    """从Claude的回复中提取JSON数据。"""
    if not diagnosis_text:
        return None
    try:
        start = diagnosis_text.find("{")
        end = diagnosis_text.rfind("}") + 1
        if start >= 0 and end > start:
            json_str = diagnosis_text[start:end]
            return json.loads(json_str)
    except json.JSONDecodeError as e:
        logger.error(f"JSON解析失败：{e}")
        logger.debug(f"原始回复内容：{diagnosis_text}")
    except Exception as e:
        logger.error(f"无法解析诊断结果：{e}")
    return None


def apply_fix(container_name, diagnosis):
    """在安全的情况下应用自动修复措施。"""
    if not AUTO_FIX:
        logger.info(f"全局禁用了自动修复功能。跳过容器{container_name}的检测操作。")
        return False

    if not diagnosis.get("auto_restart_safe"):
        logger.info(f"Claude提示：重新启动容器{container_name}存在安全风险。跳过此操作。")
        return False

    # 确保每小时不会重复重启同一个容器超过3次
    recent_fixes = [
        t for t in fix_history[container_name]
        if t > datetime.now() - timedelta(hours=1)
    ]
    if len(recent_fixes) >= 3:
        logger.warning(
            f"容器{container_name}在今天已经重新启动了{len(recent_fixes)}次。可能存在更严重的问题。跳过此次检测操作。"
        )
        send_slack_alert(
            container_name, diagnosis,
            extra="连续失败：该容器在过去1小时内已被重启3次以上。需要人工干预。"
        )
        return False

    try:
        container = get_docker_client().containers.get(container_name)
        logger.info(f"正在重新启动容器{container_name}...")
        container.restart(timeout=30)
        fix_history[container_name].append(datetime.now())
        logger.info(f"容器{container_name}已成功重启。")

        # 确认容器确实已经重新启动并运行正常
        time.sleep(5)
        container.reload()
        if container.status != "running":
            logger.error(f"容器{container_name}重启后仍然无法正常运行。")
            return False

        return True
    except Exception as e:
        logger.error(f"重新启动容器{container_name}时发生错误：{e}")
        return False


def send_slack_alert(container_name, diagnosis, extra=""):
    """将诊断结果发送到Slack。"""
    if not SLACK_WEBHOOK:
        return

    severity_emoji = {
        "low": "🟡",
        "medium": "🟠",
        "high": "🔴"
    }

    severity = diagnosis.get("severity", "未知")
    emoji = severity_emoji.get(severity, "⚪")

    blocks = [
        {
            "type": "header",
            "text": {
                "type": "plain_text",
                "text": f"{emoji} 容器诊断警报：{container_name}"
            }
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*严重程度：* {severity}"},
                {"type": "mrkdwn", "text": f"*容器名称：* `{container_name}`"},
                {"type": "mrkdwn", "text": f"*根本原因：* {diagnosis.get('root_cause', '未知')}"},
                {"type": "mrkdwn", "text": f"*修复建议：* {diagnosis.get('suggested_fix', 'N/A')}"
            ]
        }
    ]

    if diagnosis.get("config_suggestions"):
        suggestions = "\n".join(
            f"• `{s}`" for s in diagnosis["config_suggestions"]
        )
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"*配置建议：*\n{suggestions}"
            }
        })

    if extra:
        blocks.append({
            "type": "section",
            "text": {"type": "mrkdwn", "text": f"*⚠️ {extra}*"}
        })

    try:
        requests.post(SLACK_WEBHOOK, json={"blocks": blocks}, timeout=10)
    except Exception as e:
        logger.error(f"Slack通知发送失败：{e}")


# --- 健康检查端点 ---
@app.route("/health")
def health():
    """用于检查容器诊断系统的健康状态。"""
    try:
        get_docker_client().ping()
        docker_ok = True
    except:
        docker_ok = False

    return jsonify({
        "status": "正常" if docker_ok else "异常",
        "docker_connected": docker_ok,
        "monitoring": TARGET_contAINERS,
        "total_diagnoses": len(diagnosis_history),
        "fixes_applied": {k: len(v) for k, v in fix_history.items()),
        "rate_limit_remaining": MAX_DIAGNOSES - sum(rate_limit_counter.values()),
        "uptime_check": datetime.now().isoformat()
    }


@app.route("/history")
def history():
    """返回最近的诊断记录。"""
    return jsonify(diagnosis_history[-50:]))


def monitor_containers():
    """主要的监控循环函数。"""
    logger.info(f"容器诊断系统正在启动...")
    logger.info(f"当前正在监控的容器数量：{TARGET_CONTAINERS}")
    logger.info(f"检查间隔时间为：{CHECK_INTERVAL}秒")
    logger.info(f"自动修复功能是否启用：{AUTO_fix}")
    logger.info(f"当前的请求限制设置为：{MAX_DIAGNOSES}次/小时")

    while True:
        for container_name in TARGET_containers:
            container_name = container_name.strip()
            if not container_name:
                continue

            logs = get_container_logs(container_name)
            if not logs:
                continue

            error_patterns = detect_errors(logs)
            if not errorpatterns:
                continue

            # 如果已经诊断过这个错误，就跳过本次检查
            if not is_new_error(container_name, logs):
                continue

            logger.warning(
                f"在容器{container_name}中检测到了以下错误：{error_patterns}"
            )

            diagnosis_text = diagnose_with_claude(
                container_name, logs, errorpatterns
            )
            if not diagnosis_text:
                continue

            diagnosis = parse_diagnosis(diagnosis_text)
            if not diagnosis:
                logger.error("无法解析Claude的诊断结果。跳过本次检查。")
                continue

            # 将诊断结果记录下来
            diagnosis_history.append({
                "container": container_name,
                "timestamp": datetime.now().isoformat(),
                "diagnosis": diagnosis,
                "patterns": error_patterns
            })

            logger.info(
                f"容器{container_name}的诊断结果为："
                f"严重程度：{diagnosis.get('severity')},
                "根本原因：{diagnosis.get('root_cause')}"
            )

            # 仅对严重程度为“高”的错误应用自动修复功能
            fixed = False
            if diagnosis.get("severity") == "high":
                fixed = apply_fix(container_name, diagnosis)

            # 必须向Slack发送通知
            send_slack_alert(
                container_name, diagnosis,
                extra="已自动修复" if fixed else ""
            )

        time.sleep(CHECK_INTERVAL)


if __name__ == "__main__":
    # 在后台运行Flask的健康检查端点
    flask_thread = Thread(
        target=lambda: app.run(host="0.0.0.0", port=8080, debug=False),
        daemon=True
    )
    flask_thread.start()
    logger.info("健康检查端点正在8080端口运行..."

    try:
        monitor_containers()
    except KeyboardInterrupt:
        logger.info("容器诊断系统正在关闭...")

这段代码确实很长，因此让我来重点讲解其中重要的部分吧。

错误去重功能（is_new_error）：这个功能是我通过惨痛的经验才学会的。如果不使用这个机制，脚本会每隔10秒就检测到相同的错误，并向Claude发送重复的请求。我会对日志输出的最后200个字符进行哈希处理，如果这些字符与我们之前记录过的错误信息相匹配，就会跳过这次检测。这个方法很简单，但却让我的API使用成本降低了大约80%。

curl http://localhost:8080/health

{
    "status": "healthy",
    "docker_connected": true,
    "monitoring": ["web", "api", "db"],
    "total_diagnoses": 14,
    "fixes_applied": {"api": 2, "web": 1},
    "rate_limit_remaining": 6,
    "uptime_check": "2026-03-15T14:30:00"
}

curl http://localhost:8080/history

version: '3.8'

services:
  container_doctor:
    build:
      context: .
      dockerfile: Dockerfile
    container_name: container_doctor
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - TARGET_contAINERS=web,api,db
      - CHECK_INTERVAL=10
      - LOG_lines=50
      - AUTO_fix=true
      - SLACK_WEBHOOK_URL=${SLACK/WebHOOK_URL}
      - MAX_DIAGNOSES_PER_HOUR=20
    ports:
      - "8080:8080"
    restart: unless-stopped
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/health"]
      interval: 30s
      timeout: 10s
      retries: 3

  web:
    image: nginx:latest
    container_name: web
    ports:
      - "80:80"
    restart: unless-stopped

  api:
    build: ./api
    container_name: api
    environment:
      - DATABASE_URL=postgres://\({POSTGRES_USER}:\){POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
      - POOL_SIZE=20
    depends_on:
      - db
    restart: unless-stopped

  db:
    image: postgres:15
    container_name: db
    environment:
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - POSTGRES_DB=${POSTGRES_DB}
    volumes:
      - db_data:/var/lib/postgresql/data
    restart: unless-stopped
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER}"]
      interval: 10s
      timeout: 5s
      retries: 5

volumes:
  db_data:

FROM python:3.12-slim

WORKDIR /app

RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY container_doctor.py .

EXPOSE 8080

CMD ["python", "-u", "container_doctor.py"]

{
    "root_cause": "进程因内存不足被OOM Killer终止。该容器在负载较高时请求的内存超出了256MB的限制。",
    "severity": "high",
    "suggested_fix": "在docker-compose.yml文件中将内存限制增加到512MB，并继续监控在更高限制下是否还会出现内存泄漏问题。",
    "auto_restart_safe": true,
    "config_suggestions": ["mem_limit: 512m", "memswap_limit: 1g"],
    "likely_recurring": true,
    "estimated_impact": "API服务会完全中断，所有请求都会收到来自nginx的502错误响应。"
}

{
    "root_cause": "SQLAlchemy连接的连接池容量仅为5个，无法满足8个并发进程的需求。每个进程在处理请求时都会占用一个连接。",
    "severity": "high",
    "suggested_fix": "将POOL_SIZE设置为20，并增加POOL_TIMEOUT为30秒。长期解决方案是使用PgBouncer作为连接池管理工具。",
    "auto_restart_safe": true,
    "config_suggestions": ["POOL_SIZE=20", "POOL_TIMEOUT=30", "POOL_RECYCLE=3600"],
    "likely_recurring": true,
    "estimated_impact": "新的API请求会等待30秒后超时；现有的请求可能会完成，但速度会很慢。"
}