CTAgent Systems

AIOps Agent 与 LangGraph

从故障注入到 Agent 启动：用 Prometheus 与 Alertmanager 构建可靠的自动告警链路

连接故障注入、Prometheus、Alertmanager 与 Agent 调查入口，并处理幂等和重试。

PrometheusAlertmanager自动告警

订单服务的数据库连接池已经耗尽，请求开始排队并超时。此时，AIOps Agent 会自动知道发生了故障吗？

不会。Agent 只会处理送到面前的任务。在它开始诊断之前，系统还需要完成一条更基础、也更容易被低估的链路：把短暂的运行时异常转换成一个经过聚合、鉴权、去重并持久化的诊断任务。

本文以 Agent Py 中的隔离 Live Eval 场景 APY-LIVE-ORDER-POOL-LEAK-001 为例，追踪一次连接池耗尽如何经过 Prometheus、Alertmanager、Nginx、FastAPI 和 PostgreSQL，最终唤醒后台 Worker 并启动 AIOps Agent。

先给出本文的核心结论：

自动告警不等于 Alertmanager 调用一次 Webhook。可靠的自动告警是一条事件处理链：Prometheus 判断异常是否持续存在，Alertmanager 管理通知生命周期，接入服务把不可信的 HTTP 请求转成持久 Incident 和后台任务，Worker 再异步启动 Agent。

1. 先看全貌：一条告警经过了哪些系统？

flowchart LR
    A[Docker Compose<br/>注入连接池耗尽故障]
    B[Order API<br/>暴露连接池与业务探针指标]
    C[Prometheus<br/>抓取指标并计算规则]
    D{异常持续满足<br/>for: 4s?}
    E[Alertmanager<br/>分组、去重与路由]
    F[Nginx<br/>Webhook 独立限流]
    G[FastAPI<br/>鉴权、校验与标准化]
    H[(PostgreSQL<br/>Incident + Event<br/>Diagnostic Task + Job)]
    I[Background Worker<br/>领取持久任务]
    J[AIOps Agent<br/>开始诊断]
    K[继续评估指标]

    A --> B --> C --> D
    D -- 否 --> K --> C
    D -- 是 --> E --> F --> G --> H --> I --> J

这条链中，每个组件的职责并不重叠：

组件输入核心职责明确不负责
Order API真实请求与故障状态暴露连接池、等待者和业务探针指标判断是否应该创建事故
Prometheus周期性抓取的时间序列计算告警表达式，判断异常是否持续存在管理通知接收方和 Incident
AlertmanagerPrometheus 发来的 firing/resolved 告警分组、去重、等待、重复通知和路由创建项目内任务或运行 Agent
Nginx + FastAPIAlertmanager Webhook限流、鉴权、校验、数据最小化和接入状态机在 HTTP 请求中执行长时间诊断
PostgreSQL标准化后的告警事件原子保存 Incident、事件、诊断任务和后台 Job替代 Prometheus 计算指标
Background Worker可领取的持久 Job通过数据库租约领取任务并调用诊断 Handler决定告警规则是否成立

一个好用的心智模型是：Prometheus 是“探测器”，Alertmanager 是“通知调度台”，Webhook 接入层是“事故登记处”，PostgreSQL 是“不会因电话挂断而消失的工单本”，Worker 才是“把工单交给 Agent 的执行者”。这个类比只用于划分职责；真正的正确性仍由告警规则、HTTP 契约和数据库事务保证。

2. Prometheus 如何把连接池指标变成告警？

2.1 单个指标通常不足以证明故障

连接池的空闲连接为零，不一定代表已经发生事故。它可能只是某个瞬间所有连接都在正常工作，很快就会归还。

因此，Agent Py 的 Live Eval 规则没有只判断 free == 0，而是要求五个条件同时成立：

- alert: OrderApiConnectionPoolExhausted
  expr: |
    (agentpy_order_pool_fault_active == 1)
    and on (service, environment, scenario_id, run_id)
    (agentpy_order_pool_free == 0)
    and on (service, environment, scenario_id, run_id)
    (agentpy_order_pool_checked_out == agentpy_order_pool_capacity)
    and on (service, environment, scenario_id, run_id)
    (agentpy_order_pool_waiter_observed == 1)
    and on (service, environment, scenario_id, run_id)
    (agentpy_order_business_probe_success == 0)
  for: 4s

