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 / quality (push) Has been cancelled
ci / container (push) Has been cancelled

This commit is contained in:
2026-08-09 16:07:06 +00:00
parent d1d67cf5ed
commit cf3a39b50a

View File

@@ -1,71 +1,97 @@
import logging import logging
import time import time
from datetime import datetime
from urllib.parse import urljoin from urllib.parse import urljoin
import httpx import httpx
from .logging_config import redact_url from app.config import Settings
from .models import CheckResult, MonitorStatus, utcnow from app.models import CurrentStatus, MonitorState, utc_now
from .security import Resolver, UnsafeTarget, system_resolver, validate_target 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: class EndpointChecker:
def __init__( def __init__(
self, self,
client: httpx.AsyncClient, settings: Settings,
timeout: float, *,
max_redirects: int, transport: httpx.AsyncBaseTransport | None = None,
max_response_bytes: int,
resolver: Resolver = system_resolver, resolver: Resolver = system_resolver,
) -> None: ) -> None:
self.client = client self.settings = settings
self.timeout = timeout self.transport = transport
self.max_redirects = max_redirects
self.max_response_bytes = max_response_bytes
self.resolver = resolver self.resolver = resolver
async def check(self, monitor_id: str, initial_url: str) -> CheckResult: async def check(self, monitor_id: str, target_url: str) -> CurrentStatus:
started = time.perf_counter() started = time.monotonic()
checked_at: datetime = utcnow() current_url = target_url
current_url = initial_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: try:
for redirect_count in range(self.max_redirects + 1): async with httpx.AsyncClient(
await validate_target(current_url, self.resolver) timeout=timeout,
async with self.client.stream( limits=limits,
"GET", current_url, follow_redirects=False, timeout=self.timeout, transport=self.transport,
headers={"User-Agent": "endpoint-monitor/1"}, follow_redirects=False,
) as response: headers={"User-Agent": self.settings.user_agent},
consumed = 0 ) as client:
async for chunk in response.aiter_bytes(): for redirect_count in range(self.settings.max_redirects + 1):
consumed += len(chunk) await validate_target(current_url, self.resolver)
if consumed >= self.max_response_bytes: async with client.stream("GET", current_url) as response:
break if response.status_code in _REDIRECTS and response.headers.get("location"):
if response.is_redirect and response.headers.get("location"): if redirect_count >= self.settings.max_redirects:
if redirect_count == self.max_redirects: raise httpx.TooManyRedirects("redirect limit exceeded")
return self._result(started, checked_at, MonitorStatus.error, error="too_many_redirects") current_url = urljoin(current_url, response.headers["location"])
current_url = urljoin(str(response.url), response.headers["location"]) continue
continue state = MonitorState.UP if response.status_code < 400 else MonitorState.DOWN
status = MonitorStatus.up if 200 <= response.status_code < 400 else MonitorStatus.down result = self._status(started, state, response.status_code)
return self._result(started, checked_at, status, response.status_code) self._log(monitor_id, current_url, result)
except UnsafeTarget as exc: return result
result = self._result(started, checked_at, MonitorStatus.error, error=str(exc)) except UnsafeTarget:
except httpx.TimeoutException: logger.warning(
result = self._result(started, checked_at, MonitorStatus.error, error="timeout") "check blocked",
except httpx.HTTPError: extra={"event": "check_blocked", "monitor_id": monitor_id, "url": redact_url(current_url)},
result = self._result(started, checked_at, MonitorStatus.error, error="network_error") )
except Exception: raise
logger.exception("check_unexpected_error", extra={"monitor_id": monitor_id, "url": redact_url(initial_url)}) except (httpx.HTTPError, OSError) as exc:
result = self._result(started, checked_at, MonitorStatus.error, error="internal_error") result = self._status(started, MonitorState.ERROR, error=str(exc)[:500])
self._log(monitor_id, initial_url, result) self._log(monitor_id, current_url, result)
return result return result
raise RuntimeError("unreachable redirect loop")
@staticmethod @staticmethod
def _result(started: float, checked_at: datetime, status: MonitorStatus, http_status: int | None = None, error: str | None = None) -> CheckResult: def _status(
return CheckResult(status=status, checked_at=checked_at, latency_ms=max(0.0, (time.perf_counter() - started) * 1000), http_status=http_status, error=error) 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 @staticmethod
def _log(monitor_id: str, url: str, result: CheckResult) -> None: def _log(monitor_id: str, url: str, result: CurrentStatus) -> None:
logger.info("check_completed", extra={"monitor_id": monitor_id, "url": url, **result.model_dump()}) 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,
},
)