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

@@ -1,107 +1,132 @@
+import asyncio
 import logging
+import socket
 import time
-from collections.abc import Awaitable, Callable
-from datetime import UTC, datetime
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass
+from typing import Protocol
 from urllib.parse import urljoin
 
-import httpx
+import aiohttp
+from aiohttp.abc import AbstractResolver
 
 from app.config import Settings
-from app.models import CurrentStatus, State
-from app.security import (
-    DestinationRejected,
-    Resolver,
-    redacted_url,
-    require_public_destination,
-    system_resolver,
-)
+from app.models import CurrentStatus, HealthState, utc_now
+from app.security import Resolver, TargetPolicyError, ValidatedTarget, redact_url, system_resolver, validate_target
 
-logger = logging.getLogger(__name__)
-RedirectCodes = {301, 302, 303, 307, 308}
+logger = logging.getLogger("monitor.checker")
+
+
+@dataclass(frozen=True)
+class HopResponse:
+    status: int
+    location: str | None
+
+
+class Fetcher(Protocol):
+    async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse: ...
+
+
+class PinnedResolver(AbstractResolver):
+    def __init__(self, hostname: str, addresses: Sequence[str]) -> None:
+        self.hostname = hostname
+        self.addresses = addresses
+
+    async def resolve(self, host: str, port: int = 0, family: int = socket.AF_UNSPEC) -> list[dict[str, object]]:
+        if host.rstrip(".").lower() != self.hostname:
+            raise OSError("unpinned hostname requested")
+        result: list[dict[str, object]] = []
+        for address in self.addresses:
+            address_family = socket.AF_INET6 if ":" in address else socket.AF_INET
+            if family in (socket.AF_UNSPEC, address_family):
+                result.append({"hostname": host, "host": address, "port": port, "family": address_family, "proto": 0, "flags": 0})
+        return result
+
+    async def close(self) -> None:
+        return None
+
+
+class AioHttpFetcher:
+    async def fetch(self, target: ValidatedTarget, settings: Settings, remaining: float) -> HopResponse:
+        resolver = PinnedResolver(target.hostname, target.addresses)
+        connector = aiohttp.TCPConnector(resolver=resolver, use_dns_cache=True, force_close=True)
+        timeout = aiohttp.ClientTimeout(
+            total=remaining,
+            connect=min(settings.connect_timeout_seconds, remaining),
+            sock_read=min(settings.read_timeout_seconds, remaining),
+        )
+        headers = {"User-Agent": settings.user_agent, "Accept": "*/*"}
+        async with aiohttp.ClientSession(connector=connector, timeout=timeout, trust_env=False) as session:
+            async with session.get(target.url, headers=headers, allow_redirects=False) as response:
+                await response.content.read(1)
+                return HopResponse(response.status, response.headers.get("Location"))
+
+
+_ERROR_MESSAGES = {
+    "blocked_target": "target rejected by network policy",
+    "timeout": "endpoint check timed out",
+    "dns_or_connect": "endpoint connection failed",
+    "tls": "endpoint TLS validation failed",
+    "protocol": "endpoint returned an invalid HTTP response",
+    "redirect_limit": "endpoint exceeded the redirect limit",
+}
 
 
 class EndpointChecker:
     def __init__(
         self,
         settings: Settings,
+        fetcher: Fetcher | None = None,
         resolver: Resolver = system_resolver,
-        transport: httpx.AsyncBaseTransport | None = None,
+        clock: Callable[[], float] = time.monotonic,
     ) -> None:
         self.settings = settings
+        self.fetcher = fetcher or AioHttpFetcher()
         self.resolver = resolver
-        self.transport = transport
+        self.clock = clock
 
-    async def check(self, monitor_id: str, url: str) -> CurrentStatus:
-        started = time.monotonic()
-        safe_url = redacted_url(url)
+    async def check(self, url: str, monitor_id: str = "unknown") -> CurrentStatus:
+        started = self.clock()
+        safe_url = redact_url(url)
+        error_code: str | None = None
+        http_status: int | None = None
         try:
