管理系统 celery采集器的设计

image.png

设计原因

1.由于客户存在多个商业级运维系统，每个运维系统都完全分开，需要每天登入各个运维系统查看是否存在告警。

2.商业级的运维系统毕竟是通用化，无法完美匹配客户的业务需求，需要自己对商业的运维系统做扩展。

3.商业级的运维系统有时候存在一些bug，对于少数设备无法进行监控和采集。

采集信息详情

• 采集存储数据
  • 主机wwn，主机组，LUN组，LUN，端口组，端口等等信息
• 采集光交数据
  • 光交硬件信息，光交wwn信息，光交端口错误，性能信息
• 采集vmware虚拟化数据
  • 采集vmware主机，宿主机，存储信息
• 采集snmp数据
  • 待定
• 接收 snmp trap数据
  • 通过trap服务接口，采集以上设备的告警信息，并写入告警系统
• 分布式存储采集
  • 采集分布式存储的容量信息，卷组，LUN等等信息
• 其他采集
  • 直接通过商业运维系统的数据库进行采集，采集需要的数据进行处理后进行存储。

---

celery的安装及设置

安装celery

1.直接pip install celery进行安装。安装redis数据库，配置redis数据库密码。

2.创建项目目录，在目录下创建 conf文件夹。在conf内创建conf.py文件，存储一些配置信息

3.创建celery_run.py

4.创建其他目录，每个目录对应一套业务系统。

from __future__ import absolute_import,unicode_literals
from celery import Celery,platforms
from celery.schedules import crontab
from conf.conf import logger,redis_ip,redis_port,redis_pass,redis_broker,redis_backend
import sys
from os import path
#允许root用户运行
platforms.C_FORCE_ROOT = True


#设置最高40个任务同时运行
CELERYD_MAX_TASKS_PER_CHILD = 40


#设置本地时间
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = False

#采用redis进行处理
app = Celery(
    'manager',
    #发布任务
    broker='redis://:%s@%s:%s/%s'%(redis_pass,redis_ip,redis_port,redis_broker),
    #返回任务结果
    backend='redis://:%s@%s:%s/%s'%(redis_pass,redis_ip,redis_port,redis_backend),
    #在每个应用目录下创建tasks.py，并将其设置到下面
    include=['Brocade.tasks','fst.tasks','alert_system.tasks'])

# celery配置存放
app.conf.update(
    #返回任务结果存储1小时
    result_expires=3600,
    #设置序列模式
    task_serializer = 'pickle',
    # result_serializer = 'pickle',
    accept_content =['pickle'],
    # accept_content =['application/json', 'pickle'],
    #设置本地时间，启用定时任务时使用
    CELERY_TIMEZONE = 'Asia/Shanghai',
    CELERY_ENABLE_UTC = False,
)

# 定时任务设置
app.conf.beat_schedule = {
    'multiply-at-some-time': {
        'task': 'alert_system.tasks.printtest',
        'schedule': crontab(hour=12, minute=00),   # 每天早上 6 点 00 分执行一次
        'args': ()                                  # 任务函数参数
    }

}
app.conf.timezone = "GMT"
if __name__ == '__main__':
    try:
        app.start()
    except Exception as e:
        logger.warning("启动失败%s"%(e))

创建完成后可以直接进行启动测试。celery -A celery_run worker -l info，启动成功正常。

---

mongodb数据库的设置。

在conf下创建mongodb.py，将数据库的操作全部放置到这里面,其他文件专注业务逻辑。需要先安装monogdb，pip install pymongo

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
from urllib.parse import quote_plus
import sys,datetime


from pymongo import MongoClient

try:
    from collect_manager.conf.conf import mong_ip,mong_user,mong_password,mong_app
except Exception:
    from .conf.conf import mong_ip, mong_user, mong_password, mong_app

#接收log日志使用。
class Logsystem():
    def __init__(self):
        self.obj = monogdb().use_logsystem()
    def warn(self,info):
        data = {
            "grade":"warn",
            "messages":info,
            "datetime":datetime.datetime.now(),
            "confirm": False,
        }
        self.obj.insert(data)
    def error(self,info):
        data = {
            "grade":"error",
            "messages":info,
            "datetime":datetime.datetime.now(),
            "confirm":False,
        }
        self.obj.insert(data)
    def info(self,info):
        data = {
            "grade":"info",
            "messages":info,
            "datetime":datetime.datetime.now(),
            "confirm":True,
        }
        self.obj.insert(data)
    def fail(self,info):
        data = {
            "grade":"fail",
            "messages":info,
            "datetime":datetime.datetime.now(),
            "confirm":False,
        }
        self.obj.insert(data)
        
        
