配置钉钉群机器人
1 点击群的设置按钮
2 点击智能群助手
3 点击添加机器人
4 点添加
5 选择自定义机器人
6 点添加
7 配置名称和安全设置 并记录下加签密钥
8 记录下webhook地址
配置自定义告警脚本
# 进入zabbix server容器
docker exec -u root -ti zabbix-server /bin/bash
# 安装python3.6
yum install python36
# 安装pip
yum install python3-pip
# 安装requests库
pip3 install requests
# 查看自定义报警脚本路径
cat /etc/zabbix/zabbix_server.conf | grep AlertScriptsPath=
# 进入自定义报警脚本目录
cd /usr/lib/zabbix/alertscripts
# 创建告警脚本
vi dingding.py
dingding.py参考(脚本来自网络 简单修改了下)
配置webhook链接 api_url = ‘webhook地址’
配置加签密钥 api_secret = ‘加签密钥’
配置是否@所有人 at_all = True
钉钉机器人webhook参考文档
#!/usr/bin/env python3
# coding:utf8
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
import sys
# webhook URL
api_url = 'webhook地址'
# secret key
api_secret = '加签密钥'
# is @all
at_all = True
def get_timestamp_sign():
timestamp = str(round(time.time() * 1000))
secret = api_secret
secret_enc = secret.encode('utf-8')
string_to_sign = '{} {}'.format(timestamp, secret)
string_to_sign_enc = string_to_sign.encode('utf-8')
hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
return timestamp, sign
def get_signed_url():
timestamp, sign = get_timestamp_sign()
webhook = api_url + "×tamp=" + timestamp + "&sign=" + sign
return webhook
def get_webhook(mode):
if mode == 0:
webhook = api_url
elif mode == 1:
webhook = get_signed_url()
else:
webhook = ""
return webhook
def get_message(text, user_info):
message = {
"msgtype": "text",
"text": {
"content": text
},
"at": {
"atMobiles": [
user_info,
],
"isAtAll": at_all
}
}
return message
def send_ding_message(text, user_info):
webhook = get_webhook(1)
header = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
message = get_message(text, user_info)
message_json = json.dumps(message)
info = requests.post(url=webhook, data=message_json, headers=header).json()
code = info["errcode"]
errmsg = info["errmsg"]
if code == 0:
print("send success return:%s %s " % (code, errmsg))
else:
print("send fail return:%s %s " % (code, errmsg))
exit(3)
if __name__ == "__main__":
text = sys.argv[3]
user_info = sys.argv[1]
send_ding_message(text, user_info)
# 给予执行权
chmod +x dingding.py
# 测试
./dingding.py 18137900000 title text
创建报警媒介类型
1 管理 - 报警媒介类型 - 创建媒体类型
2 配置媒介参数
3 切换到Message templates页 添加消息模板
4 添加告警消息模板
主题:
收到告警: {EVENT.NAME}
消息:
收到告警
ID: {EVENT.ID}
主机: {HOST.NAME}
告警: {EVENT.NAME}
时间: {EVENT.DATE} {EVENT.TIME}
级别: {EVENT.SEVERITY}
数据: {EVENT.OPDATA}
5 添加告警恢复模板
主题:
告警恢复: {EVENT.NAME}
消息:
告警已恢复
ID: {EVENT.ID}
主机: {HOST.NAME}
告警: {EVENT.NAME}
时间: {EVENT.RECOVERY.DATE} {EVENT.RECOVERY.TIME}
级别: {EVENT.SEVERITY}
持续时间: {EVENT.DURATION}
测试
1 点击报警媒介后面的测试按钮
2 输入收件人手机号点击测试
3 成功收到消息
配置钉钉告警
1 User settings - Profile - 报警媒介 - 添加
2 配置告警
创建动作
1 配置 - 动作 - Trigger actions - 创建动作
2 输入名称
3 切换到操作页 添加告警操作和恢复操作
4 告警操作 选择用户组或用户 仅送到钉钉
5 恢复操作 选择用户组或用户 仅送到钉钉
6 手动shutdown一个接口测试成功
附:企业微信告警脚本
#!/usr/bin/env python3
# coding:utf8
import requests
import json
import sys
# webhook URL
api_url = 'webhook地址'
# is @all
at_all = True
def get_message(text, user_info):
mentioned_mobile_list = []
if at_all:
mentioned_mobile_list.append('@all')
else:
mentioned_mobile_list.append(user_info)
message = {
"msgtype": "text",
"text": {
"content": text,
"mentioned_mobile_list": mentioned_mobile_list
}
}
return message
def send_wechat_message(text, user_info):
header = {
"Content-Type": "application/json",
"Charset": "UTF-8"
}
message = get_message(text, user_info)
message_json = json.dumps(message)
info = requests.post(url=api_url, data=message_json, headers=header).json()
code = info["errcode"]
errmsg = info["errmsg"]
if code == 0:
print("send success return:%s %s " % (code, errmsg))
else:
print("send fail return:%s %s " % (code, errmsg))
exit(3)
if __name__ == "__main__":
text = sys.argv[3]
user_info = sys.argv[1]
send_wechat_message(text, user_info)
———————————————
页面更新:2024-04-30
本站资料均由网友自行发布提供,仅用于学习交流。如有版权问题,请与我联系,QQ:4156828
© CopyRight 2008-2024 All Rights Reserved. Powered By bs178.com 闽ICP备11008920号-3
闽公网安备35020302034844号