Python+Flask构建轻量级服务器监控系统

1. 项目背景与核心需求最近在管理公司600多台服务器时经常遇到需要快速确认哪些服务器离线的情况。传统方式是通过SSH逐个连接或手动ping测试效率极低且容易遗漏。于是决定用Python开发一个轻量级的Web监控系统实现以下核心功能自动批量检测服务器在线状态实时更新数据库中的主机状态通过Web界面直观展示监控结果支持按服务器类型分类查看这个方案特别适合中小型运维团队无需部署复杂的监控系统用最简单的技术栈就能实现核心需求。我选择了PythonFlaskMySQL的技术组合全部代码不到200行却能显著提升运维效率。2. 技术方案设计2.1 整体架构设计系统采用三层架构数据采集层使用fping进行批量ping检测数据处理层Python脚本更新数据库状态展示层Flask构建的Web界面graph TD A[fping批量检测] -- B[结果写入文件] B -- C[Python解析结果] C -- D[更新MySQL状态] D -- E[Flask展示数据]2.2 关键技术选型fping替代传统ping原生ping命令逐个检测效率太低fping支持批量检测600个IP只需2-3秒安装简单yum install fping -y数据库设计CREATE TABLE ip_table( id INT AUTO_INCREMENT, ipaddress CHAR(15) NOT NULL, application CHAR(24), status CHAR(6), PRIMARY KEY(id) );Web框架选择放弃Django选择更轻量的Flask配合Bootstrap快速构建UI所需依赖仅三个包pip install flask flask_bootstrap pymysql3. 核心实现细节3.1 批量检测实现fping脚本(fping.sh)#!/bin/bash rm -f result.txt cat ipmi_ping.txt | fping result.txtPython调用逻辑import os import pymysql def update_status(): # 从数据库读取IP列表 conn pymysql.connect(hostlocalhost, userroot, password123456, dbcheckip) cursor conn.cursor() cursor.execute(SELECT ipaddress FROM ip_table) # 写入临时文件供fping使用 with open(ipmi_ping.txt, w) as f: for ip in cursor.fetchall(): f.write(ip[0] \n) # 执行检测 os.system(sh fping.sh) # 解析结果 with open(result.txt) as f: for line in f: ip line[:line.index(is)-1] status 离线 if unreachable in line else 在线 cursor.execute(fUPDATE ip_table SET status{status} WHERE ipaddress{ip}) conn.commit() conn.close()3.2 Web界面开发Flask主程序(app.py)from flask import Flask, render_template import pymysql app Flask(__name__) app.route(/) def index(): conn pymysql.connect(hostlocalhost, userroot, password123456, dbcheckip) cur conn.cursor() cur.execute(SELECT * FROM ip_table) hosts cur.fetchall() conn.close() return render_template(index.html, hostshosts) if __name__ __main__: app.run(host0.0.0.0, port5000)前端模板(index.html){% extends bootstrap/base.html %} {% block content %} div classcontainer table classtable table-striped thead tr thID/th thIP地址/th th服务类型/th th状态/th /tr /thead tbody {% for host in hosts %} tr class{{ success if host[3]在线 else danger }} td{{ host[0] }}/td td{{ host[1] }}/td td{{ host[2] }}/td td{{ host[3] }}/td /tr {% endfor %} /tbody /table /div {% endblock %}4. 部署与优化4.1 后台服务部署使用nohup让两个核心进程常驻运行# 检测服务 nohup python ping_monitor.py ping.log 21 # Web服务 nohup python app.py web.log 21 4.2 性能优化技巧检测频率控制import time while True: update_status() time.sleep(300) # 5分钟检测一次数据库连接池from DBUtils.PooledDB import PooledDB pool PooledDB(pymysql, 5, hostlocalhost, userroot, password123456, dbcheckip)结果缓存from flask_caching import Cache cache Cache(app, config{CACHE_TYPE: simple}) app.route(/) cache.cached(timeout60) def index(): # ...5. 常见问题解决方案5.1 fping检测超时问题现象部分IP检测时间过长解决添加超时参数fping -t 500 ipmi_ping.txt # 500ms超时5.2 数据库连接失败错误pymysql.err.OperationalError: (2003, Cant connect to MySQL server)排查检查MySQL服务状态验证用户名密码确认网络连通性5.3 Web界面加载慢优化方案添加分页功能from flask_sqlalchemy import Pagination app.route(/) def index(): page request.args.get(page, 1, typeint) per_page 50 # 分页查询逻辑启用Gzip压缩from flask_compress import Compress Compress(app)6. 扩展功能建议告警通知import smtplib def send_alert(ip): server smtplib.SMTP(smtp.example.com) server.sendmail(alertexample.com, adminexample.com, f主机 {ip} 离线)历史记录CREATE TABLE status_history ( id INT AUTO_INCREMENT, ipaddress CHAR(15), status CHAR(6), check_time DATETIME, PRIMARY KEY(id) );API接口app.route(/api/status/ip) def get_status(ip): conn pool.connection() cursor conn.cursor() cursor.execute(SELECT status FROM ip_table WHERE ipaddress%s, (ip,)) result cursor.fetchone() conn.close() return jsonify({status: result[0] if result else unknown})这个项目虽然简单但涵盖了Python开发中的多个实用技能点系统监控、数据库操作、Web开发等。我在实际部署后发现对于500台以下规模的主机监控这个方案完全能满足需求且资源占用极低内存100MB。