Grafana告警系统深度定制从AlertManager到Grafana Alerting的迁移方案与高级配置技巧Grafana 从 8.0 版本开始内置了全新的 Grafana Alerting 系统与 Prometheus AlertManager 形成了新老交替的格局。作为运维工程师你需要的不只是知道怎么切换更需要理解迁移过程中的每个陷阱以及如何利用新系统的多维度告警路由能力构建一套真正智能的告警体系。一、两种告警架构的差异分析1.1 架构对比在开始迁移之前必须理解两种架构之间的根本差异。AlertManager 是 Prometheus 生态的独立组件而 Grafana Alerting 将告警定义、评估、通知全链路整合在 Grafana 内部。graph TB subgraph 传统方案: Prometheus AlertManager P1[Prometheusbr/抓取指标评估规则] --|告警触发| A1[AlertManagerbr/分组/抑制/静默/路由] A1 --|多渠道分发| N1[钉钉/企微/邮件/PagerDuty] A1 --|Webhook| E1[外部系统] end subgraph 新方案: Grafana Alerting D1[Grafana统一数据源br/Prometheus/Loki/InfluxDB] -- R1[告警规则引擎br/在Grafana内部评估] R1 --|告警状态| S1[告警状态管理器br/State Manager] S1 --|通知模板| NP1[通知策略br/Notification Policies] NP1 --|多渠道分发| N2[钉钉/企微/邮件/PagerDuty] NP1 --|Webhook| E2[外部系统] end style A1 fill:#F56C6C,color:#fff style R1 fill:#409EFF,color:#fff style S1 fill:#67C23A,color:#fff1.2 关键差异对照表维度AlertManagerGrafana Alerting规则定义位置Prometheus Rule文件Grafana UI / Provisioning规则语法PromQLPromQL 多数据源查询数据源支持Prometheus专用支持多种数据源告警分组基于标签的灵活分组基于Label Matcher的分组静默管理amtool silenceGrafana UI直接管理告警历史无持久化存储内置告警状态历史模板系统Go TemplateGo Template 内置变量二、迁移方案设计与执行2.1 迁移前的评估清单#!/bin/bash # Grafana Alerting迁移前置检查脚本 set -e echo echo Grafana Alerting 迁移前置检查 echo # 1. 检查Grafana版本(需要 9.0) GRAFANA_VERSION$(grafana-server -v 2/dev/null | grep -oP \d\.\d || echo 未知) echo echo [1] Grafana版本检查: $GRAFANA_VERSION if [ $(echo $GRAFANA_VERSION 9.0 | bc -l 2/dev/null) 1 ]; then echo [✓] 版本满足要求 else echo [✗] 版本过低建议升级到Grafana 10.x fi # 2. 检查当前AlertManager规则数量 echo echo [2] AlertManager告警规则统计 ALERTMANAGER_URL${ALERTMANAGER_URL:-http://localhost:9093} ALERTS_COUNT$(curl -s ${ALERTMANAGER_URL}/api/v2/alerts 2/dev/null | \ python3 -c import json,sys; djson.load(sys.stdin); print(len(d)) 2/dev/null || echo 0) echo 当前活跃告警数: $ALERTS_COUNT # 3. 统计Prometheus规则文件 echo echo [3] Prometheus规则文件清单 PROMETHEUS_RULES_DIR${PROMETHEUS_RULES_DIR:-/etc/prometheus/rules} if [ -d $PROMETHEUS_RULES_DIR ]; then find $PROMETHEUS_RULES_DIR -name *.yml -o -name *.yaml | while read -r rule_file; do rule_count$(grep -c alert: $rule_file 2/dev/null || echo 0) echo $rule_file: ${rule_count}条规则 done else echo 规则目录不存在: $PROMETHEUS_RULES_DIR fi # 4. 检查通知渠道 echo echo [4] 当前通知渠道检查 AM_CONFIG$(curl -s ${ALERTMANAGER_URL}/api/v2/status 2/dev/null | \ python3 -c import json,sys; djson.load(sys.stdin); print(d.get(config,{})) 2/dev/null || echo 无法获取) echo AlertManager配置状态: $AM_CONFIG echo echo [5] Grafana Plugin检查 echo 检查Grafana是否已安装所需通知渠道插件... grafana-cli plugins ls 2/dev/null | grep -E (dingtalk|wecom|pagerduty) || \ echo 未发现额外通知插件基础通知渠道已内置 echo echo echo 检查完成 echo 2.2 规则转换工具从 Prometheus Rule 文件转换到 Grafana Alerting Provisioning 格式#!/usr/bin/env python3 Prometheus告警规则 → Grafana Alerting Provisioning 格式转换器 import yaml import json import argparse from typing import Dict, List, Any from pathlib import Path class RuleConverter: Prometheus规则文件到Grafana Provisioning格式的转换器 支持转换: - Prometheus Rule Group → Grafana Alert Rule Group - 告警表达式保留 - 标签和注解迁移 def __init__(self, folder_uid: str migrated-alerts): Args: folder_uid: Grafana中存放转换后规则的文件夹UID self.folder_uid folder_uid self.errors: List[str] [] self.warnings: List[str] [] def convert_file(self, prom_rules_file: str) - Dict[str, Any]: 转换单个Prometheus规则文件 Args: prom_rules_file: Prometheus规则文件路径 Returns: Grafana Alerting Provisioning格式的字典 Raises: FileNotFoundError: 文件不存在时抛出 yaml.YAMLError: YAML解析失败时抛出 file_path Path(prom_rules_file) if not file_path.exists(): raise FileNotFoundError(f规则文件不存在: {prom_rules_file}) try: with open(file_path, r, encodingutf-8) as f: prom_data yaml.safe_load(f) except yaml.YAMLError as e: raise yaml.YAMLError(fYAML解析失败: {e}) if not prom_data or groups not in prom_data: self.errors.append(f文件 {prom_rules_file} 不包含有效的groups字段) return {} grafana_rules [] for group in prom_data.get(groups, []): group_name group.get(name, migrated-group) rules group.get(rules, []) for idx, rule in enumerate(rules): if not rule.get(alert): # 跳过recording rules self.warnings.append( f跳过recording rule: {rule.get(record, unknown)} ) continue grafana_rule self._convert_single_rule( rule, group_name, idx ) if grafana_rule: grafana_rules.append(grafana_rule) return { apiVersion: 1, groups: [ { orgId: 1, name: fmigrated-{file_path.stem}, folder: Migrated Alerts, folderUid: self.folder_uid, interval: 1m, rules: grafana_rules } ] } def _convert_single_rule( self, rule: Dict, group_name: str, index: int ) - Dict[str, Any]: 转换单条告警规则 alert_name rule.get(alert, falert-{index}) expr rule.get(expr, ) duration rule.get(for, 5m) labels rule.get(labels, {}) annotations rule.get(annotations, {}) if not expr: self.errors.append( f规则 {alert_name} 缺少expr字段已跳过 ) return {} # 安全地转换duration字符串为秒 duration_seconds self._parse_duration(duration) grafana_rule { uid: fmigrated-{alert_name}-{index}, title: alert_name, condition: A, data: [ { refId: A, relativeTimeRange: { from: 600, to: 0 }, datasourceUid: prometheus, model: { expr: expr, intervalMs: 1000, maxDataPoints: 43200, refId: A } } ], noDataState: NoData, execErrState: Error, for: f{duration_seconds}s, annotations: annotations, labels: labels, isPaused: False # 迁移后默认启用 } return grafana_rule staticmethod def _parse_duration(duration_str: str) - int: 解析Prometheus风格的duration字符串 Args: duration_str: 如 5m, 1h, 30s, 2d Returns: 秒数 if not duration_str: return 300 # 默认5分钟 duration_str duration_str.strip() multipliers { s: 1, m: 60, h: 3600, d: 86400, w: 604800 } # 去除末尾字母获取数字部分 unit number_part for i, char in enumerate(duration_str): if char.isdigit() or char .: number_part char else: unit duration_str[i:] break try: value float(number_part) if number_part else 0 except ValueError: return 300 for suffix, multiplier in multipliers.items(): if suffix in unit: return int(value * multiplier) return 300 def get_report(self) - Dict: 获取转换报告 return { errors: self.errors, warnings: self.warnings, error_count: len(self.errors), warning_count: len(self.warnings) } def main(): parser argparse.ArgumentParser( descriptionPrometheus规则 → Grafana Alerting转换器 ) parser.add_argument( input_file, helpPrometheus规则文件路径 ) parser.add_argument( -o, --output, defaultgrafana-alerts-provisioning.yaml, help输出文件路径 ) parser.add_argument( --folder-uid, defaultmigrated-alerts, helpGrafana文件夹UID ) args parser.parse_args() try: converter RuleConverter(folder_uidargs.folder_uid) result converter.convert_file(args.input_file) if not result: print(转换失败未生成有效输出) report converter.get_report() for err in report[errors]: print(f 错误: {err}) return 1 with open(args.output, w, encodingutf-8) as f: yaml.dump(result, f, allow_unicodeTrue, default_flow_styleFalse) print(f转换完成: {args.output}) report converter.get_report() print(f错误: {report[error_count]}, 警告: {report[warning_count]}) if report[warnings]: print(\n警告详情:) for warn in report[warnings]: print(f - {warn}) return 0 except Exception as e: print(f转换过程中发生异常: {e}) return 1 if __name__ __main__: exit(main())三、通知策略高级配置3.1 多维标签路由Grafana Alerting 的通知策略Notification Policies支持基于标签的树状路由可以实现比 AlertManager 更精细的通知控制# grafana-notification-policies.yaml # Grafana Alerting通知策略Provisioning配置 apiVersion: 1 # 默认通知策略兜底路由 policies: - orgId: 1 receiver: default-email group_by: [alertname, severity] group_wait: 30s group_interval: 5m repeat_interval: 4h # 子策略按severity标签分支路由 routes: # P0级别立即通知 企业微信 PagerDuty - receiver: critical-oncall group_wait: 10s # P0只等10秒就发送 group_interval: 1m # 每分钟重复 repeat_interval: 5m # 每5分钟重复 matchers: - severity critical # P1级别邮件 企业微信群 - receiver: warning-team group_wait: 30s group_interval: 5m repeat_interval: 1h matchers: - severity warning # 按团队标签路由 - receiver: platform-team matchers: - team platform - severity critical - receiver: data-team matchers: - team data - severity ! info # 通知渠道Contact Points定义 contactPoints: - orgId: 1 name: critical-oncall receivers: - uid: wecom-critical type: wecom settings: url: ${WECOM_WEBHOOK_CRITICAL} messageType: markdown - uid: pagerduty-critical type: pagerduty settings: integrationKey: ${PD_INTEGRATION_KEY} severity: critical - name: warning-team receivers: - uid: wecom-warning type: wecom settings: url: ${WECOM_WEBHOOK_WARNING} messageType: markdown - name: platform-team receivers: - uid: slack-platform type: slack settings: url: ${SLACK_WEBHOOK_URL} channel: #platform-alerts - name: data-team receivers: - uid: email-data type: email settings: addresses: data-engexample.com subject: [告警] {{ .GroupLabels.alertname }}3.2 通知模板自定义{{/* Grafana Alerting 自定义通知模板 */}} {{ define wecom.message }} {{ if eq .Status firing }} ## 告警触发 {{ else }} ## 告警恢复 {{ end }} **告警名称**: {{ .CommonLabels.alertname }} **严重级别**: {{ .CommonLabels.severity }} **触发时间**: {{ .StartsAt.Format 2006-01-02 15:04:05 }} {{ if eq .Status resolved }} **恢复时间**: {{ .EndsAt.Format 2006-01-02 15:04:05 }} {{ end }} {{ range .Alerts }} --- **实例**: {{ .Labels.instance }} **环境**: {{ .Labels.env }} **服务**: {{ .Labels.service }} **描述**: {{ .Annotations.description }} **当前值**: {{ .Annotations.value }} {{ end }} Grafana Dashboard: [查看详情]({{ .ExternalURL }}) {{ end }}四、迁移后验证与监控4.1 迁移后检查清单#!/bin/bash # Grafana Alerting迁移后验证脚本 GRAFANA_URL${GRAFANA_URL:-http://localhost:3000} API_KEY${GRAFANA_API_KEY:-} echo Grafana Alerting 迁移后验证 # 1. 检查告警规则是否成功导入 echo echo [1] 告警规则状态检查 if [ -n $API_KEY ]; then curl -s -H Authorization: Bearer $API_KEY \ ${GRAFANA_URL}/api/v1/provisioning/alert-rules \ | python3 -c import json, sys data json.load(sys.stdin) total len(data) paused sum(1 for r in data if r.get(isPaused)) print(f 总规则数: {total}) print(f 已暂停: {paused}) print(f 活跃: {total - paused}) 2/dev/null || echo 无法连接Grafana API else echo 未设置GRAFANA_API_KEY跳过API检查 fi # 2. 验证通知渠道 echo echo [2] 验证通知渠道 # 发送测试通知 curl -s -X POST -H Authorization: Bearer $API_KEY \ -H Content-Type: application/json \ ${GRAFANA_URL}/api/v1/provisioning/contact-points/test \ -d {name:test-receiver} 2/dev/null \ || echo 跳过通知测试 # 3. 对比规则覆盖率 echo echo [3] 规则覆盖率检查 echo 请人工核对以下内容: echo - [ ] 所有Prometheus规则已转换 echo - [ ] 所有通知渠道已配置 echo - [ ] 静默规则已迁移 echo - [ ] 告警分组逻辑保持一致 echo echo 验证完成 4.2 告警质量监控迁移后需要建立对告警系统自身的监控监控指标PromQL示例合理阈值告警规则评估延迟grafana_alerting_rule_evaluation_duration_secondsP99 30s通知发送失败率grafana_alerting_notification_attempts_total{statuserror} 1%状态管理器操作延迟grafana_alerting_state_latency_secondsP99 100ms活跃告警数量grafana_alerting_alerts{stateAlerting}视规模而定五、总结从 AlertManager 迁移到 Grafana Alerting 不是简单的切换开关而是需要完整规划的系统工程。本文从架构差异分析、迁移工具开发、通知策略配置到迁移后验证提供了一套完整的实操方案。迁移带来的核心价值体现在三个方面一是告警规则与 Dashboard 的统一管理二是多数据源告警能力可以基于 Loki 日志、Elasticsearch 日志直接触发告警三是更灵活的通知策略树状路由。建议采用灰度迁移策略先选取 20% 的低优先级告警规则迁移到 Grafana Alerting稳定运行一周后再逐步扩大范围。同时在 AlertManager 上保留 48 小时的静默窗口作为回滚的缓冲期。