Shell只能干苦力?加上Python和Ansible,这套巡检方案绝了!

运维圈子里有个说法："Shell脚本是运维的祖传手艺，不会写Shell就不算真正的运维工程师。"这话有一定道理，但不全对。Shell确实强大，但在大规模服务器巡检场景下，纯Shell方案往往会陷入"写脚本两小时，改bug一整天"的困境。

今天聊一套经过生产环境验证的巡检方案：Shell负责数据采集，Ansible负责批量调度，Python负责数据处理。三者各司其职，既能发挥各自优势，又能形成完整的自动化闭环。

一、为什么单一工具搞不定巡检

Shell的局限性

Shell脚本在单台服务器上执行命令确实高效，但面对几十台甚至上百台服务器时，问题接踵而至：

• • 并发控制能力弱：手动写循环串行执行，100台服务器巡检一遍可能要几十分钟
• • 结果汇总困难：每台服务器生成一个日志文件，人工查看效率极低
• • 跨平台兼容性差：Linux用Bash，Windows得换PowerShell，维护两套脚本成本翻倍

Ansible的短板

Ansible作为自动化运维利器，在批量执行方面确实出色，但直接用它做数据分析并不顺手。Ansible的debug模块输出的是文本格式，想要生成可视化报表或进行阈值分析，还需要额外的数据处理环节。

Python的定位

Python擅长数据处理和报表生成，但如果直接用它连接远程服务器执行命令，需要引入Paramiko等库处理SSH连接、身份认证、并发控制，反而把简单的事情复杂化了。

结论很明确：没有银弹，让合适的工具做合适的事。

二、方案架构设计

这套方案的核心思路是"分层解耦"：

数据流向如下：

目标服务器(Shell采集) 
    ↓
Ansible控制节点(批量收集)
    ↓
原始数据文件(JSON/CSV格式)
    ↓
Python处理脚本(分析+报表)
    ↓
巡检报告(Excel/HTML) + 告警通知

三、具体实现步骤

第一步：编写Shell采集脚本

在目标服务器上部署采集脚本collect.sh，采集核心系统指标：

#!/bin/bash
# 采集时间
CHECK_TIME=$(date"+%Y-%m-%d %H:%M:%S")
HOSTNAME=$(hostname)
IP=$(hostname -I | awk '{print $1}')

# CPU使用率（取1分钟平均负载）
CPU_LOAD=$(uptime | awk -F'load average:''{print $2}' | awk '{print $1}' | sed 's/,//')
CPU_CORES=$(nproc)
CPU_USAGE=$(echo"scale=2; $CPU_LOAD / $CPU_CORES * 100" | bc)

# 内存使用率
MEM_TOTAL=$(free -m | awk '/Mem:/ {print $2}')
MEM_USED=$(free -m | awk '/Mem:/ {print $3}')
MEM_USAGE=$(echo"scale=2; $MEM_USED / $MEM_TOTAL * 100" | bc)

# 磁盘使用率（检查根分区）
DISK_USAGE=$(df -h / | awk 'NR==2 {print $5}' | sed 's/%//')

# 关键服务状态（根据实际业务调整）
NGINX_STATUS=$(systemctl is-active nginx 2>/dev/null || echo"unknown")
MYSQL_STATUS=$(systemctl is-active mysql 2>/dev/null || echo"unknown")

# 输出JSON格式，方便后续处理
cat <<EOF
{
    "hostname": "$HOSTNAME",
    "ip": "$IP",
    "check_time": "$CHECK_TIME",
    "cpu_usage": "$CPU_USAGE",
    "mem_usage": "$MEM_USAGE",
    "disk_usage": "$DISK_USAGE",
    "nginx_status": "$NGINX_STATUS",
    "mysql_status": "$MYSQL_STATUS"
}
EOF

这个脚本的关键点是输出标准化JSON，避免Ansible收集后再做复杂的文本解析。

第二步：编写Ansible Playbook

创建inspection.yml，实现批量采集：

---
-name:服务器自动化巡检
hosts:all
gather_facts:no
vars:
output_dir:"/opt/inspection/reports"
local_report_dir:"./reports"

tasks:
-name:确保远程采集脚本存在
copy:
src:./collect.sh
dest:/tmp/collect.sh
mode:'0755'

-name:执行采集脚本
shell:/tmp/collect.sh
register:inspection_result
ignore_errors:yes

-name:在控制节点创建报告目录
file:
path:"{{ local_report_dir }}"
state:directory
delegate_to:localhost
run_once:yes

-name:保存采集结果到本地
copy:
content:"{{ inspection_result.stdout }}"
dest:"{{ local_report_dir }}/{{ inventory_hostname }}.json"
delegate_to:localhost
when:inspection_result.rc==0

-name:记录失败主机
lineinfile:
path:"{{ local_report_dir }}/failed_hosts.txt"
line:"{{ inventory_hostname }} - {{ inspection_result.stderr }}"
create:yes
delegate_to:localhost
when:inspection_result.rc!=0

几点实践经验：

• • 使用delegate_to: localhost将结果汇总到Ansible控制节点，避免分散在各目标服务器上
• • ignore_errors: yes确保单台服务器故障不影响整体巡检流程
• • 失败主机单独记录，便于后续排查

