2、告警源接入：常见告警源的接入方式与协议解析

告警源接入，说白了就是让我们的告警平台能听懂各个监控系统在说什么。我刚开始做运维那会儿，公司里监控系统五花八门，Zabbix、Prometheus、ELK各玩各的，告警信息散落在不同群里，值班同学每天光翻消息就得花半小时。后来我们统一做了告警源接入，才算是把散落的拼图拼到了一起。

这一章，我就带你看看几种主流告警源的接入方式和协议解析。嗯，这里要注意，不同告警源的协议差异很大，但核心思路是一致的——把异构的告警数据，转换成统一的告警事件模型。

2.1 告警源接入的核心思路

先说说我的个人习惯。我一般把告警源接入分成三步：

  1. 接收：通过Webhook、API轮询或Agent方式，拿到原始告警数据
  2. 解析：把JSON、XML或自定义格式的数据，提取出关键字段
  3. 标准化：映射到统一的告警模型（告警名称、级别、时间、来源、详情等）

你想想看，如果每个告警源都单独写一套处理逻辑，那维护成本得多高？所以标准化这一步，是绝对不能省的。

核心告警模型字段：

  • alert_id：告警唯一ID
  • alert_name：告警名称
  • severity：级别（critical/warning/info）
  • status：状态（firing/resolved）
  • source：来源系统
  • timestamp：触发时间
  • description：描述信息
  • labels：标签（主机、服务、地域等）

2.2 Zabbix 接入方式

Zabbix 是我最早接触的监控系统，说实话它的告警机制有点「老派」，但胜在稳定。我在项目中遇到过不少团队还在用 Zabbix 3.x 版本，接入时要注意版本差异。

2.2.1 Webhook 方式（推荐）

Zabbix 从 4.4 版本开始支持 Webhook 类型的告警媒介。配置起来其实不复杂：

# Zabbix Webhook 配置示例（JavaScript脚本）
# 在 Zabbix 前端：管理 → 告警媒介类型 → 创建

var params = JSON.parse(value);
var webhook_url = params.HOOK_URL;

var payload = {
    "alert_id": "{EVENT.ID}",
    "alert_name": "{TRIGGER.NAME}",
    "severity": "{TRIGGER.SEVERITY}",
    "status": "{EVENT.STATUS}",
    "host": "{HOST.NAME}",
    "timestamp": "{EVENT.DATE} {EVENT.TIME}",
    "description": "{TRIGGER.DESCRIPTION}"
};

// 发送HTTP POST请求
HttpClient.post(webhook_url, JSON.stringify(payload), {
    "Content-Type": "application/json"
});

小技巧：Zabbix 的宏变量里，{TRIGGER.SEVERITY} 返回的是数字（0-5），我一般会在脚本里做个映射，转成 readable 的级别名称，比如 3 对应 "warning"。

2.2.2 数据库轮询方式

有些老版本 Zabbix 不支持 Webhook，那就只能走数据库轮询了。我曾经维护过一个 Zabbix 3.2 的环境，就是靠定时任务查数据库来拿告警的。

# Python 示例：轮询 Zabbix 数据库
import pymysql
import time

def poll_zabbix_alerts():
    conn = pymysql.connect(host='zabbix-db', user='zabbix', password='xxx', db='zabbix')
    cursor = conn.cursor()
    
    # 查询最近5分钟内的告警
    sql = """
    SELECT e.eventid, t.description, t.severity, e.clock, h.name
    FROM events e
    JOIN triggers t ON e.objectid = t.triggerid
    JOIN hosts h ON t.hostid = h.hostid
    WHERE e.clock > UNIX_TIMESTAMP(NOW() - INTERVAL 5 MINUTE)
    AND e.source = 0
    """
    cursor.execute(sql)
    rows = cursor.fetchall()
    
    for row in rows:
        alert = {
            "alert_id": row[0],
            "alert_name": row[1],
            "severity": map_severity(row[2]),
            "timestamp": row[3],
            "host": row[4]
        }
        # 发送到告警平台
        send_to_alert_platform(alert)
    
    cursor.close()
    conn.close()

注意：数据库轮询方式对 Zabbix 数据库有压力，轮询间隔不要太短，我建议至少 30 秒以上。另外，要处理好已恢复告警的标记，避免重复推送。

2.3 Prometheus 接入方式

Prometheus 的告警机制比 Zabbix 现代得多，它通过 Alertmanager 统一管理告警。说白了，Alertmanager 就是 Prometheus 生态里的「告警路由器」。

