第325篇：配置变更审批与回滚流程自动化

关键词：变更审批、工作流引擎、自动回滚、变更窗口、Risk Assessment、ITIL、审批流、自动化

一、引言

网络变更是运维工作中风险最高的操作之一。据统计，超过 70% 的网络故障由配置变更引起。传统的变更审批流程依赖邮件或工单系统，审批周期长、信息传递不完整、回滚依赖手工执行。

将变更审批与回滚流程自动化，意味着从"变更申请"到"变更验证"的每个环节都由系统自动推进，人工仅参与决策点（审批/拒绝/放行），执行点（部署/回滚/验证）完全由自动化引擎完成。

二、变更审批工作流设计

2.1 变更分类与审批矩阵

2.2 审批流程状态机

stateDiagram-v2
    [*] --> Draft: 工程师创建变更申请
    Draft --> Submitted: 提交审批
    Submitted --> TechReview: 技术评审
    TechReview --> Rejected: 技术评审不通过
    TechReview --> MgmtApproval: 通过 → 管理层审批
    MgmtApproval --> Rejected
    MgmtApproval --> Approved: 审批通过
    Approved --> Scheduled: 排入变更窗口
    Scheduled --> InProgress: 变更窗口到达，开始执行
    InProgress --> Success: 验证通过
    InProgress --> Rollback: 验证失败/超时
    Rollback --> RollbackSuccess: 回滚成功
    Rollback --> RollbackFailed: 回滚失败 → 紧急升级
    RollbackSuccess --> Completed
    RollbackFailed --> Escalation
    Success --> Completed: 变更后观察期结束
    Rejected --> [*]: 流程终止
    Completed --> [*]: 归档
    Escalation --> [*]: 人工介入后归档

2.3 变更申请单数据结构

# change_request.yaml
change_request:
  id: CHG-2026-0901-001
  title: "PE-BJ 区域 BGP Timer 参数调整"
  type: P2                    # 变更类型
  risk_level: low             # 风险等级
  requestor: zhangsan         # 申请人
  team: network_core          # 所属团队

  description: |
    将 PE-BJ 区域所有 PE 设备的 BGP keepalive timer
    从 30s 调整为 10s，holdtime 从 90s 调整为 30s，
    以加快 BGP 会话故障检测速度。

  scope:
    devices:
      - PE-BJ-01
      - PE-BJ-02
      - PE-BJ-03
    config_files:
      - configs/production/pe-routers/PE-BJ-01.yaml
      - configs/production/pe-routers/PE-BJ-02.yaml
      - configs/production/pe-routers/PE-BJ-03.yaml

  change_window:
    start: "2026-09-05T02:00:00+08:00"
    end: "2026-09-05T04:00:00+08:00"

  rollback_plan:
    method: "config_revert"
    trigger: "auto"           # auto | manual
    timeout: 300              # 验证超时 300s 后自动回滚
    rollback_files:
      - backups/PE-BJ-01.pre-chg.cfg
      - backups/PE-BJ-02.pre-chg.cfg
      - backups/PE-BJ-03.pre-chg.cfg

  approval_chain:
    - role: tech_lead
      approver: lisi
      status: pending
    - role: change_manager
      approver: wangwu
      status: pending

  health_checks:
    - name: "BGP Session Count"
      command: "display bgp peer all | include Established"
      expected: ">= 4"
    - name: "Interface Error Rate"
      command: "display interface brief | include Error"
      expected: "0"

三、审批自动化引擎实现

3.1 基于 Python 的审批引擎

# scripts/change_workflow.py
import yaml
import json
import hashlib
from datetime import datetime, timedelta
from enum import Enum

class ChangeStatus(Enum):
    DRAFT = "draft"
    SUBMITTED = "submitted"
    TECH_REVIEW = "tech_review"
    MGMT_APPROVAL = "mgmt_approval"
    APPROVED = "approved"
    SCHEDULED = "scheduled"
    IN_PROGRESS = "in_progress"
    SUCCESS = "success"
    ROLLBACK = "rollback"
    COMPLETED = "completed"
    REJECTED = "rejected"