这五个信号分别回答不同的问题：

  1. fault_active == 1：隔离实验中的故障注入确实处于开启状态；
  2. free == 0：连接池没有空闲连接；
  3. checked_out == capacity：连接池容量已经全部借出；
  4. waiter_observed == 1：至少观察到请求等待连接；
  5. business_probe_success == 0：资源饱和已经影响业务探针。

组合条件比单指标更接近“资源已耗尽并产生业务影响”的事实，也能减少把正常高利用率误判成事故的机会。

2.2 for: 4s 解决的是瞬时抖动

Prometheus 规则的状态可以简化为：

Inactive ──表达式成立──> Pending ──持续 4 秒──> Firing
    ^                         │
    └──────表达式不再成立─────┘

表达式第一次成立时，告警进入 Pending，而不是立刻通知。只有异常连续满足 for: 4s，它才进入 Firing 并发送给 Alertmanager。如果等待期间指标恢复，告警会回到 Inactive。

生产环境中的持续窗口应该根据采集周期、业务 SLO、容忍抖动和故障影响重新标定。这里选择 4 秒，是为了让 Live Eval 能够在可控时间内完成，并不意味着生产连接池告警也应该等待 4 秒。

3. Alertmanager 为什么不能被一个普通 HTTP 请求替代？

Prometheus 擅长计算“现在是否满足告警条件”，但通知本身还有另一组问题：同一批告警如何合并、第一次等待多久、多久重复一次、恢复后是否通知，以及应该发给哪个接收方。

项目当前的 Alertmanager 路由配置为：

route:
  receiver: agent-py-webhook
  group_by: [alertname, service, environment, scenario_id, run_id]
  group_wait: 1s
  group_interval: 5s
  repeat_interval: 1m

receivers:
  - name: agent-py-webhook
    webhook_configs:
      - url: http://nginx/aiops/alerts/webhook/alertmanager/local-alertmanager
        send_resolved: true
        max_alerts: 50

这些字段分别控制：

  • group_by：哪些标签相同的告警属于同一通知组；
  • group_wait: 1s：新分组出现后先短暂等待，让同组告警有机会一起发送；
  • group_interval: 5s：同一分组内容发生变化时，两次通知之间的最小间隔；
  • repeat_interval: 1m：告警持续未恢复时，多久再次提醒；
  • send_resolved: true：恢复后把 resolved 状态也发送给 Webhook。

Alertmanager 因此解决了“如何投递通知”的问题，但它仍然不是 Agent Py 的 Incident 数据库。它不知道哪个用户和知识库应当接收诊断任务，也不负责保存项目内的审计事件。

这里还有一个关键差别：Alertmanager 的重复通知是正常机制，不是异常流量。接入系统必须天然幂等，不能假设每个 Webhook 只会收到一次。

4. Webhook 如何从不可信请求变成可靠 Incident？

4.1 Nginx 先保护入口

Webhook 使用独立于普通 API 和 SSE 的限流预算：

limit_req_zone $binary_remote_addr zone=alert_webhook_per_ip:10m rate=5r/s;

location ~ ^/aiops/alerts/webhook/alertmanager/[A-Za-z0-9._-]+$ {
    limit_req zone=alert_webhook_per_ip burst=20 nodelay;
    client_max_body_size 256k;
    include /etc/nginx/includes/proxy-common.conf;
    proxy_pass http://agent_py_backend;
}

独立限流有两个作用：告警风暴不会直接占满普通用户 API 的预算，普通 API 的流量也不会挤掉告警入口。client_max_body_size 256k 则在网关层拒绝明显过大的请求。

Nginx 限流是入口保护，不是业务幂等。即使每秒只收到一个请求，同一告警仍可能被重复投递；是否创建第二个任务必须由后面的状态机和数据库约束决定。

4.2 FastAPI 只做有界、快速的接入工作

Webhook 路由是：

POST /aiops/alerts/webhook/alertmanager/{source_id}

