当前位置:首页>python>写了个Python脚本监控服务器,再也不怕半夜被告警吵醒

写了个Python脚本监控服务器,再也不怕半夜被告警吵醒

  • 2026-09-10 11:28:45
写了个Python脚本监控服务器,再也不怕半夜被告警吵醒

这个场景是不是似曾相识:凌晨三点十四分,手机狂震。

从床上弹起来,心跳飙到120,打开企业微信一看——

⚠️ 告警:web-03 CPU使用率 82.3%

八十二。

我盯着屏幕看了五秒钟,把手机摔回枕头底下,翻了个身继续睡。

不是我摆烂,是这个破告警我见过太多次了。web-03是台老机器,跑的是个Java应用,JVM一起来CPU就没低于过70%。设了80%的告警阈值,结果一天能报十几次,每次打开看都啥事没有。

这种“狼来了”式的告警,比不告警还可怕——它让你对所有告警都失去信任,等真正出故障的时候,你反而不当回事了。

被折腾了半年之后,我终于下决心自己写了个监控脚本。核心思路就一个:与其让告警通知我,不如让告警只在该通知的时候通知我。

现在跑了快一年了,半夜被吵醒的次数从每月四五次降到基本为零。今天把这套东西整理出来,代码不多,但坑踩了不少。

往期阅读>>>

Python 为什么会成为AI时代的头部语言

Python 40个常用的列表推导式

Python 50个提高代码开发效率的方法

Python 自动检测服务HTTPS证书过期时间并发送预警

Python 自动化操作Redis的15个实用脚本

Python 自动化管理Jenkins的15个实用脚本,提升效率

Python copyparty搭建轻量的文件服务器的方法

Python 实现2FA认证的方法,提升安全性

Python 封装20个常用API接口,提升开发效率

App2Docker:如何无需编写Dockerfile也可以创建容器镜像

Python 集成 Nacos 配置中心的方法

Python 35个JSON数据处理方法

Python 字典与列表的20个核心技巧

Python 15个文本分析的库,提升效率

Python 15个Pandas技巧,提升数据分析效率

Python 运维中30个常用的库,提升效率

Python调用远程接口的方法

Python 提取HTML文本的方法,提升效率

Python 应用容器化方法:实现“一次部署,处处运行”

Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率

Python 5个常见的异步任务处理框架

Python数据科学常见的30个库

Python 50个实用代码片段,优雅高效


先说基础:psutil这个库真的好用

Python做服务器监控,psutil 是标配。这个库基本上把你能想到的系统指标都封装好了:

import psutil# CPU使用率(取1秒的采样)cpu = psutil.cpu_percent(interval=1)# 内存mem = psutil.virtual_memory()print(f"内存使用率: {mem.percent}%")print(f"可用内存: {mem.available / 1024**3:.1f}GB")# 磁盘disk = psutil.disk_usage('/')print(f"磁盘使用率: {disk.percent}%")# 网络net = psutil.net_io_counters()print(f"累计发送: {net.bytes_sent / 1024**2:.1f}MB")print(f"累计接收: {net.bytes_recv / 1024**2:.1f}MB")# 进程级别的监控也行(注意:第一次调用cpu_percent返回0,需调两次)psutil.process_iter(['pid', 'name', 'cpu_percent'])  # 第一次,预热time.sleep(1)for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):if proc.info['cpu_percent'] andproc.info['cpu_percent'] >50:print(f"高CPU进程: {proc.info['name']} (PID:{proc.info['pid']})")

pip install psutil 装完就能用,跨平台,Linux/macOS/Windows都行。不需要什么特殊权限,普通用户就能跑大部分指标。

有了数据源,接下来就是怎么判断“该不该告警”。这才是重点。


告警策略,这才是重点

大多数入门级监控教程的告警逻辑是这样的:

if cpu>80:send_alert("CPU过高!")

我当初也是这么写的。然后就被现实教育了。

问题出在哪?这种“一刀切”的阈值完全不考虑实际情况。CPU 82%在空闲时段确实该报警,但如果这台机器本身就在跑批处理任务,82%完全正常。你得让脚本有点“判断力”。

我后来改成了分级策略,思路很简单:

# config.py — 监控配置集中管理THRESHOLDS = {'cpu': {'warning': 80,       # 提醒级别,不紧急'critical': 95,      # 严重级别,需要关注'sustained_minutes': 5# 连续超过阈值N分钟才算真的有问题    },'memory': {'warning': 85,'critical': 95,'sustained_minutes': 3    },'disk': {'warning': 85,'critical': 95,'sustained_minutes': 0# 磁盘不需要持续判断,超了就是超了    }}# 告警级别定义ALERT_LEVELS = {'info': '仅记录日志','warning': '发群消息,不@人','critical': '发群消息,@值班人'}

