摘要：在日常的 Linux 运维工作中，我们往往面临着“救火式”的困境：磁盘满了才去清理，服务挂了才去重启。虽然市面上有 Zabbix、Prometheus 等专业监控工具，但对于轻量级需求或特定业务场景，编写高效的 Shell 脚本依然是最直接的手段。
然而，单纯的 Shell 脚本存在逻辑处理能力弱、报警风暴难以抑制、缺乏统一视图等痛点。本文将带你跳出基础脚本的舒适区，通过优化 Shell 脚本并结合 Python 进行二次封装，打造一个集“监控、管理、控制”于一体的轻量级自动化运维平台。
详细内容请查阅下文。
一、基石构建：Shell 脚本的进阶优化
在迈向平台化之前，需要先夯实基础。原有的脚本虽然能用，但在生产环境中显得过于简陋。我们需要引入更严谨的逻辑判断和更通用的配置管理。1. 邮件报警服务的标准化配置
报警是监控的眼睛。为了避免在每个脚本中重复配置 SMTP 信息，建议将邮件发送功能封装为一个独立的函数或公共脚本 send_alert.sh。# send_alert.shMAIL_TO="17587257@qq.com"SMTP_SERVER="smtp.163.com"FROM_ADDR="tjf020@163.com"send_alert() {localsubject=$1localmessage=$2#使用 mailx 发送，生产环境建议配置 s-nail 或 sendmail 以提高成功率echo"$message" | mail -s "$subject" "$MAIL_TO"}
2. 系统负载监控：从“数值”到“趋势”
原有的脚本仅判断当前负载是否超过阈值，容易产生误报。进阶版应结合 CPU 核心数动态计算阈值，并增加“持续时间”判断（例如：连续 3 次检查均超标才报警）。精准提取：使用 awk 和 tr 处理 uptime 输出，去除逗号干扰。# check_load_pro.shCORES=$(grep -c 'model name' /proc/cpuinfo)THRESHOLD=$(echo "$CORES * 1.5" | bc)LOAD=$(uptime | awk -F'load average:' '{print $2}' | awk '{print $1}' | tr -d ',')使用 bc 进行浮点数比较IS_HIGH=$(echo "$LOAD > $THRESHOLD" | bc)if [ "$IS_HIGH" -eq 1 ]; thenMSG="严重：服务器负载过高！当前:$LOAD (阈值: $THRESHOLD)"#调用报警函数source./send_alert.sh && send_alert "【严重】系统负载报警" "$MSG"fi
3. 服务状态监控：引入“自愈”机制
监控的目的不仅是发现问题，更是为了解决问题。对于关键服务（如 Nginx、MySQL），脚本应具备“看门狗”功能：发现服务停止 -> 尝试自动重启 -> 重启失败再报警。# check_service_pro.shSERVICES=("nginx" "mysqld")for SERVICE in "${SERVICES[@]}"; doif! systemctl is-active --quiet "$SERVICE"; thenecho"服务 $SERVICE 异常，正在尝试重启..."systemctlrestart "$SERVICE"sleep3#二次检查if! systemctl is-active --quiet "$SERVICE"; thensend_alert"【紧急】服务自愈失败" "服务 $SERVICE 重启失败，请立即人工介入！"fifidone
4. 磁盘与日志：精细化治理
磁盘监控：使用 df -P 确保跨平台兼容性，并过滤掉 tmpfs 等虚拟文件系统，避免误报。日志清理：find 命令是清理利器，但 rm 操作极其危险。进阶版脚本应先记录要删除的文件列表到日志中，确认无误后再执行删除，防止误删业务数据。二、核心升级：Python 赋能，打造统一监控平台
Shell 脚本适合处理单一任务，但当我们需要汇总数据、生成报表或进行复杂的逻辑判断时，Python 的优势就体现出来了。我们将编写一个 Python 主程序 monitor_center.py，通过调用上述 Shell 脚本或直接使用 Python 库来获取系统信息。1. 架构设计
数据采集层：使用 Python 的 psutil 库直接获取系统底层数据，比调用 Shell 命令更高效、更跨平台。- 展示与通知层：生成 HTML 格式的日报，或推送到企业微信。
2. Python 监控核心代码实现
我们需要安装 psutil 库：pip install psutil。# monitor_center.pyimport psutilimport socketimport smtplibfrom email.mime.text import MIMETextfrom datetime import datetimeclass SystemMonitor:def__init__(self):self.hostname= socket.gethostname()self.alerts= []defcheck_cpu_load(self):#获取 1 分钟负载load_1,_, _ = psutil.getloadavg()cpu_count= psutil.cpu_count()#计算负载百分比load_percent= (load_1 / cpu_count) * 100ifload_percent > 150:阈值 150%self.alerts.append(f"CPU负载过高: {load_1} (核心数: {cpu_count})")defcheck_disk_usage(self):partitions= psutil.disk_partitions()forpartition in partitions:try:usage= psutil.disk_usage(partition.mountpoint)percent= usage.percentifpercent > 85:self.alerts.append(f"磁盘空间不足:{partition.mountpoint} 使用率 {percent:.1f}%")exceptPermissionError:continuedefcheck_memory(self):mem= psutil.virtual_memory()ifmem.percent > 90:self.alerts.append(f"内存使用率过高:{mem.percent}%")defsend_report(self):#这里可以集成发送邮件、钉钉 webhook 等逻辑status= "异常" if self.alerts else "正常"content= f"服务器巡检报告\n\n-主机名: {self.hostname}\n-时间: {datetime.now()}\n-状态: {status}\n\n"ifself.alerts:content+= "报警详情：\n" + "\n".join(self.alerts)else:content+= "所有指标均正常。"print(content)实际生产中替换为邮件发送函数if __name__ == "__main__":monitor= SystemMonitor()monitor.check_cpu_load()monitor.check_disk_usage()monitor.check_memory()monitor.send_report()
三、架构演进：基于 Flask 构建集中式监控平台
单机脚本的局限性在于“信息孤岛”。当服务器数量增长到几十甚至上百台时，逐台登录查看日志或执行脚本变得不切实际。我们需要一个中心化的“驾驶舱”。1. 架构设计：Agent + Server 模式
2. 开发 Agent：让服务器“开口说话”
在被监控服务器上安装 Flask：pip install flask psutil。from flask import Flask, jsonify, requestimport psutilimport socketimport subprocessimport datetimeimport logging配置日志，方便排查问题logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')logger = logging.getLogger(__name__)app = Flask(__name__)【安全】设置一个简单的 API Key，防止任意重启服务生产环境建议从配置文件读取或使用环境变量API_KEY = "your-secret-key-here"def get_local_ip():"""【优化】获取真实的局域网IP原方法socket.gethostbyname 经常返回 127.0.0.1"""try:#创建一个 UDP socket 连接到外部地址（不需要真的连通），以此获取本机出口 IPs= socket.socket(socket.AF_INET, socket.SOCK_DGRAM)s.connect(("8.8.8.8",80))ip= s.getsockname()[0]s.close()returnipexceptException:#如果网络不通，回退到 hostname 解析returnsocket.gethostbyname(socket.gethostname())def get_system_info():"""采集本机系统信息"""try:#【健壮性】兼容处理 load average，防止旧系统报错try:cpu_load= psutil.getloadavg()load_1= round(cpu_load[0], 2)except(AttributeError, OSError):load_1= 0.0memory= psutil.virtual_memory()disk= psutil.disk_usage('/')return{"hostname":socket.gethostname(),"ip":get_local_ip(),【修复】使用新的 IP 获取方法"cpu_load_1":load_1,"memory_percent":memory.percent,"disk_percent":disk.percent,"timestamp":datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}exceptException as e:logger.error(f"采集系统信息失败:{e}")return{"error": "采集失败"}@app.route('/api/status', methods=['GET'])def api_status():"""提供系统状态数据接口"""returnjsonify(get_system_info())@app.route('/api/restart_service/', methods=['POST'])def api_restart_service(service_name):"""提供远程重启服务的接口"""#【安全】简单的 Header 鉴权auth_key= request.headers.get('X-API-Key')ifauth_key != API_KEY:logger.warning(f"未授权的重启尝试:{request.remote_addr}")returnjsonify({"status": "error", "msg": "未授权"}), 403#【安全】简单过滤非法字符，防止命令注入ifnot service_name.isalnum() and '_' not in service_name and '-' not in service_name:returnjsonify({"status": "error", "msg": "服务名包含非法字符"}), 400try:logger.info(f"正在重启服务:{service_name}")#增加超时时间，防止命令卡死导致 Agent 阻塞result= subprocess.run(["sudo","systemctl", "restart", service_name],check=True,timeout=10,capture_output=True,text=True)returnjsonify({"status": "success", "msg": f"服务 {service_name} 重启成功"})exceptsubprocess.CalledProcessError as e:logger.error(f"重启服务失败:{e.stderr}")returnjsonify({"status": "error", "msg": f"重启失败: {e.stderr}"}), 500exceptsubprocess.TimeoutExpired:returnjsonify({"status": "error", "msg": "重启超时"}), 504if __name__ == '__main__':#【优化】生产环境建议关闭 debug，使用 gunicorn 部署更佳#这里保持原生运行，但关闭 debug 模式以防泄露源码app.run(host='0.0.0.0', port=5000, debug=False)
备注：执行指令#nohup python3 /opt/monitor/agent.py > nohup.out 2>&1 & &启动Agent。现在这台服务器就拥有了一个 http://:5000/api/status 的接口，任何人都可以通过 HTTP 请求获取其实时状态。3. 开发 Server：打造统一监控仪表盘
在监控中心服务器上，创建 dashboard.py。它负责收集所有 Agent 的数据。from flask import Flask, render_template_stringimport requestsimport threadingimport timeapp = Flask(__name__)被监控服务器列表TARGET_SERVERS = ["http://172.16.24.252:5000/api/status","http://172.16.24.253:5000/api/status",#可以继续添加更多服务器]全局变量存储监控数据MONITOR_DATA = []def fetch_data():"""后台线程：定时从所有Agent 拉取数据"""whileTrue:temp_data= []forurl in TARGET_SERVERS:try:#设置超时时间，防止某个节点挂掉导致整个程序卡死resp= requests.get(url, timeout=5)ifresp.status_code == 200:temp_data.append(resp.json())exceptException as e:#节点不可达，记录错误信息hostname= url.split('/')[2].split(':')[0]temp_data.append({"hostname":hostname, "error": "节点离线或无法访问"})globalMONITOR_DATAMONITOR_DATA= temp_datatime.sleep(10)每 10 秒更新一次简单的 HTML 模板，用于展示仪表盘HTML_TEMPLATE = """服务器集群监控仪表盘{%for host in hosts %}5.0 else 'normal' }}">{{host.get('cpu_load_1', '-') }}90 else 'normal' }}">{{host.get('memory_percent', '-') }}%90 else 'normal' }}">{{host.get('disk_percent', '-') }}%{%endfor %}数据每 10 秒自动刷新"""@app.route('/')def index():returnrender_template_string(HTML_TEMPLATE, hosts=MONITOR_DATA)if __name__ == '__main__':thread= threading.Thread(target=fetch_data, daemon=True)thread.start()#添加 ssl_context 参数app.run(host='0.0.0.0',port=8080, debug=False, ssl_context=('cert.pem', 'key.pem'))
4. 实战体验
现在，只需在浏览器访问监控中心服务器的 https://<监控中心IP>:8080，你就能看到所有被监控服务器的实时状态。- 一目了然：所有服务器的负载、内存、磁盘状态尽收眼底。
- 故障定位：哪台服务器离线、哪个指标超标，通过颜色高亮（红/绿）瞬间识别。
- 扩展性：新增服务器？只需在 TARGET_SERVERS 列表中添加其 Agent 地址即可，无需登录新服务器部署复杂环境。
5. 注意事项
集中管理的关键是agent的配置，本项目是将agent.py程序存到被监控的服务器/opt/monitor/目录中，然后运行命令nohup python3 /opt/monitor/agent.py > nohup.out 2>&1 & 启动agent代理程序。在此之前需要为被监控的服务器配置一下Python环境，
第二步：执行 pip3 install flask安装flask框架第三步：执行 yum install -y gcc python3-devel安装gcc，为下一步安装psutil做好准备，第四步：执行 pip3 install psutil==5.9.0安装psutil库。四、经验总结
1.脚本化：用 Shell 解决了单机的重复劳动。2.平台化：用 Python 封装了复杂的逻辑处理。3.集中化：用 Flask 构建了可视化的集群管理视图。运维之路，始于脚本，终于架构。希望这套从 Shell 到 Flask 的演进方案，能助你构建出更稳定、更智能的运维体系。📢总结与分享
总而言之，通过不断练习这些命令，你将能体会到 Vim 带来的高效与便捷，真正成为 Linux 环境下的高手。
如果本文对您有帮助，欢迎：
3.🔄转发，在Linux系统中设计打造你自己的自动化监控脚本提高团队的工作效率。
10个月宝宝每天需要喝多少奶粉？
