3. IP层与路由诊断
说实话,SNMP 这个协议在网工圈里已经活了几十年了。我刚开始做网络监控那会儿,最头疼的就是怎么把设备上的 CPU、内存、接口流量这些数据给"掏"出来。后来发现,Python 的 pysnmp 库就是干这个的利器。
这一章,咱们就动手写代码。我会带着你一步步实现 SNMP 的 Get、GetBulk 和 Walk 操作。嗯,代码跑通了,你就能从任何支持 SNMP 的设备上采集数据了。
3.1 环境准备:装好 pysnmp
先装库。我个人习惯用 pip 直接装:
pip install pysnmp
如果你用的是 Python 3.7 以上版本,基本不会遇到兼容问题。我曾在 CentOS 6 的老机器上折腾过,那才叫一个酸爽——Python 版本太低,pysnmp 死活装不上。所以,建议你至少用 Python 3.8。
3.2 SNMP Get:单点查询,精准打击
SNMP Get 是最基础的操作。说白了,就是问设备:"嘿,你这个 OID 的值是多少?" 设备会直接回答你。
来看一个最简单的例子,查询设备的系统描述:
from pysnmp.hlapi import *
def snmp_get(host, community, oid):
errorIndication, errorStatus, errorIndex, varBinds = next(
getCmd(SnmpEngine(),
CommunityData(community),
UdpTransportTarget((host, 161)),
ContextData(),
ObjectType(ObjectIdentity(oid)))
)
if errorIndication:
print(f"错误: {errorIndication}")
return None
elif errorStatus:
print(f"SNMP错误: {errorStatus.prettyPrint()}")
return None
else:
for varBind in varBinds:
return varBind[1].prettyPrint()
# 查询系统描述 OID: 1.3.6.1.2.1.1.1.0
result = snmp_get('192.168.1.1', 'public', '1.3.6.1.2.1.1.1.0')
print(f"系统描述: {result}")
这段代码的核心逻辑其实就四步:
- 创建 SNMP 引擎
- 指定团体名(相当于密码)
- 指定目标设备 IP 和端口
- 发送 Get 请求,拿回结果
小提示: 团体名 'public' 是只读的,'private' 是读写的。生产环境千万别用默认的 public,我见过太多因为没改团体名被扫到然后被篡改配置的案例了。
3.3 SNMP GetBulk:批量获取,效率翻倍
Get 一次只能查一个 OID。如果你要查 CPU、内存、接口流量三个指标,就得发三次请求。这太慢了。
GetBulk 就是干这个的。它允许你一次查询多个 OID,设备会一次性返回所有结果。我在监控 2000 台交换机的时候,全靠 GetBulk 撑着的,否则请求量能把设备 CPU 打满。
def snmp_getbulk(host, community, oids, max_repetitions=10):
oid_list = [ObjectType(ObjectIdentity(oid)) for oid in oids]
errorIndication, errorStatus, errorIndex, varBinds = next(
bulkCmd(SnmpEngine(),
CommunityData(community),
UdpTransportTarget((host, 161)),
ContextData(),
0, max_repetitions,
*oid_list)
)
if errorIndication:
print(f"错误: {errorIndication}")
return None
results = {}
for varBind in varBinds:
oid = varBind[0].prettyPrint()
value = varBind[1].prettyPrint()
results[oid] = value
return results
# 同时查询系统描述、系统运行时间、系统联系人
oids = ['1.3.6.1.2.1.1.1.0',
'1.3.6.1.2.1.1.3.0',
'1.3.6.1.2.1.1.4.0']
result = snmp_getbulk('192.168.1.1', 'public', oids)
for oid, value in result.items():
print(f"{oid}: {value}")
注意: max_repetitions 参数控制每次批量返回的最大行数。设太大,设备可能扛不住;设太小,又得发多次请求。我一般设 10-20,具体看设备性能。曾经有个客户把 max_repetitions 设成 1000,直接把核心交换机搞重启了……
3.4 SNMP Walk:遍历整个MIB树
Walk 操作就像是在 MIB 树里"散步"。你给一个根 OID,它会把这个节点下的所有子节点都遍历出来。这在采集接口流量时特别有用——你不需要知道每个接口的 OID,Walk 会自动找到所有接口。
def snmp_walk(host, community, root_oid):
results = []
for errorIndication, errorStatus, errorIndex, varBinds in \
nextCmd(SnmpEngine(),
CommunityData(community),
UdpTransportTarget((host, 161)),
ContextData(),
ObjectType(ObjectIdentity(root_oid)),
lexicographicMode=False):
if errorIndication:
print(f"错误: {errorIndication}")
break
elif errorStatus:
print(f"SNMP错误: {errorStatus.prettyPrint()}")
break
else:
for varBind in varBinds:
oid = varBind[0].prettyPrint()
value = varBind[1].prettyPrint()
results.append((oid, value))
return results
# 遍历接口描述表 (ifDescr)
interfaces = snmp_walk('192.168.1.1', 'public', '1.3.6.1.2.1.2.2.1.2')
for oid, desc in interfaces:
print(f"{oid} -> {desc}")
这里有个坑:lexicographicMode=False 这个参数。如果不设,pysnmp 默认按字典序遍历,可能会漏掉一些节点。我踩过这个坑,查了两天才发现是参数问题。
3.5 实战:采集CPU、内存、接口流量
好了,理论知识够了。咱们直接上实战,写一个完整的采集脚本。
先看看常见的 OID 表:
| 指标 | OID | 说明 |
|---|---|---|
| CPU 使用率 | 1.3.6.1.4.1.9.9.109.1.1.1.1.7.1 | Cisco 设备专用 |
| 内存使用率 | 1.3.6.1.4.1.9.9.48.1.1.1.5.1 | Cisco 设备专用 |
| 接口入流量 | 1.3.6.1.2.1.2.2.1.10 | ifInOctets,字节数 |
| 接口出流量 | 1.3.6.1.2.1.2.2.1.16 | ifOutOctets,字节数 |
重要提醒: 不同厂商的 OID 不一样。上面列的是 Cisco 的。华为用 hh3c 私有 MIB,H3C 用 h3c 私有 MIB。采集前一定要先确认设备型号对应的 OID。我建议你先用 MIB Browser 手动查一遍,确认 OID 有效再写代码。
完整采集脚本:
from pysnmp.hlapi import *
import time
class SNMPCollector:
def __init__(self, host, community):
self.host = host
self.community = community
def get_cpu_usage(self):
"""采集CPU使用率"""
oid = '1.3.6.1.4.1.9.9.109.1.1.1.1.7.1'
result = self._snmp_get(oid)
return int(result) if result else None
def get_memory_usage(self):
"""采集内存使用率"""
# 先获取已用内存
used_oid = '1.3.6.1.4.1.9.9.48.1.1.1.5.1'
# 再获取总内存
total_oid = '1.3.6.1.4.1.9.9.48.1.1.1.6.1'
used = int(self._snmp_get(used_oid) or 0)
total = int(self._snmp_get(total_oid) or 1)
return (used / total) * 100
def get_interface_traffic(self, if_index):
"""采集指定接口的流量"""
# ifInOctets
in_oid = f'1.3.6.1.2.1.2.2.1.10.{if_index}'
# ifOutOctets
out_oid = f'1.3.6.1.2.1.2.2.1.16.{if_index}'
in_bytes = int(self._snmp_get(in_oid) or 0)
out_bytes = int(self._snmp_get(out_oid) or 0)
return in_bytes, out_bytes
def _snmp_get(self, oid):
errorIndication, errorStatus, errorIndex, varBinds = next(
getCmd(SnmpEngine(),
CommunityData(self.community),
UdpTransportTarget((self.host, 161)),
ContextData(),
ObjectType(ObjectIdentity(oid)))
)
if errorIndication or errorStatus:
return None
return varBinds[0][1].prettyPrint()
# 使用示例
collector = SNMPCollector('192.168.1.1', 'public')
cpu = collector.get_cpu_usage()
mem = collector.get_memory_usage()
in_traffic, out_traffic = collector.get_interface_traffic(1)
print(f"CPU: {cpu}%")
print(f"内存: {mem:.2f}%")
print(f"接口1入流量: {in_traffic} 字节")
print(f"接口1出流量: {out_traffic} 字节")
3.6 核心流程:一张图看懂
下面这张图,把整个 SNMP 采集的流程串起来了。你照着这个思路写代码,不会乱。
3.7 避坑指南
写 SNMP 采集代码,有几个坑我不得不提:
- 超时问题: 默认超时是 1 秒。网络不好的时候,经常超时。我建议设成 3-5 秒:
UdpTransportTarget((host, 161), timeout=3, retries=2) - OID 写错: 少写一个点,或者多写一个 0,结果就完全不对。我曾经因为把 .1.3.6 写成了 .1.3.06,排查了半小时。
- 返回值类型: SNMP 返回的值可能是整数、字符串、IP 地址等。用 prettyPrint() 转成字符串最保险。
- 设备负载: 别太频繁地采集。我一般设 5 分钟一次。曾经有个项目设成 10 秒一次,结果设备 CPU 飙到 90%,被客户骂惨了。
我的经验: 刚开始做 SNMP 采集,建议先用 MIB Browser 手动验证 OID 和返回值。等确认无误了,再写代码。这样能省下大量调试时间。
好了,这一章的内容就到这里。代码你都拿到了,找个设备试试吧。记住,先从小处着手——先跑通 Get,再试 GetBulk,最后上 Walk。一步步来,稳得很。