Prometheus+Alertmanager进行飞书机器人webhook告警

原创

空洞的盒子

修改于 2024-07-16 16:03:40

1640

修改于 2024-07-16 16:03:40

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 删除。

Prometheus 监控服务

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

Prometheus 监控服务

#Alertmanager

登录后参与评论

0 条评论

热度