Python自动化巡检工具开发：从SSH采集到告警报告

大半夜被短信轰炸叫起来看告警的日子我估计不少运维兄弟都经历过。我刚接手几十台服务器的时候每天早上一睁眼第一件事就是挨个登上去看CPU、磁盘、内存碰上业务高峰期还要盯着日志。后来我实在受不了这种重复劳动花了两周时间写了一套自动化巡检工具。那之后不光早上能多睡半小时很多潜在故障反而比用户先发现。这篇就把这套工具从设计思路、代码实现到落地部署的所有过程包括踩过的坑一次性说清楚。先说清楚这套工具能干什么它能按照你设定的频率批量采集服务器的CPU、内存、磁盘、网络、关键进程这些核心指标自动判断状态是否健康把异常情况推送到你的手机上最后还会生成一份可以直接发给领导的巡检日报。整个过程完全无人值守跑起来之后你只需要在收到告警的时候看一眼。文章适合有一定基础、开始独立维护多台机器的技术人员不涉及复杂平台全程用Python标准库加少量第三方库就能实现。1. 整体设计与思路拆解1.1 先搞清楚一个问题为什么不用现成的监控系统每个人问我自动化运维工具开发第一反应都是“你装个开源监控平台不就行了干嘛自己写”。这话对了一半。对于大型集群直接上成熟的监控平台确实更划算组件多、生态全、图表漂亮。但现实往往是我当初那个处境一个机房几十台机器业务规模不算大开放平台要么占用资源偏重想完成跑通最少也得折腾两三天给非技术同事看报告还得教他们怎么打开平台。而且最烦的一点是平台给你预留的告警规则和报告模板改起来很痛苦。所以我在设计整个方案之前先问了自己三个问题这工具核心要解决什么答日常巡检和异常通知。工具运行在哪里答在运维跳板机上用SSH协议去采集所有目标服务器状态。输出给谁看答给执行巡检的工程师看也给不写代码的业务负责人看。基于这三个答案设计出了一个很明确的思路——轻量、低依赖、规则透明、报告可读。1.2 模块划分与数据流动把整个工具按照功能拆成四个独立模块这个设计在后面维护时省了无数事采集模块负责建立SSH连接远程执行命令收集原始指标数据。解析模块把每台机器返回的字符串结果解析成统一格式的字典数据。判定模块读取配置中的阈值对照数据做出健康/警告/严重判定。通知与报告模块把判定结果写入日志推送告警信息到IM生成当日巡检报告。这四个模块之间用最简单的函数调用连接不用消息队列不用数据库全部结果直接存在内存中最后一次性落盘。数据流动是配置解析 → 采集循环 → 逐台判定 → 汇总输出。这套流程是我在所有自动化运维工具中反复验证过的最稳定路径。2. 核心模块划分与目录结构2.1 项目目录怎么组织最合理我当时建项目目录时特意把代码、配置、输出分开避免日志和报告跟源码混在一起。目录结构用了一个很经典的布局ops_tool/ ├── config.yaml # 所有可调整参数集中在此 ├── requirements.txt # 依赖清单 ├── main.py # 入口负责调度整个流程 ├── hosts.yaml # 服务器清单 ├── collectors/ │ ├── __init__.py │ ├── ssh_client.py # SSH连接与命令执行 │ └── metric_reader.py # 各指标的具体采集函数 ├── parsers/ │ ├── __init__.py │ └── linux_parser.py # 解析Linux命令行输出 ├── checks/ │ ├── __init__.py │ └── threshold.py # 阈值判定逻辑 └── notify/ ├── __init__.py ├── messenger.py # 通知渠道抽象 └── report.py # 生成Markdown和HTML报告这个目录结构看起来简单但它解决了一个核心问题职责清晰。每个目录只干一件事后续想加Windows支持只需要在collectors下面加一个win_metric_reader想加短信通知只需在notify下面新增一个channel。我后来加新功能时基本只需要动两个文件。2.2 配置文件设计的巧与坑所有参数集中管理是这类工具的第一原则。我在配置里不写死任何阈值全部用YAML维护# config.yaml job: name: daily-inspect ssh_timeout: 10 connect_retries: 2 thresholds: cpu_warn: 75 cpu_critical: 90 mem_warn: 80 mem_critical: 92 disk_warn: 85 disk_critical: 95 notify: channel: wecom # wecom / dingtalk / email webhook_url: https://example.invalid/hook alert_interval: 300 # 相同告警最小重发间隔秒 report: output_dir: ./reports date_format: %Y-%m-%d这里有一个非常关键的细节告警重发间隔。刚开始我没设计这个参数结果有一次某台机器磁盘持续满了两个小时我的手机每五分钟被轰炸一次。这个参数的作用就是记录每条告警指纹在指定时间窗口内重复出现就不再推送既不会漏又不会烦。2.3 依赖取舍的底层逻辑整个工具我只用了三个第三方依赖PyYAML用于解析配置paramiko用于SSH连接jinja2用于渲染HTML报告模板。其余全部使用Python标准库。有人可能觉得paramiko有点重换subprocess调系统ssh命令可行但那样要额外处理密钥、known_hosts、超时反而更麻烦。paramiko直接支持私钥加载、超时控制、命令执行是SSH自动化方案的目前最优解。关于版本锁定我在requirements.txt里全部锁到了具体的补丁版本pyyaml6.0.1 paramiko3.4.0 jinja23.1.4锁版本是因为这类工具往往在跳板机上常年不升级有一次paramiko从2.x升到3.x算法的默认行为有变化直接导致几十台机器批量连接失败。从那以后凡是部署长期运行的脚本我都锁死版本。3. 核心采集逻辑与判定标准实现3.1 SSH批量连接的正确姿势采集的第一步就是把hosts.yaml里定义的服务器列表逐个连接。hosts.yaml长这样servers: - name: web-01 host: 192.168.10.11 port: 22 user: ops key_path: /home/ops/.ssh/id_ed25519 - name: db-01 host: 192.168.10.12 port: 22 user: ops key_path: /home/ops/.ssh/id_ed25519写SSH采集的时候最需要控制的是批量并发量和超时时间。如果几十台机器串行跑每台耗时三秒一轮下来就是几分钟性能很差但如果同时并发又容易把跳板机连接数打满。我最后采用了信号量控制并发为5配合单机超时10秒的方案。以下是核心的连接执行代码这个函数是我抽出来给所有采集指令用的公共方法# collectors/ssh_client.py import paramiko from concurrent.futures import ThreadPoolExecutor, as_completed import threading class SSHClientPool: 带并发控制的SSH客户端池避免瞬间创建太多连接 def __init__(self, max_workers5): self.semaphore threading.Semaphore(max_workers) def execute(self, server, command, timeout10): with self.semaphore: client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect( hostnameserver[host], portserver.get(port, 22), usernameserver[user], key_filenameserver[key_path], timeouttimeout, banner_timeouttimeout, auth_timeouttimeout ) stdin, stdout, stderr client.exec_command(command, timeouttimeout) out stdout.read().decode(utf-8, errorsignore) err stderr.read().decode(utf-8, errorsignore) return {server: server[name], ok: True, stdout: out, stderr: err} except Exception as exc: return {server: server[name], ok: False, error: str(exc)} finally: client.close() def run_all(self, servers, command): results {} with ThreadPoolExecutor(max_workersself.MAX_WORKERS) as pool: futures {pool.submit(self.execute, srv, command): srv[name] for srv in servers} for fut in as_completed(futures): res fut.result() results[res[server]] res return results这段代码里有几个细节值得说明。connect时把banner_timeout和auth_timeout都设置成跟timeout一致是因为有些机器在SSH握手阶段会卡住只设置connect的timeout不够容易发生长时间挂起。读取输出时用了errorsignore实际工作中遇到过语言环境不同有个别机器返回了非UTF-8编码导致脚本崩溃直接忽略编码错误最省事。3.2 用什么命令采集指标才最可靠采集指标时我全部选择Linux系统自带命令没有用psutil。有两个原因一是很多服务器为了安全起见面没有Python的psutil库也不能随意装包二是在远程环境里自带命令标准输出更容易被解析。这一套命令是我调试多次后稳定可靠的方案# collectors/metric_reader.py COMMANDS { cpu_usage: top -bn1 | grep Cpu(s) | awk -Fid, {print 100 - $2}, mem_usage: free -m | awk NR2{printf \%.1f\, $3*100/$2}, disk_usage: df -P / | awk NR2{print $5} | tr -d %, load_avg: cat /proc/loadavg | awk {print $1}, tcp_estab: ss -ant | grep ESTAB | wc -l, }这里有一处比较容易踩坑的是CPU使用率。很多参考代码用top取第一行没有加循环次数在某些系统上top第一次运行的不是进程状态而是平均值表现不稳定。用top -bn1并且只用CPU空闲比例倒推实测下来是兼容性最好的方案。内存使用率用free的第三列除以第二列输出加了一位小数方便后面做数值判定。磁盘用df -P /这个-P参数很有用它会强制使用POSIX格式输出避免因为路径折叠导致解析错行。3.3 阈值判定背后的算法思路拿到指标之后判定模块要做的事很简单数值比较但比较逻辑要设计好。每个判定返回三个字段状态level、当前值value、阈值limit。# checks/threshold.py def check_cpu(usage, warn75, critical90): if usage critical: return {level: critical, value: usage, threshold: critical} if usage warn: return {level: warning, value: usage, threshold: warn} return {level: ok, value: usage, threshold: critical}关键思想是每个阈值规则放到独立函数里传参用配置值。这样无论是想加内存阈值还是磁盘阈值逻辑完全一致。判定完成后把所有结果汇总成一个列表每个元素包含服务器名、指标名、level、value、message。这张列表就是后续通知和报告的唯一数据源。4. 通知机制与报告生成实操4.1 多通道消息推送的抽象设计这一节可能是大家最关心的。告警消息推送我设计了一个统一接口定义了send方法子类分别实现企业微信、钉钉、邮件的具体逻辑。以最容易调试的企业微信机器人为例# notify/messenger.py import requests, time, hashlib class ChannelBase: def send(self, title, content): raise NotImplementedError class WecomChannel(ChannelBase): def __init__(self, webhook_url): self.webhook_url webhook_url def send(self, title, content): payload { msgtype: markdown, markdown: { content: f### {title}\n{content} } } resp requests.post(self.webhook_url, jsonpayload, timeout5) return resp.status_code 200由于工具部署在内网环境要确保跳板机能访问到webhook地址。实际测试中会碰到webhook地址配置错误、端口被防火墙挡掉等情况所以send方法必须有返回值并且调用方要把发送失败记录到日志。这里还要提一个项目里必须实现的方法——告警去重。实现思路是维护一个字典key是告警指纹服务器名指标名value是上次发送时间秒级差值小于配置的alert_interval就直接跳过。这套“指纹去重”机制在连续抖动场景下非常关键日志抖动和真实故障的最大区别就是消息频率没有这个机制你会被重复警报淹没。4.2 自动生成日报从Markdown到HTML巡检报告是这套工具的另一大产出。每天9点跑完后生成一份报告Markdown用于留存查阅HTML用于发送给需要看结果的人。这里的关键是让报告生成函数接收一个统一的结果列表而不是每个模块单独去读文件。# notify/report.py from jinja2 import Environment, FileSystemLoader from datetime import datetime def build_markdown(results): lines [# 巡检日报, f生成时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)}, ] for item in results: lines.append(f- [{item[level].upper()}] {item[server]}: {item[metric]} {item[value]}% (阈值 {item[threshold]}%)) return \n.join(lines) def build_html(markdown_text, output_path): env Environment(loaderFileSystemLoader(./templates)) template env.get_template(report.html.j2) html template.render(contentmarkdown_text, title巡检日报) with open(output_path, w, encodingutf-8) as f: f.write(html)HTML模板我用了jinja2模板文件里只需要一个渲染content的位置即可。Markdown转HTML这一步如果你想做得更严谨可以用markdown库把表格和列表转成真正的HTML但我实测下来用空格保留格式的原始Markdown文本放在pre标签里反而在可读性和兼容性上更好尤其内容包含多行代码时渲染不容易乱。4.3 Crontab定时调度的完整配置工具跑起来之后剩下的就是定时调度问题。我直接用了Cron因为这台工具所在跳板机本身有在运行。# 每天上午9点执行巡检并生成日报 0 9 * * * cd /opt/ops_tool /usr/bin/python3 main.py --daily /var/log/ops_tool.log 21 # 每5分钟执行一次快速健康检查只有异常才会推送 */5 * * * * cd /opt/ops_tool /usr/bin/python3 main.py --quick /var/log/ops_tool.log 21注意main.py要有参数解析逻辑区分全量巡检和快速检查。快速检查意味着只采集CPU、磁盘两个指标其余跳过。这样可以大大降低采集频率对目标机器的干扰因为每五分钟全量采集一次相对较重对DB机器业务会有一定影响。我把主程序入口写成下面这种形式支持--daily与--quick两种模式# main.py import argparse, yaml, logging from collectors.metric_reader import collect_metrics from checks.threshold import judge_all from notify.messenger import get_channel from notify.report import build_markdown, build_html def load_config(pathconfig.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def run_daily(cfg): hosts load_hosts() raw collect_metrics(hosts, cfg) results judge_all(raw, cfg[thresholds]) md build_markdown(results) html build_html(md) channel get_channel(cfg[notify]) issues [r for r in results if r[level] ! ok] if issues: channel.send(巡检发现异常, md) save_report(md, html, cfg[report][output_dir]) def run_quick(cfg): # 快速模式仅检测cpu和disk异常时直接推送 pass if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--daily, actionstore_true) parser.add_argument(--quick, actionstore_true) args parser.parse_args() cfg load_config() logging.basicConfig(levellogging.INFO) if args.daily: run_daily(cfg) elif args.quick: run_quick(cfg)这里还有一个小细节日志输出中特别打印一条“巡检完成正常N台异常M台”方便每天早上去看一眼日志确认工具真的执行了。如果哪一天没看到这条日志说明调度或脚本有问题这个比看监控面板还可靠。5. 常见问题与排查技巧实录5.1 SSH连接卡死或超时的终极排查这个工具上线后最常见的问题是某几台机器总是连接超时。排查顺序我整理了一份基本可以解决95%的问题现象可能原因处理方式所有机器超时跳板机出网受限在跳板机上手动ssh一台测试检查防火墙和安全组单台机器超时目标机器负载过高登录目标机器看是否内存耗尽导致sshd拒绝握手可以登录但exec_command挂起命令本身执行慢把timeout从10秒调大到30秒或者改用异步方式偶发连接失败ssh密钥权限不对检查.ssh目录和私钥文件权限是否为600其实还有一个经验丰富的排查技巧在SSH连接失败时立刻手动执行ssh -v userhost观察握手日志比任何代码内日志都快。可以用paramiko的日志级别调整为DEBUG打印更多细节。5.2 采集命令在个别机器上返回乱码这个问题在我维护的机器中真实遇到过。有一些机器的语言环境是中文的df输出的表头中文导致第五列解析错位。解决方案是在命令前加LANGC强制所有命令使用英文输出环境。修改后的磁盘命令变成LANGC df -P / | awk NR2{print $5} | tr -d %最后我在所有采集命令前都统一加上LANGC这样避免了后续所有解析端的语言环境差异问题。这是自动化脚本的通用经验跨机器执行任何命令先强制固定环境变量避免因为系统语言不同导致解析结果不可预测。5.3 告警风暴与漏报的平衡技巧报警阈值定得不好工具会被打入冷宫。我第一版阈值参考某文档设了cpu警告50%结果每天全是告警因为很多批处理任务在夜里会瞬时跑满CPU。后来我学到一个办法统计一周的基础指标取均值加上2倍标准差作为警告阈值这个“基线漂移”的思路让告警准确率高了很多。同时还要设计“连续N次才告警”的机制防止瞬时尖峰误报。我在判定模块中加了一个计数逻辑同一台机器同一指标连续两次触发warning以上才会推送。实现代码很简单只要维护一个全局计数器在快速检查模式下实时更新达到连续次数才触发。5.4 报告乱码或模板渲染失败报告模板渲染失败的原因九成是缺失变量。比如模板中写了{{ item.metric }}但数据里字段名拼写有误。这类报错很好查毕竟有明确的NameError。乱码问题则来自写入HTML文件时的编码必须显式指定encodingutf-8。有不少人写的代码只顾着打开文件却忘了指定编码生成的报告在Windows浏览器打开就会乱成一片。5.5 日志文件不断膨胀的清理策略工具一旦长期运行日志文件不清理会吃满磁盘。我在设计方案时特意加了一个rotate逻辑每天归档当天的日志保留最近30天超期自动删除。最简单的写法是Crontab加一行# 每月1号清理超过30天的日志保留最新的压缩包 0 3 1 * * find /var/log/ops_tool* -mtime 30 -delete这里要说明一点不要在启用日志rotate前让工具的日志文件无限增长。我见过最惨的例子是监控工具自身把磁盘吃完导致被监控的服务也受影响那真是黑色幽默。6. 我应该从哪里开始改造这套工具做完第一版后我就意识到这套工具的扩展点非常多。第一个方向是资产管理把后端稳定运行的主机清单接入配置生成流程让新机器开完机自动纳入巡检这是目前投入产出比最高的改造。第二个方向是SQLite存储历史指标目前巡检数据只存在当天内存中一旦有了历史库就能画趋势图和做更好的异常检测。第三个方向是权限收敛目前所有跳板机都用同一把私钥后期可以切换到每个目标机一个临时密钥、定期轮换的方案让自动化运维所在的基础安全环境更可控。我自己在实际运行这套工具半年后最大的体会是不要一开始就追求大而全先让“每天能自动跑、异常能收到、报告能看懂”这三件事稳定运转再逐步扩展。说实话自动化运维工具的价值不在于它多复杂而在于它替你省掉了多少重复劳动以及帮你在故障发生前抢到多少时间。现在每天早上我做的第一件事只是拿手机看一眼有没有异常推送没有就安心开始一天的工作这种感觉比任何技术指标都来得实在。