在zabbix的和业务系统的日常维护中免不了要频繁屏蔽掉告警，这里最方便的就是禁用掉相应的告警动作，由于要相当频繁的进行屏蔽，如果日常工作繁杂的话很容易忽略掉恢复，这里会有很大的隐患，今天笔者将带来一款对zabbix的告警动作进行审计的脚本，如下所示：
|
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 |
#!/usr/bin/env python # encoding=utf-8 import logging import requests import time from conf import action_id_list import json import sys def logger_getter(): today = time.strftime("%Y-%m-%d", time.localtime()) logger = logging.getLogger() if not len(logger.handlers): logger.setLevel(logging.DEBUG) formatter = logging.Formatter("%(asctime)s ||| %(levelname)s ||| %(lineno)d ||| %(funcName)s ||| %(message)s", datefmt='%Y-%m-%d %H:%M:%S') file_handler = logging.FileHandler('./logs/debug.log' + '.' + today) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger def send_msg_to_wework(chat_id, content): print("hello") diag = {"chatid": chat_id, "msgtype": "markdown", "markdown": { "content": content}} headers = {"Content-Type": "application/json"} diag = json.dumps(diag) requests.post('http://xxx.weixin.qq.com/cgi-bin/webhook/send?key=xxxxx', data=diag, headers=headers) def auth(api_url): post_data_login = { "jsonrpc" : "2.0", "method" : "user.login", "params" : { "user" : "xxxxx", "password" : "xxxxxxxx" }, "id" : 1 } """获取 zabbix 登录令牌""" ret = requests.post(api_url, data = json.dumps(post_data_login), headers = post_headers) if 'result' in ret.text: login_code = json.loads(ret.text).get("result") print('auth success! , zabbix login code: %s' %login_code) logger_getter().info('auth success! , zabbix login code: %s' %login_code) return str(login_code) elif 'error' in ret.text: print('auth fails! , exiting') logger_getter().error('auth fails! , exiting') sys.exit(-1) def zabbix_action_status_get(auth_code): post_update_action = { "jsonrpc": "2.0", "method": "action.get", "params": { "output": "extend", "selectOperations": "extend", "selectRecoveryOperations": "extend", "selectFilter": "extend" }, "auth": auth_code, "id": 1 } ret = requests.post(url, data = json.dumps(post_update_action), headers = post_headers) result=json.loads(ret.text) return result if __name__ == '__main__': url = 'http://192.168.2.1/zabbix/api_jsonrpc.php' post_headers = {'Content-Type': 'application/json'} chat_id_prod = "xxxxxxxxxx" auth_code = auth(url) disabled_action_list = [] all_action_list = zabbix_action_status_get(auth_code)['result'] # print(all_action_list) for action_specified_id in action_id_list: for action_all in all_action_list: # print("***") action_all_id = action_all['actionid'] action_all_name = action_all['name'] action_all_status = action_all['status'] if action_specified_id == action_all_id and action_all_status == '1': disabled_action_list.append('触发器名字：' + action_all_name) if not disabled_action_list: send_msg_to_wework(chat_id_prod,"# <font color='info'>** Zabbix告警动作审计正常! <@xxxx> <@xxxx><@xxxx>**</font>") else: send_msg_to_wework(chat_id_prod,"# <font color='warning'>** 以下Zabbix告警动作审计异常,请及时进行绑定!<@xxxx> <@xxxx><@xxxx>**</font>\n{0}".format('\n'.join(disabled_action_list))) |
上述脚本将对禁用掉的告警动作通过企业微信机器人发出消息通知。
你需要准备一个叫conf.py的配置文件，里面配置上你需要进行审计的告警动作的id,如下：
|
1 |
action_id_list = ['11','23','34','45'] |
然后修改上述脚本的如下内容：
-
send_msg_to_wework()函数中企业微信机器人的api地址
-
auth()函数中zabbix的用户名和密码
- 第79行的zabbix server的api地址
- 第81行企业微信群的群聊id
- 倒数最后四行中你需要at的同事的企业微信英文名
子舒博客