zabbix-dingdingtalk-bot

发表于 更新于 分类于 Disqus： 本文字数： 805 阅读时长 ≈ 3 分钟
# zabbix 对接钉钉企业内部机器人

由于钉钉自定义机器人下线 见此.新同事需要无法再添加告警机器人,需要通过另一种方式实现钉钉机器人和zabbix告警对接.通过查阅,可以发现使用钉钉付费的企业内部机器人也可以进行告警通知.下文主要描述实现方式.

获取权限

  1. 获取企业内部开发者权限
  2. 创建钉钉应用—开通机器人
  3. 机器人授予相关权限,请见钉钉开发者文档
  4. 记录关键信息. 包括应用 Client ID Client Secret RobotCode
  5. 机器人消息接收模式此项不需要勾选(此为服务端接受消息通道)

## 告警脚本实现

放置于zabbix server 的 alertscripts 相关目录下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
#! /usr/bin/python3
import json
import requests
import sys
import logging
import ast

# 配置常量
API_BASE_URL = "https://api.dingtalk.com/v1.0"
APP_KEY = "*************"
APP_SECRET = "************************"
ROBOT_CODE = "********************"

def get_access_token():
    url = f"{API_BASE_URL}/oauth2/accessToken"
    headers = {"Content-Type": "application/json"}
    data = {"appKey": APP_KEY, "appSecret": APP_SECRET}

    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        json_response = response.json()
        return json_response.get("accessToken")
    except requests.exceptions.RequestException as e:
        logging.error(f"Error getting access token: {e}")
        return None

def send_to_message(access_token, user_ids, msg_param_data):
    url = f"{API_BASE_URL}/robot/oToMessages/batchSend"
    headers = {"Content-Type": "application/json", "x-acs-dingtalk-access-token": access_token}
    data = {"robotCode": ROBOT_CODE, "userIds": user_ids, "msgKey": "officialMarkdownMsg", "msgParam": msg_param_data}

    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        print(response.text)
    except requests.exceptions.RequestException as e:
        logging.error(f"Error sending message: {e}")

def get_user_id(access_token, phone):
    url = f"{API_BASE_URL}/topapi/v2/user/getbymobile?access_token={access_token}"
    headers = {"Content-Type": "application/json"}
    data = {"mobile": phone}

    try:
        response = requests.post(url, json=data, headers=headers)
        response.raise_for_status()
        json_response = response.json()
        return json_response.get("result", {}).get("userid")
    except requests.exceptions.RequestException as e:
        logging.error(f"Error getting user ID: {e}")
        return None

def get_trigger_color(severity):
    color_dict = {
        "Not classified": "#808080",
        "Information": "#008000",
        "Warning": "#FFFF00",
        "Average": "#FFA500",
        "High": "#FF0000",
        "Disaster": "#8B0000",
        "default_color": "#808080"
    }
    return color_dict.get(severity, color_dict["default_color"])

def main():
    # 获取企业内部应用的accessToken
    access_token = get_access_token()

    if not access_token:
        logging.error("Failed to obtain access token. Exiting.")
        sys.exit(1)

    # 获取用户手机号
    phone_id = sys.argv[1]

    # 根据用户手机号获取userid
    user_id = get_user_id(access_token, phone_id)

    if not user_id:
        logging.error("Failed to obtain user ID. Exiting.")
        sys.exit(1)

    user_ids = [user_id]

    # 获取需要发送的消息
    subject = sys.argv[2]
    message = sys.argv[3]

    message_data = json.loads(message)

    # 把字符串转为字典
    message_dictionary  = ast.literal_eval(message)
    HOSTNAME = ': '.join(('> 告警主机',message_dictionary["HOSTNAME"]))
    HOSTIP = ': '.join(('告警地址',message_dictionary["HOSTIP"]))
    ITEMNAME = ': '.join(('监控项目',message_dictionary["ITEMNAME"]))
    ITEMLASTVALUE = ': '.join(('监控取值',message_dictionary["ITEMLASTVALUE"]))
    TRIGGERSEVERITY = ': '.join(('告警等级',message_dictionary["TRIGGERSEVERITY"]))
    TRIGGERSTATUS = ': '.join(('当前状态',message_dictionary["TRIGGERSTATUS"]))
    TRIGGERNAME = ': '.join(('告警信息',message_dictionary["TRIGGERNAME"]))
    EVENTTIME = ': '.join(('告警时间',message_dictionary["EVENTTIME"]))
    EVENTAGE = ': '.join(('持续时间',message_dictionary["EVENTAGE"]))
    EVENTID = ': '.join(('事件ID',message_dictionary["EVENTID"]))
    sendtext = '\n\n >'.join((HOSTNAME,HOSTIP,ITEMNAME,ITEMLASTVALUE,TRIGGERSEVERITY,TRIGGERSTATUS,TRIGGERNAME,EVENTTIME,EVENTAGE,EVENTID))


    # 获取触发器严重程度对应的颜色代码
    trigger_severity = message_data.get('TRIGGERSEVERITY', None)
    rgb_code = get_trigger_color(trigger_severity)

    msg_data = {
    "title": "zabbix告警消息",
    "text": ''.join(('<font color=' + rgb_code + ' ' + 'face="黑体">'
                     +
                        subject
                     +
                     '</font>' 
                     +
                     '\n\n'
                     + 
                     sendtext 
                     + 
                    '\n\n'
                    '#### [点击转到ZABBIX](https://zabbix.wenet.com.cn/)'
                    ))
    }

    # 将字典转换为 JSON 格式的字符串
    msg_data_string = json.dumps(msg_data)

    # 调用发送消息的接口
    send_to_message(access_token, user_ids, msg_data_string)

    # 设置日志记录
    logging.info(f"user_id: {user_id}")
    logging.info(f"subject: {subject}")
    logging.info(f"message: {message}")

if __name__ == "__main__":
    main()

zabbix 配置

  1. 建立新告警媒介: ( 消息模板里面勾引问题以及问题恢复)
    img
  2. 动作中添加触发器动作:
    img
    img 
    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    15
     主题:
    发生故障: {EVENT.NAME}
    消息:
    {
     "HOSTNAME":"{HOST.NAME}",
     "HOSTIP":"{HOST.IP}",
     "ITEMNAME":"{ITEM.NAME}",
     "ITEMLASTVALUE":"{ITEM.LASTVALUE}",
     "TRIGGERSEVERITY":"{TRIGGER.SEVERITY}",
     "TRIGGERSTATUS":"{TRIGGER.STATUS}",
     "TRIGGERNAME":"{TRIGGER.NAME}",
     "EVENTTIME":"{EVENT.DATE} {EVENT.TIME}",
     "EVENTAGE":"{EVENT.AGE}",
     "EVENTID":"{EVENT.ID}"
     }
  3. 在具体用户上添加告警媒介
    img

实现效果

img

参考链接

  1. 如何优雅地用Python3发送Zabbix告警推送