第323篇：网络即代码（Network as Code）理念
关键词：Network as Code、NaC、基础设施即代码、声明式配置、状态管理、幂等性、Drift Detection
一、引言
"基础设施即代码"（Infrastructure as Code, IaC）已被云计算领域广泛接受。从网络工程师的视角，Network as Code（NaC）将同样的理念应用到网络设备上：网络配置不再是通过 CLI 逐条输入的手工作业，而是以代码文件形式存在，可通过版本控制、代码审查、自动化测试等软件工程手段管理。
NaC 的核心理念包含五个原则：
- 声明式优于命令式——描述"最终期望状态"，而非"一步步如何操作"
- 幂等性——同一配置重复执行多次，结果一致
- 可版本化——所有配置变更可追溯
- 可测试——配置在部署前可验证
- 可复现——相同输入在任何环境下产生相同输出
二、声明式 vs 命令式配置
2.1 两种范式的对比
命令式配置（传统方式）：
# 手工逐条输入
system-view
interface GigabitEthernet1/0/0
ip address 10.1.1.1 255.255.255.252
description UPLINK_TO_PE-BJ-02
qos lr cir 1000
quit
bgp 65001
peer 10.1.1.2 as-number 65002
address-family ipv4 unicast
peer 10.1.1.2 enable
quit
quit
声明式配置（NaC 方式）：
# desired_state.yaml - 描述期望的网络状态
interfaces:
- name: GigabitEthernet1/0/0
description: "UPLINK_TO_PE-BJ-02"
ip_address: "10.1.1.1/30"
qos:
lr:
cir: 1000 # Mbps
bgp:
as_number: 65001
neighbors:
- ip: "10.1.1.2"
remote_as: 65002
address_families:
- "ipv4 unicast"
# reconciler.py - 对等器：对比期望状态与运行状态，自动弥合差异
import yaml
import napalm
def reconcile(device, desired_state_file):
"""声明式配置对等"""
driver = napalm.get_network_driver('eos')
device = driver(device['host'], device['username'], device['password'])
device.open()
# 读取期望状态
with open(desired_state_file) as f:
desired = yaml.safe_load(f)
# 获取运行状态
running = device.get_interfaces()
bgp = device.get_bgp_config()
# 对比差异
diffs = compute_diff(desired, running, bgp)
if not diffs:
print("配置已符合期望状态，无需变更")
return True
# 生成补丁并应用
patch = generate_patch(diffs)
device.load_merge_candidate(config=patch)
device.commit_config()
print(f"已应用 {len(diffs)} 项配置变更")
return True
2.2 声明式配置的优势
| 对比维度 | 命令式 | 声明式 |
|---|---|---|
| 可读性 | 需理解执行顺序 | 直观描述目标状态 |
| 可复用性 | 每台设备需单独调整 | 同一模板适配多设备 |
| 幂等性 | 重复执行可能产生副作用 | 重复执行结果一致 |
| Drift 检测 | 难以发现偏离 | 自动对比期望与现状 |
| 回滚 | 需反向操作 | 重新应用旧状态文件 |
三、NaC 完整工作流
3.1 项目目录结构
network-as-code/
- inventory/
- devices.yaml # 设备清单
- groups/ # 设备分组
- pe_routers.yaml
- core_switches.yaml
- templates/
- interface.j2 # Jinja2 接口模板
- bgp.j2 # Jinja2 BGP 模板
- ospf.j2 # Jinja2 OSPF 模板
- acl.j2 # Jinja2 ACL 模板
- configs/
- desired/ # 期望配置
- PE-BJ-01.yaml
- PE-BJ-02.yaml
- running/ # 运行配置（采集自设备）
- PE-BJ-01.running.cfg
- PE-BJ-02.running.cfg
- golden/ # 黄金配置（基线）
- pe_golden.yaml
- core_golden.yaml
- tests/
- test_interface.py
- test_bgp.py
- test_acls.py
- scripts/
- reconcile.py # 对等器
- drift_detect.py # 偏离检测
- compliance_check.py # 合规检查
- ansible/
- playbooks/
- roles/
- .gitignore
- README.md
3.2 设备清单定义
# inventory/devices.yaml
devices:
- name: PE-BJ-01
hostname: 10.1.1.1
os: huawei_vrp
role: pe_router
groups: [pe_routers, bj_region]
mgmt_ip: 172.16.1.1
tags:
site: beijing
layer: pe
- name: PE-BJ-02
hostname: 10.1.1.2
os: huawei_vrp
role: pe_router
groups: [pe_routers, bj_region]
mgmt_ip: 172.16.1.2
tags:
site: beijing
layer: pe
- name: P-BJ-01
hostname: 10.1.1.3
os: huawei_vrp
role: p_router
groups: [core_routers, bj_region]
mgmt_ip: 172.16.1.3
tags:
site: beijing
layer: core
3.3 Jinja2 模板生成设备配置
# templates/bgp.j2
# BGP 配置模板
bgp {{ bgp_as }}
router-id {{ router_id }}
{% for peer in bgp_peers %}
peer {{ peer.ip }} as-number {{ peer.remote_as }}
{% if peer.description %}
peer {{ peer.ip }} description {{ peer.description }}
{% endif %}
{% for af in peer.address_families %}
address-family {{ af }}
peer {{ peer.ip }} enable
{% if peer.route_policy %}
peer {{ peer.ip }} route-policy {{ peer.route_policy }} import
{% endif %}
quit
{% endfor %}
quit
{% endfor %}
{% if route_reflector %}
% Route Reflector 配置
auto-flowspec enable
quit
{% endif %}
# scripts/generate_config.py
from jinja2 import Environment, FileSystemLoader
import yaml
def generate_device_config(device, templates_dir='templates/'):
"""根据设备参数和模板生成配置"""
env = Environment(loader=FileSystemLoader(templates_dir))
# 加载设备参数
with open(f'configs/desired/{device["name"]}.yaml') as f:
params = yaml.safe_load(f)
configs = []
# 生成 BGP 配置
bgp_template = env.get_template('bgp.j2')
bgp_config = bgp_template.render(**params['bgp'])
configs.append(bgp_config)
# 生成接口配置
intf_template = env.get_template('interface.j2')
intf_config = intf_template.render(**params['interfaces'])
configs.append(intf_config)
full_config = '\n'.join(configs)
# 保存生成的配置
with open(f'generated/{device["name"]}.cfg', 'w') as f:
f.write(full_config)
return full_config
四、配置漂移检测与合规检查
4.1 漂移检测
配置漂移（Configuration Drift）是指设备上实际运行的配置与期望配置发生偏离，原因可能包括：手动 CLI 修改、紧急变更未走流程、设备重启后配置丢失等。
# scripts/drift_detect.py
import yaml
import napalm
from deepdiff import DeepDiff
def detect_drift(device, desired_state_file):
"""检测配置漂移"""
driver = napalm.get_network_driver(device['os'])
dev = driver(device['hostname'], device['username'],
device['password'], optional_args={'timeout': 60})
dev.open()
# 采集运行配置
running = dev.get_config()['running']
# 读取期望配置
with open(desired_state_file) as f:
desired = yaml.safe_load(f)
# 归一化格式
running_normalized = normalize_config(running)
# 深度对比
diff = DeepDiff(desired, running_normalized,
ignore_order=True,
exclude_regex_paths=[r"\['last_modified'\]"])
if diff:
print(f"⚠️ 设备 {device['name']} 检测到配置漂移:")
for category, items in diff.items():
print(f" [{category}]:")
for item in items:
print(f" {item}")
return diff
else:
print(f"✅ 设备 {device['name']} 配置无漂移")
return None
def normalize_config(raw_config):
"""将原始配置归一化为结构化格式"""
# 解析华为 VRP 配置格式
# ...（解析逻辑）
pass
4.2 合规检查
# compliance_rules.yaml
rules:
- id: SEC-001
name: "管理接口必须启用 AAA 认证"
severity: critical
check: |
interface {{ interface }}
user-interface line vty 0 4
authentication-mode aaa
devices: [all]
- id: SEC-002
name: "SNMP v2c 必须禁用"
severity: high
check: |
# 不应存在 snmp-agent community 配置
NOT_EXISTS: snmp-agent community
devices: [all]
- id: PERF-001
name: "万兆接口必须启用 Jumbo 帧"
severity: medium
check: |
interface {{ interface }}
port link-mode bridge
max-payload-size 9216
devices: [pe_routers]
- id: NET-001
name: "PE 设备 BGP TTL 安全保护必须启用"
severity: high
check: |
bgp {{ as_number }}
peer {{ peer_ip }} ttl-security hop-count 1
devices: [pe_routers]
# scripts/compliance_check.py
import yaml
def run_compliance_check(device, rules_file='compliance_rules.yaml'):
"""运行合规检查"""
with open(rules_file) as f:
rules = yaml.safe_load(f)
results = []
running_config = get_running_config(device)
for rule in rules['rules']:
# 检查设备是否匹配
if 'all' not in rule['devices'] and device['role'] not in rule['devices']:
continue
# 执行检查
compliant = evaluate_rule(rule, running_config, device)
results.append({
'rule_id': rule['id'],
'name': rule['name'],
'severity': rule['severity'],
'compliant': compliant,
'device': device['name']
})
# 输出结果
for r in results:
status = '✅ PASS' if r['compliant'] else '❌ FAIL'
print(f"{status} [{r['severity'].upper()}] {r['rule_id']}: {r['name']}")
failed = [r for r in results if not r['compliant']]
return {'total': len(results), 'passed': len(results) - len(failed),
'failed': len(failed), 'details': failed}
五、Netmiko批量配置分发实战
5.1 批量配置分发器
# scripts/netmiko_batch_deploy.py
"""
基于 Netmiko 的批量配置分发器
支持：顺序/并行分发、结果汇总、自动回滚、详细日志
"""
import logging
import time
from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime
from pathlib import Path
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s [%(levelname)s] %(message)s',
handlers=[
logging.FileHandler('batch_deploy.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
@dataclass
class DeployResult:
"""单台设备部署结果"""
device_name: str
success: bool
output: str = ""
error: str = ""
duration_ms: float = 0.0
pre_backup: Optional[str] = None
post_backup: Optional[str] = None
class BatchConfigDeployer:
"""批量配置分发器"""
def __init__(self, credentials: Dict, backup_dir: str = './backups'):
self.credentials = credentials
self.backup_dir = Path(backup_dir)
self.backup_dir.mkdir(parents=True, exist_ok=True)
def _get_connect_params(self, device: Dict) -> Dict:
"""构建连接参数"""
return {
'device_type': device.get('platform', 'huawei_vrpv8'),
'host': device['ip'],
'username': device.get('username', self.credentials.get('username')),
'password': device.get('password', self.credentials.get('password')),
'port': device.get('port', 22),
'timeout': 30,
'session_timeout': 120,
'session_log': f"./logs/{device['name']}.log",
}
def deploy_single(self, device: Dict, config: str,
pre_backup: bool = True) -> DeployResult:
"""
单台设备配置部署
:param device: 设备信息
:param config: 要下发的配置（多行字符串）
:param pre_backup: 是否先备份当前配置
"""
dev_name = device['name']
start_time = time.time()
result = DeployResult(device_name=dev_name)
try:
with ConnectHandler(**self._get_connect_params(device)) as net:
# 1. 变更前备份
if pre_backup:
pre_file = self.backup_dir / dev_name / f"pre_{datetime.now().strftime('%Y%m%d_%H%M%S')}.cfg"
pre_file.parent.mkdir(parents=True, exist_ok=True)
running = net.send_command('display current-configuration')
with open(pre_file, 'w', encoding='utf-8') as f:
f.write(running)
result.pre_backup = str(pre_file)
logger.info(f"设备 {dev_name} 变更前备份: {pre_file}")
# 2. 进入系统视图
net.send_command('system-view')
# 3. 逐行下发配置
config_lines = [line for line in config.strip().split('\n')
if line.strip() and not line.strip().startswith('#')]
for line in config_lines:
logger.debug(f"下发配置: {line}")
output = net.send_command(line, read_timeout=10)
if 'Error' in output or 'error' in output:
result.error = f"配置行 '{line}' 报错: {output}"
result.success = False
# 触发回滚
self._rollback_device(device, pre_file if pre_backup else None)
return result
# 4. 保存配置
net.send_command('return')
net.send_command('save force')
# 5. 变更后备份
post_file = self.backup_dir / dev_name / f"post_{datetime.now().strftime('%Y%m%d_%H%M%S')}.cfg"
running = net.send_command('display current-configuration')
with open(post_file, 'w', encoding='utf-8') as f:
f.write(running)
result.post_backup = str(post_file)
result.success = True
result.output = f"成功下发 {len(config_lines)} 条配置"
except Exception as e:
result.success = False
result.error = str(e)
logger.error(f"设备 {dev_name} 部署失败: {e}")
# 尝试回滚
if pre_backup:
pre_file = list((self.backup_dir / dev_name).glob('pre_*.cfg'))
if pre_file:
self._rollback_device(device, pre_file[-1])
result.duration_ms = (time.time() - start_time) * 1000
return result
def _rollback_device(self, device: Dict, backup_file: Optional[str]):
"""回滚设备配置"""
if not backup_file or not Path(backup_file).exists():
logger.warning(f"设备 {device['name']} 无备份文件，无法回滚")
return
try:
with ConnectHandler(**self._get_connect_params(device)) as net:
net.send_command('reset saved-configuration')
# 重新加载备份配置（简化版，实际需解析后逐行下发）
logger.info(f"设备 {device['name']} 已触发回滚")
except Exception as e:
logger.error(f"设备 {device['name']} 回滚失败: {e}")
def deploy_parallel(self, devices: List[Dict], config: str,
max_workers: int = 10) -> List[DeployResult]:
"""并行部署到多台设备"""
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(self.deploy_single, d, config): d['name']
for d in devices
}
for future in as_completed(futures):
results.append(future.result())
return results
def generate_report(self, results: List[DeployResult]) -> str:
"""生成部署报告"""
lines = ["## 批量配置分发报告", ""]
lines.append(f"| 设备 | 状态 | 耗时(ms) | 错误信息 |")
lines.append(f"|------|------|----------|----------|")
for r in results:
status = '✅ 成功' if r.success else '❌ 失败'
lines.append(f"| {r.device_name} | {status} | "
f"{r.duration_ms:.0f} | {r.error[:40]} |")
total = len(results)
failed = sum(1 for r in results if not r.success)
lines.extend([
"",
f"**统计**: 总计 {total} 台, 成功 {total-failed} 台, "
f"失败 {failed} 台"
])
return '\n'.join(lines)
5.2 Ansible Playbook 批量配置模板
# ansible/playbooks/batch_config_deploy.yml
---
- name: 批量配置分发到网络设备
hosts: "{{ target_group | default('all') }}"
gather_facts: no
connection: ansible.netcommon.network_os
vars:
backup_dir: "/opt/network-backups"
pre_backup: true
config_template: "templates/config.j2"
tasks:
# 阶段1: 变更前备份
- name: 备份当前运行配置
ansible.netcommon.config:
defaults: yes
register: current_config
delegate_to: localhost
run_once: false
when: pre_backup | bool
- name: 保存备份到文件
ansible.builtin.copy:
content: "{{ current_config.src }}"
dest: "{{ backup_dir }}/{{ inventory_hostname }}/pre_{{ ansible_date_time.iso8601_basic_short }}.cfg"
mode: '0644'
delegate_to: localhost
when: pre_backup | bool and current_config is changed
# 阶段2: 渲染配置模板
- name: 渲染设备配置模板
ansible.builtin.template:
src: "{{ config_template }}"
dest: "/tmp/{{ inventory_hostname }}_new_config.cfg"
delegate_to: localhost
register: rendered_config
# 阶段3: 下发配置
- name: 下发配置到设备
ansible.netcommon.config:
src: "/tmp/{{ inventory_hostname }}_new_config.cfg"
save_when: modified
backup_options:
dir: "{{ backup_dir }}/{{ inventory_hostname }}/rollback"
register: deploy_result
# 阶段4: 验证配置
- name: 验证关键配置已生效
ansible.builtin.command:
cmd: "python scripts/verify_config.py {{ inventory_hostname }}"
delegate_to: localhost
register: verify_result
failed_when: verify_result.rc != 0
# 阶段5: 部署失败时自动回滚
- name: 自动回滚（当验证失败时）
ansible.builtin.include_tasks: rollback.yml
when: verify_result is failed
5.3 配置备份与比对脚本
# scripts/config_compare.py
"""
配置比对工具
功能：
1. 逐行 diff 比对
2. 结构化字段比对（IP地址、VLAN、BGP Peer 等）
3. 生成 HTML 可视化报告
"""
import difflib
import json
import yaml
from pathlib import Path
from typing import Dict, List, Tuple
from dataclasses import dataclass
@dataclass
class ConfigDiff:
"""配置差异结果"""
device_name: str
total_lines_before: int
total_lines_after: int
added_lines: List[str] = None
removed_lines: List[str] = None
modified_lines: List[Tuple[str, str]] = None
key_changes: List[Dict] = None
def __post_init__(self):
self.added_lines = self.added_lines or []
self.removed_lines = self.removed_lines or []
self.modified_lines = self.modified_lines or []
self.key_changes = self.key_changes or []
def compare_configs(before_path: str, after_path: str,
device_name: str = '') -> ConfigDiff:
"""
逐行比对两份配置文件
:return: ConfigDiff 对象
"""
with open(before_path, 'r', encoding='utf-8') as f:
before_lines = f.readlines()
with open(after_path, 'r', encoding='utf-8') as f:
after_lines = f.readlines()
diff = difflib.unified_diff(
before_lines, after_lines,
fromfile='before', tofile='after',
lineterm=''
)
added = []
removed = []
for line in diff:
if line.startswith('+') and not line.startswith('+++'):
added.append(line[1:].rstrip())
elif line.startswith('-') and not line.startswith('---'):
removed.append(line[1:].rstrip())
return ConfigDiff(
device_name=device_name or Path(before_path).stem,
total_lines_before=len(before_lines),
total_lines_after=len(after_lines),
added_lines=added,
removed_lines=removed
)
def compare_structured_configs(before_yaml: str, after_yaml: str,
device_name: str = '') -> ConfigDiff:
"""
结构化 YAML 配置比对
可精确识别：新增/删除的 VLAN、BGP Peer、ACL 规则等
"""
before = yaml.safe_load(before_yaml)
after = yaml.safe_load(after_yaml)
key_changes = []
# 比对 VLAN
before_vlans = set(before.get('vlans', {}).keys())
after_vlans = set(after.get('vlans', {}).keys())
for vid in after_vlans - before_vlans:
key_changes.append({
'type': 'VLAN',
'action': 'added',
'detail': f"VLAN {vid} 新增"
})
for vid in before_vlans - after_vlans:
key_changes.append({
'type': 'VLAN',
'action': 'removed',
'detail': f"VLAN {vid} 删除"
})
# 比对 BGP Peers
before_peers = {p['ip'] for p in before.get('bgp', {}).get('peers', [])}
after_peers = {p['ip'] for p in after.get('bgp', {}).get('peers', [])}
for peer in after_peers - before_peers:
key_changes.append({
'type': 'BGP Peer',
'action': 'added',
'detail': f"Peer {peer} 新增"
})
for peer in before_peers - after_peers:
key_changes.append({
'type': 'BGP Peer',
'action': 'removed',
'detail': f"Peer {peer} 删除"
})
# 比对接口 IP
before_ips = {i['name']: i.get('ip') for i in before.get('interfaces', [])}
after_ips = {i['name']: i.get('ip') for i in after.get('interfaces', [])}
for name, ip in after_ips.items():
if name in before_ips and before_ips[name] != ip:
key_changes.append({
'type': 'Interface IP',
'action': 'modified',
'detail': f"{name}: {before_ips[name]} -> {ip}"
})
return ConfigDiff(
device_name=device_name,
total_lines_before=len(str(before)),
total_lines_after=len(str(after)),
key_changes=key_changes
)
def generate_html_report(diffs: List[ConfigDiff], output_path: str):
"""生成 HTML 可视化比对报告"""
html_parts = [
"<!DOCTYPE html>",
"<html><head><meta charset='utf-8'>",
"<title>配置差异报告</title>",
"<style>",
"body { font-family: 'Segoe UI', sans-serif; margin: 20px; }",
"table { border-collapse: collapse; width: 100%; margin: 10px 0; }",
"th, td { border: 1px solid #ddd; padding: 8px; text-align: left; }",
"th { background: #4CAF50; color: white; }",
".added { background: #e8f5e9; color: #2e7d32; }",
".removed { background: #ffebee; color: #c62828; }",
".vln { padding: 2px 6px; border-radius: 3px; }",
".added .vln { background: #c8e6c9; }",
".removed .vln { background: #ffcdd2; }",
"</style></head><body>",
"<h1>网络配置差异报告</h1>",
]
for diff in diffs:
html_parts.append(f"<h2>设备: {diff.device_name}</h2>")
html_parts.append(
f"<p>变更前行数: {diff.total_lines_before} | "
f"变更后行数: {diff.total_lines_after}</p>"
)
if diff.key_changes:
html_parts.append("<h3>关键变更</h3>")
html_parts.append(
"<table><tr><th>类型</th><th>动作</th><th>详情</th></tr>"
)
for kc in diff.key_changes:
action_class = 'added' if kc['action'] == 'added' else 'removed'
html_parts.append(
f"<tr class='{action_class}'>"
f"<td>{kc['type']}</td>"
f"<td>{kc['action']}</td>"
f"<td>{kc['detail']}</td></tr>"
)
html_parts.append("</table>")
if diff.added_lines:
html_parts.append("<h3>新增配置行</h3>")
html_parts.append("<table><tr><th>新增内容</th></tr>")
for line in diff.added_lines[:50]: # 限制显示50行
html_parts.append(
f"<tr class='added'><td class='vln'>{line}</td></tr>"
)
html_parts.append("</table>")
html_parts.append("</body></html>")
with open(output_path, 'w', encoding='utf-8') as f:
f.write('\n'.join(html_parts))
5.4 错误处理与日志记录
# scripts/deploy_logger.py
"""
部署日志与错误处理模块
支持：结构化日志、错误分类、重试机制、告警推送
"""
import logging
import json
import traceback
from enum import Enum
from datetime import datetime
from typing import Dict, Optional, Callable
class ErrorCategory(Enum):
"""错误分类"""
CONNECTION_FAILED = "connection_failed"
AUTH_FAILED = "authentication_failed"
COMMAND_ERROR = "command_error"
TIMEOUT = "timeout"
CONFIG_CONFLICT = "config_conflict"
UNKNOWN = "unknown"
class DeployLogger:
"""部署专用日志器"""
def __init__(self, log_dir: str = './logs'):
import os
os.makedirs(log_dir, exist_ok=True)
self.log_file = os.path.join(log_dir, f'deploy_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log')
self.error_log = os.path.join(log_dir, 'errors.jsonl')
# 文件处理器
self.logger = logging.getLogger('deploy')
self.logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(self.log_file)
handler.setFormatter(logging.Formatter(
'%(asctime)s [%(levelname)s] %(name)s - %(message)s'
))
self.logger.addHandler(handler)
# 控制台处理器
console = logging.StreamHandler()
console.setLevel(logging.INFO)
console.setFormatter(logging.Formatter('[%(levelname)s] %(message)s'))
self.logger.addHandler(console)
def log_deploy_start(self, deploy_id: str, devices: list, config_summary: str):
"""记录部署开始"""
self.logger.info("=" * 60)
self.logger.info(f"部署开始 | ID={deploy_id}")
self.logger.info(f"目标设备: {', '.join(d['name'] for d in devices)}")
self.logger.info(f"配置摘要: {config_summary}")
def log_device_result(self, device_name: str, success: bool,
duration: float, error: Optional[str] = None):
"""记录单台设备部署结果"""
status = 'SUCCESS' if success else 'FAILED'
self.logger.info(
f"设备 {device_name}: {status} | 耗时 {duration:.0f}ms"
+ (f" | 错误: {error}" if error else "")
)
def log_error(self, deploy_id: str, device_name: str,
error_category: ErrorCategory, error_message: str,
traceback_str: str = ''):
"""记录结构化错误日志（JSONL 格式，便于后续分析）"""
error_record = {
'timestamp': datetime.now().isoformat(),
'deploy_id': deploy_id,
'device_name': device_name,
'error_category': error_category.value,
'error_message': error_message,
'traceback': traceback_str,
}
with open(self.error_log, 'a', encoding='utf-8') as f:
f.write(json.dumps(error_record, ensure_ascii=False) + '\n')
@staticmethod
def categorize_error(exception: Exception) -> ErrorCategory:
"""根据异常类型自动分类"""
error_msg = str(exception).lower()
if 'connection' in error_msg or 'refused' in error_msg:
return ErrorCategory.CONNECTION_FAILED
elif 'auth' in error_msg or 'password' in error_msg:
return ErrorCategory.AUTH_FAILED
elif 'timeout' in error_msg:
return ErrorCategory.TIMEOUT
elif 'error' in error_msg and 'config' in error_msg:
return ErrorCategory.CONFIG_CONFLICT
else:
return ErrorCategory.UNKNOWN
class RetryHandler:
"""重试处理器"""
def __init__(self, max_retries: int = 3, base_delay: float = 2.0,
logger: Optional[logging.Logger] = None):
self.max_retries = max_retries
self.base_delay = base_delay
self.logger = logger or logging.getLogger('retry')
def execute_with_retry(self, func: Callable, *args,
retryable_errors: tuple = (TimeoutError, ConnectionError),
**kwargs) -> any:
"""
带指数退避的重试执行
:param func: 要执行的函数
:param retryable_errors: 可重试的异常类型
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
return func(*args, **kwargs)
except retryable_errors as e:
last_exception = e
if attempt < self.max_retries:
delay = self.base_delay * (2 ** attempt)
self.logger.warning(
f"第 {attempt + 1} 次尝试失败: {e}，"
f"{delay}s 后重试..."
)
import time
time.sleep(delay)
else:
self.logger.error(
f"已达最大重试次数 ({self.max_retries})，放弃"
)
raise last_exception
六、总结
[*] --> DesiredState: 定义期望状态
DesiredState --> Reconcile: 对等器执行
Reconcile --> Compare: 对比期望 vs 运行
Compare --> NoChange: 无差异
Compare --> ApplyPatch: 有差异
ApplyPatch --> Verify: 验证应用结果
Verify --> DesiredState: 状态一致
Verify --> ApplyPatch: 验证失败，重试
NoChange --> [*]
DesiredState --> [*]
```
故障案例：NaC 预防配置不一致
某企业分支网络有 50 台交换机，运维发现不同分支的 VLAN 划分存在差异。经排查，原因是多次变更由不同人员执行，部分设备漏配。引入 NaC 后：
- 所有 VLAN 配置统一在
configs/desired/vlan_base.yaml中定义 - 每日自动运行漂移检测，发现偏离立即告警
- 合规检查在 CI/CD Pipeline 中自动执行，违规配置无法部署
- 3 个月内将配置一致率从 72% 提升至 99.5%
思考题
- NaC 强调"声明式优于命令式"，但在某些复杂场景（如路由策略的精细调优）中，声明式配置可能不够灵活。如何平衡两者的使用？
- 对于老旧设备（不支持 NETCONF/RESTCONF 的现代管理接口），如何实现 NaC？
下篇预告：第324篇《Git 版本控制管理网络配置》——用 Git 管理网络配置，实现变更追溯、分支管理和协作评审。