Linux 服务器巡检不再难，这个 Python 脚本全搞定

在企业级生产环境中，服务器的稳定性和可靠性至关重要。定期对服务器进行巡检，及时发现并解决潜在问题，是保障业务正常运行的关键。本文将介绍如何使用 Python3 编写一个功能完善的 Linux 服务器自动巡检脚本，帮助运维工程师实现自动化监控。

一、脚本功能设计

我们的巡检脚本将包含以下核心功能：

1. 检查项目

• CPU 使用率：监控系统 CPU 负载情况
• 内存使用率：检查内存使用情况，避免内存不足
• 磁盘使用率：监控磁盘空间，防止磁盘满导致服务异常
• 进程状态：检查关键进程是否正常运行
• 网络端口：验证服务端口是否正常开放
• 系统启动时间：了解服务器运行时长和启动时间
• 僵尸进程数量：检测系统中是否存在僵尸进程
• 网络状态：监控网络接口流量、连接数和网络延迟
• 安全状态：检查登录失败次数、最近登录记录和关键文件权限
• 自动生成巡检报告：将检查结果整理成格式化报告
• 发送告警通知：通过钉钉或企业微信发送异常告警

二、脚本实现

1. 环境准备

确保服务器已安装 Python 3.6+ 及以下依赖：

安装依赖包

对于 CentOS/RHEL/Rocky 系统：

## yum 或者 dnf 安装
yum install python-pip
pip install psutil requests -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple

对于 Ubuntu/Debian 系统：

apt-get install python-pip
pip install psutil requests -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple

使用清华镜像源可以提高安装速度，避免网络问题导致的安装失败。

三、脚本使用指南

1. 配置说明

• 进程检查：在脚本开头的 CHECK_PROCESSES 列表中添加需要监控的进程名称
• 端口检查：在脚本开头的 CHECK_PORTS 列表中添加需要监控的端口号
• 告警配置：在脚本开头的 ALERT_CONFIG 字典中配置
• type：选择告警类型，支持 dingtalk（钉钉）或 wecom（企业微信）
• webhook：填写对应的 webhook 地址

2. 运行方式

# 直接运行
python server_monitor.py

# 添加到定时任务（每小时执行一次）
crontab -e
0 * * * * python /path/to/server_monitor.py

3. 告警设置

钉钉告警设置

1. 打开钉钉群，点击「群设置」→「智能群助手」→「添加机器人」
2. 选择「自定义」机器人，填写名称并获取 webhook 地址
3. 将 webhook 地址填入脚本的 alert_config 中

企业微信告警设置

1. 登录企业微信后台，进入「应用管理」→「自建」→「创建应用」
2. 获取应用的 AgentId、CorpId 和 Secret
3. 使用企业微信机器人 webhook 或 API 发送消息

四、生优化建议

1. 权限管理：脚本应使用非 root 用户运行，避免权限过高带来的安全风险
2. 日志记录：添加详细的日志记录，便于问题排查
3. 错误处理：增强脚本的错误处理能力，确保脚本稳定运行
4. 监控频率：根据服务器重要性调整巡检频率，核心服务器可设置更频繁的检查
5. 阈值调整：根据服务器配置和业务需求，调整各项指标的告警阈值
6. 报告归档：定期归档巡检报告，便于历史数据查询和分析

