diff --git a/changelog.d/21.fixed.md b/changelog.d/21.fixed.md new file mode 100644 index 0000000..6769b2a --- /dev/null +++ b/changelog.d/21.fixed.md @@ -0,0 +1 @@ +Disable a log destination after repeated consecutive emit failures, restoring stdout if none remain — an observability sink can no longer degrade its host service's request path. diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 3813450..76b4725 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -9,6 +9,13 @@ from .google_cloud_logging import GoogleCloudLoggingDestination from .stdout import StdoutJsonDestination +# A destination that fails this many consecutive emits is disabled for the +# rest of the process. Emission is synchronous on the caller's (request) path, +# so a persistently failing destination — e.g. a credential exchange that +# errors after long internal retries — must not keep charging every request; +# an observability sink can never be allowed to degrade the host service. +DESTINATION_FAILURE_LIMIT = 3 + class LogDestinationManager: def __init__( @@ -25,6 +32,7 @@ def __init__( self.on_failure = on_failure self.destinations: list[LogDestination] = [] self.configured = False + self._consecutive_failures: dict[int, int] = {} def configure(self) -> None: failures: list[tuple[str, BaseException]] = [] @@ -62,6 +70,7 @@ def emit( severity: str, ) -> None: emitted_payload = {**payload, "severity": severity} + tripped: list[LogDestination] = [] for destination in self._ensure_destinations(): try: destination.emit( @@ -69,13 +78,23 @@ def emit( log_type=log_type, severity=severity, ) + self._consecutive_failures.pop(id(destination), None) except BaseException as exc: + failures = ( + self._consecutive_failures.get(id(destination), 0) + 1 + ) + self._consecutive_failures[id(destination)] = failures + if failures >= DESTINATION_FAILURE_LIMIT: + tripped.append(destination) self.on_failure( "logging.destination_emit", exc, destination=getattr(destination, "name", None), log_type=log_type, + consecutive_failures=failures, ) + for destination in tripped: + self._disable_destination(destination) def _ensure_destinations(self) -> list[LogDestination]: if not self.configured: @@ -87,6 +106,24 @@ def _ensure_destinations(self) -> list[LogDestination]: self.on_failure("logging.destination_config", exc) return self.destinations + def _disable_destination(self, destination: LogDestination) -> None: + self.destinations = [ + existing + for existing in self.destinations + if existing is not destination + ] + self._consecutive_failures.pop(id(destination), None) + self.on_failure( + "logging.destination_disabled", + RuntimeError( + "Disabling observability log destination after " + f"{DESTINATION_FAILURE_LIMIT} consecutive emit failures." + ), + destination=getattr(destination, "name", None), + ) + if not self.destinations: + self.destinations.append(self._stdout_destination()) + def _build_destination(self, destination_name: str) -> LogDestination: normalized = destination_name.strip().lower().replace("-", "_") if normalized == "stdout": diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 7fe9a36..10c1941 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -100,3 +100,111 @@ def test_google_destination_writes_structured_log_with_bounded_labels( } assert "request_id" not in kwargs["labels"] assert "path" not in kwargs["labels"] + + +# ── Destination circuit breaker ────────────────────────────────────────── + + +class FlakyDestination: + name = "flaky" + + def __init__(self, fail_first: int | None = None) -> None: + self.calls = 0 + self.fail_first = fail_first + + def emit(self, payload, *, log_type, severity) -> None: + self.calls += 1 + if self.fail_first is None or self.calls <= self.fail_first: + raise RuntimeError("emit failed") + + +class RecordingDestination: + name = "recording" + + def __init__(self) -> None: + self.payloads = [] + + def emit(self, payload, *, log_type, severity) -> None: + self.payloads.append(payload) + + +def _manager(destinations): + import json + import logging + + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations.manager import ( + LogDestinationManager, + ) + + failures = [] + manager = LogDestinationManager( + config=ObservabilityConfig(), + loggers={"event": logging.getLogger("test-destinations")}, + serializer=json.dumps, + on_failure=lambda operation, exc, **fields: failures.append( + (operation, fields) + ), + ) + manager.destinations = list(destinations) + manager.configured = True + return manager, failures + + +def test_destination_disabled_after_consecutive_emit_failures() -> None: + from policyengine_observability.destinations.manager import ( + DESTINATION_FAILURE_LIMIT, + ) + + flaky = FlakyDestination() + healthy = RecordingDestination() + manager, failures = _manager([flaky, healthy]) + + for _ in range(DESTINATION_FAILURE_LIMIT + 2): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + + assert flaky.calls == DESTINATION_FAILURE_LIMIT + assert flaky not in manager.destinations + assert len(healthy.payloads) == DESTINATION_FAILURE_LIMIT + 2 + assert any(op == "logging.destination_disabled" for op, _ in failures) + + +def test_emit_success_resets_the_failure_counter() -> None: + from policyengine_observability.destinations.manager import ( + DESTINATION_FAILURE_LIMIT, + ) + + flaky = FlakyDestination(fail_first=DESTINATION_FAILURE_LIMIT - 1) + manager, failures = _manager([flaky]) + + for _ in range(DESTINATION_FAILURE_LIMIT + 2): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + + assert flaky in manager.destinations + counts = [ + fields["consecutive_failures"] + for op, fields in failures + if op == "logging.destination_emit" + ] + assert max(counts) == DESTINATION_FAILURE_LIMIT - 1 + + +def test_sole_disabled_destination_falls_back_to_stdout() -> None: + from policyengine_observability.destinations.manager import ( + DESTINATION_FAILURE_LIMIT, + ) + from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + ) + + flaky = FlakyDestination() + manager, failures = _manager([flaky]) + + for _ in range(DESTINATION_FAILURE_LIMIT + 1): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + + assert flaky not in manager.destinations + assert any( + isinstance(destination, StdoutJsonDestination) + for destination in manager.destinations + )