改了俩地方:阈值分成了 warning 和 critical 两档,不同级别走不同的通知通道;另外加了个持续时长判断——CPU偶尔飙一下太正常了,连续5分钟都高才是真有问题。


持续时长判断,这个功能救了我的命

这是整套脚本里我最满意的一个功能,也是让告警量直接砍掉80%的关键。

原理很简单:每次检测到指标超标时,不急着报警,先记下来。如果接下来连续N次检测都超标,才真正发出告警。

import timefrom collections import defaultdictclass AlertStateManager:"""管理每个指标的告警状态"""def __init__(self):# 记录每个指标首次超标的时间self.first_breach = {}# 记录已经告警过的指标,避免重复self.alerted = set()def check(self, metric_name, value, threshold, sustained_seconds=300):"""        检查指标是否需要告警        返回: None(不告警), 'warning'(首次告警), 'repeat'(重复告警)        """key = f"{metric_name}:{threshold}"if value>threshold:now = time.time()if key not in self.first_breach:# 第一次超标,记下来,先不报警self.first_breach[key] = nowreturn None# 已经超标一段时间了,判断是否达到持续时长duration = now-self.first_breach[key]if duration>= sustained_seconds:if key not in self.alerted:self.alerted.add(key)return 'warning'return 'repeat'# 持续超标但已经报过了return None# 还没到持续时长,继续观察else:# 指标恢复正常,清除记录if key in self.first_breach:del self.first_breach[key]if key in self.alerted:self.alerted.discard(key)return 'recovered'# 可选:发一条恢复通知return None

这段代码跑起来的逻辑是:CPU第一次飙到阈值以上,脚本默默记下来。过了5分钟还在阈值以上,才真正发告警。如果中间掉回正常值以下,计时器清零,从头再来。

光这一个功能,就把那些“瞬时波动”引发的假告警全干掉了。


告警聚合,别让群消息炸了

另一个让我头疼的问题是:一旦服务器出问题,往往是连锁反应。CPU高了,内存也跟着飙,磁盘IO也跟着打满。按以前的逻辑,三个指标各发一条告警,群里瞬间三条消息。如果十台服务器同时出问题,那就是三十条。

所以我在发送端做了一层聚合:

from collections import defaultdictimport threadingclass AlertAggregator:"""在一个时间窗口内收集告警,合并发送"""def __init__(self, window_seconds=60):self.window = window_secondsself.pending = defaultdict(list)self.timer = Noneself.lock = threading.Lock()def add(self, host, metric, value, level):with self.lock:self.pending[host].append({'metric': metric,'value': value,'level': level            })# 每次来新告警都重置定时器(标准debounce策略)if self.timer is not None:self.timer.cancel()self.timer = threading.Timer(self.window, self._flush)self.timer.start()def _flush(self):"""把攒了一分钟的告警合并成一条消息发出去"""with self.lock:if not self.pending:self.timer = Nonereturnlines = ["## ⚠️ 服务器告警汇总\n"]for host, alerts in self.pending.items():lines.append(f"**{host}**:")for a in alerts:emoji = "🔴"ifa['level'] == 'critical'else"🟡"lines.append(f"> {emoji} {a['metric']}:{a['value']}")lines.append("")message = "\n".join(lines)send_to_wechat(message)  # 你之前写的通知函数self.pending.clear()self.timer = None

这样做的好处是:一分钟内产生的所有告警,会合并成一条消息发出来。群里清爽,信息量还更大。

我之前有次一台服务器挂了,连锁反应触发了12条告警。用了聚合之后,群里只收到一条消息,但里面12个指标全列出来了。同事说这个体验比被消息刷屏好太多了。


静默期:半夜就别吵我了

这个功能是写这套脚本的初衷——我真的不想半夜被告警吵醒。

但也不能真的把告警全关了,万一半夜服务器宕机呢?所以得区分“重要”和“不重要”。

