使用pysnmp解码SNMP v3陷阱时内存泄漏问题排查求助

问题描述 #

我们用pysnmp解码SNMPv3陷阱以提取securityEngineId，但当大量陷阱持续涌入时，应用内存占用不断攀升，最终触达Docker容器内存上限；即使扩容容器内存，内存仍会持续增长。尝试过手动调用gc.collect()回收内存，但完全无法阻止内存增长，已通过docker stats命令确认该现象。

核心解码代码如下：

if int(msg_version) == 3:
    msg, rest_of_whole_msg = decoder.decode(wholeMsg, asn1Spec=v3Message())
    security_parameters = msg.getComponentByPosition(0)
    security_parameters, rest = decoder.decode(
            security_parameters, asn1Spec=UsmSecurityParameters()
        )

2024年5月29日更新：

• 问题仅在处理SNMPv3陷阱时出现，SNMPv2c等版本无此现象
• 初步怀疑问题源于pysnmp，但移除上述解码代码后，内存增长问题仍未解决
• 期望应用内存稳定在启动基础占用加陷阱处理的合理范围内，而非线性持续增长

可能的原因及排查/解决建议 #

1. pysnmp内部状态缓存未自动清理 #

pysnmp处理SNMPv3时会维护USM用户缓存、引擎ID缓存、nonce值追踪等上下文状态，这些缓存默认不会自动过期，当大量不同代理发送陷阱时，缓存会持续膨胀。

• 检查是否复用了SnmpEngine、decoder等核心实例，避免每次处理陷阱都创建新实例（重复创建会导致资源无法回收）
• 尝试手动清理内部缓存：

  # 清理SnmpEngine的MIB缓存
  snmp_engine.msgAndPduDsp.mibInstrumController.mibBuilder.clearCache()
  # 清理USM用户缓存
  from pysnmp.hlapi.v3arch import UsmUserData
  UsmUserData.clear_cache()
  

2. 解码对象被意外持有引用 #

解码返回的msg、security_parameters等对象如果被全局变量、回调闭包、未清理的容器（如列表/字典）持有引用，会导致GC无法回收。

• 检查代码中是否有将解码对象存储到全局容器且未及时清理的逻辑
• 处理完陷阱后显式解除引用：

  if int(msg_version) == 3:
      msg, rest_of_whole_msg = decoder.decode(wholeMsg, asn1Spec=v3Message())
      security_parameters = msg.getComponentByPosition(0)
      security_parameters, rest = decoder.decode(
              security_parameters, asn1Spec=UsmSecurityParameters()
          )
      # 显式删除引用，帮助GC回收
      del msg, rest_of_whole_msg, rest
  

3. pysnmp版本存在已知泄漏bug #

部分旧版本的pysnmp在处理大量SNMPv3陷阱时存在内存泄漏问题，尤其是在USM上下文处理逻辑中。

• 升级pysnmp到最新稳定版：pip install --upgrade pysnmp
• 检查pysnmp官方仓库的issue记录，确认是否有同类泄漏问题已被修复

4. 陷阱验证逻辑的隐藏资源泄漏 #

你提到用pysnmp做代理陷阱验证，验证逻辑可能持有了未清理的资源（如会话上下文、密钥缓存、验证状态等），即使移除了解码代码，这部分资源仍在占用内存。

• 检查验证相关代码，是否有全局存储的代理信息或验证状态，处理完后及时清理
• 确认验证过程中是否创建了持久化会话对象，未显式关闭

5. 用tracemalloc定位具体泄漏点 #

如果以上方法都无效，直接用Python内置的tracemalloc追踪内存分配，找到具体的泄漏来源：

import tracemalloc

# 启动内存追踪
tracemalloc.start()

# 在陷阱处理循环中定期打印内存快照
def check_memory():
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    print("\n[Top 10内存分配点]")
    for stat in top_stats[:10]:
        print(stat)

# 比如每处理1000个陷阱调用一次check_memory()

内容的提问来源于stack exchange，提问作者justTesting