前言
《事件推送網關：讓cmdb告别“花瓶”》一文我們将cmdb事件推送的參數進行了解析，并結合redis做了去重，避免事件的重複推送。接下來的任務我們計劃是cmdb和zabbix進行資産同步，功能如下：
- cmdb的業務、叢集、子產品分組資訊同步至zabbix主機分組；
- cmdb資産删除同步至zabbix進行相應的主機删除；
- cmdb資産建立同步至zabbix，并根據相應的資訊綁定不同的模闆；
此時cmdb隻是為zabbix提供了資産建立、分組、變更等基礎資料，zabbix的告警資訊如何與其關聯，讓運維能夠通過告警内容就能直接知道哪塊業務出現問題才是我們真正想要的。
為此，我們做了如下定義：
- 定義zabbix的分組規則為：業務_叢集_子產品；
- zabbix agent的Hostname參數統一使用ip位址，友善事件推送網關通過zabbix api根據進行查詢及其他相關操作；
- 定義zabbix 主機的可見名稱規則為：叢集_子產品_ip；由于我們的叢集基本都放在同一個業務下，可見名稱規則中不加業務所屬；
通過以上規則，我們可以将告警主機的ip、分組、可見名稱自由添加到告警動作中，實作了告警資訊的可讀性。
其中：
- 叢集：中間件；
- 子產品：nginx-lan；
- IP：10.164.193.189；
通過告警内容，就可以快速定位問題，後期可結合故障自愈實作，更好的提高我們處理問題的效率。
目錄結構
開發過程中随着功能的不斷增多，我們需要對不同情況生成相應格式化參數。
D:\WORK\BLUEKING\GATEWAY\GATEWAY
│ manage.py
│
├─.vscode
│ settings.json
│
├─gateway
│ asgi.py
│ settings.py
│ urls.py
│ wsgi.py
│ __init__.py
│
└─gw_cmdb
│ admin.py
│ apps.py
│ models.py
│ tests.py
│ urls.py
│ views.py
│ __init__.py
│
├─common
│ hostidentifier_cmdb.py
│ main.py
│ module_cmdb.py
│ select_cmdb.py
└─zabbix
group.py
host.py
main.py
template.py
其中：
common目錄是與cmdb相關的子產品：
- main 接收事件推送網關推送過來的參數；
- hostidentifier_cmdb 針對主機相關事件推送傳回格式化參數；
- module_cmdb 針對子產品相關事件推送傳回格式化參數；
- select_cmdb 查詢cmdb内容如叢集、業務、作業系統等輔助資訊；
zabbix目錄是與zabbix相關的子產品：
- main 解析格式化參數，對zabbix做相應的處理；
- group 與主機分組操作相關的子產品，如查詢、建立、删除、更新等；
- host 與主機操作相關子產品，如查詢、建立、更新、删除等；
- template 與模闆操作相關的子產品，如查詢等；
由于分組同步可以在不影響zabbix使用的情況下操作，是以我們在此着重介紹此功能。
分組同步
分組同步主要有以下幾個功能需求：
- 業務、叢集、子產品分組同步，如：建立、更新、删除，對應分組名為：業務_叢集_子產品；
- 主機子產品更新同步，主機子產品轉移需要在zabbix中修改主機相應的分組，保持和cmdb一緻；
注意：分組同步的前提是通過zabbix的IP查詢并操作主機，是以要求zabbix agent 的Hostname必須為IP，并且和cmdb中的IP一一對應。
通過分組同步最終實作的效果如下：
1.接收cmdb推送參數
cmdb事件推送将參數發送給網關，由views.py接收
vim gw_cmdb/views.py
from django.http import HttpResponse
from .common.main import main
from .zabbix.main import zabbix_main
from .jumpserver.main import jumpserver_main
import json
import logging
logger = logging.getLogger('log')
# Create your views here
def cmdb_request(request):
if request.method == 'POST':
data = json.loads(request.body)
logger.info('cmdb發送消息：{}'.format(data))
## 整理cmdb資料
res=main(data)
##是否需要關聯zabbix及jumpserver
if res['result'] == 1:
return HttpResponse("ok")
else:
logger.info(res)
#zabbix同步
zabbix_request = zabbix_main(res)
logger.info('zabbix 同步完畢:{}'.format(zabbix_request))
#jumpserver同步
return HttpResponse("ok")
else:
logger.info('本接口隻支援POST模式')
return HttpResponse("本接口隻支援POST模式")
由于cmdb推送過來的參數内容過多，是以我們需要将其進行格式化并且将其解析成不同的動作，以便zabbix進行操作。
2.解析參數
（1）common将views接收的請求進行。
vim common/main.py
import logging
from .hostidentifier_cmdb import hostidentifier
from .module_cmdb import module_action
logger = logging.getLogger('log')
def main(data):
result = {'result': 1,'data': 0}
## 子產品操作
if data['obj_type'] == 'module':
return module_action(data)
## 主機辨別操作
elif data['obj_type'] == 'hostidentifier':
if data['action'] == 'update' and data['event_type'] == 'relation' :
logger.info("主機辨別更新: {}".format(data))
return hostidentifier(data)
else:
logger.info("主機辨別未知操作: {}".format(data))
else:
logger.info("未知操作: {}".format(data))
return result
注意：cmdb側事件推送的觸發動作比較多，如主機、業務、組織架構等，但其實在zabbix側我們沒有必要一一和以上事件進行對應，而是隻需要和最小操作保持一緻即可。
舉例：
- 根據分組規則：業務_叢集_子產品，最小級的子產品存在的前提肯定是有叢集和業務，是以隻要最小級的子產品建立、更新、删除時，觸發zabbix動作建立主機群組即可，而空業務、空叢集對zabbix操作沒有實際意義。
- 對于錄入cmdb的主機，無論是在資源池或業務下的空閑資源池，此時都沒有必要觸發zabbix操作，而是當主機轉移到相應的子產品下，辨別已經配置設定使用了，此時我們将觸發zabbix操作，如建立主機、綁模闆、增加主機分組等操作。
是以zabbix側動作觸發，隻需要在子產品和主機辨別更新情況下觸發即可。
（2）子產品操作解析
vim common/module_cmdb.py
import json
import logging
from .select_cmdb import select_cmdb_biz,select_cmdb_set
logger = logging.getLogger('log')
def module_action(data):
logger.info("子產品更新: {}".format(data))
##定義資料格式
datajson= {'action': '','obj_type': '','data': {'cur_data': {'group': ''},'pre_data': {'group': ''}},'result': 1}
if data['action'] == "create":
pass
elif data['action'] == "update":
datajson['obj_type']=data['obj_type']
datajson['action']= data['action']
groupname_biz_name = select_cmdb_biz(data['data'][0]['cur_data']['bk_biz_id'])
groupname_set_name = select_cmdb_set(data['data'][0]['cur_data']['bk_biz_id'], data['data'][0]['cur_data']['bk_set_id'])
groupname_module_new_name = data['data'][0]['cur_data']['bk_module_name']
groupname_module_old_name = data['data'][0]['pre_data']['bk_module_name']
logger.info("業務id:{},叢集id:{},子產品新名:{},子產品舊名:{}".format(groupname_biz_name,groupname_set_name,groupname_module_new_name,groupname_module_old_name))
if groupname_module_new_name != groupname_module_old_name:
datajson['data']['cur_data']['group']= groupname_biz_name+"_"+groupname_set_name+"_"+groupname_module_new_name
datajson['data']['pre_data']['group']= groupname_biz_name+"_"+groupname_set_name+"_"+groupname_module_old_name
result = {
'result': 0,
'data': datajson
}
return result
elif data['action'] == "delete":
datajson['obj_type']=data['obj_type']
datajson['action']= data['action']
groupname_biz_name = select_cmdb_biz(data['data'][0]['pre_data']['bk_biz_id'])
groupname_set_name = select_cmdb_set(data['data'][0]['pre_data']['bk_biz_id'], data['data'][0]['pre_data']['bk_set_id'])
groupname_module_name = data['data'][0]['pre_data']['bk_module_name']
datajson['data']['pre_data']['group']= groupname_biz_name+"_"+groupname_set_name+"_"+groupname_module_name
result = {
'result': 0,
'data': datajson
}
return result
else:
pass
return datajson
（3）主機辨別解析
cmdb主機子產品轉移可能會觸發多次請求，是以我們需要借助redis将請求進行去重。
vim common/hostidentifier_cmdb.py
import redis
import json
import hashlib
import logging
logger = logging.getLogger('log')
r = redis.StrictRedis(host='127.0.0.1',port=6379,db=1)
## 子產品變更擷取主機所有模闆定制分組
def hostidentifier(data):
##定義資料格式
datajson= {'tex_id': '','action': data['action'],'obj_type': data['obj_type'],'data': {'cur_data': {'ip': '','group': []},'bk_host_id':data['data']['cur_data']['bk_host_id'],'pre_data': 'None'},'result': 1}
## 擷取主機組資訊，并清理記錄reids去除重複會話
for i in data['data']:
datajson['data']['cur_data']['ip'] = i['cur_data']['bk_host_innerip']
grouplist = i['cur_data']['associations']
for j in grouplist:
groupname = grouplist[j]['bk_biz_name']+"_"+grouplist[j]['bk_set_name']+"_"+grouplist[j]['bk_module_name']
datajson['data']['cur_data']['group'].append(groupname)
datajson['tex_id']= hashlib.md5((data['request_id']+ i['cur_data']['bk_host_innerip']).encode('utf-8')).hexdigest()
rkey = r.hget('cmdb',datajson['tex_id'])
logger.info(rkey)
if rkey is None:
r.hset('cmdb',datajson['tex_id'],json.dumps(datajson['data']))
datajson['result'] = 0
logger.info(datajson)
return datajson
3.zabbix操作
當common将cmdb事件的參數解析後，就可傳給zabbix子產品進行相應的分組同步操作了。
（1）zabbix操作入口
vim zabbix/main.py
import json
import logging
from django.conf import settings
from urllib import request, parse
from .group import select_zabbix_group,create_zabbix_group,main_zabbix_group, update_zabbix_group, delete_zabbix_group
from .host import select_zabbix_host, select_zabbix_host_group,update_zabbix_host_group
logger = logging.getLogger('log')
def zabbix_auth(ZABBIX_USERNAME,ZABBIX_PASSWORD,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
# auth user and password
data = {
"jsonrpc": "2.0",
"method": "user.login",
"params": {
"user": ZABBIX_USERNAME,
"password": ZABBIX_PASSWORD
},
"id": 1,
}
# 由于API接收的是json字元串，故需要轉化一下
value = json.dumps(data).encode('utf-8')
# 對請求進行包裝
req = request.Request(url, headers=header, data=value)
# 驗證并擷取Auth ID
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("Auth Failed, Please Check Your Name And Password:", e)
else:
response = result.read()
# 上面擷取的是bytes類型資料，故需要decode轉化成字元串
page = response.decode('utf-8')
# 将此json字元串轉化為python字典
page = json.loads(page)
result.close()
logger.info("Auth Successful. The Auth ID Is: {}".format(page.get('result')))
return page.get('result')
def zabbix_main(result):
# 擷取zabbix 認證
auth_code = zabbix_auth(settings.ZABBIX_USERNAME,settings.ZABBIX_PASSWORD,settings.ZABBIX_URL)
logger.info("zabbix授權id:{}".format(auth_code))
# 子產品操作
if result['obj_type'] == 'module':
#####zabbix舊組是否存在，不存在直接建立新組
result_data = select_zabbix_group(result['data']['pre_data']['group'],auth_code,settings.ZABBIX_URL)
###zabbix建立
if result['action'] == 'create' :
logger.info("zabbix組建立: {}".format(result))
###zabbix修改
elif result['action'] == 'update' and result['data']['pre_data'] != None:
if len(result_data) == 0:
logger.info("zabbix組舊子產品不存在: {}".format(result['data']['pre_data']))
res = create_zabbix_group(result['data']['cur_data']['group'],auth_code,settings.ZABBIX_URL)
if "err" in res:
logger.error('zabbix新增加組失敗:{}'.format(result['data']['cur_data']))
else:
logger.info('zabbix新增加組成功:{}'.format(result['data']['cur_data']['group']))
return result['data']['cur_data']['group']
else:
res = update_zabbix_group(result_data[0]['groupid'],result['data']['cur_data']['group'],auth_code,settings.ZABBIX_URL)
logger.info('zabbix組修改完畢:{}'.format(result['data']['cur_data']['group']))
return result['data']['cur_data']['group']
###子產品删除
elif result['action'] == 'delete':
logger.info("zabbix組删除: {}".format(result))
result_data = select_zabbix_group(result['data']['pre_data']['group'],auth_code,settings.ZABBIX_URL)
if len(result_data) == 0:
logger.info("zabbix組舊子產品不存在: {}".format(result['data']['pre_data']))
else:
res = delete_zabbix_group(result_data[0]['groupid'],auth_code,settings.ZABBIX_URL)
logger.info("zabbix組删除完畢: {}".format(result))
###子產品其他操作
else:
logger.info("子產品未知操作: {}".format(result))
## 主機辨別操作
elif result['obj_type'] == 'hostidentifier':
if result['action'] == 'update':
logger.info("主機辨別更新: {}".format(result))
result_Groupidlist = main_zabbix_group(result,auth_code)
if result_Groupidlist['result'] == 0:
res = update_zabbix_host_group(result_Groupidlist['hostid'],result_Groupidlist['grouplistid'],auth_code,settings.ZABBIX_URL)
logger.info('資料同步成功：{}'.format(res))
else:
logger.info("主機辨別未知操作: {}".format(result))
else:
logger.info("未知操作: {}".format(result))
return result
（2）主機群組操作
vim zabbix/group.py
import json
import logging
from urllib import request, parse
from django.conf import settings
from .template import select_zabbix_template
from .host import create_zabbix_host, select_zabbix_host
from common.select_cmdb import select_cmdb_host
logger = logging.getLogger('log')
def select_zabbix_group(Groupname,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"method": "hostgroup.get",
"params": {
"output": "extend",
"filter": {
"name": Groupname
}
},
"auth": auth_code,
"id": 1
}
value = json.dumps(data).encode('utf-8')
req = request.Request(url, headers=header, data=value)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
result.close()
logger.info("group_result: {}".format(response['result']))
return response['result']
def create_zabbix_group(Groupname,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"method": "hostgroup.create",
"params": {
"name": Groupname
},
"auth": auth_code,
"id": 1
}
value = json.dumps(data).encode('utf-8')
logger.info("zabbix組建立資訊: {}".format(value))
req = request.Request(url, headers=header, data=value)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("主機組結構體建立失敗:", e)
else:
response = json.loads(result.read().decode('utf-8'))
result.close()
logger.info("group_result: {}".format(response))
return response
def update_zabbix_group(Groupid,Groupname,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"method": "hostgroup.update",
"params": {
"groupid": Groupid,
"name": Groupname,
},
"auth": auth_code,
"id": 1
}
value = json.dumps(data).encode('utf-8')
logger.info("zabbix組資訊修改如下: {}".format(value))
req = request.Request(url, headers=header, data=value)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
result.close()
logger.info("group_result: {}".format(response['result']))
def delete_zabbix_group(Groupid,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"method": "hostgroup.delete",
"params": [Groupid],
"auth": auth_code,
"id": 1
}
value = json.dumps(data).encode('utf-8')
logger.info("zabbix組删除資訊如下: {}".format(value))
req = request.Request(url, headers=header, data=value)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
result.close()
logger.info("group_result: {}".format(response['result']))
def main_zabbix_group(result,auth_code):
zabbix_host_group_id = {'hostid':'','grouplistid':[],'zabbix_host_name':'','result':1}
Groupidlist = []
##擷取zabbix中的組是否存在
for i in result['data']['cur_data']['group']:
zabbix_host_name= "[" +i+ "_" + result['data']['cur_data']['ip'] + "]" + zabbix_host_name
Groupname_result = select_zabbix_group(i,auth_code,settings.ZABBIX_URL)
logger.info(Groupname_result)
if len(Groupname_result) == 0:
res = create_zabbix_group(i,auth_code,settings.ZABBIX_URL)
if "errr" in res:
logger.error('zabbix增加組失敗:{}'.format(i))
else:
logger.info('zabbix增加組:{}'.format(i))
Groupidlist.append(res['result']['groupids'][0])
else:
Groupidlist.append(Groupname_result[0]['groupid'])
logger.info('zabbix分組已存在:{}'.format(i))
# 查詢zabbix 主機資訊擷取組id，及主機id
res = select_zabbix_host( result['data']['cur_data']['ip'],auth_code,settings.ZABBIX_URL)
if len(res) != 0:
logger.info('zabbix主機{}查詢資訊如下:{}'.format(result['data']['cur_data']['ip'],res))
# 增加原始組id
#for i in res[0]['groups']:
# Groupidlist.append(i['groupid'])
zabbix_host_group_id['hostid'] = res[0]['hostid']
zabbix_host_group_id['grouplistid'] = Groupidlist
zabbix_host_group_id['result'] = 0
return zabbix_host_group_id
else:
logger.info('zabbix主機{}查詢資訊為空:{}'.format(result['data']['cur_data']['ip']))
cmdb_host_os_type = select_cmdb_host(result['data']['cur_data']['bk_host_id'])
zabbix_template_name = "Template OS " + cmdb_host_os_type
zabbix_template_id = select_zabbix_template(zabbix_template_name,auth_code,settings.ZABBIX_URL)
logger.info('zabbix建立主機：{},主機系統類型：{}，主機模闆id：{}'.format(result['data']['cur_data']['ip'],cmdb_host_os_type,zabbix_template_id))
res = create_zabbix_host(result['data']['cur_data']['ip'],zabbix_host_name,Groupidlist,zabbix_template_id,auth_code,settings.ZABBIX_URL)
return zabbix_host_group_id
（3）主機操作
vim zabbix/host.py
import json
import logging
from urllib import request, parse
logger = logging.getLogger('log')
def select_zabbix_host(Host,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = json.dumps({
"jsonrpc": "2.0",
"method": "host.get",
"params": {
"output": ["hostid","name","status","host","groupids"],
"selectGroups":"extend",
"selectInterfaces":["ip"],
"filter":{"host":Host}
},
"id": 1,
"auth": auth_code}).encode('utf-8')
req = request.Request(url, headers=header, data=data)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
result.close()
logger.info("group_result: {}".format(response['result']))
return response['result']
def select_zabbix_host_group(Host,auth_code,ZABBIX_URL):
group_list = []
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = json.dumps({
"jsonrpc": "2.0",
"method": "host.get",
"params": {
"output": ["hostid","name","status","host","groupids"],
"selectGroups":"extend",
"selectInterfaces":["ip"],
"filter":{"host":Host}
},
"id": 1,
"auth": auth_code
}).encode('utf-8')
req = request.Request(url, headers=header, data=data)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
# 上面擷取的是bytes類型資料，故需要decode轉化成字元串
result.close()
logger.info("group_result: {}".format(response['result']))
if (response != 0) and (len(response) != 0):
groups = response['result'][0]['groups']
for group in groups:
var = {}
var['groupid'] = group['groupid']
group_list.append(var)
# return host['groups'][0]['groupid']
return group_list
else:
return ""
return group_list
def update_zabbix_host():
pass
def update_zabbix_host_group(Hostid,Host_Group_list,auth_code, ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = json.dumps({
"jsonrpc": "2.0",
"method": "host.update",
"params": {
# "hostid": '10202',
"hostid": Hostid,
"groups": Host_Group_list,
"status": 0,
},
"auth": auth_code,
"id": 1
}).encode('utf-8')
logger.info("更新資料json如下：{}".format(data))
req = request.Request(url, headers=header, data=data)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
logger.info("傳回資料如下：{}".format(response))
result.close()
if 'error' in response.keys():
logger.info("host組資訊修改失敗")
else:
return response['result']
def create_zabbix_host(Hostip,Hostname,Host_Group_list,templateid,auth_code,ZABBIX_URL):
url = "{}/api_jsonrpc.php".format(ZABBIX_URL)
header = {"Content-Type": "application/json"}
data = json.dumps({
"jsonrpc": "2.0",
"method": "host.create",
"params": {
"host": Hostip,
"name": Hostname,
"interfaces": [
{
"type": 1,
"main": 1,
"useip": 1,
"ip": Hostip,
"dns": "",
"port": "10050"
}
],
"groups": Host_Group_list,
"templates": [
{
"templateid": templateid
}
]
},
"id": 1,
"auth": auth_code}).encode('utf-8')
logger.info("更新資料json如下：{}".format(data))
req = request.Request(url, headers=header, data=data)
try:
# 打開包裝過的url
result = request.urlopen(req)
except Exception as e:
logger.error("result封裝：", e)
else:
response = json.loads(result.read().decode('utf-8'))
logger.info("傳回資料如下：{}".format(response))
result.close()
if 'error' in response.keys():
logger.info("zabbix主機建立失敗：{}".format(response))
else:
return response['result']
return response['result']
總結
通過事件推送網關我們初步實作了cmdb和zabbix的分組同步，我們隻需要在cmdb一側就可以管理主機資源，不僅節省了我們跨平台處理的時間，而且規範了我們基礎設施的管理。