python-cmdb資產管理項目2-使用工廠模式模塊化代碼

時間 2021-02-14

標籤 html python json api 服務器多線程併發 app ssh ide 欄目 Python 简体版

原文原文鏈接

python-cmdb資產管理項目2-使用工廠模式模塊化代碼

概要:python-cmdb資產管理項目豐富中控指令,收集更多信息html

1 工廠模式雛形

建立一個python項目,命令爲autoclient,並建立一個名爲lib的目錄做爲模塊目錄python

整個目錄結構以下:json

1.1 模塊代碼

開發三個代碼塊,初步做爲收集的信息api

/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py服務器

"""
用於採集硬盤信息
"""
class DiskPlugin(object):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'Disk':'100G'}

/home/ningherui/PycharmProjects/autoclient/lib/plugins/memory.py多線程

"""
用於採集內存信息
"""
class MemoryPlugin(object):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'Memory':'1G'}

/home/ningherui/PycharmProjects/autoclient/lib/plugins/network.py併發

"""
用於採集網卡信息
"""
class NetworkPlugin(object):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'IP':'192.168.100.101'}

1.2 使用配置文件setting.py

構建一個新的setting.py文件,做爲配置文件app

/home/ningherui/PycharmProjects/autoclient/setting.pyssh

#做爲一個配置文件
# PLUGIN_CLASS_LIST = [
#     'lib.plugins.disk.DiskPlugin',
#     'lib.plugins.memory.MemoryPlugin',
#     'lib.plugins.network.NetworkPlugin',
# ]

PLUGIN_CLASS_DICT = {
    "disk": 'lib.plugins.disk.DiskPlugin',
    "memory": 'lib.plugins.memory.MemoryPlugin',
    "network": 'lib.plugins.network.NetworkPlugin',
}

View Code

主代碼ide

/home/ningherui/PycharmProjects/autoclient/app.py

# 建立類庫/模塊目錄lib
# from lib.plugins.disk    import DiskPlugin
# from lib.plugins.memory  import MemoryPlugin
# from lib.plugins.network import NetworkPlugin
# 
# def run():
#     #實例化對象
#     obj1 = DiskPlugin()
#     disk_info = obj1.process()
# 
#     obj2 = MemoryPlugin()
#     memory_info = obj2.process()
#     
#     obj3 = NetworkPlugin()
#     Network_info = obj3.process()
#     
# if __name__ == '__main__':
#     run()
import importlib
from  setting import PLUGIN_CLASS_DICT

def run():
   for key,path in PLUGIN_CLASS_DICT.items():
        #key = "disk", path = "lib.plugins.disk.DiskPlugin"
        """
            使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
            lib.plugins.disk  DiskPlugin
        """
        module_path,class_name = path.rsplit('.',maxsplit=1)
#根據一個字符串導入一個模塊,須要使用importlib的模塊
        module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
        #獲取一個類的對象
        cls = getattr(module,class_name)
        #print(key,cls())
        plugin_object = cls()
        info = plugin_object.process()
        print(key,info)


if __name__ == '__main__':
    run()

View Code

執行

disk {'Disk': '100G'}
memory {'Memory': '1G'}
network {'IP': '192.168.100.101'}

基於反射和工廠模式實現可擴展

這樣的方式有利於模塊的擴展,當添加一個新的模塊,只須要在setting的[配置文件加入相應的模塊配置,就可使用,刪除也是如此

1.3 使用約束類,約束模塊使用的方法

在上述的模塊中都使用的process方法,可是當在添加的新的模塊中,若是定義新的方法,不叫process,這時添加到setting的配置中,並不能生效,則能夠定一個約束類

代碼以下:

class BasePlugin(object):
    """
    基類,用於作約束.約束子類中必須實現process方法
    """
    def process(self):
        raise NotImplementedError("%s中必須實現Process方法" %self.__class__.__name__)
class DiskPlugin(BasePlugin):
    pass
class MemoryPlugin(BasePlugin):
    pass
obj = DiskPlugin()
obj.process()

執行後結果以下:

raise NotImplementedError("%s中必須實現Process方法" %self.__class__.__name__)
NotImplementedError: DiskPlugin中必須實現Process方法

整理後代碼以下

/home/ningherui/PycharmProjects/autoclient/lib/plugins/base.py

class BasePlugin(object):
    """
    基類,用於作約束.約束子類中必須實現process方法
    """
    def process(self):
        raise NotImplementedError("%s中必須實現Process方法" %self.__class__.__name__)

/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py

"""
用於採集硬盤信息
"""
from .base import BasePlugin
class DiskPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'Disk':'100G'}

