☰
用Nornir批量采集网络设备信息:从Inventory到结果落盘
2026/10/5 5:09:08 网站建设 项目流程

干运维最烦的事之一,就是登录网络设备敲命令收集信息。设备一多,厂商一杂,这个动作就特别消耗人。我现在的做法是用 Nornir 写一个 Python 脚本,批量对设备执行show/display命令,抓版本、接口状态、序列号这些信息,然后落成 JSON、CSV。Nornir 是一个纯 Python 的网络自动化框架,比起 Ansible 更接近开发者的习惯,适合写巡检、采集、备份这类工具脚本。这篇文章就把我从零搭起来的过程完整盘一遍,包括 Inventory 设计、并发参数、结果处理和踩坑记录,适合正在把手工巡检改成自动化的朋友参考。

最早我写批量采集用的是 Paramiko,一个for循环连设备、等回显、切分输出,再自己维护线程池。设备少还行,设备一多问题全来了:有的设备响应慢导致超时,有的命令输出被分页打断,还有的厂商命令不一样,脚本里全是if分支。后来切到 Nornir,思路一下清楚了很多。Nornir 把“连哪些设备”“怎么连”“执行什么任务”三件事拆开,Inventory 管设备清单,连接插件管 SSH,任务就是一个普通 Python 函数,剩下的并发调度框架都替你做完了。

1. 为什么用 Nornir 收集设备信息,而不是自己攒一堆脚本

1.1 信息收集这件事,到底麻烦在哪

网络设备信息收集看着简单,就是登上去敲几条命令,但落到批量场景就变味了。先说数量:一百台交换机、路由器,每台三五个命令,手工敲一遍可能要一整天,还得靠人去核对输出。再说厂商差异:Cisco 是show version,华为是display version,命令不一样,输出格式也不一样,同一个字段在不同厂商设备上的位置可能完全不同。

更麻烦的是输出不结构化。show version打印出来是一坨文本,你要提取序列号、系统版本、运行时间,就得写正则。如果只是临时查一台,正则无所谓;但如果你想做资产台账、版本合规检查,每次都要从纯文本里扒字段,后期维护成本会很高。

我用 Nornir 之后,把采集逻辑分成了三层:

  • 设备层:Inventory 文件里定义设备 IP、平台、账号、角色。
  • 连接层:用 Netmiko 或 NAPALM 插件负责 SSH、命令执行、回显处理。
  • 业务层:自己写 Python 函数,决定每台设备跑什么命令、返回值怎么处理。

这样每一层都能独立替换,加新设备不用改代码,加新采集项也不用改连接方式。

1.2 Nornir 对比 Ansible 和纯脚本的取舍

很多朋友第一反应是问:为什么不用 Ansible?Ansible 当然可以做网络设备信息收集,而且生态成熟,但它和 Nornir 的定位不一样。

对比项Ansible纯 Python + Paramiko/NetmikoNornir
核心形态YAML Playbook手写循环脚本Python 库
编程灵活性中,逻辑复杂时要写自定义模块高高
并发调度内置自己写线程池或串行内置 Runner
设备清单管理Inventory 文件自己定义Inventory 文件
与现有系统集成需要通过 API 或命令行调动直接 import直接 import
学习成本需要熟悉 Ansible 语法只要会 Python只要会 Python

如果你本身是 Python 技术栈,或者想把采集能力嵌到自己的巡检平台、工单系统里,Nornir 会顺手很多。它本质上就是一个可以import的并发任务框架,而不是一套独立运行的系统。

还有一个点是并发模型。Nornir 默认用多线程 Runner,配置里写一个num_workers就能派发任务,不用自己ThreadPoolExecutor。而且它处理任务结果的方式很统一:每台设备返回一个MultiResult,里面有成功状态、输出内容和异常信息,遍历results就能汇总,调试起来非常直观。

2. 环境搭好后,Inventory 文件才是会不会踩坑的分水岭

2.1 最小安装与项目目录