# 登入monogdb使用
class monogdb():
    def __init__(self):
        """
        初始化登入系统
        """
        self.url = "mongodb://%s:%s@%s" % (quote_plus(mong_user),quote_plus(mong_password),mong_ip)
        self.client = MongoClient(self.url,connect=False)
        self.app = mong_app
    def Logindatabase(self):
        """
        返回登入得数据库
        :return:
        """
        dataname = self.client[self.app]
        return dataname
    def use_bro_san_tables(self):
        """
        使用bro_san表
        :return:
        """
        obj = self.Logindatabase()
        return obj["bro_san"]
    def use_soft_log_tables(self):
        obj = self.Logindatabase()
        return obj["soft_log"]
    def use_test_tables(self):
        obj = self.Logindatabase()
        return obj["test"]
    def use_virtual_machine_tables(self):
        obj = self.Logindatabase()
        return obj["virtual_machine"]
    def use_vmwarestorage_tables(self):
        obj = self.Logindatabase()
        return obj["vmware_storage"]
    def use_vmwarehost_tables(self):
        obj = self.Logindatabase()
        return obj["vmware_host"]
    def use_logsystem(self):
        """
            类型：
            messages
            qrade       错误类型，debug info warn  error fail
            is_confirm   是否确认
            is_delete   是否删除
            label       标注，确认是否为重复信息，采集mysql id信息。
            datetime  日期
        :return:
        """
        obj = self.Logindatabase()
        return obj["log_system"]

snmp trap的接收服务设置

负责接收trap的信息，由于 设备种类太多，没必要对MIB做解析，直接写入设备IP到日志系统内。

#!/usr/bin/python
# -*- coding: UTF-8 -*-

import sys


sys.path.append(".")


from pysnmp.entity import engine, config
from pysnmp.proto import api

from pyasn1.codec.ber import decoder
from pysnmp.carrier.asynsock.dgram import udp, udp6

from pysnmp.carrier.asynsock.dispatch import AsynsockDispatcher

from pysnmp.entity.rfc3413 import ntfrcv
try:
    from monogdb_api import Logsystem
except:
    from .monogdb_api import Logsystem

Logsystem = Logsystem()

snmpEngine = engine.SnmpEngine()

def cbFun(transportDispatcher, transportDomain, transportAddress, wholeMsg):
    while wholeMsg:
        msgVer = int(api.decodeMessageVersion(wholeMsg))
        if msgVer in api.protoModules:
            pMod = api.protoModules[msgVer]
        else:
            print('Unsupported SNMP version %s' % msgVer)
            return
        reqMsg, wholeMsg = decoder.decode(
            wholeMsg, asn1Spec=pMod.Message(),
            )
        print('Notification message from %s:%s: ' % (
            transportDomain, transportAddress
            )
        )
        ipdress = transportAddress[0]
        reqPDU = pMod.apiMessage.getPDU(reqMsg)
        if reqPDU.isSameTypeWith(pMod.TrapPDU()):
            if msgVer == api.protoVersion1:
                try:
                    strlist = 'Enterprise: {},Agent Address: {},Generic Trap: {},Specific Trap: {},Uptime: {}'.format(
                        pMod.apiTrapPDU.getEnterprise(reqPDU).prettyPrint(),
                        pMod.apiTrapPDU.getAgentAddr(reqPDU).prettyPrint(),
                        pMod.apiTrapPDU.getGenericTrap(reqPDU).prettyPrint(),
                        pMod.apiTrapPDU.getSpecificTrap(reqPDU).prettyPrint(),
                        pMod.apiTrapPDU.getTimeStamp(reqPDU).prettyPrint()
                        )
                    print(strlist)
                except Exception as e:
                    strlist = ("messages error,error({})").format(e)
                varBinds = pMod.apiTrapPDU.getVarBindList(reqPDU)
            else:
                strlist= "error messages is not"
                varBinds = pMod.apiPDU.getVarBindList(reqPDU)
        Logsystem.error((u"ip ({}) is receive  trap error, Please check the device details immediately.\n {}").format(ipdress,strlist))

            # print('Var-binds:')
            # print(str(varBinds))
    #         for oid, val in varBinds:
    #             #a = oid.prettyPrint().strip()
    #             b = val.prettyPrint().strip().split('\n')
    #             #print(a)
    #             for line in b:
    #                 item = line.strip()
    #                 if item.startswith('string-value'):
    #                     print('string-value='+item.replace('string-value=0x','').decode('hex'))
    #                 else:
    #                     print(item)
    return wholeMsg


