有时候被ZABBIX监控的主机可能需要关机重启等维护操作,为了在此期间不触发告警,需要创建主机的维护任务,以免出现误告警
ZABBIX本身有这个API可供调用(不同版本细节略有不同,本次用的ZABBIX6.*),实现批量化建立主机的维护任务
无论哪种方式(IP列表,主机描述,或IP子网)创建维护,都是向maintenance.create这个方法传递参数的过程,除了起始和终止的时间,最主要的一个参数就是hostids这个参数,它是一个列表(也可以是单个hostid)
def create_host_maintenance(self,names,ips,starttimes,stoptimes):starts=self.retimes(starttimes)stops=self.retimes(stoptimes)data = json.dumps({"jsonrpc":"2.0","method":"maintenance.create","params": {"name": names,"active_since": starts,"active_till": stops,#"hostids": [ "12345","54321" ],"hostids": ips,"timeperiods": [{"timeperiod_type": 0,"start_date": starts,#"start_time": 0,zabbix6不用这个参数,低版本的有用"period": int(int(stops)-int(starts))}]},"auth": self.authID,"id": 1,})request = urllib2.Request(self.url, data)for key in self.header:request.add_header(key, self.header[key])try:result = urllib2.urlopen(request)except URLError as e:print "Error as ", eelse:response = json.loads(result.read())result.close()print "maintenance is created!"
1.简单说明hostids这个列表参数产生的过程
按IP列表产生hostids的LIST,并调用方法创建维护
def djmaintenanceiplist(self,ipliststr="strs",area="none",byip="none"):lists = []if area=="none": #通过IP列表来创建维护,只向函数传递ipliststr这个参数,其它参数为空for line in ipliststr.splitlines():con = line.split()ipaddr = con[0].strip()hostids = self.host_getbyip(ipaddr)if hostids != "error" and hostids not in lists:lists.append(hostids)return listselse:if area !="ip" and ipliststr != "strs": #按主机描述匹配创建维护,ipliststr参数因定为strs,area参数为主机描述,byip参数为空(因为网络环境的原因,本例弃用这个条件)sqls="select hostid from zabbixdb.hosts where name like '%"+area+"%' "tests=pysql.conn_mysql()datas=tests.query_mysqlrelists(sqls)if datas != "error":for ids, in datas:if ids not in lists:lists.append(ids)else:if byip != "none": #按主机IP子网创建维护,byip参数不为空(因为网络环境的原因,本例弃用这个条件)sqls = "select hosts.hostid from zabbixdb.hosts,zabbixdb.interface where `hosts`.hostid=interface.hostid and (interface.ip like '%" + byip + "%' or hosts.name like '%" + byip + "%')"tests = pysql.conn_mysql()datas = tests.query_mysqlrelists(sqls)if datas != "error":for ids, in datas:if ids not in lists:lists.append(ids)return listsdef djiplist(self,starttime,stoptime,strlins): #strlins为IP列表的参数,可由WEB前端传递进来test = ZabbixTools()test.user_login()lists = test.djmaintenanceiplist(strlins)nowtime = str(time.strftime('%Y-%m-%d-%H-%M-%S', time.localtime(time.time())))test.create_host_maintenance(nowtime, lists, starttime, stoptime)
按主机描述和IP子网产生hostids的LIST,并调用方法创建维护
def djdescript(self,starttime,stoptime,descstr,ipnets="none"):lists = []if descstr != "ip": #descstr参数不为ip的时候,表示根据主机的描述信息匹配创建维护,此时不传递innets参数for line in descstr.splitlines():con = line.split()descstrnew = con[0].strip()sqls = "select hostid from dbname.hosts where name like '%" + descstrnew + "%' "tests = pysql.conn_mysql()datas = tests.query_mysqlrelists(sqls)if datas != "error":for ids, in datas:if ids not in lists:lists.append(ids)else:if ipnets != "none": #ipnets参数不为空,表示按照IP子网匹配创建维护,此时descstr参数一定为"ip"sqls = "select hosts.hostid from dbname.hosts,dbname.interface where `hosts`.hostid=interface.hostid and (interface.ip like '%" + ipnets + "%' or hosts.name like '%" + ipnets + "%')"tests = pysql.conn_mysql()datas = tests.query_mysqlrelists(sqls)if datas != "error":for ids, in datas:if ids not in lists:lists.append(ids)test = ZabbixTools()test.user_login()nowtime = str(time.strftime('%Y-%m-%d-%H-%M-%S', time.localtime(time.time())))test.create_host_maintenance(nowtime, lists, starttime, stoptime)2.create_host_maintenance和djmaintenanceiplist,及djdescript函数调用方法的说明
时间转换函数self.retimes,将用户传递/的"%Y-%m-%d %H:%M:%S"日期时间转换为时间时间戳
def retimes(self,times):timeArray = time.strptime(times, "%Y-%m-%d %H:%M:%S")timestamp = time.mktime(timeArray)return timestampself.host_getbyip(ipaddr)通过主机IP获取zabbix的hostid,这个方法应用很广泛
函数中的self.authID通过zabbix的API 的"user.login"方法获取,参考ZABBIX官方文档
def host_getbyip(self,hostip):data = json.dumps({"jsonrpc":"2.0","method":"host.get","params":{"output":["hostid","name","status","available"],"filter": {"ip": hostip,"custom_interfaces":0}},"auth":self.authID,"id":1,})request = urllib2.Request(self.url, data)for key in self.header:request.add_header(key, self.header[key])try:result = urllib2.urlopen(request)except URLError as e:if hasattr(e, 'reason'):print 'We failed to reach a server.'print 'Reason: ', e.reasonelif hasattr(e, 'code'):print 'The server could not fulfill the request.'print 'Error code: ', e.codeelse:response = json.loads(result.read())result.close()lens=len(response['result'])if lens > 0:return response['result'][0]['hostid']else:print "error "+hostipreturn "error"pysql.conn_mysql()的实现
#coding:utf-8
import pymysql
class conn_mysql:def __init__(self):self.db_host = 'zabbixdbip'self.db_user = 'dbusername'self.db_passwd = 'dbuserpassword'self.database = "dbname"def query_mysqlrelists(self, sql):conn = pymysql.connect(host=self.db_host, user=self.db_user, passwd=self.db_passwd,database=self.database)cur = conn.cursor()cur.execute(sql)data = cur.fetchall()cur.close()conn.close()#print data# 查询到有数据则进入行判断，row有值且值有效则取指定行数据，无值则默认第一行if data != None and len(data) > 0:return data#调用方式:for ids, in datas:else:return "error"#此方法返回的数据不含数据库字段的列表,如果要返回包含列名(KEY)的字典列表,则:conn = pymysql.connect(host=self.db_host, user=self.db_user, passwd=self.db_passwd,database=self.database,cursorclass=pymysql.cursors.DictCursor) 解决:tuple indices must be integers, not str
3.VIEWS.PY及前端和前端伟递参数的说明
views.py对应的方法
#@login_required
def zabbixmanage(request):if request.method=="POST":uptime = request.POST.get('uptime')downtime= request.POST.get('downtime')UTC_FORMAT = "%Y-%m-%dT%H:%M"utcTime = datetime.datetime.strptime(uptime, UTC_FORMAT)uptime = str(utcTime + datetime.timedelta(hours=0))utcTime = datetime.datetime.strptime(downtime, UTC_FORMAT)downtime = str(utcTime + datetime.timedelta(hours=0))if request.POST.has_key('iplists'):try:sqlstr=request.POST.get("sqlstr")u1=upproxy.ZabbixTools() #前面的python文件名为upproxy.py,类名为ZabbixToolsu1.djiplist(uptime,downtime,sqlstr)except Exception:return render(request,"zbx1.html",{"login_err":"FAILSTEP1"})if request.POST.has_key('descs'):try:sqlstr = request.POST.get("sqlstr")u1 = upproxy.ZabbixTools() u1.djdescript(uptime,downtime,sqlstr)except Exception:return render(request,"zbx1.html",{"login_err":"FAILSTEP2"})if request.POST.has_key('ipnets'):try:sqlstr = request.POST.get("sqlstr")u1 = upproxy.ZabbixTools()u1.djdescript(uptime,downtime,"ip",sqlstr)except Exception:return render(request,"zbx1.html",{"login_err":"FAILSTEP3"})
HTML的简单写法,不太会写,很潦草
<!DOCTYPE html>
<html lang="en">
<head><meta charset="utf-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>ZABBIXMANAGE</title>
</head>
<body><div id="container" class="cls-container"><div id="bg-overlay" ></div><div class="cls-header cls-header-lg"><div class="cls-brand"><h3>ZABBIX主机维护模式</h3><h4>开始时间小于结束时间</h4></div></div><div class="cls-content"><div class="cls-content-sm panel"><div class="panel-body"><form id="loginForm" action="{% url 'zabbixmanage' %}" method="POST"> {% csrf_token %}<div class="form-group"><div class="input-group"><div class="input-group-addon"><i class="fa fa-user"></i></div><textarea type="text" class="form-control" name="sqlstr" placeholder="IP列表,关键字或者IP网段" style="width:300px;height:111px"></textarea></br></div></div></br></br>开始时间:<input type="datetime-local" name="uptime"></br></br>结束时间:<input type="datetime-local" name="downtime"></br></br><p class="pad-btm">按IP列表维护:按行输入IP列表加入维护,一行一个IP,多行输入</p><p class="pad-btm">按关键字匹配:主机描述信息的关键字匹配的加入维护,一般用于虚拟机宿主机和关键字匹配</p><p class="pad-btm">匹配子网关键字维护:IP网段匹配的加入维护,一次填写一个子网,多个子网多次设置,写法示例:172.16.</p><button class="btn btn-success btn-block" type="submit" name="iplists"><b>按IP列表维护一般情况用这个就行</b></button></br></br><button class="btn btn-success btn-block" type="submit" name="descs"><b>按关键字匹配</b></button><button class="btn btn-success btn-block" type="submit" name="ipnets"><b>匹配子网关键字维护</b></button><h4 style="color: red"><b>{{ login_err }}</b></h4></form></div></div></div> </div>
</body>
</html>跑起来大约是这个界面,用户填写主机IP列表,或匹配的描述,或子网的信息,选好时间,点个按钮就能实现批量的维护任务
![[AI]文心一言爆火的同时，ChatGPT-4.5的正确打开方式](http://pic.xiahunao.cn/[AI]文心一言爆火的同时，ChatGPT-4.5的正确打开方式)
)
)
