Disable a log destination after repeated consecutive emit failures

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.

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,20 +70,31 @@ def emit(
         severity: str,
     ) -> None:
         emitted_payload = {**payload, "severity": severity}
+        tripped: list[LogDestination] = []
         for destination in self._ensure_destinations():
             try:
                 destination.emit(
                     emitted_payload,
                     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":

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
+    )