使用Python实现批量启用或禁用zabbix的告警动作

前言

当你的企业的zabbix系统用了很多年之后肯定是积攒下来少说十几个多则几十个zabbix告警动作，由于历史原因经常有一半被永远禁用了，但是你又不敢随意的删掉它们，这样每次维护监控在禁用之前的时候你就需要记下来哪些是启用的，然后再在维护完之后小心翼翼的全部启用，确保一个不落，但是人的眼睛总是会走神，难免会有遗漏，即使是有双人复核恐怕也不能幸免。本文实现了用Python脚本批量一次性禁用和启用，原理就是提前把这些告警动作id多次或N次反反复复复核后写到配置文件里面，然后用Python脚本调用zabbix的rest api来批量操作。

实战

$ mkdir logs

1	$ mkdir logs

新建一个配置文件模块conf.py，如下：

action_id_list = ['11', '12', '13', '14', '15', '16', '17', '...']

1	action_id_list = ['11', '12', '13', '14', '15', '16', '17', '...']

上面这些id可以在点开每个告警动作的配置页面，然后在页面的url里面有个actionid=的查询字符串，等号后面就是id，自己记下来然后写入上述配置模块中。

下面直接上脚本：

#!/usr/bin/env python
# encoding=utf-8

import logging
import argparse
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 auth():
    post_data_login = {
    "jsonrpc" : "2.0",
    "method" : "user.login",
    "params" : {
        "user" : "admin",
        "password" : "admin12345"
    },
    "id" : 1
}
    """获取 zabbix 登录令牌"""
    ret = requests.post(url, data = json.dumps(post_data_login), headers = post_headers)
    if 'result' in ret.text: 
        login_code = json.loads(ret.text).get("result")
        logger_getter().info('auth success! , zabbix login code: %s' %login_code)
        return login_code
    elif 'error' in ret.text:
        logger_getter().error('auth fails! , exiting')
        sys.exit(-1) 

def switch_zabbix_action(auth_code, action_id, action_flag):
    post_update_action = {
    "jsonrpc": "2.0",
    "method": "action.update",
    "params": {
        "actionid": action_id,
        "status": action_flag
    },
    "auth": auth_code,
    "id": 1
   }

    ret = requests.post(url, data = json.dumps(post_update_action), headers = post_headers)
    result=json.loads(ret.text)
    logger_getter().debug(result)

    if 'result' in ret.text:
        if action_flag == '1':
            print('%s %s disable success!' %(action_flag, action_id))
            logger_getter().info('%s %s disable success!' %(action_flag, action_id))
        if action_flag == '0':
            print('%s %s enable success!' %(action_flag, action_id))
            logger_getter().info('%s %s enable success!' %(action_flag, action_id))
    elif 'error' in ret.text:
        print('%s %s fail! exiting' %(action_flag, action_id))
        logger_getter().error('%s %s fail! exiting' %(action_flag, action_id))
        sys.exit(-3)

if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description="A tool to switch zabbix action's status.")

    parser.add_argument(
        '-E', '--enable',
        action='store_true',
        help="enable the zabbix's action")

    parser.add_argument(
        '-D', '--disable',
        action='store_true',
        help="disable the zabbix's action")
    args = parser.parse_args()

    
    url = 'http://192.168.0.1/zabbix/api_jsonrpc.php'
    post_headers = {'Content-Type': 'application/json'}

    auth_code = auth()
    if args.disable:
        for action_id in action_id_list:
            switch_zabbix_action(auth_code, action_id, '1')
        sys.exit(0)
    if args.enable:
        for action_id in action_id_list:
            switch_zabbix_action(auth_code, action_id, '0')
        sys.exit(0)

100

101

102

103

#!/usr/bin/env python

# encoding=utf-8

import logging

import argparse

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 auth():

post_data_login = {

"jsonrpc" : "2.0",

"method" : "user.login",

"params" : {

"user" : "admin",

"password" : "admin12345"

"id" : 1

}

"""获取 zabbix 登录令牌"""

ret = requests.post(url, data = json.dumps(post_data_login), headers = post_headers)

if 'result' in ret.text:

login_code = json.loads(ret.text).get("result")

logger_getter().info('auth success! , zabbix login code: %s' %login_code)

return login_code

elif 'error' in ret.text:

logger_getter().error('auth fails! , exiting')

sys.exit(-1)

def switch_zabbix_action(auth_code, action_id, action_flag):

post_update_action = {

"jsonrpc": "2.0",

"method": "action.update",

"params": {

"actionid": action_id,

"status": action_flag

"auth": auth_code,

"id": 1

}

ret = requests.post(url, data = json.dumps(post_update_action), headers = post_headers)

result=json.loads(ret.text)

logger_getter().debug(result)

if 'result' in ret.text:

if action_flag == '1':

print('%s %s disable success!' %(action_flag, action_id))

logger_getter().info('%s %s disable success!' %(action_flag, action_id))

if action_flag == '0':

print('%s %s enable success!' %(action_flag, action_id))

logger_getter().info('%s %s enable success!' %(action_flag, action_id))

elif 'error' in ret.text:

print('%s %s fail! exiting' %(action_flag, action_id))

logger_getter().error('%s %s fail! exiting' %(action_flag, action_id))

sys.exit(-3)

if __name__ == '__main__':

parser = argparse.ArgumentParser(

description="A tool to switch zabbix action's status.")

parser.add_argument(

'-E', '--enable',

action='store_true',

help="enable the zabbix's action")

parser.add_argument(

'-D', '--disable',

action='store_true',

help="disable the zabbix's action")

args = parser.parse_args()

url = 'http://192.168.0.1/zabbix/api_jsonrpc.php'

post_headers = {'Content-Type': 'application/json'}

auth_code = auth()

if args.disable:

for action_id in action_id_list:

switch_zabbix_action(auth_code, action_id, '1')

sys.exit(0)

if args.enable:

for action_id in action_id_list:

switch_zabbix_action(auth_code, action_id, '0')

sys.exit(0)

使用方式：

在自行替换上述脚本中的zabbix server的ip和账号密码后执行下述命令：

# 禁用
$ python xxx.py -D

# 启用
$ python xxx.py -E

# 禁用

$ python xxx.py -D

# 启用

$ python xxx.py -E

本作品采用知识共享署名 4.0 国际许可协议进行许可