3、日志记录与告警:使用logging模块记录异常,配置邮件/钉钉/企业微信告警

3.1 日志记录——部署工程师的“黑匣子”

模型部署上线后,最怕什么?

我最怕半夜被电话吵醒,说服务挂了。但更怕的是,服务挂了却不知道为啥挂的。

日志,就是我们的“黑匣子”。它记录了系统从启动到崩溃的全过程。我个人习惯,每个部署项目的第一件事,就是把日志系统搭好。没有日志,你就像在黑暗中修车,全靠摸。

Python 自带的 logging 模块,其实够用了。别一上来就上 ELK,先把手头的活干利索。

3.2 用 logging 模块记录异常

先看一个最基础的配置。我在项目中通常这样写:

import logging
import sys
from logging.handlers import RotatingFileHandler

# 创建 logger
logger = logging.getLogger("model_deploy")
logger.setLevel(logging.DEBUG)

# 控制台输出
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)

# 文件输出,按大小轮转
file_handler = RotatingFileHandler(
    "deploy.log", maxBytes=10*1024*1024, backupCount=5
)
file_handler.setLevel(logging.DEBUG)

# 定义格式
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
)
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

# 添加处理器
logger.addHandler(console_handler)
logger.addHandler(file_handler)

嗯,这里要注意几个点:

  • 日志级别要分层:DEBUG 写文件,INFO 以上才打印到控制台。不然生产环境会被日志刷爆。
  • 文件要轮转RotatingFileHandler 能按大小切分文件。我曾经见过一个项目,单日志文件 50GB,打开都费劲。
  • 格式要带行号%(filename)s:%(lineno)d 这俩字段,能帮你快速定位到出错的代码行。
我的小技巧:在模型推理的关键路径上,加一条 logger.debug("Input shape: %s", input_tensor.shape)。平时不输出,出问题时改成 DEBUG 级别,就能复现现场。

3.3 异常捕获与记录的最佳实践

光有日志配置还不够,你得知道在哪儿记、记什么。

我见过很多新手,在 except 块里只写一句 logger.error("error")。这等于没记。你想想看,出错了你连堆栈都不知道,怎么排查?

正确的做法是:

try:
    result = model.predict(input_data)
except Exception as e:
    logger.exception("模型推理失败,输入数据: %s", input_data)
    # 这里会自动记录完整的 traceback
    raise  # 或者返回错误响应

logger.exception() 会自动把异常堆栈打进去。这是我最常用的方法,没有之一。

避坑指南:千万不要在 except 里直接 pass。我曾经接手过一个项目,所有异常都被吞掉了,服务返回 200 但结果全是错的。排查了三天,才发现是模型加载时某个权重文件损坏了。

3.4 配置告警——让系统主动找你

日志写好了,但你不能天天盯着日志看吧?

告警的意义在于:让系统在出问题时主动找你,而不是你去找问题

我个人习惯,把告警分为三个等级:

级别 触发条件 通知方式 响应时间
P0(致命) 服务完全不可用、模型加载失败 电话 + 短信 + 群消息 5分钟内
P1(严重) 推理延迟超过阈值、错误率飙升 群消息 + 邮件 30分钟内
P2(警告) 内存使用率过高、磁盘空间不足 邮件 当天处理

3.5 邮件告警配置

邮件告警虽然慢,但胜在稳定。适合 P2 级别的通知。

import smtplib
from email.mime.text import MIMEText

def send_email_alert(subject, content):
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = 'deploy@yourcompany.com'
    msg['To'] = 'ops@yourcompany.com'

    with smtplib.SMTP('smtp.yourcompany.com', 587) as server:
        server.starttls()
        server.login('deploy@yourcompany.com', 'password')
        server.send_message(msg)
注意:密码别硬编码。用环境变量或者配置中心管理。我一般用 os.getenv("SMTP_PASSWORD") 来读取。

3.6 钉钉告警配置

钉钉机器人是目前国内用得最多的。配置简单,响应快。

import requests
import json

def send_dingtalk_alert(message):
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=YOUR_TOKEN"
    payload = {
        "msgtype": "text",
        "text": {
            "content": f"【模型部署告警】\n{message}"
        }
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(webhook_url, data=json.dumps(payload), headers=headers)
    return response.json()

钉钉有个好处——可以 @ 指定的人。我通常会在 P0 告警里 @ 整个值班群:

payload = {
    "msgtype": "text",
    "text": {
        "content": f"【P0告警】服务不可用!请立即处理!\n{message}"
    },
    "at": {
        "atMobiles": ["13800138000"],
        "isAtAll": True
    }
}

3.7 企业微信告警配置

企业微信的机器人跟钉钉类似,但格式略有不同。

def send_wechat_alert(message):
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    payload = {
        "msgtype": "markdown",
        "markdown": {
            "content": f"## 模型部署告警\n{message}\n> 请相关同事及时处理"
        }
    }
    headers = {"Content-Type": "application/json"}
    response = requests.post(webhook_url, data=json.dumps(payload), headers=headers)
    return response.json()
核心要点:企业微信支持 markdown 格式,可以加粗、加标题、引用。比纯文本好看很多。我一般用 markdown 格式发告警,重点信息一目了然。

3.8 统一告警封装

实际项目中,我不会每个地方都单独调告警函数。我会封装一个统一的告警类:

class AlertManager:
    def __init__(self):
        self.channels = []

    def add_channel(self, channel_func):
        self.channels.append(channel_func)

    def send(self, level, title, message):
        full_msg = f"[{level}] {title}\n{message}"
        for channel in self.channels:
            try:
                channel(full_msg)
            except Exception as e:
                logger.error(f"告警通道发送失败: {e}")

# 使用示例
alert = AlertManager()
alert.add_channel(send_dingtalk_alert)
alert.add_channel(send_email_alert)

# 在异常捕获中调用
try:
    model.predict(data)
except Exception as e:
    logger.exception("推理异常")
    alert.send("P1", "模型推理失败", str(e))

这样做的好处是:想加新的告警通道,只需要写一个函数,然后 add_channel 就行。不用改业务代码。

3.9 总结

日志和告警,说白了就是两件事:

  • 日志:把现场记录下来,方便事后复盘。
  • 告警:在出事的第一时间通知到你。

我见过太多团队,模型上线后日志不记、告警不配。出问题了全靠人工排查,效率极低。说白了,这是对自己代码的不自信。

把日志和告警当成基础设施来建设。前期多花一小时,后期能省一整天。

最后一句:告警不是越多越好。我曾经把每个 INFO 日志都配了告警,结果一天收到几百条消息,最后全被屏蔽了。记住,告警要精准,不要狼来了