[系统运维]cmdb事件推送实现zabbix资产同步

前言

《事件推送网关：让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一一对应。

通过分组同步最终实现的效果如下：
![在这里插入图片描述](https://img-blog.csdnimg.cn/20210714224139923.png#pic_center

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一侧就可以管理主机资源，不仅节省了我们跨平台处理的时间，而且规范了我们基础设施的管理。