先装依赖。我习惯用虚拟环境,避免污染系统 Python:

python3 -m venv nornir-venv source nornir-venv/bin/activate pip install nornir nornir-netmiko nornir-napalm nornir-utils

这里几个包的分工是:

  • nornir:核心框架,负责 Inventory、Runner、结果管理。
  • nornir-netmiko:提供 Netmiko 相关的任务函数,比如netmiko_send_command。
  • nornir-napalm:提供 NAPALM 相关任务函数,比如napalm_get,用于拉结构化数据。
  • nornir-utils:一些辅助工具,比如print_result打印结果。

项目目录我建议这样组织:

nornir-demo/ ├── config.yaml ├── inventory/ │ ├── hosts.yaml │ ├── groups.yaml │ └── defaults.yaml ├── collect_info.py └── output/

config.yaml是 Nornir 的入口配置,告诉框架 Inventory 文件在哪、并发数多少、用什么 Runner。我给了一个最小可用版本:

inventory: plugin: SimpleInventory options: host_file: "inventory/hosts.yaml" group_file: "inventory/groups.yaml" defaults_file: "inventory/defaults.yaml" runner: plugin: threaded options: num_workers: 20

第一次跑通的时候,num_workers不要照抄我,先改成 5 到 10 就好,原因后面说。

2.2 hosts.yaml / groups.yaml / defaults.yaml 怎么设计

Inventory 是 Nornir 的核心,也是新手最容易瞎搞的地方。我的建议是不要把设备信息塞到一个文件里,而是把“共性”和“特性”分开。

hosts.yaml定义每台设备独有的属性,比如 IP、角色、区域:

core-sw-01: hostname: 192.168.10.11 groups: - huawei data: role: core core-rtr-01: hostname: 192.168.10.12 groups: - cisco data: role: core access-sw-20: hostname: 192.168.20.20 groups: - huawei data: role: access

注意这里的hostname是实际连接地址,设备名是 YAML 顶层的core-sw-01。这两个概念在 Nornir 里是分开的,刚开始容易混。

groups.yaml定义同一类设备的公共属性,比如平台、厂商、连接参数:

huawei: platform: huawei data: vendor: huawei cisco: platform: cisco_ios data: vendor: cisco

defaults.yaml放所有设备默认都有的连接信息:

username: admin password: admin123 port: 22

这样分组的好处是以后加一台设备,只需要在hosts.yaml里写一行,平台和厂商信息自动继承。如果你有一组设备的账号密码不同,可以在hosts.yaml对应设备下直接覆盖username、password。

F(role="core")这种方式能过滤data里的字段。分组信息同样参与继承,所以我习惯在 group 里放data.vendor,过滤设备时用得很顺手。

2.3 连接参数:platform 字段不能拍脑袋

platform这个字段在 Nornir 里非常关键,Netmiko 插件会拿它去匹配对应的设备驱动。Cisco IOS 在 Netmiko 里的平台名是cisco_ios,华为一般是huawei,不同厂商写法不一样,拼错一个字连接就失败。

另外要注意,如果后面用到 NAPALM 拉结构化数据,它的平台名和 Netmiko 不一定一样。比如 Cisco IOS 在 Netmiko 里是cisco_ios,在 NAPALM 里是ios。所以我现在的做法是:用 Netmiko 为主做命令采集,用 NAPALM 做结构化 getter 采集时,单独维护一组连接参数,不让同一个platform值硬扛两个插件。后面如果报“driver 不存在”之类的错,第一个怀疑对象就是平台名。

3. 信息收集的核心写法:从一条命令到多设备并行抓取

3.1 第一条命令:先单台跑通,再谈批量

很多人一上来就写全设备扫描,结果报错信息都分不清是哪台设备挂的。我的习惯是先锁一台设备跑通全链路。

from nornir import InitNornir from nornir.core.filter import F from nornir_netmiko import netmiko_send_command from nornir_utils.plugins.functions import print_result nr = InitNornir(config_file="config.yaml") # 先只挑一台设备 one_device = nr.filter(F(name="core-rtr-01")) result = one_device.run( task=netmiko_send_command, command_string="show version", ) print_result(result)