完整脚本代码

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Linux 服务器自动巡检脚本
功能：检查 CPU、内存、磁盘、进程、网络端口状态，并生成报告和发送告警
"""

import os
import time
import psutil
import socket
import json
import requests
from datetime import datetime

# 配置参数
# 告警配置
ALERT_CONFIG = {
'type': 'dingtalk',  # 可选: 'dingtalk' 或 'wecom'
'webhook': 'https://oapi.dingtalk.com/robot/send?access_token=your_token'
}

# 要检查的进程和端口
CHECK_PROCESSES = ['sshd', 'nginx', 'mysql', 'redis']
CHECK_PORTS = [22, 80, 443, 3306, 6379]

classServerMonitor:
def__init__(self):
        self.hostname = socket.gethostname()
        self.ip = self.get_local_ip()
        self.results = {}

defget_local_ip(self):
"""获取本地 IP 地址"""
try:
            s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            s.connect(('8.8.8.8', 80))
            ip = s.getsockname()[0]
            s.close()
return ip
except:
return'127.0.0.1'

defcheck_cpu(self):
"""检查 CPU 使用率"""
        cpu_percent = psutil.cpu_percent(interval=1)
        status = "正常"if cpu_percent < 80else"警告"
        self.results['cpu'] = {
'使用率': f"{cpu_percent}%",
'状态': status
        }
return cpu_percent

defcheck_memory(self):
"""检查内存使用率"""
        mem = psutil.virtual_memory()
        mem_percent = mem.percent
        status = "正常"if mem_percent < 80else"警告"
        self.results['内存'] = {
'使用率': f"{mem_percent}%",
'总内存': f"{mem.total / 1024 / 1024 / 1024:.2f}GB",
'可用内存': f"{mem.available / 1024 / 1024 / 1024:.2f}GB",
'状态': status
        }
return mem_percent

defcheck_disk(self):
"""检查磁盘使用率"""
        disk_info = []
for partition in psutil.disk_partitions():
if partition.fstype:
try:
                    usage = psutil.disk_usage(partition.mountpoint)
                    usage_percent = usage.percent
                    status = "正常"if usage_percent < 85else"警告"
                    disk_info.append({
'挂载点': partition.mountpoint,
'文件系统': partition.fstype,
'使用率': f"{usage_percent}%",
'总空间': f"{usage.total / 1024 / 1024 / 1024:.2f}GB",
'可用空间': f"{usage.free / 1024 / 1024 / 1024:.2f}GB",
'状态': status
                    })
except:
pass
        self.results['磁盘'] = disk_info
return disk_info

defcheck_processes(self, process_names):
"""检查进程状态"""
        process_status = []
for name in process_names:
            found = False
for proc in psutil.process_iter(['name']):
try:
if name.lower() in proc.info['name'].lower():
                        found = True
break
except:
pass
            status = "运行中"if found else"未运行"
            process_status.append({
'进程名': name,
'状态': status
            })
        self.results['进程'] = process_status
return process_status

defcheck_ports(self, ports):
"""检查网络端口"""
        port_status = []
for port in ports:
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex((self.ip, port))
            status = "开放"if result == 0else"关闭"
            sock.close()
            port_status.append({
'端口': port,
'状态': status
            })
        self.results['端口'] = port_status
return port_status

defcheck_uptime(self):
"""检查系统启动时间"""
        uptime_seconds = time.time() - psutil.boot_time()
        days = int(uptime_seconds // 86400)
        hours = int((uptime_seconds % 86400) // 3600)
        minutes = int((uptime_seconds % 3600) // 60)
        uptime_str = f"{days}天{hours}小时{minutes}分钟"
        boot_time = datetime.fromtimestamp(psutil.boot_time()).strftime("%Y-%m-%d %H:%M:%S")
        self.results['系统启动时间'] = {
'运行时间': uptime_str,
'启动时间': boot_time
        }
return uptime_str

defcheck_zombie_processes(self):
"""检查僵尸进程数量"""
        zombie_count = 0
for proc in psutil.process_iter(['status']):
try:
if proc.info['status'] == psutil.STATUS_ZOMBIE:
                    zombie_count += 1
except:
pass
        status = "正常"if zombie_count == 0else"警告"
        self.results['僵尸进程'] = {
'数量': zombie_count,
'状态': status
        }
return zombie_count

defcheck_network_status(self):
"""检查网络状态"""
        network_info = []
        net_io = psutil.net_io_counters()
        bytes_sent = net_io.bytes_sent / 1024 / 1024# MB
        bytes_recv = net_io.bytes_recv / 1024 / 1024# MB

# 获取网络接口信息
for interface, addrs in psutil.net_if_addrs().items():
if interface != 'lo':  # 排除本地回环接口
                interface_info = {
'接口': interface,
'发送流量': f"{bytes_sent:.2f} MB",
'接收流量': f"{bytes_recv:.2f} MB"
                }
                network_info.append(interface_info)

# 检查网络连接数
        connections = psutil.net_connections()
        connection_count = len(connections)

# 检查网络延迟
        ping_result = self.ping_test('8.8.8.8')

        self.results['网络状态'] = {
'接口信息': network_info,
'连接数': connection_count,
'网络延迟': f"{ping_result} ms",
'状态': "正常"if ping_result < 100else"警告"
        }
return network_info

defping_test(self, host):
"""测试网络延迟"""
import subprocess
try:
            output = subprocess.check_output(
                ['ping', '-c', '1', host],
                stderr=subprocess.STDOUT,
                universal_newlines=True
            )
# 提取延迟信息
for line in output.split('\n'):
if'time='in line:
return float(line.split('time=')[1].split(' ')[0])
return9999
except:
return9999

defcheck_security(self):
"""检查安全状态"""
# 检查登录失败次数和失败的用户（从日志文件中获取）
        failed_login_count, failed_users = self.get_failed_login_count()

# 检查异常登录（简单实现：检查最近登录记录）
        recent_logins = self.get_recent_logins()

# 检查权限变更（简单实现：检查关键文件权限）
        critical_files = ['/etc/shadow', '/etc/sudoers']  # 移除 /etc/passwd
        permission_issues = []
for file_path in critical_files:
if os.path.exists(file_path):
try:
                    stat = os.stat(file_path)
# 检查权限是否过于宽松
if stat.st_mode & 0o007 != 0:  # 其他用户有写权限
                        permission_issues.append(f"{file_path} 权限过于宽松")
except:
pass

        status = "正常"if failed_login_count < 5and len(permission_issues) == 0else"警告"

        self.results['安全状态'] = {
'登录失败次数': failed_login_count,
'登录失败的用户': failed_users,
'最近登录': recent_logins,
'权限问题': permission_issues,
'状态': status
        }
return status

defget_failed_login_count(self):
"""获取登录失败次数和失败的用户"""
import subprocess
try:
# 检查 /var/log/auth.log 或 /var/log/secure
            log_files = ['/var/log/auth.log', '/var/log/secure']
            failed_count = 0
            failed_users = []

for log_file in log_files:
if os.path.exists(log_file):
try:
                        output = subprocess.check_output(
                            ['grep', 'Failed password', log_file],
                            stderr=subprocess.STDOUT,
                            universal_newlines=True
                        )
                        lines = output.split('\n')
                        failed_count += len(lines) - 1
# 提取失败的用户名
for line in lines:
if'for'in line and'from'in line:
# 解析用户名
                                parts = line.split()
if'for'in parts:
                                    user_index = parts.index('for') + 1
if user_index < len(parts):
# 处理 "invalid user" 情况
if parts[user_index] == 'invalid'and user_index + 1 < len(parts) and parts[user_index + 1] == 'user':
# 格式：Failed password for invalid user test from...
if user_index + 2 < len(parts):
                                                user = parts[user_index + 2]
if user notin failed_users:
                                                    failed_users.append(user)
else:
# 格式：Failed password for test from...
                                            user = parts[user_index]
if user notin failed_users:
                                                failed_users.append(user)
except:
pass
return failed_count, failed_users
except:
return0, []

defget_recent_logins(self):
"""获取最近登录记录"""
import subprocess
try:
            output = subprocess.check_output(
                ['last', '-n', '5'],
                stderr=subprocess.STDOUT,
                universal_newlines=True
            )
return output.strip()
except:
return"无法获取登录记录"

defgenerate_report(self):
"""生成巡检报告"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        report = f"""===========================================
服务器巡检报告
===========================================
主机名: {self.hostname}
IP地址: {self.ip}
巡检时间: {timestamp}