请求进入 FastAPI 后依次经过：

  1. 根据 source_id 查找已启用的 Source；
  2. 使用固定 Source 配置中的 Bearer Token 鉴权；
  3. 同时检查声明长度和实际读取字节数；
  4. 解析 Alertmanager Webhook v4 Payload；
  5. 只保留允许的 labels、annotations 和安全 URL origin；
  6. 对原始 groupKey 和请求体分别计算 SHA-256；
  7. 应用 Source 的标签过滤条件；
  8. 进入 Redis 短租约和 PostgreSQL 状态机；
  9. 数据提交后尝试唤醒 Background Runtime；
  10. 返回不含敏感字段的 202 响应。

接入层不会相信 Payload 中的 ownerUserId、knowledgeBaseId 或恢复权限。这些权限信息必须来自服务端 Source 配置，否则任何能调用 Webhook 的发送方都可能把任务写入其他用户空间，甚至尝试绕过恢复审批。

项目也不会持久化原始 groupKey 和原始请求体：

group_key_hash = sha256(group_key.encode("utf-8")).hexdigest()
payload_sha256 = sha256(raw_body).hexdigest()

前者用于识别同一 Incident 生命周期，后者用于识别相同 Delivery。真正进入数据库和后续诊断输入的，是经过字段白名单、长度限制和安全清洗后的结构。

5. 为什么一定要先写 PostgreSQL，再启动 Agent？

如果 Webhook 收到 firing 后直接调用 Agent，会出现一个危险窗口：Agent 已经开始运行，但 HTTP 连接突然断开，Alertmanager 不知道请求是否成功，于是再次投递。系统可能由此启动两个诊断任务。

更严重的是，如果后端进程在返回响应前退出，而任务只存在于内存中，那么告警已经被接收，诊断工作却永久丢失。

Agent Py 的处理方式是先在一个 PostgreSQL 事务中创建四类记录：

Alert Incident
    ├── Alert Event
    ├── Diagnostic Task
    └── Background Job

只有事务提交成功，Webhook 才把请求视为已经安全接收。这里有三层互补的幂等机制：

5.1 active Incident 唯一约束

同一个 owner + source + group_key_hash 生命周期最多只有一个 active Incident。首次 firing 使用 PostgreSQL 的 INSERT ... ON CONFLICT DO NOTHING 竞争创建权。

并发请求中只有成功创建 Incident 的事务会继续创建 Diagnostic Task 和 Background Job。其他请求读取已经存在的 active Incident，更新它的 last_seen_at 和 delivery_count，但复用原来的任务。

项目的 PostgreSQL 测试会并发发送 20 次 firing，并验证最终只有一个 Incident、一个 Diagnostic Task 和一个 Job。这比“先查询、再插入”的应用层判断可靠，因为后者在并发下可能让多个请求同时看到“尚不存在”。

5.2 稳定 Alert Event ID

Alert Event ID 由 owner、source、delivery status 和 payload_sha256 派生。网络层对完全相同 Payload 的重试不会重复添加审计事件，但每次经过验证的 Delivery 仍可更新 Incident 的投递计数和最后观察时间。

5.3 Redis 不是最终正确性来源

Redis 在这里提供约 2 秒的短租约，用于降低相同 groupKey 并发进入 PostgreSQL 时的竞争。获取不到租约的请求只会短暂等待；Redis 超时或不可用时，流程仍会进入 PostgreSQL。

因此，Redis 的角色是性能优化，PostgreSQL 唯一约束和事务才是正确性底座。把 Redis 锁当作唯一幂等保证，会让 Redis 故障、租约过期或客户端超时直接演变成重复任务。

6. 首次 firing、重复 firing 和 resolved 分别发生什么？

sequenceDiagram
    participant P as Prometheus
    participant A as Alertmanager
    participant W as Webhook API
    participant DB as PostgreSQL
    participant R as Background Worker
    participant G as AIOps Agent

    P->>A: firing
    A->>W: Webhook v4 firing
    W->>DB: 原子创建 Incident、Event、Task、Job
    DB-->>W: incident_created
    W-->>A: 202 Accepted
    W-)R: commit 后唤醒
    R->>DB: claim Job with lease
    R->>G: 启动诊断 Handler

    A->>W: 重复 firing
    W->>DB: 更新 last_seen_at / delivery_count
    DB-->>W: duplicate_updated + 原 Task ID
    W-->>A: 202 Accepted

    P->>A: resolved
    A->>W: Webhook v4 resolved
    W->>DB: Incident 标记 resolved，追加 Event
    DB-->>W: incident_resolved
    W-->>A: 202 Accepted
    Note over R,G: 已经开始的诊断不会因 resolved 被强制取消

