第352篇：网络健康检查体系设计与自动化实践
关键词
健康检查、网络巡检、自动化检查、KPI、性能基线、健康评分、定期巡检、巡检报告
一、案例背景
1.1 为什么需要健康检查
某企业网络健康检查需求：
痛点：
❌ 手动巡检耗时（2 人 × 2 天/月） ❌ 检查项目不一致（每次查的都不一样） ❌ 问题发现滞后（故障后才查原因） ❌ 缺乏历史对比（无法发现性能劣化） ❌ 报告格式混乱（不同人写的格式不同）
目标：
✅ 全自动化巡检 ✅ 统一的检查标准和评分 ✅ 主动发现风险（故障前预警） ✅ 历史趋势分析 ✅ 标准化报告输出
1.2 健康检查分层
网络健康检查分层模型：
L1：设备级健康检查（每日/每周）
✓ 设备 CPU 利用率 ✓ 设备内存利用率 ✓ 设备温度 ✓ 电源状态 ✓ 风扇状态 ✓ 接口状态（up/down）
L2：协议级健康检查（每周/每月）
✓ OSPF 邻居状态 ✓ BGP 邻居状态 ✓ 路由表条目数量 ✓ STP 拓扑变化 ✓ VRRP 状态 ✓ M-LAG 状态
L3：性能级健康检查（每月/每季）
✓ 接口带宽利用率 ✓ 流量趋势分析 ✓ 错误包统计 ✓ 丢包率 ✓ 延迟变化 ✓ 队列深度
L4：安全级健康检查（每月）
✓ ACL 策略命中率 ✓ 端口安全状态 ✓ 非法 DHCP 服务器检测 ✓ ARP 攻击检测 ✓ 登录失败记录
二、健康检查系统设计
2.1 评分模型
健康评分模型：
总分 = 100
扣分项：
┌──────────────────────────────────────────┐
│ 扣分 扣分值 │
├──────────────────────────────────────────┤
│ 设备 CPU > 80% -10 │
│ 设备 CPU > 90% -20 │
│ 内存 > 80% -10 │
│ 内存 > 90% -20 │
│ 接口 down（非管理 shutdown） -5/个 │
│ BGP 邻居 down -15 │
│ OSPF 邻居 down -10 │
│ 错误包增长 > 100% -5 │
│ 温度 > 阈值 -10 │
│ 电源/风扇异常 -15 │
│ 路由条目变化 > 20% -10 │
│ STP 拓扑变化频繁 -5 │
└──────────────────────────────────────────┘
评分等级：
┌──────────────────────────────────────────┐
│ 等级 分数 说明 │
├──────────────────────────────────────────┤
│ Excellent 90-100 健康，无需关注 │
│ Good 75-89 良好，持续监控 │
│ Warning 60-74 异常，需要关注 │
│ Critical 0-59 严重，立即处理 │
└──────────────────────────────────────────┘
2.2 自动化健康检查脚本
#!/usr/bin/env python3
"""
网络健康检查自动化框架
支持多层级检查，输出评分和报告
"""
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
import json
import math
@dataclass
class CheckResult:
"""检查结果"""
item: str
status: str # pass / fail / warning
value: Any
threshold: Any
score_deduction: float
detail: str = ""
@dataclass
class DeviceHealth:
"""设备健康状态"""
device_name: str
device_role: str # core / aggregation / access
ip_address: str
vendor: str
results: List[CheckResult] = field(default_factory=list)
total_score: float = 100.0
def add_result(self, result: CheckResult):
self.results.append(result)
self.total_score -= result.score_deduction
def get_grade(self) -> str:
if self.total_score >= 90:
return "Excellent"
elif self.total_score >= 75:
return "Good"
elif self.total_score >= 60:
return "Warning"
else:
return "Critical"
class HealthChecker:
"""健康检查器"""
def __init__(self):
self.devices: List[DeviceHealth] = []
self.history: Dict[str, List[DeviceHealth]] = {}
def add_device(self, device: DeviceHealth):
self.devices.append(device)
def check_cpu(self, device: DeviceHealth, cpu_util: float):
"""检查 CPU"""
if cpu_util > 90:
result = CheckResult(
"CPU 利用率", "fail", f"{cpu_util}%",
"90%", 20,
f"CPU 利用率 {cpu_util}% 超过 90% 阈值"
)
elif cpu_util > 80:
result = CheckResult(
"CPU 利用率", "warning", f"{cpu_util}%",
"80%", 10,
f"CPU 利用率 {cpu_util}% 超过 80% 警告线"
)
else:
result = CheckResult(
"CPU 利用率", "pass", f"{cpu_util}%",
"80%", 0
)
device.add_result(result)
def check_memory(self, device: DeviceHealth, mem_util: float):
"""检查内存"""
if mem_util > 90:
result = CheckResult(
"内存利用率", "fail", f"{mem_util}%",
"90%", 20,
f"内存利用率 {mem_util}% 超过 90%"
)
elif mem_util > 80:
result = CheckResult(
"内存利用率", "warning", f"{mem_util}%",
"80%", 10
)
else:
result = CheckResult(
"内存利用率", "pass", f"{mem_util}%",
"80%", 0
)
device.add_result(result)
def check_interfaces(self, device: DeviceHealth,
interfaces: Dict[str, bool]):
"""检查接口状态"""
down_interfaces = [
name for name, status in interfaces.items()
if not status
]
deduction = min(len(down_interfaces) * 5, 30)
device.add_result(CheckResult(
"接口状态", "warning" if down_interfaces else "pass",
f"{len(down_interfaces)} 个接口 down",
"0", deduction,
f"DOWN 接口: {', '.join(down_interfaces[:5])}"
))
def check_bgp(self, device: DeviceHealth, bgp_peers: Dict[str, bool]):
"""检查 BGP 邻居"""
down_peers = [
peer for peer, status in bgp_peers.items()
if not status
]
deduction = len(down_peers) * 15
device.add_result(CheckResult(
"BGP 邻居", "fail" if down_peers else "pass",
f"{len(down_peers)} 个 BGP 邻居 down",
"0", min(deduction, 30),
f"DOWN BGP: {', '.join(down_peers[:5])}"
))
def check_temperature(self, device: DeviceHealth,
temp: float, threshold: float = 65):
"""检查温度"""
if temp > threshold:
device.add_result(CheckResult(
"设备温度", "fail", f"{temp}°C",
f"{threshold}°C", 10,
f"温度 {temp}°C 超过 {threshold}°C"
))
else:
device.add_result(CheckResult(
"设备温度", "pass", f"{temp}°C",
f"{threshold}°C", 0
))
def generate_device_report(self, device: DeviceHealth) -> str:
"""生成单设备报告"""
report = f"""
设备健康检查报告
{'=' * 60}
设备信息:
名称: {device.device_name}
角色: {device.device_role}
IP: {device.ip_address}
厂商: {device.vendor}
综合评分: {device.total_score:.1f}/100
评级: {device.get_grade()}
检查明细:
"""
for r in device.results:
status_map = {
"pass": "✅",
"warning": "⚠",
"fail": "❌"
}
symbol = status_map.get(r.status, "❓")
report += (
f" {symbol} {r.item:<20} "
f"值: {r.value:<15} "
f"阈值: {r.threshold:<10} "
f"扣分: {r.score_deduction:.0f}\n"
)
if r.detail:
report += f" {r.detail}\n"
return report
def generate_summary_report(self) -> str:
"""生成汇总报告"""
total = len(self.devices)
grades = {}
for d in self.devices:
grade = d.get_grade()
grades[grade] = grades.get(grade, 0) + 1
summary = f"""
网络健康检查汇总报告
{'=' * 60}
检查时间: {datetime.now().isoformat()}
检查设备数: {total}
健康分布:
"""
for grade in ["Excellent", "Good", "Warning", "Critical"]:
count = grades.get(grade, 0)
bar = "█" * count
summary += f" {grade:<12} {bar} {count}\n"
summary += f"""
平均评分: {sum(d.total_score for d in self.devices) / total:.1f}
问题设备（需要立即关注）:
"""
critical = [
d for d in self.devices
if d.get_grade() in ("Warning", "Critical")
]
if critical:
for d in critical:
summary += f" ❌ {d.device_name} "
summary += f"(评分: {d.total_score:.1f}, "
summary += f"角色: {d.device_role})\n"
else:
summary += " 无\n"
return summary
def main():
"""主函数"""
checker = HealthChecker()
# 模拟设备数据
devices_data = [
{
"name": "Core-1", "role": "core",
"ip": "10.0.0.1", "vendor": "Huawei",
"cpu": 65.0, "mem": 55.0, "temp": 52,
"interfaces": {
"10GE1/0/1": True, "10GE1/0/2": True,
"10GE1/0/3": False, "10GE1/0/4": True
},
"bgp": {"ISP-1": True, "ISP-2": True, "Core-2": True}
},
{
"name": "Core-2", "role": "core",
"ip": "10.0.0.2", "vendor": "Huawei",
"cpu": 85.0, "mem": 82.0, "temp": 68,
"interfaces": {
"10GE1/0/1": True, "10GE1/0/2": False
},
"bgp": {"ISP-1": True, "ISP-2": False, "Core-1": True}
},
]
for dev in devices_data:
device = DeviceHealth(
device_name=dev["name"],
device_role=dev["role"],
ip_address=dev["ip"],
vendor=dev["vendor"]
)
checker.check_cpu(device, dev["cpu"])
checker.check_memory(device, dev["mem"])
checker.check_interfaces(device, dev["interfaces"])
checker.check_bgp(device, dev["bgp"])
checker.check_temperature(device, dev["temp"])
checker.add_device(device)
# 打印单设备报告
print(checker.generate_device_report(device))
# 打印汇总报告
print(checker.generate_summary_report())
if __name__ == "__main__":
main()
三、巡检周期与策略
巡检周期建议：
日检（5 分钟）：
✅ 设备在线状态 ✅ 核心接口 up/down ✅ CPU 和内存利用率 ✅ BGP/OSPF 邻居状态 ❌ 无需日报，异常时告警
周检（15 分钟）：
✅ 所有设备硬件健康 ✅ 接口错误包统计 ✅ 路由表条目变化 ✅ 温度趋势 ✅ 日志错误分析 ✅ 输出周报
月检（1 小时）：
✅ 接口带宽利用率峰值 ✅ 流量趋势分析（同环比） ✅ 性能基线对比 ✅ 安全策略审计 ✅ 配置合规检查 ✅ 输出月报
季检（半日）：
✅ 全面性能评估 ✅ 容量规划分析 ✅ 架构优化建议 ✅ 固件版本审计 ✅ EOS/EOL 设备检查 ✅ 输出季度健康报告
四、健康检查最佳实践
健康检查体系落地经验：
工具化： ┌──────────────────────────────────────────┐ │ 推荐工具链： │ │ └─ 数据采集：Ansible + Netmiko │ │ └─ 数据存储：InfluxDB + Prometheus │ │ └─ 可视化：Grafana │ │ └─ 告警：AlertManager │ │ └─ 报告：Jinja2 模板 + WeasyPrint │ └──────────────────────────────────────────┘
告警策略：
告警级别 响应时间 通知方式 Critical 5 分钟 电话+短信 High 15 分钟 IM 消息 Medium 1 小时 邮件 Low 下一个工作日 日报
持续改进： ┌──────────────────────────────────────────┐ │ □ 每月复盘检查项的有效性 │ │ □ 根据故障案例增加新的检查项 │ │ □ 定期调整阈值（根据历史数据） │ │ □ 优化检查脚本性能 │ │ □ 培训团队理解健康检查体系 │ └──────────────────────────────────────────┘
五、总结
健康检查体系建设关键要点：
1. 分层检查
└─ L1 设备级（基础运行状态）
└─ L2 协议级（路由协议健康）
└─ L3 性能级（流量和性能趋势）
└─ L4 安全级（安全策略评估）
2. 评分量化
└─ 标准化评分模型（0-100）
└─ 明确的扣分规则
└─ 等级划分（Excellent/Critical）
3. 持续自动化
└─ 定时自动执行
└─ 异常自动告警
└─ 报告自动生成
└─ 趋势自动分析
4. 闭环改进
└─ 检查发现问题
└─ 分析根因
└─ 制定改进方案
└─ 验证改进效果
下篇预告：第353篇《网络性能瓶颈分析与优化实战案例》——通过实际案例讲解网络性能瓶颈的分析方法和优化策略。
下篇预告：第353篇《网络性能瓶颈分析与优化实战案例》——通过实际案例讲解网络性能瓶颈的分析方法和优化策略。