/home/ningherui/PycharmProjects/autoclient/lib/plugins/memory.py

"""
用於採集內存信息
"""
from .base import BasePlugin
class MemoryPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'Memory':'1G'}

/home/ningherui/PycharmProjects/autoclient/lib/plugins/network.py

"""
用於採集網卡信息
"""
from .base import BasePlugin
class NetworkPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self):
        # 假設執行命令
        return {'IP':'192.168.100.101'}

當模塊的方法不是process就會報錯,並會提示是那一個模塊的方法錯誤

文件結構

把獲取的信息進行彙總,單獨放出來

創建一個__init__.py文件

/home/ningherui/PycharmProjects/autoclient/lib/plugins/__init__.py

import importlib
import setting

def get_server_info():
    server_info = {}
    for key,path in setting.PLUGIN_CLASS_DICT.items():
        #key = "disk", path = "lib.plugins.disk.DiskPlugin"
        """
            使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
            lib.plugins.disk  DiskPlugin
        """
        module_path,class_name = path.rsplit('.',maxsplit=1)
        #根據一個字符串導入一個模塊,須要使用importlib的模塊
        module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
        #獲取一個類的對象
        cls = getattr(module,class_name)
        #print(key,cls())
        plugin_object = cls()
        info = plugin_object.process()
        server_info[key] = info
    return server_info

View Code

主文件

/home/ningherui/PycharmProjects/autoclient/app.py

# 建立類庫/模塊目錄lib
# from lib.plugins.disk    import DiskPlugin
# from lib.plugins.memory  import MemoryPlugin
# from lib.plugins.network import NetworkPlugin
# 
# def run():
#     #實例化對象
#     obj1 = DiskPlugin()
#     disk_info = obj1.process()
# 
#     obj2 = MemoryPlugin()
#     memory_info = obj2.process()
#     
#     obj3 = NetworkPlugin()
#     Network_info = obj3.process()
#     
# if __name__ == '__main__':
#     run()
import importlib
import setting
from lib.plugins import get_server_info
def ssh(hostname,cmd):
    import paramiko
    private_key = paramiko.RSAKey.from_private_key_file('/home/ningherui/.ssh/id_rsa')
    # 建立SSH對象
    ssh = paramiko.SSHClient()
    # 容許鏈接不在know_hosts文件中的主機
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 鏈接服務器
    ssh.connect(hostname=hostname, port=setting.SSH_PORT, username=setting.SSH_USER, pkey=setting.SSH_KEY)
    # 執行命令
    stdin, stdout, stderr = ssh.exec_command(cmd)
    # 獲取命令結果
    result = stdout.read()
    # 關閉鏈接
    ssh.close()
    return result
# def run():
#    for key,path in setting.PLUGIN_CLASS_DICT.items():
#         #key = "disk", path = "lib.plugins.disk.DiskPlugin"
#         """
#             使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
#             lib.plugins.disk  DiskPlugin
#         """
#         module_path,class_name = path.rsplit('.',maxsplit=1)
# #根據一個字符串導入一個模塊,須要使用importlib的模塊
#         module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
#         #獲取一個類的對象
#         cls = getattr(module,class_name)
#         #print(key,cls())
#         plugin_object = cls()
#         info = plugin_object.process()
#         print(key,info)
def run():
    #獲取__init__.py的get_server_info方法的到的數據
    server_info = get_server_info()
    print(server_info)

if __name__ == '__main__':
    run()

View Code

執行

/usr/bin/python3 /home/ningherui/PycharmProjects/autoclient/app.py
{'disk': {'Disk': '100G'}, 'memory': {'Memory': '1G'}, 'network': {'IP': '192.168.100.101'}}

