Use ping RTT for SNMP checker latency

Ping the host first to check availability and measure actual network
latency. Only proceed with SNMP query if the host is reachable.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-01-26 19:49:26 +01:00
parent 624e3e7633
commit b1ba8b9fab

View File

@@ -6,7 +6,9 @@ récupérer sa température.
"""
import asyncio
import time
import platform
import re
import subprocess
from pysnmp.hlapi.v1arch.asyncio import (
Slim,
@@ -49,12 +51,43 @@ class SnmpChecker(BaseChecker):
"""
return asyncio.run(self._async_check())
def _ping(self, host: str, timeout: int) -> tuple[bool, float | None]:
"""
Exécute un ping vers l'hôte pour vérifier sa disponibilité.
Args:
host: Adresse IP ou nom d'hôte.
timeout: Délai d'attente en secondes.
Returns:
Tuple (succès, RTT en ms ou None).
"""
if platform.system().lower() == "windows":
cmd = ["ping", "-n", "1", "-w", str(timeout * 1000), host]
else:
cmd = ["ping", "-c", "1", "-W", str(timeout), host]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout + 5
)
if result.returncode == 0:
match = re.search(r"time[=<]([\d.]+)\s*ms", result.stdout)
return True, float(match.group(1)) if match else None
return False, None
except Exception:
return False, None
async def _async_check(self) -> CheckResult:
"""
Exécute la requête SNMP de manière asynchrone.
Exécute la vérification SNMP de manière asynchrone.
Interroge l'OID principal et optionnellement l'OID de température
en une seule requête SNMP GET.
Effectue d'abord un ping pour vérifier la disponibilité et mesurer
la latence, puis interroge l'OID principal et optionnellement
l'OID de température via SNMP.
Returns:
CheckResult contenant le résultat et la température si configurée.
@@ -66,14 +99,22 @@ class SnmpChecker(BaseChecker):
temperature_oid = self.config.get("temperature_oid")
timeout_val = self.config.get("timeout", 5)
# Ping d'abord pour vérifier la disponibilité et mesurer la latence
ping_success, response_time = self._ping(host, timeout_val)
if not ping_success:
return CheckResult(
success=False,
message="Host is unreachable",
response_time=None
)
# Requête SNMP
try:
with Slim() as slim:
# Construit la liste des OIDs à interroger
oids = [ObjectType(ObjectIdentity(oid))]
if temperature_oid:
oids.append(ObjectType(ObjectIdentity(temperature_oid)))
start = time.time()
error_indication, error_status, error_index, var_binds = await slim.get(
community,
host,
@@ -82,30 +123,27 @@ class SnmpChecker(BaseChecker):
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
response_time=response_time
)
elif error_status:
return CheckResult(
success=False,
message=f"SNMP error: {error_status.prettyPrint()}",
response_time=None
response_time=response_time
)
else:
# Inclut uniquement l'OID principal dans les détails
details = {str(var_binds[0][0]): str(var_binds[0][1])}
# Extrait la température si configurée (deuxième OID)
if temperature_oid and len(var_binds) >= 2:
try:
details["temperature"] = int(var_binds[1][1])
except (ValueError, TypeError):
pass # Ignore si la valeur n'est pas un entier
pass
return CheckResult(
success=True,
@@ -117,5 +155,5 @@ class SnmpChecker(BaseChecker):
return CheckResult(
success=False,
message=f"SNMP error: {e}",
response_time=None
response_time=response_time
)