第 18 章 · CloudWatch 监控与 CI/CD

本章目标：

• 将 Agent 请求日志结构化输出到 CloudWatch Logs
• 创建自定义 Metrics（token 用量、延迟、错误率）并配置 Alarms
• 编写完整的 GitHub Actions CI/CD pipeline（test → lint → build → deploy to Fargate）
• 使用 CloudWatch Logs Insights 进行日志查询与排障

---

在前面的章节中，我们已经完成了 ECS/Fargate 部署和 SQS 消息队列的搭建。一个生产级的 Agent 服务不仅要把代码跑起来，还要确保它能被监控、告警、持续交付。本章我们将这些能力串联起来。

18.1 结构化日志输出到 CloudWatch

CloudWatch Logs 是 AWS 原生的日志服务。要让日志有价值，必须结构化——否则就是一堆无法搜索的文本。

18.1.1 日志格式约定

所有 Agent 请求应使用 JSON 格式输出，关键字段如下：

# agent_prod/app/logging.py
import json
import logging
import time
from datetime import datetime, timezone
from uuid import uuid4

class StructuredFormatter(logging.Formatter):
    """结构化日志格式化器，输出 JSON 行"""
    
    def format(self, record: logging.LogRecord) -> str:
        base = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
            "request_id": getattr(record, "request_id", uuid4().hex[:8]),
            "service": "agent-prod",
        }
        
        # 附加 exc_info 用于异常
        if record.exc_info:
            base["exception"] = self.formatException(record.exc_info)
        
        # 附加 extra 字段
        for key in ("agent_id", "thread_id", "tool_name", "latency_ms", "error_code"):
            if hasattr(record, key):
                base[key] = getattr(record, key)
        
        return json.dumps(base, ensure_ascii=False)


def get_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    
    handler = logging.StreamHandler()
    handler.setFormatter(StructuredFormatter())
    logger.addHandler(handler)
    
    return logger

使用时只需在 logger 上设置 extra 字段：

# agent_prod/app/agent_service.py
import logging

logger = get_logger(__name__)

async def handle_agent_request(thread_id: str, user_input: str) -> dict:
    start = time.time()
    request_id = uuid4().hex[:8]
    
    try:
        result = await run_graph(thread_id, user_input)
        latency_ms = int((time.time() - start) * 1000)
        
        logger.info(
            "Agent request completed",
            extra={
                "request_id": request_id,
                "thread_id": thread_id,
                "latency_ms": latency_ms,
                "tool_name": result.get("final_tool"),
            },
        )
        return result
    except Exception as e:
        latency_ms = int((time.time() - start) * 1000)
        logger.error(
            f"Agent request failed: {e}",
            extra={
                "request_id": request_id,
                "thread_id": thread_id,
                "latency_ms": latency_ms,
                "error_code": type(e).__name__,
            },
        )
        raise

18.1.2 发送到 CloudWatch Logs

使用 boto3 配合 CloudWatch Logs Client，或在容器启动时通过 Amazon CloudWatch Agent 自动采集。

最小化的 boto3 实现：

# agent_prod/app/cloudwatch_logger.py
import asyncio
import boto3
import logging
from datetime import datetime, timezone
from typing import Dict, Any

class CloudWatchLogHandler(logging.Handler):
    """将结构化日志推送到 CloudWatch Logs"""
    
    def __init__(self, log_group: str, log_stream: str | None = None):
        super().__init__()
        self.client = boto3.client("logs", region_name="us-east-1")
        self.log_group = log_group
        self.log_stream = log_stream or f"{datetime.utcnow():%Y/%m/%d}/{uuid4().hex[:8]}"
        self._batch: list[dict] = []
        self._sequence_token: str | None = None
    
    def emit(self, record: logging.LogRecord) -> None:
        log_event = {
            "timestamp": int(datetime.fromisoformat(
                record.created.isoformat()
            ).timestamp() * 1000),
            "message": self.format(record),
        }
        self._batch.append(log_event)
        
        if len(self._batch) >= 10_000 or len(str(self._batch)) >= 1024 * 1024:
            self._flush()
    
    def _flush(self) -> None:
        if not self._batch:
            return
        response = self.client.put_log_events(
            logGroupName=self.log_group,
            logStreamName=self.log_stream,
            logEvents=self._batch,
            sequenceToken=self._sequence_token,
        )
        self._sequence_token = response.get("nextSequenceToken")
        self._batch.clear()

注意：生产环境推荐使用 AWS Distro for OpenTelemetry (ADOT) 或 CloudWatch Agent，它们支持自动批处理与重试。

