背景
正常来讲，我们在自动化发布的过程中难免会造成告警风暴，那么我们可以做些什么来规避这个问题呢，一个比较简单的方式就是在发布前禁用掉zabbix的触发器，发布完再启用，本篇博客就是在这个背景下进行创作的，本次用到的脚本如下所示：
~/Downloads/tmp tree
.
├── main.py
├── zhtrigfinder.py
└── ztrigswitcher.py
0 directories, 3 files代码分两块，一个是main.py，一个是github上的开源脚本zhtrigfinder.py和ztrigswitcher.py，仓库名叫q1x/zabbix-gnomes，感谢，该仓库年久失修，为了防止失联本人fork了一下，也在下面把源码直接贴了出来，请把两块代码放到同一级目录。
整个项目依赖python2，主要是这个七八年前的上古开源脚本zhtrigfinder.py和ztrigswitcher.py也依赖python2。
依赖安装
python2 -m pip install pyzabbix==1.0.0 requests==2.26.0使用
上述脚本的后两个脚本依赖一个叫zbx_prod.conf的下面格式的配置文件：
[Zabbix API]
username=johndoe
password=verysecretpassword
api=https://zabbix.mycompany.com/path/to/zabbix/frontend/
no_verify=true主代码main.py命令行参数如下：
~/Downloads/tmp python main.py -h
usage: main.py [-h] [-k KEYWORDS] [-D] [-E]
A wrapper for zabbix tool.
optional arguments:
-h, --help show this help message and exit
-k KEYWORDS, --keywords KEYWORDS
search keywords
-D, --disabled disable the trigger
-E, --enabled enable the trigger主代码main.py的运行方式如下:
python2 main.py -D -k 'XXX-DOCKER-WEB'
python2 main.py -E -k 'XXX-DOCKER-WEB'代码
main.py如下：
#!/usr/bin/env python2
# -*- encoding: utf-8 -*-
import sys
import time
import logging
import argparse
import subprocess
import traceback
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('/data/logs/debug.log' + '.' + today,encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
def check(keywords, host_ip):
try:
logger_getter().debug('Finding the system of {0} on host {1}'.format(keywords,host_ip))
trigger_id_list = subprocess.check_output("python2 /data/srv/salt/zhtrigfinder.py -c /data/srv/salt/zbx_prod.conf -s '{0}' -n '{1}'".format(keywords, host_ip), shell=True)
return trigger_id_list.strip()
except subprocess.CalledProcessError:
logger_getter().debug('Not found the system of {0} on host {1}'.format(keywords,host_ip))
return ''
def take_action(keywords, host_ip_list):
count = 0
# 命中host机器
target_host_ip = ''
for host_ip in host_ip_list:
trigger_id_list = check(keywords, host_ip)
if trigger_id_list:
count = count + 1
target_host_ip = host_ip
trigger_id_list = [id_ for id_ in trigger_id_list.strip().split('\n')]
for id_ in trigger_id_list:
try:
if args.disabled:
switcher_result = subprocess.check_output("python2 /data/srv/salt/ztrigswitcher.py -c /data/srv/salt/zbx_prod.conf -D -t " + str(id_), shell=True).strip()
if switcher_result == 'Disabled':
logger_getter().debug("1st: The trigger of {0}'s disable on host {1} is confirmed OK by API status!".format(keywords, host_ip))
elif args.enabled:
switcher_result = subprocess.check_output("python2 /data/srv/salt/ztrigswitcher.py -c /data/srv/salt/zbx_prod.conf -E -t " + str(id_),shell=True).strip()
if switcher_result == 'Enabled':
logger_getter().debug("1st: The trigger of {0}'s enable on host {1} is confirmed OK by both db and API status, and the value is 0!".format(keywords, host_ip))
except subprocess.CalledProcessError:
logger_getter().debug("Exception: Unknown error happened for switch the trigger status, the keywords is {0} and the host ip is {1}".format(args.keywords, host_ip))
logger_getter().debug(traceback.format_exc())
sys.exit(0)
break
else:
continue
if count == 0:
logger_getter().debug('Exception: no trigger id is found on all hosts for the keywords of {0}.'.format(keywords))
else:
if args.disabled:
logger_getter().debug("The system's trigger of {0} is disabled on {1}!".format(keywords, target_host_ip))
elif args.enabled:
logger_getter().debug("The system's trigger of {0} is enabled on {1}!".format(keywords,target_host_ip))
def main(args):
keywords = args.keywords
host_ip_list = ['192.168.1.1','192.168.1.2']
take_action(keywords, host_ip_list)
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description='A wrapper for zabbix tool.')
parser.add_argument(
'-k','--keywords',
help='search keywords',
type=str)
parser.add_argument(
'-D','--disabled',
help='disable the trigger',
action='store_true')
parser.add_argument(
'-E','--enabled',
help='enable the trigger',
action='store_true')
args = parser.parse_args()
main(args)
整个代码你只需要关注并修改一下logging库写日志的路径，以及前面提到的上古脚本的放置路径，以及main()中的host_ip_list = [‘192.168.1.1’,‘192.168.1.2’]，自行修改为你们生产环境的zabbix agent探测机器的节点即可。
zhtrigfinder.py如下：
#!/usr/bin/env python
#
# import needed modules.
# pyzabbix is needed, see https://github.com/lukecyca/pyzabbix
#
import argparse
import ConfigParser
import os
import os.path
import sys
import distutils.util
from pyzabbix import ZabbixAPI
# define config helper function
def ConfigSectionMap(section):
dict1 = {}
options = Config.options(section)
for option in options:
try:
dict1[option] = Config.get(section, option)
if dict1[option] == -1:
DebugPrint("skip: %s" % option)
except:
print("exception on %s!" % option)
dict1[option] = None
return dict1
# set default vars
defconf = "~/.zbx.conf"
username = ""
password = ""
api = ""
noverify = ""
# Define commandline arguments
parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,description='Tries to find triggers configured for the specified Zabbix host.', epilog="""
This program can use .ini style configuration files to retrieve the needed API connection information.
To use this type of storage, create a conf file (the default is $HOME/.zbx.conf) that contains at least the [Zabbix API] section and any of the other parameters:
[Zabbix API]
username=johndoe
password=verysecretpassword
api=https://zabbix.mycompany.com/path/to/zabbix/frontend/
no_verify=true
""")
group = parser.add_mutually_exclusive_group(required=False)
group2 = parser.add_mutually_exclusive_group(required=False)
parser.add_argument('hostname', help='Hostname to find the configured triggers on')
parser.add_argument('-u', '--username', help='User for the Zabbix api')
parser.add_argument('-p', '--password', help='Password for the Zabbix api user')
parser.add_argument('-a', '--api', help='Zabbix API URL')
parser.add_argument('--no-verify', help='Disables certificate validation when using a secure connection',action='store_true')
parser.add_argument('-c','--config', help='Config file location (defaults to $HOME/.zbx.conf)')
group.add_argument('-n', '--numeric', help='Return numeric triggerids instead of descriptions',action='store_true')
group.add_argument('-e', '--extended', help='Returns trigger id, value, status, state, severity, description and expression separated by ":". See https://www.zabbix.com/documentation/2.2/manual/api/reference/trigger/object for more information.',action='store_true')
group2.add_argument('-s', '--search', help='Show only triggers with a description containing this search string')
group2.add_argument('-A', '--active', help='Show only active triggers',action='store_true')
args = parser.parse_args()
# load config module
Config = ConfigParser.ConfigParser()
Config
# if configuration argument is set, test the config file
if args.config:
if os.path.isfile(args.config) and os.access(args.config, os.R_OK):
Config.read(args.config)
# if not set, try default config file
else:
if os.path.isfile(defconf) and os.access(defconf, os.R_OK):
Config.read(defconf)
# try to load available settings from config file
try:
username=ConfigSectionMap("Zabbix API")['username']
password=ConfigSectionMap("Zabbix API")['password']
api=ConfigSectionMap("Zabbix API")['api']
noverify=bool(distutils.util.strtobool(ConfigSectionMap("Zabbix API")["no_verify"]))
except:
pass
# override settings if they are provided as arguments
if args.username:
username = args.username
if args.password:
password = args.password
if args.api:
api = args.api
if args.no_verify:
noverify = args.no_verify
# test for needed params
if not username:
sys.exit("Error: API User not set")
if not password:
sys.exit("Error: API Password not set")
if not api:
sys.exit("Error: API URL is not set")
# Setup Zabbix API connection
zapi = ZabbixAPI(api)
if noverify is True:
zapi.session.verify = False
# Login to the Zabbix API
zapi.login(username, password)
##################################
# Start actual API logic
##################################
# set the hostname we are looking for
host_name = args.hostname
hosts = zapi.host.get(output="extend", filter={"host": host_name})
if hosts:
# Find triggers
if args.search:
triggers = zapi.trigger.get(filter={'host':host_name},output='extend',search={'description':args.search},expandExpression=1,expandDescription=1)
elif args.active:
triggers = zapi.trigger.get(filter={'host':host_name,'value':1},output='extend',monitored=1,active=1,expandExpression=1,expandDescription=1)
else:
triggers = zapi.trigger.get(filter={'host':host_name},output='extend',expandExpression=1,expandDescription=1)
if triggers:
if args.extended:
# print(ids and descriptions)
for trigger in triggers:
print(format(trigger["triggerid"])+":"+format(trigger["value"])+":"+format(trigger["status"])+":"+format(trigger["state"])+":"+format(trigger["priority"])+":"+format(trigger["description"])+":"+format(trigger["expression"]))
else:
if args.numeric:
# print(ids)
for trigger in triggers:
print(format(trigger["triggerid"]))
else:
# print(descriptions)
for trigger in triggers:
print(format(trigger["description"]))
else:
sys.exit("Error: No matching triggers found on "+ host_name)
else:
sys.exit("Error: Could not find host "+ host_name)
# And we're done...
ztrigswitcher.py如下：
#!/usr/bin/env python
#
# import needed modules.
# pyzabbix is needed, see https://github.com/lukecyca/pyzabbix
#
import argparse
import ConfigParser
import os
import os.path
import sys
import distutils.util
from pyzabbix import ZabbixAPI
# define config helper function
def ConfigSectionMap(section):
dict1 = {}
options = Config.options(section)
for option in options:
try:
dict1[option] = Config.get(section, option)
if dict1[option] == -1:
DebugPrint("skip: %s" % option)
except:
print("exception on %s!" % option)
dict1[option] = None
return dict1
# set default vars
defconf = "~/.zbx.conf"
username = ""
password = ""
api = ""
noverify = ""
# Define commandline arguments
parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,description='Switches the host inventory mode for the specified host(s) or hostgroup(s). The default setting is to switch to "automatic" mode.', epilog="""
This program can use .ini style configuration files to retrieve the needed API connection information.
To use this type of storage, create a conf file (the default is $HOME/.zbx.conf) that contains at least the [Zabbix API] section and any of the other parameters:
[Zabbix API]
username=johndoe
password=verysecretpassword
api=https://zabbix.mycompany.com/path/to/zabbix/frontend/
no_verify=true
""")
group = parser.add_mutually_exclusive_group(required=True)
parser.add_argument('-t','--triggerid', help='Numeric trigger ID to change status on')
parser.add_argument('-u', '--username', help='User for the Zabbix api')
parser.add_argument('-p', '--password', help='Password for the Zabbix api user')
parser.add_argument('-a', '--api', help='Zabbix API URL')
parser.add_argument('--no-verify', help='Disables certificate validation when nventory_mode using a secure connection',action='store_true')
parser.add_argument('-c','--config', help='Config file location (defaults to $HOME/.zbx.conf)')
group.add_argument('-E', '--enable', help='Set trigger to enabled', action='store_true')
group.add_argument('-D', '--disable',help='Set trigger to disabled', action='store_true')
args = parser.parse_args()
# load config module
Config = ConfigParser.ConfigParser()
Config
# if configuration argument is set, test the config file
if args.config:
if os.path.isfile(args.config) and os.access(args.config, os.R_OK):
Config.read(args.config)
# if not set, try default config file
else:
if os.path.isfile(defconf) and os.access(defconf, os.R_OK):
Config.read(defconf)
# try to load available settings from config file
try:
username=ConfigSectionMap("Zabbix API")['username']
password=ConfigSectionMap("Zabbix API")['password']
api=ConfigSectionMap("Zabbix API")['api']
noverify=bool(distutils.util.strtobool(ConfigSectionMap("Zabbix API")["no_verify"]))
except:
pass
# override settings if they are provided as arguments
if args.username:
username = args.username
if args.password:
password = args.password
if args.api:
api = args.api
if args.no_verify:
noverify = args.no_verify
# test for needed params
if not username:
sys.exit("Error: API User not set")
if not password:
sys.exit("Error: API Password not set")
if not api:
sys.exit("Error: API URL is not set")
# Setup Zabbix API connection
zapi = ZabbixAPI(api)
if noverify is True:
zapi.session.verify = False
# Login to the Zabbix API
zapi.login(username, password)
##################################
# Start actual API logic
##################################
if not args.triggerid:
sys.exit("Error: Triggerid not found")
if args.enable:
status=int('0')
elif args.disable:
status=int('1')
else:
sys.exit("Error: Trigger status not provided")
trigger=zapi.trigger.get(filter={'triggerid':args.triggerid},output='triggerid')
if trigger:
result=zapi.trigger.update(triggerid=args.triggerid, status=status)
check=zapi.trigger.get(filter={'triggerid':args.triggerid}, output=['description','status'], expandDescription=1)
if check[0]['status'] == '0':
mode="Enabled"
print(mode)
elif check[0]['status'] == '1':
mode="Disabled"
print(mode)
else:
sys.exit("Error: Something went wrong!")
#print(format(mode) + " : " + format(check[0]['description']))
else:
sys.exit("Error: Trigger not found")
# sys.exit("Error: No trigger id provided.")
# try:
# # Apply the linkage
# result=zapi.host.massupdate(hosts=hlookup,inventory_mode=invm)
# except:
# sys.exit("Error: Something went wrong while performing the update")
#
#if args.extended:
# hosts=zapi.host.get(output='extend',hostids=result['hostids'])
# hostnames=""
# for host in range(len(hosts)):
# if not hostnames:
# hostnames = str(hosts[host]['host'])
# else:
# hostnames = hostnames + ", " + str(hosts[host]['host'])
# print("Inventory mode switched to \"" + args.mode + "\" on: " + hostnames)
#
# And we're done...