class ChangeWorkflow:
    def __init__(self, cr_file):
        with open(cr_file) as f:
            self.cr = yaml.safe_load(f)['change_request']
        self.status = ChangeStatus.DRAFT
        self.history = []
        self.backup_created = False

    def log_event(self, action, user, detail=""):
        """记录审批日志"""
        self.history.append({
            'timestamp': datetime.now().isoformat(),
            'action': action,
            'user': user,
            'detail': detail
        })

    def submit(self, user):
        """提交变更申请"""
        # 自动检查：配置 diff 是否已生成
        diff_hash = self.generate_diff_hash()
        if not diff_hash:
            raise ValueError("无法生成配置 diff，请检查 config_files")

        self.status = ChangeStatus.SUBMITTED
        self.log_event("submit", user, f"diff_hash={diff_hash}")
        return self.status

    def approve(self, approver_role, user, comment=""):
        """审批动作"""
        # 验证审批人身份
        expected_role = self._get_next_approver_role()
        if approver_role != expected_role:
            raise PermissionError(
                f"当前需要 {expected_role} 审批，而非 {approver_role}")

        # 更新审批状态
        for step in self.cr['approval_chain']:
            if step['role'] == approver_role:
                step['status'] = 'approved'
                step['approver'] = user
                step['comment'] = comment
                step['approved_at'] = datetime.now().isoformat()
                break

        # 推进到下一状态
        if self.status == ChangeStatus.TECH_REVIEW:
            # 检查是否还有管理审批
            remaining = [s for s in self.cr['approval_chain']
                        if s['status'] == 'pending' and
                        s['role'] == 'change_manager']
            if remaining:
                self.status = ChangeStatus.MGMT_APPROVAL
            else:
                self.status = ChangeStatus.APPROVED

        elif self.status == ChangeStatus.MGMT_APPROVAL:
            self.status = ChangeStatus.APPROVED

        self.log_event("approve", user, comment)
        return self.status

    def reject(self, approver_role, user, reason):
        """拒绝变更"""
        self.status = ChangeStatus.REJECTED
        self.log_event("reject", user, reason)
        return self.status

    def execute_change(self, user):
        """执行变更"""
        if self.status != ChangeStatus.APPROVED:
            raise ValueError(f"变更状态为 {self.status.value}，无法执行")

        # 1. 创建变更前备份
        self.create_pre_change_backup()
        self.backup_created = True

        # 2. 状态变更
        self.status = ChangeStatus.IN_PROGRESS
        self.log_event("execute_start", user)

        # 3. 执行配置下发（实际调用 Ansible/Netmiko）
        deploy_result = self._deploy_config()

        if not deploy_result['success']:
            self.trigger_rollback("部署失败")
            return self.status

        # 4. 执行健康检查
        health_result = self._run_health_checks()

        if not health_result['healthy']:
            self.trigger_rollback(f"健康检查失败: {health_result['failed_checks']}")
            return self.status

        # 5. 进入观察期
        self.status = ChangeStatus.SUCCESS
        self.log_event("execute_success", user)
        return self.status

    def trigger_rollback(self, reason):
        """触发回滚"""
        self.status = ChangeStatus.ROLLBACK
        self.log_event("rollback_trigger", "system", reason)

        if not self.backup_created:
            self.log_event("rollback_error", "system", "未找到变更前备份")
            return self.status

        rollback_result = self._rollback_config()

        if rollback_result['success']:
            self.log_event("rollback_success", "system")
        else:
            self.log_event("rollback_failed", "system",
                          f"回滚失败: {rollback_result.get('error')}")

        return self.status

    def generate_diff_hash(self):
        """生成配置 diff 的哈希值"""
        import difflib
        diff_content = ""
        for config_file in self.cr['scope']['config_files']:
            # 读取 Git 仓库中的当前版本与新版本差异
            diff_content += f.read(config_file)
        return hashlib.sha256(diff_content.encode()).hexdigest()

    def _get_next_approver_role(self):
        """获取下一个审批角色"""
        for step in self.cr['approval_chain']:
            if step['status'] == 'pending':
                return step['role']
        return None

    def create_pre_change_backup(self):
        """创建变更前配置备份"""
        for device in self.cr['scope']['devices']:
            backup_path = f"backups/{device}.pre-chg-{self.cr['id']}.cfg"
            # 通过 SSH/SNMP 从设备导出运行配置
            self._backup_device_config(device, backup_path)

    def get_audit_trail(self):
        """获取完整审计日志"""
        return {
            'change_id': self.cr['id'],
            'final_status': self.status.value,
            'timeline': self.history,
            'approvals': self.cr['approval_chain']
        }

3.2 审批流程 API（RESTful）

