脚本如何汇总查询告警记录

wen 实用脚本 2026-06-27 27

自动化运维的实战指南

目录导读

为什么需要脚本汇总告警记录？ —— 运维场景痛点与价值分析
脚本汇总的核心架构 —— 数据源、处理逻辑、输出形式
主流实现方案对比 —— Shell、Python、Ansible、PromQL 适用场景
四步实战：用Python脚本汇总告警记录 —— 从API采集到邮件推送
常见问题与解答 —— 权限、性能、断点续传、多系统集成
SEO优化建议 —— 标题、关键词、结构化数据与多媒体嵌入

为什么需要脚本汇总告警记录？

运维场景痛点

在复杂的IT基础设施中,告警系统可能来自Zabbix、Prometheus、Grafana、云监控（如阿里云、腾讯云）或自研监控平台，手动登录不同控制台查询告警记录，会面临以下问题：

脚本如何汇总查询告警记录

效率低下：工程师需要花费30%以上的工作时间切换多个平台。
遗漏风险：告警阈值调整、重复告警、误报导致关键事件被淹没。
审计难题：合规审计需要提供指定时间段内的完整告警记录，手工导出格式杂乱。

脚本汇总的价值

通过编写脚本（Shell/Python/Go）统一调用API或读取日志文件，可以实现：

自动汇总24小时内的告警记录，去重、分级、归类。
输出结构化表格（CSV/Excel/JSON）或发送至企业微信、钉钉、邮件。
支持定时任务（Cron/Windows任务计划）实现每日早报。

问答环节
Q：小型团队是否需要脚本汇总？
A：即使只有5台服务器，手动查询告警记录每月浪费约10小时，一个简单的Python脚本仅需2小时开发，后续每月节省8小时。脚本汇总是将重复劳动自动化的最佳起点。

脚本汇总的核心架构

一个完整的脚本汇总查询系统包含三个层次：

层级	组件	常见实现
数据采集层	监控系统API/日志接口	requests调用HTTP API、subprocess执行CLI命令
处理层	数据清洗、去重、分级	pandas、awk/sed、正则表达式
输出层	文件/消息/数据库	CSV写入、SMTP邮件、Webhook推送

设计原则：

幂等性：脚本反复执行不产生重复告警记录。
增量采集：仅查询上次执行时间之后的告警，降低API压力。
异常处理：API超时、权限失败时记录错误日志，不影响已有数据。

主流实现方案对比

方案A：Shell + curl/jq（轻量级）

适用于Linux环境,快速从RESTful API抓取JSON格式告警。

#!/bin/bash
# 从Prometheus Alertmanager获取当前告警
curl -s -u "admin:password" 'http://alertmanager:9093/api/v2/alerts' | jq '.'

优点：无需安装额外解释器，适合嵌入系统。
缺点：复杂逻辑（去重、多表连接）难以实现。

方案B：Python + requests + pandas（通用型）

适合多数场景,尤其需要数据分析或输出Excel。

import requests, pandas as pd
response = requests.get('https://monitor.example.com/api/alerts', headers={'Authorization': 'Bearer token'})
df = pd.DataFrame(response.json())
df.to_csv('alerts_daily.csv')

优点：生态丰富，可对接数据库、消息队列。
缺点：需Python环境，依赖包管理。

方案C：Ansible Playbook（配置管理场景）

如果已经使用Ansible管理服务器,可以直接用模板引擎渲染告警统计。

- name: 收集host告警
  uri:
    url: "http://{{ item }}:9090/api/v1/alerts"
    return_content: yes
  loop: "{{ groups['monitors'] }}"

方案D：PromQL + Grafana API（实时查询）

直接通过Prometheus HTTP API查询历史告警。

/api/v1/query?query=ALERTS{severity="critical"}

四步实战：用Python脚本汇总告警记录

步骤1：定义数据源

假设我们使用Prometheus Alertmanager，API返回格式如下：

{
  "alerts": [
    {
      "status": "firing",
      "labels": {"alertname": "HighCPU", "severity": "critical"},
      "startsAt": "2025-03-01T10:00:00Z",
      "endsAt": "0001-01-01T00:00:00Z",
      "annotations": {"summary": "CPU > 90%"}
    }
  ]
}

步骤2：编写采集函数

import requests, datetime, logging
from typing import List, Dict
def fetch_alerts(since_time: str) -> List[Dict]:
    url = "https://alertmanager.example.com/api/v2/alerts"
    params = {"since": since_time}  # 增量查询
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        return resp.json().get("alerts", [])
    except Exception as e:
        logging.error(f"API请求失败: {e}")
        return []

步骤3：数据清洗与汇总

def process_alerts(alerts: List[Dict]) -> pd.DataFrame:
    records = []
    for alert in alerts:
        records.append({
            "alertname": alert["labels"]["alertname"],
            "severity": alert["labels"].get("severity", "unknown"),
            "status": alert["status"],
            "start_time": alert["startsAt"],
            "summary": alert["annotations"].get("summary", ""),
            "source": alert["labels"].get("instance", "unknown")
        })
    df = pd.DataFrame(records)
    # 去重：相同alertname+instance+start_time视为重复
    df.drop_duplicates(subset=['alertname', 'source', 'start_time'], inplace=True)
    # 按严重级别排序
    severity_order = {"critical": 0, "warning": 1, "info": 2}
    df['severity_order'] = df['severity'].map(severity_order)
    df.sort_values('severity_order', inplace=True)
    return df

步骤4：输出与推送

def send_report(df: pd.DataFrame):
    # 生成CSV文件
    filename = f"alerts_{datetime.date.today()}.csv"
    df.to_csv(filename, index=False)
    # 发送邮件（示例使用smtplib）
    import smtplib
    from email.mime.multipart import MIMEMultipart
    from email.mime.base import MIMEBase
    # ... 此处省略具体邮件配置，可参考常见smtp封装

完整脚本建议存放在Git仓库，配合CI/CD或定时任务运行。