跑完之后,print_result会打印类似这样的内容:

core-rtr-01 ** changed : False **************************************************** ***** changed : False **************************************************** >>>>>>>>>> 执行结果 >>>>>>>>>> Cisco IOS Software, C880 Software (C880-ADVIPSERVICESK9-M), Version 15.4(3)M8...

result是一个AggregatedResult,以设备名为 key,设备名就能拿到对应结果对象。这一行代码就完成了 SSH 连接、命令执行、回显获取的整个过程。

3.2 用 NAPALM getter 拿结构化数据

用netmiko_send_command拿到的是纯文本,如果你想拿设备厂商、型号、系统版本、序列号这些字段,用 NAPALM 的getters会更省事。

from nornir_napalm.plugins.tasks import napalm_get result = nr.run( task=napalm_get, getters=["facts", "interfaces"], )

facts会返回一个字典,包括vendor、model、os_version、serial_number等字段,interfaces会返回接口状态和描述信息。因为返回的是 Python 字典,后面做资产汇总、版本核对非常方便,不需要对命令输出做正则解析。

这也是我建议把 NAPALM 加进技术栈的原因。它不是用来替代 Netmiko 的,而是互补:Netmiko 适合执行任意命令、拿原始输出;NAPALM 适合拿那些已经定义好的结构化模型。

3.3 按角色或厂商过滤设备

实际巡检时,我不一定每次都全量采集。比如今天只核对核心设备的版本,那就没必要去打扰几十台接入交换机。Nornir 的filter在这里很好用。

core_devices = nr.filter(F(role="core")) result = core_devices.run( task=netmiko_send_command, command_string="show version", )

如果只想处理某个厂商的设备,可以利用 group 里定义的vendor:

cisco_devices = nr.filter(F(vendor="cisco"))

F能过滤设备属性和data里的字段,比自己写循环判断优雅太多。配合命令行参数,还能把过滤条件做成可选的,比如默认采集全部,加--role core就只采核心设备。

3.4 把不同厂商命令映射到同一个任务里

真实环境很少只有单一厂商。Cisco 用show version,华为用display version,同一个脚本怎么兼容?我一般会写一个自定义任务函数,按task.host.platform分发命令。

def collect_system_info(task): commands = { "huawei": ["display version", "display interface brief"], "cisco_ios": ["show version", "show ip interface brief"], } platform = task.host.platform cmd_list = commands.get(platform, ["show version"]) outputs = {} for cmd in cmd_list: try: partial = task.run( task=netmiko_send_command, command_string=cmd, ) outputs[cmd] = partial[0].result except Exception: outputs[cmd] = None return outputs result = nr.run(task=collect_system_info)

这里的思路是:collect_system_info是一个普通 Python 函数,Nornir 会把它作为任务分发到每台设备上。函数内部再用task.run调底层的 Netmiko 任务。返回值是字典,最后可以在result[设备名].result里拿到。

这个写法比把所有命令写死在nr.run调用里要清晰很多。以后要新增一个厂商,只需在commands字典里加一行;要新增采集项,改cmd_list就行,完全不用动主流程。

4. 拿到结果之后:批量保存、异常处理与结果过滤

4.1 把命令输出落成 JSON

采集完不保存等于白采。我一般会把当次所有命令输出保存成一个带时间戳的 JSON 文件,既方便回溯,也方便后续写报告。

from datetime import datetime from pathlib import Path import json output_dir = Path("output") output_dir.mkdir(exist_ok=True) result = nr.run(task=collect_system_info) backup_file = output_dir / f"commands_{datetime.now():%Y%m%d_%H%M%S}.json" collected = {} for device_name, device_result in result.items(): collected[device_name] = { "platform": nr.inventory.hosts[device_name].platform, "success": not device_result.failed, "data": device_result.result, } backup_file.write_text( json.dumps(collected, ensure_ascii=False, indent=2, default=str), encoding="utf-8", ) print(f"已保存到 {backup_file}")

