第325篇：网络变更自动化与回滚
关键词
网络变更、变更自动化、配置部署、回滚机制、变更管理、变更审批、批量变更、安全变更
一、网络变更的挑战
1.1 变更为什么总让人紧张
网络变更的"恐惧因素"：
范围大：
┌─ 一次变更涉及多台设备
├─ 配置间存在隐式依赖
├─ 影响面难以准确评估
└─ 出问题可能影响全网
不可逆风险：
┌─ "commit" 后设备配置被覆盖
├─ 回滚也可能引入新问题
├─ 配置无法精确恢复到前一秒
└─ 部分命令（如 format）不可逆
人工操作失误：
┌─ 敲错命令（VLAN 100 打成 VLAN 1000）
├─ 敲错设备（在核心上配了接入的配置）
├─ 漏配某个参数
└─ 顺序错误（未配 VLAN 先配接口）
缺乏验证：
┌─ 变更是"盲操作"
├─ 无法在变更前 100% 模拟效果
├─ 变更后靠人工逐条验证
└─ 问题可能在数小时或数天后才暴露
1.2 自动化变更的目标
自动化变更的四个目标：
1. 可预测（Predictable）
┌─ 同样输入 = 同样输出
├─ 配置通过模板预生成
├─ 变更前可预览完整配置
└─ 有自动化检查环节把关
2. 可回滚（Reversible）
┌─ 每个变更都有对应的回滚方案
├─ 回滚配置预先生成
├─ 一键执行回滚
└─ 回滚后自动验证
3. 可追溯（Auditable）
┌─ 谁在何时改了什么都记录
├─ 变更原因关联工单号
├─ 变更前后的配置快照保存
└─ 变更结果记录到 CMDB
4. 可验证（Verifiable）
┌─ 变更中验证（每条命令执行后）
├─ 变更后验证（全量检查）
├─ 业务级验证（ping/trace）
└─ 持续验证（变更后一段时间内）
二、自动化变更架构
2.1 变更流水线
自动化变更流水线：
┌──────────────────────────────────────────┐
│ 阶段 1：变更请求（Change Request） │
│ ┌─ 提交变更工单 │
│ ├─ 描述变更内容和目的 │
│ ├─ 关联故障/需求编号 │
│ └─ 指定变更窗口 │
│ │
│ 阶段 2：变更方案生成（Plan） │
│ ┌─ 基于模板生成配置 │
│ ├─ 自动生成变更脚本 │
│ ├─ 生成回滚脚本 │
│ └─ 自动评估影响范围 │
│ │
│ 阶段 3：变更审批（Review） │
│ ┌─ 变更方案自动检查 │
│ ├─ 合规规则检查 │
│ ├─ 冲突检测 │
│ └─ 人工审批（Peer Review） │
│ │
│ 阶段 4：变更实施（Execute） │
│ ┌─ 变更前快照 │
│ ├─ 逐设备/逐命令执行 │
│ ├─ 实时验证（命令级） │
│ └─ 变更后快照 │
│ │
│ 阶段 5：变更验证（Verify） │
│ ┌─ 配置验证（对比期望） │
│ ├─ 状态验证（邻居/路由/接口） │
│ ├─ 业务验证（连通性/延迟） │
│ └─ CI 测试结果检查 │
│ │
│ 阶段 6：变更关闭（Close） │
│ ┌─ 更新 CMDB │
│ ├─ 标记变更完成 │
│ ├─ 发送变更报告 │
│ └─ 知识沉淀（变更总结） │
└──────────────────────────────────────────┘
2.2 变更数据模型
#!/usr/bin/env python3
# change_model.py — 变更数据模型
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
from enum import Enum
from datetime import datetime
class ChangeStatus(Enum):
"""变更状态"""
DRAFT = "草稿"
PENDING_REVIEW = "待审批"
APPROVED = "已批准"
REJECTED = "已拒绝"
IN_PROGRESS = "执行中"
SUCCESS = "成功"
FAILED = "失败"
ROLLED_BACK = "已回滚"
CANCELLED = "已取消"
class ChangeType(Enum):
"""变更类型"""
CONFIG_MODIFY = "配置修改"
SW_UPGRADE = "软件升级"
HW_REPLACEMENT = "硬件替换"
NETWORK_OPTIMIZE = "网络优化"
EMERGENCY_FIX = "紧急修复"
class DeviceConfig:
"""设备配置快照"""
def __init__(
self,
device_name: str,
commands: List[str],
rollback_commands: Optional[List[str]] = None,
verification: Optional[Dict] = None,
):
self.device_name = device_name
self.commands = commands
self.rollback_commands = rollback_commands or []
self.verification = verification or {}
self.pre_change_snapshot: str = ""
self.post_change_snapshot: str = ""
self.status: str = "pending"
@dataclass
class ChangeRequest:
"""变更请求"""
id: str # CR-20240101-001
title: str # 变更标题
change_type: ChangeType # 变更类型
description: str # 变更描述
reason: str # 变更原因
ticket_id: str # 关联工单号
requester: str # 申请人
reviewer: str = "" # 审批人
implementer: str = "" # 实施人
scheduled_time: Optional[str] = None # 计划变更时间
window_start: str = "02:00" # 变更窗口开始
window_end: str = "06:00" # 变更窗口结束
status: ChangeStatus = ChangeStatus.DRAFT
devices: List[DeviceConfig] = field(default_factory=list)
impact_scope: str = "" # 影响范围
rollback_plan: str = "" # 回滚方案
created_at: str = field(
default_factory=lambda: datetime.now().isoformat()
)
risk_level: str = "LOW" # LOW/MEDIUM/HIGH/CRITICAL
tags: List[str] = field(default_factory=list)
def add_device_config(
self,
device_name: str,
commands: List[str],
rollback_commands: Optional[List[str]] = None,
):
"""添加设备配置"""
self.devices.append(
DeviceConfig(device_name, commands, rollback_commands)
)
三、变更引擎实现
3.1 核心变更引擎
#!/usr/bin/env python3
# change_engine.py — 变更执行引擎
from netmiko import (
ConnectHandler,
NetMikoTimeoutException,
NetMikoAuthenticationException,
)
from typing import List, Dict, Optional
from datetime import datetime
import time
import json
import logging
from change_model import ChangeRequest, DeviceConfig, ChangeStatus
logger = logging.getLogger(__name__)
class ChangeResult:
"""单设备变更结果"""
def __init__(self, device_name: str):
self.device_name = device_name
self.success = False
self.commands_executed: List[str] = []
self.commands_failed: List[Dict] = []
self.execution_log: List[Dict] = []
self.pre_snapshot: str = ""
self.post_snapshot: str = ""
self.rolled_back = False
self.error: str = ""
def log(self, action: str, detail: str, status: str = "info"):
self.execution_log.append({
"time": datetime.now().isoformat(),
"action": action,
"detail": detail,
"status": status,
})
class ChangeExecutor:
"""变更执行器"""
@staticmethod
def execute_change(
device_info: Dict,
change: DeviceConfig,
verify_after_each: bool = True,
dry_run: bool = False,
) -> ChangeResult:
"""执行设备变更"""
result = ChangeResult(change.device_name)
try:
conn = ConnectHandler(**device_info)
conn.enable()
result.log("connect", "设备连接成功", "info")
# 1. 变更前快照
result.pre_snapshot = conn.send_command(
"display current-configuration"
)
result.log("pre_snapshot", "变更前快照已采集", "info")
if dry_run:
result.log("dry_run", "模拟执行模式，不下发配置", "info")
for cmd in change.commands:
result.commands_executed.append(f"[DRY-RUN] {cmd}")
result.success = True
conn.disconnect()
return result
# 2. 逐条执行配置命令
for i, cmd in enumerate(change.commands):
try:
output = conn.send_config_set(
[cmd], read_timeout=30
)
result.commands_executed.append(cmd)
result.log("command", f"执行: {cmd}", "success")
if "Error" in output:
result.commands_failed.append({
"command": cmd,
"output": output,
"index": i,
})
result.log(
"command_error",
f"命令执行错误: {output}",
"error",
)
# 执行后验证
if verify_after_each and change.verification:
verify_cmd = change.verification.get("command", "")
if verify_cmd:
time.sleep(2)
v_out = conn.send_command(verify_cmd)
result.log(
"verify",
f"验证: {verify_cmd} -> {v_out[:200]}",
"info",
)
except Exception as e:
result.commands_failed.append({
"command": cmd,
"error": str(e),
})
result.log(
"command_failed",
f"命令执行异常: {e}",
"error",
)
result.error = str(e)
conn.disconnect()
return result
# 3. 保存配置
conn.save_config()
result.log("save", "配置已保存", "info")
# 4. 变更后快照
result.post_snapshot = conn.send_command(
"display current-configuration"
)
result.log("post_snapshot", "变更后快照已采集", "info")
result.success = len(result.commands_failed) == 0
conn.disconnect()
except NetMikoTimeoutException as e:
result.error = f"设备连接超时: {e}"
result.log("error", result.error, "error")
except NetMikoAuthenticationException as e:
result.error = f"设备认证失败: {e}"
result.log("error", result.error, "error")
except Exception as e:
result.error = f"变更异常: {e}"
result.log("error", result.error, "error")
return result
@staticmethod
def rollback_change(
device_info: Dict,
change: DeviceConfig,
change_result: ChangeResult,
) -> ChangeResult:
"""回滚变更"""
rollback_result = ChangeResult(change.device_name)
try:
conn = ConnectHandler(**device_info)
conn.enable()
for cmd in change.rollback_commands:
try:
output = conn.send_config_set(
[cmd], read_timeout=30
)
rollback_result.commands_executed.append(cmd)
rollback_result.log(
"rollback", f"回滚命令: {cmd}", "success"
)
except Exception as e:
rollback_result.log(
"rollback_failed",
f"回滚命令失败: {cmd} - {e}",
"error",
)
conn.save_config()
rollback_result.success = True
rollback_result.rolled_back = True
change_result.rolled_back = True
conn.disconnect()
except Exception as e:
rollback_result.error = f"回滚失败: {e}"
rollback_result.log("error", rollback_result.error, "error")
return rollback_result
class BatchChangeExecutor:
"""批量变更执行器"""
def __init__(
self,
change_request: ChangeRequest,
devices_info: Dict[str, Dict],
):
self.change_request = change_request
self.devices_info = devices_info
self.results: Dict[str, ChangeResult] = {}
def execute_all(
self,
serial: bool = False,
verify_after_each: bool = True,
dry_run: bool = False,
) -> Dict[str, ChangeResult]:
"""执行所有设备变更"""
self.change_request.status = ChangeStatus.IN_PROGRESS
if serial:
# 串行执行（安全模式）
for device_config in self.change_request.devices:
dev_name = device_config.device_name
if dev_name not in self.devices_info:
self.results[dev_name] = ChangeResult(dev_name)
self.results[dev_name].error = "设备信息缺失"
continue
result = ChangeExecutor.execute_change(
self.devices_info[dev_name],
device_config,
verify_after_each,
dry_run,
)
self.results[dev_name] = result
# 变更失败自动回滚
if not result.success and not dry_run:
self._auto_rollback(dev_name)
else:
# 并行执行（快速模式）
from concurrent.futures import (
ThreadPoolExecutor, as_completed
)
with ThreadPoolExecutor(max_workers=5) as pool:
futures = {}
for device_config in self.change_request.devices:
dev_name = device_config.device_name
if dev_name not in self.devices_info:
continue
future = pool.submit(
ChangeExecutor.execute_change,
self.devices_info[dev_name],
device_config,
verify_after_each,
dry_run,
)
futures[future] = dev_name
for future in as_completed(futures):
dev_name = futures[future]
try:
self.results[dev_name] = future.result()
except Exception as e:
result = ChangeResult(dev_name)
result.error = f"并行执行异常: {e}"
self.results[dev_name] = result
all_success = all(r.success for r in self.results.values())
any_rolled_back = any(r.rolled_back for r in self.results.values())
if all_success:
self.change_request.status = ChangeStatus.SUCCESS
elif any_rolled_back:
self.change_request.status = ChangeStatus.ROLLED_BACK
else:
self.change_request.status = ChangeStatus.FAILED
return self.results
def _auto_rollback(self, device_name: str):
"""自动回滚失败的变更"""
logger.warning(f"{device_name} 变更失败，自动回滚...")
change = next(
(d for d in self.change_request.devices
if d.device_name == device_name),
None,
)
if not change or not change.rollback_commands:
logger.error(f"{device_name} 无回滚方案")
return
ChangeExecutor.rollback_change(
self.devices_info[device_name],
change,
self.results[device_name],
)
四、安全的回滚策略
4.1 回滚方案分级
回滚方案分级：
级别 1：命令级回滚（最细粒度）
原理：每条变更命令都有逆操作 示例： ┌─ 变更: port default vlan 100 └─ 回滚: undo port default vlan 适用：配置参数修改 风险：低（最安全）
级别 2：配置段回滚
原理：替换整个配置段 示例： ┌─ 保存变更前的接口配置段 └─ 回滚时整体替换 适用：接口/协议配置修改 风险：中
级别 3：全量配置回滚（最粗粒度）
原理：将整个 running-config 恢复到变更前 示例： ┌─ 变更前保存 startup-config └─ 回滚时 reload 设备 适用：重大变更的最后防线 风险：高（设备重启影响所有业务）
4.2 回滚策略选择
# rollback_strategy.py — 回滚策略选择
from typing import List, Dict, Optional
from dataclasses import dataclass
import json
class RollbackStrategy:
"""回滚策略"""
@staticmethod
def generate_command_rollback(
change_commands: List[str]
) -> List[str]:
"""生成命令级回滚（自动推导 undo 命令）"""
rollback = []
# 反转命令顺序（后配的先回滚）
for cmd in reversed(change_commands):
# 华为 VRP undo 推导
if cmd.startswith("interface "):
continue # 不做 undo interface
elif cmd.startswith("vlan batch "):
vlans = cmd.split("vlan batch ")[1]
rollback.append(f"undo vlan batch {vlans}")
elif cmd.startswith("port default vlan"):
rollback.append(cmd.replace(
"port default vlan",
"undo port default vlan"
))
elif cmd.startswith("ip address"):
rollback.append(f"undo {cmd}")
elif cmd.startswith("ospf"):
rollback.append(f"undo {cmd}")
elif cmd.startswith("bgp"):
rollback.append(f"undo {cmd}")
else:
rollback.append(f"undo {cmd}")
return rollback
@staticmethod
def generate_snapshot_based_rollback(
pre_change_config: str,
changed_sections: List[str],
) -> Dict:
"""基于快照的回滚方案"""
sections = {}
current_section = ""
for line in pre_change_config.splitlines():
if line.startswith("interface "):
current_section = line
sections[current_section] = []
elif line.startswith("bgp ") or line.startswith("ospf "):
current_section = line
sections[current_section] = []
elif current_section:
sections[current_section].append(line)
# 只返回变更部分的回滚
rollback_config = {}
for section in changed_sections:
if section in sections:
rollback_config[section] = sections[section]
return {
"method": "snapshot",
"affected_sections": changed_sections,
"rollback_sections": rollback_config,
}
4.3 回滚流程
标准回滚流程：
检测到失败：
- 停止当前变更（不再继续执行后续命令）
- 记录失败现场（保存当前配置和错误）
- 通知变更工程师
自动回滚（可选）：
- 检查是否有回滚方案
- 优先使用预先生成的回滚脚本
- 执行回滚（命令级 → 配置段级）
- 验证回滚效果
- 记录回滚结果
人工介入（自动回滚失败时）：
- 发送紧急通知（电话/短信）
- 提供现场配置信息和变更记录
- 高级工程师远程介入
- 必要时执行全量配置恢复（reload）
回滚后行动：
- 确认业务恢复
- 更新变更单状态（ROLLED_BACK）
- 通知相关方变更已被回滚
- 复盘分析根因，更新回滚方案
五、变更验证
5.1 分层次验证
#!/usr/bin/env python3
# change_verification.py — 变更验证
from netmiko import ConnectHandler
from typing import Dict, List, Optional
from datetime import datetime
import re
class ChangeVerifier:
"""变更验证器"""
@staticmethod
def verify_connectivity(
device_info: Dict,
targets: List[str],
count: int = 3,
) -> Dict:
"""验证连通性"""
results = {}
try:
conn = ConnectHandler(**device_info)
for target in targets:
output = conn.send_command(
f"ping {target} -c {count}"
)
success = "!" in output or "packet loss" not in output
results[target] = {
"reachable": success,
"output": output[:200],
}
conn.disconnect()
except Exception as e:
for target in targets:
results[target] = {
"reachable": False,
"error": str(e),
}
return results
@staticmethod
def verify_bgp_state(
device_info: Dict,
expected_peers: int,
) -> Dict:
"""验证 BGP 状态"""
try:
conn = ConnectHandler(**device_info)
output = conn.send_command("display bgp peer")
# 统计 Established 的 peer 数量
established = len(re.findall(
r"Established", output
))
idle = len(re.findall(r"Idle", output))
conn.disconnect()
return {
"passed": established == expected_peers,
"established": established,
"expected": expected_peers,
"idle": idle,
"has_issues": idle > 0,
}
except Exception as e:
return {
"passed": False,
"error": str(e),
}
@staticmethod
def verify_interface_state(
device_info: Dict,
interfaces: List[str],
) -> Dict:
"""验证接口状态"""
results = {}
try:
conn = ConnectHandler(**device_info)
output = conn.send_command("display interface brief")
for intf in interfaces:
for line in output.splitlines():
if intf in line:
up = "up" in line.lower()
results[intf] = {
"up": up,
"line": line.strip(),
}
break
else:
results[intf] = {
"up": False,
"error": "接口未找到",
}
conn.disconnect()
except Exception as e:
for intf in interfaces:
results[intf] = {"up": False, "error": str(e)}
return results
def verify_change(
self,
device_info: Dict,
expected_state: Dict,
) -> Dict:
"""综合验证"""
results = {
"timestamp": datetime.now().isoformat(),
"checks": {},
}
# 连通性验证
if "connectivity" in expected_state:
results["checks"]["connectivity"] = \
self.verify_connectivity(
device_info,
expected_state["connectivity"],
)
# BGP 状态验证
if "bgp_peers" in expected_state:
results["checks"]["bgp"] = self.verify_bgp_state(
device_info,
expected_state["bgp_peers"],
)
# 接口状态验证
if "interfaces" in expected_state:
results["checks"]["interfaces"] = \
self.verify_interface_state(
device_info,
expected_state["interfaces"],
)
# 总体判定
all_passed = all(
check.get("passed", True)
for check in results["checks"].values()
if isinstance(check, dict)
)
results["passed"] = all_passed
return results
六、变更报告与审计
6.1 变更报告生成
#!/usr/bin/env python3
# change_report.py — 变更报告生成
from typing import Dict, List
from datetime import datetime
import json
class ChangeReportGenerator:
"""变更报告生成器"""
@staticmethod
def generate_html_report(
change_request: "ChangeRequest",
results: Dict[str, "ChangeResult"],
output_file: str = "change_report.html",
):
"""生成 HTML 变更报告"""
total_devices = len(change_request.devices)
success_devices = sum(
1 for r in results.values() if r.success
)
failed_devices = sum(
1 for r in results.values()
if not r.success and not r.rolled_back
)
rolled_back_devices = sum(
1 for r in results.values() if r.rolled_back
)
html = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<title>变更执行报告</title>
<style>
body {{ font-family: Arial, sans-serif; margin: 20px; }}
.header {{ background: #1a237e; color: white; padding: 20px;
border-radius: 8px; margin-bottom: 20px; }}
.success {{ color: #4CAF50; }} .failed {{ color: #f44336; }}
.warning {{ color: #ff9800; }}
table {{ width: 100%; border-collapse: collapse; }}
th, td {{ border: 1px solid #ddd; padding: 10px; text-align: left; }}
th {{ background: #283593; color: white; }}
.log-info {{ color: #2196F3; }}
.log-error {{ color: #f44336; }}
.log-success {{ color: #4CAF50; }}
</style>
</head>
<body>
<div class="header">
<h1>变更执行报告</h1>
<p>变更 ID: {change_request.id}</p>
<p>标题: {change_request.title}</p>
<p>执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
</div>
<h2>执行摘要</h2>
<table>
<tr><td>总设备数</td><td>{total_devices}</td></tr>
<tr><td>成功</td><td class="success">{success_devices}</td></tr>
<tr><td>失败</td><td class="failed">{failed_devices}</td></tr>
<tr><td>已回滚</td><td class="warning">{rolled_back_devices}</td></tr>
</table>
<h2>设备执行详情</h2>
<table>
<tr>
<th>设备</th>
<th>状态</th>
<th>已执行命令</th>
<th>已回滚</th>
<th>执行日志</th>
</tr>
"""
for dev_name, result in results.items():
status = "✅ 成功" if result.success else "❌ 失败"
commands = "<br>".join(result.commands_executed[:5])
if len(result.commands_executed) > 5:
commands += f"<br>...共 {len(result.commands_executed)} 条"
logs = "<br>".join(
f'<span class="log-{l["status"]}">[{l["status"]}] {l["action"]}</span>'
for l in result.execution_log[:5]
)
html += f"""
<tr>
<td>{dev_name}</td>
<td>{status}</td>
<td>{commands}</td>
<td>{'是' if result.rolled_back else '否'}</td>
<td>{logs}</td>
</tr>"""
html += """
</table>
</body>
</html>
"""
with open(output_file, "w", encoding="utf-8") as f:
f.write(html)
return output_file
七、最佳实践总结
自动化变更最佳实践：
1. 变更前充分准备
┌─ 生成可预览的配置（dry-run 先模拟）
├─ 每个变更必须有对应的回滚方案
├─ 评估影响范围并通知相关方
└─ 选择低风险窗口（凌晨业务低谷）
2. 变更中严格管控
┌─ 串行执行比并行安全（优先串行）
├─ 每条命令执行后验证结果
├─ 自动回滚机制作为安全网
└─ 保留完整的执行日志
3. 变更后全面验证
┌─ 配置级验证（期望 vs 实际）
├─ 协议级验证（BGP/OSPF 邻居）
├─ 业务级验证（连通性/延迟）
└─ 持续监控（变更后 24 小时）
4. 从失败中学习
┌─ 每个失败变更都做复盘
├─ 完善回滚方案
├─ 更新自动化检查和验证规则
└─ 沉淀变更知识库
5. 度量与持续改进
┌─ 跟踪变更成功率
├─ 跟踪回滚触发率
├─ 跟踪变更平均耗时
└─ 定期回顾优化变更流程
下篇预告：第326篇 — 多厂商设备统一管理框架，介绍如何通过统一框架管理多厂商网络设备的配置与运维。
下篇预告：第326篇 — 多厂商设备统一管理框架，介绍如何通过统一框架管理多厂商网络设备的配置与运维。