# api/change_api.py
from flask import Flask, request, jsonify
import yaml

app = Flask(__name__)
changes = {}  # 实际生产中应使用数据库

@app.route('/api/changes', methods=['POST'])
def create_change():
    """创建变更申请"""
    data = request.json
    cr_id = data.get('id', generate_change_id())
    workflow = ChangeWorkflow(f"changes/{cr_id}.yaml")
    changes[cr_id] = workflow
    return jsonify({'id': cr_id, 'status': 'draft'}), 201

@app.route('/api/changes/<cr_id>/submit', methods=['POST'])
def submit_change(cr_id):
    """提交变更"""
    workflow = changes[cr_id]
    user = request.json.get('user')
    status = workflow.submit(user)
    return jsonify({'id': cr_id, 'status': status.value})

@app.route('/api/changes/<cr_id>/approve', methods=['POST'])
def approve_change(cr_id):
    """审批变更"""
    workflow = changes[cr_id]
    data = request.json
    status = workflow.approve(
        data['approver_role'], data['user'], data.get('comment'))
    return jsonify({'id': cr_id, 'status': status.value})

@app.route('/api/changes/<cr_id>/execute', methods=['POST'])
def execute_change(cr_id):
    """执行变更"""
    workflow = changes[cr_id]
    status = workflow.execute_change(request.json.get('user'))
    return jsonify({'id': cr_id, 'status': status.value})

@app.route('/api/changes/<cr_id>/audit', methods=['GET'])
def get_audit(cr_id):
    """获取审计日志"""
    workflow = changes[cr_id]
    return jsonify(workflow.get_audit_trail())

四、自动回滚策略

4.1 回滚触发条件

4.2 三层回滚机制

# scripts/three_tier_rollback.py
def three_tier_rollback(change_id):
    """三层回滚机制"""

    # 第一层：配置级回滚（秒级）
    # 直接恢复变更前备份的配置
    tier1_result = config_rollback(change_id)
    if tier1_result['success']:
        return {'tier': 1, 'status': 'success', **tier1_result}

    # 第二层：命令级回滚（分钟级）
    # 逐条反向执行变更命令
    tier2_result = command_rollback(change_id)
    if tier2_result['success']:
        return {'tier': 2, 'status': 'success', **tier2_result}

    # 第三层：整机级回滚（小时级）
    # 重新加载出厂配置 + 基础配置模板
    tier3_result = device_reset(change_id)
    return {'tier': 3, 'status': tier3_result.get('success')}

4.3 变更窗口管理

# change_windows.yaml
maintenance_windows:
  - name: "周日凌晨维护窗口"
    schedule: "0 2 * * 0"      # 每周日凌晨 2:00
    duration: 120               # 分钟
    max_changes: 5              # 同一窗口最多 5 个变更
    auto_approve_types: [P4]    # P4 标准变更自动审批

  - name: "月度变更窗口"
    schedule: "0 0 1 * *"       # 每月 1 日午夜
    duration: 480               # 分钟
    max_changes: 20
    requires_approval: true

五、多线程设备信息采集与定时调度

5.1 多线程设备信息采集脚本

# scripts/multi_thread_collector.py
"""
多线程设备信息采集脚本
功能：并发采集设备运行信息、支持采集模板、自动重试、超时控制
输出：CSV / JSON / 数据库
"""
import csv
import json
import logging
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from datetime import datetime
from netmiko import ConnectHandler
from typing import Dict, List, Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)


@dataclass
class DeviceInfo:
    """设备采集信息模型"""
    device_name: str
    ip_address: str
    platform: str = ""
    uptime: str = ""
    cpu_usage: float = 0.0
    memory_usage: float = 0.0
    bgp_sessions: int = 0
    ospf_neighbors: int = 0
    interface_count: int = 0
    up_interfaces: int = 0
    down_interfaces: int = 0
    temperature: float = 0.0
    fan_status: str = ""
    alerts: List[str] = field(default_factory=list)
    collection_time: str = field(default_factory=lambda: datetime.now().isoformat())
    errors: List[str] = field(default_factory=list)


