FabledSteward/plugins/snmp/poller.py

# plugins/snmp/poller.py
"""
Asynchronous SNMP GET helper.

Requires pysnmp-lextudio (the maintained pysnmp fork), bundled into the Docker
image via the `snmp` extra (`pip install .[snmp]`):
    pip install 'steward[snmp]'

If pysnmp is not installed, poll_device() returns an empty dict and logs a
warning — SNMP polling is then simply disabled, nothing else breaks.
"""
from __future__ import annotations
import logging

logger = logging.getLogger(__name__)


def _pysnmp_available() -> bool:
    try:
        import pysnmp  # noqa: F401
        return True
    except ImportError:
        return False


def _mp_model(version: str) -> int:
    """Map an SNMP version string to pysnmp's mpModel int (0 = v1, 1 = v2c)."""
    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(
    host: str,
    port: int,
    community: str,
    version: str,
    oids: list[dict],
) -> dict[str, float]:
    """Perform an SNMP GET for each OID and return ``{label: float_value}``.

    Non-numeric OIDs (strings, etc.) are skipped. Returns an empty dict on any
    error (unreachable host, wrong community, …) so a flaky device never breaks
    the poll loop.

    pysnmp's HLAPI is asyncio-only as of v6. The import path moved between major
    versions, so we support both rather than let a dependency bump silently
    re-break polling:
      • pysnmp-lextudio 6.2.x → ``pysnmp.hlapi.asyncio`` (``getCmd`` + a directly
        constructed ``UdpTransportTarget``).
      • canonical pysnmp 7.x  → ``pysnmp.hlapi.v3arch.asyncio`` (``get_cmd`` + the
        async ``UdpTransportTarget.create``).
    """
    if not _pysnmp_available():
        logger.warning("pysnmp not installed — SNMP polling disabled. "
                       "Install with: pip install 'steward[snmp]'")
        return {}

    try:
        # canonical pysnmp 7.x
        from pysnmp.hlapi.v3arch.asyncio import (
            CommunityData,
            ContextData,
            ObjectIdentity,
            ObjectType,
            SnmpEngine,
            UdpTransportTarget,
            get_cmd as _get_cmd,
        )
        _transport_is_async = True
    except ImportError:
        # pysnmp-lextudio 6.2.x
        from pysnmp.hlapi.asyncio import (
            CommunityData,
            ContextData,
            ObjectIdentity,
            ObjectType,
            SnmpEngine,
            UdpTransportTarget,
            getCmd as _get_cmd,
        )
        _transport_is_async = False

    engine = SnmpEngine()

    # Always release the engine's UDP socket — see _close_engine for why.
    try:
        # Same host/port for every OID on this device, so build the transport once.
        try:
            if _transport_is_async:
                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:
                error_indication, error_status, error_index, var_binds = await _get_cmd(
                    engine,
                    CommunityData(community, mpModel=_mp_model(version)),
                    transport,
                    ContextData(),
                    ObjectType(ObjectIdentity(oid)),
                )
            except Exception as exc:
                logger.debug("SNMP GET %s@%s OID %s failed: %s", host, port, oid, exc)
                continue

            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)