2.3.1 Alertmanager Webhook Receiver

这是最标准的接入方式。Alertmanager 原生支持 webhook_config，配置起来非常简洁：

# alertmanager.yml 配置
route:
  receiver: 'my-webhook'
  group_wait: 10s
  group_interval: 5m
  repeat_interval: 4h

receivers:
- name: 'my-webhook'
  webhook_configs:
  - url: 'http://your-alert-platform:8080/webhook/prometheus'
    send_resolved: true
    http_config:
      basic_auth:
        username: 'admin'
        password: 'password123'

Prometheus 发过来的告警数据格式是固定的 JSON，我贴一个典型的 payload：

{
  "receiver": "my-webhook",
  "status": "firing",
  "alerts": [
    {
      "status": "firing",
      "labels": {
        "alertname": "HighCpuUsage",
        "instance": "192.168.1.100:9090",
        "job": "node_exporter",
        "severity": "critical"
      },
      "annotations": {
        "summary": "CPU usage above 90%",
        "description": "Instance 192.168.1.100 CPU usage is 95%"
      },
      "startsAt": "2024-01-15T10:30:00Z",
      "endsAt": "0001-01-01T00:00:00Z",
      "generatorURL": "http://prometheus:9090/graph?g0..."
    }
  ],
  "groupLabels": {"alertname": "HighCpuUsage"},
  "commonLabels": {"severity": "critical"},
  "commonAnnotations": {"summary": "CPU usage above 90%"},
  "externalURL": "http://alertmanager:9093"
}

解析要点：

  • status 字段：firing 表示触发，resolved 表示恢复
  • labels 里的 alertname 是告警规则名称，severity 是级别
  • annotations 里通常放人类可读的描述信息
  • startsAt 和 endsAt 是时间戳，注意 endsAt 为 0001-01-01 时表示还在触发中

2.3.2 自定义 Exporter 方式

有些场景下，我们需要把非 Prometheus 体系的告警也接入进来。我做过一个方案：写一个自定义 Exporter，把其他系统的告警数据暴露成 Prometheus metrics，然后通过 Prometheus 自身的告警规则来触发。

# Python 自定义 Exporter 示例
from prometheus_client import start_http_server, Gauge
import time

# 定义告警指标
alert_gauge = Gauge('custom_alert', 'Custom alert from legacy system',
                    ['alert_name', 'severity', 'source'])

def update_alerts():
    # 从遗留系统获取告警
    alerts = fetch_from_legacy_system()
    for alert in alerts:
        alert_gauge.labels(
            alert_name=alert['name'],
            severity=alert['severity'],
            source='legacy'
        ).set(1 if alert['status'] == 'firing' else 0)

if __name__ == '__main__':
    start_http_server(8000)
    while True:
        update_alerts()
        time.sleep(30)

2.4 ELK（Elasticsearch）接入方式

ELK 本身不是专门的告警系统，但很多人用它做日志告警。我记得有一次线上故障，就是靠 ELK 的日志告警发现的——某个服务的错误日志在 5 分钟内暴增了 100 倍。

2.4.1 Watcher 告警（X-Pack）

Elasticsearch 的 Watcher 功能可以基于查询结果触发告警。配置一个 Webhook 动作：

# Elasticsearch Watcher 配置示例
PUT _watcher/watch/log_error_watch
{
  "trigger": {
    "schedule": {
      "interval": "5m"
    }
  },
  "input": {
    "search": {
      "request": {
        "indices": ["logs-*"],
        "body": {
          "query": {
            "bool": {
              "filter": [
                {"range": {"@timestamp": {"gte": "now-5m"}}},
                {"term": {"level": "ERROR"}}
              ]
            }
          },
          "aggs": {
            "error_count": {"value_count": {"field": "_index"}}
          }
        }
      }
    }
  },
  "condition": {
    "compare": {
      "ctx.payload.aggregations.error_count.value": {"gte": 100}
    }
  },
  "actions": {
    "webhook": {
      "webhook": {
        "method": "POST",
        "host": "your-alert-platform",
        "port": 8080,
        "path": "/webhook/elk",
        "body": "{\"alert_name\":\"LogErrorSpike\",\"count\":{{ctx.payload.aggregations.error_count.value}},\"timestamp\":\"{{ctx.trigger.scheduled_time}}\"}"
      }
    }
  }
}

避坑指南：Watcher 的 body 里用 Mustache 模板语法，注意转义双引号。我曾经在这里踩过坑，少转义了一个引号，导致 Webhook 一直发不出去。

