zabbix-server 配置告警

基于python环境

yum install python3
pip3 install requests

告警配置文件在

cd /usr/lib/zabbix/alertscripts/ cd /usr/local/share/zabbix/alertscripts

[root@localhost alertscripts]# ls
dingding.py

#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import json import sys import os import datetime webhook = "https://oapi.dingtalk.com/robot/send?access_token=自己的token11ca4841903f166d24e853502730985133d267ae0f1f8ea6c95d545edef4d768" log_path = "/usr/lib/zabbix/alertscripts/log/dingding.log" if len(sys.argv) != 4: print("参数错误：python3 dingding.py 手机号 标题 内容") sys.exit(1) user = str(sys.argv[1]).strip() subject = sys.argv[2] text = sys.argv[3] data = { "msgtype": "text", "text": { "content": f"{subject}{text}" }, "at": { "atMobiles": [], "isAtAll": False } } headers = {'Content-Type': 'application/json'} try: x = requests.post(url=webhook, data=json.dumps(data), headers=headers, timeout=10) res = x.json() print("钉钉接口返回：", res) except Exception as e: res = {"errcode": -999, "errmsg": f"请求异常:{str(e)}"} print("请求异常：", e) log_dir = os.path.dirname(log_path) if not os.path.exists(log_dir): os.makedirs(log_dir) with open(log_path, "a+", encoding="utf-8") as f: f.write("\n" + "--" * 30) now = str(datetime.datetime.now()) if res["errcode"] == 0: f.write(f"\n{now} 推送成功 内容:{subject}{text}\n")

测试

[root@localhost alertscripts]# ls dingding.py [root@localhost alertscripts]# python3 /usr/local/share/zabbix/alertscripts/dingding.py 11111111111 告警 来自Zabbix的测试消息 钉钉接口返回： {'errcode': 0, 'errmsg': 'ok'}

配置完成

钉钉加签设置

#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import json import sys import os import datetime import time import hmac import hashlib import base64 # 钉钉机器人配置 webhook = "https://oapi.dingtalk.com/robot/send?access_token=自己的token11ca4841903f166d24e853502730985133d267ae0f1f8ea6c95d545edef4d768" secret = "自己secSEC111c25df8a37e56253eded6c7f51ddc96e414acd27258ac16cc96192d1769d7c1" log_path = "/usr/local/share/zabbix/alertscripts/log" # 参数校验 if len(sys.argv) != 4: print("参数错误：python3 dingding.py 手机号 标题 内容") sys.exit(1) user = str(sys.argv[1]).strip() subject = sys.argv[2] text = sys.argv[3] # 消息内容，如需关键词可拼接 content = f"告警 {subject}{text}" data = { "msgtype": "text", "text": { "content": f"{subject}{text}" }, "at": { "atMobiles": [user], "isAtAll": False } } headers = {'Content-Type': 'application/json'} # 加签核心逻辑 timestamp = str(round(time.time() * 1000)) sign_str = f"{timestamp}\n{secret}" hmac_obj = hmac.new(secret.encode("utf-8"), sign_str.encode("utf-8"), digestmod=hashlib.sha256) sign = base64.b64encode(hmac_obj.digest()).decode("utf-8") # 拼接带签名、时间戳的完整请求地址 send_url = f"{webhook}&timestamp={timestamp}&sign={sign}" try: x = requests.post(url=send_url, data=json.dumps(data), headers=headers, timeout=10) res = x.json() print("钉钉接口返回：", res) except Exception as e: res = {"errcode": -999, "errmsg": f"请求异常:{str(e)}"} print("请求异常：", e) # 自动创建日志目录写入日志 log_dir = os.path.dirname(log_path) if not os.path.exists(log_dir): os.makedirs(log_dir) with open(log_path, "a+", encoding="utf-8") as f: f.write("\n" + "--" * 30) now = str(datetime.datetime.now()) if res["errcode"] == 0: f.write(f"\n{now} 推送成功 接收人:{user} 内容:{subject}{text}\n") else: f.write(f"\n{now} 推送失败 接收人:{user} 错误信息:{res} 内容:{subject}{text}\n")