脚本程序监控:
通过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方法即可