from datetime import datetimeclass QuietHoursManager:def __init__(self, quiet_start=23, quiet_end=8):self.quiet_start = quiet_startself.quiet_end = quiet_enddef is_quiet_now(self):hour = datetime.now().hourreturn hour>= self.quiet_startorhour<self.quiet_enddef should_alert(self, level):"""        静默期间的告警策略:        - critical 级别:照常告警,该叫就叫        - warning 级别:只记日志,等早上统一汇报        - info 级别:直接忽略        """if not self.is_quiet_now():returnTrue# 非静默期,全部告警if level == 'critical':returnTrue# 严重问题,半夜也得叫起来# warning 和 info 在静默期只记日志log_to_file(f"[静默期-抑制] {level} 级别告警已记录,待早上汇报")return False

逻辑很直接:晚上11点到早上8点之间,只有 critical 级别的告警才会推送到群里。warning 的存到日志里,第二天早上发一条汇总。

我还在早上8点自动发一条“夜间简报”:

def morning_summary():"""每天早上8点发送夜间告警汇总"""overnight_alerts = get_alerts_since(hours_ago=9)if not overnight_alerts:send_to_wechat("## ☀️ 早间简报\n> 昨晚一切正常,无告警记录。")else:lines = [f"## 🌅 夜间告警汇总(共{len(overnight_alerts)}条)\n"]for alert in overnight_alerts:lines.append(f"> [{alert['time']}] {alert['host']} - {alert['metric']}: {alert['value']}")send_to_wechat("\n".join(lines))

这条消息现在是我们组每天早上的“早餐伴侣”,大部分人看一眼“无告警”就去干活了。


进程级别的监控,有时候比系统指标更有用

系统级的CPU、内存告警有时候不够精准。比如一台服务器CPU只有60%,但某个进程自己吃掉了40%——系统层面看着没事,但这个进程可能已经在影响业务了。

所以我又加了一层进程监控:

import psutildef check_top_processes(top_n=5):"""找出占资源最多的进程"""processes = []for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent', 'status']):try:info = proc.infoif info['cpu_percent'] andinfo['cpu_percent'] >20:processes.append(info)except (psutil.NoSuchProcess, psutil.AccessDenied):continue# 按CPU排序,取前Nprocesses.sort(key=lambda x: x.get('cpu_percent', 0), reverse=True)return processes[:top_n]def monitor_specific_process(process_name, max_memory_mb=2048):"""监控特定进程的内存使用,防止内存泄漏"""for procinpsutil.process_iter(['pid', 'name', 'memory_info']):try:if proc.info['name'] == process_name:mem_mb = proc.info['memory_info'].rss/1024/1024if mem_mb>max_memory_mb:return {'alert': True,'message': f"进程 {process_name}(PID:{proc.info['pid']}) 内存占用 {mem_mb:.0f}MB,超过阈值 {max_memory_mb}MB"                    }except (psutil.NoSuchProcess, psutil.AccessDenied):continuereturn {'alert': False}

这个功能在排查Java应用的内存泄漏时帮了大忙。有次我们的Spring Boot应用内存一直在涨,涨到3G多还没触发系统级告警(因为总内存有16G),但进程级监控直接报了。提前发现,提前处理,没酿成线上事故。


把上面的东西串起来

最后写一个主循环,把所有模块串起来。这里只展示CPU检查的完整逻辑,内存和磁盘同理:

import timeimport psutil# 初始化各模块alert_state = AlertStateManager()aggregator = AlertAggregator(window_seconds=60)quiet_hours = QuietHoursManager(quiet_start=23, quiet_end=8)def check_and_alert(metric_name, value, config):"""统一的告警检查逻辑"""host = get_hostname()# 先检查critical级别status = alert_state.check(f'{metric_name}_critical', value,config['critical'],sustained_seconds=config['sustained_minutes'] *60    )if status == 'warning':if quiet_hours.should_alert('critical'):aggregator.add(host, metric_name, f'{value}%(严重)', 'critical')return# critical已触发,不用再检查warning# 再检查warning级别status = alert_state.check(f'{metric_name}_warning', value,config['warning'],sustained_seconds=config['sustained_minutes'] *60    )if status == 'warning':if quiet_hours.should_alert('warning'):aggregator.add(host, metric_name, f'{value}%', 'warning')def run_monitor(interval=60):"""主监控循环"""print("监控已启动,按 Ctrl+C 停止...")whileTrue:try:cpu = psutil.cpu_percent(interval=1)mem = psutil.virtual_memory().percentdisk = psutil.disk_usage('/').percentcheck_and_alert('CPU', cpu, THRESHOLDS['cpu'])check_and_alert('内存', mem, THRESHOLDS['memory'])# 磁盘sustained_minutes=0,超了就报check_and_alert('磁盘', disk, THRESHOLDS['disk'])# 检查关键进程java_status = monitor_specific_process('java', max_memory_mb=3072)if java_status.get('alert'):if quiet_hours.should_alert('warning'):aggregator.add(get_hostname(), 'Java进程',java_status['message'], 'warning')except Exceptionase:# 监控脚本自身出错也要通知,不能悄悄挂了send_to_wechat("监控脚本异常: {e}")time.sleep(interval)

