alertmanager 直接配置 飞书 的 webhook ,发现并不满足飞书接口的 json 格式。报错如下
level=error ts=2025-08-28T04:57:02.734Z caller=dispatch.go:310 component=dispatcher msg="Notify for alerts failed" num_alerts=23 err="prometheusalert-webhook/webhook[0]: notify retry canceled due to unrecoverable error after 1 attempts: unexpected status code 400: https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxx"
上网查询发现开源项目 prometheusalert 按照官方文档配置配置飞书地址,v4.9.1 版本默认的模板和网上找到的模板 空卡片的情况,如下
然后就打算自己写个 python 查询,接收 alertmanager 的消息体,做修改转发给 飞书。
cm.yaml
apiVersion: v1
kind: ConfigMap
metadata:name: alert-flask-cmnamespace: monitor
data:app.py: |from flask import Flask, request, jsonifyimport jsonimport requestsimport loggingfrom datetime import datetimeapp = Flask(__name__)# 飞书机器人 Webhook 地址webhook_url = "https://open.feishu.cn/open-apis/bot/v2/hook/xxxxxxxx"# 日志配置logging.basicConfig(level=logging.INFO)def format_time(timestr):"""把 2025-08-08T06:55:43.825666166Z → 2025-08-08 06:55:43"""try:dt = datetime.fromisoformat(timestr.replace("Z", "+00:00"))return dt.strftime("%Y-%m-%d %H:%M:%S")except Exception:return timestr@app.route('/alert', methods=['POST'])def receive_alert():try:alert_response = request.jsonlogging.info("收到 Alertmanager 消息体: %s", json.dumps(alert_response, ensure_ascii=False))alerts = alert_response.get("alerts", [])if not alerts:logging.info("没有告警")return "no alerts", 200send_status = []for alert in alerts:status = alert.get("status", "firing")if status == "firing":msg_json = format_alert_to_feishu(alert)elif status == "resolved":msg_json = format_resolved_to_feishu(alert)else:logging.info("未知状态: %s", status)continuelogging.info("生成飞书消息体: %s", msg_json)response = send_alert(msg_json)if response is None:send_status.append("发送失败")else:send_status.append(f"发送成功:{response.status_code}")return "; ".join(send_status), 200except Exception as e:logging.exception("处理告警异常")return f"error: {str(e)}", 500def send_alert(json_data):try:response = requests.post(webhook_url, json=json.loads(json_data), timeout=5)response.raise_for_status()logging.info("发送飞书成功,状态码: %s", response.status_code)return responseexcept requests.exceptions.RequestException as e:logging.error("发送飞书失败: %s", e)return Nonedef format_alert_to_feishu(alert):labels = alert.get("labels", {})annotations = alert.get("annotations", {})alert_name = labels.get("alertname", "Unknown")instance = labels.get("instance", "Unknown")severity = labels.get("severity", "N/A")summary = annotations.get("summary", "")description = annotations.get("description", "无描述")start_time = format_time(alert.get("startsAt", "Unknown"))lines = [f"**告警名称**:{alert_name}",f"**告警实例**:{instance}",f"**告警级别**:{severity}",]if summary:lines.append(f"**告警摘要**:{summary}")lines.append(f"**告警描述**:{description}")lines.append(f"**触发时间**:{start_time}")content = "\n".join(lines)webhook_msg = {"msg_type": "interactive","card": {"header": {"title": {"tag": "plain_text", "content": "===== == 告警 == ====="},"template": "red"},"elements": [{"tag": "div", "text": {"tag": "lark_md", "content": content}}]}}return json.dumps(webhook_msg, ensure_ascii=False)def format_resolved_to_feishu(alert):labels = alert.get("labels", {})annotations = alert.get("annotations", {})alert_name = labels.get("alertname", "Unknown")instance = labels.get("instance", "Unknown")summary = annotations.get("summary", "")success_msg = annotations.get("success", "告警已恢复")description = annotations.get("description", "无描述")end_time = format_time(alert.get("endsAt", "Unknown"))lines = [f"**告警名称**:{alert_name}",f"**告警实例**:{instance}",]if summary:lines.append(f"**告警摘要**:{summary}")lines.append(f"**告警描述**:{description}")lines.append(f"**恢复说明**:{success_msg}")lines.append(f"**恢复时间**:{end_time}")content = "\n".join(lines)webhook_msg = {"msg_type": "interactive","card": {"header": {"title": {"tag": "plain_text", "content": "===== == 恢复 == ====="},"template": "green"},"elements": [{"tag": "div", "text": {"tag": "lark_md", "content": content}}]}}return json.dumps(webhook_msg, ensure_ascii=False)if __name__ == '__main__':app.run(host='0.0.0.0', port=4000)
deployment 中的镜像需要自己构建,随便找个 python 镜像作为 base pip 安装 flask、requetsts 即可
deploy-svc.yaml
apiVersion: apps/v1
kind: Deployment
metadata:name: alert-flasknamespace: monitor
spec:replicas: 1selector:matchLabels:app: alert-flasktemplate:metadata:labels:app: alert-flaskspec:containers:- name: alert-flaskimage: python:3.11-slim-bookworm-flaskcommand: ["python", "/app/app.py"]ports:- containerPort: 4000volumeMounts:- name: app-cmmountPath: /appvolumes:- name: app-cmconfigMap:name: alert-flask-cm
---
apiVersion: v1
kind: Service
metadata:name: alert-flask-svcnamespace: monitor
spec:selector:app: alert-flaskports:- name: httpport: 4000targetPort: 4000type: ClusterIP
上面的 configmap、deploy、service 部署好后,更改 alertmanager 的配置
receivers:
- name: feishuwebhook_configs:- send_resolved: trueurl: http://alert-flask-svc:4000/alert
然后飞书就能收到告警了