fix(snmp): close SnmpEngine after each poll to stop fd leak
A fresh SnmpEngine() was created on every poll_device() call and never closed. pysnmp opens a UDP transport socket per engine and doesn't release it on GC, so each scheduler tick (default 60s, per device) leaked a file descriptor. Over hours of polling the process hit its fd ceiling and the listening socket could no longer accept connections — OSError: [Errno 24] Too many open files on socket.accept(), locking up the app. Wrap the engine in try/finally and release its transport socket via a new _close_engine() helper that probes both pysnmp API shapes (6.2.x lextudio camelCase, canonical 7.x snake_case); all close paths are best-effort so a failed close never breaks the poll loop. Regression tests cover both shapes and the never-raises contract. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
+81
-39
@@ -28,6 +28,44 @@ def _mp_model(version: str) -> int:
|
|||||||
return 0 if version == "1" else 1
|
return 0 if version == "1" else 1
|
||||||
|
|
||||||
|
|
||||||
|
def _close_engine(engine) -> None:
|
||||||
|
"""Release the engine's UDP transport socket.
|
||||||
|
|
||||||
|
pysnmp opens a UDP socket per ``SnmpEngine`` and never closes it on its own.
|
||||||
|
Since we build a fresh engine for every poll, an unclosed engine leaks one
|
||||||
|
file descriptor each scheduler tick; over hours of polling that exhausts the
|
||||||
|
process fd limit (``OSError: [Errno 24] Too many open files``), which also
|
||||||
|
takes down the app's listening socket. So close it explicitly here.
|
||||||
|
|
||||||
|
The close method/attribute names differ across pysnmp majors (6.2.x lextudio
|
||||||
|
is camelCase, canonical 7.x is snake_case), so probe for whatever exists.
|
||||||
|
All paths are best-effort — a failed close must never break the poll loop.
|
||||||
|
"""
|
||||||
|
# 7.x exposes a convenience close directly on the engine.
|
||||||
|
for meth in ("close_dispatcher", "closeDispatcher"):
|
||||||
|
fn = getattr(engine, meth, None)
|
||||||
|
if callable(fn):
|
||||||
|
try:
|
||||||
|
fn()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return
|
||||||
|
# 6.2.x: go through the transport dispatcher.
|
||||||
|
for attr in ("transport_dispatcher", "transportDispatcher"):
|
||||||
|
dispatcher = getattr(engine, attr, None)
|
||||||
|
if dispatcher is None:
|
||||||
|
continue
|
||||||
|
for meth in ("close_dispatcher", "closeDispatcher"):
|
||||||
|
fn = getattr(dispatcher, meth, None)
|
||||||
|
if callable(fn):
|
||||||
|
try:
|
||||||
|
fn()
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return
|
||||||
|
return
|
||||||
|
|
||||||
|
|
||||||
async def poll_device(
|
async def poll_device(
|
||||||
host: str,
|
host: str,
|
||||||
port: int,
|
port: int,
|
||||||
@@ -81,49 +119,53 @@ async def poll_device(
|
|||||||
|
|
||||||
engine = SnmpEngine()
|
engine = SnmpEngine()
|
||||||
|
|
||||||
# Same host/port for every OID on this device, so build the transport once.
|
# Always release the engine's UDP socket — see _close_engine for why.
|
||||||
try:
|
try:
|
||||||
if _transport_is_async:
|
# Same host/port for every OID on this device, so build the transport once.
|
||||||
transport = await UdpTransportTarget.create((host, port), timeout=5, retries=1)
|
|
||||||
else:
|
|
||||||
transport = UdpTransportTarget((host, port), timeout=5, retries=1)
|
|
||||||
except Exception as exc:
|
|
||||||
logger.debug("SNMP transport setup failed for %s:%s: %s", host, port, exc)
|
|
||||||
return {}
|
|
||||||
|
|
||||||
results: dict[str, float] = {}
|
|
||||||
|
|
||||||
for oid_cfg in oids:
|
|
||||||
oid = oid_cfg["oid"]
|
|
||||||
label = oid_cfg.get("label") or oid
|
|
||||||
scale = float(oid_cfg.get("scale", 1.0))
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
error_indication, error_status, error_index, var_binds = await _get_cmd(
|
if _transport_is_async:
|
||||||
engine,
|
transport = await UdpTransportTarget.create((host, port), timeout=5, retries=1)
|
||||||
CommunityData(community, mpModel=_mp_model(version)),
|
else:
|
||||||
transport,
|
transport = UdpTransportTarget((host, port), timeout=5, retries=1)
|
||||||
ContextData(),
|
|
||||||
ObjectType(ObjectIdentity(oid)),
|
|
||||||
)
|
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.debug("SNMP GET %s@%s OID %s failed: %s", host, port, oid, exc)
|
logger.debug("SNMP transport setup failed for %s:%s: %s", host, port, exc)
|
||||||
continue
|
return {}
|
||||||
|
|
||||||
if error_indication:
|
results: dict[str, float] = {}
|
||||||
logger.debug("SNMP error %s@%s OID %s: %s", host, port, oid, error_indication)
|
|
||||||
continue
|
for oid_cfg in oids:
|
||||||
if error_status:
|
oid = oid_cfg["oid"]
|
||||||
logger.debug("SNMP status %s@%s OID %s: %s at %s",
|
label = oid_cfg.get("label") or oid
|
||||||
host, port, oid, error_status.prettyPrint(),
|
scale = float(oid_cfg.get("scale", 1.0))
|
||||||
error_index and var_binds[int(error_index) - 1][0] or "?")
|
|
||||||
continue
|
|
||||||
|
|
||||||
for _, val in var_binds:
|
|
||||||
try:
|
try:
|
||||||
results[label] = float(val) * scale
|
error_indication, error_status, error_index, var_binds = await _get_cmd(
|
||||||
except (TypeError, ValueError):
|
engine,
|
||||||
# Non-numeric type (e.g. OctetString description) — skip.
|
CommunityData(community, mpModel=_mp_model(version)),
|
||||||
logger.debug("SNMP non-numeric value for %s label=%s: %r", oid, label, val)
|
transport,
|
||||||
|
ContextData(),
|
||||||
|
ObjectType(ObjectIdentity(oid)),
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
logger.debug("SNMP GET %s@%s OID %s failed: %s", host, port, oid, exc)
|
||||||
|
continue
|
||||||
|
|
||||||
return results
|
if error_indication:
|
||||||
|
logger.debug("SNMP error %s@%s OID %s: %s", host, port, oid, error_indication)
|
||||||
|
continue
|
||||||
|
if error_status:
|
||||||
|
logger.debug("SNMP status %s@%s OID %s: %s at %s",
|
||||||
|
host, port, oid, error_status.prettyPrint(),
|
||||||
|
error_index and var_binds[int(error_index) - 1][0] or "?")
|
||||||
|
continue
|
||||||
|
|
||||||
|
for _, val in var_binds:
|
||||||
|
try:
|
||||||
|
results[label] = float(val) * scale
|
||||||
|
except (TypeError, ValueError):
|
||||||
|
# Non-numeric type (e.g. OctetString description) — skip.
|
||||||
|
logger.debug("SNMP non-numeric value for %s label=%s: %r", oid, label, val)
|
||||||
|
|
||||||
|
return results
|
||||||
|
finally:
|
||||||
|
_close_engine(engine)
|
||||||
|
|||||||
@@ -30,3 +30,47 @@ def test_poll_device_without_pysnmp_returns_empty(monkeypatch):
|
|||||||
poller.poll_device("192.0.2.1", 161, "public", "2c", [{"oid": "1.3.6.1.2.1.1.3.0"}])
|
poller.poll_device("192.0.2.1", 161, "public", "2c", [{"oid": "1.3.6.1.2.1.1.3.0"}])
|
||||||
)
|
)
|
||||||
assert out == {}
|
assert out == {}
|
||||||
|
|
||||||
|
|
||||||
|
# --- _close_engine: the fd-leak guard ---------------------------------------
|
||||||
|
# A fresh SnmpEngine per poll leaks its UDP socket unless closed; _close_engine
|
||||||
|
# must release it across both pysnmp API shapes. These fakes stand in for the
|
||||||
|
# engine since the unit lane has no pysnmp.
|
||||||
|
|
||||||
|
def test_close_engine_uses_engine_level_close_7x():
|
||||||
|
# canonical pysnmp 7.x: close method lives directly on the engine.
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
class Engine:
|
||||||
|
def close_dispatcher(self):
|
||||||
|
calls.append("engine")
|
||||||
|
|
||||||
|
poller._close_engine(Engine())
|
||||||
|
assert calls == ["engine"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_close_engine_falls_back_to_dispatcher_6x():
|
||||||
|
# pysnmp-lextudio 6.2.x: no engine-level close; go through the dispatcher.
|
||||||
|
calls = []
|
||||||
|
|
||||||
|
class Dispatcher:
|
||||||
|
def closeDispatcher(self):
|
||||||
|
calls.append("dispatcher")
|
||||||
|
|
||||||
|
class Engine:
|
||||||
|
transportDispatcher = Dispatcher()
|
||||||
|
|
||||||
|
poller._close_engine(Engine())
|
||||||
|
assert calls == ["dispatcher"]
|
||||||
|
|
||||||
|
|
||||||
|
def test_close_engine_never_raises():
|
||||||
|
# A failing close must not break the poll loop.
|
||||||
|
class Engine:
|
||||||
|
def close_dispatcher(self):
|
||||||
|
raise OSError("boom")
|
||||||
|
|
||||||
|
poller._close_engine(Engine()) # no exception
|
||||||
|
|
||||||
|
# No close path at all (defensive) is also fine.
|
||||||
|
poller._close_engine(object())
|
||||||
|
|||||||
Reference in New Issue
Block a user