Python3 easysnmp查询多交换机时随机超时，与查询顺序相关

我用easysnmp.Session查询3台交换机的SNMPv3数据时，出现随机超时问题，超时情况和查询顺序有关。每台交换机都能偶尔响应，说明设备本身没问题；但用bash脚本连续查询时全部正常。

测试代码 #

import easysnmp

switches_dns1 = {
            "switch1": "switch-mBvk1-1.obis.ns.nl",
            "switch2": "switch-mBvk1-2.obis.ns.nl",
            "switch3": "switch-abv5-1.obis.ns.nl"
            }

switches_dns2 = {
            "switch3": "switch-abv5-1.obis.ns.nl",
            "switch1": "switch-mBvk1-1.obis.ns.nl",
            "switch2": "switch-mBvk1-2.obis.ns.nl"          
            }

switches_dns3 = {
            "switch2": "switch-mBvk1-2.obis.ns.nl",
            "switch3": "switch-abv5-1.obis.ns.nl",
            "switch1": "switch-mBvk1-1.obis.ns.nl"                       
            }

switches_dns4 = {
            "switch2": "switch-abv5-1.obis.ns.nl",
            "switch3": "switch-mBvk1-2.obis.ns.nl",
            "switch1": "switch-mBvk1-1.obis.ns.nl"                       
            }

print("====> Testcase 1")   
for switch, hostname_dns in  switches_dns1.items():
    print(switch)
    try:
        snmp_session = easysnmp.Session(hostname=hostname_dns, security_username='test_user', auth_protocol='MD5', auth_password='blahblahblah', version=3, security_level='auth_with_privacy', privacy_password='blahblahblah', privacy_protocol='DES')
        result=snmp_session.get('.1.3.6.1.4.1.37072.302.3.1.1.3.6.0')
        print("Switch: {}, Result: {}".format(switch, result))
    except easysnmp.exceptions.EasySNMPTimeoutError as err:
        print("Switch: {}, result: TIMEOUT ERROR".format(switch))

测试结果 #

调整查询顺序做了四次测试，每台交换机都有成功响应的情况；bash脚本执行结果全部正常：

./test.sh
SNMPv2-SMI::enterprises.37072.302.3.1.1.3.6.0 = IpAddress: 10.160.59.131
SNMPv2-SMI::enterprises.37072.302.3.1.1.3.6.0 = IpAddress: 10.160.59.132
SNMPv2-SMI::enterprises.37072.302.3.1.1.3.6.0 = IpAddress: 10.160.59.133

已尝试的方法 #

• 在SNMP调用间添加sleep，无效
• 查找easysnmp的Session.close()方法，未找到
• 因文档不推荐，未尝试update_session()方法

解决方案 #

1. 复用Session对象（修改hostname切换目标） #

循环内重复创建Session可能导致底层套接字资源泄漏，复用一个Session并动态修改hostname可以避免这个问题：

import easysnmp

switches_dns1 = {
            "switch1": "switch-mBvk1-1.obis.ns.nl",
            "switch2": "switch-mBvk1-2.obis.ns.nl",
            "switch3": "switch-abv5-1.obis.ns.nl"
            }

# 初始化基础Session，只创建一次
base_session = easysnmp.Session(
    security_username='test_user',
    auth_protocol='MD5',
    auth_password='blahblahblah',
    version=3,
    security_level='auth_with_privacy',
    privacy_password='blahblahblah',
    privacy_protocol='DES',
    timeout=3,
    retries=2
)

print("====> Testcase 1")   
for switch, hostname_dns in switches_dns1.items():
    print(switch)
    try:
        # 更新目标主机
        base_session.hostname = hostname_dns
        result = base_session.get('.1.3.6.1.4.1.37072.302.3.1.1.3.6.0')
        print("Switch: {}, Result: {}".format(switch, result))
    except easysnmp.exceptions.EasySNMPTimeoutError as err:
        print("Switch: {}, result: TIMEOUT ERROR".format(switch))

2. 增加超时与重试次数 #

默认超时时间较短，显式设置更长的超时和重试次数，可降低网络波动导致的随机超时：

snmp_session = easysnmp.Session(
    hostname=hostname_dns,
    security_username='test_user',
    auth_protocol='MD5',
    auth_password='blahblahblah',
    version=3,
    security_level='auth_with_privacy',
    privacy_password='blahblahblah',
    privacy_protocol='DES',
    timeout=5,  # 秒
    retries=3   # 重试次数
)

3. 强制释放底层会话资源 #

通过ctypes调用net-snmp的snmp_close函数，强制释放底层资源，避免套接字占用：

import easysnmp
import ctypes

# 加载net-snmp库（Linux下为libnetsnmp.so，Windows下需调整路径）
netsnmp = ctypes.CDLL('libnetsnmp.so')

switches_dns1 = {
            "switch1": "switch-mBvk1-1.obis.ns.nl",
            "switch2": "switch-mBvk1-2.obis.ns.nl",
            "switch3": "switch-abv5-1.obis.ns.nl"
            }

print("====> Testcase 1")   
for switch, hostname_dns in switches_dns1.items():
    print(switch)
    snmp_session = None
    try:
        snmp_session = easysnmp.Session(
            hostname=hostname_dns,
            security_username='test_user',
            auth_protocol='MD5',
            auth_password='blahblahblah',
            version=3,
            security_level='auth_with_privacy',
            privacy_password='blahblahblah',
            privacy_protocol='DES'
        )
        result = snmp_session.get('.1.3.6.1.4.1.37072.302.3.1.1.3.6.0')
        print("Switch: {}, Result: {}".format(switch, result))
    except easysnmp.exceptions.EasySNMPTimeoutError as err:
        print("Switch: {}, result: TIMEOUT ERROR".format(switch))
    finally:
        # 无论成功失败，都关闭底层会话
        if snmp_session:
            netsnmp.snmp_close(snmp_session.session_ptr)

4. 并行查询（线程/进程） #

用线程创建独立的查询会话，避免同一进程内的套接字资源竞争：

import easysnmp
import threading

def query_switch(switch, hostname_dns):
    try:
        snmp_session = easysnmp.Session(
            hostname=hostname_dns,
            security_username='test_user',
            auth_protocol='MD5',
            auth_password='blahblahblah',
            version=3,
            security_level='auth_with_privacy',
            privacy_password='blahblahblah',
            privacy_protocol='DES'
        )
        result = snmp_session.get('.1.3.6.1.4.1.37072.302.3.1.1.3.6.0')
        print("Switch: {}, Result: {}".format(switch, result))
    except easysnmp.exceptions.EasySNMPTimeoutError as err:
        print("Switch: {}, result: TIMEOUT ERROR".format(switch))

switches_dns1 = {
            "switch1": "switch-mBvk1-1.obis.ns.nl",
            "switch2": "switch-mBvk1-2.obis.ns.nl",
            "switch3": "switch-abv5-1.obis.ns.nl"
            }

print("====> Testcase 1")   
threads = []
for switch, hostname_dns in switches_dns1.items():
    t = threading.Thread(target=query_switch, args=(switch, hostname_dns))
    threads.append(t)
    t.start()

# 等待所有线程完成
for t in threads:
    t.join()

内容的提问来源于stack exchange，提问作者7th Sin