if __name__ == '__main__':
    transportDispatcher = AsynsockDispatcher()

    transportDispatcher.registerRecvCbFun(cbFun)

    # UDP/IPv4
    transportDispatcher.registerTransport(
        udp.domainName, udp.UdpSocketTransport().openServerMode(('0.0.0.0', 162))
    )

    # UDP/IPv6
    transportDispatcher.registerTransport(
        udp6.domainName, udp6.Udp6SocketTransport().openServerMode(('::1', 162))
    )

    transportDispatcher.jobStarted(1)

    try:
        # Dispatcher will never finish as job#1 never reaches zero
        transportDispatcher.runDispatcher()
    except:
        transportDispatcher.closeDispatcher()
        raise

存储数据的采集

在目录下创建应用目录storage，并在下面创建对应的子目录，比如华为存储设置huawei,富士通存储设置fst.并且在storage下创建tasks.py。存储数据通过web主动点击的方式进行采集。

设备品牌	采集方式	备注
富士通	ssh	通过ssh进行数据采集
华为	ssh	通过ssh进行数据采集
日立	web	手工采集
EMC	web	手工采集

这个代码相对简单，直接ssh到设备进行采集就行，对于一些低端的设备通过登入web进行采集。

光交数据的采集

每天采集一次，通过ssh的方式进行采集，采集后进行清洗存储。

设备品牌	采集方式	备注
博科	ssh	通过ssh进行数据采集
思科	ssh	通过ssh进行数据采集

vmware虚拟化数据的采集

采集vmware的数据进行存储，对接客户的备份系统，明确查看虚拟化是否成功备份。

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
import atexit
import requests
from pyVmomi import vim
from pyVim import connect
import json
from conf.conf import logger
def sizeof_fmt(num):
    """
    Returns the human readable version of a file size

    :param num:
    :return:
    """
    for item in ['bytes', 'KB', 'MB', 'GB']:
        if num < 1024.0:
            return "%3.1f%s" % (num, item)
        num /= 1024.0
    return "%3.1f%s" % (num, 'TB')