第三步：Python数据处理与报表生成

创建generate_report.py，实现数据分析和可视化：

import json
import os
import pandas as pd
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

defload_data(report_dir):
"""加载所有JSON巡检结果"""
    data = []
for filename in os.listdir(report_dir):
if filename.endswith('.json'):
withopen(os.path.join(report_dir, filename), 'r') as f:
try:
                    record = json.load(f)
                    data.append(record)
except json.JSONDecodeError:
print(f"解析失败: {filename}")
return data

defanalyze_data(data):
"""数据分析与阈值检查"""
    alerts = []
for item in data:
# CPU告警阈值：80%
iffloat(item.get('cpu_usage', 0)) > 80:
            alerts.append({
'hostname': item['hostname'],
'ip': item['ip'],
'item': 'CPU使用率',
'value': item['cpu_usage'],
'threshold': '80%',
'level': 'warning'
            })

# 内存告警阈值：85%
iffloat(item.get('mem_usage', 0)) > 85:
            alerts.append({
'hostname': item['hostname'],
'ip': item['ip'],
'item': '内存使用率',
'value': item['mem_usage'],
'threshold': '85%',
'level': 'critical'
            })

# 磁盘告警阈值：90%
iffloat(item.get('disk_usage', 0)) > 90:
            alerts.append({
'hostname': item['hostname'],
'ip': item['ip'],
'item': '磁盘使用率',
'value': item['disk_usage'],
'threshold': '90%',
'level': 'critical'
            })

# 服务状态检查
if item.get('nginx_status') != 'active':
            alerts.append({
'hostname': item['hostname'],
'ip': item['ip'],
'item': 'Nginx服务',
'value': item['nginx_status'],
'threshold': 'active',
'level': 'critical'
            })

return alerts

defgenerate_excel(data, alerts, output_file):
"""生成Excel巡检报告"""
with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
# 汇总 sheet
        df_summary = pd.DataFrame(data)
        df_summary.to_excel(writer, sheet_name='巡检汇总', index=False)

# 告警 sheet
if alerts:
            df_alerts = pd.DataFrame(alerts)
            df_alerts.to_excel(writer, sheet_name='告警详情', index=False)
else:
            pd.DataFrame({'状态': ['无告警']}).to_excel(
                writer, sheet_name='告警详情', index=False
            )

defsend_email(report_file, recipients):
"""发送邮件通知"""
    msg = MIMEMultipart()
    msg['Subject'] = f"服务器巡检报告 - {datetime.now().strftime('%Y-%m-%d')}"
    msg['From'] = 'ops@company.com'
    msg['To'] = ', '.join(recipients)

# 邮件正文
    body = f"""
    各位好，

    附件为 {datetime.now().strftime('%Y-%m-%d')} 的服务器巡检报告，请查收。

    如有异常，请及时处理。
    """
    msg.attach(MIMEText(body, 'plain', 'utf-8'))

# 附件
withopen(report_file, 'rb') as f:
        attachment = MIMEApplication(f.read())
        attachment.add_header(
'Content-Disposition', 
'attachment', 
            filename=os.path.basename(report_file)
        )
        msg.attach(attachment)

# 发送（根据实际环境配置SMTP服务器）
# server = smtplib.SMTP('smtp.company.com', 25)
# server.send_message(msg)
# server.quit()

if __name__ == '__main__':
    report_dir = './reports'
    output_file = f"inspection_report_{datetime.now().strftime('%Y%m%d')}.xlsx"

# 加载数据
    data = load_data(report_dir)

# 分析告警
    alerts = analyze_data(data)

# 生成报表
    generate_excel(data, alerts, output_file)
print(f"报告已生成: {output_file}")

# 发送邮件（可选）
# send_email(output_file, ['admin@company.com'])

第四步：定时任务配置

使用crontab实现每日自动巡检：

# 每天凌晨2点执行巡检
0 2 * * * cd /opt/inspection && ansible-playbook -i hosts inspection.yml && python3 generate_report.py

四、适用场景

解决了哪些痛点

1. 1. 效率提升：从手动登录每台服务器检查，到一键批量执行，100台服务器的巡检时间从2小时缩短到5分钟以内
2. 2. 结果标准化：JSON格式的结构化数据，避免了Shell文本解析的繁琐和错误
3. 3. 可追溯性：每日生成的报告文件留存，形成历史趋势数据，便于容量规划
4. 4. 扩展性强：新增巡检项只需修改Shell采集脚本，无需改动整体架构

适用场景

• • 中小企业：没有预算采购商业监控软件，但需要基本的巡检报告
• • 混合云环境：同时管理本地机房和云服务器，Ansible的SSH方式通用性强
• • 合规要求：需要定期生成纸质或电子巡检报告存档的行业（如金融、政务）

这套方案不是最酷炫的，但确实是最务实的。它不需要部署Agent，不依赖特定厂商的监控平台，用开源工具组合解决了实际问题。

当然，如果企业规模扩大到数千台服务器，或者需要实时监控而非定时巡检，建议引入Prometheus、Zabbix等专业监控方案。但在那之前，Shell + Ansible + Python的组合，足以支撑大多数场景的运维需求。