2.4.2 ElastAlert 方案（开源）

如果没买 X-Pack，可以用 ElastAlert 这个开源工具。它支持多种告警规则类型，配置起来也很灵活：

# ElastAlert 配置文件示例
# config.yaml
rules_folder: rules
run_every:
  minutes: 1
buffer_time:
  minutes: 5
es_host: elasticsearch.example.com
es_port: 9200
writeback_index: elastalert_status

# rules/error_rate.yaml
name: "High Error Rate"
type: frequency
index: logs-*
num_events: 50
timeframe:
  minutes: 5
filter:
- query:
    query_string:
      query: "level: ERROR"
alert:
- "command"
command: ["curl", "-X", "POST", "-d", "{\"alert_name\":\"HighErrorRate\"}", "http://alert-platform:8080/webhook/elastalert"]

2.5 云监控接入方式

云监控这块，各家有各家的玩法。我以阿里云和腾讯云为例，说说通用的接入思路。

2.5.1 阿里云 CMS

阿里云支持通过 Webhook 回调来推送告警。配置路径：云监控 → 告警规则 → 通知方式 → 选择 Webhook。

# 阿里云 Webhook 回调示例 payload
{
  "alertName": "CPU使用率过高",
  "alertState": "ALERT",
  "curValue": "95.0",
  "dimensions": {
    "instanceId": "i-xxxxx",
    "userId": "123456"
  },
  "expression": "Average >= 90",
  "metricName": "cpu_utilization",
  "namespace": "acs_ecs",
  "region": "cn-hangzhou",
  "timestamp": "2024-01-15 10:30:00"
}

2.5.2 腾讯云 CM

腾讯云的告警回调格式略有不同，但核心字段大同小异：

# 腾讯云 Webhook 回调示例 payload
{
  "sessionId": "xxxxx",
  "alarmId": "alarm-xxxxx",
  "alarmStatus": "1",  // 1:触发 0:恢复
  "alarmType": "metric",
  "alarmObjInfo": {
    "region": "ap-guangzhou",
    "namespace": "qce/cvm",
    "dimensions": {
      "InstanceId": "ins-xxxxx"
    }
  },
  "alarmPolicyName": "CPU告警",
  "alarmPolicyType": "cvm",
  "alarmLevel": "2",  // 1:致命 2:严重 3:警告 4:提醒
  "metricName": "CPUUsage",
  "metricValue": "95.0",
  "occurTime": "2024-01-15 10:30:00"
}

云监控接入要点：

  • 云厂商的告警格式差异主要在字段命名上，建议写一个适配层做字段映射
  • 注意云监控的告警恢复通知，有些云厂商的恢复通知格式和触发通知不一样
  • 云监控的 Webhook 地址需要公网可达，如果告警平台在内网，可以考虑用云函数做中转

2.6 告警源接入架构图

下面这张图是我自己总结的告警源接入架构，你可以参考一下：

告警源接入架构图 告警源 Zabbix Prometheus ELK 云监控 接入层（协议解析与适配） Webhook/DB轮询 Alertmanager Webhook Watcher/ElastAlert 云厂商Webhook 标准化层（统一告警模型） 字段映射 · 级别转换 · 去重 · 富化 告警平台（统一处理与分发）

2.7 接入协议解析要点

最后，我总结一下接入协议解析的几个关键点，这些都是我在实际项目中踩过的坑：

协议类型 常见格式 解析难点 我的建议
HTTP Webhook JSON / XML 字段命名不统一，嵌套层级深 写一个通用的 JSON Path 提取器
数据库轮询 SQL 结果集 数据库压力，增量数据识别 用时间戳或自增ID做增量标记
Agent 推送 自定义二进制/文本 协议不标准，需要定制解析 优先转成 JSON 再处理
消息队列 Kafka/RabbitMQ 消息顺序，重复消费 做好幂等处理，记录 offset

重要提醒：不管用什么协议接入，一定要做好以下几点：

  • 超时处理：Webhook 请求设置 5 秒超时，避免阻塞
  • 重试机制：失败后至少重试 3 次，间隔递增
  • 日志记录：记录原始 payload，方便排查问题
  • 监控告警：接入层本身也要有监控，接入挂了要能发现

嗯，告警源接入这块，说白了就是「翻译」工作。把不同监控系统的语言，翻译成告警平台能理解的语言。只要把协议解析和标准化做好了，后面的事情就顺了。我个人习惯是先接一个告警源跑通全流程，再逐步扩展，这样风险最小。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →