97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import dataclasses
|
|
import logging
|
|
import time
|
|
from typing import List, Optional
|
|
|
|
import dns.exception
|
|
import requests
|
|
|
|
from . import dns_resolver
|
|
|
|
logger = logging.getLogger("uptime_monitor.checker")
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class CheckResult:
|
|
endpoint_name: str
|
|
host: str
|
|
scheme: str
|
|
port: int
|
|
success: bool
|
|
status_code: Optional[int]
|
|
expected_status: int
|
|
response_time_ms: Optional[float]
|
|
resolved_ip: Optional[str]
|
|
error: Optional[str]
|
|
timestamp_ns: int
|
|
|
|
|
|
def check_endpoint(ep, default_timeout: float, timestamp_ns: int) -> CheckResult:
|
|
timeout = ep.timeout_seconds or default_timeout
|
|
start = time.monotonic()
|
|
|
|
if dns_resolver.is_ip_literal(ep.host):
|
|
# Nothing to resolve — connect straight to the given IP. dns_server
|
|
# is not required/used in this case (see config.py).
|
|
resolved_ip = ep.host
|
|
pin = contextlib.nullcontext()
|
|
else:
|
|
try:
|
|
resolved_ip = dns_resolver.resolve(ep.host, ep.dns_server, timeout)
|
|
except dns.exception.DNSException as e:
|
|
return CheckResult(
|
|
endpoint_name=ep.name, host=ep.host, scheme=ep.scheme, port=ep.port,
|
|
success=False, status_code=None, expected_status=ep.expected_status,
|
|
response_time_ms=None, resolved_ip=None,
|
|
error=f"dns_error: {e}", timestamp_ns=timestamp_ns,
|
|
)
|
|
pin = dns_resolver.pin_resolution(ep.host, resolved_ip)
|
|
|
|
url = f"{ep.scheme}://{dns_resolver.format_host_for_url(ep.host)}:{ep.port}{ep.path}"
|
|
try:
|
|
with pin:
|
|
resp = requests.get(
|
|
url, timeout=timeout, verify=ep.verify_tls, allow_redirects=True
|
|
)
|
|
elapsed_ms = (time.monotonic() - start) * 1000
|
|
success = resp.status_code == ep.expected_status
|
|
return CheckResult(
|
|
endpoint_name=ep.name, host=ep.host, scheme=ep.scheme, port=ep.port,
|
|
success=success, status_code=resp.status_code, expected_status=ep.expected_status,
|
|
response_time_ms=elapsed_ms, resolved_ip=resolved_ip,
|
|
error=None if success else f"unexpected_status_code:{resp.status_code}",
|
|
timestamp_ns=timestamp_ns,
|
|
)
|
|
except requests.RequestException as e:
|
|
elapsed_ms = (time.monotonic() - start) * 1000
|
|
return CheckResult(
|
|
endpoint_name=ep.name, host=ep.host, scheme=ep.scheme, port=ep.port,
|
|
success=False, status_code=None, expected_status=ep.expected_status,
|
|
response_time_ms=elapsed_ms, resolved_ip=resolved_ip,
|
|
error=str(e), timestamp_ns=timestamp_ns,
|
|
)
|
|
|
|
|
|
def log_result(result: CheckResult) -> None:
|
|
level = logging.INFO if result.success else logging.WARNING
|
|
logger.log(
|
|
level,
|
|
"endpoint=%s success=%s status=%s expected=%s ip=%s time_ms=%s error=%s",
|
|
result.endpoint_name, result.success, result.status_code,
|
|
result.expected_status, result.resolved_ip,
|
|
None if result.response_time_ms is None else round(result.response_time_ms, 1),
|
|
result.error,
|
|
)
|
|
|
|
|
|
def check_all(endpoints, default_timeout: float, timestamp_ns: int) -> List[CheckResult]:
|
|
results = []
|
|
for ep in endpoints:
|
|
result = check_endpoint(ep, default_timeout, timestamp_ns)
|
|
log_result(result)
|
|
results.append(result)
|
|
return results
|