📚 课程目录第3章告警模拟基础

3. 告警模拟基础：定义告警数据结构、使用Python生成模拟告警JSON

告警模拟是Runbook自动化单元测试的起点。只有构造出结构完整、字段真实的模拟告警，后续的触发匹配、自愈动作验证才有可靠的输入。本章将定义一套标准化的告警数据结构，并给出Python生成模拟告警JSON的完整方法。

3.1 告警数据结构的核心字段

一个通用的告警数据结构应包含标识、来源、严重度、时间、上下文和状态六大类信息。下表列出核心字段及其说明：

分类	字段名	类型	说明
标识	`alert_id`	string	全局唯一告警ID，如UUID
来源	`source`	string	告警产生系统，如Prometheus、Zabbix
来源	`monitor_name`	string	监控项名称，如CPU使用率
严重度	`severity`	string	枚举值：critical / warning / info
时间	`started_at`	datetime	告警触发时间，ISO 8601格式
上下文	`labels`	dict	标签键值对，如环境、区域、实例ID
上下文	`annotations`	dict	可读描述，如当前值、阈值、建议操作
状态	`status`	string	firing / resolved

设计原则：字段应保持最小完备。labels用于路由匹配，annotations用于人工阅读和Runbook参数注入。避免将动态数据（如当前值）放在固定字段中，统一放入annotations。

3.2 告警数据结构的JSON Schema

为保障模拟告警的合法性，建议定义JSON Schema进行校验。以下是一个精简但完整的Schema定义：

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "AlertEvent",
  "type": "object",
  "required": ["alert_id", "source", "severity", "started_at", "status"],
  "properties": {
    "alert_id": {
      "type": "string",
      "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$"
    },
    "source": {
      "type": "string",
      "enum": ["Prometheus", "Zabbix", "CloudWatch", "自定义"]
    },
    "monitor_name": { "type": "string" },
    "severity": {
      "type": "string",
      "enum": ["critical", "warning", "info"]
    },
    "started_at": {
      "type": "string",
      "format": "date-time"
    },
    "labels": {
      "type": "object",
      "additionalProperties": { "type": "string" }
    },
    "annotations": {
      "type": "object",
      "additionalProperties": { "type": "string" }
    },
    "status": {
      "type": "string",
      "enum": ["firing", "resolved"]
    }
  }
}

3.3 使用Python生成模拟告警JSON

基于上述数据结构，我们可以编写一个Python生成器，支持随机生成、参数化定制和批量输出。核心实现如下：

import uuid
import json
import random
from datetime import datetime, timezone, timedelta

class AlertGenerator:
    """模拟告警生成器"""

    SOURCES = ["Prometheus", "Zabbix", "CloudWatch"]
    SEVERITIES = ["critical", "warning", "info"]
    MONITORS = {
        "CPU使用率": {"threshold": 90, "unit": "%"},
        "内存使用率": {"threshold": 85, "unit": "%"},
        "磁盘IO延迟": {"threshold": 200, "unit": "ms"},
        "HTTP 5xx错误率": {"threshold": 5, "unit": "%"},
    }

    @staticmethod
    def generate_one(severity=None, source=None, status="firing"):
        """生成单条模拟告警"""
        monitor_name = random.choice(list(AlertGenerator.MONITORS.keys()))
        current_value = random.randint(
            AlertGenerator.MONITORS[monitor_name]["threshold"] - 10,
            AlertGenerator.MONITORS[monitor_name]["threshold"] + 20
        )

        alert = {
            "alert_id": str(uuid.uuid4()),
            "source": source or random.choice(AlertGenerator.SOURCES),
            "monitor_name": monitor_name,
            "severity": severity or random.choice(AlertGenerator.SEVERITIES),
            "started_at": datetime.now(timezone.utc).isoformat(),
            "labels": {
                "env": random.choice(["prod", "staging", "dev"]),
                "region": random.choice(["us-east-1", "ap-southeast-2", "eu-west-1"]),
                "instance": f"i-{uuid.uuid4().hex[:8]}"
            },
            "annotations": {
                "current_value": str(current_value),
                "threshold": str(AlertGenerator.MONITORS[monitor_name]["threshold"]),
                "unit": AlertGenerator.MONITORS[monitor_name]["unit"],
                "summary": f"{monitor_name} 当前值 {current_value}，超过阈值"
            },
            "status": status
        }
        return alert

    @staticmethod
    def generate_batch(count=10, **kwargs):
        """批量生成模拟告警"""
        return [AlertGenerator.generate_one(**kwargs) for _ in range(count)]

# 使用示例：生成3条critical级别告警并输出JSON
if __name__ == "__main__":
    alerts = AlertGenerator.generate_batch(count=3, severity="critical")
    print(json.dumps(alerts, indent=2, ensure_ascii=False))

输出示例（单条）：

{
  "alert_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "source": "Prometheus",
  "monitor_name": "CPU使用率",
  "severity": "critical",
  "started_at": "2025-03-21T14:30:00+00:00",
  "labels": {
    "env": "prod",
    "region": "ap-southeast-2",
    "instance": "i-1a2b3c4d"
  },
  "annotations": {
    "current_value": "95",
    "threshold": "90",
    "unit": "%",
    "summary": "CPU使用率 当前值 95，超过阈值"
  },
  "status": "firing"
}

3.4 告警模拟的完整流程

下图展示了从定义数据结构到生成模拟告警，再到注入Runbook测试管线的完整流程：

3.5 参数化定制与场景覆盖

在实际测试中，需要覆盖多种告警场景。下表列出常见的参数化维度及其用途：

参数维度	典型取值	测试场景
severity	critical / warning / info	验证不同严重度的路由与响应策略
source	Prometheus / Zabbix / CloudWatch	验证多监控源适配器解析逻辑
status	firing / resolved	验证告警触发与恢复的完整生命周期
labels.env	prod / staging / dev	验证环境隔离与告警抑制规则
annotations.current_value	阈值附近、远超阈值、波动值	验证自愈动作的触发条件判断

常见陷阱：模拟告警中的时间字段（started_at）必须使用UTC时间，避免时区转换导致的测试断言失败。另外，labels中的值应全部为字符串类型，即使表示数字（如"95"），否则JSON Schema校验会报错。

3.6 模拟告警的校验与持久化

生成模拟告警后，建议立即进行Schema校验，并将合法数据持久化到文件或测试夹具中，便于后续测试用例复用。

import jsonschema
from jsonschema import validate

# 加载Schema
with open("alert_schema.json", "r") as f:
    schema = json.load(f)

# 生成并校验
alerts = AlertGenerator.generate_batch(count=5, severity="warning")
for alert in alerts:
    try:
        validate(instance=alert, schema=schema)
        print(f"✅ {alert['alert_id']} 校验通过")
    except jsonschema.ValidationError as e:
        print(f"❌ {alert['alert_id']} 校验失败: {e.message}")

# 持久化到文件
with open("mock_alerts.json", "w") as f:
    json.dump(alerts, f, indent=2, ensure_ascii=False)