🙏夜莺nightingale v6版本如何配置飞书告警通道@人功能

Question

环境：夜莺nightingale v6版本
需求：配置飞书告警通道，实现@人的功能

凉拌毛豆 · Answer

1、新建feishu_card通知媒介

2、启动通知脚本

脚本内容：

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys
import json
import requests
import re

class Sender(object):

    @classmethod
    def send_feishu_card(cls, payload):
        users = payload.get('event').get("notify_users_obj")
        tokens = {}
        phones = {}

        for u in users:
            if u.get("phone"):
                phones[u.get("phone")] = 1

            contacts = u.get("contacts")
            if contacts.get("feishu_robot_token", ""):
                tokens[contacts.get("feishu_robot_token", "")] = 1
        info_annotations =  payload.get('event').get('annotations')
        user_at_message = ""
        if info_annotations.get('FeishuAts',""):
            userAt = info_annotations.get("FeishuAts")
            for i in userAt.split(","):
                user_at_message = user_at_message + ""
                print("user_at_message: %s" % user_at_message)
        if user_at_message == "":
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "
" + "**查看详情**: [告警平台](http://xx.xx.xx.xx/alert-cur-events)",
                "tag": "lark_md"
            }
        else:
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "
" + "**查看详情**: [告警平台](http://xx.xx.xx.xx/alert-cur-events)" + "
" + user_at_message,
                "tag": "lark_md"
            }
        if re.findall("S1 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "red"
            u_content = "🔔 告警通知"
        elif re.findall("S2 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "yellow"
            u_content = "🔔 告警通知"
        elif re.findall("S3 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "purple"
            u_content = "🔔 告警通知"
        else:
            color = "green"
            u_content = "🔔 告警恢复"
        headers = {
            "Content-Type": "application/json;charset=utf-8",
            "Host": "open.feishu.cn"
        }
        for t in tokens:
            url = "https://open.feishu.cn/open-apis/bot/v2/hook/{}".format(t)
            body = {
                "msg_type": "interactive",
                "card": {
                    "config": {
                        "wide_screen_mode": True
                    },
                    "header": {
                        "title": {
                            "tag": "plain_text",
                            "content": u_content
                        },
                        "template": color
                    },
                    "elements": [{"tag": "div", "text": message }]
                }
            }
            response = requests.post(url, headers=headers, data=json.dumps(body))
            print(f"notify_ifeishu: token={t} status_code={response.status_code} response_text={response.text}")

def main():
    payload = json.load(sys.stdin)
    with open(".payload", 'w') as f:
        f.write(json.dumps(payload, indent=4))
    for ch in payload.get('event').get('notify_channels'):
        send_func_name = "send_{}".format(ch.strip())
        if not hasattr(Sender, send_func_name):
            print("function: {} not found", send_func_name)
            continue
        send_func = getattr(Sender, send_func_name)
        send_func(payload)


if __name__ == "__main__":
    main()

注意：
替换脚本中的告警平台的地址，另外n9e容器中默认没有requests和re库，需要进入容器手动安装

docker exec -it n9e pip install requests  -i  http://pypi.douban.com/simple --trusted-host=pypi.douban.com
docker exec -it n9e pip install re  -i  http://pypi.douban.com/simple --trusted-host=pypi.douban.com

3、启动通知配置

其中获取查看FeishuAts的用户user id参考：https://www.cnblogs.com/mxcl/p/16359730.html

配置完成之后触发告警收到的通知就变成这样：

打工人小余 · Answer

替换脚本中的告警平台的地址，另外n9e容器中默认没有requests和re库，需要进入容器手动安装

大佬，夜莺v6版本容器内的python版本是python 3.11，re库是内置的，然后requests可以直接用urllib.request来替代，urllib也是内置的，下面是我按照你提供的脚本做的修改，这样就不需要在容器内部进行额外的操作了，毕竟在k8s里面搞这个，不大方便

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
import sys
import json
import urllib.request
import urllib.parse
import re

class Sender(object):

    @classmethod
    def send_feishu_card(cls, payload):
        users = payload.get('event').get("notify_users_obj")
        tokens = {}
        phones = {}
        host = 'http://192.168.10.89:17000/alert-cur-events'
        for u in users:
            if u.get("phone"):
                phones[u.get("phone")] = 1
            contacts = u.get("contacts")
            if contacts.get("feishu_robot_token", ""):
                tokens[contacts.get("feishu_robot_token", "")] = 1
        info_annotations =  payload.get('event').get('annotations')
        user_at_message = ""
        if info_annotations.get('FeishuAts', ""):
            userAt = info_annotations.get("FeishuAts")
            for i in userAt.split(","):
                user_at_message = user_at_message + ""
                print("user_at_message: %s" % user_at_message)
        if user_at_message == "":
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "
" + "**查看详情**: [告警平台](" + host + ")",
                "tag": "lark_md"
            }
        else:
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "
" + "**查看详情**: [告警平台](" + host + ")" + "
" + user_at_message,
                "tag": "lark_md"
            }
        if re.findall("S1 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "red"
            u_content = "🔔  告警通知"
        elif re.findall("S2 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "yellow"
            u_content = "🔔  告警通知"
        elif re.findall("S3 Triggered", payload.get('tpls').get("feishu", "feishu not found")):
            color = "purple"
            u_content = "🔔  告警通知"
        else:
            color = "green"
            u_content = "✅  告警解除"
        headers = {
            "Content-Type": "application/json;charset=utf-8",
            "Host": "open.feishu.cn"
        }
        for t in tokens:
            url = "https://open.feishu.cn/open-apis/bot/v2/hook/{}".format(t)
            body = {
                "msg_type": "interactive",
                "card": {
                    "config": {
                        "wide_screen_mode": True
                    },
                    "header": {
                        "title": {
                            "tag": "plain_text",
                            "content": u_content
                        },
                        "template": color
                    },
                    "elements": [{"tag": "div", "text": message }]
                }
            }
            data = json.dumps(body).encode('utf-8')
            request = urllib.request.Request(url, data=data, headers=headers)
            response = urllib.request.urlopen(request)
            print(f"notify_ifeishu: token={t} status_code={response.status} response_text={response.read().decode('utf-8')}")

def main():
    payload = json.load(sys.stdin)
    with open(".payload", 'w') as f:
        f.write(json.dumps(payload, indent=4))
    for ch in payload.get('event').get('notify_channels'):
        send_func_name = "send_{}".format(ch.strip())
        if not hasattr(Sender, send_func_name):
            print("function: {} not found", send_func_name)
            continue
        send_func = getattr(Sender, send_func_name)
        send_func(payload)


if __name__ == "__main__":
    main()

ulricqin · Answer

这个我没具体做，不过我看了一下飞书的文档，有一些思路，分享如下：

v6版本的夜莺，告警规则配置页面，最下面，有个自定义字段的功能。可以把想要at的人写在这里，写法类似下面这样（来自飞书文档）：

举个例子：

然后在系统配置-通知模板中，配置飞书的通知模板，把这个 FeishuAts 引用一下，样例如下：

我没有测试，你试试？如果有结论可以分享一下。

更推荐的做法是使用 FlashDuty，支持告警聚合降噪、排班、认领、升级、协同等等，而且可以支持飞书应用、钉钉应用，告警通过卡片的方式发送，而且可以在飞书/钉钉里直接认领处理，或者执行屏蔽等操作。

FlashDuty

tigerL · Answer

感谢@ulricqin 提供的解决方案
我这边测试了下，发送的通知消息并不能@人，收到的消息是混乱的，效果如下:

猜测可能的原因是消息体不符合飞书消息规范（json在转义时发生异常）

【=====update========】
经过不懈努力，通过自定义脚本方式实现了需求，这里简单说下实现步骤：
1）在【告警规则】配置中添加【附加信息】配置，如下

2）feishu通知模板不做修改，保持默认即可
3）添加自定义通知媒介【ifeishu】

4）【告警规则】配置中通知配置选择【ifeishu】通知媒介

5）通知设置中启用【通知脚本】功能，配置脚本路径

最终实现发送效果：

贴下核心脚本内容：
1）判断附加信息字段，按飞书消息体规则将@人员id组合起来

2）判断@人员消息非空，进行组装发送消息体，采用消息卡片格式进行发送

以上，感谢@ulricqin 大佬提供的思路，如有同学需要完整脚本，可以私聊我

710leo · Answer

测试了下，将 {{index .AnnotationsJSON "FeishuAts"}} 改为 {{$feishuAt := index .AnnotationsJSON "FeishuAts"}} {{$feishuAt|unescaped}} 就可以实现@的效果了

uglyliu · Answer

大佬，飞书消息卡片的通知脚本能否贴下呢？

jicki · Answer

通过 email At 更好，不需要去获取 open_id

🙏夜莺nightingale v6版本如何配置飞书告警通道@人功能

7 Answers