-            result = await self._request(url, started)
-        except DestinationRejected:
-            self._log(monitor_id, safe_url, State.ERROR, started, None)
-            raise
-        except (httpx.TimeoutException, httpx.NetworkError, httpx.ProtocolError) as exc:
-            result = self._error(started, self._safe_error(exc))
-        self._log(monitor_id, safe_url, result.state, started, result.status_code)
-        return result
+            async with asyncio.timeout(self.settings.total_timeout_seconds):
+                current = url
+                for redirects in range(self.settings.max_redirects + 1):
+                    target = await validate_target(current, self.resolver)
+                    remaining = max(0.001, self.settings.total_timeout_seconds - (self.clock() - started))
+                    response = await self.fetcher.fetch(target, self.settings, remaining)
+                    http_status = response.status
+                    if response.status not in {301, 302, 303, 307, 308} or not response.location:
+                        break
+                    if redirects == self.settings.max_redirects:
+                        error_code = "redirect_limit"
+                        break
+                    current = urljoin(current, response.location)
+                state = HealthState.up if error_code is None and http_status is not None and 200 <= http_status < 400 else HealthState.down
+        except TargetPolicyError:
+            error_code, state = "blocked_target", HealthState.down
+        except TimeoutError:
+            error_code, state = "timeout", HealthState.down
+        except aiohttp.ClientConnectorCertificateError:
+            error_code, state = "tls", HealthState.down
+        except (aiohttp.ClientConnectorError, socket.gaierror, OSError):
+            error_code, state = "dns_or_connect", HealthState.down
+        except aiohttp.ClientError:
+            error_code, state = "protocol", HealthState.down
 
-    async def _request(self, initial_url: str, started: float) -> CurrentStatus:
-        timeout = httpx.Timeout(
-            self.settings.request_timeout_seconds,
-            connect=self.settings.connect_timeout_seconds,
-        )
-        current = initial_url
-        async with httpx.AsyncClient(
-            timeout=timeout, follow_redirects=False, transport=self.transport
-        ) as client:
-            for hop in range(self.settings.max_redirects + 1):
-                await require_public_destination(current, self.resolver)
-                async with client.stream("GET", current) as response:
-                    if response.status_code not in RedirectCodes:
-                        state = State.UP if 200 <= response.status_code < 400 else State.DOWN
-                        return CurrentStatus(
+        latency = max(0.0, (self.clock() - started) * 1000)
+        status = CurrentStatus(
             state=state,
-                            checked_at=datetime.now(UTC),
-                            status_code=response.status_code,
-                            latency_ms=self._elapsed(started),
+            checked_at=utc_now(),
+            latency_ms=round(latency, 3),
+            http_status=http_status,
+            error_code=error_code,
+            error_message=_ERROR_MESSAGES.get(error_code) if error_code else None,
         )
-                    location = response.headers.get("location")
-                if not location:
-                    return self._error(started, "redirect response omitted Location")
-                if hop == self.settings.max_redirects:
-                    return self._error(started, "redirect limit exceeded")
-                current = urljoin(current, location)
-        return self._error(started, "redirect limit exceeded")  # pragma: no cover
-
-    @staticmethod
-    def _elapsed(started: float) -> float:
-        return round((time.monotonic() - started) * 1000, 3)
-
-    def _error(self, started: float, message: str) -> CurrentStatus:
-        return CurrentStatus(
-            state=State.ERROR,
-            checked_at=datetime.now(UTC),
-            latency_ms=self._elapsed(started),
-            error=message[:200],
-        )
-
-    @staticmethod
-    def _safe_error(exc: Exception) -> str:
-        if isinstance(exc, httpx.TimeoutException):
-            return "outbound request timed out"
-        return f"outbound request failed: {type(exc).__name__}"
-
-    @staticmethod
-    def _log(
-        monitor_id: str, url: str, state: State, started: float, status_code: int | None
-    ) -> None:
         logger.info(
             "endpoint check completed",
-            extra={
-                "event": "endpoint_check",
-                "monitor_id": monitor_id,
-                "url": url,
-                "state": state,
-                "latency_ms": EndpointChecker._elapsed(started),
-                "status_code": status_code,
-            },
+            extra={"event": "check.completed", "monitor_id": monitor_id, "url": safe_url, "outcome": state, "error_code": error_code, "latency_ms": status.latency_ms},
         )
+        return status