class vsphere:
    def __init__(self,host,user,password,port=443):
        """
        初始化实例，
        :param host:vc主机名
        :param user: vc用户
        :param password: vc密码
        :param port: vc端口，默认443
        """
        try:
            self.service_instance = connect.SmartConnectNoSSL(host=host,user = user,pwd = password,port = int(port))
            atexit.register(connect.Disconnect, self.service_instance)
            self.content = self.service_instance.RetrieveContent()
        except Exception as e:
            logger.error("to vsphere fail - %s"%(e))
    #v1.0
    def getvm(self):
        """
        {'bootTime': None,
        'Bios UUID': '423f330a-a578-2c6e-3fb1-dc514a38184d',
        'Annotation': '',
        'Name': 'GZGL_10.239.37.57',
        'VMware-tools': 'toolsOld',
        'Template': False,
        'memorySizeMB': 16384,
        'numdisk': 2,
        'Path': '[X10_K01_HITACHI_Cluster03_LUN15] GZGL_10.239.37.57/GZGL_10.239.37.57.vmx',
        'IP': 'toolsOld',
        'Instance UUID':
        '503fd0c6-1379-f1d0-c2ce-2a6ca446b34c',
        'Guest': 'Red Hat Enterprise Linux 6 (64-bit)',
        'State': 'poweredOn',
        'numCpu': 4}
        :return: list[dict]
        """
        container = self.content.rootFolder  # starting point to look into
        viewType = [vim.VirtualMachine]  # object types to look for
        recursive = True  # whether we should look into it recursively
        containerView = self.content.viewManager.CreateContainerView(
            container, viewType, recursive)
        children = containerView.view
        data = []
        for child in children:
            summary = child.summary
            try:
                tempdata = {
                "Name":summary.config.name,
                "vm_tag":str(summary.vm),
                "Template":summary.config.template,
                "Path":summary.config.vmPathName,
                "Guest":summary.config.guestFullName,
                "Instance UUID":summary.config.instanceUuid,
                "Bios UUID": summary.config.uuid,
                "Annotation":summary.config.annotation,
                "State":summary.runtime.powerState,
                "VMware-tools":summary.guest.toolsStatus,
                "IP":summary.guest.ipAddress,
                "memorySizeMB":summary.config.memorySizeMB,
                "numCpu":summary.config.numCpu,
                "numdisk":summary.config.numVirtualDisks,
                "bootTime":summary.runtime.bootTime,
                "exsi_tag":str(summary.runtime.host),
                "exsi_ip":summary.runtime.host.name,
                "storage_committed": summary.storage.committed,
                "storage_uncommitted":summary.storage.uncommitted,
                "storage_unshared":summary.storage.unshared,
                "quickStats_status":summary.quickStats.guestHeartbeatStatus,
                "quickStats_uptimeSeconds":summary.quickStats.uptimeSeconds,
                "quickStats_hostMemoryUsage":summary.quickStats.hostMemoryUsage,
            }
            except Exception as e:
                tempdata = {
                    "Name": summary.config.name,
                    "vm_tag": str(summary.vm),
                    "Template": summary.config.template,
                    "Path": summary.config.vmPathName,
                    "Guest": summary.config.guestFullName,
                    "Instance UUID": summary.config.instanceUuid,
                    "Bios UUID": summary.config.uuid,
                    "Annotation": summary.config.annotation,
                    "State": summary.runtime.powerState,
                    "VMware-tools": summary.guest.toolsStatus,
                    "IP": summary.guest.ipAddress,
                    "memorySizeMB": summary.config.memorySizeMB,
                    "numCpu": summary.config.numCpu,
                    "numdisk": summary.config.numVirtualDisks,
                    "bootTime": summary.runtime.bootTime,
                    "exsi_tag": str(summary.runtime.host),
                    "exsi_ip": summary.runtime.host.name,
                    #"storage_committed": summary.storage.committed,
                    #"storage_uncommitted": summary.storage.uncommitted,
                    #"storage_unshared": summary.storage.unshared,
                    "quickStats_status": summary.quickStats.guestHeartbeatStatus,
                    "quickStats_uptimeSeconds": summary.quickStats.uptimeSeconds,
                    "quickStats_hostMemoryUsage": summary.quickStats.hostMemoryUsage,
                }
            data.append(tempdata)
        return data

    def getvm_uuid(self):
        container = self.content.rootFolder  # starting point to look into
        viewType = [vim.VirtualMachine]  # object types to look for
        recursive = True  # whether we should look into it recursively
        containerView = self.content.viewManager.CreateContainerView(
            container, viewType, recursive)
        children = containerView.view
        data = []
        for child in children:
            summary = child.summary
            tempdata = {
                "Instance UUID":summary.config.instanceUuid,
            }
            data.append(tempdata)
        return data
    #废弃
    def getexsihost_storage(self):
        """
        {
            ip:
                磁盘名称。
                    {磁盘信息}

        }
        :return:dict
        """
        objview = self.content.viewManager.CreateContainerView(self.content.rootFolder,
                                                          [vim.HostSystem],
                                                          True)
        esxi_hosts = objview.view
        objview.Destroy()
        datastores = []
        for esxi_host in esxi_hosts:
            # All Filesystems on ESXi host
            # print(esxi_host)
            storage_system = esxi_host.configManager.storageSystem
            host_file_sys_vol_mount_info = storage_system.fileSystemVolumeInfo.mountInfo
            datastore_dict = {}
            # Map all filesystems

            for host_mount_info in host_file_sys_vol_mount_info:
                # Extract only VMFS volumes

                if host_mount_info.volume.type == "VMFS":
                    extents = host_mount_info.volume.extent
                    try:
                        uuid = host_mount_info.volume.uuid
                        uuids = uuid.replace("-","")
                    except Exception as e:
                        uuids = ""
                    datastore_details = {
                        'uuid': uuid,
                        'capacity': host_mount_info.volume.capacity,
                        'vmfs_version': host_mount_info.volume.version,
                        'local': host_mount_info.volume.local,
                        'ssd': host_mount_info.volume.ssd,
                    }
                    extent_arr = []
                    extent_count = 0
                    for extent in extents:
                        extent_count += 1
                        extent_arr.append(extent.diskName)
                        datastore_details['extents'] = extent_arr
                        datastore_details["storagename"] = host_mount_info.volume.name
                        datastore_details["exsi_host"] = esxi_host.name

                datastores.append(datastore_details)

        return datastores
    def getstorage01(self):
        """
        "Capacity(GB)": "1.1TB",
          "URL": "ds:///vmfs/volumes/5b126d7e-fbee7582-8d66-5c546d5798c3/",
          "VmNum": "0",
          "hostNum": "1",
          "free cap(GB)": "1.1TB",
          "Name": "datastore1 (25)"
          完成
        :return:[{}]
        """
        ds_obj_list = self.content.viewManager.CreateContainerView(self.content.rootFolder,
                                                          [vim.Datastore],
                                                          True)
        datastore = []
        for ds in ds_obj_list.view:
            summary = ds.summary
            ds_capacity = summary.capacity
            ds_freespace = summary.freeSpace
            # ipdata = []
            # vmdata = []
            esxiip = []
            vmip = []
            for i in ds.host:
                # ipdata.append({str(i.key):i.key.name})
                esxiip.append({
                    "esxiip_tag":str(i.key),
                    "esxiip":i.key.name
                })
            for i in ds.vm:
                # vmdata.append({str(i):i.name})
                vmip.append(
                    {
                        "vmip_tag":str(i),
                        "vmipname":i.name,
                    }
                )
            datadict  = {
                "Name":format(summary.name),
                "URL":format(summary.url),
                "Capacity":format(sizeof_fmt(ds_capacity)),
                "free cap":format(sizeof_fmt(ds_freespace)),
                "hostNum":format(len(ds.host)),
                "VmNum":format(len(ds.vm)),
                "esxiip":esxiip,
                "vmip":vmip
            }
            if datadict:
                datastore.append(datadict)
        return datastore
    #v1.0
    def gethost(self):
        """
        :return:list
        """
        objview = self.content.viewManager.CreateContainerView(self.content.rootFolder,
                                                          [vim.HostSystem],
                                                          True)
        esxi_hosts = objview.view
        objview.Destroy()
        datastores = []
        for esxi_host in esxi_hosts:
            systemInfo = esxi_host.hardware.systemInfo
            cpuInfo = esxi_host.hardware.cpuInfo
            storage_system = esxi_host.configManager.storageSystem
            host_file_sys_vol_mount_info = storage_system.fileSystemVolumeInfo.mountInfo
            data = {
            "esxi_host":esxi_host.name
            ,"esxi_tag":str(esxi_host)
            ,"vendor":systemInfo.vendor
            ,"model":systemInfo.model
            ,"host_uuid":systemInfo.uuid
            ,"numCpuPackages":cpuInfo.numCpuPackages
            ,"numCpuCores":cpuInfo.numCpuCores
            ,"numCpuThreads":cpuInfo.numCpuThreads
            ,"hz":cpuInfo.hz
            ,"memorySize":esxi_host.hardware.memorySize
            }
            datastores.append(data)
        return datastores
    #v1.0
    def getcluster(self):
        """
        获取集群下的IP信息
        :return:
        """
        objview = self.content.viewManager.CreateContainerView(self.content.rootFolder,
                                                          [vim.ClusterComputeResource],
                                                          True)
        datavcenter = {}
        for i in objview.view:
            datahost = []
            for k in i.host:
                datahost.append({str(k):k.name})
            datavcenter[i.name] = datahost
        # print(json.dumps(datavcenter,indent=1))
        return datavcenter