1.4 使用shh鏈接遠程執行操做得到數據

代碼以下:

主代碼

/home/ningherui/PycharmProjects/autoclient/app.py

# 建立類庫/模塊目錄lib
# from lib.plugins.disk    import DiskPlugin
# from lib.plugins.memory  import MemoryPlugin
# from lib.plugins.network import NetworkPlugin
# 
# def run():
#     #實例化對象
#     obj1 = DiskPlugin()
#     disk_info = obj1.process()
# 
#     obj2 = MemoryPlugin()
#     memory_info = obj2.process()
#     
#     obj3 = NetworkPlugin()
#     Network_info = obj3.process()
#     
# if __name__ == '__main__':
#     run()
import importlib
import setting
from lib.plugins import get_server_info
def ssh(hostname,cmd):
    import paramiko
    private_key = paramiko.RSAKey.from_private_key_file('/home/ningherui/.ssh/id_rsa')
    # 建立SSH對象
    ssh = paramiko.SSHClient()
    # 容許鏈接不在know_hosts文件中的主機
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 鏈接服務器
    ssh.connect(hostname=hostname, port=setting.SSH_PORT, username=setting.SSH_USER, pkey=private_key)
    # 執行命令
    stdin, stdout, stderr = ssh.exec_command(cmd)
    # 獲取命令結果
    result = stdout.read()
    # 關閉鏈接
    ssh.close()
    return result
# def run():
#    for key,path in setting.PLUGIN_CLASS_DICT.items():
#         #key = "disk", path = "lib.plugins.disk.DiskPlugin"
#         """
#             使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
#             lib.plugins.disk  DiskPlugin
#         """
#         module_path,class_name = path.rsplit('.',maxsplit=1)
# #根據一個字符串導入一個模塊,須要使用importlib的模塊
#         module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
#         #獲取一個類的對象
#         cls = getattr(module,class_name)
#         #print(key,cls())
#         plugin_object = cls()
#         info = plugin_object.process()
#         print(key,info)
def run():
    #獲取__init__.py的get_server_info方法的到的數據,獲取兩個參數
    hostname = '192.168.100.101'
    server_info = get_server_info(ssh,hostname)
    print(server_info)

if __name__ == '__main__':
    run()

View Code

__init__代碼:

import importlib
import setting

def get_server_info(ssh,hostname):
    server_info = {}
    for key,path in setting.PLUGIN_CLASS_DICT.items():
        #key = "disk", path = "lib.plugins.disk.DiskPlugin"
        """
            使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
            lib.plugins.disk  DiskPlugin
        """
        module_path,class_name = path.rsplit('.',maxsplit=1)
        #根據一個字符串導入一個模塊,須要使用importlib的模塊
        module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
        #獲取一個類的對象
        cls = getattr(module,class_name)
        #print(key,cls())
        plugin_object = cls()
        info = plugin_object.process(ssh,hostname)
        server_info[key] = info
    return server_info

View Code

模塊代碼:

-#base.py
class BasePlugin(object):
    """
    基類,用於作約束.約束子類中必須實現process方法
    """
    def process(self,ssh,hostname):
        raise NotImplementedError("%s中必須實現Process方法" %self.__class__.__name__)
#disk.py
"""
用於採集硬盤信息
"""
from .base import BasePlugin
class DiskPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self,ssh,hostname):
        # 假設執行命令
        # return {'Disk':'100G'}
        result = ssh(hostname,' df  -H|grep root|awk \'{print $2}\'')
        return result.decode('utf-8')
#network.py
"""
用於採集網卡信息
"""
from .base import BasePlugin
class NetworkPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self,ssh,hostname):
        result = ssh(hostname, 'ifconfig ens33|grep -v inet6|grep inet|awk \'{print $2}\'')
        return result.decode('utf-8')
#memory.py
"""
用於採集內存信息
"""
from .base import BasePlugin
class MemoryPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self,ssh,hostname):
        result = ssh(hostname, "free -h |grep Mem|awk '{print $2}'")
        return result.decode('utf-8')

View Code

在setting.py中配置,能夠在主代碼中調用配置

SSH_USER='root'
SSH_PORT='22'