18.2 自定义 Metrics：Token、延迟与错误率

CloudWatch Metrics 可以存储任意数值。我们为 Agent 服务定义三个核心指标。

18.2.1 指标定义

指标名称	类型	单位	描述
AgentTokenUsage	Counter	tokens	每次请求消耗的总 token 数
AgentRequestLatency	Gauge	milliseconds	请求处理延迟
AgentErrorRate	Rate	count	错误发生次数

# agent_prod/app/metrics.py
import boto3
from botocore.config import Config

CW_CLIENT = boto3.client(
    "cloudwatch",
    region_name="us-east-1",
    config=Config(retries={"max_attempts": 3}),
)

NAMESPACE = "AgentProd"


def put_token_usage(agent_id: str, tokens: int) -> None:
    CW_CLIENT.put_metric_data(
        Namespace=NAMESPACE,
        MetricData=[
            {
                "MetricName": "AgentTokenUsage",
                "Value": float(tokens),
                "Unit": "Count",
                "Dimensions": [{"Name": "AgentId", "Value": agent_id}],
            }
        ],
    )


def put_latency(agent_id: str, latency_ms: float) -> None:
    CW_CLIENT.put_metric_data(
        Namespace=NAMESPACE,
        MetricData=[
            {
                "MetricName": "AgentRequestLatency",
                "Value": latency_ms,
                "Unit": "Milliseconds",
                "Dimensions": [{"Name": "AgentId", "Value": agent_id}],
            }
        ],
    )


def put_error(agent_id: str) -> None:
    CW_CLIENT.put_metric_data(
        Namespace=NAMESPACE,
        MetricData=[
            {
                "MetricName": "AgentErrorRate",
                "Value": 1.0,
                "Unit": "Count",
                "Dimensions": [{"Name": "AgentId", "Value": agent_id}],
            }
        ],
    )

在 Agent 处理函数中注入：

# agent_prod/app/agent_service.py
from .metrics import put_token_usage, put_latency, put_error

async def handle_agent_request(thread_id: str, user_input: str) -> dict:
    start = time.time()
    request_id = uuid4().hex[:8]
    
    try:
        result = await run_graph(thread_id, user_input)
        latency_ms = (time.time() - start) * 1000
        
        # 上报指标
        token_count = result.get("token_usage", 0)
        put_token_usage(result.get("agent_id", "default"), token_count)
        put_latency(result.get("agent_id", "default"), latency_ms)
        
        logger.info(
            "Agent request completed",
            extra={"request_id": request_id, "latency_ms": latency_ms},
        )
        return result
    except Exception as e:
        agent_id = result.get("agent_id", "default") if "result" in dir() else "default"
        put_error(agent_id)
        raise

18.3 CloudWatch Alarms：延迟与错误率告警

当指标超过阈值时，CloudWatch Alarms 可以触发 SNS 通知或自动执行操作。

18.3.1 延迟告警

// agent_prod/deploy/cloudwatch-alarms.json
{
  "AlarmName": "AgentRequestLatencyHigh",
  "AlarmDescription": "Agent 请求延迟超过 5 秒",
  "MetricName": "AgentRequestLatency",
  "Namespace": "AgentProd",
  "Statistic": "Average",
  "Period": 60,
  "EvaluationPeriods": 3,
  "Threshold": 5000,
  "ComparisonOperator": "GreaterThanThreshold",
  "Dimensions": [
    { "Name": "AgentId", "Value": "*" }
  ],
  "AlarmActions": [
    "arn:aws:sns:us-east-1:123456789012:agent-alerts"
  ]
}

用 AWS CLI 部署：

aws cloudwatch put-metric-alarm \
  --cli-input-json file://cloudwatch-alarms.json

18.3.2 错误率告警

# agent_prod/deploy/create_alarms.py
import boto3
import json

cw = boto3.client("cloudwatch", region_name="us-east-1")
sns_topic = "arn:aws:sns:us-east-1:123456789012:agent-alerts"

alarms = [
    {
        "AlarmName": "AgentErrorRateHigh",
        "MetricName": "AgentErrorRate",
        "Namespace": "AgentProd",
        "Statistic": "Sum",
        "Period": 60,
        "EvaluationPeriods": 2,
        "Threshold": 10,  # 每分钟超过 10 个错误
        "ComparisonOperator": "GreaterThanThreshold",
        "Dimensions": [{"Name": "AgentId", "Value": "*"}],
        "AlarmActions": [sns_topic],
    },
    {
        "AlarmName": "AgentLatencyHigh",
        "MetricName": "AgentRequestLatency",
        "Namespace": "AgentProd",
        "Statistic": "Average",
        "Period": 60,
        "EvaluationPeriods": 3,
        "Threshold": 5000,
        "ComparisonOperator": "GreaterThanThreshold",
        "Dimensions": [{"Name": "AgentId", "Value": "*"}],
        "AlarmActions": [sns_topic],
    },
]

