1. 项目背景与需求分析
在服务器测试工作中，存储设备的检测是基础但极其重要的一环。传统的人工逐块检查硬盘的方式效率低下，特别是在面对12块甚至更多硬盘的服务器配置时，测试工程师往往需要花费大量时间重复执行相同的检查步骤。这不仅消耗人力成本，还容易因疲劳导致人为错误。
我曾在一次数据中心巡检中，遇到需要手动检查48台服务器的384块硬盘的情况。当时采用传统方法，三个工程师整整花费了两天时间才完成基础检测。正是这次经历让我意识到，必须开发一个自动化工具来解决这个痛点。
存储自动检测脚本的核心需求可以归纳为：
- 自动识别服务器上所有存储设备（包括NVMe、SATA、SAS等不同类型）
- 采集每块硬盘的关键参数（容量、型号、固件版本、SMART状态等）
- 生成统一格式的检测报告，便于后续分析和存档
- 支持定时执行和异常告警功能
2. 环境准备与工具选型
2.1 Python环境配置
选择Python作为开发语言主要基于以下考虑：
- 跨平台兼容性：Python在Linux服务器上运行良好
- 丰富的系统管理库：如psutil、pySMART等
- 简洁的语法：快速开发原型和后期维护
推荐使用Python 3.6+版本，可通过以下命令检查当前环境：
python3 --version pip3 --version对于开发环境，我强烈建议使用virtualenv创建隔离环境：
python3 -m venv storage-checker source storage-checker/bin/activate2.2 关键依赖库安装
本脚本需要以下Python库支持：
pip install psutil pySMART pandas tabulate其中：
- psutil：用于获取系统信息
- pySMART：专业的SMART数据读取库
- pandas：数据处理和报告生成
- tabulate：美化终端输出
注意：在生产环境部署时，建议使用requirements.txt固定版本：
pip freeze > requirements.txt
3. 核心功能实现
3.1 存储设备枚举
使用lsblk命令获取所有块设备信息是最可靠的方法。在Python中可以通过subprocess调用：
import subprocess import json def get_block_devices(): cmd = "lsblk -J -o NAME,MODEL,SIZE,TYPE,ROTA,TRAN" result = subprocess.run(cmd.split(), capture_output=True, text=True) return json.loads(result.stdout)这个实现有几个关键点：
- -J参数输出JSON格式便于解析
- 包含的关键字段：
- NAME：设备名称（如sda、nvme0n1）
- MODEL：硬盘型号
- SIZE：容量
- TYPE：设备类型（disk/part）
- ROTA：是否旋转设备（SSD为0）
- TRAN：传输协议（sata/nvme等）
3.2 NVMe设备特殊处理
NVMe设备需要额外获取更多信息，可以使用nvme-cli工具：
def get_nvme_info(device): cmd = f"nvme list -o json" result = subprocess.run(cmd.split(), capture_output=True, text=True) nvme_data = json.loads(result.stdout) # 提取特定设备信息 for dev in nvme_data["Devices"]: if dev["DevicePath"] == f"/dev/{device}": return { "firmware": dev["Firmware"], "serial": dev["SerialNumber"], "namespace": dev["NameSpace"] } return {}3.3 SMART数据采集
对于传统硬盘，使用pySMART库获取SMART信息：
from pySMART import DeviceList def get_smart_data(device): dev_list = DeviceList() disk = dev_list.get_dev_by_name(device) return { "health": disk.assessment, "temperature": disk.temperature, "power_on_hours": disk.attributes[9].raw, "reallocated_sectors": disk.attributes[5].raw }4. 报告生成与输出
4.1 数据结构设计
建议使用如下数据结构存储所有硬盘信息：
class StorageDevice: def __init__(self, name, model, size, device_type, protocol): self.name = name self.model = model self.size = size self.type = device_type self.protocol = protocol self.smart = {} self.nvme_info = {}4.2 终端表格输出
使用tabulate库生成美观的终端输出：
from tabulate import tabulate def print_terminal_report(devices): table = [] for dev in devices: row = [ dev.name, dev.model, dev.size, "Healthy" if dev.smart.get("health") == "PASS" else "Warning", dev.smart.get("temperature", "N/A") ] table.append(row) headers = ["Device", "Model", "Size", "Health", "Temp(℃)"] print(tabulate(table, headers=headers, tablefmt="grid"))4.3 Excel报告生成
使用pandas生成更专业的Excel报告：
import pandas as pd def generate_excel_report(devices, filename="storage_report.xlsx"): data = [] for dev in devices: data.append({ "Device": dev.name, "Model": dev.model, "Size": dev.size, "Type": dev.type, "Protocol": dev.protocol, "Health": dev.smart.get("health", "N/A"), "Temperature": dev.smart.get("temperature", "N/A"), "PowerOnHours": dev.smart.get("power_on_hours", "N/A"), "Firmware": dev.nvme_info.get("firmware", "N/A") }) df = pd.DataFrame(data) df.to_excel(filename, index=False)5. 完整脚本集成
5.1 主程序逻辑
def main(): print("Storage Health Checker v1.0") print("="*50) # 获取所有块设备 block_data = get_block_devices() devices = [] for device in block_data["blockdevices"]: if device["type"] != "disk": continue # 创建设备对象 dev = StorageDevice( name=device["name"], model=device.get("model", "Unknown"), size=device["size"], device_type="SSD" if device.get("rota") == "0" else "HDD", protocol=device.get("tran", "Unknown") ) # 获取详细信息 if dev.protocol == "nvme": dev.nvme_info = get_nvme_info(dev.name) else: dev.smart = get_smart_data(dev.name) devices.append(dev) # 输出报告 print_terminal_report(devices) generate_excel_report(devices) print("\nReport generated: storage_report.xlsx") if __name__ == "__main__": main()5.2 异常处理增强
在实际运行中，需要考虑各种异常情况：
try: smart_data = get_smart_data(device) except Exception as e: print(f"Warning: Failed to get SMART data for {device}: {str(e)}") smart_data = {"health": "UNKNOWN"}6. 实际应用与优化建议
6.1 定时任务集成
可以将脚本设置为cron定时任务，例如每天凌晨3点执行：
0 3 * * * /path/to/python /path/to/storage_checker.py6.2 邮件告警功能
添加邮件发送功能，当检测到异常硬盘时自动发送告警：
import smtplib from email.mime.text import MIMEText def send_alert(subject, content): msg = MIMEText(content) msg["Subject"] = subject msg["From"] = "storage-checker@yourdomain.com" msg["To"] = "admin@yourdomain.com" with smtplib.SMTP("smtp.yourdomain.com") as server: server.send_message(msg)6.3 性能优化技巧
- 并行采集：对于多块硬盘，可以使用多线程并行采集数据
from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=4) as executor: results = list(executor.map(get_smart_data, device_names))结果缓存：将上次检测结果缓存，只检查变化部分
日志记录：添加详细的运行日志，便于问题排查
7. 常见问题与解决方案
7.1 权限问题处理
脚本可能需要root权限才能读取SMART数据。解决方法：
- 使用sudo运行脚本
- 将用户加入disk组
sudo usermod -aG disk $USER7.2 NVMe工具缺失
如果nvme-cli未安装，可以使用以下命令安装：
# Ubuntu/Debian sudo apt install nvme-cli # RHEL/CentOS sudo yum install nvme-cli7.3 中文乱码问题
在终端输出时可能出现中文乱码，解决方法：
import locale locale.setlocale(locale.LC_ALL, "en_US.UTF-8")8. 脚本扩展思路
8.1 硬盘性能测试集成
可以扩展加入基础性能测试功能：
def test_disk_speed(device): # 使用dd命令测试顺序读写 cmd = f"dd if=/dev/{device} of=/dev/null bs=1M count=1024 iflag=direct" result = subprocess.run(cmd.split(), capture_output=True, text=True) # 解析输出获取速度信息 ...8.2 与监控系统集成
将采集到的数据推送到Prometheus等监控系统：
from prometheus_client import Gauge, push_to_gateway disk_temp = Gauge("disk_temperature", "Disk temperature in Celsius", ["device"]) disk_temp.labels(device="sda").set(45) push_to_gateway("localhost:9091", job="storage_check")8.3 Web可视化界面
使用Flask等框架开发简单的Web界面：
from flask import Flask, render_template app = Flask(__name__) @app.route("/") def dashboard(): devices = get_all_devices() return render_template("dashboard.html", devices=devices)在实际服务器测试工作中，这个脚本已经帮助我们的团队将存储检测时间从平均30分钟/台缩短到不到2分钟，同时大大降低了人为错误的概率。特别是在大规模部署前的验收测试中，自动化检测的优势更加明显。
