bvandeusen
be4654fc72
- Add HTTP monitor plugin: endpoint checking with status history, latency tracking, time-range views - Add SNMP plugin: OID polling, device management, metric recording - Add Docker plugin: container status, resource usage, widget - Traefik: access log widget, request chart widget, expanded routes - UniFi: clients widget, devices widget, expanded poll routes - UPS: history widget, additional routes - Update plugin index with new entries Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
# plugins/snmp/poller.py
|
|
"""
|
|
Synchronous SNMP GET helper, run via executor.
|
|
|
|
Requires pysnmp-lextudio (maintained pysnmp fork):
|
|
pip install 'fabledscryer[snmp]'
|
|
|
|
If pysnmp is not installed, poll_device() returns an empty dict and logs a warning.
|
|
"""
|
|
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 version string to pysnmp mpModel integer."""
|
|
return 0 if version == "1" else 1
|
|
|
|
|
|
def poll_device_sync(
|
|
host: str,
|
|
port: int,
|
|
community: str,
|
|
version: str,
|
|
oids: list[dict],
|
|
) -> dict[str, float]:
|
|
"""
|
|
Perform SNMP GET for each OID and return {label: float_value}.
|
|
Non-numeric OIDs (strings, etc.) are skipped.
|
|
Returns empty dict on any error.
|
|
"""
|
|
if not _pysnmp_available():
|
|
logger.warning("pysnmp not installed — SNMP polling disabled. "
|
|
"Install with: pip install 'fabledscryer[snmp]'")
|
|
return {}
|
|
|
|
from pysnmp.hlapi import (
|
|
CommunityData,
|
|
ContextData,
|
|
ObjectIdentity,
|
|
ObjectType,
|
|
SnmpEngine,
|
|
UdpTransportTarget,
|
|
getCmd,
|
|
)
|
|
|
|
results: dict[str, float] = {}
|
|
engine = SnmpEngine()
|
|
|
|
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 = next(
|
|
getCmd(
|
|
engine,
|
|
CommunityData(community, mpModel=_mp_model(version)),
|
|
UdpTransportTarget((host, port), timeout=5, retries=1),
|
|
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
|