Prometheus+Alertmanager进行飞书机器人webhook告警
原创Prometheus+Alertmanager进行飞书机器人webhook告警
原创
Alertmanager介绍
Alertmanager 是 Prometheus 生态系统中的一个核心组件，负责处理由 Prometheus 服务器发送的告警通知。其主要功能包括告警的去重、分组、抑制、和路由到不同的通知接收端（如邮件、Slack、PagerDuty 等）。本文中，我们主要使用webhook用于接收Alertmanager发出的告警。
Alertmanager的核心功能
1. 告警去重（De-duplication）：
• 当 Prometheus 服务器检测到告警条件满足时，会生成告警并发送到 Alertmanager。Alertmanager 会根据告警的标签和其他信息进行去重，以防止重复告警的发送。
2. 告警分组（Grouping）：
• Alertmanager 可以将相似的告警分组，以减少接收到的告警数量。例如，可以将同一服务或同一主机上的告警分组到一起，并发送一个综合告警通知。
3. 告警抑制（Silencing）：
• Alertmanager 支持设置抑制规则，以在特定条件下抑制某些告警的通知。例如，可以在计划维护期间抑制特定服务的告警，避免收到不必要的通知。
4. 告警路由（Routing）：
• Alertmanager 根据配置的路由规则，将告警通知发送到不同的接收端。可以根据告警的标签、严重性等信息设置不同的路由规则。
5. 告警通知（Notification）：
• Alertmanager 支持多种通知渠道，包括电子邮件、Slack、PagerDuty、OpsGenie、Webhook 等。用户可以配置不同的接收端，以便在不同的场景下接收到告警通知。
告警的产生与推送
Prometheus的告警与触达一直分为以下三个阶段：
定义告警规则
根据业务需求，我们可以在 Prometheus 配置文件中定义告警规则。
这里我们简单定义一个磁盘使用率告警的规则。
groups:
- name: disk_usage_alerts
rules:
- alert: DiskUsageHigh
expr: 100 * (node_filesystem_size_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"} - node_filesystem_avail_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"}) / node_filesystem_size_bytes{fstype!~"tmpfs|fuse.lxcfs|overlay"} > 40
for: 5m
labels:
severity: warning
annotations:
summary: "Disk usage is high on {{ $labels.instance }}"
description: "Disk usage on {{ $labels.instance }} is above 40% (current value: {{ $value }}%)"在prometheus.yml中进行装配我们前面定义的告警规则。
# Load rules once and periodically evaluate them according to the global 'evaluation_interval'.
rule_files:
- "/data/prometheus/alert.rules.yml"我们可以在http://127.0.0.1:9090/alerts中查看我们的规则是否装配成功。
评估告警条件
Prometheus会对我们配置的告警规则进行定期评估。满足计算规则与持续时间后就会生成告警信息，同时附带我们在规则中添加的附加告警信息。
1. 定期评估：
• Prometheus 服务器按照配置文件中的规则，定期评估告警规则。默认评估间隔为 1 分钟，可以通过配置文件进行调整。
2. 执行 PromQL 表达式：
• 在每次评估时，Prometheus 执行告警规则中的 PromQL 表达式。例如，在上述示例中，Prometheus 会计算过去 5 分钟内 CPU 空闲时间的平均值，并将其与 80% 进行比较。
3. 判断条件是否满足：
• Prometheus 检查表达式的计算结果。如果表达式返回的结果满足条件（例如，CPU 使用率高于 80%），则该条件被认为是满足的。
4. 处理 for 字段：
• 如果告警规则中定义了 for 字段（例如 5 分钟），则 Prometheus 需要在这个持续时间内持续满足告警条件，才会触发告警。如果条件在这个时间段内被打断，计时会重置。
告警信息触发
在Prometheus生成告警后，会由Alertmanager进行告警的接收，解析，与发送。
在处理Prometheus生成的告警之前，我们首先要在prometheus.yml中配置Alertmanager。
# Alertmanager configuration
alerting:
alertmanagers:
- static_configs:
- targets: ["127.0.0.1:9093"]Alertmanager告警相关配置
global：全局配置，适用于所有通知。
- resolve_timeout：设置告警从触发到解决的超时时间。
route：告警路由规则。
- receiver：默认接收端。
- group_by：告警分组依据的标签。
- group_wait：等待分组的时间。
- group_interval：分组发送的间隔时间。
- repeat_interval：重复发送告警的间隔时间。
routes：子路由规则。
- match 和 match_re：匹配告警标签。
- receiver：匹配规则对应的接收端。
receivers：接收端配置。
- email_configs：邮件接收端配置。
- pagerduty_configs：PagerDuty 接收端配置。
- slack_configs：Slack 接收端配置。
inhibit_rules：抑制规则。
- source_match：源告警标签匹配条件。
- target_match：目标告警标签匹配条件。
- equal：需要相等的标签。
Alertmanager告警处理流程
1. 接收告警：
• Alertmanager 接收到 Prometheus 发送的告警数据，并存储在内部队列中等待处理。
2. 去重（De-duplication）：
• Alertmanager 会对接收到的告警进行去重处理，确保同一告警不会被重复发送。
3. 分组（Grouping）：
• Alertmanager 根据配置文件中的分组规则，将相似的告警进行分组。分组是为了减少告警通知的数量，提高告警处理的效率。
4. 抑制（Silencing）：
• Alertmanager 会应用抑制规则，抑制不需要的告警。例如，在计划维护期间，可以设置抑制规则，避免发送大量不必要的告警通知。
5. 路由（Routing）：
• 根据配置文件中的路由规则，Alertmanager 将告警分配到不同的接收端。路由规则可以根据告警的标签、严重性等信息进行灵活配置。
6. 发送通知：
• Alertmanager 将处理后的告警通知发送到配置的接收端。常见的接收端包括邮件、Slack、PagerDuty、OpsGenie、Webhook 等。
告警的发送
这里我们选择webhook的告警方式。让机器人在飞书群中自动推送告警信息。
以下为基础的配置样例：
route:
group_by: ['disk_usage_alerts']
group_wait: 30s
group_interval: 5m
repeat_interval: 1h
receiver: 'webhook_receiver'
receivers:
- name: 'webhook_receiver'
webhook_configs:
- url: 'http://127.0.0.1:4000/alert'由于飞书机器人自身的数据结构限制，如果我们直接在webhook的URL中配置机器人的地址，此时是无法进行告警信息的发送与触达的。所以我们还需要自行开发服务，来调用飞书机器人进行告警。
这里我们使用Python来进行告警相关逻辑的开发与实现。
方案1：定时拉取
通过schedule定时请求Alertmanager的告警接口，来解析Prometheus推送至Alertmanager的告警信息。
以下为定时请求Alertmanager API，拉取告警信息进行封装发送的逻辑实现。
import requests
import json
import time
from datetime import datetime, timedelta
# Author: Empty Box
alert_manager_url = "http://127.0.0.1:9093/api/v2/alerts"
webhook_url = "https://open.feishu.cn/自己的飞书机器人地址"
# 获取告警
def get_alerts(alert_manager_url):
try:
response = requests.get(alert_manager_url)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error fetching alerts: {e}")
return []
# 监控告警
def monitor_alerts(alert_manager_url, poll_interval=60):
known_alerts = set()
while True:
alerts = get_alerts(alert_manager_url)
new_alerts = []
for alert in alerts:
alert_id = alert["labels"].get("alertname")
if alert_id not in known_alerts:
known_alerts.add(alert_id)
new_alerts.append(alert)
if new_alerts:
print(f"New alerts found: {len(new_alerts)}")
for alert in new_alerts:
print(json.dumps(alert, indent=4))
time.sleep(poll_interval)
# 发送告警
def send_alert(alert_manager_url):
try:
response = requests.get(alert_manager_url)
if response.status_code == 200:
alert_messages = response.json()
for alert_message in alert_messages:
labels = alert_message['labels']
# 告警类型
alert_name = alert_message['labels']['alertname']
# 任务类型
job = alert_message['labels']['job']
# 告警级别
severity = alert_message['labels']['severity']
# 告警内容
description = alert_message['annotations']['description']
# 获取告警状态，用于判断是否重复告警
state = alert_message['status']['state']
# 构造机器人告警消息体
webhook_msg = {}
elements = []
elements_json = {}
alert_name_json = {}
alert_name_text_json = {}
description_json = {}
description_text_json = {}
job_json = {}
job_text_json = {}
msg_type = "interactive"
webhook_msg["msg_type"] = msg_type
alert_name_json["tag"] = "div"
alert_name_text_json["content"] = "**告警类型**：" + alert_name
alert_name_text_json["tag"] = "lark_md"
alert_name_json["text"] = alert_name_text_json
elements.append(alert_name_json)
start_time_json["tag"] = "div"
start_time_text_json["content"] = "**触发时间**：" + start_time
start_time_text_json["tag"] = "lark_md"
start_time_json["text"] = start_time_text_json
elements.append(start_time_json)
severity_json["tag"] = "div"
severity_text_json["content"] = "**告警级别**：" + severity
severity_text_json["tag"] = "lark_md"
severity_json["text"] = severity_text_json
elements.append(severity_json)
instance_id_json["tag"] = "div"
instance_id_text_json["content"] = "**触发实例**：" + instance_id
instance_id_text_json["tag"] = "lark_md"
instance_id_json["text"] = instance_id_text_json
elements.append(instance_id_json)
job_json["tag"] = "div"
job_text_json["content"] = "**监控类型**：" + job
job_text_json["tag"] = "lark_md"
job_json["text"] = job_text_json
elements.append(job_json)
description_json["tag"] = "div"
description_text_json["content"] = "**告警内容**：" + description
description_text_json["tag"] = "lark_md"
description_json["text"] = description_text_json
elements.append(description_json)
elements_json["elements"] = elements
webhook_msg["card"] = elements_json
json_data = json.dumps(webhook_msg)
requests.post(webhook_url, headers={'Content-Type': 'application/json'}, data=json_data)
else:
print('告警信息获取失败')
except Exception as e:
print(f"error is :{e}")
if __name__ == "__main__":
poll_interval = 60 # 轮询间隔，单位为秒
monitor_alerts(alert_manager_url, poll_interval)
在这种方式中，我们主要使用request库来实现机器人API的调用与Alertmanager API的调用。使用JSON库来进行告警信息的解析与机器人消息的封装。在某些离线任务的状态监测上可以使用这种定时拉取的方式。在某些实时性要求较高的场景下，定时拉取告警信息，可能就会出现，告警不及时等问题。
方案2：实时触发
在实时触发的场景下，我们可以通过自行开发web服务，让Alertmanager实时请求，然后我们在自定义逻辑中，对Alertmanager实时发送的告警进行解析与处理。
from flask import Flask, request, jsonify
from datetime import datetime, timedelta
import json
import time
import requests
app = Flask(__name__)
webhook_url = "https://open.feishu.cn/自己的机器人地址"
@app.route('/alert', methods=['POST'])
def receive_alert():
# 供Alert调用，解析Alert返回的json信息
try:
msg_json = {}
alert_response = request.json
alerts = alert_response["alerts"]
if len(alerts) > 0:
for alert in alerts:
alert_name = alert["labels"]["alertname"]
# 根据告警类型执行不同的告警信息消息体
if "DiskUsageHigh" == alert_name:
msg_json = disk_usage_high_json(alert)
# 发送告警
response = send_alert(msg_json)
print("告警原数据: ", alert)
continue
return str(response.status_code)
except Exception as e:
return e
def send_alert(json_data):
response = requests.post(webhook_url, headers={'Content-Type': 'application/json'}, data=json_data)
return response
def disk_usage_high_json(alerts):
alert_name = alerts["labels"]["alertname"]
job = alerts["labels"]["job"]
description = alerts["annotations"]["description"]
webhook_msg = {}
elements = []
elements_json = {}
alert_name_json = {}
alert_name_text_json = {}
description_json = {}
description_text_json = {}
job_json = {}
job_text_json = {}
msg_type = "interactive"
webhook_msg["msg_type"] = msg_type
alert_name_json["tag"] = "div"
alert_name_text_json["content"] = "**告警类型**：" + alert_name
alert_name_text_json["tag"] = "lark_md"
alert_name_json["text"] = alert_name_text_json
elements.append(alert_name_json)
start_time_json["tag"] = "div"
start_time_text_json["content"] = "**触发时间**：" + start_time
start_time_text_json["tag"] = "lark_md"
start_time_json["text"] = start_time_text_json
elements.append(start_time_json)
severity_json["tag"] = "div"
severity_text_json["content"] = "**告警级别**：" + severity
severity_text_json["tag"] = "lark_md"
severity_json["text"] = severity_text_json
elements.append(severity_json)
description_json["tag"] = "div"
description_text_json["content"] = "**告警内容**：" + description
description_text_json["tag"] = "lark_md"
description_json["text"] = description_text_json
elements.append(description_json)
elements_json["elements"] = elements
webhook_msg["card"] = elements_json
json_data = json.dumps(webhook_msg)
return json_data
if __name__ == '__main__':
app.run(host='0.0.0.0', port=4000)
在实时告警的自定义逻辑中，我们主要使用Flask来进行Python web服务的开发，通过解析Alertmanager实时请求web接口的JSON，来解析告警内容，根据告警类别进行判断，来调用不同的告警消息体构造方法来封装告警内容并进行触达。
方案2告警效果如下图所示：
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。
如有侵权，请联系 cloudcommunity@tencent.com 删除。
腾讯云开发者