# 采集模板：定义不同角色的采集命令
COLLECTION_TEMPLATES = {
    'pe_router': {
        'commands': [
            ('display version', 'version_info'),
            ('display uptime', 'uptime'),
            ('display cpu-usage', 'cpu'),
            ('display memory-usage', 'memory'),
            ('display bgp peer all | include Established', 'bgp_count'),
            ('display ospf peer | include Full', 'ospf_count'),
            ('display interface brief', 'interface_brief'),
            ('display temperature', 'temperature'),
            ('display fan', 'fan_status'),
            ('display alarm active', 'active_alarms'),
        ]
    },
    'core_router': {
        'commands': [
            ('display version', 'version_info'),
            ('display uptime', 'uptime'),
            ('display cpu-usage', 'cpu'),
            ('display memory-usage', 'memory'),
            ('display bgp peer all', 'bgp_all'),
            ('display isis peer', 'isis_peer'),
            ('display interface brief', 'interface_brief'),
            ('display mpls lsp', 'mpls_lsp'),
        ]
    },
    'access_switch': {
        'commands': [
            ('display version', 'version_info'),
            ('display cpu-usage', 'cpu'),
            ('display memory-usage', 'memory'),
            ('display interface brief', 'interface_brief'),
            ('display stp brief', 'stp_brief'),
            ('display vlan', 'vlan_list'),
        ]
    }
}


def parse_cpu_usage(output: str) -> float:
    """从命令输出中解析 CPU 利用率"""
    for line in output.split('\n'):
        if 'CPU' in line.upper() and '%' in line:
            parts = line.split('%')
            try:
                return float(parts[0].split()[-1])
            except (ValueError, IndexError):
                pass
    return 0.0


def parse_memory_usage(output: str) -> float:
    """从命令输出中解析内存利用率"""
    for line in output.split('\n'):
        if 'Memory' in line.upper() or '内存' in line:
            parts = line.split('%')
            try:
                return float(parts[0].split()[-1])
            except (ValueError, IndexError):
                pass
    return 0.0


def parse_interface_stats(output: str) -> Dict[str, int]:
    """解析接口统计：总数、Up数、Down数"""
    stats = {'total': 0, 'up': 0, 'down': 0}
    for line in output.split('\n'):
        parts = line.split()
        if len(parts) >= 3 and parts[0] not in ('Interface', '-------------------------------------------------------------------'):
            stats['total'] += 1
            status = parts[-1] if parts else ''
            if 'up' in status.lower():
                stats['up'] += 1
            elif 'down' in status.lower():
                stats['down'] += 1
    return stats


def collect_single_device(device: Dict, template_name: str = 'pe_router') -> DeviceInfo:
    """采集单台设备信息"""
    info = DeviceInfo(
        device_name=device['name'],
        ip_address=device['ip'],
        platform=device.get('platform', 'unknown')
    )

    template = COLLECTION_TEMPLATES.get(template_name, COLLECTION_TEMPLATES['pe_router'])
    commands = template['commands']

    try:
        with ConnectHandler(
            device_type=device.get('platform', 'huawei_vrpv8'),
            host=device['ip'],
            username=device.get('username', 'admin'),
            password=device.get('password', ''),
            timeout=30,
            session_timeout=60,
        ) as net:
            for cmd, key in commands:
                try:
                    output = net.send_command(cmd, read_timeout=30)
                    if key == 'cpu':
                        info.cpu_usage = parse_cpu_usage(output)
                    elif key == 'memory':
                        info.memory_usage = parse_memory_usage(output)
                    elif key in ('bgp_count', 'bgp_all'):
                        info.bgp_sessions = sum(1 for l in output.split('\n') if 'Established' in l)
                    elif key == 'interface_brief':
                        stats = parse_interface_stats(output)
                        info.interface_count = stats['total']
                        info.up_interfaces = stats['up']
                        info.down_interfaces = stats['down']
                    elif key == 'uptime':
                        for line in output.split('\n'):
                            if 'uptime' in line.lower():
                                info.uptime = line.strip()
                                break
                except Exception as cmd_err:
                    info.errors.append(f"命令 '{cmd}' 失败: {str(cmd_err)}")
    except Exception as conn_err:
        info.errors.append(f"连接失败: {str(conn_err)}")
        logger.error(f"设备 {device['name']} 连接失败: {conn_err}")
    return info


def collect_batch(devices: List[Dict], template: str = 'pe_router',
                  max_workers: int = 20) -> List[DeviceInfo]:
    """多线程批量采集"""
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(collect_single_device, d, template): d['name']
            for d in devices
        }
        for future in as_completed(futures):
            try:
                results.append(future.result())
            except Exception as e:
                logger.error(f"采集异常: {e}")
    return results