default=str是防止结果里有datetime这类不能直接序列化的对象。设备返回的原始回显可能会有大小写差异,先统一存下来,后续做分析时再清洗。

4.2 失败任务不能直接吞掉

批量跑一百台设备,有几台连接失败太正常了。脚本最忌讳的是某个设备挂了导致整个进程崩掉,或者失败之后毫无提示。Nornir 默认不会因为单台设备失败就终止全局任务,它会记录失败状态,跑完再告诉你哪些失败了。

我通常会在结果里单独收集失败设备:

failed_hosts = [] for device_name, device_result in result.items(): if device_result.failed: failed_hosts.append(device_name) if failed_hosts: print(f"以下设备采集失败: {failed_hosts}") else: print("全部设备采集成功")

注意nr.data.failed_hosts也能拿到失败设备列表,我两种都会用。前者适合在result结果集里保留现场,后者适合快速判断有没有失败。

如果希望某个关键设备失败了就中断整个任务,可以在config.yaml里加core.raise_on_error相关配置,或者直接在自定义任务里raise。不过网络采集场景我一般不建议这么做,宁可记录失败,也不要因为一台设备导致后面全不跑。

4.3 把设备 facts 落成 CSV

命令输出适合 JSON,资产台账类字段适合 CSV。用 NAPALM 拿到 facts 之后,可以直接写 CSV 给资产系统用。

