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

Viewed 789

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

7 Answers

1、新建feishu_card通知媒介
image.png

2、启动通知脚本
image.png

脚本内容:

#!/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 + "<at id=" + i + "></at>"
                print("user_at_message: %s" % user_at_message)
        if user_at_message == "":
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "\n" + "**查看详情**: [告警平台](http://xx.xx.xx.xx/alert-cur-events)",
                "tag": "lark_md"
            }
        else:
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "\n" + "**查看详情**: [告警平台](http://xx.xx.xx.xx/alert-cur-events)" + "\n" + 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、启动通知配置
image.png
其中获取查看FeishuAts的用户user id参考:https://www.cnblogs.com/mxcl/p/16359730.html

配置完成之后触发告警收到的通知就变成这样:
image.png

image.png

很强,按照模板一下子就完事了,多谢大佬

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

替换脚本中的告警平台的地址,另外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 + "<at id=" + i + "></at>"
                print("user_at_message: %s" % user_at_message)
        if user_at_message == "":
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "\n" + "**查看详情**: [告警平台](" + host + ")",
                "tag": "lark_md"
            }
        else:
            message = {
                "content": payload.get('tpls').get("feishu", "feishu.tpl not found") + "\n" + "**查看详情**: [告警平台](" + host + ")" + "\n" + 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()

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

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

image.png

举个例子:

image.png

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

image.png

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


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

FlashDuty

感谢大佬提供思路,已通过其它方式实现需求

感谢@ulricqin 提供的解决方案
我这边测试了下,发送的通知消息并不能@人,收到的消息是混乱的,效果如下:
image.png
猜测可能的原因是消息体不符合飞书消息规范(json在转义时发生异常)

【=====update========】
经过不懈努力,通过自定义脚本方式实现了需求,这里简单说下实现步骤:
1)在【告警规则】配置中添加【附加信息】配置,如下
image.png
2)feishu通知模板不做修改,保持默认即可
3)添加自定义通知媒介【ifeishu】
image.png
4)【告警规则】配置中通知配置选择【ifeishu】通知媒介
image.png
5)通知设置中启用【通知脚本】功能,配置脚本路径
image.png
最终实现发送效果:
image.png
image.png

贴下核心脚本内容:
1)判断附加信息字段,按飞书消息体规则将@人员id组合起来
image.png
2)判断@人员消息非空,进行组装发送消息体,采用消息卡片格式进行发送
image.png

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

挺好的,github提供gist功能,把你的脚本作为gist提供出来如何?或者直接贴到回答里,用markdown的code段

可以的,我看下gist如何使用

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

按上述修改测试了下,确实可以实现@效果,不过由于默认实现的发送消息是文本格式而非卡片格式,还是使用上面自定义脚本方式了,感谢@710leo 提供的解决方案

大佬,能否提供下脚本哇 earnmoney6666@qq.com

@tigerl 大佬,麻烦提供下脚本 634826182@qq.com

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

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

<at email="xxxx@flashcat.cloud"></at><at email="yyyy@flashcat.cloud"></at>