
这个场景是不是似曾相识:凌晨三点十四分,手机狂震。
从床上弹起来,心跳飙到120,打开企业微信一看——
⚠️ 告警:web-03 CPU使用率 82.3%
八十二。
我盯着屏幕看了五秒钟,把手机摔回枕头底下,翻了个身继续睡。
不是我摆烂,是这个破告警我见过太多次了。web-03是台老机器,跑的是个Java应用,JVM一起来CPU就没低于过70%。设了80%的告警阈值,结果一天能报十几次,每次打开看都啥事没有。
这种“狼来了”式的告警,比不告警还可怕——它让你对所有告警都失去信任,等真正出故障的时候,你反而不当回事了。
被折腾了半年之后,我终于下决心自己写了个监控脚本。核心思路就一个:与其让告警通知我,不如让告警只在该通知的时候通知我。
现在跑了快一年了,半夜被吵醒的次数从每月四五次降到基本为零。今天把这套东西整理出来,代码不多,但坑踩了不少。
往期阅读>>>
Python 自动化管理Jenkins的15个实用脚本,提升效率
App2Docker:如何无需编写Dockerfile也可以创建容器镜像
Python 自动化识别Nginx配置并导出为excel文件,提升Nginx管理效率
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