【CPU 状态】
使用率: {self.results['cpu']['使用率']}
状态: {self.results['cpu']['状态']}

【内存状态】
使用率: {self.results['内存']['使用率']}
总内存: {self.results['内存']['总内存']}
可用内存: {self.results['内存']['可用内存']}
状态: {self.results['内存']['状态']}

【磁盘状态】
"""

for disk in self.results['磁盘']:
            report += f"挂载点: {disk['挂载点']}\n"
            report += f"文件系统: {disk['文件系统']}\n"
            report += f"使用率: {disk['使用率']}\n"
            report += f"总空间: {disk['总空间']}\n"
            report += f"可用空间: {disk['可用空间']}\n"
            report += f"状态: {disk['状态']}\n\n"

        report += "【进程状态】\n"
for process in self.results['进程']:
            report += f"进程名: {process['进程名']}\n"
            report += f"状态: {process['状态']}\n"

        report += "\n【端口状态】\n"
for port in self.results['端口']:
            report += f"端口: {port['端口']}\n"
            report += f"状态: {port['状态']}\n"

        report += "\n【系统启动时间】\n"
        report += f"运行时间: {self.results['系统启动时间']['运行时间']}\n"
        report += f"启动时间: {self.results['系统启动时间']['启动时间']}\n"

        report += "\n【僵尸进程】\n"
        report += f"数量: {self.results['僵尸进程']['数量']}\n"
        report += f"状态: {self.results['僵尸进程']['状态']}\n"

        report += "\n【网络状态】\n"
        report += f"连接数: {self.results['网络状态']['连接数']}\n"
        report += f"网络延迟: {self.results['网络状态']['网络延迟']}\n"
        report += f"状态: {self.results['网络状态']['状态']}\n"
        report += "接口信息:\n"
for interface in self.results['网络状态']['接口信息']:
            report += f"  - {interface['接口']}: 发送 {interface['发送流量']}, 接收 {interface['接收流量']}\n"

        report += "\n【安全状态】\n"
        report += f"登录失败次数: {self.results['安全状态']['登录失败次数']}\n"
if self.results['安全状态']['登录失败的用户']:
            report += f"登录失败的用户: {', '.join(self.results['安全状态']['登录失败的用户'])}\n"
        report += f"状态: {self.results['安全状态']['状态']}\n"
if self.results['安全状态']['权限问题']:
            report += "权限问题:\n"
for issue in self.results['安全状态']['权限问题']:
                report += f"  - {issue}\n"
        report += f"最近登录:\n{self.results['安全状态']['最近登录']}\n"

        report += "==========================================="

# 保存报告到文件
        report_file = f"server_monitor_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
with open(report_file, 'w') as f:
            f.write(report)

return report, report_file

defsend_dingtalk_alert(self, webhook, report):
"""发送钉钉告警"""
try:
            message = {
"msgtype": "text",
"text": {
"content": f"服务器巡检异常\n{report}"
                }
            }
            response = requests.post(webhook, json=message)
return response.status_code == 200
except:
returnFalse

defsend_wecom_alert(self, webhook, report):
"""发送企业微信告警"""
try:
            message = {
"msgtype": "text",
"text": {
"content": f"服务器巡检异常\n{report}"
                }
            }
            response = requests.post(webhook, json=message)
return response.status_code == 200
except:
returnFalse

defrun(self, process_names=None, ports=None, alert_config=None):
"""执行巡检"""
if process_names isNone:
            process_names = ['sshd', 'nginx', 'mysql']
if ports isNone:
            ports = [22, 80, 443]

# 执行各项检查
        self.check_cpu()
        self.check_memory()
        self.check_disk()
        self.check_processes(process_names)
        self.check_ports(ports)
        self.check_uptime()
        self.check_zombie_processes()
        self.check_network_status()
        self.check_security()

# 生成报告
        report, report_file = self.generate_report()
        print(report)
        print(f"报告已保存至: {report_file}")

# 检查是否有异常
        has_alert = False
        alert_message = ""

if self.results['cpu']['状态'] == "警告":
            has_alert = True
            alert_message += f"CPU 使用率异常: {self.results['cpu']['使用率']}\n"

if self.results['内存']['状态'] == "警告":
            has_alert = True
            alert_message += f"内存使用率异常: {self.results['内存']['使用率']}\n"

for disk in self.results['磁盘']:
if disk['状态'] == "警告":
                has_alert = True
                alert_message += f"磁盘 {disk['挂载点']} 使用率异常: {disk['使用率']}\n"

for process in self.results['进程']:
if process['状态'] == "未运行":
                has_alert = True
                alert_message += f"进程 {process['进程名']} 未运行\n"

for port in self.results['端口']:
if port['状态'] == "关闭":
                has_alert = True
                alert_message += f"端口 {port['端口']} 未开放\n"

if self.results['僵尸进程']['状态'] == "警告":
            has_alert = True
            alert_message += f"僵尸进程数量异常: {self.results['僵尸进程']['数量']}\n"

if self.results['网络状态']['状态'] == "警告":
            has_alert = True
            alert_message += f"网络状态异常: 延迟 {self.results['网络状态']['网络延迟']}\n"

if self.results['安全状态']['状态'] == "警告":
            has_alert = True
            alert_message += f"安全状态异常: 登录失败 {self.results['安全状态']['登录失败次数']} 次\n"
if self.results['安全状态']['登录失败的用户']:
                alert_message += f"  - 登录失败的用户: {', '.join(self.results['安全状态']['登录失败的用户'])}\n"
if self.results['安全状态']['权限问题']:
for issue in self.results['安全状态']['权限问题']:
                    alert_message += f"  - {issue}\n"

# 发送告警
if has_alert and alert_config:
if alert_config.get('type') == 'dingtalk'and alert_config.get('webhook'):
                success = self.send_dingtalk_alert(alert_config['webhook'], alert_message)
                print(f"钉钉告警发送{'成功'if success else'失败'}")
elif alert_config.get('type') == 'wecom'and alert_config.get('webhook'):
                success = self.send_wecom_alert(alert_config['webhook'], alert_message)
                print(f"企业微信告警发送{'成功'if success else'失败'}")

return has_alert

if __name__ == '__main__':
# 执行巡检
    monitor = ServerMonitor()
    monitor.run(
        process_names=CHECK_PROCESSES,
        ports=CHECK_PORTS,
        alert_config=ALERT_CONFIG
    )

五、最后

通过本文介绍的 Python 脚本，基本可以实现 Linux 服务器的自动化巡检，及时发现并解决潜在问题。脚本不仅包含了基础的系统指标检查，还提供了自动生成报告和发送告警的功能，大大提高了运维效率。