Python批量服务器巡检脚本

#!/usr/bin/env python3 # -*- coding: utf-8 -*- # ========================================== # 批量服务器巡检脚本 # 用途：通过SSH批量巡检多台Linux服务器，输出汇总报告 # 依赖：pip install paramiko # 使用方法：python3 batch_server_check.py # ========================================== import paramiko import json from datetime import datetime from concurrent.futures import ThreadPoolExecutor, as_completed # ========== 服务器列表配置 ========== SERVERS = [ { "name": "Web服务器1", "host": "192.168.1.10", "port": 22, "username": "root", "password": "your_password", # 或使用 key_file # "key_file": "/root/.ssh/id_rsa", }, { "name": "数据库服务器", "host": "192.168.1.20", "port": 22, "username": "root", "password": "your_password", }, # 在此添加更多服务器 ] # 要检查的服务 CHECK_SERVICES = ["nginx", "mysql", "docker", "redis"] # 并发数 MAX_WORKERS = 5 # ========================================== def ssh_exec(client, command, timeout=10): # 执行SSH命令 try: stdin, stdout, stderr = client.exec_command(command, timeout=timeout) return stdout.read().decode('utf-8', errors='ignore').strip() except Exception as e: return f"ERROR: {e}" def check_server(server): # 检查单台服务器 result = { "name": server["name"], "host": server["host"], "status": "unknown", "cpu": {}, "memory": {}, "disk": [], "services": {}, "uptime": "", "errors": [] } client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: # 连接 connect_kwargs = { "hostname": server["host"], "port": server.get("port", 22), "username": server["username"], "timeout": 10, } if "key_file" in server: connect_kwargs["key_filename"] = server["key_file"] else: connect_kwargs["password"] = server["password"] client.connect(**connect_kwargs) result["status"] = "online" # 主机名和运行时间 result["hostname"] = ssh_exec(client, "hostname") result["uptime"] = ssh_exec(client, "uptime -p 2>/dev/null || uptime") # CPU信息 cpu_model = ssh_exec(client, "grep 'model name' /proc/cpuinfo | head -1 | cut -d: -f2 | sed 's/^ *//'") cpu_cores = ssh_exec(client, "nproc") cpu_load = ssh_exec(client, "cat /proc/loadavg | awk '{print $1, $2, $3}'") result["cpu"] = { "model": cpu_model, "cores": cpu_cores, "load": cpu_load } # 内存 mem_info = ssh_exec(client, "free -m | awk 'NR==2{printf "%s,%s,%s,%s", $2, $3, $4, $7}'") if mem_info and "," in mem_info: parts = mem_info.split(",") result["memory"] = { "total_mb": parts[0], "used_mb": parts[1], "free_mb": parts[2], "available_mb": parts[3], "usage_percent": f"{int(parts[1]) / int(parts[0]) * 100:.1f}%" if parts[0] != "0" else "0%" } # 磁盘 disk_info = ssh_exec(client, "df -hP | grep -v tmpfs | grep -v devtmpfs | awk 'NR>1{print $2","$3","$4","$5","$6}'") for line in disk_info.split("\n"): if line and "," in line: parts = line.split(",") if len(parts) >= 5: usage = int(parts[3].replace("%", "")) result["disk"].append({ "total": parts[0], "used": parts[1], "free": parts[2], "usage": parts[3], "mount": parts[4], "warning": usage > 80 }) # 服务状态 for svc in CHECK_SERVICES: status = ssh_exec(client, f"systemctl is-active {svc} 2>/dev/null || echo 'not-found'") result["services"][svc] = status # 检查是否有高CPU/内存进程 top_cpu = ssh_exec(client, "ps aux --sort=-%cpu | head -4 | awk 'NR>1{print $11, $3"%"}'") top_mem = ssh_exec(client, "ps aux --sort=-%mem | head -4 | awk 'NR>1{print $11, $4"%"}'") result["top_cpu"] = top_cpu.split("\n") result["top_mem"] = top_mem.split("\n") except Exception as e: result["status"] = "offline" result["errors"].append(str(e)) finally: client.close() return result def generate_report(results): # 生成巡检报告 report = [] report.append("=" * 70) report.append(" 批量服务器巡检报告") report.append(f" 巡检时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f" 服务器数量: {len(results)}") report.append("=" * 70) report.append("") online = sum(1 for r in results if r["status"] == "online") offline = sum(1 for r in results if r["status"] == "offline") report.append(f"【汇总】在线: {online}, 离线: {offline}") report.append("") for r in results: report.append("-" * 70) report.append(f" {r['name']} ({r['host']})") report.append("-" * 70) if r["status"] == "offline": report.append(f" ❌ 无法连接: {'; '.join(r['errors'])}") report.append("") continue report.append(f" 主机名: {r.get('hostname', 'N/A')}") report.append(f" 运行时间: {r.get('uptime', 'N/A')}") report.append("") # CPU cpu = r.get("cpu", {}) report.append(f" 【CPU】{cpu.get('model', 'N/A')} | 核心: {cpu.get('cores', 'N/A')} | 负载: {cpu.get('load', 'N/A')}") # 内存 mem = r.get("memory", {}) if mem: report.append(f" 【内存】总计: {mem.get('total_mb', '?')}MB | 已用: {mem.get('used_mb', '?')}MB | 使用率: {mem.get('usage_percent', '?')}") # 磁盘 report.append(" 【磁盘】") for d in r.get("disk", []): warn = " ⚠️" if d.get("warning") else "" report.append(f" {d['mount']:>15} | 总计: {d['total']:>6} | 已用: {d['used']:>6} | 使用率: {d['usage']:>5}{warn}") # 服务 report.append(" 【服务状态】") for svc, status in r.get("services", {}).items(): icon = "🟢" if status == "active" else "🔴" if status in ("failed", "inactive", "not-found") else "🟡" report.append(f" {icon} {svc:>12}: {status}") report.append("") # 保存JSON with open(f"server_check_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json", "w", encoding="utf-8") as f: json.dump(results, f, ensure_ascii=False, indent=2) report.append("=" * 70) report.append(" 巡检完成，详细数据已保存为JSON文件") report.append("=" * 70) return "\n".join(report) def main(): print("开始批量巡检...") print(f"目标服务器: {len(SERVERS)} 台") print() results = [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: futures = {executor.submit(check_server, s): s for s in SERVERS} for future in as_completed(futures): server = futures[future] try: result = future.result() results.append(result) status_icon = "🟢" if result["status"] == "online" else "🔴" print(f" {status_icon} {server['name']} ({server['host']}) - {result['status']}") except Exception as e: print(f" ❌ {server['name']} 检查异常: {e}") print() report = generate_report(results) print(report) if __name__ == "__main__": main()

[{"name": "SERVERS", "label": "\u670d\u52a1\u5668\u5217\u8868(JSON\u683c\u5f0f)", "type": "textarea", "default": "[{\"name\":\"server1\",\"host\":\"192.168.1.10\",\"port\":22,\"username\":\"root\",\"password\":\"pass\"}]"}]