def save_results(results: List[DeviceInfo], output_dir: str = './output',
                 fmt: str = 'csv'):
    """保存采集结果"""
    import os
    os.makedirs(output_dir, exist_ok=True)
    ts = datetime.now().strftime('%Y%m%d_%H%M%S')

    if fmt == 'csv':
        filename = os.path.join(output_dir, f'device_info_{ts}.csv')
        flat = [asdict(r) for r in results]
        for f in flat:
            f.pop('alerts', None)
            f.pop('errors', None)
        if flat:
            df = pd.DataFrame(flat)
            df.to_csv(filename, index=False, encoding='utf-8-sig')
    elif fmt == 'json':
        filename = os.path.join(output_dir, f'device_info_{ts}.json')
        with open(filename, 'w', encoding='utf-8') as fh:
            json.dump([asdict(r) for r in results], fh, indent=2, ensure_ascii=False, default=str)
    logger.info(f"结果已保存: {filename}")

5.2 数据采集模板

# collection_templates.yaml
# 定义不同设备角色的采集命令模板

templates:
  pe_router:
    description: "PE 路由器全面采集"
    commands:
      - cmd: display version
        key: version_info
        parse: text
      - cmd: display cpu-usage
        key: cpu
        parse: percentage
        threshold:
          warning: 70
          critical: 85
      - cmd: display memory-usage
        key: memory
        parse: percentage
        threshold:
          warning: 75
          critical: 90
      - cmd: display bgp peer all | include Established
        key: bgp_sessions
        parse: line_count
      - cmd: display ospf peer | include Full
        key: ospf_neighbors
        parse: line_count
      - cmd: display interface brief
        key: interfaces
        parse: table

  core_router:
    description: "核心路由器全面采集"
    commands:
      - cmd: display cpu-usage
        key: cpu
        parse: percentage
      - cmd: display memory-usage
        key: memory
        parse: percentage
      - cmd: display bgp peer all
        key: bgp_all
        parse: table
      - cmd: display isis peer
        key: isis_peers
        parse: table
      - cmd: display mpls lsp
        key: mpls_lsp
        parse: table
      - cmd: display lsp-all
        key: lsp_all
        parse: count

  access_switch:
    description: "接入交换机基础采集"
    commands:
      - cmd: display cpu-usage
        key: cpu
        parse: percentage
      - cmd: display memory-usage
        key: memory
        parse: percentage
      - cmd: display interface brief
        key: interfaces
        parse: table
      - cmd: display stp brief
        key: stp_status
        parse: table

  firewall:
    description: "防火墙安全采集"
    commands:
      - cmd: display cpu usage
        key: cpu
        parse: percentage
      - cmd: display session statistics
        key: sessions
        parse: number
      - cmd: display security-policy statistics
        key: policy_hits
        parse: table
      - cmd: display zone
        key: zones
        parse: list

5.3 数据分析和可视化代码

# scripts/data_analysis.py
"""
采集数据分析与可视化
功能：设备健康评分、趋势分析、异常检测、报表生成
"""
import json
import pandas as pd
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
from typing import List, Dict
from datetime import datetime


def health_score(device: Dict) -> float:
    """计算设备健康评分（0-100）"""
    score = 100.0
    cpu = device.get('cpu_usage', 0)
    mem = device.get('memory_usage', 0)

    if cpu > 80: score -= 30
    elif cpu > 60: score -= 15
    elif cpu > 40: score -= 5

    if mem > 85: score -= 30
    elif mem > 70: score -= 15
    elif mem > 50: score -= 5

    total = max(device.get('interface_count', 1), 1)
    down = device.get('down_interfaces', 0)
    down_rate = down / total
    if down_rate > 0.2: score -= 25
    elif down_rate > 0.1: score -= 15
    elif down_rate > 0.05: score -= 5

    alerts = len(device.get('alerts', []))
    score -= min(alerts * 5, 20)

    return max(score, 0)


def analyze_health(results: List[Dict]) -> Dict:
    """分析所有设备健康状态"""
    df = pd.DataFrame(results)
    df['health_score'] = df.apply(health_score, axis=1)

    return {
        'total_devices': len(df),
        'healthy_count': int((df['health_score'] >= 80).sum()),
        'warning_count': int(((df['health_score'] >= 50) & (df['health_score'] < 80)).sum()),
        'critical_count': int((df['health_score'] < 50).sum()),
        'avg_cpu': float(df['cpu_usage'].mean()),
        'avg_memory': float(df['memory_usage'].mean()),
        'total_down_interfaces': int(df['down_interfaces'].sum()),
    }