執行結果

/usr/bin/python3 /home/ningherui/PycharmProjects/autoclient/app.py
{'disk': '40G\n', 'memory': '1.9G\n', 'network': '192.168.100.101\n'}

1.5 收集多個機器信息

收集多個機器,並傳入api,顯示,api的代碼參考上一章autoserver(https://www.cnblogs.com/zyxnhr/p/14391361.html)

# 建立類庫/模塊目錄lib
# from lib.plugins.disk    import DiskPlugin
# from lib.plugins.memory  import MemoryPlugin
# from lib.plugins.network import NetworkPlugin
# 
# def run():
#     #實例化對象
#     obj1 = DiskPlugin()
#     disk_info = obj1.process()
# 
#     obj2 = MemoryPlugin()
#     memory_info = obj2.process()
#     
#     obj3 = NetworkPlugin()
#     Network_info = obj3.process()
#     
# if __name__ == '__main__':
#     run()
import importlib
import requests
import setting
from lib.plugins import get_server_info
def ssh(hostname,cmd):
    import paramiko
    private_key = paramiko.RSAKey.from_private_key_file('/home/ningherui/.ssh/id_rsa')
    # 建立SSH對象
    ssh = paramiko.SSHClient()
    # 容許鏈接不在know_hosts文件中的主機
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 鏈接服務器
    ssh.connect(hostname=hostname, port=setting.SSH_PORT, username=setting.SSH_USER, pkey=private_key)
    # 執行命令
    stdin, stdout, stderr = ssh.exec_command(cmd)
    # 獲取命令結果
    result = stdout.read()
    # 關閉鏈接
    ssh.close()
    return result
# def run():
#    for key,path in setting.PLUGIN_CLASS_DICT.items():
#         #key = "disk", path = "lib.plugins.disk.DiskPlugin"
#         """
#             使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
#             lib.plugins.disk  DiskPlugin
#         """
#         module_path,class_name = path.rsplit('.',maxsplit=1)
# #根據一個字符串導入一個模塊,須要使用importlib的模塊
#         module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
#         #獲取一個類的對象
#         cls = getattr(module,class_name)
#         #print(key,cls())
#         plugin_object = cls()
#         info = plugin_object.process()
#         print(key,info)
def run():
    #獲取__init__.py的get_server_info方法的到的數據,獲取兩個參數
    # hostname = '192.168.100.101'
    host_list = [
        '192.168.100.101',
        '192.168.100.102',
        '192.168.100.103',
    ]
    for hostname in host_list:
        server_info = get_server_info(ssh,hostname)
        result = requests.post(
            url='http://127.0.0.1:8000/api/get_data',
            json={'host': hostname, 'info': server_info}
        )
        # print(server_info)
    # server_info = get_server_info(ssh,hostname)
    # print(server_info)
    print('把資產信息發送到API', result.text)
if __name__ == '__main__':
    run()

View Code

執行:

server端顯示

2 使用線程池多線程

測試腳本

import time
from concurrent.futures import ThreadPoolExecutor

def task(i):
    time.sleep(1)
    print(i)

pool = ThreadPoolExecutor(10)

for i in range(87):
    pool.submit(task,i)

View Code

app.py腳本以下:

# 建立類庫/模塊目錄lib
# from lib.plugins.disk    import DiskPlugin
# from lib.plugins.memory  import MemoryPlugin
# from lib.plugins.network import NetworkPlugin
# 
# def run():
#     #實例化對象
#     obj1 = DiskPlugin()
#     disk_info = obj1.process()
# 
#     obj2 = MemoryPlugin()
#     memory_info = obj2.process()
#     
#     obj3 = NetworkPlugin()
#     Network_info = obj3.process()
#     
# if __name__ == '__main__':
#     run()
import importlib
import requests
import setting
from lib.plugins import get_server_info
def ssh(hostname,cmd):
    import paramiko
    private_key = paramiko.RSAKey.from_private_key_file('/home/ningherui/.ssh/id_rsa')
    # 建立SSH對象
    ssh = paramiko.SSHClient()
    # 容許鏈接不在know_hosts文件中的主機
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    # 鏈接服務器
    ssh.connect(hostname=hostname, port=setting.SSH_PORT, username=setting.SSH_USER, pkey=private_key)
    # 執行命令
    stdin, stdout, stderr = ssh.exec_command(cmd)
    # 獲取命令結果
    result = stdout.read()
    # 關閉鏈接
    ssh.close()
    return result
# def run():
#    for key,path in setting.PLUGIN_CLASS_DICT.items():
#         #key = "disk", path = "lib.plugins.disk.DiskPlugin"
#         """
#             使用下面的語句從右向左根據地一個點(.)分紅兩個字符串,
#             lib.plugins.disk  DiskPlugin
#         """
#         module_path,class_name = path.rsplit('.',maxsplit=1)
# #根據一個字符串導入一個模塊,須要使用importlib的模塊
#         module = importlib.import_module(module_path)   #等價於import lib.plugins.disk
#         #獲取一個類的對象
#         cls = getattr(module,class_name)
#         #print(key,cls())
#         plugin_object = cls()
#         info = plugin_object.process()
#         print(key,info)
def run():
    #獲取__init__.py的get_server_info方法的到的數據,獲取兩個參數
    # hostname = '192.168.100.101'
    host_list = [
        '192.168.100.101',
        '192.168.100.102',
        '192.168.100.103',
    ]
    for hostname in host_list:
        server_info = get_server_info(ssh,hostname)
        result = requests.post(
            url='http://127.0.0.1:8000/api/get_data',
            json={'host': hostname, 'info': server_info}
        )
        # print(server_info)
    # server_info = get_server_info(ssh,hostname)
    # print(server_info)
    print('把資產信息發送到API', result.text)
if __name__ == '__main__':
    run()

View Code

測試結果

server端 api顯示:

3 單例模式使用

3.1單例模式示例

class Singletoton(object):
    instance = None #靜態字段,類變量
    def __init__(self,name):
        #初始化對象
        self.name = name
    def __new__(cls, *args, **kwargs):
        #建立對象
        if not cls.instance:
            cls.instance = object.__new__(cls)
        return cls.instance
obj1 = Singletoton('alex')
obj2 = Singletoton('erick')

print(obj1)
print(obj2)

View Code

執行結果

多線程致使建立多個對象

import time
class Singletoton(object):
    instance = None #靜態字段,類變量
    def __init__(self,name):
        #初始化對象
        self.name = name
    def __new__(cls, *args, **kwargs):
        #建立對象
        if not cls.instance:
            time.sleep(0.2)
            cls.instance = object.__new__(cls)
        return cls.instance


import threading
def func():
    obj = Singletoton('xxx')
    print(obj)
for i in range(10):
    thread = threading.Thread(target=func)
    thread.start()

View Code

執行結果

緣由:

是由於當進程執行到建立對象時,10個線程都執行這一步,可是前面的線程沒有執行完,就致使建立新的對象

3.2 加鎖解決多線程問題

import time
import threading
class Singletoton(object):
    instance = None #靜態字段,類變量
    lock = threading.RLock()
    def __init__(self,name):
        #初始化對象
        self.name = name
    def __new__(cls, *args, **kwargs):

        #建立對象
        with cls.lock:
            if not cls.instance:
                time.sleep(0.2)
                cls.instance = object.__new__(cls)
            return cls.instance

def func():
    obj = Singletoton('xxx')
    print(obj)
for i in range(10):
    thread = threading.Thread(target=func)
    thread.start()

View Code

執行結果

這種條件下,當在併發多線程的過程當中,當有線程外的進程建立對象時,就會有鎖鎖住,沒法建立新的對象,效率比較低

3.3 代碼優化

import time
import threading
class Singletoton(object):
    instance = None #靜態字段,類變量
    lock = threading.RLock()
    def __init__(self,name):
        #初始化對象
        self.name = name
    def __new__(cls, *args, **kwargs):

        #建立對象
        if cls.instance:
            return cls.instance
        with cls.lock:
            if not cls.instance:
                time.sleep(0.2)
                cls.instance = object.__new__(cls)
            return cls.instance

def func():
    obj = Singletoton('xxx')
    print(obj)
for i in range(10):
    thread = threading.Thread(target=func)
    thread.start()

View Code

經過模塊實現單例模式

# 文件1
# danli.py
class Foo(object):
    pass

site = Foo()

# 文件2
from danli.py import site

print(site)  #單例模式

4 日誌處理(異常堆棧信息)

4.1 traceback模塊

捕獲異常的堆棧信息

import traceback
try:
    int('dauchdasukd')
except Exception as e:
    # print(e)
    print(traceback.format_exc())

執行看錯誤信息:

4.2 單個日誌處理

import logging
logging.basicConfig(filename='log.txt',
                    format='%(asctime)s - %(name)s - %(levelname)s - %(module)s: %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S %p',
                    level=10)
logging.error('xxxxxxxxl')

執行後,會生成log.txt文件,內容以下:

2021-02-14 11:16:23 AM - root - ERROR - test: xxxxxxxxl

對於上述記錄日誌的功能,只能將日誌記錄自愛單個文件中,若是要設置多個日誌文件,logging.basicConfig將沒法完成,須要自定義文件和日誌操做對象

4.3 多日誌處理

import logging

file_haddler = logging.FileHandler('run.log','a',encoding='utf-8')
fmt = logging.Formatter(fmt='%(asctime)s - %(name)s - %(levelname)s - %(module)s: %(message)s')
file_haddler.setFormatter(fmt)

file_haddler_1 = logging.FileHandler('run_1.log','a',encoding='utf-8')
fmt = logging.Formatter()
file_haddler_1.setFormatter(fmt)

logger1 = logging.Logger('ssssxxxxx',level=logging.DEBUG)
logger1.addHandler(file_haddler)
logger1.addHandler(file_haddler_1)

logger1.error('asdzxcqwe')

執行會生成兩個日誌文件

4.4 引用日誌文件

建立一個utils的公共庫目錄,在該目錄下建立一個log模塊的庫,同時建立一個log目錄,同時log的名字爲cmdb.log

/home/ningherui/PycharmProjects/autoclient/lib/utils/log.py

import logging
import setting
class Logger():
    def __init__(self,log_file_path,level):
        file_haddler = logging.FileHandler(log_file_path, 'a', encoding='utf-8')
        fmt = logging.Formatter(fmt='%(asctime)s - %(name)s - %(levelname)s - %(module)s: %(message)s')
        file_haddler.setFormatter(fmt)

        self.logger = logging.Logger('cmdb',level=level)
        self.logger.addHandler(file_haddler)

    def error(self,msg):
        self.logger.error(msg)

#調用這個logger模塊,寫入日誌
logger = Logger(setting.LOGGING_PATH,logging.DEBUG)

disk磁盤日誌

/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py

"""
用於採集硬盤信息
"""
from .base import BasePlugin
from lib.utils.log import logger
import traceback
class DiskPlugin(BasePlugin):
    """
    採集硬盤信息
    """
    def process(self,ssh,hostname):
        result = None
        try:
            result = int('cdacas')
        except Exception as e:
            logger.error(traceback.format_exc())
        # 假設執行命令
        # return {'Disk':'100G'}
        #result = ssh(hostname,' df  -H|grep root|awk \'{print $2}\'')

        # return result.decode('utf-8')
        return result

View Code

執行app後,生成的日誌文件

/home/ningherui/PycharmProjects/autoclient/log/cmdb.log

2021-02-14 14:25:23,577 - cmdb - ERROR - log: Traceback (most recent call last):
  File "/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py", line 14, in process
    result = int('cdacas')
ValueError: invalid literal for int() with base 10: 'cdacas'

2021-02-14 14:25:23,578 - cmdb - ERROR - log: Traceback (most recent call last):
  File "/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py", line 14, in process
    result = int('cdacas')
ValueError: invalid literal for int() with base 10: 'cdacas'

2021-02-14 14:25:23,578 - cmdb - ERROR - log: Traceback (most recent call last):
  File "/home/ningherui/PycharmProjects/autoclient/lib/plugins/disk.py", line 14, in process
    result = int('cdacas')
ValueError: invalid literal for int() with base 10: 'cdacas'