运维这活儿说到底就是跟重复劳动斗智斗勇。我常年一人守着几十台服务器每天早上的第一件事就是SSH登录逐台看负载、看磁盘、看进程看得多了就忍不住想骂人。后来我干脆花了两个晚上从零开始SOLO实战独立写出了一个属于自己的运维小工具专门把“巡检”这件事自动化。整个过程从需求梳理、技术选型、代码编写、测试验证到上线维护全程都是一个人搞定。这篇文章就是我当时的最真实记录希望能给同样打算自己动手写工具的运维朋友一点参考。1. 为什么单独写运维小工具需求与边界1.1 我面对的真实痛点监控系统并不万能先说背景我手上有一批服务器数量不多不少。要说没有监控吧公司也搭过一些开源监控平台但问题是那套系统偏重日常看板、告警规则、图表面板都要配真正出问题的时候一堆报警刷屏反而分不清轻重。尤其是每天早上例行巡检我需要确认每台机器的CPU、内存、磁盘、关键服务状态平台虽然有但是要在一堆图表里自己找不如我直接看几行数字来得快。真正压迫我动手的导火索是有一次磁盘告警没触发结果某台测试机日志文件把根分区写满了数据库起不来。事后追查发现是监控平台的采集频率太低加上我配置的规则有问题。我当时的第一个念头是“为什么不自己写一个更贴合我的工具”。这个念头一旦冒出来就再也按不住了。1.2 为什么选择SOLO而非直接采购或找人协助有人会问既然公司有平台为什么不请开发部门帮忙写或者直接用商业软件我当时的判断有几个第一这个需求太小了。小到不值得走一遍需求评审、排期、开发、测试的漫长流程。很多运维工具本质上就是“今天早上我要跑的几条命令”自己一个人写从想法到落地可能一个晚上就够了。第二独立开发意味着完全可控。我不需要等别人改代码不需要跟其他部门协调接口工具的所有行为都长在我自己的经验上。我一个人既当产品经理又当开发这种SOLO模式在开发效率上其实非常高前提是项目边界足够小。第三写工具本身也是学习过程。一开始我连并发SSH都觉得麻烦写完之后Paramiko、并发线程、YAML配置、Webhook通知这些技能就都成了自己的东西。以后遇到更大的需求我至少知道该从哪里下手。1.3 小工具的设计边界克制是第一位写运维工具最容易犯的毛病就是越写越大。我一开始甚至想过给工具加上资产管理、内网探测、历史趋势图后来全部砍掉了。原因很简单一个人维护的工具功能越多出故障的概率越大最后反而成了新的维护负担。所以我的最终边界非常收敛这个工具只做三件事。批量采集目标机器的核心指标包括CPU负载、内存使用率、磁盘空间、关键端口或进程状态。对照我预设的阈值做异常判定的规则一旦某个指标超限就记下来。把结果整理成报告异常项直接通过机器人Webhook推到聊天群里。至于历史数据存储、可视化大屏、权限管理这些统统不管。这个决定帮我节省了大量时间也让我能在一个周末就把第一版跑通。2. 技术选型与整体方案设计2.1 从Shell到Python我为什么最终选择了Python最开始我的本能是用Shell脚本写因为运维服务器上都有Shell不用装环境。但仔细想了下还是放弃了原因很现实Shell处理服务器列表、循环、并发、异常捕获都相对吃力。几十台机器串行跑完早高峰都过去了。JSON或YAML解析在Shell里并非不能做但代码会变得很绕可读性差维护起来费劲。Python生态里有现成的库用PyYAML读取配置用Paramiko做SSH连接用concurrent.futures做并发采集。每个模块都有清晰的边界出了问题容易定位。当然Python的代价是需要运行环境有Python 3.6以上并且要pip安装依赖。对于我自己维护的机器这完全不是问题。如果你面对的是一批不能随便装环境的服务器那可能又要回到Shell方案。选型没有绝对好只看自己的约束条件。2.2 整体模块划分把工具拆成四个小零件我把整个工具拆成了四个独立模块配置加载、采集执行、结果判定、通知输出。之所以拆得这么干脆是为了以后升级时能单点替换。比如今天通知走聊天机器人明天如果想改成邮件只需要改“通知输出”那一个模块其他代码不用动。模块之间通过统一的“主机对象”和“检查结果对象”传递数据。主机对象包含主机名、IP、SSH端口、登录用户、采集项目检查结果对象则包含每个指标的实际值和判定状态。这样设计能保证每个函数只做一件事也有利于测试。2.3 配置文件与参数设计一个人写工具也要有“可配置”的意识否则每次换机器就要改代码迟早改出事故。我用了YAML格式做配置文件结构大概是这样# config.yaml interval_seconds: 3600 connect_timeout: 10 max_workers: 5 thresholds: cpu_load_warn: 4.0 memory_usage_warn: 85 disk_usage_warn: 85 port_down_warn: true hosts: - name: web-01 ip: 192.168.1.11 ssh_port: 22 user: ops key_file: ~/.ssh/id_rsa checks: - cpu_load - memory_usage - disk_usage - port port: 8080 - name: db-01 ip: 192.168.1.12 ssh_port: 22 user: ops key_file: ~/.ssh/id_rsa checks: - cpu_load - memory_usage - disk_usage你可能注意到了我把监控机器信息、登录方式、检查项都放进了配置里。运行的时候工具只要读这个文件就能知道自己该去查什么、阈值是什么。这样我换一台机器只改配置文件不碰代码。我更建议你把密钥路径参数化不要直接写死。目录结构也许就是monit-tool/ main.py config.yaml requirements.txt README.md modules/ __init__.py loader.py collector.py judge.py notifier.py3. 核心代码实现与关键步骤3.1 从配置读取到生成主机对象配置读取的逻辑说实话很朴素但有几个细节值得注意。文件路径要和main.py所在目录联动不要用相对当前工作目录的写法否则你在别的目录下执行脚本就会报找不到文件。import os import yaml from dataclasses import dataclass BASE_DIR os.path.dirname(os.path.abspath(__file__)) CONFIG_PATH os.path.join(BASE_DIR, config.yaml) dataclass class Host: name: str ip: str ssh_port: int user: str key_file: str checks: list port: int 0 def load_config(): with open(CONFIG_PATH, r, encodingutf-8) as f: data yaml.safe_load(f) hosts [] for item in data.get(hosts, []): hosts.append(Host( nameitem[name], ipitem[ip], ssh_portitem.get(ssh_port, 22), useritem[user], key_fileitem.get(key_file, ~/.ssh/id_rsa), checksitem[checks], portitem.get(port, 0), )) return data, hostsdataclass是我比较喜欢的方式它比直接用字典更清晰。读取主机配置之后生成Host对象后面的采集函数只需要接收这个对象不需要关心YAML结构逻辑自然就顺了。3.2 并发SSH采集又快又能控住资源第一次写的时候我用最笨的办法一个for循环挨个登录服务器。结果20台机器跑完花了将近4分钟很多时间都浪费在等待网络连接上。后来换成ThreadPoolExecutor并发采集耗时直接降到30秒左右效果非常明显。from concurrent.futures import ThreadPoolExecutor, as_completed import paramiko import sys def collect_on_host(host): result {host: host.name, ip: host.ip, ok: False, metrics: {}} try: client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) key_file os.path.expanduser(host.key_file) client.connect( hostnamehost.ip, porthost.ssh_port, usernamehost.user, key_filenamekey_file, timeout10, banner_timeout10, auth_timeout10, ) result[metrics][cpu_load] get_cpu_load(client) result[metrics][memory_usage] get_memory_usage(client) result[metrics][disk_usage] get_disk_usage(client) if port in host.checks: result[metrics][port_ok] check_port(client, host.port) client.close() result[ok] True except Exception as e: result[error] str(e) return result def collect_all(hosts, max_workers5): results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: future_map {executor.submit(collect_on_host, host): host for host in hosts} for future in as_completed(future_map): result future.result() results.append(result) print(f完成 {result[host]}: ok{result[ok]}) return results这里有一个关键点max_workers不能盲目调到很大。我试过开20个线程去连20台服务器结果有一半连接因为目标机sshd并发限制而失败。后来把并发数控制在5到8反而是最稳的。线程不是越多越好它是一个典型的“木桶效应”问题目标服务器的处理能力才是瓶颈。3.3 告警判定与结果汇总阈值不要拍脑袋对于采集回来的指标判定逻辑我最开始是直接在函数里写死if cpu_load 4.0: warn(cpu load too high)后来发现这种写死方式太脆了不同机器、不同业务对阈值的容忍度完全不一样。比如数据库服务器可能CPU超过2就该报警而编译服务器跑到8也只是正常。所以我全部改成从配置读阈值。def judge_results(results, thresholds): summary {total: len(results), success: 0, failed: 0, warnings: []} for item in results: if not item[ok]: summary[failed] 1 summary[warnings].append({ host: item[host], type: connect_failed, detail: item.get(error, ), }) continue summary[success] 1 metrics item[metrics] if metrics.get(cpu_load, 0) thresholds[cpu_load_warn]: summary[warnings].append({ host: item[host], type: cpu_load, detail: fload{metrics[cpu_load]}, threshold: thresholds[cpu_load_warn], }) if metrics.get(memory_usage, 0) thresholds[memory_usage_warn]: summary[warnings].append({ host: item[host], type: memory_usage, detail: fusage{metrics[memory_usage]}%, threshold: thresholds[memory_usage_warn], }) if metrics.get(disk_usage, 0) thresholds[disk_usage_warn]: summary[warnings].append({ host: item[host], type: disk_usage, detail: fusage{metrics[disk_usage]}%, threshold: thresholds[disk_usage_warn], }) if port_ok in metrics and not metrics[port_ok]: summary[warnings].append({ host: item[host], type: port_down, detail: fport{item.get(port, unknown)} not reachable, }) return summary阈值我后来都没有拍脑袋而是先跑了一周“只记录不告警”的观察模式把每台服务器的正常波动范围摸出来再按“正常峰值的1.5倍”和“可用性经验”设定。这样能少很多噪声告警尤其是CPU负载在低峰和高峰完全不是一个数量级。3.4 报告输出与定时调度让工具自己跑起来采集和判定结果我最后会格式化成一封简明报告。当时我选的是最简单的文本格式用来推到聊天机器人的Webhook效果已经足够。以下是通知函数的大致写法import requests def send_webhook(url, content): payload {msgtype: text, text: {content: content}} resp requests.post(url, jsonpayload, timeout5) if resp.status_code ! 200: raise RuntimeError(fwebhook send failed: {resp.status_code}) def build_report(summary): lines [] lines.append(f巡检结果总计 {summary[total]} 台成功 {summary[success]} 台异常 {len(summary[warnings])} 项) for w in summary[warnings]: lines.append(f[{w[type]}] {w[host]}: {w[detail]}阈值 {w.get(threshold, N/A)}) return \n.join(lines) if lines else 巡检结果全部正常定时调度我用的是最简单的crontab没有引入额外工具。每天凌晨4点和下午4点各跑一次避开业务高峰同时保证到早上能看到前一夜的情况。0 4 * * * cd /opt/monit-tool /usr/bin/python3 main.py /var/log/monit-tool.log 21 0 16 * * * cd /opt/monit-tool /usr/bin/python3 main.py /var/log/monit-tool.log 21如果你管理的机器systemd更顺手也可以写一个timer单元效果差不多。我之所以crontab优先纯粹是因为我手头所有机器都有cron不依赖额外的服务启动方式。4. 测试、上线与踩坑实录4.1 我踩过的几个坑一张速查表独立开发最大的风险不是代码写不出来而是没人帮你测试所有问题都要自己正面刚。我把这几次踩过的坑整理成了一张表希望你能绕开现象可能原因解决办法SSH连接报“Host key not found”目标主机未加入known_hosts使用AutoAddPolicy()生产环境建议改为load_host_keys并预先录入连接极慢每次等待几十秒服务器配置了反向DNS解析SSH客户端连接参数中设置look_for_keysFalse并让服务端关闭UseDNS采集过程中部分主机连接超时并发线程数过高超过sshd限制调低max_workers增加超时参数connect、banner、auth中文乱码或编码异常远程命令输出包含非UTF8字符远程命令前加LANGen_US.UTF-8本地用decode(utf-8, errorsignore)执行命令后没有输出或通道卡住未正确读取stdout或stderr关闭时机不对统一用exec_command后先读stdout再读stderr最后等待channel exit定时任务跑了但通知没收到脚本执行异常被cron吞掉没有日志cron命令中保留 logfile 21程序内设置全局异常捕获磁盘使用率误报把临时挂载点也算进去了比如/proc和/dev采集时过滤掉虚拟文件系统和只读挂载点4.2 异常处理与重试机制别让一次失败毁掉整个任务用Python写多主机采集最怕遇到“一台机器连接失败导致整个程序崩溃”。我一开始没有做任何异常保护某个内网机器短时不可达工具直接traceback退出后面的机器全都没采集。这种体验让我养成了两个习惯。第一个习惯是“每台主机单独隔离”。collect_on_host函数内部把所有可能出错的环节包括连接、执行命令、解析输出、关闭连接都包在try-except里。函数无论如何都会返回一个带okFalse和error字段的结果不会向上抛异常。这样即使一台机器挂了也只是那台机器标记为失败其它机器照常采集。第二个习惯是“连接失败重试一次”。网络抖动是常态直接判定失败容易造成误报。我后来在采集主流程里加了最多重试1次的逻辑只有重试仍然失败才标记为异常。重试时有2秒的休眠避免立刻去撞还在恢复中的服务。def collect_with_retry(host, retry_times1): for attempt in range(retry_times 1): result collect_on_host(host) if result[ok]: return result if attempt retry_times: time.sleep(2) return result4.3 日志与调试技巧没有打印就没有排查能力一个人写工具调试基本靠日志。我第一版用了大量print后来发现信息密度太低。一台机器成功、一台机器失败混在一起根本分不清谁是谁。于是我把所有输出统一改造为标准库logging并加了一个--debug参数。python3 main.py --debug在--debug模式下日志会详细输出每一条SSH命令、每一个返回码、每一次解析的原始数据。正常运行时只输出WARNING及以上级别免得每天刷屏。import logging def setup_logger(debugFalse): level logging.DEBUG if debug else logging.WARNING logging.basicConfig( levellevel, format%(asctime)s [%(levelname)s] %(name)s: %(message)s, datefmt%Y-%m-%d %H:%M:%S, )调试时还有一个技巧单独建一个hosts_debug.yaml只放一台测试机这样跑起来速度快输出干净能定位问题在哪里。如果直接拿全量机器来debug几十台主机的结果混在一起眼睛会看瞎。4.4 上线策略先观察再告警这个工具上线时我没有直接开告警。第一个星期运行的是“静默模式”所有异常只打印到日志和本地文件不推送到聊天群。目的很简单先看工具会不会误报会不会因为某些采集逻辑不完善而遗漏真正的问题。结果这一周还真发现了两个问题。一是某台机器的SSH握手太慢老是超时二是某些机器存在多个磁盘挂载点我只取了根分区的使用率导致数据盘满了也没察觉。如果一上来就开告警群里肯定天天响最后大家就不信这个工具了。运维工具最怕的就是狼来了。5. 单人维护的几条心得5.1 代码要写给明天的自己看写这个工具的时候我给自己定了一条规矩任何代码都要像写给一个陌生人看一样该写注释的地方必须写。尤其是那些看起来“很绕”的逻辑比如为什么要加banner_timeout10为什么要过滤/proc挂载点这些在当时很容易懂但隔两周再看可能就忘得一干二净。我在关键函数顶部都补上了Docstring说明这个函数的输入输出和副作用。这不是为了形式好看而是为了降低“下次改动时把系统改坏”的概率。长一点的函数我会拆成多个短函数一个函数只干一件事。之前见过很多人的脚本写得跟一团线一样改一个变量牵连全身这种代码在单人项目里最致命。5.2 版本管理要趁早哪怕只有一个人很多人觉得“我一个人的脚本git可有可无”。我的经验是等改到第三版你就知道版本管理有多必要。我用Git管理这个工具每完成一个功能点就commit一次提交信息写清楚“加了什么、为什么加”。回到那段改配置格式然后导致工具崩溃的日子我只需要git revert就能恢复而不是靠脑补之前哪行代码是对的。分支策略也很简单main分支是稳定版本dev分支专门放实验性改动。我一般先在dev上跑两天确认没问题才合并到main。这样即使某个改动出了问题也不会影响定时任务实际运行的稳定版本。5.3 工具未来的扩展方向加什么为什么不加这个巡检工具目前已经稳定运行了三个多月。我心中的扩展方向其实是有的但优先级不同。支持多阈值模板比如给数据库机器用更严格的内存阈值给测试机器用宽松阈值。这个非常实用已经列入计划。把历史结果存入SQLite方便做简单的趋势对比。但SQLite文件需要定期清理和维护我并不急着做。增加采集项比如磁盘inode、特定日志文件error统计。不过每个采集项都会增加SSH命令执行时间所以要进一步考虑执行效率。我不会再往这个工具里加Web UI、权限系统或者复杂的仪表盘。因为那又回到了“重平台”的老路会让我重新陷入维护泥潭。SOLO实战最重要的能力不是写多少代码而是知道哪些代码不值得写。很多工具看起来功能丰富实际上百分之八十的功能都是摆设。我一个人维护必须把精力放在最稳定、最有价值的那百分之二十上。如果你也正打算自己动手写一个运维小工具我的建议是从最让你难受的重复操作开始。别急着追求大而全先写一个能用起来的东西再慢慢迭代。这个过程里学到的东西有时候比工具本身更有价值。