第305篇：NAPALM——统一网络设备管理接口
关键词：NAPALM、多厂商抽象、配置管理、路由表、接口状态、网络即代码
一、引言
Netmiko解决了"连接设备"的问题，但它仍然要求工程师针对不同厂商编写差异化的命令。如果需要采集路由表、接口统计信息、ARP表等结构化数据，工程师仍需自行解析不同厂商的CLI输出格式。
NAPALM（Network Automation and Programmability Abstraction Layer with Multivendor Support）在Netmiko和Python SSH Lib之上构建了一层标准化API，将不同厂商的设备统一抽象为相同的接口。无论底层设备是华为VRP、思科IOS-XE还是Juniper JunOS，NAPALM返回的数据结构完全一致。
二、NAPALM 架构与安装
2.1 安装
pip install napalm
# 华为VRP驱动（非内置）
pip install napalm-huawei-881
2.2 架构分层
2.3 驱动通信方式对比
| 驱动 | 通信协议 | 优势 | 劣势 |
|---|---|---|---|
| Huawei (Netmiko) | SSH/CLI | 兼容老设备 | 需解析文本 |
| Cisco IOS (Netmiko) | SSH/CLI | 广泛支持 | 需解析文本 |
| Juniper JunOS (NCClient) | NETCONF | 原生结构化数据 | 需设备支持NETCONF |
| Arista EOS (EAPI) | HTTP/REST | 高性能 | 需配置HTTP服务 |
三、基础连接与设备信息
3.1 初始化连接
import napalm
# 方式1：直接传参
driver = napalm.get_network_driver("huawei") # 华为VRP
device = driver(
hostname="192.168.1.1",
username="admin",
password="H@wa123"
)
# 方式2：使用字典参数
device_params = {
"hostname": "192.168.1.1",
"username": "admin",
"password": "H@wa123",
"optional_args": {
"delay_factor": 1,
"timeout": 30,
}
}
device = driver(**device_params)
# 打开连接
device.open()
# 关闭连接
device.close()
3.2 获取设备基本信息
import napalm
driver = napalm.get_network_driver("huawei")
device = driver(hostname="192.168.1.1", username="admin", password="H@wa123")
device.open()
# 获取厂商信息
facts = device.get_facts()
print(facts)
# {
# "hostname": "Core-SW-01",
# "model": "CE6865",
# "serial_number": "210235Q8D210BXXXX",
# "os_version": "V200R019C10",
# "vendor": "Huawei",
# "uptime": 1234567,
# }
# 获取硬件信息
hardware = device.get_hardware()
print(hardware)
device.close()
3.3 跨厂商统一接口
import napalm
# 定义异构设备清单
devices = [
{"driver": "huawei", "host": "192.168.1.1", "user": "admin", "pwd": "H@wa123"},
{"driver": "cisco_ios", "host": "192.168.1.2", "user": "admin", "pwd": "C1sc0!"},
{"driver": "juniper", "host": "192.168.1.3", "user": "admin", "pwd": "Jun1per!"},
]
# 同一套API采集所有设备信息
for dev_info in devices:
driver = napalm.get_network_driver(dev_info["driver"])
device = driver(
hostname=dev_info["host"],
username=dev_info["user"],
password=dev_info["pwd"]
)
device.open()
facts = device.get_facts()
print(f"{facts['vendor']:12s} | {facts['model']:15s} | "
f"{facts['os_version']:20s} | {facts['hostname']}")
device.close()
# 输出示例:
# Huawei | CE6865 | V200R019C10 | Core-SW-01
# Cisco | ISR4451 | 16.12.4 | Branch-Rtr-01
# Juniper | EX4300-48T | 21.2R3-S4 | DC-SW-01
四、配置管理
4.1 获取配置
device.open()
# 获取运行配置
running = device.get_config(retrieve="running")
print(running["running"][:200]) # 前200字符
# 获取启动配置
startup = device.get_config(retrieve="startup")
print(startup["startup"][:200])
# 获取候选配置（Juniper特有概念）
candidate = device.get_config(retrieve="candidate")
4.2 配置合并与替换
# 配置合并（Merge）：将新配置添加到现有配置中
merge_config = """
interface GigabitEthernet0/0/20
description To-New-Server
port link-type access
port default vlan 100
"""
# 预览变更（Dry Run），不实际下发
changes = device.compare_config(merge_config)
print("预览变更:")
print(changes)
# 加载配置到候选区
device.load_merge_candidate(config=merge_config)
# 确认并应用配置
device.commit_config()
print("配置已提交")
# 如果确认前发现异常，可回滚
# device.rollback_config()
4.3 合规性检查（Compliance Checker）
# 定义期望配置模板
desired_config = {
"interfaces": {
"GigabitEthernet0/0/1": {
"description": "To-Core-SW-01",
"enabled": True,
},
"GigabitEthernet0/0/2": {
"description": "To-Core-SW-02",
"enabled": True,
},
},
"vlans": {
"10": {"name": "USER_VLAN"},
"20": {"name": "SERVER_VLAN"},
}
}
# 加载期望配置
device.load_replace_candidate(config=napalm.utils.pytest_get_fixture(desired_config))
# 对比差异
diff = device.compare_config()
print("配置差异:")
print(diff)
# 如果无差异，说明合规
if not diff.strip():
print("设备配置合规")
else:
print("设备配置不合规，发现差异")
五、网络状态采集
5.1 接口信息
device.open()
# 获取所有接口信息
interfaces = device.get_interfaces()
for name, data in interfaces.items():
print(f"{name:35s} | Up: {str(data['is_up']):5s} | "
f"Speed: {data['speed']:6d} | "
f"MAC: {data['mac_address']}")
# 示例输出:
# GigabitEthernet0/0/1 | Up: True | Speed: 1000 | MAC: 00:1e:10:xx:xx:xx
# GigabitEthernet0/0/2 | Up: True | Speed: 1000 | MAC: 00:1e:10:xx:xx:xx
# GigabitEthernet0/0/20 | Up: True | Speed: 1000 | MAC: 00:1e:10:xx:xx:xx
5.2 接口统计信息
# 获取接口流量统计
interface_stats = device.get_interface_stats()
for iface, stats in interface_stats.items():
rx_bytes = stats["rx_bytes"]
tx_bytes = stats["tx_bytes"]
rx_errors = stats["rx_errors"]
tx_errors = stats["tx_errors"]
# 计算带宽利用率（假设5分钟间隔采样）
rx_mbps = (rx_bytes * 8) / (5 * 60 * 1000000)
tx_mbps = (tx_bytes * 8) / (5 * 60 * 1000000)
print(f"{iface:35s} | RX: {rx_mbps:8.2f} Mbps | "
f"TX: {tx_mbps:8.2f} Mbps | "
f"RX Err: {rx_errors} | TX Err: {tx_errors}")
5.3 路由表
# 获取IPv4路由表
routes = device.get_routes_v4()
for prefix, route in routes.items():
for next_hop_info in route:
print(f"{'=' * 60}")
print(f"前缀: {prefix}")
print(f" 下一跳: {next_hop_info['next_hop']}")
print(f" 出接口: {next_hop_info['interface']}")
print(f" 管理距离: {next_hop_info['preference']}")
print(f" 度量值: {next_hop_info['metric']}")
print(f" 协议: {next_hop_info['protocol']}")
5.4 ARP表
# 获取ARP表
arp_table = device.get_arp_table()
for entry in arp_table:
print(f"IP: {entry['ip']:16s} | "
f"MAC: {entry['mac']:18s} | "
f"接口: {entry['interface']:35s} | "
f"年龄: {entry['age']}")
5.5 LLDP邻居
# 获取LLDP邻居信息
lldp_neighbors = device.get_lldp_neighbors()
for local_iface, neighbors in lldp_neighbors.items():
for neighbor in neighbors:
print(f"{local_iface:30s} -> {neighbor['hostname']:20s} | "
f"端口: {neighbor['port']}")
六、BGP与STP信息
6.1 BGP邻居状态
# 获取BGP邻居
bgp_neighbors = device.get_bgp_neighbors()
print(f"BGP进程: {bgp_neighbors['global']['router_id']}")
print(f"本地AS: {bgp_neighbors['global']['local_as']}")
for peer, data in bgp_neighbors["global"]["neighbors"].items():
print(f"\n邻居: {peer}")
print(f" 状态: {data['is_up']}")
print(f" 对端AS: {data['remote_as']}")
print(f" 保持时间: {data['holdtimer']}")
print(f" 已建立时长: {data['uptime']}秒")
# 打印地址族信息
for afi_safi, routes in data["address_flows"].items():
print(f" 地址族: {afi_safi}")
print(f" 接收前缀: {routes['received']}")
print(f" 发送前缀: {routes['sent']}")
6.2 STP信息
# 获取STP信息
stp = device.get_stp_counters()
print(stp)
# STP接口信息
stp_interfaces = device.get_stp_vlan_mappings()
for vlan, ports in stp_interfaces.items():
print(f"VLAN {vlan}: {len(ports)} 个端口参与STP")
七、批量采集与报告生成
7.1 多设备批量采集
import napalm
import json
from datetime import datetime
import csv
# 从CSV加载设备清单
devices_csv = []
with open("devices.csv", "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
devices_csv.append(row)
# 批量采集
all_data = {}
for dev_row in devices_csv:
driver = napalm.get_network_driver(dev_row["driver"])
device = driver(
hostname=dev_row["host"],
username=dev_row["user"],
password=dev_row["pwd"]
)
try:
device.open()
device_data = {
"facts": device.get_facts(),
"interfaces_count": len(device.get_interfaces()),
"routes_count": len(device.get_routes_v4()),
"bgp_peers": len(device.get_bgp_neighbors()["global"]["neighbors"]),
"vlans": device.get_vlans(),
}
all_data[dev_row["host"]] = device_data
print(f"[+] {dev_row['host']} 采集完成")
except Exception as e:
print(f"[-] {dev_row['host']} 采集失败: {e}")
all_data[dev_row["host"]] = {"error": str(e)}
finally:
device.close()
# 保存报告
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
with open(f"napalm_report_{ts}.json", "w", encoding="utf-8") as f:
json.dump(all_data, f, indent=2, ensure_ascii=False, default=str)
print(f"\n报告已保存: napalm_report_{ts}.json")
7.2 接口健康度报表
def generate_health_report(all_data):
"""生成设备健康度报告"""
report_lines = [
f"{'设备IP':16s} | {'接口总数':8s} | {'Down接口':8s} | "
f"{'路由数':8s} | {'BGP邻居':8s} | {'状态'}",
"-" * 90,
]
for host, data in all_data.items():
if "error" in data:
report_lines.append(f"{host:16s} | {'N/A':8s} | {'N/A':8s} | "
f"{'N/A':8s} | {'N/A':8s} | 异常: {data['error']}")
continue
facts = data["facts"]
interfaces = data.get("interfaces", {})
# 计算Down接口数
down_count = sum(1 for i in interfaces.values() if not i["is_up"])
total = len(interfaces)
# 健康状态
status = "健康" if down_count == 0 else f"异常({down_count} Down)"
report_lines.append(
f"{host:16s} | {total:8d} | {down_count:8d} | "
f"{data['routes_count']:8d} | {data['bgp_peers']:8d} | {status}"
)
print("\n".join(report_lines))
generate_health_report(all_data)
八、故障案例：NAPALM路由表采集性能瓶颈
8.1 故障现象
采集50台核心路由器的路由表时，单台设备get_routes_v4()耗时超过3分钟，导致整体采集超时。
8.2 排查分析
# 查看NAPALM华为驱动的源码逻辑：
# 1. 执行 'display ip routing-table'（全量输出）
# 2. 逐行正则解析
# 3. 转换为标准字典格式
# 问题：核心路由器路由表条目超过50000条，CLI输出约15MB
# 文本解析耗时 = 传输时间 + 逐行解析时间 > 180秒
8.3 解决方案
方案1：使用NETCONF协议采集（推荐）
# 如果设备支持NETCONF，使用ncclient驱动替代CLI驱动
driver = napalm.get_network_driver("huawei881") # NETCONF驱动
device = driver(
hostname="192.168.1.1",
username="admin",
password="H@wa123",
optional_args={"port": 830}
)
device.open()
routes = device.get_routes_v4() # 结构化数据，无需解析
device.close()
方案2：增量采集
# 只采集感兴趣的网段
device.open()
all_routes = device.get_routes_v4()
# 过滤出关键前缀
critical_prefixes = ["10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"]
filtered = {}
for prefix in all_routes:
for cp in critical_prefixes:
if prefix.startswith(cp.split("/")[0].rsplit(".", 1)[0]):
filtered[prefix] = all_routes[prefix]
break
device.close()
八（续）、高级主题
8.1 NAPALM CLI 快速验证
NAPALM 提供了一个命令行工具，无需编写 Python 代码即可快速验证设备连接和采集能力：
# 安装 napalm-automation-cli
pip install napalm-automation-cli
# 快速采集设备 facts
napalm-cli --user admin --password H@wa123 --vendor huawei get facts 192.168.1.1
# 采集接口信息
napalm-cli --user admin --password H@wa123 --vendor huawei get interfaces 192.168.1.1
# 采集路由表
napalm-cli --user admin --password H@wa123 --vendor huawei get routes_v4 192.168.1.1
# 采集 BGP 邻居
napalm-cli --user admin --password H@wa123 --vendor huawei get bgp_neighbors 192.168.1.1
# 列出设备支持的驱动
napalm-cli --list-drivers
8.2 NAPALM 驱动能力矩阵
| 方法 | 华为(SSH) | 华为(NETCONF) | Cisco IOS | Juniper | Arista EOS |
|---|---|---|---|---|---|
get_facts |
✅ | ✅ | ✅ | ✅ | ✅ |
get_interfaces |
✅ | ✅ | ✅ | ✅ | ✅ |
get_interfaces_counters |
✅ | ✅ | ✅ | ✅ | ✅ |
get_lldp_neighbors |
✅ | ✅ | ✅ | ✅ | ✅ |
get_arp_table |
✅ | ✅ | ✅ | ✅ | ✅ |
get_mac_address_table |
✅ | ✅ | ✅ | ✅ | ✅ |
get_routes_v4 |
✅ | ✅ | ✅ | ✅ | ✅ |
get_bgp_neighbors |
✅ | ✅ | ✅ | ✅ | ✅ |
get_bgp_routes |
⚠️ | ✅ | ✅ | ✅ | ✅ |
get_config |
✅ | ✅ | ✅ | ✅ | ✅ |
load_merge_candidate |
✅ | ✅ | ✅ | ✅ | ✅ |
commit_config |
✅ | ✅ | ✅ | ✅ | ✅ |
rollback_config |
✅ | ✅ | ✅ | ✅ | ✅ |
get_stp_counters |
⚠️ | ⚠️ | ✅ | ✅ | ✅ |
get_vlans |
✅ | ✅ | ✅ | ✅ | ✅ |
get_optics |
⚠️ | ✅ | ✅ | ⚠️ | ✅ |
get_env_stats |
⚠️ | ⚠️ | ⚠️ | ✅ | ✅ |
⚠️ = 部分支持，取决于具体驱动版本和设备型号
8.3 NAPALM 与 Ansible Network 集成
在 Ansible Playbook 中使用 NAPALM 模块（替代传统 CLI 模块）：
---
- name: NAPALM 驱动的网络自动化
hosts: network_devices
gather_facts: no
tasks:
# 1. 采集接口信息（统一结构）
- name: 采集所有接口信息
napalm_alias:
module: get
method: get_interfaces
delegate_to: "{{ inventory_hostname }}"
register: napalm_interfaces
# 2. 基于采集结果做条件判断
- name: 检查 Down 接口
napalm_alias:
module: get
method: get_interfaces
register: iface_data
delegate_to: "{{ inventory_hostname }}"
- name: 报告 Down 接口
debug:
msg: >-
{{ item.key }} is DOWN on {{ inventory_hostname }}
loop: "{{ iface_data.result | dict2items }}"
when: not item.value.is_up
# 3. 配置合规检查
- name: 加载期望配置
napalm_alias:
module: configure
method: load_merge_candidate
config: |
interface GigabitEthernet0/0/1
description To-Core-SW-01-Standard
delegate_to: "{{ inventory_hostname }}"
- name: 预览配置差异
napalm_alias:
module: configure
method: compare_config
register: config_diff
delegate_to: "{{ inventory_hostname }}"
- name: 有条件地提交配置
napalm_alias:
module: configure
method: commit_config
delegate_to: "{{ inventory_hostname }}"
when: config_diff.result | length > 0
8.4 NAPALM 性能优化策略
| 优化策略 | 说明 | 效果 |
|---|---|---|
| 使用 NETCONF 驱动 | 结构化数据直接传输，跳过文本解析 | 路由表采集提速 5-10 倍 |
| 减少采集频率 | 对静态数据（facts、interfaces）降低采集频次 | 降低设备 CPU 负载 |
| 并行采集 | 使用 ThreadPoolExecutor 并发连接 |
批量采集总时长缩短 |
| 选择性采集 | 仅调用需要的 get_* 方法，不遍历全量 API |
减少不必要的 SSH 交互 |
| 连接复用 | 单台设备的多次采集共用一个连接 | 减少 SSH 握手开销 |
# 连接复用示例
device.open()
try:
facts = device.get_facts()
interfaces = device.get_interfaces()
arp = device.get_arp_table()
# 所有采集共用一次 SSH 连接
except Exception as e:
print(f"采集异常: {e}")
finally:
device.close()
九、总结
| 知识点 | 要点 |
|---|---|
| get_network_driver | 按厂商获取驱动，支持huawei/cisco_ios/juniper等 |
| get_facts | 统一获取设备基本信息（厂商、型号、版本等） |
| get_config | 获取running/startup/candidate配置 |
| load_merge_candidate | 加载合并配置到候选区 |
| compare_config | 预览配置差异（Dry Run） |
| commit_config / rollback_config | 提交或回滚配置 |
| get_interfaces / get_routes_v4 | 统一采集接口和路由信息 |
| 驱动选择 | CLI驱动（兼容老设备） vs NETCONF驱动（高性能结构化数据） |
思考题
- NAPALM的
load_merge_candidate()和load_replace_candidate()有何本质区别？在什么场景下应该使用Replace模式？ - 如果一个NAPALM驱动不支持
get_bgp_neighbors()方法，调用时会发生什么？如何优雅地处理这种情况？ - 对比Netmiko和NAPALM：在"批量下发VLAN配置"和"采集全网路由表"两个场景中，分别选择哪个工具更合适？为什么？
下篇预告
第306篇：Jinja2 模板引擎——配置模板化。学习如何用模板将设备配置参数化，实现"一份模板、适配百台设备"的配置生成能力。