对应的 Incident 生命周期如下：

stateDiagram-v2
    [*] --> Active: 无 active + firing\n创建 Incident / Task / Job
    Active --> Active: 重复 firing\n更新计数，不建新任务
    Active --> Resolved: resolved\n关闭 Incident
    [*] --> OrphanResolved: 无 active + resolved\n只记录审计事件
    Resolved --> Active: 再次 firing\n创建新生命周期
    OrphanResolved --> [*]

这里最容易误解的是 resolved：它表示 Prometheus 观察到告警条件已经不再成立，并不证明 Agent 的诊断已经无意义，也不代表某个恢复动作一定成功。因此，resolved 会关闭 Incident 的告警生命周期，但不会粗暴取消已经开始的诊断任务，也不会额外启动一次 LLM 调用。

7. HTTP 状态码其实在表达“是否已经安全接收”

Webhook 的返回码不是普通的接口成功或失败提示，它会影响发送方如何理解本次投递：

情况返回码系统含义
首次、重复、filtered、resolved 或 orphan resolved 已安全处理202请求已完成接入处理；不代表诊断已经完成
Bearer Token 缺失或错误401未鉴权，不持久化
Source 不存在或禁用404路由配置无效，不持久化
请求体超过限制413入口拒绝，不读取为业务事件
JSON 或 Webhook Schema 非法422Payload 无法进入状态机
PostgreSQL 事务失败503尚未安全持久化，应保留发送方后续重投的可能
Redis 不可用，但 PostgreSQL 提交成功202已降级处理，正确性不受 Redis 影响
PostgreSQL 已提交，但 Worker 即时唤醒失败202Job 已持久化，可由启动恢复或下一轮调度领取

401、404、413 和 422 通常意味着凭据、Source 或 Payload 配置需要被修正；机械重发同一请求并不会消除原因。503 则明确表示关键持久化没有完成，不能返回一个误导发送方的 2xx。

为什么 Worker 唤醒失败仍然返回 202？因为正确性的分界线是 PostgreSQL commit，而不是内存中的 runtime.start() 是否立即成功。只要 Job 已经写入数据库，后端重启或下一次调度就还能看到它。

8. Worker 如何保证重启后还能继续任务？

Background Worker 不靠“某个 Python 协程还活着”维持任务，而是通过 PostgreSQL 租约领取 Job：

  1. Worker 调用 claim_next 领取一个可执行 Job；
  2. 领取成功后写入 Worker ID 和租约到期时间；
  3. 执行期间按照租约时长的三分之一周期续租；
  4. 处理成功后标记 succeeded；
  5. 普通失败按照 min(30, 2 ** attempt) 秒进行有界指数退避；
  6. Worker 崩溃后租约不再续期，其他 Worker 可以接管过期 Job。

对应测试会让两个 Worker 并发争抢同一个过期租约，并验证最终只有一个 Worker 接管。这样既允许失败恢复，也避免两个 Worker 同时执行同一逻辑任务。

这也解释了为什么不能在 Webhook 里使用裸 asyncio.create_task()：请求结束或进程退出后，内存任务没有可恢复的事实记录。持久 Job 和数据库租约才把“接收告警”与“执行诊断”可靠地解耦开。

9. 总结：自动告警真正闭环了什么？

回到开头的问题：连接池耗尽以后，Agent 为什么能够自动开始工作？

不是因为 Agent 一直在扫描所有系统，也不是因为 Alertmanager 直接调用了某个智能体方法，而是因为系统建立了一条职责清晰、失败语义明确的事件链：

运行时指标
→ Prometheus 持续条件判断
→ Alertmanager 通知生命周期管理
→ Nginx/FastAPI 安全接入
→ PostgreSQL 原子持久化和最终幂等
→ 租约 Worker 可靠领取
→ AIOps Agent 启动

这条链最重要的设计不是“组件足够多”，而是任何一步失败时都能回答三个问题：事件是否已经持久化？发送方是否应该重投？是否可能重复启动 Agent？

当这三个问题有确定答案，自动报警才不再是一次脆弱的 Webhook 调用，而是一个可以审计、恢复和继续演进的 AIOps 入口。。

陈涛 · Agent Application Developer

杭州 · 2026