跑起来之后,脚本每分钟检查一次所有指标。大部分时候它安安静静的,只有真正出了问题才会说话。


踩过的坑,都是凌晨三点换来的

坑一:psutil.cpu_percent() 第一次调用永远不准

这个坑我踩了好久。psutil.cpu_percent(interval=1) 第一次调用的结果往往偏高或者偏低,因为它需要一个时间窗口来计算CPU使用率,第一次采样没有对比基准。

解决办法:启动时先调一次丢掉结果:

# 启动时预热一次(interval=1 会阻塞1秒,自带采样)psutil.cpu_percent(interval=1)# 从第二次调用开始才是有效数据

文档里有提到这个,但我当时没仔细看,浪费了一个下午排查“为什么脚本一启动就报警”。

坑二:监控脚本自己挂了没人知道

这是个很讽刺的问题——你写了个监控脚本,结果监控脚本自己挂了,然后没人发现。

所以我做了两件事:一是脚本自身用 supervisor 管理,挂了自动拉起来;二是加了心跳机制——脚本每10分钟往一个文件里写一次时间戳,再写一个独立的小脚本每15分钟检查一次这个文件,如果时间戳太久没更新,就发告警说“监控脚本可能挂了”。

套娃是套娃了点,但管用。

坑三:时区问题又出现了

对,又是时区。我的静默期逻辑用的是 datetime.now().hour,结果服务器是UTC时区。我以为设置了晚上11点到早上8点静默,实际上是UTC 23:00到UTC 8:00,换算成北京时间是早上7点到下午4点。

也就是说,工作时间的告警全被吞了,半夜反而正常推送。

修复很简单:

from datetime import datetimeimport pytzdef get_local_hour():tz = pytz.timezone('Asia/Shanghai')return datetime.now(tz).hour

坑四:阈值设太低,新机器一上来就疯狂告警

换了台新服务器,配置比老机器好很多,但监控脚本直接复用了老配置。结果新机器跑批处理的时候CPU能到90%(虽然完全没问题),老脚本不管这些,照报不误。

后来我给每台机器单独配了阈值,放在 config_{hostname}.yaml 里。别偷懒所有机器共用一份配置,硬件不一样的机器,“正常”的标准也不一样。


一张图看清告警流程

指标采集(每分钟)    ↓阈值判断(warning / critical)    ↓持续时长检查(连续超标N分钟?)    ↓ 是                    ↓ 否告警聚合器            继续观察(计时器清零)(60秒窗口内合并)    ↓静默期判断(当前是夜间?)    ↓ 是                    ↓ 否只推送critical         正常推送warning记日志    ↓早上8点发送夜间汇总

现在这套监控跑了快一年了

说几个数字吧:

刚上线那会儿,每天群里的告警消息大概30-40条。上了持续时长判断之后,降到了5-8条。再加上聚合和静默期,现在平均每天2-3条,每条都是有实际意义的。

半夜被告警叫醒的次数,从之前每月四五次,变成了过去11个月总共被叫了2次——那两次还确实是真故障,幸亏叫醒了。

整套代码加起来大概300多行,没有用什么高大上的框架,就是一个Python脚本加crontab。但它给我的感觉就是——终于有人替我值夜班了。

当然这脚本也不是万能的。上个月有次磁盘IO打满导致应用卡死,它就没检测到——因为我压根没写IO监控。后来补上了,但这件事提醒我:监控永远没有“做完”的一天,你永远在补下一个盲区。

如果你也被告警噪音困扰过,建议先从“持续时长判断”这个功能做起。就这一个改动,能让你的告警质量直线上升。

你们线上用的是Prometheus+Grafana这种专业方案,还是和我一样自己糊脚本?有用过告警聚合或者静默期策略的吗?欢迎在评论区交流下你的做法。

“无他,惟手熟尔”!有需要的用起来!
------加入知识库与更多人一起学习------

https://ima.qq.com/wiki/?shareId=f2628818f0874da17b71ffa0e5e8408114e7dbad46f1745bbd1cc1365277631c

最新文章

随机文章