From cf3a39b50aeb6e3ee5fb8fb41687c7071d34a9c7 Mon Sep 17 00:00:00 2001 From: demo-bot Date: Sun, 9 Aug 2026 16:07:06 +0000 Subject: [PATCH] decomposer: generate deliverable files for Define the service contract and project architecture for the FastAPI endpoint monitoring service.; Implement the typed monitor CRUD API and concurrency-safe in-memory state according to the service design.; Implement secure on-demand endpoint checks with status updates, latency measurement, robust error handling, and redacted structured logs.; Add operational API endpoints and environment-driven runtime configuration to the monitoring service.; Create automated tests for the monitoring service.; Package the service with Docker and developer documentation.; Validate the complete project. --- app/checker.py | 126 +++++++++++++++++++++++++++++-------------------- 1 file changed, 76 insertions(+), 50 deletions(-) diff --git a/app/checker.py b/app/checker.py index 87ff0b8..f40f9ee 100644 --- a/app/checker.py +++ b/app/checker.py @@ -1,71 +1,97 @@ 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 +from app.config import Settings +from app.models import CurrentStatus, MonitorState, utc_now +from app.security import Resolver, UnsafeTarget, redact_url, system_resolver, validate_target -logger = logging.getLogger("monitor.checker") +logger = logging.getLogger("endpoint_monitor.checker") +_REDIRECTS = {301, 302, 303, 307, 308} class EndpointChecker: def __init__( self, - client: httpx.AsyncClient, - timeout: float, - max_redirects: int, - max_response_bytes: int, + settings: Settings, + *, + transport: httpx.AsyncBaseTransport | None = None, resolver: Resolver = system_resolver, ) -> None: - self.client = client - self.timeout = timeout - self.max_redirects = max_redirects - self.max_response_bytes = max_response_bytes + self.settings = settings + self.transport = transport 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 + async def check(self, monitor_id: str, target_url: str) -> CurrentStatus: + started = time.monotonic() + current_url = target_url + timeout = httpx.Timeout( + connect=self.settings.connect_timeout_seconds, + read=self.settings.read_timeout_seconds, + write=self.settings.write_timeout_seconds, + pool=self.settings.pool_timeout_seconds, + ) + limits = httpx.Limits(max_connections=self.settings.max_connections) 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 + async with httpx.AsyncClient( + timeout=timeout, + limits=limits, + transport=self.transport, + follow_redirects=False, + headers={"User-Agent": self.settings.user_agent}, + ) as client: + for redirect_count in range(self.settings.max_redirects + 1): + await validate_target(current_url, self.resolver) + async with client.stream("GET", current_url) as response: + if response.status_code in _REDIRECTS and response.headers.get("location"): + if redirect_count >= self.settings.max_redirects: + raise httpx.TooManyRedirects("redirect limit exceeded") + current_url = urljoin(current_url, response.headers["location"]) + continue + state = MonitorState.UP if response.status_code < 400 else MonitorState.DOWN + result = self._status(started, state, response.status_code) + self._log(monitor_id, current_url, result) + return result + except UnsafeTarget: + logger.warning( + "check blocked", + extra={"event": "check_blocked", "monitor_id": monitor_id, "url": redact_url(current_url)}, + ) + raise + except (httpx.HTTPError, OSError) as exc: + result = self._status(started, MonitorState.ERROR, error=str(exc)[:500]) + self._log(monitor_id, current_url, result) + return result + raise RuntimeError("unreachable redirect loop") @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) + def _status( + started: float, + state: MonitorState, + http_status: int | None = None, + error: str | None = None, + ) -> CurrentStatus: + return CurrentStatus( + state=state, + checked_at=utc_now(), + latency_ms=round((time.monotonic() - started) * 1000, 3), + 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()}) + def _log(monitor_id: str, url: str, result: CurrentStatus) -> None: + logger.info( + "check completed", + extra={ + "event": "check_completed", + "monitor_id": monitor_id, + "url": redact_url(url), + "outcome": result.state.value, + "latency_ms": result.latency_ms, + "http_status": result.http_status, + "error": result.error, + }, + )