Python监控脚本异常和进程异常

脚本程序监控:
通过python装饰器监控脚本是否异常,如果异常发送报警邮件通知开发人员

脚本进程监控:
因为使用的是supervisor进程守护的脚本,所以通过init.py来初始化获取supervisor配置文件获取对应的执行文件名,使用subprocess库执行对应的shell命令,获取进程状态,状态异常,发送报警邮件

init.py文件如下

import os
# 获取当前文件目录
BaseDir = os.path.dirname(os.path.abspath(__file__))
# 获取对应文件所在目录
PathDir = os.path.join(BaseDir, '../supervisor_conf')
# 获取文件列表
FileList = os.listdir(path=PathDir)
CmdList = []
for file in FileList:
    file_path = PathDir + "/" + file
    # 逐行读取文件内容
    for content in open(file_path).readlines():
        if content.startswith("command=python"):
            cmd = content.split(" ")[-1].split("/")[-1].split(".")[0]
            CmdList.append(cmd)

monitor.py文件如下

import time
import smtplib
import asyncio
import datetime
import traceback
import subprocess
from functools import wraps
from email.header import Header
from email.mime.text import MIMEText
from .import CmdList
# 进程状态对应说明字典
process_status_dict = {"R": u"运行状态", "S": u"可中断睡眠状态", "D": u"不可中断睡眠状态", "T": u"暂停状态", "Z": u"退出状态"}


async def send_mail(mail_list: list, content: str, subject: str):
    """
    功能:异步发送邮件
    -----------------------------
    :param mail_list: 接收人邮箱列表
    :param content: 邮件内容
    :param subject: 邮件主题
    """
    # 邮箱服务器类型
    server_type = "smtp.exmail.qq.com"
    # 发件人邮箱和密码
    sender = "********"
    password = "******"
    receivers = mail_list
    subject = f"脚本程序报警邮件:{subject}"
    # 参数分别是邮件内容、文本格式、编码
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['from'] = Header(sender)
    msg['to'] = Header("script_monitor")
    msg['subject'] = Header(subject)
    email_client = smtplib.SMTP_SSL(server_type, 465)  # SSL加密
    email_client.login(sender, password)  # 登录服务器
    email_client.sendmail(sender, receivers, msg.as_string())
    email_client.quit()  # 退出服务器


def app_monitor(user):
    """
    功能: 程序监控装饰器
    ------------------------
    user: 开发人员账号ID
    """

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                func(*args, **kwargs)
            except Exception as e:
                error_info = f"异常时间:\n{datetime.datetime.now()}\n异常详情:\n{traceback.format_exc()}"
                # 根据开发人员账号ID获取邮箱以及上级负责人邮箱
                mail_list = ["********", "********"]
                # 异步调用发邮件-声明一个特定的新loop
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                loop.run_until_complete(send_mail(mail_list=mail_list, content=error_info, subject=str(e)))
                loop.close()  # 关闭loop

        return wrapper

    return decorator


def process_monitor():
    """
    功能: 监控程序进程运行情况
    ------------------------
    说明:进程异常发送报警邮件
    """
    while True:
        for cmd_name in CmdList:
            # shell命令,通过执行文件名字获取对应进程的状态
            cmd = "ps -efo stat,args |grep -v grep|grep %s |awk '{print $1}'" % cmd_name
            # 执行shell命令并获取返回结果数据
            process_ret = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE).stdout.read()
            process_status = process_ret.decode("utf-8").split("\n")[0]
            if process_status != "S":
                # 非运行状态发送报警邮件
                status_name = process_status_dict.get(process_status)
                error_info = f"进程名称:\n{cmd_name}\n异常时间:\n{datetime.datetime.now()}\n异常状态:\n{status_name}"
                # 根据开发人员账号ID获取邮箱以及上级负责人邮箱
                mail_list = ["******", "*********"]
                # 异步调用发邮件-声明一个特定的新loop
                loop = asyncio.new_event_loop()
                asyncio.set_event_loop(loop)
                loop.run_until_complete(send_mail(mail_list=mail_list, content=error_info, subject=""))
                loop.close()  # 关闭loop
        # 所有进程检测一遍休眠5s
        time.sleep(5)

使用说明:
1. 脚本程序监控,直接使用装饰器app_monitor装饰对应的程序方法即可
2. 脚本进程监控,直接运行process_montior方法即可