自动化运维脚本：使用Shell/Python监控模型服务健康状态

自动化运维脚本使用Shell/Python监控模型服务健康状态最近在星图GPU平台上部署了一个图像着色服务cv_unet_image-colorization跑得挺稳定。但作为运维心里总有点不踏实——万一半夜服务挂了怎么办GPU显存悄悄爆了谁告诉我总不能24小时盯着屏幕吧。所以我花点时间折腾了一套自动化监控脚本。核心思路很简单让脚本定时去“看看”服务还活着没资源占用高不高日志里有没有报错。一旦发现问题就自动发消息通知我。这样一来我就能睡个安稳觉把精力放在更重要的优化工作上。这篇文章我就手把手带你写这套脚本。你不用有太深的运维背景只要会点基本的Shell和Python跟着步骤走就能给自己的模型服务也加上一个“健康小护士”。1. 我们要监控什么在动手写代码之前得先想清楚到底要监控服务的哪些方面。对于部署在GPU服务器上的模型服务我主要关心三件事第一服务是不是真的在“服务”。也就是健康检查。最简单的方法就是定期去调用一下服务的某个API接口比如一个简单的/health或/predict看看它能不能正常返回结果。如果超时了或者返回了错误码那基本就是服务出问题了。第二资源有没有被“吃光”。模型推理尤其是图像处理非常消耗GPU资源。我们需要监控GPU的显存使用率、利用率以及服务器的CPU、内存情况。如果显存使用率长期在90%以上或者突然飙升那可能就是内存泄漏或者有异常请求需要及时处理。第三服务自己有没有“喊疼”。也就是日志监控。服务在运行过程中会把各种信息包括错误Error、警告Warning写到日志文件里。我们可以让脚本定期去扫描最新的日志看看有没有出现特定的错误关键字比如“OutOfMemoryError”、“CUDA error”、“Failed to predict”等。把这三点监控起来我们就能对服务的状态有一个比较全面的把握了。2. 环境与工具准备我们的监控脚本会用到一些现成的工具和Python库先来准备好。系统环境一台Linux服务器我们的模型服务就部署在上面。Bash Shell环境。Python 3.6或以上版本。需要安装的Python包我们主要用requests来调用API用psutil来获取系统资源信息用GPUtil来获取GPU信息如果你用的是NVIDIA GPU。通过pip安装即可pip install requests psutil # 如果是NVIDIA GPU安装GPUtil pip install gputil如果你的Python环境里没有pip可以先安装它sudo apt-get install python3-pip适用于Ubuntu/Debian。关于通知渠道脚本发现问题后需要通知我们。这里介绍两种最常用的方式邮件通知简单通用适合个人或小团队。需要你知道SMTP服务器的地址、端口和邮箱密码或授权码。钉钉群机器人通知团队协作时非常方便信息能即时推送到手机。需要在钉钉群里添加一个自定义Webhook机器人获取它的Webhook地址。你可以根据实际情况选择一种或两种都配置。下面我们会分别给出示例代码。好了工具备齐接下来我们进入核心部分——写脚本。3. 编写健康检查脚本Shell篇我们先用一个简单的Shell脚本来实现最基础的健康检查。Shell脚本轻量、快速适合执行简单的命令和逻辑判断。假设我们的图像着色服务提供了一个用于着色的API地址是http://localhost:5000/predict它接受一个包含图片路径的JSON。我们可以用curl命令来测试它#!/bin/bash # 监控脚本基础健康检查 # 作者你的名字 # 日期$(date) # 1. 定义服务地址和检查超时时间 API_URLhttp://localhost:5000/predict TIMEOUT10 # 超时时间单位秒 # 2. 准备一个简单的测试请求这里用一张示例图片 TEST_IMAGE_PATH/path/to/your/test_image.jpg # 注意这里需要确保test_image.jpg文件存在或者你有一个不需要图片的health端点。 # 更通用的做法是服务应该提供一个专门的 /health 或 /status 端点只返回状态码。 # 我们假设 /predict 需要图片这里仅作连接性测试的示例。 # 使用curl进行健康检查 # -s: 静默模式不输出进度信息 # -o /dev/null: 将正常输出重定向到空设备不显示 # -w %{http_code}: 只输出最后的HTTP状态码 # --max-time: 请求超时时间 response_code$(curl -s -o /dev/null -w %{http_code} --max-time $TIMEOUT $API_URL) # 3. 判断状态码 if [ $response_code -eq 200 ]; then echo $(date): 服务健康检查通过状态码: 200 exit 0 # 退出状态码0表示成功 elif [ $response_code -eq 000 ]; then # curl返回000通常表示网络连接失败如服务未启动 echo $(date): [严重] 服务无法连接可能已宕机 # 这里可以触发告警我们先记录到文件 echo $(date): 服务宕机 /tmp/service_monitor_error.log exit 1 # 退出状态码非0表示失败 else echo $(date): [警告] 服务返回异常状态码: $response_code echo $(date): 状态码异常: $response_code /tmp/service_monitor_warn.log exit 2 fi这个脚本非常基础它只检查服务是否能连通。你可以把它保存为check_service_health.sh并赋予执行权限chmod x check_service_health.sh。然后手动运行一下试试./check_service_health.sh。如何让它定时运行使用Linux系统的crontab。执行crontab -e编辑定时任务添加一行# 每分钟执行一次健康检查脚本并将输出追加到日志文件 * * * * * /home/yourname/check_service_health.sh /var/log/service_health.log 21这样每分钟脚本都会运行一次并将输出记录到日志里。但只有日志还不够我们需要更强大的功能和更灵活的通知机制这就轮到Python上场了。4. 编写综合监控脚本Python篇Python脚本功能更强大我们可以把健康检查、资源监控、日志扫描和告警通知都集成在一起。4.1 监控脚本主体框架我们先搭建一个主函数定义好要监控的项#!/usr/bin/env python3 # -*- coding: utf-8 -*- import requests import time import subprocess import smtplib from email.mime.text import MIMEText from email.header import Header import json import psutil import os import sys # 尝试导入GPUtil如果非NVIDIA环境可注释掉 try: import GPUtil HAS_GPU True except ImportError: HAS_GPU False print(警告: 未找到GPUtil库将跳过GPU监控。如需监控GPU请安装: pip install gputil) class ModelServiceMonitor: def __init__(self): # 服务配置 self.api_url http://localhost:5000/predict self.health_check_timeout 5 # 日志文件路径假设你的服务日志在这里 self.log_file_path /var/log/cv_unet_service.log # 要监控的错误关键字列表 self.error_keywords [ERROR, Error, failed, exception, out of memory, CUDA error] # 资源监控阈值百分比 self.cpu_threshold 80.0 self.memory_threshold 85.0 self.gpu_memory_threshold 90.0 # 告警配置稍后填充 self.mail_config None self.dingtalk_webhook None def check_service_health(self): 检查API服务是否可用 try: # 这里可以发送一个更简单的请求比如给/health端点 # 如果没有health端点可以尝试一个轻量级的predict请求或直接GET response requests.get(self.api_url, timeoutself.health_check_timeout) if response.status_code 200: return True, f服务状态正常 (HTTP {response.status_code}) else: return False, f服务返回异常状态码: {response.status_code} except requests.exceptions.ConnectionError: return False, 服务连接失败可能未启动或网络问题 except requests.exceptions.Timeout: return False, f服务请求超时{self.health_check_timeout}秒 except Exception as e: return False, f健康检查发生未知错误: {str(e)} def check_system_resources(self): 检查系统资源使用情况 warnings [] # 检查CPU使用率 cpu_percent psutil.cpu_percent(interval1) if cpu_percent self.cpu_threshold: warnings.append(fCPU使用率过高: {cpu_percent:.1f}% (阈值: {self.cpu_threshold}%)) # 检查内存使用率 memory psutil.virtual_memory() if memory.percent self.memory_threshold: warnings.append(f内存使用率过高: {memory.percent:.1f}% (阈值: {self.memory_threshold}%)) # 检查GPU显存如果可用 if HAS_GPU: try: gpus GPUtil.getGPUs() for gpu in gpus: if gpu.memoryUtil * 100 self.gpu_memory_threshold: # memoryUtil是0-1之间的小数 warnings.append(fGPU {gpu.name} 显存使用率过高: {gpu.memoryUtil*100:.1f}% (阈值: {self.gpu_memory_threshold}%)) except Exception as e: warnings.append(f获取GPU信息时出错: {str(e)}) if warnings: return False, ; .join(warnings) else: return True, 系统资源使用正常 def check_service_logs(self): 扫描服务日志中的错误关键字 if not os.path.exists(self.log_file_path): return True, f日志文件不存在: {self.log_file_path}跳过日志检查 try: # 读取日志文件的最后N行例如1000行避免读取整个大文件 with open(self.log_file_path, r, encodingutf-8, errorsignore) as f: # 这是一个简单的方法对于大文件效率不高。生产环境建议用tail命令或记录上次读取位置。 lines f.readlines()[-1000:] recent_errors [] for line in lines: for keyword in self.error_keywords: if keyword.lower() in line.lower(): # 避免重复添加完全相同的错误行 if line not in recent_errors: recent_errors.append(line.strip()) break # 找到任意一个关键字就跳出内层循环 if recent_errors: # 只返回最近发现的几条错误 error_sample \n.join(recent_errors[-3:]) # 最后3条 return False, f在日志中发现错误关键字:\n{error_sample} else: return True, 日志中未发现关键错误信息 except Exception as e: return False, f读取或分析日志文件时出错: {str(e)} def send_dingtalk_alert(self, title, message): 通过钉钉机器人发送告警 if not self.dingtalk_webhook: print(钉钉Webhook未配置跳过发送) return False headers {Content-Type: application/json} data { msgtype: markdown, markdown: { title: title, text: f## {title}\n\n{message}\n\n**时间**: {time.strftime(%Y-%m-%d %H:%M:%S)} } } try: resp requests.post(self.dingtalk_webhook, headersheaders, datajson.dumps(data), timeout10) if resp.status_code 200: print(钉钉告警发送成功) return True else: print(f钉钉告警发送失败状态码: {resp.status_code}) return False except Exception as e: print(f发送钉钉告警时出错: {str(e)}) return False def send_email_alert(self, subject, body): 通过邮件发送告警 if not self.mail_config: print(邮件配置未设置跳过发送) return False try: msg MIMEText(body, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] self.mail_config[from_addr] msg[To] self.mail_config[to_addr] # 使用SMTP_SSL更安全端口通常465 server smtplib.SMTP_SSL(self.mail_config[smtp_server], self.mail_config[smtp_port]) server.login(self.mail_config[username], self.mail_config[password]) server.sendmail(self.mail_config[from_addr], [self.mail_config[to_addr]], msg.as_string()) server.quit() print(邮件告警发送成功) return True except Exception as e: print(f发送邮件告警时出错: {str(e)}) return False def run_monitor_cycle(self): 执行一次完整的监控检查 print(f\n 开始监控检查 {time.strftime(%Y-%m-%d %H:%M:%S)} ) all_ok True alert_messages [] # 1. 检查服务健康 health_ok, health_msg self.check_service_health() print(f服务健康: {✓ if health_ok else ✗} - {health_msg}) if not health_ok: all_ok False alert_messages.append(f[服务健康] {health_msg}) # 2. 检查系统资源 resource_ok, resource_msg self.check_system_resources() print(f系统资源: {✓ if resource_ok else ✗} - {resource_msg}) if not resource_ok: all_ok False alert_messages.append(f[系统资源] {resource_msg}) # 3. 检查服务日志 log_ok, log_msg self.check_service_logs() print(f服务日志: {✓ if log_ok else ✗} - {log_msg[:100]}...) # 只打印前100字符 if not log_ok: all_ok False alert_messages.append(f[服务日志] {log_msg}) # 4. 如果有问题发送告警 if not all_ok and alert_messages: alert_title f【告警】模型服务监控异常 alert_body \n\n.join(alert_messages) alert_body f\n\n检查时间: {time.strftime(%Y-%m-%d %H:%M:%S)} alert_body f\n服务地址: {self.api_url} # 同时发送钉钉和邮件如果配置了 self.send_dingtalk_alert(alert_title, alert_body) self.send_email_alert(alert_title, alert_body) print(f 监控检查完成状态: {正常 if all_ok else 异常} \n) return all_ok if __name__ __main__: monitor ModelServiceMonitor() # 在这里配置你的告警方式 # 1. 钉钉机器人配置可选 # monitor.dingtalk_webhook https://oapi.dingtalk.com/robot/send?access_tokenYOUR_TOKEN # 2. 邮件配置可选 # monitor.mail_config { # smtp_server: smtp.163.com, # smtp_port: 465, # username: your_email163.com, # password: 你的授权码, # 注意不是邮箱密码是SMTP授权码 # from_addr: your_email163.com, # to_addr: adminyourcompany.com # } # 运行一次检查 monitor.run_monitor_cycle()这个Python脚本已经具备了核心的监控功能。你需要根据实际情况修改几个地方self.api_url: 你的模型服务真实地址。self.log_file_path: 你的服务日志文件路径。在__main__部分配置你的钉钉Webhook或邮件SMTP信息。4.2 让Python脚本也定时运行同样我们可以用crontab来定时执行这个Python脚本。# 编辑crontab crontab -e # 添加一行每5分钟执行一次监控脚本并记录日志 */5 * * * * /usr/bin/python3 /home/yourname/model_service_monitor.py /var/log/model_monitor.log 21这样每5分钟脚本就会自动检查一次服务状态。5. 一些实用技巧和进阶想法基础的监控跑起来之后你可以根据实际需求再做一些增强1. 避免告警风暴如果服务连续出问题脚本会每分钟都发告警这可能会“轰炸”你。我们可以加一个简单的逻辑比如记录上次发送告警的时间如果相同问题在短时间内重复出现就抑制一下只发一次。2. 更聪明的日志分析上面的日志检查只是简单匹配关键字。你可以把它做得更智能比如使用正则表达式匹配更具体的错误模式。记录上次检查到的日志行号下次只检查新的日志提高效率。对错误信息进行归类统计看看哪种错误出现得最频繁。3. 添加性能指标监控除了“是否可用”我们还可以监控“性能如何”。比如在健康检查时记录API的响应时间。如果响应时间持续变慢可能意味着服务负载过高或存在性能瓶颈这可以作为一个预警指标。4. 将监控数据可视化你可以把每次检查收集到的数据CPU、内存、GPU使用率API响应时间写入到一个时间序列数据库比如InfluxDB然后用Grafana制作一个漂亮的监控仪表盘。这样就能直观地看到服务资源使用的历史趋势和实时状态。5. 与自动化修复结合监控发现问题后除了告警还可以尝试自动修复一些简单问题。比如如果检测到服务进程不存在脚本可以自动尝试重启服务。当然这要非常小心确保重启逻辑是安全的。6. 总结从头到尾走了一遍你会发现给模型服务搭建一个基础的自动化监控体系并没有想象中那么复杂。核心就是“定期检查 异常通知”。我们从最简单的Shell脚本健康检查开始逐步扩展到一个功能比较全面的Python监控脚本覆盖了服务连通性、资源使用和日志错误这三个关键维度。这套脚本在我自己的项目上运行了一段时间效果不错。最大的感受就是心里有底了晚上不用再惦记着服务器。当真的出现问题时钉钉消息能立刻推送到手机让我可以快速响应。当然这只是一个起点。监控是一个可以不断深入和优化的领域。你可以根据自己的业务重要性、团队规模和技术栈选择更专业的监控方案比如Prometheus Grafana的生态。但对于大多数个人开发者或小团队来说今天介绍的这套“自研”脚本已经能解决80%的日常监控需求了。建议你先从最核心的健康检查做起让它跑起来。然后再逐步添加资源监控和日志分析。最重要的是一定要配置好告警通知让监控真正发挥作用。希望这篇文章能帮你更好地守护你的AI服务。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。