def generate_visualization(df: pd.DataFrame, output_dir: str = './output'):
    """生成可视化图表"""
    import os
    os.makedirs(output_dir, exist_ok=True)

    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
    plt.rcParams['axes.unicode_minus'] = False

    # CPU 分布
    axes[0, 0].hist(df['cpu_usage'].dropna(), bins=20, color='#4CAF50', edgecolor='black', alpha=0.7)
    axes[0, 0].set_title('CPU 使用率分布')
    axes[0, 0].axvline(80, color='red', linestyle='--', label='告警阈值 80%')
    axes[0, 0].legend()

    # 内存分布
    axes[0, 1].hist(df['memory_usage'].dropna(), bins=20, color='#2196F3', edgecolor='black', alpha=0.7)
    axes[0, 1].set_title('内存使用率分布')
    axes[0, 1].axvline(85, color='red', linestyle='--', label='告警阈值 85%')
    axes[0, 1].legend()

    # 健康评分排名
    df['health_score'] = df.apply(health_score, axis=1)
    bottom_10 = df.nsmallest(10, 'health_score')
    colors = ['red' if s < 50 else 'orange' if s < 80 else 'green' for s in bottom_10['health_score']]
    axes[1, 0].barh(bottom_10['device_name'], bottom_10['health_score'], color=colors)
    axes[1, 0].set_title('健康评分最低的 10 台设备')
    axes[1, 0].set_xlim(0, 100)

    # 接口状态饼图
    up = int(df['up_interfaces'].sum())
    down = int(df['down_interfaces'].sum())
    axes[1, 1].pie([up, down], labels=['Up', 'Down'], colors=['#4CAF50', '#F44336'], autopct='%1.1f%%')
    axes[1, 1].set_title('接口状态总览')

    plt.tight_layout()
    plt.savefig(os.path.join(output_dir, 'health_dashboard.png'), dpi=150)


def generate_report(df: pd.DataFrame, output_path: str):
    """生成 Markdown 汇总报告"""
    summary = analyze_health(df.to_dict('records'))
    df['health_score'] = df.apply(health_score, axis=1)

    lines = [
        "# 设备采集分析报告",
        f"\n**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
        "",
        "## 概览统计", "",
        "| 指标 | 数值 |", "|------|------|",
        f"| 设备总数 | {summary['total_devices']} |",
        f"| 健康设备 (≥80分) | {summary['healthy_count']} |",
        f"| 警告设备 (50-80分) | {summary['warning_count']} |",
        f"| 严重设备 (<50分) | {summary['critical_count']} |",
        f"| 平均 CPU 使用率 | {summary['avg_cpu']:.1f}% |",
        f"| 平均内存使用率 | {summary['avg_memory']:.1f}% |",
        "",
    ]

    critical = df[df['health_score'] < 80].sort_values('health_score')
    if not critical.empty:
        lines.extend(["## 需要关注的设备", "",
            "| 设备 | CPU% | MEM% | BGP会话 | 接口Down | 健康评分 |",
            "|------|------|------|---------|----------|----------|"])
        for _, row in critical.iterrows():
            lines.append(
                f"| {row['device_name']} | {row['cpu_usage']:.1f} | "
                f"{row['memory_usage']:.1f} | {row['bgp_sessions']} | "
                f"{row['down_interfaces']} | {row['health_score']:.0f} |")
    else:
        lines.append("所有设备健康评分均在正常范围内。")

    with open(output_path, 'w', encoding='utf-8') as f:
        f.write('\n'.join(lines))

5.4 定时任务调度配置

# scripts/scheduler.py
"""
定时采集任务调度器
支持：APScheduler 定时任务、cron 表达式、任务持久化
"""
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
import yaml
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


