实现在自动发布前后禁用或启用zabbix触发器

实现在自动发布前后禁用或启用zabbix触发器

背景

正常来讲,我们在自动化发布的过程中难免会造成告警风暴,那么我们可以做些什么来规避这个问题呢,一个比较简单的方式就是在发布前禁用掉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...

 

0 评论 在 “实现在自动发布前后禁用或启用zabbix触发器

发表评论

电子邮件地址不会被公开。 必填项已用*标注

captcha