Python 自动化巡检脚本要能解释自己发现了什么一、巡检脚本不是跑一遍然后 outputs 0 就算成功很多团队的自动化巡检脚本写好之后就扔定时任务里跑。脚本的逻辑是检查各项指标如果有异常就 outputs 1否则 outputs 0。然后定时任务捕获 outputs 1 就发告警。这种方式的致命缺陷是告警里只有 outputs 1没有解释为什么。值班人员收到告警后还要手动去服务器上跑脚本、看日志、分析输出才能理解脚本到底发现了什么。如果脚本的作者已经离职这份理解成本会更高。好的巡检脚本应该能自己解释自己发现了什么。不光要报告有问题还要报告什么问题、在哪里、可能影响什么。二、巡检脚本的输出规范结构化 可解释flowchart TD A[巡检脚本启动] -- B[检查各项指标] B -- C{发现异常?} C --|是| D[生成结构化报告] C --|否| E[生成健康报告] D -- F[输出到 stdout / 文件 / API] E -- F F -- G[告警系统 / 监控平台] G -- H[值班人员收到可解释告警] I[巡检脚本规范] -- J[输出必须是 JSON] I -- K[每条异常必须带 evidence] I -- L[必须能独立运行] I -- M[必须有退出码规范] style D fill:#bfb,stroke:#333 style H fill:#f9f,stroke:#333 style I fill:#bbf,stroke:#333我们团队规定所有巡检脚本的输出必须是JSON 格式包含以下字段{ check_name: disk_usage_check, service: order-db, status: warning, severity: P2, message: 磁盘使用率超过 85% 阈值, evidence: { metric: disk_usage_percent, current_value: 87.3, threshold: 85.0, trend: increasing, predicted_full_in_days: 3 }, suggested_action: 清理 7 天前的 binlog 或扩容磁盘, owner: database-team, runbook_url: https://wiki.example.com/order-db/disk-full-runbook }关键设计点evidence字段必须包含可验证的数据不能只说磁盘快满了要给出当前使用率、阈值、趋势预测。这样值班人员看到告警后可以立即判断严重性。suggested_action要具体不能只说请处理要说清理 7 天前的 binlog。runbook_url是救命稻草如果告警在凌晨 3 点触发值班人员可能迷迷糊糊不知道怎么处理。runbook 里应该有详细的步骤和命令。三、Python 巡检脚本的工程化实践一个生产级的 Python 巡检脚本应该包含以下模块#!/usr/bin/env python3 自动化巡检脚本模板 使用方法: python3 check_disk.py --service order-db --threshold 85 import argparse import json import sys import logging from datetime import datetime from typing import Dict, Any, Optional # 配置日志要包含 trace_id方便关联 logging.basicConfig( format%(asctime)s %(levelname)s [%(trace_id)s] %(message)s, levellogging.INFO ) logger logging.getLogger(__name__) class CheckResult: 巡检结果封装类 def __init__(self, check_name: str, service: str): self.check_name check_name self.service service self.status healthy # healthy / warning / critical self.severty: Optional[str] None self.message: Optional[str] None self.evidence: Dict[str, Any] {} self.suggested_action: Optional[str] None self.owner: Optional[str] None self.runbook_url: Optional[str] None def to_json(self) - str: 序列化为 JSON return json.dumps({ check_name: self.check_name, service: self.service, status: self.status, severity: self.severty, message: self.message, evidence: self.evidence, suggested_action: self.suggested_action, owner: self.owner, runbook_url: self.runbook_url, checked_at: datetime.now().isoformat(), }, ensure_asciiFalse, indent2) def is_healthy(self) - bool: return self.status healthy def check_disk_usage(service: str, threshold: float) - CheckResult: 检查磁盘使用率 result CheckResult(disk_usage_check, service) # 模拟查询监控系统的 API metrics query_metrics( serviceservice, metricdisk_usage_percent, time_range5m ) current_usage metrics[current_value] trend metrics[trend] # increasing / stable / decreasing # 判断是否超过阈值 if current_usage threshold: result.status warning if current_usage 95 else critical result.severty P2 if current_usage 95 else P0 result.message f磁盘使用率 {current_usage:.1f}% 超过阈值 {threshold:.1f}% # 关键提供证据 result.evidence { metric: disk_usage_percent, current_value: current_usage, threshold: threshold, trend: trend, predicted_full_in_days: predict_disk_full(current_usage, trend), } # 关键提供建议动作 if service.startswith(db): result.suggested_action 执行: PURGE BINARY LOGS BEFORE DATE_SUB(NOW(), INTERVAL 7 DAY); else: result.suggested_action 联系 SRE 团队评估是否需要扩容 result.owner database-team if service.startswith(db) else sre-team result.runbook_url fhttps://wiki.example.com/{service}/disk-runbook else: result.status healthy return result def query_metrics(service: str, metric: str, time_range: str) - Dict[str, Any]: 查询监控系统 API示例代码 # 实际实现要对接 Prometheus / VictoriaMetrics / Datadog 等 import random return { current_value: random.uniform(80, 98), trend: random.choice([increasing, stable, decreasing]), } def predict_disk_full(current_usage: float, trend: str) - Optional[int]: 预测磁盘多久会满简化版 if trend ! increasing: return None # 假设每天增长 1% days_remaining int((100 - current_usage) / 1) return max(0, days_remaining) def main(): parser argparse.ArgumentParser(description磁盘使用率巡检) parser.add_argument(--service, requiredTrue, help服务名) parser.add_argument(--threshold, typefloat, default85.0, help告警阈值) parser.add_argument(--output, choices[json, prometheus], defaultjson) args parser.parse_args() # 执行检查 result check_disk_usage(args.service, args.threshold) # 输出结果 if args.output json: print(result.to_json()) elif args.output prometheus: # 输出 Prometheus 格式的指标 print(fdisk_check_status{{service\{args.service}\}} {0 if result.is_healthy() else 1}) # 设置退出码方便定时任务判断 sys.exit(0 if result.is_healthy() else 1) if __name__ __main__: main()四、巡检脚本的运维化管理单个巡检脚本写好之后还要考虑批量管理、定时调度、结果聚合。我们的做法是所有巡检脚本放在 Git 仓库里用 Terraform 管理定时任务AWS CloudWatch Events 或 K8s CronJob。巡检结果统一推送到 S3 或 ELK方便事后分析和趋势查看。定期 review 巡检命中率多少告警是真的需要处理多少是误报根据 review 结果调整阈值或删除无用的检查项。# Kubernetes CronJob 示例每天凌晨 2 点运行磁盘巡检 apiVersion: batch/v1 kind: CronJob metadata: name: disk-check spec: schedule: 0 2 * * * jobTemplate: spec: template: spec: containers: - name: checker image: python:3.11-slim command: - python3 - /scripts/check_disk.py - --service - order-db - --threshold - 85 volumeMounts: - name: scripts mountPath: /scripts volumes: - name: scripts configMap: name: inspection-scripts restartPolicy: OnFailure五、总结Python 自动化巡检脚本的价值不在于能跑而在于能解释自己发现了什么。输出要结构化JSON异常要带证据建议动作要具体还要有 runbook 链接。落地时的关键三点输出必须是 JSON、证据要比结论详细、所有脚本必须进 Git 定时任务统一管理。做到这三点巡检脚本才是运维的助手做不到就只是另一个 outputs 0/1的黑盒程序。