SCHEDULER_CONFIG = {
    'tasks': [
        {
            'name': '每小时设备状态采集',
            'trigger': 'cron',
            'cron': {'hour': '*'},
            'devices_file': 'inventory/devices.yaml',
            'template': 'pe_router',
            'max_workers': 20,
            'output_dir': './output',
            'format': 'csv',
        },
        {
            'name': '每日全面采集',
            'trigger': 'cron',
            'cron': {'hour': 2, 'minute': 0},
            'devices_file': 'inventory/all_devices.yaml',
            'template': 'core_router',
            'max_workers': 30,
            'output_dir': './output/daily',
            'format': 'json',
        },
        {
            'name': '每5分钟核心设备监控',
            'trigger': 'interval',
            'interval_args': {'minutes': 5},
            'devices_file': 'inventory/core_devices.yaml',
            'template': 'core_router',
            'max_workers': 10,
            'output_dir': './output/monitoring',
            'format': 'json',
        },
    ]
}


def run_collection_task(task_config: dict):
    """执行采集任务"""
    from multi_thread_collector import collect_batch, save_results
    logger.info(f"开始执行采集任务: {task_config['name']}")

    with open(task_config['devices_file'], 'r', encoding='utf-8') as f:
        devices = yaml.safe_load(f)['devices']

    results = collect_batch(
        devices=devices,
        template=task_config.get('template', 'pe_router'),
        max_workers=task_config.get('max_workers', 20)
    )
    save_results(results, task_config.get('output_dir', './output'),
                 task_config.get('format', 'csv'))
    logger.info(f"采集完成: 共 {len(results)} 台设备")


def main():
    scheduler = BlockingScheduler()
    for task in SCHEDULER_CONFIG['tasks']:
        if task['trigger'] == 'cron':
            trigger = CronTrigger(**task['cron'])
        elif task['trigger'] == 'interval':
            trigger = IntervalTrigger(**task['interval_args'])
        else:
            continue
        scheduler.add_job(
            run_collection_task, trigger=trigger, args=[task],
            id=task['name'], max_instances=1, misfire_grace_time=300)
        logger.info(f"已注册定时任务: {task['name']}")
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()


# crontab 替代方案：
# */5 * * * * /usr/bin/python3 /opt/scripts/monitoring_collector.py
# 0 * * * *   /usr/bin/python3 /opt/scripts/hourly_collector.py
# 0 2 * * *   /usr/bin/python3 /opt/scripts/daily_collector.py

# scheduler_config.yaml
tasks:
  - name: "每小时设备状态采集"
    trigger: cron
    cron: { hour: "*" }
    devices_file: inventory/devices.yaml
    template: pe_router
    max_workers: 20
    format: csv
    alert_thresholds:
      cpu_warning: 70
      cpu_critical: 85
      memory_warning: 75
      memory_critical: 90

  - name: "每日全面采集"
    trigger: cron
    cron: { hour: 2, minute: 0 }
    devices_file: inventory/all_devices.yaml
    template: core_router
    max_workers: 30
    format: json
    generate_report: true
    send_email:
      recipients: ["network-team@example.com"]

六、总结

participant Eng as 工程师
participant System as 审批系统
participant Reviewer as 审批人
participant Auto as 自动化引擎
participant Device as 网络设备

Eng->>System: 提交变更申请
System->>Reviewer: 通知技术评审
Reviewer->>System: 审批通过
System->>Auto: 排入变更窗口
Auto->>Device: 变更前备份配置
Auto->>Device: 下发新配置
Auto->>Auto: 执行健康检查
alt 健康检查通过
    Auto->>Auto: 进入观察期
    Auto->>Eng: 变更成功通知
else 健康检查失败
    Auto->>Device: 自动回滚配置
    Auto->>Eng: 回滚告警通知
end

```

故障案例：自动化回滚拯救核心变更

某电信运营商在月度变更窗口中对核心路由器执行 BGP 路由策略更新。变更执行到第 3 台设备时，系统检测到该设备的 BGP 会话数从 12 降至 2。自动回滚机制立即触发：

1. 秒级响应：第 3 台设备自动回滚到变更前配置，BGP 会话恢复正常
2. 批次冻结：后续 17 台设备未执行变更
3. 根因分析：回滚后对比 diff，发现路由策略中的 deny 规则优先级设置错误
4. 修复后重提：修正路由策略后重新提交变更申请，经额外一轮技术评审后成功部署

思考题

1. 在设计变更审批流程时，如何平衡"审批效率"与"审批严格度"？标准变更（P4）的预审批机制是否存在安全隐患？
2. 自动回滚是否总是最优选择？在什么场景下，自动回滚可能导致更严重的后果？

下篇预告：第326篇《自动化测试：用 pytest 验证网络状态》——用软件测试的方法验证网络配置是否正确生效，确保变更质量。