利用supervisor进行进程管理

supervisor安装

1.apt-get install supervisor直接进行安装 。
2.echo_supervisord_conf > /etc/supervisord.conf，生成配置文件

3.配置：

    key在配置文件底部添加
        [include]
files=/etc/supervisor/*.conf #若你本地无/etc/supervisor目录，直接创建。

4.开机自启动

配置进程开机自启动“
    chmod +x /etc/init.d/supervisord
# in debian based:
sudo update-rc.d supervisord defaults
# in redhat
chkconfig --add supervisord
    设置开机启动：chkconfig supervisord on
    查看是否成功：chkconfig --list | grep supervisord

5. 配置程序,在/etc/supervisor内创建3个conf配置文件。并写入以下3个文件。

[program:sanmanager]
directory=/home/collect_manager
command= celery -A run_celery worker -l info
stdout_logfile=/home/collect_manager/run.log
user = root
autostart=true
autorestart=true
startsecs=60
stopasgroup=true
ikillasgroup=true
startretries=1
redirect_stderr=true

[program:sanmanagerbreat]
directory=/home/collect_manager
command= celery -A run_celery heart -l info
stdout_logfile=/home/collect_manager/breat.log
user = root
autostart=true
autorestart=true
startsecs=60
stopasgroup=true
ikillasgroup=true
startretries=1
redirect_stderr=true

[program:trapserver]
directory=/home/collect_manager
command= python3 trapserver.py
stdout_logfile=/home/collect_manager/trapserver.log
user = root
autostart=true
autorestart=true
startsecs=60
stopasgroup=true
ikillasgroup=true
startretries=1
redirect_stderr=true

6.用法

supervisorctl的用法
supervisord : 启动
supervisor supervisorctl reload :修改完配置文件后重新启动
supervisor supervisorctl status :查看supervisor监管的进程状态 
supervisorctl start 进程名 ：启动XXX进程 
supervisorctl stop 进程名 ：停止XXX进程
 supervisorctl stop all：停止全部进程，注：start、restart、stop都不会载入最新的配置文件。 
supervisorctl update：根据最新的配置文件，启动新配置或有改动的进程，配置没有改动的进程不会受影响而重启