Antoine Van Elstraete
f70bf69133
Migrate from deprecated hlapi to v1arch.asyncio with Slim wrapper. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import asyncio
|
|
import time
|
|
|
|
from pysnmp.hlapi.v1arch.asyncio import (
|
|
Slim,
|
|
ObjectIdentity,
|
|
ObjectType,
|
|
)
|
|
|
|
from .base import BaseChecker, CheckResult
|
|
|
|
|
|
class SnmpChecker(BaseChecker):
|
|
def check(self) -> CheckResult:
|
|
return asyncio.run(self._async_check())
|
|
|
|
async def _async_check(self) -> CheckResult:
|
|
host = self.config["host"]
|
|
port = self.config.get("port", 161)
|
|
community = self.config.get("community", "public")
|
|
oid = self.config.get("oid", "1.3.6.1.2.1.1.1.0") # sysDescr
|
|
timeout_val = self.config.get("timeout", 5)
|
|
|
|
start = time.time()
|
|
try:
|
|
with Slim() as slim:
|
|
error_indication, error_status, error_index, var_binds = await slim.get(
|
|
community,
|
|
host,
|
|
port,
|
|
ObjectType(ObjectIdentity(oid)),
|
|
timeout=timeout_val,
|
|
retries=1
|
|
)
|
|
|
|
response_time = (time.time() - start) * 1000 # ms
|
|
|
|
if error_indication:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {error_indication}",
|
|
response_time=None
|
|
)
|
|
elif error_status:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {error_status.prettyPrint()}",
|
|
response_time=None
|
|
)
|
|
else:
|
|
values = {str(oid): str(val) for oid, val in var_binds}
|
|
return CheckResult(
|
|
success=True,
|
|
message="SNMP response OK",
|
|
response_time=response_time,
|
|
details=values
|
|
)
|
|
except Exception as e:
|
|
return CheckResult(
|
|
success=False,
|
|
message=f"SNMP error: {e}",
|
|
response_time=None
|
|
)
|