4. 告警源对接——Prometheus Alertmanager：Webhook 接收与解析实战

说到告警自动化，第一步总是最关键的。你得先让系统“听得见”告警。

Prometheus Alertmanager 的 Webhook 机制，说白了就是一个 HTTP 回调。出问题时，Alertmanager 会把告警信息打包成 JSON，发到你指定的 URL。

这一章，我们就亲手把这个接收端搭起来。解析 JSON、验签，一步到位。

4.1 你的第一个 Webhook 接收端点

用什么写？我个人习惯用 Flask，轻量、快速、适合这活儿。

我们不搞花架子，直接写一个最简的接收端点：

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/webhook', methods=['POST'])
def webhook_receiver():
    # 接收 Alertmanager 的 POST 请求
    data = request.get_json()
    if not data:
        return jsonify({"status": "error", "message": "No JSON payload"}), 400

    # 打印出来，先看看长什么样
    print(f"收到告警：{data.get('groupKey')}")
    print(f"告警数量：{len(data.get('alerts', []))}")

    return jsonify({"status": "ok"}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

这里有个细节：request.get_json() 必须是 POST 请求，而且 Content-Type 要带 application/json。Alertmanager 默认就是 JSON 格式，所以不会有问题。但我在项目中遇到过一种情况——对方把 JSON 字符串包成了纯文本，结果 get_json() 直接返回 None。嗯，这时候你得用 request.data 手动解析。

4.2 拆解 Alertmanager 的 JSON 结构

搞明白数据长什么样，比直接写逻辑更重要。你想想看，连告警字段都不清楚，后面怎么判断严重等级、怎么分组？

Alertmanager 发来的 JSON 结构如下：

再看一个具体的告警对象（alerts[0]）内部的字段：

4.3 实战解析：提取你需要的信息

来，我们写一个更完整的解析函数。它会把告警里的关键字段抽出来，转成内部统一的结构。这样后面的处置逻辑就不用关心原始 JSON 长什么样了。

def parse_alertmanager_payload(data):
    parsed_alerts = []
    group_key = data.get('groupKey', '')
    status = data.get('status', 'unknown')

    for alert in data.get('alerts', []):
        labels = alert.get('labels', {})
        annotations = alert.get('annotations', {})

        parsed_alert = {
            'fingerprint': alert.get('fingerprint'),
            'status': alert.get('status', status),
            'alertname': labels.get('alertname', 'unknown'),
            'severity': labels.get('severity', 'info'),
            'instance': labels.get('instance', ''),
            'summary': annotations.get('summary', ''),
            'description': annotations.get('description', ''),
            'starts_at': alert.get('startsAt'),
            'generator_url': alert.get('generatorURL'),
            'group_key': group_key
        }
        parsed_alerts.append(parsed_alert)

    return parsed_alerts

为什么要把 group_key 塞进每个告警里？我个人的习惯是保留上下文，后面做聚合或关联分析时，能直接追溯到它属于哪个组。你想想看，一个进程挂了的告警，和它相关的网络告警可能在同一个组里，没有 group_key 你怎么关联？

4.4 验证签名：防伪造的关键一步

接收 Webhook 很简单，但你怎么确定这个请求真的来自你自己的 Alertmanager，而不是某个顺手扫端口的人发的垃圾？

Alertmanager 官方支持配置一个 secret，每次发 Webhook 时会在 HTTP 头里带上签名。我们只用把这个头和本地计算的签名比对一下就行。

这里是我用过的签名验证方案：

import hmac
import hashlib

SECRET_KEY = "your-secret-key-here"  # 和 Alertmanager 里配置的一致

def verify_signature(request):
    # 从 Header 里拿签名
    received_sig = request.headers.get('X-Alertmanager-Signature')
    if not received_sig:
        return False

    # 计算本地签名
    payload = request.get_data()
    expected_sig = hmac.new(
        SECRET_KEY.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    # 安全比对，防止时序攻击
    return hmac.compare_digest(received_sig, expected_sig)

然后在接收端点里这样用：

@app.route('/webhook', methods=['POST'])
def webhook_receiver():
    if not verify_signature(request):
        return jsonify({"status": "error", "message": "Invalid signature"}), 403

    data = request.get_json()
    alerts = parse_alertmanager_payload(data)
    # 后面的处置逻辑...
    return jsonify({"status": "ok"}), 200

我曾经在线上调试时，信誓旦旦地加了签名校验，结果忘了两边配一样的 secret。前端告警发过来，全部 403 拦截。嗯，排查了快一个小时才反应过来——原来 secret 里有个尾随的空格。这种坑，踩过一次你就永远不会忘了。

4.5 整体工作流程

我把这一章的核心流程整理成了一张图，你一看就明白了：

4.6 把这些串起来：一个能上线的接收端

有了解析函数，有了签名验证，下面把它们组装起来就是水到渠成的事。我通常会再加一个简单的日志记录，方便排查。

import logging
from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)
SECRET_KEY = "your-secret-key-here"
logging.basicConfig(level=logging.INFO)

def verify_signature(request):
    received_sig = request.headers.get('X-Alertmanager-Signature')
    if not received_sig:
        return False
    payload = request.get_data()
    expected_sig = hmac.new(
        SECRET_KEY.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(received_sig, expected_sig)

def parse_alertmanager_payload(data):
    parsed_alerts = []
    group_key = data.get('groupKey', '')
    for alert in data.get('alerts', []):
        labels = alert.get('labels', {})
        annotations = alert.get('annotations', {})
        parsed_alerts.append({
            'fingerprint': alert.get('fingerprint'),
            'alertname': labels.get('alertname', 'unknown'),
            'severity': labels.get('severity', 'info'),
            'instance': labels.get('instance', ''),
            'summary': annotations.get('summary', ''),
            'description': annotations.get('description', ''),
            'starts_at': alert.get('startsAt'),
            'group_key': group_key
        })
    return parsed_alerts

@app.route('/webhook', methods=['POST'])
def webhook_receiver():
    if not verify_signature(request):
        app.logger.warning("签名验证失败，丢弃请求")
        return jsonify({"status": "error", "message": "Invalid signature"}), 403

    data = request.get_json()
    alerts = parse_alertmanager_payload(data)
    app.logger.info(f"收到 {len(alerts)} 条告警，来源组：{data.get('groupKey')}")

    # 这里你可以把 alerts 传给后续的过滤、路由、通知模块
    # ...

    return jsonify({"status": "ok", "alert_count": len(alerts)}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

我建议你先把这段代码部署到一个测试环境，然后用 curl 模拟 Alertmanager 发几条告警试试。相信我，当你看到终端老老实实打印出解析后的告警信息时，那种成就感是很实在的。