for alarm in alarms:
    cw.put_metric_alarm(**alarm)
    print(f"Created alarm: {alarm['AlarmName']}")

18.4 GitHub Actions CI/CD Pipeline

从代码提交到 Fargate 部署的完整流水线：

# .github/workflows/ci-cd.yml
name: CI/CD - Agent Prod

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

env:
  REGISTRY: public.ecr.aws
  IMAGE_NAME: agent-prod
  ECS_CLUSTER: agent-prod-cluster
  ECS_SERVICE: agent-prod-service
  AWS_REGION: us-east-1

jobs:
  test:
    name: Run Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: pip install pytest pytest-asyncio
      - run: pytest tests/ -v --tb=short
      - run: coverage run -m pytest tests/
      - run: coverage report

  lint:
    name: Lint & Type Check
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install ruff mypy
      - run: ruff check agent_prod/
      - run: ruff format --check agent_prod/
      - run: mypy agent_prod/ --strict

  build:
    name: Build & Push Image
    needs: [test, lint]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ env.AWS_REGION }}
      - name: Login to ECR
        uses: aws-actions/amazon-ecr-login@v2
      - name: Build & push
        run: |
          docker build -t ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest .
          docker tag ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest \
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
          docker push ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
          docker push ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}

  deploy:
    name: Deploy to Fargate
    needs: [build]
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4
      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: ${{ env.AWS_REGION }}
      - name: Update ECS service
        run: |
          aws ecs update-service \
            --cluster ${{ env.ECS_CLUSTER }} \
            --service ${{ env.ECS_SERVICE }} \
            --force-new-deployment \
            --task-definition agent-prod-task

18.4.1 secrets 配置

在 GitHub Repository Settings → Secrets and variables → Actions 中设置：

AWS_ACCESS_KEY_ID     # ECS access key
AWS_SECRET_ACCESS_KEY # ECS secret key
ECR_REPO              # ecr repository URI

18.5 CloudWatch Logs Insights 查询

使用结构化日志后，可以通过 Logs Insights 快速分析。

18.5.1 查询最近 1 小时的平均延迟

fields @timestamp, @message
| filter @logGroup like /agent-prod/
| parse @message /"latency_ms": (?<latency>\d+)/
| stats avg(latency) as avg_latency by bin(5m)
| sort @timestamp desc

18.5.2 查找错误请求

fields @timestamp, request_id, error_code
| filter @logGroup like /agent-prod/
| filter @message like /error/i
| sort @timestamp desc
| limit 20

18.5.3 Token 用量 Top Agent

fields @timestamp
| parse @message /"agent_id": "(?<agent_id>[^"]+)"/
| parse @message /"latency_ms": (?<latency>\d+)/
| stats sum(latency) as total_latency by agent_id
| sort total_latency desc
| limit 10

---

本章小结

• 结构化日志：使用 JSON 格式输出，包含 request_id、latency_ms、error_code 等字段
• 自定义 Metrics：通过 put_metric_data 上报 token 用量、延迟、错误率
• CloudWatch Alarms：配置延迟 > 5s 或错误率 > 10/min 的告警，触发 SNS 通知
• CI/CD Pipeline：GitHub Actions 执行 test → lint → build → deploy to Fargate
• Logs Insights：使用查询语法快速分析日志与指标

🛠️ 动手实践

1. 完善告警配置：修改 cloudwatch-alarms.json，增加一个基于 AgentTokenUsage 的告警，当每小时 token 消耗超过 100 万时触发通知。

2. 扩展 CI/CD：在 ci-cd.yml 的 deploy job 中，增加一个等待 ECS 服务稳定性的步骤（轮询 describe-services 直到 status 为 ACTIVE）。

3. 日志查询实战：在 CloudWatch Console 中使用 Logs Insights 查询过去 24 小时内所有 error_code 不为空的请求，导出结果到 S3。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. CloudWatch Alarms 的 EvaluationPeriods 参数是什么含义？

2. 以下哪个不是 CloudWatch Logs Insights 支持的语法关键字？

3. GitHub Actions 中，如何实现只在 main 分支部署？

4. 结构化日志的主要优势是什么？

---

📌 本章内容假设你已经完成 FastAPI ch22 的 Docker 基础，本章在此基础上讲述如何将其部署到 AWS 并加上监控与 CI/CD。