import csv facts_result = nr.run(task=napalm_get, getters=["facts"]) csv_file = output_dir / "device_facts.csv" with open(csv_file, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( f, fieldnames=["device", "vendor", "model", "os_version", "serial_number"], ) writer.writeheader() for device_name, device_result in facts_result.items(): facts = (device_result.result or {}).get("facts", {}) writer.writerow({ "device": device_name, "vendor": facts.get("vendor", ""), "model": facts.get("model", ""), "os_version": facts.get("os_version", ""), "serial_number": facts.get("serial_number", ""), })

这样导出来的表格,可以直接交给资产管理系统,也可以作为版本合规检查的原始数据。

4.4 定时运行与 CI/CD 扩展

脚本跑通之后,下一步就是让它定时自动跑。我现在的做法是把项目打包成 Docker 镜像,镜像里装好 Python 依赖和脚本,然后用定时任务或 GitLab CI/CD 的 scheduled pipeline 每天执行一次,采集结果自动归档到存储目录。

这种方式的好处是执行环境完全隔离,不会因为服务器上 Python 包版本漂移导致脚本突然挂掉。镜像构建过程也很简单:

FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["python", "collect_info.py"]

每次执行生成的 JSON 文件名带时间戳,天然形成历史记录,后面想做设备配置漂移对比,直接读文件就行。

5. 实际使用中我反复踩的几个坑

5.1 Nornir 3.x 与旧教程的 API 差异

看 Nornir 教程最痛苦的就是版本不一致。网上很多老文章还是 Nornir 2.x 的写法,任务函数直接放在nornir.plugins.tasks.networking里,导入路径是from nornir.plugins.tasks.networking import netmiko_send_command。到了 Nornir 3.x,网络任务已经拆到独立插件里了,正确的导入是:

from nornir_netmiko import netmiko_send_command from nornir_napalm.plugins.tasks import napalm_get

如果你照着旧教程写,大概率会报ModuleNotFoundError: No module named 'nornir.plugins.tasks.networking'。遇到这个错,不用怀疑环境,先检查导入路径。

同样的,config.yaml里的 Runner 配置也有变化。老版本喜欢用runner: threaded,新版本则是在runner.options里指定num_workers。建议直接以官方文档和你安装的nornir版本对应文档为准。

5.2 并发数不是越大越好

Nornir 默认并发数是 100,听起来很猛,但对网络设备来说往往不是好事。我第一次跑全量采集就是直接默认值,结果核心交换机 CPU 直接飙高,还有几台设备 SSH 连接数太多,被安全策略限制了。

网络设备不像服务器,并发 SSH 连接很容易触发它的连接数限制或 CPU 过高。我现在通用的配置是num_workers: 20,对低端接入交换机,甚至会降到 5。如果采集的是几十台核心设备,跑得更慢不可怕,把设备搞出问题才是大事。

5.3 命令输出带分页符和终端宽度问题

网络设备默认输出往往会分页,比如---- More ----这种。Netmiko 一般会自动处理分页问题,但我也遇到过一个情况:某个型号的交换机开启了终端宽度限制,导致display interface的输出被截断,接口描述信息不完整。

解决办法是采集前先关掉分页,并设置终端长度为 0。Netmiko 连接时通常默认会做这个动作,但如果遇到不标准的设备,可以在命令前拼接分号,或者用terminal length 0这种命令先初始化。

还有一个坑是命令输出里有颜色控制符,这些符号会混在结果里。我处理的办法是保存前用正则把 ANSI 转义字符清掉,或者用 Netmiko 自带的一些参数关闭颜色显示。

5.4 设备密码不要明文提交到 Git

刚开始做脚本,我直接把密码写在defaults.yaml里,结果有一次不小心把仓库推到远端才发现。现在我已经改成环境变量注入的方式:

import os for host in nr.inventory.hosts.values(): host.password = os.getenv("DEVICE_PASSWORD", host.password)

hosts.yaml里只保留用户名和 IP,真正的密码从环境变量读取。CI/CD 里可以通过 CI 变量或 Secret 机制注入,本地开发自己 export 一下就行。如果你的环境已经接入了 Vault 等密钥管理平台,也可以写一个transform_function去拉取密钥,这里就不展开了。

5.5 用 Netmiko 解析输出时,优先考虑结构化参数

如果采集 Cisco 设备,netmiko_send_command支持use_textfsm=True,能把很多show命令输出直接解析成结构化数据。比如:

result = nr.filter(F(vendor="cisco")).run( task=netmiko_send_command, command_string="show ip interface brief", use_textfsm=True, )

返回结果就不再是一大段文本,而是一个字典列表,比如:

[ {"intf": "GigabitEthernet0/1", "ipaddr": "192.168.10.1", "status": "up", "proto": "up"}, ... ]

这比写正则舒服太多了。不过前提是 Netmiko 内置的 TextFSM 模板支持该命令,不支持的只能退回到普通文本解析。

5.6 大结果集打印会刷屏,先写文件再分析

调试时用print_result很方便,但上百台设备的结果打印出来会非常长,尤其是包含完整命令输出的场景。我现在调试阶段会先打印失败设备清单和耗时,具体输出直接落盘:

import time start = time.perf_counter() result = nr.run(task=collect_system_info) elapsed = time.perf_counter() - start print(f"总耗时: {elapsed:.2f} 秒") print(f"失败设备: {list(nr.data.failed_hosts)}")

等确认设备全通、命令都正确之后,再去看 JSON 文件里的详细结果。这样既不会刷屏,也不会漏掉关键信息。

最后再分享一个实际操作中的体会

Nornir 的入门门槛其实不高,真正花时间的不是框架本身,而是 Inventory 设计和任务函数的边界设计。我刚上手的时候总觉得先想办法把所有设备连上就赢了,结果后面每次加采集项都要改主流程,维护成本很高。后来我把任务函数收敛成“按平台分发命令、返回字典”这个模式之后,整个脚本就稳定下来了。

如果你正准备做网络设备信息收集,建议从一个小范围开始:先 3 台设备、2 个命令、JSON 落盘跑通,再加过滤、加并发、加定时。不要一开始就追求覆盖所有厂商和所有命令。后面还可以在这个基础上继续做配置备份、版本合规检查、配置漂移对比,Nornir 这套架子都不用换,只需要往里面加任务函数就行。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询