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.
Some checks failed
ci / test (push) Has been cancelled
Some checks failed
ci / test (push) Has been cancelled
This commit is contained in:
166
app/checker.py
166
app/checker.py
@@ -1,132 +1,50 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import socket
|
||||
import time
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass
|
||||
from typing import Protocol
|
||||
from time import monotonic
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import aiohttp
|
||||
from aiohttp.abc import AbstractResolver
|
||||
import httpx
|
||||
|
||||
from app.config import Settings
|
||||
from app.models import CurrentStatus, HealthState, utc_now
|
||||
from app.security import Resolver, TargetPolicyError, ValidatedTarget, redact_url, system_resolver, validate_target
|
||||
from .config import Settings
|
||||
from .logging import log_event, redacted_url
|
||||
from .models import CheckSnapshot, State, now
|
||||
from .security import UnsafeDestination, validate_destination
|
||||
|
||||
logger = logging.getLogger("monitor.checker")
|
||||
logger = logging.getLogger("endpoint_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})
|
||||
async def check_url(url: str, timeout: float, settings: Settings,
|
||||
transport: httpx.AsyncBaseTransport | None = None) -> CheckSnapshot:
|
||||
started = monotonic()
|
||||
current = url
|
||||
try:
|
||||
async with httpx.AsyncClient(transport=transport, follow_redirects=False,
|
||||
timeout=httpx.Timeout(timeout)) as client:
|
||||
for redirect_count in range(settings.max_redirects + 1):
|
||||
await validate_destination(current)
|
||||
response = await client.get(current)
|
||||
if response.is_redirect:
|
||||
location = response.headers.get("location")
|
||||
if not location:
|
||||
break
|
||||
if redirect_count == settings.max_redirects:
|
||||
raise httpx.TooManyRedirects("redirect limit exceeded")
|
||||
current = urljoin(current, location)
|
||||
continue
|
||||
elapsed = round((monotonic() - started) * 1000, 3)
|
||||
state = State.up if 200 <= response.status_code < 400 else State.down
|
||||
result = CheckSnapshot(state=state, checked_at=now(), latency_ms=elapsed,
|
||||
http_status=response.status_code)
|
||||
log_event(logger, "check_complete", url=redacted_url(current), state=state,
|
||||
latency_ms=elapsed, http_status=response.status_code)
|
||||
return result
|
||||
raise httpx.RemoteProtocolError("redirect response missing location")
|
||||
except UnsafeDestination:
|
||||
log_event(logger, "check_blocked", url=redacted_url(current))
|
||||
raise
|
||||
except (httpx.HTTPError, TimeoutError) as exc:
|
||||
elapsed = round((monotonic() - started) * 1000, 3)
|
||||
result = CheckSnapshot(state=State.error, checked_at=now(), latency_ms=elapsed,
|
||||
error=type(exc).__name__)
|
||||
log_event(logger, "check_error", url=redacted_url(current), state=State.error,
|
||||
error=type(exc).__name__, latency_ms=elapsed)
|
||||
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,
|
||||
clock: Callable[[], float] = time.monotonic,
|
||||
) -> None:
|
||||
self.settings = settings
|
||||
self.fetcher = fetcher or AioHttpFetcher()
|
||||
self.resolver = resolver
|
||||
self.clock = clock
|
||||
|
||||
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:
|
||||
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
|
||||
|
||||
latency = max(0.0, (self.clock() - started) * 1000)
|
||||
status = CurrentStatus(
|
||||
state=state,
|
||||
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,
|
||||
)
|
||||
logger.info(
|
||||
"endpoint check completed",
|
||||
extra={"event": "check.completed", "monitor_id": monitor_id, "url": safe_url, "outcome": state, "error_code": error_code, "latency_ms": status.latency_ms},
|
||||
)
|
||||
return status
|
||||
|
||||
Reference in New Issue
Block a user