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:
108
app/checker.py
108
app/checker.py
@@ -1,50 +1,76 @@
|
|||||||
import logging
|
import logging
|
||||||
from time import monotonic
|
import time
|
||||||
|
from datetime import UTC, datetime
|
||||||
from urllib.parse import urljoin
|
from urllib.parse import urljoin
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
|
||||||
from .config import Settings
|
from app.config import Settings
|
||||||
from .logging import log_event, redacted_url
|
from app.models import MonitorState, StatusSnapshot
|
||||||
from .models import CheckSnapshot, State, now
|
from app.security import Resolver, UnsafeTarget, redact_url, system_resolver, validate_target
|
||||||
from .security import UnsafeDestination, validate_destination
|
|
||||||
|
|
||||||
logger = logging.getLogger("endpoint_monitor.checker")
|
logger = logging.getLogger("monitor.checker")
|
||||||
|
|
||||||
|
|
||||||
async def check_url(url: str, timeout: float, settings: Settings,
|
class EndpointChecker:
|
||||||
transport: httpx.AsyncBaseTransport | None = None) -> CheckSnapshot:
|
def __init__(
|
||||||
started = monotonic()
|
self, settings: Settings, resolver: Resolver = system_resolver,
|
||||||
current = url
|
transport: httpx.AsyncBaseTransport | None = None,
|
||||||
try:
|
) -> None:
|
||||||
async with httpx.AsyncClient(transport=transport, follow_redirects=False,
|
self.settings = settings
|
||||||
timeout=httpx.Timeout(timeout)) as client:
|
self.resolver = resolver
|
||||||
for redirect_count in range(settings.max_redirects + 1):
|
self.transport = transport
|
||||||
await validate_destination(current)
|
|
||||||
response = await client.get(current)
|
async def check(self, monitor_id: str, url: str) -> StatusSnapshot:
|
||||||
if response.is_redirect:
|
started = time.perf_counter()
|
||||||
location = response.headers.get("location")
|
current = url
|
||||||
if not location:
|
try:
|
||||||
break
|
timeout = httpx.Timeout(
|
||||||
if redirect_count == settings.max_redirects:
|
self.settings.check_timeout_seconds,
|
||||||
raise httpx.TooManyRedirects("redirect limit exceeded")
|
connect=self.settings.connect_timeout_seconds,
|
||||||
current = urljoin(current, location)
|
)
|
||||||
continue
|
async with httpx.AsyncClient(
|
||||||
elapsed = round((monotonic() - started) * 1000, 3)
|
timeout=timeout, transport=self.transport, follow_redirects=False,
|
||||||
state = State.up if 200 <= response.status_code < 400 else State.down
|
) as client:
|
||||||
result = CheckSnapshot(state=state, checked_at=now(), latency_ms=elapsed,
|
for hop in range(self.settings.max_redirects + 1):
|
||||||
http_status=response.status_code)
|
await validate_target(current, self.resolver)
|
||||||
log_event(logger, "check_complete", url=redacted_url(current), state=state,
|
async with client.stream("GET", current) as response:
|
||||||
latency_ms=elapsed, http_status=response.status_code)
|
if response.is_redirect:
|
||||||
return result
|
location = response.headers.get("location")
|
||||||
raise httpx.RemoteProtocolError("redirect response missing location")
|
if not location:
|
||||||
except UnsafeDestination:
|
return self._result(MonitorState.ERROR, started, error="bad_redirect")
|
||||||
log_event(logger, "check_blocked", url=redacted_url(current))
|
if hop == self.settings.max_redirects:
|
||||||
raise
|
return self._result(MonitorState.ERROR, started, error="too_many_redirects")
|
||||||
except (httpx.HTTPError, TimeoutError) as exc:
|
current = urljoin(current, location)
|
||||||
elapsed = round((monotonic() - started) * 1000, 3)
|
continue
|
||||||
result = CheckSnapshot(state=State.error, checked_at=now(), latency_ms=elapsed,
|
state = MonitorState.UP if 200 <= response.status_code < 400 else MonitorState.DOWN
|
||||||
error=type(exc).__name__)
|
result = self._result(state, started, response.status_code)
|
||||||
log_event(logger, "check_error", url=redacted_url(current), state=State.error,
|
self._log(monitor_id, url, result)
|
||||||
error=type(exc).__name__, latency_ms=elapsed)
|
return result
|
||||||
|
except UnsafeTarget as exc:
|
||||||
|
result = self._result(MonitorState.BLOCKED, started, error=str(exc))
|
||||||
|
except httpx.TimeoutException:
|
||||||
|
result = self._result(MonitorState.ERROR, started, error="timeout")
|
||||||
|
except httpx.HTTPError:
|
||||||
|
result = self._result(MonitorState.ERROR, started, error="transport_error")
|
||||||
|
self._log(monitor_id, url, result)
|
||||||
return result
|
return result
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _result(
|
||||||
|
state: MonitorState, started: float, status: int | None = None,
|
||||||
|
error: str | None = None,
|
||||||
|
) -> StatusSnapshot:
|
||||||
|
return StatusSnapshot(
|
||||||
|
state=state, checked_at=datetime.now(UTC),
|
||||||
|
latency_ms=round((time.perf_counter() - started) * 1000, 3),
|
||||||
|
http_status=status, error=error,
|
||||||
|
)
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _log(monitor_id: str, url: str, result: StatusSnapshot) -> None:
|
||||||
|
logger.info("check_complete", extra={
|
||||||
|
"monitor_id": monitor_id, "target": redact_url(url),
|
||||||
|
"state": result.state, "latency_ms": result.latency_ms,
|
||||||
|
"http_status": result.http_status, "error": result.error,
|
||||||
|
})
|
||||||
|
|||||||
Reference in New Issue
Block a user