72 lines
3.3 KiB
Python
72 lines
3.3 KiB
Python
import logging
|
|
import time
|
|
from datetime import datetime
|
|
from urllib.parse import urljoin
|
|
|
|
import httpx
|
|
|
|
from .logging_config import redact_url
|
|
from .models import CheckResult, MonitorStatus, utcnow
|
|
from .security import Resolver, UnsafeTarget, system_resolver, validate_target
|
|
|
|
logger = logging.getLogger("monitor.checker")
|
|
|
|
|
|
class EndpointChecker:
|
|
def __init__(
|
|
self,
|
|
client: httpx.AsyncClient,
|
|
timeout: float,
|
|
max_redirects: int,
|
|
max_response_bytes: int,
|
|
resolver: Resolver = system_resolver,
|
|
) -> None:
|
|
self.client = client
|
|
self.timeout = timeout
|
|
self.max_redirects = max_redirects
|
|
self.max_response_bytes = max_response_bytes
|
|
self.resolver = resolver
|
|
|
|
async def check(self, monitor_id: str, initial_url: str) -> CheckResult:
|
|
started = time.perf_counter()
|
|
checked_at: datetime = utcnow()
|
|
current_url = initial_url
|
|
try:
|
|
for redirect_count in range(self.max_redirects + 1):
|
|
await validate_target(current_url, self.resolver)
|
|
async with self.client.stream(
|
|
"GET", current_url, follow_redirects=False, timeout=self.timeout,
|
|
headers={"User-Agent": "endpoint-monitor/1"},
|
|
) as response:
|
|
consumed = 0
|
|
async for chunk in response.aiter_bytes():
|
|
consumed += len(chunk)
|
|
if consumed >= self.max_response_bytes:
|
|
break
|
|
if response.is_redirect and response.headers.get("location"):
|
|
if redirect_count == self.max_redirects:
|
|
return self._result(started, checked_at, MonitorStatus.error, error="too_many_redirects")
|
|
current_url = urljoin(str(response.url), response.headers["location"])
|
|
continue
|
|
status = MonitorStatus.up if 200 <= response.status_code < 400 else MonitorStatus.down
|
|
return self._result(started, checked_at, status, response.status_code)
|
|
except UnsafeTarget as exc:
|
|
result = self._result(started, checked_at, MonitorStatus.error, error=str(exc))
|
|
except httpx.TimeoutException:
|
|
result = self._result(started, checked_at, MonitorStatus.error, error="timeout")
|
|
except httpx.HTTPError:
|
|
result = self._result(started, checked_at, MonitorStatus.error, error="network_error")
|
|
except Exception:
|
|
logger.exception("check_unexpected_error", extra={"monitor_id": monitor_id, "url": redact_url(initial_url)})
|
|
result = self._result(started, checked_at, MonitorStatus.error, error="internal_error")
|
|
self._log(monitor_id, initial_url, result)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _result(started: float, checked_at: datetime, status: MonitorStatus, http_status: int | None = None, error: str | None = None) -> CheckResult:
|
|
return CheckResult(status=status, checked_at=checked_at, latency_ms=max(0.0, (time.perf_counter() - started) * 1000), http_status=http_status, error=error)
|
|
|
|
@staticmethod
|
|
def _log(monitor_id: str, url: str, result: CheckResult) -> None:
|
|
logger.info("check_completed", extra={"monitor_id": monitor_id, "url": url, **result.model_dump()})
|