From 70b1c53f8b4b605306a46bab14b4a0379af1a730 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Tue, 7 Jul 2026 21:51:38 +0200 Subject: [PATCH 1/7] Bound Google Cloud Logging writes with an explicit timeout Logger.log_struct exposes no call options, so writes inherited the transport defaults: a 60-second retry deadline and 60-second timeout that let a degraded Logging API stall the calling request far longer than any observability write is worth. Build the log entries directly (keeping the Logger for its full_name and platform-detected default resource) and write through the gapic layer with retry disabled and a configurable timeout (OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS, default 2s), falling back to the unbounded write_entries call on non-gapic transports rather than dropping logs. Adds emit_batch so multiple records cost one bounded call. Refs #22 Co-Authored-By: Claude Fable 5 --- changelog.d/22.changed.md | 1 + policyengine_observability/config.py | 5 + .../destinations/google_cloud_logging.py | 96 +++++++++-- .../destinations/manager.py | 1 + tests/test_destinations.py | 159 ++++++++++++++---- 5 files changed, 222 insertions(+), 40 deletions(-) create mode 100644 changelog.d/22.changed.md diff --git a/changelog.d/22.changed.md b/changelog.d/22.changed.md new file mode 100644 index 0000000..6ce4776 --- /dev/null +++ b/changelog.d/22.changed.md @@ -0,0 +1 @@ +Google Cloud Logging writes now carry an explicit per-call timeout with retries disabled (default 2 seconds, `OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS`), replacing the transport defaults that could block a caller for up to 60 seconds per record when the Logging API degrades. The destination also gains `emit_batch`, writing multiple records in one bounded call. diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index 2e37860..ce84666 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -87,6 +87,7 @@ class ObservabilityConfig: log_destinations: tuple[str, ...] = ("stdout",) google_cloud_project: str | None = None google_cloud_log_name: str = "policyengine-observability" + google_log_timeout_seconds: float = 2.0 @classmethod def from_env( @@ -166,6 +167,10 @@ def from_env( os.getenv("OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME") or cls.google_cloud_log_name ), + google_log_timeout_seconds=float_from_env( + "OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS", + cls.google_log_timeout_seconds, + ), ) diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 8d8d918..13fbb9b 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -1,6 +1,6 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Sequence from typing import Any, Protocol from policyengine_observability.google_credentials import ( @@ -10,13 +10,12 @@ from .base import normalize_payload +DEFAULT_WRITE_TIMEOUT_SECONDS = 2.0 + class GoogleCloudLogger(Protocol): - def log_struct( - self, - payload: dict[str, Any], - **kwargs: Any, - ) -> None: ... + full_name: str + default_resource: Any class GoogleCloudLoggingClient(Protocol): @@ -32,6 +31,16 @@ def logger(self, log_name: str) -> GoogleCloudLogger: ... class GoogleCloudLoggingDestination: + """Writes structured payloads to Google Cloud Logging. + + Writes go through the transport layer directly with an explicit + per-call timeout and retries disabled: ``Logger.log_struct`` offers no + call options, and the underlying defaults (60s retry deadline) would + let a degraded Logging API stall the caller far longer than any + observability write is worth. The ``Logger`` object is still used for + its ``full_name`` and platform-detected ``default_resource``. + """ + name = "google_cloud_logging" def __init__( @@ -40,9 +49,11 @@ def __init__( project: str | None, log_name: str, client_factory: GoogleCloudLoggingClientFactory | None = None, + timeout_seconds: float = DEFAULT_WRITE_TIMEOUT_SECONDS, ) -> None: self.project = project self.log_name = log_name + self.timeout_seconds = timeout_seconds credentials = load_google_credentials(prefer_workload_identity=True) if credentials is None: configure_google_application_credentials() @@ -69,18 +80,81 @@ def emit( log_type: str, severity: str, ) -> None: + self._write( + [self._build_entry(payload, log_type=log_type, severity=severity)] + ) + + def emit_batch( + self, + records: Sequence[tuple[dict[str, Any], str, str]], + ) -> None: + """Write ``(payload, log_type, severity)`` records in one call.""" + entries = [ + self._build_entry(payload, log_type=log_type, severity=severity) + for payload, log_type, severity in records + ] + if entries: + self._write(entries) + + def _build_entry( + self, + payload: dict[str, Any], + *, + log_type: str, + severity: str, + ) -> dict[str, Any]: normalized = normalize_payload(payload) - kwargs: dict[str, Any] = { - "severity": severity, + entry: dict[str, Any] = { + "logName": self.logger.full_name, + "resource": _resource_dict(self.logger.default_resource), + "jsonPayload": normalized, + "severity": str(severity).upper(), "labels": _labels(normalized, log_type=log_type), } trace_id = normalized.get("trace_id") if trace_id and self.project: - kwargs["trace"] = f"projects/{self.project}/traces/{trace_id}" + entry["trace"] = f"projects/{self.project}/traces/{trace_id}" span_id = normalized.get("span_id") if span_id: - kwargs["span_id"] = span_id - self.logger.log_struct(normalized, **kwargs) + entry["spanId"] = str(span_id) + return entry + + def _write(self, entries: list[dict[str, Any]]) -> None: + api = self.client.logging_api + gapic_api = getattr(api, "_gapic_api", None) + if gapic_api is not None: + try: + from google.cloud.logging_v2._gapic import ( + _log_entry_mapping_to_pb, + ) + from google.cloud.logging_v2.types import ( + WriteLogEntriesRequest, + ) + except ImportError: # pragma: no cover - exotic installs only + gapic_api = None + else: + request = WriteLogEntriesRequest( + entries=[ + _log_entry_mapping_to_pb(entry) for entry in entries + ], + partial_success=True, + ) + gapic_api.write_log_entries( + request=request, + retry=None, + timeout=self.timeout_seconds, + ) + return + # Non-gapic transports expose only the unbounded call; keep the + # destination functional there rather than dropping logs. + api.write_entries(entries, partial_success=True) + + +def _resource_dict(resource: Any) -> Any: + to_dict = getattr(resource, "_to_dict", None) + if callable(to_dict): + return to_dict() + return resource def _labels(payload: dict[str, Any], *, log_type: str) -> dict[str, str]: diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 76b4725..7821aa2 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -132,6 +132,7 @@ def _build_destination(self, destination_name: str) -> LogDestination: return GoogleCloudLoggingDestination( project=self.config.google_cloud_project, log_name=self.config.google_cloud_log_name, + timeout_seconds=self.config.google_log_timeout_seconds, ) raise ValueError( f"Unknown observability log destination: {destination_name}" diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 10c1941..4176af5 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -12,25 +12,66 @@ def __str__(self) -> str: raise RuntimeError("cannot stringify") +class FakeResource: + def _to_dict(self) -> dict: + return {"type": "global", "labels": {}} + + class FakeLogger: def __init__(self) -> None: + self.full_name = "projects/resolved-project/logs/test-log" + self.default_resource = FakeResource() + + +class FakeGapicApi: + def __init__(self) -> None: + self.calls = [] + + def write_log_entries(self, *, request, retry, timeout) -> None: + self.calls.append((request, retry, timeout)) + + +class FakeLoggingApi: + def __init__(self, *, gapic: bool) -> None: self.calls = [] + if gapic: + self._gapic_api = FakeGapicApi() - def log_struct(self, payload, **kwargs) -> None: - self.calls.append((payload, kwargs)) + def write_entries(self, entries, *, partial_success) -> None: + self.calls.append((entries, partial_success)) class FakeClient: - def __init__(self) -> None: + def __init__(self, *, gapic: bool = True) -> None: self.project = "resolved-project" self.fake_logger = FakeLogger() self.log_names = [] + self.logging_api = FakeLoggingApi(gapic=gapic) def logger(self, log_name: str) -> FakeLogger: self.log_names.append(log_name) return self.fake_logger +def _destination(monkeypatch, client, **kwargs): + monkeypatch.setattr( + google_cloud_logging, + "load_google_credentials", + lambda *, prefer_workload_identity: None, + ) + monkeypatch.setattr( + google_cloud_logging, + "configure_google_application_credentials", + lambda: None, + ) + return GoogleCloudLoggingDestination( + project=None, + log_name="policyengine-observability", + client_factory=lambda _project, _credentials: client, + **kwargs, + ) + + def test_normalize_payload_recursively_stringifies_unsafe_values() -> None: normalized = normalize_payload( { @@ -49,25 +90,9 @@ def test_normalize_payload_recursively_stringifies_unsafe_values() -> None: assert normalized["nested"]["object"].startswith(" None: - monkeypatch.setattr( - google_cloud_logging, - "load_google_credentials", - lambda *, prefer_workload_identity: None, - ) - monkeypatch.setattr( - google_cloud_logging, - "configure_google_application_credentials", - lambda: None, - ) +def test_google_destination_writes_bounded_gapic_entry(monkeypatch) -> None: client = FakeClient() - destination = GoogleCloudLoggingDestination( - project=None, - log_name="policyengine-observability", - client_factory=lambda _project, _credentials: client, - ) + destination = _destination(monkeypatch, client) destination.emit( { @@ -85,21 +110,97 @@ def test_google_destination_writes_structured_log_with_bounded_labels( severity="ERROR", ) - payload, kwargs = client.fake_logger.calls[0] assert client.log_names == ["policyengine-observability"] - assert payload["object"].startswith(" None: + client = FakeClient(gapic=False) + destination = _destination(monkeypatch, client) + + destination.emit( + {"trace_id": None, "event": "x"}, + log_type="event", + severity="info", + ) + + entries, partial_success = client.logging_api.calls[0] + assert partial_success is True + (entry,) = entries + assert entry["severity"] == "INFO" + assert entry["resource"] == {"type": "global", "labels": {}} + assert "trace" not in entry + assert "spanId" not in entry + + +def test_google_destination_respects_configured_timeout(monkeypatch) -> None: + client = FakeClient() + destination = _destination(monkeypatch, client, timeout_seconds=0.5) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + _request, _retry, timeout = client.logging_api._gapic_api.calls[0] + assert timeout == 0.5 + + +def test_google_destination_emit_batch_writes_one_call(monkeypatch) -> None: + client = FakeClient() + destination = _destination(monkeypatch, client) + + destination.emit_batch( + [ + ({"event": "a"}, "event", "INFO"), + ({"event": "b"}, "request", "WARNING"), + ] + ) + destination.emit_batch([]) + + assert len(client.logging_api._gapic_api.calls) == 1 + request, _retry, _timeout = client.logging_api._gapic_api.calls[0] + from google.logging.type.log_severity_pb2 import LogSeverity + + assert len(request.entries) == 2 + assert LogSeverity.Name(request.entries[1].severity) == "WARNING" + + +def test_google_destination_handles_plain_mapping_resource( + monkeypatch, +) -> None: + client = FakeClient(gapic=False) + client.fake_logger.default_resource = { + "type": "gce_instance", + "labels": {}, + } + destination = _destination(monkeypatch, client) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + (entry,), _ = client.logging_api.calls[0] + assert entry["resource"] == {"type": "gce_instance", "labels": {}} # ── Destination circuit breaker ────────────────────────────────────────── From ea67dfae3dc17e34d3e9f2331c6ed70190a72f32 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Tue, 7 Jul 2026 21:54:20 +0200 Subject: [PATCH 2/7] Add agent-native google stdout format OBSERVABILITY_STDOUT_FORMAT=google emits JSON lines carrying the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, time, logging.googleapis.com/trace, spanId, labels), so agent-collected stdout gets full-fidelity ingestion with no in-process network emission. The bounded-labels helper moves to destinations.base for reuse by both destinations. Default format is unchanged. Refs #22 Co-Authored-By: Claude Fable 5 --- changelog.d/22.added.md | 1 + policyengine_observability/config.py | 6 + .../destinations/base.py | 18 +++ .../destinations/google_cloud_logging.py | 18 +-- .../destinations/manager.py | 2 + .../destinations/stdout.py | 53 ++++++- tests/test_destinations.py | 130 ++++++++++++++++++ 7 files changed, 210 insertions(+), 18 deletions(-) create mode 100644 changelog.d/22.added.md diff --git a/changelog.d/22.added.md b/changelog.d/22.added.md new file mode 100644 index 0000000..3f0ef4a --- /dev/null +++ b/changelog.d/22.added.md @@ -0,0 +1 @@ +Added an agent-native stdout format (`OBSERVABILITY_STDOUT_FORMAT=google`): JSON lines carry the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, time, trace, span, labels), enabling full-fidelity Cloud Logging ingestion from stdout with no in-process network emission. Added a background log emitter (`OBSERVABILITY_LOG_EMIT_MODE=async`): log calls enqueue onto a bounded buffer and a worker thread batches writes off the request path, with flush-on-shutdown, post-fork/snapshot `restart_observability()`, and the existing destination circuit breaker preserved. diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index ce84666..7436fc5 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -88,6 +88,7 @@ class ObservabilityConfig: google_cloud_project: str | None = None google_cloud_log_name: str = "policyengine-observability" google_log_timeout_seconds: float = 2.0 + stdout_format: str = "plain" @classmethod def from_env( @@ -171,6 +172,11 @@ def from_env( "OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS", cls.google_log_timeout_seconds, ), + stdout_format=( + os.getenv("OBSERVABILITY_STDOUT_FORMAT") or cls.stdout_format + ) + .strip() + .lower(), ) diff --git a/policyengine_observability/destinations/base.py b/policyengine_observability/destinations/base.py index 72e798e..00c7fe1 100644 --- a/policyengine_observability/destinations/base.py +++ b/policyengine_observability/destinations/base.py @@ -35,3 +35,21 @@ def normalize_payload(value: Any) -> Any: return str(value) except BaseException: return f"" + + +def bounded_labels( + payload: dict[str, Any], + *, + log_type: str, +) -> dict[str, str]: + labels = {"log_type": log_type} + for key in ( + "service_name", + "service_role", + "environment", + "schema_version", + ): + value = payload.get(key) + if value is not None: + labels[key] = str(value) + return labels diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 13fbb9b..98d8d75 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -8,7 +8,7 @@ load_google_credentials, ) -from .base import normalize_payload +from .base import bounded_labels, normalize_payload DEFAULT_WRITE_TIMEOUT_SECONDS = 2.0 @@ -109,7 +109,7 @@ def _build_entry( "resource": _resource_dict(self.logger.default_resource), "jsonPayload": normalized, "severity": str(severity).upper(), - "labels": _labels(normalized, log_type=log_type), + "labels": bounded_labels(normalized, log_type=log_type), } trace_id = normalized.get("trace_id") if trace_id and self.project: @@ -155,17 +155,3 @@ def _resource_dict(resource: Any) -> Any: if callable(to_dict): return to_dict() return resource - - -def _labels(payload: dict[str, Any], *, log_type: str) -> dict[str, str]: - labels = {"log_type": log_type} - for key in ( - "service_name", - "service_role", - "environment", - "schema_version", - ): - value = payload.get(key) - if value is not None: - labels[key] = str(value) - return labels diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 7821aa2..e965edb 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -142,4 +142,6 @@ def _stdout_destination(self) -> StdoutJsonDestination: return StdoutJsonDestination( loggers=self.loggers, serializer=self.serializer, + output_format=self.config.stdout_format, + google_cloud_project=self.config.google_cloud_project, ) diff --git a/policyengine_observability/destinations/stdout.py b/policyengine_observability/destinations/stdout.py index 5bc01b1..085c0be 100644 --- a/policyengine_observability/destinations/stdout.py +++ b/policyengine_observability/destinations/stdout.py @@ -4,10 +4,22 @@ from collections.abc import Callable, Mapping from typing import Any -from .base import normalize_payload +from .base import bounded_labels, normalize_payload + +GOOGLE_TRACE_KEY = "logging.googleapis.com/trace" +GOOGLE_SPAN_ID_KEY = "logging.googleapis.com/spanId" +GOOGLE_LABELS_KEY = "logging.googleapis.com/labels" class StdoutJsonDestination: + """Writes structured payloads as JSON lines through stdlib loggers. + + In ``google`` output format the line carries the special keys the + Cloud Run / GKE logging agent promotes to first-class LogEntry fields + (severity, time, trace, span, labels), so agent-collected stdout gets + full-fidelity ingestion with no in-process network emission. + """ + name = "stdout" def __init__( @@ -15,9 +27,13 @@ def __init__( *, loggers: Mapping[str, logging.Logger], serializer: Callable[[dict[str, Any]], str], + output_format: str = "plain", + google_cloud_project: str | None = None, ) -> None: self.loggers = loggers self.serializer = serializer + self.output_format = output_format + self.google_cloud_project = google_cloud_project def emit( self, @@ -27,10 +43,43 @@ def emit( severity: str, ) -> None: logger = self.loggers.get(log_type) or self.loggers["event"] - message = self.serializer(normalize_payload(payload)) + normalized = normalize_payload(payload) + if self.output_format == "google": + normalized = self._google_line( + normalized, + log_type=log_type, + severity=severity, + ) + message = self.serializer(normalized) if severity in {"ERROR", "CRITICAL"}: logger.error(message) elif severity == "WARNING": logger.warning(message) else: logger.info(message) + + def _google_line( + self, + normalized: dict[str, Any], + *, + log_type: str, + severity: str, + ) -> dict[str, Any]: + line = dict(normalized) + line["severity"] = str(severity).upper() + created_at = normalized.get("created_at") + if created_at: + line["time"] = created_at + trace_id = normalized.get("trace_id") + if trace_id and self.google_cloud_project: + line[GOOGLE_TRACE_KEY] = ( + f"projects/{self.google_cloud_project}/traces/{trace_id}" + ) + span_id = normalized.get("span_id") + if span_id: + line[GOOGLE_SPAN_ID_KEY] = str(span_id) + line[GOOGLE_LABELS_KEY] = bounded_labels( + normalized, + log_type=log_type, + ) + return line diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 4176af5..f3fe8ab 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -309,3 +309,133 @@ def test_sole_disabled_destination_falls_back_to_stdout() -> None: isinstance(destination, StdoutJsonDestination) for destination in manager.destinations ) + + +# ── Stdout output formats ──────────────────────────────────────────────── + + +class RecordingLogger: + def __init__(self) -> None: + self.lines = [] + + def info(self, message) -> None: + self.lines.append(("info", message)) + + def warning(self, message) -> None: + self.lines.append(("warning", message)) + + def error(self, message) -> None: + self.lines.append(("error", message)) + + +def _stdout_destination(**kwargs): + import json + + from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + ) + + logger = RecordingLogger() + destination = StdoutJsonDestination( + loggers={"event": logger}, + serializer=json.dumps, + **kwargs, + ) + return destination, logger + + +def test_stdout_google_format_maps_agent_keys() -> None: + import json + + destination, logger = _stdout_destination( + output_format="google", + google_cloud_project="central-project", + ) + + destination.emit( + { + "created_at": "2026-07-07T00:00:00+00:00", + "trace_id": "abc123", + "span_id": "def456", + "service_name": "svc", + "path": "/calculate", + }, + log_type="event", + severity="WARNING", + ) + + level, message = logger.lines[0] + line = json.loads(message) + assert level == "warning" + assert line["severity"] == "WARNING" + assert line["time"] == "2026-07-07T00:00:00+00:00" + assert ( + line["logging.googleapis.com/trace"] + == "projects/central-project/traces/abc123" + ) + assert line["logging.googleapis.com/spanId"] == "def456" + assert line["logging.googleapis.com/labels"] == { + "log_type": "event", + "service_name": "svc", + } + assert line["path"] == "/calculate" + + +def test_stdout_google_format_omits_trace_without_project() -> None: + import json + + destination, logger = _stdout_destination(output_format="google") + + destination.emit( + {"trace_id": "abc123"}, + log_type="event", + severity="ERROR", + ) + + _level, message = logger.lines[0] + line = json.loads(message) + assert "logging.googleapis.com/trace" not in line + assert "time" not in line + assert line["severity"] == "ERROR" + + +def test_stdout_plain_format_adds_no_agent_keys() -> None: + import json + + destination, logger = _stdout_destination() + + destination.emit( + {"trace_id": "abc123", "severity": "INFO"}, + log_type="event", + severity="INFO", + ) + + level, message = logger.lines[0] + line = json.loads(message) + assert level == "info" + assert line == {"trace_id": "abc123", "severity": "INFO"} + + +def test_manager_stdout_fallback_carries_configured_format() -> None: + import json + import logging + + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations.manager import ( + LogDestinationManager, + ) + + manager = LogDestinationManager( + config=ObservabilityConfig( + stdout_format="google", + google_cloud_project="central-project", + ), + loggers={"event": logging.getLogger("test-stdout-format")}, + serializer=json.dumps, + on_failure=lambda *args, **kwargs: None, + ) + + destination = manager._stdout_destination() + + assert destination.output_format == "google" + assert destination.google_cloud_project == "central-project" From a5d55665c89daa007625f1b28f15639350463775 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Tue, 7 Jul 2026 21:58:01 +0200 Subject: [PATCH 3/7] Add the background log emitter BackgroundEmitDestination installs the standard accept/emit split around any destination: emit() appends to a bounded buffer and returns immediately, and a lazy pid-aware daemon worker drains batches to the wrapped destination (preferring emit_batch, so a batch costs one bounded API call). Overflow drops the newest record with counted, throttled reporting; after the failure limit of consecutive failed batches the wrapper trips and subsequent emit() calls raise, flowing through the manager's existing circuit breaker and stdout fallback unchanged. flush(deadline) drains from the caller's thread with an atexit hook registered on first worker start; restart() clears buffer and trip state for forked or snapshot-restored processes whose threads did not survive. Refs #22 Co-Authored-By: Claude Fable 5 --- .../destinations/background.py | 226 ++++++++++++++++ tests/test_destinations.py | 253 ++++++++++++++++++ 2 files changed, 479 insertions(+) create mode 100644 policyengine_observability/destinations/background.py diff --git a/policyengine_observability/destinations/background.py b/policyengine_observability/destinations/background.py new file mode 100644 index 0000000..4cf1500 --- /dev/null +++ b/policyengine_observability/destinations/background.py @@ -0,0 +1,226 @@ +from __future__ import annotations + +import atexit +import os +import threading +import time +from collections import deque +from collections.abc import Callable +from typing import Any + +from .base import LogDestination + +DEFAULT_QUEUE_SIZE = 1000 +DEFAULT_BATCH_SIZE = 10 +DEFAULT_BATCH_LATENCY_SECONDS = 0.25 +DEFAULT_FLUSH_DEADLINE_SECONDS = 5.0 +DEFAULT_FAILURE_LIMIT = 3 +DROP_REPORT_EVERY = 100 + +_LogRecord = tuple[dict[str, Any], str, str] + + +class BackgroundEmitDestination: + """Decouples log acceptance from emission. + + ``emit()`` appends to a bounded in-memory buffer and returns + immediately — it never blocks on the network and never raises for + sink trouble. A daemon worker drains the buffer in batches to the + wrapped destination (preferring its ``emit_batch``). Overflow drops + the newest record and counts it; after ``failure_limit`` consecutive + failed batches the wrapper trips, and subsequent ``emit()`` calls + raise so the manager's existing circuit breaker disables it through + the same path as a synchronous destination (stdout fallback + preserved). + + The worker starts lazily on first emit and is pid-aware, so a forked + or snapshot-restored process (where threads do not survive) starts a + fresh worker automatically; ``restart()`` additionally clears the + buffer and trip state for consumers that restore process memory. + """ + + def __init__( + self, + wrapped: LogDestination, + *, + on_failure: Callable[..., None], + queue_size: int = DEFAULT_QUEUE_SIZE, + batch_size: int = DEFAULT_BATCH_SIZE, + batch_latency_seconds: float = DEFAULT_BATCH_LATENCY_SECONDS, + flush_deadline_seconds: float = DEFAULT_FLUSH_DEADLINE_SECONDS, + failure_limit: int = DEFAULT_FAILURE_LIMIT, + ) -> None: + self.wrapped = wrapped + self.name = getattr(wrapped, "name", "background") + self.on_failure = on_failure + self.queue_size = queue_size + self.batch_size = max(1, batch_size) + self.batch_latency_seconds = batch_latency_seconds + self.flush_deadline_seconds = flush_deadline_seconds + self.failure_limit = failure_limit + self._buffer: deque[_LogRecord] = deque() + self._lock = threading.Lock() + self._start_lock = threading.Lock() + self._wake = threading.Event() + self._tripped = threading.Event() + self._closed = threading.Event() + self._worker: threading.Thread | None = None + self._pid: int | None = None + self._consecutive_failures = 0 + self._dropped = 0 + self._atexit_registered = False + + def emit( + self, + payload: dict[str, Any], + *, + log_type: str, + severity: str, + ) -> None: + if self._tripped.is_set(): + raise RuntimeError( + f"Background emitter for {self.name} is tripped after " + f"{self.failure_limit} consecutive batch failures." + ) + self._ensure_worker() + dropped_total: int | None = None + with self._lock: + if len(self._buffer) >= self.queue_size: + self._dropped += 1 + if ( + self._dropped == 1 + or self._dropped % DROP_REPORT_EVERY == 0 + ): + dropped_total = self._dropped + else: + self._buffer.append((payload, log_type, severity)) + if dropped_total is not None: + self.on_failure( + "logging.destination_queue_overflow", + RuntimeError( + "Background emitter buffer is full; dropping newest " + "log records." + ), + destination=self.name, + dropped_total=dropped_total, + ) + self._wake.set() + + def flush(self, deadline_seconds: float | None = None) -> None: + """Drain the buffer from the caller's thread, bounded by a + deadline; reports any undelivered remainder.""" + if deadline_seconds is None: + deadline_seconds = self.flush_deadline_seconds + deadline = time.monotonic() + deadline_seconds + while not self._tripped.is_set() and time.monotonic() < deadline: + if not self._drain_once(): + break + with self._lock: + remaining = len(self._buffer) + if remaining: + self.on_failure( + "logging.destination_flush_incomplete", + RuntimeError( + "Background emitter could not deliver all buffered " + "log records before the flush deadline." + ), + destination=self.name, + remaining=remaining, + ) + + def close(self) -> None: + """Stop the worker without flushing.""" + self._closed.set() + self._wake.set() + + def restart(self) -> None: + """Reset for a process whose memory was restored or forked: + drop buffered records, clear trip state, start a fresh worker on + the next emit.""" + with self._start_lock: + self._closed.set() + self._wake.set() + self._worker = None + self._pid = None + with self._lock: + self._buffer.clear() + self._dropped = 0 + self._consecutive_failures = 0 + self._tripped.clear() + + def _ensure_worker(self) -> None: + worker = self._worker + if ( + worker is not None + and worker.is_alive() + and self._pid == os.getpid() + ): + return + with self._start_lock: + worker = self._worker + if ( + worker is not None + and worker.is_alive() + and self._pid == os.getpid() + ): + return + closed = threading.Event() + self._closed = closed + self._wake = threading.Event() + thread = threading.Thread( + target=self._run, + args=(closed,), + name=f"observability-emit-{self.name}", + daemon=True, + ) + self._pid = os.getpid() + self._worker = thread + thread.start() + if not self._atexit_registered: + atexit.register(self.flush) + self._atexit_registered = True + + def _run(self, closed: threading.Event) -> None: + while not closed.is_set(): + self._wake.wait(timeout=self.batch_latency_seconds) + self._wake.clear() + while not closed.is_set() and self._drain_once(): + pass + + def _drain_once(self) -> bool: + """Write one batch; returns True while more work may remain.""" + with self._lock: + if not self._buffer: + return False + batch = [ + self._buffer.popleft() + for _ in range(min(self.batch_size, len(self._buffer))) + ] + return self._write_batch(batch) + + def _write_batch(self, batch: list[_LogRecord]) -> bool: + try: + emit_batch = getattr(self.wrapped, "emit_batch", None) + if callable(emit_batch): + emit_batch(batch) + else: + for payload, log_type, severity in batch: + self.wrapped.emit( + payload, + log_type=log_type, + severity=severity, + ) + except BaseException as exc: + self._consecutive_failures += 1 + self.on_failure( + "logging.destination_emit_async", + exc, + destination=self.name, + consecutive_failures=self._consecutive_failures, + ) + if self._consecutive_failures >= self.failure_limit: + self._tripped.set() + self._closed.set() + return not self._tripped.is_set() + self._consecutive_failures = 0 + return True diff --git a/tests/test_destinations.py b/tests/test_destinations.py index f3fe8ab..816fc8f 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -439,3 +439,256 @@ def test_manager_stdout_fallback_carries_configured_format() -> None: assert destination.output_format == "google" assert destination.google_cloud_project == "central-project" + + +# ── Background emitter ─────────────────────────────────────────────────── + + +class BatchRecordingDestination: + name = "batch-recording" + + def __init__(self) -> None: + self.batches = [] + self.single_emits = [] + + def emit(self, payload, *, log_type, severity) -> None: + self.single_emits.append((payload, log_type, severity)) + + def emit_batch(self, records) -> None: + self.batches.append(list(records)) + + +class FailingDestination: + name = "failing" + + def emit(self, payload, *, log_type, severity) -> None: + raise RuntimeError("sink down") + + +def _background(wrapped, **kwargs): + from policyengine_observability.destinations.background import ( + BackgroundEmitDestination, + ) + + failures = [] + destination = BackgroundEmitDestination( + wrapped, + on_failure=lambda operation, exc, **fields: failures.append( + (operation, fields) + ), + **kwargs, + ) + return destination, failures + + +def _inline(destination, monkeypatch): + """Disable the worker thread so tests drive draining synchronously.""" + monkeypatch.setattr(destination, "_ensure_worker", lambda: None) + return destination + + +def test_background_emit_enqueues_without_writing(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, _failures = _background(wrapped, batch_size=2) + _inline(destination, monkeypatch) + + for index in range(3): + destination.emit({"event": index}, log_type="event", severity="INFO") + + assert wrapped.batches == [] + assert len(destination._buffer) == 3 + + +def test_background_drain_prefers_emit_batch(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, _failures = _background(wrapped, batch_size=2) + _inline(destination, monkeypatch) + + for index in range(3): + destination.emit({"event": index}, log_type="event", severity="INFO") + while destination._drain_once(): + pass + + assert [len(batch) for batch in wrapped.batches] == [2, 1] + assert wrapped.single_emits == [] + assert wrapped.batches[0][0] == ({"event": 0}, "event", "INFO") + + +def test_background_drain_falls_back_to_single_emits(monkeypatch) -> None: + wrapped = RecordingDestination() + destination, _failures = _background(wrapped, batch_size=2) + _inline(destination, monkeypatch) + + destination.emit({"event": "a"}, log_type="event", severity="INFO") + while destination._drain_once(): + pass + + assert wrapped.payloads == [{"event": "a"}] + + +def test_background_overflow_drops_newest_and_reports(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped, queue_size=2) + _inline(destination, monkeypatch) + + for index in range(4): + destination.emit({"event": index}, log_type="event", severity="INFO") + + kept = [payload["event"] for payload, _, _ in destination._buffer] + assert kept == [0, 1] + overflow = [ + fields + for operation, fields in failures + if operation == "logging.destination_queue_overflow" + ] + assert len(overflow) == 1 + assert overflow[0]["dropped_total"] == 1 + assert destination._dropped == 2 + + +def test_background_trips_after_consecutive_batch_failures( + monkeypatch, +) -> None: + import pytest + + destination, failures = _background( + FailingDestination(), batch_size=1, failure_limit=3 + ) + _inline(destination, monkeypatch) + + for index in range(5): + destination.emit({"event": index}, log_type="event", severity="INFO") + drained = True + while drained: + drained = destination._drain_once() + + counts = [ + fields["consecutive_failures"] + for operation, fields in failures + if operation == "logging.destination_emit_async" + ] + assert counts == [1, 2, 3] + with pytest.raises(RuntimeError, match="tripped"): + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + +def test_background_trip_flows_through_manager_breaker(monkeypatch) -> None: + from policyengine_observability.destinations.manager import ( + DESTINATION_FAILURE_LIMIT, + ) + from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + ) + + destination, _failures = _background( + FailingDestination(), batch_size=1, failure_limit=1 + ) + _inline(destination, monkeypatch) + manager, manager_failures = _manager([destination]) + + manager.emit({"event": "seed"}, log_type="event", severity="INFO") + destination._drain_once() + for _ in range(DESTINATION_FAILURE_LIMIT): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + + assert destination not in manager.destinations + assert any( + operation == "logging.destination_disabled" + for operation, _fields in manager_failures + ) + assert any( + isinstance(existing, StdoutJsonDestination) + for existing in manager.destinations + ) + + +def test_background_flush_drains_everything(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped, batch_size=2) + _inline(destination, monkeypatch) + + for index in range(5): + destination.emit({"event": index}, log_type="event", severity="INFO") + destination.flush() + + assert sum(len(batch) for batch in wrapped.batches) == 5 + assert len(destination._buffer) == 0 + assert failures == [] + + +def test_background_flush_reports_undelivered_remainder( + monkeypatch, +) -> None: + destination, failures = _background( + FailingDestination(), batch_size=1, failure_limit=1 + ) + _inline(destination, monkeypatch) + + for index in range(4): + destination.emit({"event": index}, log_type="event", severity="INFO") + destination.flush() + + incomplete = [ + fields + for operation, fields in failures + if operation == "logging.destination_flush_incomplete" + ] + assert len(incomplete) == 1 + assert incomplete[0]["remaining"] == 3 + + +def test_background_restart_clears_trip_and_buffer(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, _failures = _background( + FailingDestination(), batch_size=1, failure_limit=1 + ) + _inline(destination, monkeypatch) + destination.emit({"event": "x"}, log_type="event", severity="INFO") + destination._drain_once() + assert destination._tripped.is_set() + + destination.restart() + destination.wrapped = wrapped + _inline(destination, monkeypatch) + + assert not destination._tripped.is_set() + assert len(destination._buffer) == 0 + destination.emit({"event": "y"}, log_type="event", severity="INFO") + destination._drain_once() + assert wrapped.batches == [[({"event": "y"}, "event", "INFO")]] + + +def test_background_worker_delivers_end_to_end() -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background( + wrapped, batch_size=10, batch_latency_seconds=0.01 + ) + + for index in range(3): + destination.emit({"event": index}, log_type="event", severity="INFO") + destination.flush(1.0) + destination.close() + + assert sum(len(batch) for batch in wrapped.batches) == 3 + assert failures == [] + assert destination._atexit_registered is True + + +def test_background_worker_restarts_after_fork(monkeypatch) -> None: + import os as os_module + + from policyengine_observability.destinations import background + + wrapped = BatchRecordingDestination() + destination, _failures = _background(wrapped) + destination.emit({"event": "a"}, log_type="event", severity="INFO") + first_worker = destination._worker + assert first_worker is not None + + real_pid = os_module.getpid() + monkeypatch.setattr(background.os, "getpid", lambda: real_pid + 1) + destination.emit({"event": "b"}, log_type="event", severity="INFO") + + assert destination._worker is not first_worker + destination.flush(1.0) + destination.close() From 4675af99961c0d0ce8ffaeae403815fdfc8ded45 Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Tue, 7 Jul 2026 22:00:07 +0200 Subject: [PATCH 4/7] Wire async emission through config, manager, runtime, and facade OBSERVABILITY_LOG_EMIT_MODE=async (default sync) wraps non-stdout destinations in the background emitter with knobs for queue size, batch size, batch latency, and flush deadline; stdout is never wrapped because it is the fallback sink. runtime.shutdown() now flushes log destinations before the OTel provider flush (which previously early-returned past everything when no providers existed), and the runtime exposes flush_log_destinations/restart_log_destinations, surfaced publicly as flush_observability/restart_observability for consumers whose processes fork or restore from memory snapshots. Refs #22 Co-Authored-By: Claude Fable 5 --- policyengine_observability/__init__.py | 10 + policyengine_observability/config.py | 36 +++ .../destinations/manager.py | 54 ++++- policyengine_observability/runtime.py | 15 ++ tests/test_destinations.py | 209 ++++++++++++++++++ 5 files changed, 320 insertions(+), 4 deletions(-) diff --git a/policyengine_observability/__init__.py b/policyengine_observability/__init__.py index 88a8fdf..e7baa68 100644 --- a/policyengine_observability/__init__.py +++ b/policyengine_observability/__init__.py @@ -105,6 +105,14 @@ def shutdown_observability() -> None: observability_runtime().shutdown() +def flush_observability(deadline_seconds: float | None = None) -> None: + observability_runtime().flush_log_destinations(deadline_seconds) + + +def restart_observability() -> None: + observability_runtime().restart_log_destinations() + + def shutdown_tracing() -> None: shutdown_observability() @@ -171,6 +179,8 @@ def collect_timings(name: str = "operation", **attrs: Any): "set_attribute", "set_observability_runtime", "shutdown_observability", + "flush_observability", + "restart_observability", "shutdown_tracing", "start_scope", "traceparent_header", diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index 7436fc5..1fb7d07 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -45,6 +45,16 @@ def csv_from_env(name: str) -> tuple[str, ...]: return tuple(part.strip() for part in raw_value.split(",") if part.strip()) +def int_from_env(name: str, default: int) -> int: + raw_value = os.getenv(name) + if raw_value is None: + return default + try: + return int(raw_value) + except ValueError: + return default + + def float_from_env(name: str, default: float) -> float: raw_value = os.getenv(name) if raw_value is None: @@ -89,6 +99,11 @@ class ObservabilityConfig: google_cloud_log_name: str = "policyengine-observability" google_log_timeout_seconds: float = 2.0 stdout_format: str = "plain" + log_emit_mode: str = "sync" + log_queue_size: int = 1000 + log_batch_size: int = 10 + log_batch_latency_seconds: float = 0.25 + log_flush_deadline_seconds: float = 5.0 @classmethod def from_env( @@ -177,6 +192,27 @@ def from_env( ) .strip() .lower(), + log_emit_mode=( + os.getenv("OBSERVABILITY_LOG_EMIT_MODE") or cls.log_emit_mode + ) + .strip() + .lower(), + log_queue_size=int_from_env( + "OBSERVABILITY_LOG_QUEUE_SIZE", + cls.log_queue_size, + ), + log_batch_size=int_from_env( + "OBSERVABILITY_LOG_BATCH_SIZE", + cls.log_batch_size, + ), + log_batch_latency_seconds=float_from_env( + "OBSERVABILITY_LOG_BATCH_LATENCY_SECONDS", + cls.log_batch_latency_seconds, + ), + log_flush_deadline_seconds=float_from_env( + "OBSERVABILITY_LOG_FLUSH_DEADLINE_SECONDS", + cls.log_flush_deadline_seconds, + ), ) diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index e965edb..fa5a269 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -5,6 +5,7 @@ from typing import Any from ..config import ObservabilityConfig +from .background import BackgroundEmitDestination from .base import LogDestination from .google_cloud_logging import GoogleCloudLoggingDestination from .stdout import StdoutJsonDestination @@ -129,15 +130,60 @@ def _build_destination(self, destination_name: str) -> LogDestination: if normalized == "stdout": return self._stdout_destination() if normalized in {"google", "google_cloud", "google_cloud_logging"}: - return GoogleCloudLoggingDestination( - project=self.config.google_cloud_project, - log_name=self.config.google_cloud_log_name, - timeout_seconds=self.config.google_log_timeout_seconds, + return self._maybe_background( + GoogleCloudLoggingDestination( + project=self.config.google_cloud_project, + log_name=self.config.google_cloud_log_name, + timeout_seconds=self.config.google_log_timeout_seconds, + ) ) raise ValueError( f"Unknown observability log destination: {destination_name}" ) + def _maybe_background(self, destination: LogDestination) -> LogDestination: + # Stdout is the fallback sink and stays synchronous; it must keep + # working when everything else (including threads) is broken. + if self.config.log_emit_mode != "async": + return destination + return BackgroundEmitDestination( + destination, + on_failure=self.on_failure, + queue_size=self.config.log_queue_size, + batch_size=self.config.log_batch_size, + batch_latency_seconds=self.config.log_batch_latency_seconds, + flush_deadline_seconds=self.config.log_flush_deadline_seconds, + failure_limit=DESTINATION_FAILURE_LIMIT, + ) + + def flush(self, deadline_seconds: float | None = None) -> None: + for destination in list(self.destinations): + flush = getattr(destination, "flush", None) + if not callable(flush): + continue + try: + flush(deadline_seconds) + except BaseException as exc: + self.on_failure( + "logging.destination_flush", + exc, + destination=getattr(destination, "name", None), + ) + + def restart(self) -> None: + for destination in list(self.destinations): + restart = getattr(destination, "restart", None) + if not callable(restart): + continue + try: + restart() + except BaseException as exc: + self.on_failure( + "logging.destination_restart", + exc, + destination=getattr(destination, "name", None), + ) + def _stdout_destination(self) -> StdoutJsonDestination: return StdoutJsonDestination( loggers=self.loggers, diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 8fbda49..11d14da 100644 --- a/policyengine_observability/runtime.py +++ b/policyengine_observability/runtime.py @@ -1070,7 +1070,22 @@ def instrument_httpx(self) -> None: except BaseException as exc: self.log_observability_failure("httpx.auto_instrument", exc) + def flush_log_destinations( + self, deadline_seconds: float | None = None + ) -> None: + try: + self.log_destination_manager.flush(deadline_seconds) + except BaseException as exc: + self.log_observability_failure("logging.flush", exc) + + def restart_log_destinations(self) -> None: + try: + self.log_destination_manager.restart() + except BaseException as exc: + self.log_observability_failure("logging.restart", exc) + def shutdown(self) -> None: + self.flush_log_destinations() providers = [ ("trace", self.tracer_provider), ("metrics", self.meter_provider), diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 816fc8f..639a400 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -692,3 +692,212 @@ def test_background_worker_restarts_after_fork(monkeypatch) -> None: assert destination._worker is not first_worker destination.flush(1.0) destination.close() + + +# ── Async wiring, lifecycle, and config knobs ──────────────────────────── + + +def _config_manager(config): + import json + import logging + + from policyengine_observability.destinations.manager import ( + LogDestinationManager, + ) + + return LogDestinationManager( + config=config, + loggers={"event": logging.getLogger("test-wiring")}, + serializer=json.dumps, + on_failure=lambda *args, **kwargs: None, + ) + + +def test_async_mode_wraps_google_destination(monkeypatch) -> None: + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations import manager as manager_mod + from policyengine_observability.destinations.background import ( + BackgroundEmitDestination, + ) + + class StubGoogle: + name = "google_cloud_logging" + + def __init__(self, **kwargs) -> None: + self.kwargs = kwargs + + def emit(self, payload, *, log_type, severity) -> None: + pass + + monkeypatch.setattr( + manager_mod, "GoogleCloudLoggingDestination", StubGoogle + ) + manager = _config_manager( + ObservabilityConfig(log_emit_mode="async", log_queue_size=7) + ) + + destination = manager._build_destination("google_cloud_logging") + + assert isinstance(destination, BackgroundEmitDestination) + assert isinstance(destination.wrapped, StubGoogle) + assert destination.queue_size == 7 + assert destination.name == "google_cloud_logging" + + +def test_sync_mode_returns_bare_destination(monkeypatch) -> None: + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations import manager as manager_mod + + class StubGoogle: + name = "google_cloud_logging" + + def __init__(self, **kwargs) -> None: + pass + + def emit(self, payload, *, log_type, severity) -> None: + pass + + monkeypatch.setattr( + manager_mod, "GoogleCloudLoggingDestination", StubGoogle + ) + manager = _config_manager(ObservabilityConfig()) + + destination = manager._build_destination("google") + + assert isinstance(destination, StubGoogle) + + +def test_async_mode_never_wraps_stdout() -> None: + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + ) + + manager = _config_manager(ObservabilityConfig(log_emit_mode="async")) + + destination = manager._build_destination("stdout") + + assert isinstance(destination, StdoutJsonDestination) + + +class FlushRecordingDestination: + name = "flush-recording" + + def __init__(self) -> None: + self.flushes = [] + self.restarts = 0 + + def emit(self, payload, *, log_type, severity) -> None: + pass + + def flush(self, deadline_seconds=None) -> None: + self.flushes.append(deadline_seconds) + + def restart(self) -> None: + self.restarts += 1 + + +def test_manager_flush_and_restart_reach_capable_destinations() -> None: + flushable = FlushRecordingDestination() + plain = RecordingDestination() + manager, failures = _manager([flushable, plain]) + + manager.flush(1.5) + manager.restart() + + assert flushable.flushes == [1.5] + assert flushable.restarts == 1 + assert failures == [] + + +def test_manager_flush_reports_but_survives_failures() -> None: + class ExplodingFlush(FlushRecordingDestination): + def flush(self, deadline_seconds=None) -> None: + raise RuntimeError("flush failed") + + manager, failures = _manager([ExplodingFlush()]) + + manager.flush() + + assert any( + operation == "logging.destination_flush" + for operation, _fields in failures + ) + + +def test_runtime_shutdown_flushes_log_destinations() -> None: + from policyengine_observability import ( + ObservabilityConfig, + ObservabilityRuntime, + ) + + runtime = ObservabilityRuntime( + ObservabilityConfig(service_name="svc", otel_enabled=False) + ) + flushable = FlushRecordingDestination() + runtime.log_destination_manager.destinations = [flushable] + runtime.log_destination_manager.configured = True + + runtime.shutdown() + + assert flushable.flushes == [None] + + +def test_public_flush_and_restart_facades(monkeypatch) -> None: + import policyengine_observability as observability + + calls = [] + + class StubRuntime: + def flush_log_destinations(self, deadline_seconds=None) -> None: + calls.append(("flush", deadline_seconds)) + + def restart_log_destinations(self) -> None: + calls.append(("restart", None)) + + monkeypatch.setattr( + observability, "observability_runtime", lambda: StubRuntime() + ) + + observability.flush_observability(2.0) + observability.restart_observability() + + assert calls == [("flush", 2.0), ("restart", None)] + assert "flush_observability" in observability.__all__ + assert "restart_observability" in observability.__all__ + + +def test_from_env_parses_emission_knobs(monkeypatch) -> None: + from policyengine_observability.config import ObservabilityConfig + + monkeypatch.setenv("OBSERVABILITY_LOG_EMIT_MODE", "Async") + monkeypatch.setenv("OBSERVABILITY_STDOUT_FORMAT", "Google") + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_SIZE", "50") + monkeypatch.setenv("OBSERVABILITY_LOG_BATCH_SIZE", "5") + monkeypatch.setenv("OBSERVABILITY_LOG_BATCH_LATENCY_SECONDS", "0.5") + monkeypatch.setenv("OBSERVABILITY_LOG_FLUSH_DEADLINE_SECONDS", "9") + monkeypatch.setenv("OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS", "1.5") + + config = ObservabilityConfig.from_env(service_name="svc") + + assert config.log_emit_mode == "async" + assert config.stdout_format == "google" + assert config.log_queue_size == 50 + assert config.log_batch_size == 5 + assert config.log_batch_latency_seconds == 0.5 + assert config.log_flush_deadline_seconds == 9.0 + assert config.google_log_timeout_seconds == 1.5 + + +def test_from_env_emission_knobs_default_and_reject_garbage( + monkeypatch, +) -> None: + from policyengine_observability.config import ObservabilityConfig + + monkeypatch.delenv("OBSERVABILITY_LOG_EMIT_MODE", raising=False) + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_SIZE", "not-a-number") + + config = ObservabilityConfig.from_env(service_name="svc") + + assert config.log_emit_mode == "sync" + assert config.log_queue_size == 1000 From 20d21611ad9a5d8db8601d0568c557529de3c9be Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Tue, 7 Jul 2026 22:02:58 +0200 Subject: [PATCH 5/7] Document emission semantics, new knobs, and kill-switches Refs #22 Co-Authored-By: Claude Fable 5 --- README.md | 55 +++++++++++++++++++ .../operations/google-cloud-stage3-runbook.md | 11 ++++ 2 files changed, 66 insertions(+) diff --git a/README.md b/README.md index 5a2ca2b..6193cfd 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,61 @@ The Google Cloud setup needs: For the fixed PolicyEngine Google Cloud destination, see [`docs/operations/google-cloud-stage3-runbook.md`](docs/operations/google-cloud-stage3-runbook.md). +## Log emission and delivery semantics + +Writes to Google Cloud Logging carry an explicit per-call timeout with +retries disabled so a degraded Logging API cannot stall the caller: + +```bash +OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS=2.0 +``` + +By default, log emission is synchronous on the caller's thread. Setting +the emit mode to `async` installs the standard accept/emit split: the +logging call only appends the record to a bounded in-memory buffer and +returns immediately, and a background worker thread drains the buffer in +batches to the network destination: + +```bash +OBSERVABILITY_LOG_EMIT_MODE=async +OBSERVABILITY_LOG_QUEUE_SIZE=1000 +OBSERVABILITY_LOG_BATCH_SIZE=10 +OBSERVABILITY_LOG_BATCH_LATENCY_SECONDS=0.25 +OBSERVABILITY_LOG_FLUSH_DEADLINE_SECONDS=5.0 +``` + +Async delivery is best-effort by design. When the buffer is full, the +newest records are dropped and the drops are counted and reported through +the internal-error channel. After three consecutive failed batches the +destination is disabled for the remainder of the process and logging +falls back to stdout, matching the synchronous circuit breaker. Buffered +records are flushed at interpreter exit and by `runtime.shutdown()`; +`flush_observability(deadline_seconds)` flushes on demand. A hard kill +loses whatever was still buffered. Stdout destinations are never wrapped: +the fallback sink stays synchronous and dependency-free. + +Processes that fork or restore from memory snapshots (for example Modal +Functions with memory snapshots enabled) do not preserve threads. The +worker is pid-aware and restarts automatically after a fork; runtimes +that restore process memory should call `restart_observability()` in +their post-restore hook to clear buffered state and revive the worker. + +On platforms whose logging agent collects stdout (Cloud Run, GKE), the +agent-native stdout format emits JSON lines carrying the special keys the +agent promotes to first-class LogEntry fields (severity, time, trace, +span, labels), giving full-fidelity Cloud Logging ingestion with no +in-process network emission: + +```bash +OBSERVABILITY_LOG_DESTINATIONS=stdout +OBSERVABILITY_STDOUT_FORMAT=google +OBSERVABILITY_GOOGLE_CLOUD_PROJECT=PROJECT_ID +``` + +Because entries are written directly rather than through +`Logger.log_struct`, the Google client library's one-time instrumentation +diagnostic entry is not emitted. + ## Release workflow Changes should include a Towncrier fragment in `changelog.d/`. Pull requests diff --git a/docs/operations/google-cloud-stage3-runbook.md b/docs/operations/google-cloud-stage3-runbook.md index 6f0ea5d..0e59c2d 100644 --- a/docs/operations/google-cloud-stage3-runbook.md +++ b/docs/operations/google-cloud-stage3-runbook.md @@ -187,3 +187,14 @@ OBSERVABILITY_GOOGLE_SERVICE_ACCOUNT_EMAIL= Date: Wed, 8 Jul 2026 18:41:27 +0200 Subject: [PATCH 6/7] Address review findings across the emission suite Recovery: manager.restart() now rebuilds destinations from config, reviving breaker-disabled destinations with fresh clients (the Modal snapshot-restore case), clearing the failure ledger; configure() closes the destinations it replaces so reconfigures cannot orphan workers, clients, or atexit hooks, and close() unregisters its atexit hook. Resilience: Google writes retry transient errors inside the bounded budget (default now 5s) instead of retry=None turning every blip into a breaker strike; the worker backs off between failed batches and reports records_lost; a one-time report fires when only the unbounded non-gapic transport is available; the buffer drops OLDEST on overflow, keeping the freshest records through an outage. Concurrency: payloads are normalized at enqueue so caller mutation cannot tear records; flush() tracks and waits for the in-flight batch and includes it in the remainder report; superseded workers are signaled before their generation event is replaced (fixing a leaked worker the fork test now asserts against); one persistent wake event removes the lost-wakeup window; the runtime's internal-error flag is thread-local; manager disable is idempotent under reentrant failure reporting; _write_batch catches Exception, letting KeyboardInterrupt propagate from main-thread flushes. Hygiene: knob clamps (queue >= 1, latency >= 10ms, batch <= 500); _drain_once returns explicit empty/delivered/failed outcomes; async wrapping moved to a single choke point; emit-mode comparison normalized; shared trace/timestamp helpers in destinations.base with event-time stamping from created_at; defaults defined once in config; Protocol declares logging_api; BackgroundEmitDestination exported; duplicate test fakes consolidated; README and fragments corrected. Refs #22 Co-Authored-By: Claude Fable 5 --- README.md | 25 ++- changelog.d/22.added.md | 2 +- changelog.d/22.changed.md | 2 +- policyengine_observability/config.py | 16 +- .../destinations/__init__.py | 2 + .../destinations/background.py | 138 ++++++++---- .../destinations/base.py | 24 ++ .../destinations/google_cloud_logging.py | 142 ++++++++---- .../destinations/manager.py | 69 ++++-- policyengine_observability/runtime.py | 13 +- tests/test_destinations.py | 207 ++++++++++++++++-- tests/test_google_credentials.py | 22 +- 12 files changed, 510 insertions(+), 152 deletions(-) diff --git a/README.md b/README.md index 6193cfd..1dd8b7a 100644 --- a/README.md +++ b/README.md @@ -105,11 +105,12 @@ For the fixed PolicyEngine Google Cloud destination, see ## Log emission and delivery semantics -Writes to Google Cloud Logging carry an explicit per-call timeout with -retries disabled so a degraded Logging API cannot stall the caller: +Writes to Google Cloud Logging carry an explicit per-call timeout, with +transient errors retried only inside that budget, so a degraded Logging +API cannot stall the caller while brief blips are still absorbed: ```bash -OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS=2.0 +OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS=5.0 ``` By default, log emission is synchronous on the caller's thread. Setting @@ -127,14 +128,18 @@ OBSERVABILITY_LOG_FLUSH_DEADLINE_SECONDS=5.0 ``` Async delivery is best-effort by design. When the buffer is full, the -newest records are dropped and the drops are counted and reported through -the internal-error channel. After three consecutive failed batches the -destination is disabled for the remainder of the process and logging -falls back to stdout, matching the synchronous circuit breaker. Buffered +oldest records are dropped — keeping the freshest, most diagnostic +records — and drops are counted and reported through the internal-error +channel. After three consecutive failed batches (with backoff between +attempts) the emitter trips; subsequent log calls surface the failure to +the manager's circuit breaker, which disables the destination and falls +back to stdout. `restart_observability()` rebuilds destinations from +config, reviving a disabled destination with a fresh client. Buffered records are flushed at interpreter exit and by `runtime.shutdown()`; -`flush_observability(deadline_seconds)` flushes on demand. A hard kill -loses whatever was still buffered. Stdout destinations are never wrapped: -the fallback sink stays synchronous and dependency-free. +`flush_observability(deadline_seconds)` flushes on demand and waits for +any in-flight batch. A hard kill loses whatever was still buffered. +Stdout destinations are never wrapped: the fallback sink stays +synchronous and dependency-free. Processes that fork or restore from memory snapshots (for example Modal Functions with memory snapshots enabled) do not preserve threads. The diff --git a/changelog.d/22.added.md b/changelog.d/22.added.md index 3f0ef4a..e8a4aa3 100644 --- a/changelog.d/22.added.md +++ b/changelog.d/22.added.md @@ -1 +1 @@ -Added an agent-native stdout format (`OBSERVABILITY_STDOUT_FORMAT=google`): JSON lines carry the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, time, trace, span, labels), enabling full-fidelity Cloud Logging ingestion from stdout with no in-process network emission. Added a background log emitter (`OBSERVABILITY_LOG_EMIT_MODE=async`): log calls enqueue onto a bounded buffer and a worker thread batches writes off the request path, with flush-on-shutdown, post-fork/snapshot `restart_observability()`, and the existing destination circuit breaker preserved. +Added an agent-native stdout format (`OBSERVABILITY_STDOUT_FORMAT=google`): JSON lines carry the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, time, trace, span, labels), enabling full-fidelity Cloud Logging ingestion from stdout with no in-process network emission. Added a background log emitter (`OBSERVABILITY_LOG_EMIT_MODE=async`): log calls enqueue onto a bounded drop-oldest buffer and a worker thread batches writes off the request path, with flush-on-shutdown that waits for in-flight batches, and `restart_observability()` rebuilding destinations from config for forked or snapshot-restored processes — including destinations the circuit breaker disabled. diff --git a/changelog.d/22.changed.md b/changelog.d/22.changed.md index 6ce4776..ce49579 100644 --- a/changelog.d/22.changed.md +++ b/changelog.d/22.changed.md @@ -1 +1 @@ -Google Cloud Logging writes now carry an explicit per-call timeout with retries disabled (default 2 seconds, `OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS`), replacing the transport defaults that could block a caller for up to 60 seconds per record when the Logging API degrades. The destination also gains `emit_batch`, writing multiple records in one bounded call. +Google Cloud Logging writes now carry an explicit per-call timeout (default 5 seconds, `OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS`) with transient errors retried only inside that budget, replacing the transport defaults that could block a caller for up to 60 seconds per record when the Logging API degrades. The destination also gains `emit_batch`, writing multiple records in one bounded call, and reports once at startup when only the unbounded non-gapic transport is available. diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index 1fb7d07..9ba5498 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -5,6 +5,12 @@ from collections.abc import Sequence from dataclasses import dataclass +DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS = 5.0 +DEFAULT_LOG_QUEUE_SIZE = 1000 +DEFAULT_LOG_BATCH_SIZE = 10 +DEFAULT_LOG_BATCH_LATENCY_SECONDS = 0.25 +DEFAULT_LOG_FLUSH_DEADLINE_SECONDS = 5.0 + DEFAULT_METRIC_ATTRIBUTE_KEYS = ( "service.name", "service.role", @@ -97,13 +103,13 @@ class ObservabilityConfig: log_destinations: tuple[str, ...] = ("stdout",) google_cloud_project: str | None = None google_cloud_log_name: str = "policyengine-observability" - google_log_timeout_seconds: float = 2.0 + google_log_timeout_seconds: float = DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS stdout_format: str = "plain" log_emit_mode: str = "sync" - log_queue_size: int = 1000 - log_batch_size: int = 10 - log_batch_latency_seconds: float = 0.25 - log_flush_deadline_seconds: float = 5.0 + log_queue_size: int = DEFAULT_LOG_QUEUE_SIZE + log_batch_size: int = DEFAULT_LOG_BATCH_SIZE + log_batch_latency_seconds: float = DEFAULT_LOG_BATCH_LATENCY_SECONDS + log_flush_deadline_seconds: float = DEFAULT_LOG_FLUSH_DEADLINE_SECONDS @classmethod def from_env( diff --git a/policyengine_observability/destinations/__init__.py b/policyengine_observability/destinations/__init__.py index 5703e23..bdffd8f 100644 --- a/policyengine_observability/destinations/__init__.py +++ b/policyengine_observability/destinations/__init__.py @@ -1,11 +1,13 @@ from __future__ import annotations +from .background import BackgroundEmitDestination from .base import LogDestination, normalize_payload from .google_cloud_logging import GoogleCloudLoggingDestination from .manager import LogDestinationManager from .stdout import StdoutJsonDestination __all__ = [ + "BackgroundEmitDestination", "GoogleCloudLoggingDestination", "LogDestination", "LogDestinationManager", diff --git a/policyengine_observability/destinations/background.py b/policyengine_observability/destinations/background.py index 4cf1500..e850362 100644 --- a/policyengine_observability/destinations/background.py +++ b/policyengine_observability/destinations/background.py @@ -8,35 +8,48 @@ from collections.abc import Callable from typing import Any -from .base import LogDestination +from ..config import ( + DEFAULT_LOG_BATCH_LATENCY_SECONDS, + DEFAULT_LOG_BATCH_SIZE, + DEFAULT_LOG_FLUSH_DEADLINE_SECONDS, + DEFAULT_LOG_QUEUE_SIZE, +) +from .base import LogDestination, normalize_payload -DEFAULT_QUEUE_SIZE = 1000 -DEFAULT_BATCH_SIZE = 10 -DEFAULT_BATCH_LATENCY_SECONDS = 0.25 -DEFAULT_FLUSH_DEADLINE_SECONDS = 5.0 DEFAULT_FAILURE_LIMIT = 3 +DEFAULT_FAILURE_BACKOFF_SECONDS = 1.0 +MIN_BATCH_LATENCY_SECONDS = 0.01 +MAX_BATCH_SIZE = 500 DROP_REPORT_EVERY = 100 +# _drain_once outcomes: each signal has exactly one meaning. +DRAIN_EMPTY = "empty" +DRAIN_DELIVERED = "delivered" +DRAIN_FAILED = "failed" + _LogRecord = tuple[dict[str, Any], str, str] class BackgroundEmitDestination: """Decouples log acceptance from emission. - ``emit()`` appends to a bounded in-memory buffer and returns - immediately — it never blocks on the network and never raises for - sink trouble. A daemon worker drains the buffer in batches to the - wrapped destination (preferring its ``emit_batch``). Overflow drops - the newest record and counts it; after ``failure_limit`` consecutive - failed batches the wrapper trips, and subsequent ``emit()`` calls - raise so the manager's existing circuit breaker disables it through - the same path as a synchronous destination (stdout fallback - preserved). + ``emit()`` normalizes the payload (snapshotting it against caller + mutation), appends it to a bounded in-memory buffer, and returns + immediately — it never blocks on the network. When the buffer is + full the oldest record is dropped (counted and reported, throttled), + keeping the freshest records: during a sink outage the most + diagnostic records are the recent ones. A daemon worker drains the + buffer in batches to the wrapped destination (preferring its + ``emit_batch``), backing off between failed batches; after + ``failure_limit`` consecutive failed batches the wrapper trips, and + subsequent ``emit()`` calls raise so the manager's circuit breaker + disables it through the same path as a synchronous destination. The worker starts lazily on first emit and is pid-aware, so a forked - or snapshot-restored process (where threads do not survive) starts a - fresh worker automatically; ``restart()`` additionally clears the - buffer and trip state for consumers that restore process memory. + process starts a fresh worker automatically; ``restart()`` + additionally clears buffered and trip state. Recovery of a tripped + and disabled destination happens at the manager level + (``LogDestinationManager.restart`` rebuilds destinations). """ def __init__( @@ -44,23 +57,29 @@ def __init__( wrapped: LogDestination, *, on_failure: Callable[..., None], - queue_size: int = DEFAULT_QUEUE_SIZE, - batch_size: int = DEFAULT_BATCH_SIZE, - batch_latency_seconds: float = DEFAULT_BATCH_LATENCY_SECONDS, - flush_deadline_seconds: float = DEFAULT_FLUSH_DEADLINE_SECONDS, + queue_size: int = DEFAULT_LOG_QUEUE_SIZE, + batch_size: int = DEFAULT_LOG_BATCH_SIZE, + batch_latency_seconds: float = DEFAULT_LOG_BATCH_LATENCY_SECONDS, + flush_deadline_seconds: float = DEFAULT_LOG_FLUSH_DEADLINE_SECONDS, failure_limit: int = DEFAULT_FAILURE_LIMIT, + failure_backoff_seconds: float = DEFAULT_FAILURE_BACKOFF_SECONDS, ) -> None: self.wrapped = wrapped self.name = getattr(wrapped, "name", "background") self.on_failure = on_failure - self.queue_size = queue_size - self.batch_size = max(1, batch_size) - self.batch_latency_seconds = batch_latency_seconds - self.flush_deadline_seconds = flush_deadline_seconds + self.queue_size = max(1, queue_size) + self.batch_size = min(max(1, batch_size), MAX_BATCH_SIZE) + self.batch_latency_seconds = max( + MIN_BATCH_LATENCY_SECONDS, batch_latency_seconds + ) + self.flush_deadline_seconds = max(0.0, flush_deadline_seconds) self.failure_limit = failure_limit - self._buffer: deque[_LogRecord] = deque() + self.failure_backoff_seconds = max(0.0, failure_backoff_seconds) + self._buffer: deque[_LogRecord] = deque(maxlen=self.queue_size) self._lock = threading.Lock() self._start_lock = threading.Lock() + # One wake event for the object's lifetime; only the closed event + # is generation-scoped, so emitters can never signal a stale one. self._wake = threading.Event() self._tripped = threading.Event() self._closed = threading.Event() @@ -68,6 +87,7 @@ def __init__( self._pid: int | None = None self._consecutive_failures = 0 self._dropped = 0 + self._in_flight = 0 self._atexit_registered = False def emit( @@ -83,22 +103,24 @@ def emit( f"{self.failure_limit} consecutive batch failures." ) self._ensure_worker() + # Snapshot now: the caller may keep mutating nested structures + # after this returns, and serialization happens on the worker. + record = (normalize_payload(payload), log_type, severity) dropped_total: int | None = None with self._lock: - if len(self._buffer) >= self.queue_size: + if len(self._buffer) == self.queue_size: self._dropped += 1 if ( self._dropped == 1 or self._dropped % DROP_REPORT_EVERY == 0 ): dropped_total = self._dropped - else: - self._buffer.append((payload, log_type, severity)) + self._buffer.append(record) if dropped_total is not None: self.on_failure( "logging.destination_queue_overflow", RuntimeError( - "Background emitter buffer is full; dropping newest " + "Background emitter buffer is full; dropping oldest " "log records." ), destination=self.name, @@ -108,15 +130,23 @@ def emit( def flush(self, deadline_seconds: float | None = None) -> None: """Drain the buffer from the caller's thread, bounded by a - deadline; reports any undelivered remainder.""" + deadline; waits for a worker-held in-flight batch and reports + any undelivered remainder.""" if deadline_seconds is None: deadline_seconds = self.flush_deadline_seconds deadline = time.monotonic() + deadline_seconds while not self._tripped.is_set() and time.monotonic() < deadline: - if not self._drain_once(): + outcome = self._drain_once() + if outcome != DRAIN_EMPTY: + continue + with self._lock: + idle = not self._buffer and self._in_flight == 0 + if idle: break + # A batch is in flight on the worker; give it a moment. + time.sleep(MIN_BATCH_LATENCY_SECONDS) with self._lock: - remaining = len(self._buffer) + remaining = len(self._buffer) + self._in_flight if remaining: self.on_failure( "logging.destination_flush_incomplete", @@ -132,11 +162,15 @@ def close(self) -> None: """Stop the worker without flushing.""" self._closed.set() self._wake.set() + if self._atexit_registered: + atexit.unregister(self.flush) + self._atexit_registered = False def restart(self) -> None: - """Reset for a process whose memory was restored or forked: - drop buffered records, clear trip state, start a fresh worker on - the next emit.""" + """Reset local state: drop buffered records, clear trip state, + start a fresh worker on the next emit. A destination the manager + already disabled cannot be revived here — use the manager-level + restart, which rebuilds destinations.""" with self._start_lock: self._closed.set() self._wake.set() @@ -164,9 +198,11 @@ def _ensure_worker(self) -> None: and self._pid == os.getpid() ): return + # Signal any superseded-but-alive worker before replacing its + # generation event, so it exits instead of leaking. + self._closed.set() closed = threading.Event() self._closed = closed - self._wake = threading.Event() thread = threading.Thread( target=self._run, args=(closed,), @@ -184,19 +220,30 @@ def _run(self, closed: threading.Event) -> None: while not closed.is_set(): self._wake.wait(timeout=self.batch_latency_seconds) self._wake.clear() - while not closed.is_set() and self._drain_once(): - pass + while not closed.is_set(): + outcome = self._drain_once() + if outcome == DRAIN_EMPTY: + break + if outcome == DRAIN_FAILED: + # Back off before hammering a failing sink again. + closed.wait(timeout=self.failure_backoff_seconds) - def _drain_once(self) -> bool: - """Write one batch; returns True while more work may remain.""" + def _drain_once(self) -> str: + """Write one batch; returns a DRAIN_* outcome.""" with self._lock: if not self._buffer: - return False + return DRAIN_EMPTY batch = [ self._buffer.popleft() for _ in range(min(self.batch_size, len(self._buffer))) ] - return self._write_batch(batch) + self._in_flight += len(batch) + try: + delivered = self._write_batch(batch) + finally: + with self._lock: + self._in_flight -= len(batch) + return DRAIN_DELIVERED if delivered else DRAIN_FAILED def _write_batch(self, batch: list[_LogRecord]) -> bool: try: @@ -210,17 +257,18 @@ def _write_batch(self, batch: list[_LogRecord]) -> bool: log_type=log_type, severity=severity, ) - except BaseException as exc: + except Exception as exc: self._consecutive_failures += 1 self.on_failure( "logging.destination_emit_async", exc, destination=self.name, consecutive_failures=self._consecutive_failures, + records_lost=len(batch), ) if self._consecutive_failures >= self.failure_limit: self._tripped.set() self._closed.set() - return not self._tripped.is_set() + return False self._consecutive_failures = 0 return True diff --git a/policyengine_observability/destinations/base.py b/policyengine_observability/destinations/base.py index 00c7fe1..1f43849 100644 --- a/policyengine_observability/destinations/base.py +++ b/policyengine_observability/destinations/base.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import Mapping, Sequence +from datetime import datetime from typing import Any, Protocol @@ -53,3 +54,26 @@ def bounded_labels( if value is not None: labels[key] = str(value) return labels + + +def trace_resource_name( + project: str | None, + trace_id: Any, +) -> str | None: + if not trace_id or not project: + return None + return f"projects/{project}/traces/{trace_id}" + + +def rfc3339_timestamp(value: Any) -> str | None: + """Return an RFC3339 timestamp for a payload-supplied value, or None + when the value cannot be interpreted as an aware datetime.""" + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + if parsed.tzinfo is None: + return None + return parsed.isoformat() diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 98d8d75..62d18b0 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -3,14 +3,20 @@ from collections.abc import Callable, Sequence from typing import Any, Protocol +from policyengine_observability.config import ( + DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS, +) from policyengine_observability.google_credentials import ( configure_google_application_credentials, load_google_credentials, ) -from .base import bounded_labels, normalize_payload - -DEFAULT_WRITE_TIMEOUT_SECONDS = 2.0 +from .base import ( + bounded_labels, + normalize_payload, + rfc3339_timestamp, + trace_resource_name, +) class GoogleCloudLogger(Protocol): @@ -20,6 +26,7 @@ class GoogleCloudLogger(Protocol): class GoogleCloudLoggingClient(Protocol): project: str | None + logging_api: Any def logger(self, log_name: str) -> GoogleCloudLogger: ... @@ -33,12 +40,15 @@ def logger(self, log_name: str) -> GoogleCloudLogger: ... class GoogleCloudLoggingDestination: """Writes structured payloads to Google Cloud Logging. - Writes go through the transport layer directly with an explicit - per-call timeout and retries disabled: ``Logger.log_struct`` offers no - call options, and the underlying defaults (60s retry deadline) would - let a degraded Logging API stall the caller far longer than any + Writes go through the transport layer directly with a bounded retry + and an explicit per-call timeout: ``Logger.log_struct`` offers no call + options, and the underlying defaults (60s retry deadline) would let a + degraded Logging API stall the caller far longer than any observability write is worth. The ``Logger`` object is still used for - its ``full_name`` and platform-detected ``default_resource``. + its ``full_name`` and platform-detected ``default_resource``. The + bounded path needs the gapic transport; when it is unavailable the + destination reports that once through ``on_failure`` and falls back + to the unbounded ``write_entries`` call rather than dropping logs. """ name = "google_cloud_logging" @@ -49,7 +59,8 @@ def __init__( project: str | None, log_name: str, client_factory: GoogleCloudLoggingClientFactory | None = None, - timeout_seconds: float = DEFAULT_WRITE_TIMEOUT_SECONDS, + timeout_seconds: float = DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS, + on_failure: Callable[..., None] | None = None, ) -> None: self.project = project self.log_name = log_name @@ -72,6 +83,63 @@ def client_factory( self.client = client_factory(project, credentials) self.project = project or getattr(self.client, "project", None) self.logger = self.client.logger(log_name) + self._full_name = self.logger.full_name + self._resource = _resource_dict(self.logger.default_resource) + self._gapic_api = None + self._gapic_tools: tuple[Any, Any] | None = None + self._retry: Any = None + self._resolve_transport(on_failure) + + def _resolve_transport( + self, on_failure: Callable[..., None] | None + ) -> None: + api = getattr(self.client, "logging_api", None) + gapic_api = getattr(api, "_gapic_api", None) + if gapic_api is not None: + try: + from google.api_core import exceptions as api_exceptions + from google.api_core.retry import ( + Retry, + if_exception_type, + ) + from google.cloud.logging_v2._gapic import ( + _log_entry_mapping_to_pb, + ) + from google.cloud.logging_v2.types import ( + WriteLogEntriesRequest, + ) + except ImportError: # pragma: no cover - exotic installs only + gapic_api = None + else: + self._gapic_api = gapic_api + self._gapic_tools = ( + _log_entry_mapping_to_pb, + WriteLogEntriesRequest, + ) + # Retry transient errors, but only inside the overall + # write budget, so a Logging API blip does not surface as + # a destination failure while a real outage stays bounded. + self._retry = Retry( + initial=0.1, + maximum=1.0, + multiplier=1.3, + timeout=self.timeout_seconds, + predicate=if_exception_type( + api_exceptions.DeadlineExceeded, + api_exceptions.InternalServerError, + api_exceptions.ServiceUnavailable, + ), + ) + if self._gapic_api is None and on_failure is not None: + on_failure( + "logging.destination_unbounded_transport", + RuntimeError( + "Google Cloud Logging gapic transport unavailable; " + "writes fall back to the unbounded write_entries " + "call." + ), + destination=self.name, + ) def emit( self, @@ -105,49 +173,39 @@ def _build_entry( ) -> dict[str, Any]: normalized = normalize_payload(payload) entry: dict[str, Any] = { - "logName": self.logger.full_name, - "resource": _resource_dict(self.logger.default_resource), + "logName": self._full_name, + "resource": self._resource, "jsonPayload": normalized, "severity": str(severity).upper(), "labels": bounded_labels(normalized, log_type=log_type), } - trace_id = normalized.get("trace_id") - if trace_id and self.project: - entry["trace"] = f"projects/{self.project}/traces/{trace_id}" + # Stamp the event time when the payload carries one; async + # emission means the server's receive time can lag the event. + timestamp = rfc3339_timestamp(normalized.get("created_at")) + if timestamp: + entry["timestamp"] = timestamp + trace = trace_resource_name(self.project, normalized.get("trace_id")) + if trace: + entry["trace"] = trace span_id = normalized.get("span_id") if span_id: entry["spanId"] = str(span_id) return entry def _write(self, entries: list[dict[str, Any]]) -> None: - api = self.client.logging_api - gapic_api = getattr(api, "_gapic_api", None) - if gapic_api is not None: - try: - from google.cloud.logging_v2._gapic import ( - _log_entry_mapping_to_pb, - ) - from google.cloud.logging_v2.types import ( - WriteLogEntriesRequest, - ) - except ImportError: # pragma: no cover - exotic installs only - gapic_api = None - else: - request = WriteLogEntriesRequest( - entries=[ - _log_entry_mapping_to_pb(entry) for entry in entries - ], - partial_success=True, - ) - gapic_api.write_log_entries( - request=request, - retry=None, - timeout=self.timeout_seconds, - ) - return - # Non-gapic transports expose only the unbounded call; keep the - # destination functional there rather than dropping logs. - api.write_entries(entries, partial_success=True) + if self._gapic_api is not None and self._gapic_tools is not None: + mapping_to_pb, request_class = self._gapic_tools + request = request_class( + entries=[mapping_to_pb(entry) for entry in entries], + partial_success=True, + ) + self._gapic_api.write_log_entries( + request=request, + retry=self._retry, + timeout=self.timeout_seconds, + ) + return + self.client.logging_api.write_entries(entries, partial_success=True) def _resource_dict(resource: Any) -> Any: diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index fa5a269..07d3234 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -10,11 +10,12 @@ 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. +# A destination that fails this many consecutive emits is disabled until a +# manager-level restart rebuilds destinations. In sync mode the emits are on +# the caller's (request) path; in async mode the same limit applies to the +# background emitter's consecutive batch failures, and a tripped emitter +# raises so this breaker fires through the same path. Either way, a +# persistently failing sink must not keep charging the host service. DESTINATION_FAILURE_LIMIT = 3 @@ -36,6 +37,9 @@ def __init__( self._consecutive_failures: dict[int, int] = {} def configure(self) -> None: + # Close destinations from a previous configure() so their worker + # threads, clients, and atexit hooks are released, not orphaned. + self._close_destinations(self.destinations) failures: list[tuple[str, BaseException]] = [] destinations: list[LogDestination] = [] for destination_name in self.config.log_destinations or ("stdout",): @@ -108,6 +112,10 @@ def _ensure_destinations(self) -> list[LogDestination]: return self.destinations def _disable_destination(self, destination: LogDestination) -> None: + # Failure reporting can re-enter emit() and reach the limit again + # before the outer call disables; make disabling idempotent. + if destination not in self.destinations: + return self.destinations = [ existing for existing in self.destinations @@ -127,24 +135,33 @@ def _disable_destination(self, destination: LogDestination) -> None: def _build_destination(self, destination_name: str) -> LogDestination: normalized = destination_name.strip().lower().replace("-", "_") + # Stdout is the fallback sink and stays synchronous; every other + # destination goes through the async-wrapping choke point below. if normalized == "stdout": return self._stdout_destination() + return self._maybe_background( + self._build_network_destination(normalized, destination_name) + ) + + def _build_network_destination( + self, + normalized: str, + destination_name: str, + ) -> LogDestination: if normalized in {"google", "google_cloud", "google_cloud_logging"}: - return self._maybe_background( - GoogleCloudLoggingDestination( - project=self.config.google_cloud_project, - log_name=self.config.google_cloud_log_name, - timeout_seconds=self.config.google_log_timeout_seconds, - ) + return GoogleCloudLoggingDestination( + project=self.config.google_cloud_project, + log_name=self.config.google_cloud_log_name, + timeout_seconds=self.config.google_log_timeout_seconds, + on_failure=self.on_failure, ) raise ValueError( f"Unknown observability log destination: {destination_name}" ) def _maybe_background(self, destination: LogDestination) -> LogDestination: - # Stdout is the fallback sink and stays synchronous; it must keep - # working when everything else (including threads) is broken. - if self.config.log_emit_mode != "async": + emit_mode = (self.config.log_emit_mode or "").strip().lower() + if emit_mode != "async": return destination return BackgroundEmitDestination( destination, @@ -171,15 +188,29 @@ def flush(self, deadline_seconds: float | None = None) -> None: ) def restart(self) -> None: - for destination in list(self.destinations): - restart = getattr(destination, "restart", None) - if not callable(restart): + """Rebuild destinations from config. Restart must revive + destinations the breaker disabled and replace clients whose + connections did not survive a fork or memory-snapshot restore, + so it reconfigures from scratch rather than poking survivors.""" + self._consecutive_failures.clear() + self.configured = False + try: + self.configure() + except BaseException as exc: + self.destinations = [self._stdout_destination()] + self.configured = True + self.on_failure("logging.destination_restart", exc) + + def _close_destinations(self, destinations: list[LogDestination]) -> None: + for destination in destinations: + close = getattr(destination, "close", None) + if not callable(close): continue try: - restart() + close() except BaseException as exc: self.on_failure( - "logging.destination_restart", + "logging.destination_close", exc, destination=getattr(destination, "name", None), ) diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 11d14da..93b5a05 100644 --- a/policyengine_observability/runtime.py +++ b/policyengine_observability/runtime.py @@ -115,7 +115,7 @@ def __init__( self.failover_events = _NoOpInstrument() self.active_requests = _NoOpInstrument() self._httpx_instrumented = False - self._emitting_internal_error = False + self._internal_error_local = threading.local() self.log_destination_manager = LogDestinationManager( config=config, loggers={ @@ -128,6 +128,17 @@ def __init__( on_failure=self._handle_destination_failure, ) + @property + def _emitting_internal_error(self) -> bool: + # Thread-local: the async log worker reports failures concurrently + # with request threads, and a shared flag would misroute one + # thread's report to stderr because another thread is mid-report. + return getattr(self._internal_error_local, "value", False) + + @_emitting_internal_error.setter + def _emitting_internal_error(self, value: bool) -> None: + self._internal_error_local.value = value + @classmethod def disabled(cls) -> ObservabilityRuntime: return cls(ObservabilityConfig(enabled=False)) diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 639a400..c4fbdd3 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -113,8 +113,9 @@ def test_google_destination_writes_bounded_gapic_entry(monkeypatch) -> None: assert client.log_names == ["policyengine-observability"] assert client.logging_api.calls == [] request, retry, timeout = client.logging_api._gapic_api.calls[0] - assert retry is None - assert timeout == 2.0 + # Transient errors retry only inside the bounded write budget. + assert retry is not None + assert timeout == 5.0 assert request.partial_success is True (entry,) = request.entries from google.logging.type.log_severity_pb2 import LogSeverity @@ -458,13 +459,6 @@ def emit_batch(self, records) -> None: self.batches.append(list(records)) -class FailingDestination: - name = "failing" - - def emit(self, payload, *, log_type, severity) -> None: - raise RuntimeError("sink down") - - def _background(wrapped, **kwargs): from policyengine_observability.destinations.background import ( BackgroundEmitDestination, @@ -506,7 +500,7 @@ def test_background_drain_prefers_emit_batch(monkeypatch) -> None: for index in range(3): destination.emit({"event": index}, log_type="event", severity="INFO") - while destination._drain_once(): + while destination._drain_once() != "empty": pass assert [len(batch) for batch in wrapped.batches] == [2, 1] @@ -520,13 +514,13 @@ def test_background_drain_falls_back_to_single_emits(monkeypatch) -> None: _inline(destination, monkeypatch) destination.emit({"event": "a"}, log_type="event", severity="INFO") - while destination._drain_once(): + while destination._drain_once() != "empty": pass assert wrapped.payloads == [{"event": "a"}] -def test_background_overflow_drops_newest_and_reports(monkeypatch) -> None: +def test_background_overflow_drops_oldest_and_reports(monkeypatch) -> None: wrapped = BatchRecordingDestination() destination, failures = _background(wrapped, queue_size=2) _inline(destination, monkeypatch) @@ -535,7 +529,7 @@ def test_background_overflow_drops_newest_and_reports(monkeypatch) -> None: destination.emit({"event": index}, log_type="event", severity="INFO") kept = [payload["event"] for payload, _, _ in destination._buffer] - assert kept == [0, 1] + assert kept == [2, 3] overflow = [ fields for operation, fields in failures @@ -552,15 +546,14 @@ def test_background_trips_after_consecutive_batch_failures( import pytest destination, failures = _background( - FailingDestination(), batch_size=1, failure_limit=3 + FlakyDestination(), batch_size=1, failure_limit=3 ) _inline(destination, monkeypatch) for index in range(5): destination.emit({"event": index}, log_type="event", severity="INFO") - drained = True - while drained: - drained = destination._drain_once() + while not destination._tripped.is_set(): + destination._drain_once() counts = [ fields["consecutive_failures"] @@ -581,7 +574,7 @@ def test_background_trip_flows_through_manager_breaker(monkeypatch) -> None: ) destination, _failures = _background( - FailingDestination(), batch_size=1, failure_limit=1 + FlakyDestination(), batch_size=1, failure_limit=1 ) _inline(destination, monkeypatch) manager, manager_failures = _manager([destination]) @@ -620,7 +613,7 @@ def test_background_flush_reports_undelivered_remainder( monkeypatch, ) -> None: destination, failures = _background( - FailingDestination(), batch_size=1, failure_limit=1 + FlakyDestination(), batch_size=1, failure_limit=1 ) _inline(destination, monkeypatch) @@ -640,7 +633,7 @@ def test_background_flush_reports_undelivered_remainder( def test_background_restart_clears_trip_and_buffer(monkeypatch) -> None: wrapped = BatchRecordingDestination() destination, _failures = _background( - FailingDestination(), batch_size=1, failure_limit=1 + FlakyDestination(), batch_size=1, failure_limit=1 ) _inline(destination, monkeypatch) destination.emit({"event": "x"}, log_type="event", severity="INFO") @@ -667,11 +660,12 @@ def test_background_worker_delivers_end_to_end() -> None: for index in range(3): destination.emit({"event": index}, log_type="event", severity="INFO") destination.flush(1.0) + assert destination._atexit_registered is True destination.close() assert sum(len(batch) for batch in wrapped.batches) == 3 assert failures == [] - assert destination._atexit_registered is True + assert destination._atexit_registered is False def test_background_worker_restarts_after_fork(monkeypatch) -> None: @@ -690,6 +684,8 @@ def test_background_worker_restarts_after_fork(monkeypatch) -> None: destination.emit({"event": "b"}, log_type="event", severity="INFO") assert destination._worker is not first_worker + first_worker.join(timeout=2.0) + assert not first_worker.is_alive() destination.flush(1.0) destination.close() @@ -797,19 +793,182 @@ def restart(self) -> None: self.restarts += 1 -def test_manager_flush_and_restart_reach_capable_destinations() -> None: +def test_manager_flush_reaches_capable_destinations() -> None: flushable = FlushRecordingDestination() plain = RecordingDestination() manager, failures = _manager([flushable, plain]) manager.flush(1.5) - manager.restart() assert flushable.flushes == [1.5] - assert flushable.restarts == 1 assert failures == [] +def test_manager_restart_rebuilds_destinations_from_config() -> None: + from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + ) + + stale = FlushRecordingDestination() + manager, _failures = _manager([stale]) + manager._consecutive_failures[id(stale)] = 2 + + manager.restart() + + assert stale not in manager.destinations + assert any( + isinstance(destination, StdoutJsonDestination) + for destination in manager.destinations + ) + assert manager._consecutive_failures == {} + + +def test_manager_restart_revives_a_disabled_destination(monkeypatch) -> None: + """The Modal snapshot-restore path: a destination tripped and + disabled before the snapshot must come back after restart, with a + fresh client.""" + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations import manager as manager_mod + from policyengine_observability.destinations.manager import ( + DESTINATION_FAILURE_LIMIT, + ) + + built = [] + + class StubGoogleRebuild: + name = "google_cloud_logging" + + def __init__(self, **kwargs) -> None: + built.append(self) + + def emit(self, payload, *, log_type, severity) -> None: + raise RuntimeError("sink down") + + monkeypatch.setattr( + manager_mod, "GoogleCloudLoggingDestination", StubGoogleRebuild + ) + manager = _config_manager( + ObservabilityConfig(log_destinations=("google_cloud_logging",)) + ) + manager.configure() + for _ in range(DESTINATION_FAILURE_LIMIT): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + assert built[0] not in manager.destinations + + manager.restart() + + assert len(built) == 2 + assert built[1] in manager.destinations + + +def test_manager_configure_closes_replaced_destinations() -> None: + class ClosableDestination(FlushRecordingDestination): + def __init__(self) -> None: + super().__init__() + self.closed = 0 + + def close(self) -> None: + self.closed += 1 + + stale = ClosableDestination() + manager, _failures = _manager([stale]) + + manager.configure() + + assert stale.closed == 1 + assert stale not in manager.destinations + + +def test_background_flush_waits_for_in_flight_batch(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped) + _inline(destination, monkeypatch) + + with destination._lock: + destination._in_flight = 1 + destination.flush(0.05) + + incomplete = [ + fields + for operation, fields in failures + if operation == "logging.destination_flush_incomplete" + ] + assert len(incomplete) == 1 + assert incomplete[0]["remaining"] == 1 + with destination._lock: + destination._in_flight = 0 + + +def test_google_destination_reports_unbounded_transport(monkeypatch) -> None: + failures = [] + monkeypatch.setattr( + google_cloud_logging, + "load_google_credentials", + lambda *, prefer_workload_identity: None, + ) + monkeypatch.setattr( + google_cloud_logging, + "configure_google_application_credentials", + lambda: None, + ) + GoogleCloudLoggingDestination( + project=None, + log_name="policyengine-observability", + client_factory=lambda _project, _credentials: FakeClient(gapic=False), + on_failure=lambda operation, exc, **fields: failures.append( + (operation, fields) + ), + ) + + assert failures[0][0] == "logging.destination_unbounded_transport" + + +def test_google_destination_stamps_event_timestamp(monkeypatch) -> None: + client = FakeClient(gapic=False) + destination = _destination(monkeypatch, client) + + destination.emit( + {"created_at": "2026-07-08T00:00:00+00:00", "event": "x"}, + log_type="event", + severity="INFO", + ) + destination.emit( + {"created_at": "not-a-timestamp", "event": "y"}, + log_type="event", + severity="INFO", + ) + + (stamped,), _ = client.logging_api.calls[0] + assert stamped["timestamp"] == "2026-07-08T00:00:00+00:00" + (unstamped,), _ = client.logging_api.calls[1] + assert "timestamp" not in unstamped + + +def test_internal_error_flag_is_thread_local() -> None: + import threading + + from policyengine_observability import ( + ObservabilityConfig, + ObservabilityRuntime, + ) + + runtime = ObservabilityRuntime( + ObservabilityConfig(service_name="svc", otel_enabled=False) + ) + runtime._emitting_internal_error = True + seen_in_thread = [] + + def read_flag() -> None: + seen_in_thread.append(runtime._emitting_internal_error) + + thread = threading.Thread(target=read_flag) + thread.start() + thread.join() + + assert runtime._emitting_internal_error is True + assert seen_in_thread == [False] + + def test_manager_flush_reports_but_survives_failures() -> None: class ExplodingFlush(FlushRecordingDestination): def flush(self, deadline_seconds=None) -> None: diff --git a/tests/test_google_credentials.py b/tests/test_google_credentials.py index c112ab0..c787767 100644 --- a/tests/test_google_credentials.py +++ b/tests/test_google_credentials.py @@ -277,11 +277,16 @@ def test_google_destination_bootstraps_application_credentials( lambda *, prefer_workload_identity: None, ) + class FakeLogger: + def __init__(self, log_name): + self.full_name = f"projects/test-project/logs/{log_name}" + self.default_resource = {"type": "global", "labels": {}} + class FakeClient: project = "test-project" def logger(self, log_name): - return log_name + return FakeLogger(log_name) destination = google_cloud_logging.GoogleCloudLoggingDestination( project=None, @@ -291,7 +296,9 @@ def logger(self, log_name): assert calls == ["configured"] assert destination.project == "test-project" - assert destination.logger == "policyengine-observability" + assert destination.logger.full_name == ( + "projects/test-project/logs/policyengine-observability" + ) def test_google_destination_passes_loaded_credentials(monkeypatch) -> None: @@ -303,11 +310,16 @@ def test_google_destination_passes_loaded_credentials(monkeypatch) -> None: calls = [] + class FakeLogger: + def __init__(self, log_name): + self.full_name = f"projects/test-project/logs/{log_name}" + self.default_resource = {"type": "global", "labels": {}} + class FakeClient: project = "test-project" def logger(self, log_name): - return log_name + return FakeLogger(log_name) destination = google_cloud_logging.GoogleCloudLoggingDestination( project="central-project", @@ -318,4 +330,6 @@ def logger(self, log_name): ) assert calls == [("central-project", "wif-credentials")] - assert destination.logger == "policyengine-observability" + assert destination.logger.full_name == ( + "projects/test-project/logs/policyengine-observability" + ) From 0253238cabeabe25ed16bbabf214c4670f08b27e Mon Sep 17 00:00:00 2001 From: Anthony Volk Date: Wed, 8 Jul 2026 20:08:38 +0200 Subject: [PATCH 7/7] Address second-round review findings Correctness: records are normalized and timestamped at enqueue and carried as (payload, log_type, severity, enqueue-time) tuples so delayed batches keep their event time and the removed created_at backdating cannot recur; drop-oldest accounting moved before the deque eviction it counts; failure counters mutate under the lock with a stale-generation skip so a superseded worker cannot trip the fresh one; the emitter trips before its report fires and includes the stranded record count; flush attempts one drain before the first deadline check so flush(0) works against a healthy sink, exits clean when drained and idle, and waits for in-flight batches; close() is terminal (no worker respawn, no atexit re-register) and reports the count of records it discards; the worker survives BaseException with a crash report; fork/snapshot handling registers instances in a WeakSet wired to os.register_at_fork, signals the superseded generation, and reports inherited records dropped in the child. Manager: configure/restart/disable serialize on a reentrant lifecycle lock; failure reports are deferred until the new destination set is live so a report's own emission cannot re-enter configuration; replaced destinations close only after their successors are installed; disabling is exactly-once under reentrant reporting and closes the destination; one absolute deadline is shared across per-destination flushes; config warnings fire for unknown emit modes and the double-ingestion stdout_format=google + google-destination combination. Google destination: transient errors retry inside the bounded budget via a single closure resolved at construction (worst case ~2x budget, documented); a one-time report fires when only the unbounded non-gapic transport is available; close() releases the client transport. Config knobs normalize in __post_init__ regardless of construction path. The runtime shutdown flushes log destinations inside the timed closure so the shutdown timeout bounds it. Stdout google mode no longer emits a time key: synchronous emission makes server receive time correct. Docs and changelog updated to match; 169 tests, branch coverage 91%. --- README.md | 29 +- changelog.d/22.added.md | 2 +- policyengine_observability/config.py | 22 +- .../destinations/background.py | 281 +++++++++++----- .../destinations/base.py | 15 - .../destinations/google_cloud_logging.py | 96 ++++-- .../destinations/manager.py | 240 ++++++++++---- .../destinations/stdout.py | 28 +- policyengine_observability/runtime.py | 8 +- tests/test_destinations.py | 309 +++++++++++++++--- 10 files changed, 743 insertions(+), 287 deletions(-) diff --git a/README.md b/README.md index 1dd8b7a..82a68e9 100644 --- a/README.md +++ b/README.md @@ -107,7 +107,9 @@ For the fixed PolicyEngine Google Cloud destination, see Writes to Google Cloud Logging carry an explicit per-call timeout, with transient errors retried only inside that budget, so a degraded Logging -API cannot stall the caller while brief blips are still absorbed: +API cannot stall the caller while brief blips are still absorbed. (The +retry machinery hands the final attempt a fresh per-attempt timeout, so +the worst-case wall time is about twice the configured budget.) ```bash OBSERVABILITY_GOOGLE_LOG_TIMEOUT_SECONDS=5.0 @@ -117,7 +119,10 @@ By default, log emission is synchronous on the caller's thread. Setting the emit mode to `async` installs the standard accept/emit split: the logging call only appends the record to a bounded in-memory buffer and returns immediately, and a background worker thread drains the buffer in -batches to the network destination: +batches to the network destination. The worker naps for the batch +latency after waking so bursts coalesce into fewer, larger writes, and +each record is written with its enqueue time as the entry timestamp so +delayed batches keep their event time: ```bash OBSERVABILITY_LOG_EMIT_MODE=async @@ -133,13 +138,15 @@ records — and drops are counted and reported through the internal-error channel. After three consecutive failed batches (with backoff between attempts) the emitter trips; subsequent log calls surface the failure to the manager's circuit breaker, which disables the destination and falls -back to stdout. `restart_observability()` rebuilds destinations from -config, reviving a disabled destination with a fresh client. Buffered -records are flushed at interpreter exit and by `runtime.shutdown()`; -`flush_observability(deadline_seconds)` flushes on demand and waits for -any in-flight batch. A hard kill loses whatever was still buffered. -Stdout destinations are never wrapped: the fallback sink stays -synchronous and dependency-free. +back to stdout. Recovery is deliberately manual: a disabled destination +stays disabled until `restart_observability()` rebuilds destinations +from config with fresh clients (automatic half-open probing is future +work). Buffered records are flushed at interpreter exit and by +`runtime.shutdown()`; `flush_observability(deadline_seconds)` flushes on +demand and waits for any in-flight batch. Closing with records still +buffered reports the count of records lost. A hard kill loses whatever +was still buffered. Stdout destinations are never wrapped: the fallback +sink stays synchronous and dependency-free. Processes that fork or restore from memory snapshots (for example Modal Functions with memory snapshots enabled) do not preserve threads. The @@ -149,8 +156,8 @@ their post-restore hook to clear buffered state and revive the worker. On platforms whose logging agent collects stdout (Cloud Run, GKE), the agent-native stdout format emits JSON lines carrying the special keys the -agent promotes to first-class LogEntry fields (severity, time, trace, -span, labels), giving full-fidelity Cloud Logging ingestion with no +agent promotes to first-class LogEntry fields (severity, trace, span, +labels), giving full-fidelity Cloud Logging ingestion with no in-process network emission: ```bash diff --git a/changelog.d/22.added.md b/changelog.d/22.added.md index e8a4aa3..da5b0ce 100644 --- a/changelog.d/22.added.md +++ b/changelog.d/22.added.md @@ -1 +1 @@ -Added an agent-native stdout format (`OBSERVABILITY_STDOUT_FORMAT=google`): JSON lines carry the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, time, trace, span, labels), enabling full-fidelity Cloud Logging ingestion from stdout with no in-process network emission. Added a background log emitter (`OBSERVABILITY_LOG_EMIT_MODE=async`): log calls enqueue onto a bounded drop-oldest buffer and a worker thread batches writes off the request path, with flush-on-shutdown that waits for in-flight batches, and `restart_observability()` rebuilding destinations from config for forked or snapshot-restored processes — including destinations the circuit breaker disabled. +Added an agent-native stdout format (`OBSERVABILITY_STDOUT_FORMAT=google`): JSON lines carry the special keys the Cloud Run/GKE logging agent promotes to first-class LogEntry fields (severity, trace, span, labels), enabling full-fidelity Cloud Logging ingestion from stdout with no in-process network emission. Added a background log emitter (`OBSERVABILITY_LOG_EMIT_MODE=async`): log calls enqueue onto a bounded drop-oldest buffer and a worker thread batches writes off the request path, with flush-on-shutdown that waits for in-flight batches, and `restart_observability()` rebuilding destinations from config for forked or snapshot-restored processes — including destinations the circuit breaker disabled. diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index 9ba5498..7139054 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -111,6 +111,20 @@ class ObservabilityConfig: log_batch_latency_seconds: float = DEFAULT_LOG_BATCH_LATENCY_SECONDS log_flush_deadline_seconds: float = DEFAULT_LOG_FLUSH_DEADLINE_SECONDS + def __post_init__(self) -> None: + # Normalize string knobs regardless of construction path, so + # programmatic configs behave like env-driven ones. + object.__setattr__( + self, + "stdout_format", + (self.stdout_format or "plain").strip().lower(), + ) + object.__setattr__( + self, + "log_emit_mode", + (self.log_emit_mode or "sync").strip().lower(), + ) + @classmethod def from_env( cls, @@ -195,14 +209,10 @@ def from_env( ), stdout_format=( os.getenv("OBSERVABILITY_STDOUT_FORMAT") or cls.stdout_format - ) - .strip() - .lower(), + ), log_emit_mode=( os.getenv("OBSERVABILITY_LOG_EMIT_MODE") or cls.log_emit_mode - ) - .strip() - .lower(), + ), log_queue_size=int_from_env( "OBSERVABILITY_LOG_QUEUE_SIZE", cls.log_queue_size, diff --git a/policyengine_observability/destinations/background.py b/policyengine_observability/destinations/background.py index e850362..0b812e8 100644 --- a/policyengine_observability/destinations/background.py +++ b/policyengine_observability/destinations/background.py @@ -4,8 +4,10 @@ import os import threading import time +import weakref from collections import deque from collections.abc import Callable +from datetime import UTC, datetime from typing import Any from ..config import ( @@ -27,28 +29,39 @@ DRAIN_DELIVERED = "delivered" DRAIN_FAILED = "failed" -_LogRecord = tuple[dict[str, Any], str, str] +# (normalized payload, log_type, severity, enqueued-at RFC3339 timestamp) +_LogRecord = tuple[dict[str, Any], str, str, str] + +# Fork handling: threads (including workers holding locks) do not survive +# fork, so every instance must rebuild its synchronization primitives and +# drop the inherited buffer copy (the parent's worker will deliver it) in +# the child while it is still single-threaded. +_INSTANCES: weakref.WeakSet[BackgroundEmitDestination] = weakref.WeakSet() + + +def _reset_instances_after_fork() -> None: # pragma: no cover - fork hook + for destination in list(_INSTANCES): + destination._reset_after_fork() + + +if hasattr(os, "register_at_fork"): + os.register_at_fork(after_in_child=_reset_instances_after_fork) class BackgroundEmitDestination: """Decouples log acceptance from emission. ``emit()`` normalizes the payload (snapshotting it against caller - mutation), appends it to a bounded in-memory buffer, and returns - immediately — it never blocks on the network. When the buffer is - full the oldest record is dropped (counted and reported, throttled), - keeping the freshest records: during a sink outage the most - diagnostic records are the recent ones. A daemon worker drains the - buffer in batches to the wrapped destination (preferring its - ``emit_batch``), backing off between failed batches; after - ``failure_limit`` consecutive failed batches the wrapper trips, and - subsequent ``emit()`` calls raise so the manager's circuit breaker - disables it through the same path as a synchronous destination. - - The worker starts lazily on first emit and is pid-aware, so a forked - process starts a fresh worker automatically; ``restart()`` - additionally clears buffered and trip state. Recovery of a tripped - and disabled destination happens at the manager level + mutation), stamps the enqueue time, appends the record to a bounded + in-memory buffer, and returns immediately — it never blocks on the + network. When the buffer is full the oldest record is dropped + (counted and reported, throttled), keeping the freshest records. A + daemon worker wakes when the buffer becomes non-empty, naps briefly + to coalesce a batch, and drains to the wrapped destination + (preferring its ``emit_batch``), backing off between failed batches; + after ``failure_limit`` consecutive failed batches the wrapper trips + and subsequent ``emit()`` calls raise so the manager's circuit + breaker disables it. Recovery happens at the manager level (``LogDestinationManager.restart`` rebuilds destinations). """ @@ -75,20 +88,24 @@ def __init__( self.flush_deadline_seconds = max(0.0, flush_deadline_seconds) self.failure_limit = failure_limit self.failure_backoff_seconds = max(0.0, failure_backoff_seconds) + self._wrapped_emit_batch = getattr(wrapped, "emit_batch", None) + if not callable(self._wrapped_emit_batch): + self._wrapped_emit_batch = None self._buffer: deque[_LogRecord] = deque(maxlen=self.queue_size) self._lock = threading.Lock() self._start_lock = threading.Lock() - # One wake event for the object's lifetime; only the closed event - # is generation-scoped, so emitters can never signal a stale one. self._wake = threading.Event() self._tripped = threading.Event() + self._stopped = threading.Event() self._closed = threading.Event() self._worker: threading.Thread | None = None self._pid: int | None = None self._consecutive_failures = 0 self._dropped = 0 + self._fork_dropped = 0 self._in_flight = 0 self._atexit_registered = False + _INSTANCES.add(self) def emit( self, @@ -97,6 +114,10 @@ def emit( log_type: str, severity: str, ) -> None: + if self._stopped.is_set(): + # Only reachable through a stale reference during a manager + # reconfigure swap; the manager no longer routes here. + return if self._tripped.is_set(): raise RuntimeError( f"Background emitter for {self.name} is tripped after " @@ -105,9 +126,17 @@ def emit( self._ensure_worker() # Snapshot now: the caller may keep mutating nested structures # after this returns, and serialization happens on the worker. - record = (normalize_payload(payload), log_type, severity) + # The timestamp preserves event time against emission delay. + record = ( + normalize_payload(payload), + log_type, + severity, + datetime.now(UTC).isoformat(), + ) dropped_total: int | None = None + fork_dropped: int | None = None with self._lock: + was_empty = not self._buffer if len(self._buffer) == self.queue_size: self._dropped += 1 if ( @@ -116,6 +145,9 @@ def emit( ): dropped_total = self._dropped self._buffer.append(record) + if self._fork_dropped: + fork_dropped = self._fork_dropped + self._fork_dropped = 0 if dropped_total is not None: self.on_failure( "logging.destination_queue_overflow", @@ -126,25 +158,41 @@ def emit( destination=self.name, dropped_total=dropped_total, ) - self._wake.set() + if fork_dropped is not None: + self.on_failure( + "logging.destination_fork_buffer_dropped", + RuntimeError( + "Dropped log records inherited across a fork; the " + "parent process delivers its own copy." + ), + destination=self.name, + dropped_total=fork_dropped, + ) + if was_empty: + self._wake.set() def flush(self, deadline_seconds: float | None = None) -> None: - """Drain the buffer from the caller's thread, bounded by a - deadline; waits for a worker-held in-flight batch and reports - any undelivered remainder.""" + """Drain the buffer from the caller's thread. Bounded by a soft + deadline (a blocking write in progress can overrun it by one + write budget); waits for a worker-held in-flight batch and + reports any undelivered remainder.""" if deadline_seconds is None: deadline_seconds = self.flush_deadline_seconds - deadline = time.monotonic() + deadline_seconds - while not self._tripped.is_set() and time.monotonic() < deadline: - outcome = self._drain_once() - if outcome != DRAIN_EMPTY: - continue - with self._lock: - idle = not self._buffer and self._in_flight == 0 - if idle: + deadline = time.monotonic() + max(0.0, deadline_seconds) + while not self._tripped.is_set(): + outcome = self._drain_once(self._closed) + if outcome == DRAIN_EMPTY: + with self._lock: + idle = not self._buffer and self._in_flight == 0 + if idle: + # Fully drained: nothing to report, even if a record + # arrives after this instant. + return + if time.monotonic() >= deadline: break - # A batch is in flight on the worker; give it a moment. - time.sleep(MIN_BATCH_LATENCY_SECONDS) + if outcome != DRAIN_DELIVERED: + # Empty-but-in-flight or a failed batch: brief pause. + time.sleep(MIN_BATCH_LATENCY_SECONDS) with self._lock: remaining = len(self._buffer) + self._in_flight if remaining: @@ -159,48 +207,75 @@ def flush(self, deadline_seconds: float | None = None) -> None: ) def close(self) -> None: - """Stop the worker without flushing.""" + """Stop permanently: no further records are accepted, the worker + exits, and undelivered records are reported and discarded.""" + self._stopped.set() self._closed.set() self._wake.set() + with self._lock: + remaining = len(self._buffer) + self._in_flight + self._buffer.clear() + if remaining: + self.on_failure( + "logging.destination_closed_pending", + RuntimeError( + "Background emitter closed with undelivered log records." + ), + destination=self.name, + remaining=remaining, + ) if self._atexit_registered: atexit.unregister(self.flush) self._atexit_registered = False + wrapped_close = getattr(self.wrapped, "close", None) + if callable(wrapped_close): + try: + wrapped_close() + except Exception as exc: + self.on_failure( + "logging.destination_close", + exc, + destination=self.name, + ) - def restart(self) -> None: - """Reset local state: drop buffered records, clear trip state, - start a fresh worker on the next emit. A destination the manager - already disabled cannot be revived here — use the manager-level - restart, which rebuilds destinations.""" - with self._start_lock: - self._closed.set() - self._wake.set() - self._worker = None - self._pid = None - with self._lock: - self._buffer.clear() - self._dropped = 0 - self._consecutive_failures = 0 - self._tripped.clear() + def _reset_after_fork(self) -> None: + """Runs in a forked child while it is single-threaded: parent + threads (possibly holding our locks) do not exist here, and the + buffer is a copy the parent will deliver itself.""" + inherited = len(self._buffer) + # Signal the superseded generation first: after a real fork no + # thread is listening (harmless); on the belt-path pid check a + # live stale worker exits instead of leaking. + self._closed.set() + self._wake.set() + self._lock = threading.Lock() + self._start_lock = threading.Lock() + self._wake = threading.Event() + self._closed = threading.Event() + self._buffer = deque(maxlen=self.queue_size) + self._in_flight = 0 + self._consecutive_failures = 0 + self._worker = None + self._pid = None + self._fork_dropped += inherited def _ensure_worker(self) -> None: + if self._pid is not None and self._pid != os.getpid(): + # Belt for exotic fork paths that bypassed the fork hook. + self._reset_after_fork() worker = self._worker - if ( - worker is not None - and worker.is_alive() - and self._pid == os.getpid() - ): + if worker is not None and worker.is_alive(): return with self._start_lock: worker = self._worker - if ( - worker is not None - and worker.is_alive() - and self._pid == os.getpid() - ): + if worker is not None and worker.is_alive(): + return + if self._stopped.is_set(): return # Signal any superseded-but-alive worker before replacing its # generation event, so it exits instead of leaking. self._closed.set() + self._wake.set() closed = threading.Event() self._closed = closed thread = threading.Thread( @@ -217,18 +292,33 @@ def _ensure_worker(self) -> None: self._atexit_registered = True def _run(self, closed: threading.Event) -> None: - while not closed.is_set(): - self._wake.wait(timeout=self.batch_latency_seconds) - self._wake.clear() + try: while not closed.is_set(): - outcome = self._drain_once() - if outcome == DRAIN_EMPTY: + # Idle costs nothing: emit() wakes us on the buffer's + # empty-to-non-empty transition. + self._wake.wait() + self._wake.clear() + if closed.is_set(): break - if outcome == DRAIN_FAILED: - # Back off before hammering a failing sink again. - closed.wait(timeout=self.failure_backoff_seconds) + # Nap briefly so nearby records coalesce into one batch. + closed.wait(timeout=self.batch_latency_seconds) + while not closed.is_set(): + outcome = self._drain_once(closed) + if outcome == DRAIN_EMPTY: + break + if outcome == DRAIN_FAILED: + closed.wait(timeout=self.failure_backoff_seconds) + except BaseException as exc: # worker must never die silently + try: + self.on_failure( + "logging.destination_worker_crashed", + exc, + destination=self.name, + ) + except Exception: # pragma: no cover - reporting best-effort + pass - def _drain_once(self) -> str: + def _drain_once(self, closed: threading.Event | None = None) -> str: """Write one batch; returns a DRAIN_* outcome.""" with self._lock: if not self._buffer: @@ -239,36 +329,63 @@ def _drain_once(self) -> str: ] self._in_flight += len(batch) try: - delivered = self._write_batch(batch) + delivered = self._write_batch(batch, closed) finally: with self._lock: self._in_flight -= len(batch) return DRAIN_DELIVERED if delivered else DRAIN_FAILED - def _write_batch(self, batch: list[_LogRecord]) -> bool: + def _write_batch( + self, + batch: list[_LogRecord], + closed: threading.Event | None, + ) -> bool: try: - emit_batch = getattr(self.wrapped, "emit_batch", None) - if callable(emit_batch): - emit_batch(batch) + if self._wrapped_emit_batch is not None: + self._wrapped_emit_batch(batch) else: - for payload, log_type, severity in batch: + for payload, log_type, severity, _timestamp in batch: self.wrapped.emit( payload, log_type=log_type, severity=severity, ) except Exception as exc: - self._consecutive_failures += 1 + stale = closed is not None and closed is not self._closed + if stale: + # A superseded generation's late failure must not poison + # the current generation's breaker state. + return False + tripped_now = False + stranded = 0 + with self._lock: + self._consecutive_failures += 1 + failures = self._consecutive_failures + if ( + failures >= self.failure_limit + and not self._tripped.is_set() + ): + tripped_now = True + stranded = len(self._buffer) + if tripped_now: + # Trip BEFORE reporting so the failure report cannot be + # enqueued into this now-doomed buffer. + self._tripped.set() + self._closed.set() + self._wake.set() + fields: dict[str, Any] = { + "destination": self.name, + "consecutive_failures": failures, + "records_lost": len(batch), + } + if tripped_now: + fields["stranded"] = stranded self.on_failure( "logging.destination_emit_async", exc, - destination=self.name, - consecutive_failures=self._consecutive_failures, - records_lost=len(batch), + **fields, ) - if self._consecutive_failures >= self.failure_limit: - self._tripped.set() - self._closed.set() return False - self._consecutive_failures = 0 + with self._lock: + self._consecutive_failures = 0 return True diff --git a/policyengine_observability/destinations/base.py b/policyengine_observability/destinations/base.py index 1f43849..28cb3ce 100644 --- a/policyengine_observability/destinations/base.py +++ b/policyengine_observability/destinations/base.py @@ -1,7 +1,6 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from datetime import datetime from typing import Any, Protocol @@ -63,17 +62,3 @@ def trace_resource_name( if not trace_id or not project: return None return f"projects/{project}/traces/{trace_id}" - - -def rfc3339_timestamp(value: Any) -> str | None: - """Return an RFC3339 timestamp for a payload-supplied value, or None - when the value cannot be interpreted as an aware datetime.""" - if not isinstance(value, str): - return None - try: - parsed = datetime.fromisoformat(value) - except ValueError: - return None - if parsed.tzinfo is None: - return None - return parsed.isoformat() diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 62d18b0..99228cf 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -14,7 +14,6 @@ from .base import ( bounded_labels, normalize_payload, - rfc3339_timestamp, trace_resource_name, ) @@ -85,14 +84,12 @@ def client_factory( self.logger = self.client.logger(log_name) self._full_name = self.logger.full_name self._resource = _resource_dict(self.logger.default_resource) - self._gapic_api = None - self._gapic_tools: tuple[Any, Any] | None = None - self._retry: Any = None - self._resolve_transport(on_failure) + self._bounded_write: Callable[[list[dict[str, Any]]], None] | None + self._bounded_write = self._resolve_transport(on_failure) def _resolve_transport( self, on_failure: Callable[..., None] | None - ) -> None: + ) -> Callable[[list[dict[str, Any]]], None] | None: api = getattr(self.client, "logging_api", None) gapic_api = getattr(api, "_gapic_api", None) if gapic_api is not None: @@ -111,15 +108,10 @@ def _resolve_transport( except ImportError: # pragma: no cover - exotic installs only gapic_api = None else: - self._gapic_api = gapic_api - self._gapic_tools = ( - _log_entry_mapping_to_pb, - WriteLogEntriesRequest, - ) - # Retry transient errors, but only inside the overall - # write budget, so a Logging API blip does not surface as - # a destination failure while a real outage stays bounded. - self._retry = Retry( + # Retry transient errors inside the write budget. Note + # the worst case is ~2x the budget: api-core hands the + # final retry attempt a fresh per-attempt timeout. + retry = Retry( initial=0.1, maximum=1.0, multiplier=1.3, @@ -130,7 +122,24 @@ def _resolve_transport( api_exceptions.ServiceUnavailable, ), ) - if self._gapic_api is None and on_failure is not None: + timeout_seconds = self.timeout_seconds + + def bounded_write(entries: list[dict[str, Any]]) -> None: + request = WriteLogEntriesRequest( + entries=[ + _log_entry_mapping_to_pb(entry) + for entry in entries + ], + partial_success=True, + ) + gapic_api.write_log_entries( + request=request, + retry=retry, + timeout=timeout_seconds, + ) + + return bounded_write + if on_failure is not None: on_failure( "logging.destination_unbounded_transport", RuntimeError( @@ -140,6 +149,7 @@ def _resolve_transport( ), destination=self.name, ) + return None def emit( self, @@ -154,24 +164,49 @@ def emit( def emit_batch( self, - records: Sequence[tuple[dict[str, Any], str, str]], + records: Sequence[tuple[dict[str, Any], str, str, str | None]], ) -> None: - """Write ``(payload, log_type, severity)`` records in one call.""" + """Write ``(payload, log_type, severity, timestamp)`` records in + one bounded call. Payloads must already be normalized and the + timestamp (RFC3339 enqueue time, or None) preserves event time + against asynchronous emission delay — this is the background + emitter's contract.""" entries = [ - self._build_entry(payload, log_type=log_type, severity=severity) - for payload, log_type, severity in records + self._build_entry( + payload, + log_type=log_type, + severity=severity, + timestamp=timestamp, + pre_normalized=True, + ) + for payload, log_type, severity, timestamp in records ] if entries: self._write(entries) + def close(self) -> None: + """Release the underlying client transport when possible.""" + close = getattr(self.client, "close", None) + if callable(close): + close() + return + api = getattr(self.client, "logging_api", None) + gapic_api = getattr(api, "_gapic_api", None) + transport = getattr(gapic_api, "transport", None) + transport_close = getattr(transport, "close", None) + if callable(transport_close): + transport_close() + def _build_entry( self, payload: dict[str, Any], *, log_type: str, severity: str, + timestamp: str | None = None, + pre_normalized: bool = False, ) -> dict[str, Any]: - normalized = normalize_payload(payload) + normalized = payload if pre_normalized else normalize_payload(payload) entry: dict[str, Any] = { "logName": self._full_name, "resource": self._resource, @@ -179,9 +214,9 @@ def _build_entry( "severity": str(severity).upper(), "labels": bounded_labels(normalized, log_type=log_type), } - # Stamp the event time when the payload carries one; async - # emission means the server's receive time can lag the event. - timestamp = rfc3339_timestamp(normalized.get("created_at")) + # Synchronous writes rely on the server's receive time; the + # background emitter supplies its enqueue time instead, so + # delayed batches keep their event time. if timestamp: entry["timestamp"] = timestamp trace = trace_resource_name(self.project, normalized.get("trace_id")) @@ -193,17 +228,8 @@ def _build_entry( return entry def _write(self, entries: list[dict[str, Any]]) -> None: - if self._gapic_api is not None and self._gapic_tools is not None: - mapping_to_pb, request_class = self._gapic_tools - request = request_class( - entries=[mapping_to_pb(entry) for entry in entries], - partial_success=True, - ) - self._gapic_api.write_log_entries( - request=request, - retry=self._retry, - timeout=self.timeout_seconds, - ) + if self._bounded_write is not None: + self._bounded_write(entries) return self.client.logging_api.write_entries(entries, partial_success=True) diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 07d3234..f4480b9 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -1,6 +1,8 @@ from __future__ import annotations import logging +import threading +import time from collections.abc import Callable, Mapping from typing import Any @@ -18,6 +20,11 @@ # persistently failing sink must not keep charging the host service. DESTINATION_FAILURE_LIMIT = 3 +_VALID_EMIT_MODES = {"sync", "async"} +_GOOGLE_DESTINATION_NAMES = {"google", "google_cloud", "google_cloud_logging"} + +_Report = tuple[str, BaseException, dict[str, Any]] + class LogDestinationManager: def __init__( @@ -35,37 +42,102 @@ def __init__( self.destinations: list[LogDestination] = [] self.configured = False self._consecutive_failures: dict[int, int] = {} + # Serializes configure/restart/disable transitions. Reentrant so + # a failure report fired while holding it can safely re-enter. + self._lifecycle_lock = threading.RLock() def configure(self) -> None: - # Close destinations from a previous configure() so their worker - # threads, clients, and atexit hooks are released, not orphaned. - self._close_destinations(self.destinations) - failures: list[tuple[str, BaseException]] = [] + with self._lifecycle_lock: + reports = self._configure_locked() + # Reports fire only after the new destination set is live, so a + # report's own emission cannot re-enter configuration. + self._fire(reports) + + def _configure_locked(self) -> list[_Report]: + previous = list(self.destinations) + reports: list[_Report] = [] + + def deferred_on_failure( + operation: str, exc: BaseException, **fields: Any + ) -> None: + reports.append((operation, exc, fields)) + destinations: list[LogDestination] = [] for destination_name in self.config.log_destinations or ("stdout",): try: - destinations.append(self._build_destination(destination_name)) + destinations.append( + self._build_destination( + destination_name, deferred_on_failure + ) + ) except BaseException as exc: - failures.append((destination_name, exc)) + reports.append( + ( + "logging.destination_config", + exc, + {"destination": destination_name}, + ) + ) if not destinations: destinations.append(self._stdout_destination()) - failures.append( + reports.append( ( - "stdout_fallback", + "logging.destination_config", RuntimeError( "No configured observability log destination " "initialized; falling back to stdout." ), + {"destination": "stdout_fallback"}, ) ) + reports.extend(self._config_warnings()) + # The rebuilt objects get a fresh failure ledger; stale entries + # would otherwise leak and can collide via id() reuse. + self._consecutive_failures.clear() self.destinations = destinations self.configured = True - for destination_name, exc in failures: - self.on_failure( - "logging.destination_config", - exc, - destination=destination_name, + # Close replaced destinations only after the new set is live, so + # no emitter can reach a closed destination through the manager. + for destination in previous: + self._close_destination(destination, deferred_on_failure) + return reports + + def _config_warnings(self) -> list[_Report]: + warnings: list[_Report] = [] + emit_mode = (self.config.log_emit_mode or "").strip().lower() + if emit_mode not in _VALID_EMIT_MODES: + warnings.append( + ( + "logging.destination_config_warning", + ValueError( + "Unknown OBSERVABILITY_LOG_EMIT_MODE " + f"{self.config.log_emit_mode!r}; emission runs " + "synchronously." + ), + {}, + ) + ) + normalized_names = { + name.strip().lower().replace("-", "_") + for name in (self.config.log_destinations or ()) + } + if ( + (self.config.stdout_format or "").strip().lower() == "google" + and "stdout" in normalized_names + and normalized_names & _GOOGLE_DESTINATION_NAMES + ): + warnings.append( + ( + "logging.destination_config_warning", + ValueError( + "stdout_format=google together with a " + "google_cloud_logging destination ingests every " + "record into Cloud Logging twice." + ), + {}, + ) ) + return warnings def emit( self, @@ -101,6 +173,63 @@ def emit( for destination in tripped: self._disable_destination(destination) + def flush(self, deadline_seconds: float | None = None) -> None: + """Flush all capable destinations against one shared deadline.""" + deadline: float | None = None + if deadline_seconds is not None: + deadline = time.monotonic() + max(0.0, deadline_seconds) + for destination in list(self.destinations): + flush = getattr(destination, "flush", None) + if not callable(flush): + continue + remaining: float | None = None + if deadline is not None: + remaining = max(0.0, deadline - time.monotonic()) + try: + flush(remaining) + except Exception as exc: + self.on_failure( + "logging.destination_flush", + exc, + destination=getattr(destination, "name", None), + ) + + def restart(self) -> None: + """Rebuild destinations from config. Restart must revive + destinations the breaker disabled and replace clients whose + connections did not survive a fork or memory-snapshot restore, + so it reconfigures from scratch rather than poking survivors.""" + with self._lifecycle_lock: + self.configured = False + try: + reports = self._configure_locked() + except BaseException as exc: + self.destinations = [self._stdout_destination()] + self.configured = True + reports = [("logging.destination_restart", exc, {})] + self._fire(reports) + + def _fire(self, reports: list[_Report]) -> None: + for operation, exc, fields in reports: + self.on_failure(operation, exc, **fields) + + def _close_destination( + self, + destination: LogDestination, + report: Callable[..., None], + ) -> None: + close = getattr(destination, "close", None) + if not callable(close): + return + try: + close() + except Exception as exc: + report( + "logging.destination_close", + exc, + destination=getattr(destination, "name", None), + ) + def _ensure_destinations(self) -> list[LogDestination]: if not self.configured: try: @@ -112,16 +241,22 @@ def _ensure_destinations(self) -> list[LogDestination]: return self.destinations def _disable_destination(self, destination: LogDestination) -> None: - # Failure reporting can re-enter emit() and reach the limit again - # before the outer call disables; make disabling idempotent. - if destination not in self.destinations: - return - self.destinations = [ - existing - for existing in self.destinations - if existing is not destination - ] - self._consecutive_failures.pop(id(destination), None) + with self._lifecycle_lock: + # Failure reporting can re-enter emit() and reach the limit + # again before the outer call disables, and two threads can + # race here; the membership check under the lock makes + # disabling exactly-once. + if destination not in self.destinations: + return + self.destinations = [ + existing + for existing in self.destinations + if existing is not destination + ] + self._consecutive_failures.pop(id(destination), None) + if not self.destinations: + self.destinations = [self._stdout_destination()] + self._close_destination(destination, self.on_failure) self.on_failure( "logging.destination_disabled", RuntimeError( @@ -130,30 +265,35 @@ def _disable_destination(self, destination: LogDestination) -> None: ), destination=getattr(destination, "name", None), ) - if not self.destinations: - self.destinations.append(self._stdout_destination()) - def _build_destination(self, destination_name: str) -> LogDestination: + def _build_destination( + self, + destination_name: str, + report: Callable[..., None], + ) -> LogDestination: normalized = destination_name.strip().lower().replace("-", "_") # Stdout is the fallback sink and stays synchronous; every other # destination goes through the async-wrapping choke point below. if normalized == "stdout": return self._stdout_destination() return self._maybe_background( - self._build_network_destination(normalized, destination_name) + self._build_network_destination( + normalized, destination_name, report + ) ) def _build_network_destination( self, normalized: str, destination_name: str, + report: Callable[..., None], ) -> LogDestination: - if normalized in {"google", "google_cloud", "google_cloud_logging"}: + if normalized in _GOOGLE_DESTINATION_NAMES: return GoogleCloudLoggingDestination( project=self.config.google_cloud_project, log_name=self.config.google_cloud_log_name, timeout_seconds=self.config.google_log_timeout_seconds, - on_failure=self.on_failure, + on_failure=report, ) raise ValueError( f"Unknown observability log destination: {destination_name}" @@ -173,48 +313,6 @@ def _maybe_background(self, destination: LogDestination) -> LogDestination: failure_limit=DESTINATION_FAILURE_LIMIT, ) - def flush(self, deadline_seconds: float | None = None) -> None: - for destination in list(self.destinations): - flush = getattr(destination, "flush", None) - if not callable(flush): - continue - try: - flush(deadline_seconds) - except BaseException as exc: - self.on_failure( - "logging.destination_flush", - exc, - destination=getattr(destination, "name", None), - ) - - def restart(self) -> None: - """Rebuild destinations from config. Restart must revive - destinations the breaker disabled and replace clients whose - connections did not survive a fork or memory-snapshot restore, - so it reconfigures from scratch rather than poking survivors.""" - self._consecutive_failures.clear() - self.configured = False - try: - self.configure() - except BaseException as exc: - self.destinations = [self._stdout_destination()] - self.configured = True - self.on_failure("logging.destination_restart", exc) - - def _close_destinations(self, destinations: list[LogDestination]) -> None: - for destination in destinations: - close = getattr(destination, "close", None) - if not callable(close): - continue - try: - close() - except BaseException as exc: - self.on_failure( - "logging.destination_close", - exc, - destination=getattr(destination, "name", None), - ) - def _stdout_destination(self) -> StdoutJsonDestination: return StdoutJsonDestination( loggers=self.loggers, diff --git a/policyengine_observability/destinations/stdout.py b/policyengine_observability/destinations/stdout.py index 085c0be..50c31a7 100644 --- a/policyengine_observability/destinations/stdout.py +++ b/policyengine_observability/destinations/stdout.py @@ -4,7 +4,7 @@ from collections.abc import Callable, Mapping from typing import Any -from .base import bounded_labels, normalize_payload +from .base import bounded_labels, normalize_payload, trace_resource_name GOOGLE_TRACE_KEY = "logging.googleapis.com/trace" GOOGLE_SPAN_ID_KEY = "logging.googleapis.com/spanId" @@ -65,21 +65,21 @@ def _google_line( log_type: str, severity: str, ) -> dict[str, Any]: - line = dict(normalized) - line["severity"] = str(severity).upper() - created_at = normalized.get("created_at") - if created_at: - line["time"] = created_at - trace_id = normalized.get("trace_id") - if trace_id and self.google_cloud_project: - line[GOOGLE_TRACE_KEY] = ( - f"projects/{self.google_cloud_project}/traces/{trace_id}" - ) + # normalize_payload returned a fresh dict; mutate it in place. + # Stdout emission is synchronous, so the agent's receive time is + # the event time and no explicit `time` key is needed. + normalized["severity"] = str(severity).upper() + trace = trace_resource_name( + self.google_cloud_project, + normalized.get("trace_id"), + ) + if trace: + normalized[GOOGLE_TRACE_KEY] = trace span_id = normalized.get("span_id") if span_id: - line[GOOGLE_SPAN_ID_KEY] = str(span_id) - line[GOOGLE_LABELS_KEY] = bounded_labels( + normalized[GOOGLE_SPAN_ID_KEY] = str(span_id) + normalized[GOOGLE_LABELS_KEY] = bounded_labels( normalized, log_type=log_type, ) - return line + return normalized diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 93b5a05..123d97b 100644 --- a/policyengine_observability/runtime.py +++ b/policyengine_observability/runtime.py @@ -1086,7 +1086,7 @@ def flush_log_destinations( ) -> None: try: self.log_destination_manager.flush(deadline_seconds) - except BaseException as exc: + except Exception as exc: self.log_observability_failure("logging.flush", exc) def restart_log_destinations(self) -> None: @@ -1096,7 +1096,6 @@ def restart_log_destinations(self) -> None: self.log_observability_failure("logging.restart", exc) def shutdown(self) -> None: - self.flush_log_destinations() providers = [ ("trace", self.tracer_provider), ("metrics", self.meter_provider), @@ -1106,10 +1105,11 @@ def shutdown(self) -> None: for name, provider in providers if provider is not None ] - if not providers: - return def flush() -> None: + # Log flush runs inside the same hard-bounded thread as the + # OTel providers so a hung sink cannot stall process exit. + self.flush_log_destinations() for name, provider in providers: try: provider.shutdown() diff --git a/tests/test_destinations.py b/tests/test_destinations.py index c4fbdd3..6520503 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -174,8 +174,8 @@ def test_google_destination_emit_batch_writes_one_call(monkeypatch) -> None: destination.emit_batch( [ - ({"event": "a"}, "event", "INFO"), - ({"event": "b"}, "request", "WARNING"), + ({"event": "a"}, "event", "INFO", "2026-07-08T00:00:00+00:00"), + ({"event": "b"}, "request", "WARNING", None), ] ) destination.emit_batch([]) @@ -369,7 +369,7 @@ def test_stdout_google_format_maps_agent_keys() -> None: line = json.loads(message) assert level == "warning" assert line["severity"] == "WARNING" - assert line["time"] == "2026-07-07T00:00:00+00:00" + assert "time" not in line assert ( line["logging.googleapis.com/trace"] == "projects/central-project/traces/abc123" @@ -396,7 +396,6 @@ def test_stdout_google_format_omits_trace_without_project() -> None: _level, message = logger.lines[0] line = json.loads(message) assert "logging.googleapis.com/trace" not in line - assert "time" not in line assert line["severity"] == "ERROR" @@ -505,7 +504,9 @@ def test_background_drain_prefers_emit_batch(monkeypatch) -> None: assert [len(batch) for batch in wrapped.batches] == [2, 1] assert wrapped.single_emits == [] - assert wrapped.batches[0][0] == ({"event": 0}, "event", "INFO") + payload, log_type, severity, timestamp = wrapped.batches[0][0] + assert (payload, log_type, severity) == ({"event": 0}, "event", "INFO") + assert isinstance(timestamp, str) and "T" in timestamp def test_background_drain_falls_back_to_single_emits(monkeypatch) -> None: @@ -528,7 +529,7 @@ def test_background_overflow_drops_oldest_and_reports(monkeypatch) -> None: for index in range(4): destination.emit({"event": index}, log_type="event", severity="INFO") - kept = [payload["event"] for payload, _, _ in destination._buffer] + kept = [payload["event"] for payload, *_ in destination._buffer] assert kept == [2, 3] overflow = [ fields @@ -555,12 +556,15 @@ def test_background_trips_after_consecutive_batch_failures( while not destination._tripped.is_set(): destination._drain_once() - counts = [ - fields["consecutive_failures"] + reports = [ + fields for operation, fields in failures if operation == "logging.destination_emit_async" ] - assert counts == [1, 2, 3] + assert [fields["consecutive_failures"] for fields in reports] == [1, 2, 3] + assert all(fields["records_lost"] == 1 for fields in reports) + assert "stranded" not in reports[0] + assert reports[2]["stranded"] == 2 with pytest.raises(RuntimeError, match="tripped"): destination.emit({"event": "x"}, log_type="event", severity="INFO") @@ -585,6 +589,7 @@ def test_background_trip_flows_through_manager_breaker(monkeypatch) -> None: manager.emit({"event": "x"}, log_type="event", severity="INFO") assert destination not in manager.destinations + assert destination._stopped.is_set() assert any( operation == "logging.destination_disabled" for operation, _fields in manager_failures @@ -630,27 +635,6 @@ def test_background_flush_reports_undelivered_remainder( assert incomplete[0]["remaining"] == 3 -def test_background_restart_clears_trip_and_buffer(monkeypatch) -> None: - wrapped = BatchRecordingDestination() - destination, _failures = _background( - FlakyDestination(), batch_size=1, failure_limit=1 - ) - _inline(destination, monkeypatch) - destination.emit({"event": "x"}, log_type="event", severity="INFO") - destination._drain_once() - assert destination._tripped.is_set() - - destination.restart() - destination.wrapped = wrapped - _inline(destination, monkeypatch) - - assert not destination._tripped.is_set() - assert len(destination._buffer) == 0 - destination.emit({"event": "y"}, log_type="event", severity="INFO") - destination._drain_once() - assert wrapped.batches == [[({"event": "y"}, "event", "INFO")]] - - def test_background_worker_delivers_end_to_end() -> None: wrapped = BatchRecordingDestination() destination, failures = _background( @@ -670,12 +654,14 @@ def test_background_worker_delivers_end_to_end() -> None: def test_background_worker_restarts_after_fork(monkeypatch) -> None: import os as os_module + import time as time_module from policyengine_observability.destinations import background wrapped = BatchRecordingDestination() destination, _failures = _background(wrapped) destination.emit({"event": "a"}, log_type="event", severity="INFO") + destination.flush(1.0) first_worker = destination._worker assert first_worker is not None @@ -686,11 +672,36 @@ def test_background_worker_restarts_after_fork(monkeypatch) -> None: assert destination._worker is not first_worker first_worker.join(timeout=2.0) assert not first_worker.is_alive() - destination.flush(1.0) + # The record must be delivered by the NEW worker, not a caller flush. + for _ in range(200): + if sum(len(batch) for batch in wrapped.batches) >= 2: + break + time_module.sleep(0.01) + delivered = [ + payload["event"] for batch in wrapped.batches for payload, *_ in batch + ] + assert "b" in delivered destination.close() -# ── Async wiring, lifecycle, and config knobs ──────────────────────────── +def test_reset_after_fork_drops_inherited_buffer(monkeypatch) -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped) + _inline(destination, monkeypatch) + destination.emit({"event": "a"}, log_type="event", severity="INFO") + + destination._reset_after_fork() + + assert len(destination._buffer) == 0 + destination.emit({"event": "b"}, log_type="event", severity="INFO") + fork_reports = [ + fields + for operation, fields in failures + if operation == "logging.destination_fork_buffer_dropped" + ] + assert fork_reports == [ + {"destination": "batch-recording", "dropped_total": 1} + ] def _config_manager(config): @@ -732,7 +743,9 @@ def emit(self, payload, *, log_type, severity) -> None: ObservabilityConfig(log_emit_mode="async", log_queue_size=7) ) - destination = manager._build_destination("google_cloud_logging") + destination = manager._build_destination( + "google_cloud_logging", lambda *args, **kwargs: None + ) assert isinstance(destination, BackgroundEmitDestination) assert isinstance(destination.wrapped, StubGoogle) @@ -758,7 +771,9 @@ def emit(self, payload, *, log_type, severity) -> None: ) manager = _config_manager(ObservabilityConfig()) - destination = manager._build_destination("google") + destination = manager._build_destination( + "google", lambda *args, **kwargs: None + ) assert isinstance(destination, StubGoogle) @@ -771,7 +786,9 @@ def test_async_mode_never_wraps_stdout() -> None: manager = _config_manager(ObservabilityConfig(log_emit_mode="async")) - destination = manager._build_destination("stdout") + destination = manager._build_destination( + "stdout", lambda *args, **kwargs: None + ) assert isinstance(destination, StdoutJsonDestination) @@ -781,7 +798,6 @@ class FlushRecordingDestination: def __init__(self) -> None: self.flushes = [] - self.restarts = 0 def emit(self, payload, *, log_type, severity) -> None: pass @@ -789,9 +805,6 @@ def emit(self, payload, *, log_type, severity) -> None: def flush(self, deadline_seconds=None) -> None: self.flushes.append(deadline_seconds) - def restart(self) -> None: - self.restarts += 1 - def test_manager_flush_reaches_capable_destinations() -> None: flushable = FlushRecordingDestination() @@ -800,7 +813,8 @@ def test_manager_flush_reaches_capable_destinations() -> None: manager.flush(1.5) - assert flushable.flushes == [1.5] + assert len(flushable.flushes) == 1 + assert 1.0 < flushable.flushes[0] <= 1.5 assert failures == [] @@ -924,6 +938,8 @@ def test_google_destination_reports_unbounded_transport(monkeypatch) -> None: def test_google_destination_stamps_event_timestamp(monkeypatch) -> None: + """Synchronous writes rely on server receive time; batch records + carry the background emitter's enqueue timestamp.""" client = FakeClient(gapic=False) destination = _destination(monkeypatch, client) @@ -932,16 +948,14 @@ def test_google_destination_stamps_event_timestamp(monkeypatch) -> None: log_type="event", severity="INFO", ) - destination.emit( - {"created_at": "not-a-timestamp", "event": "y"}, - log_type="event", - severity="INFO", + destination.emit_batch( + [({"event": "y"}, "event", "INFO", "2026-07-08T01:00:00+00:00")] ) - (stamped,), _ = client.logging_api.calls[0] - assert stamped["timestamp"] == "2026-07-08T00:00:00+00:00" - (unstamped,), _ = client.logging_api.calls[1] - assert "timestamp" not in unstamped + (sync_entry,), _ = client.logging_api.calls[0] + assert "timestamp" not in sync_entry + (batch_entry,), _ = client.logging_api.calls[1] + assert batch_entry["timestamp"] == "2026-07-08T01:00:00+00:00" def test_internal_error_flag_is_thread_local() -> None: @@ -1060,3 +1074,202 @@ def test_from_env_emission_knobs_default_and_reject_garbage( assert config.log_emit_mode == "sync" assert config.log_queue_size == 1000 + + +def test_background_flush_zero_deadline_attempts_one_drain( + monkeypatch, +) -> None: + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped) + _inline(destination, monkeypatch) + destination.emit({"event": "a"}, log_type="event", severity="INFO") + + destination.flush(0) + + assert sum(len(batch) for batch in wrapped.batches) == 1 + assert failures == [] + + +def test_background_flush_waits_for_completing_in_flight_batch( + monkeypatch, +) -> None: + import threading + + wrapped = BatchRecordingDestination() + destination, failures = _background(wrapped) + _inline(destination, monkeypatch) + with destination._lock: + destination._in_flight = 1 + + def complete() -> None: + with destination._lock: + destination._in_flight = 0 + + timer = threading.Timer(0.05, complete) + timer.start() + destination.flush(1.0) + timer.join() + + assert failures == [] + + +def test_background_worker_survives_base_exception_with_report( + monkeypatch, +) -> None: + import threading + + class ExitingDestination: + name = "exiting" + + def emit(self, payload, *, log_type, severity) -> None: + raise SystemExit(1) + + destination, failures = _background(ExitingDestination(), batch_size=1) + _inline(destination, monkeypatch) + destination.emit({"event": "a"}, log_type="event", severity="INFO") + closed = threading.Event() + + destination._run(closed) + + crashes = [ + operation + for operation, _fields in failures + if operation == "logging.destination_worker_crashed" + ] + assert crashes == ["logging.destination_worker_crashed"] + + +def test_background_close_reports_pending_and_is_terminal( + monkeypatch, +) -> None: + class ClosableWrapped(BatchRecordingDestination): + def __init__(self) -> None: + super().__init__() + self.closed = 0 + + def close(self) -> None: + self.closed += 1 + + wrapped = ClosableWrapped() + destination, failures = _background(wrapped) + _inline(destination, monkeypatch) + destination.emit({"event": "a"}, log_type="event", severity="INFO") + + destination.close() + + pending = [ + fields + for operation, fields in failures + if operation == "logging.destination_closed_pending" + ] + assert pending == [{"destination": "batch-recording", "remaining": 1}] + assert wrapped.closed == 1 + assert len(destination._buffer) == 0 + # Terminal: no acceptance, no worker respawn, no atexit re-register. + destination.emit({"event": "b"}, log_type="event", severity="INFO") + assert len(destination._buffer) == 0 + assert destination._worker is None + assert destination._atexit_registered is False + + +def test_manager_reports_unknown_emit_mode(monkeypatch) -> None: + from policyengine_observability.config import ObservabilityConfig + + manager = _config_manager(ObservabilityConfig(log_emit_mode="asnyc")) + warnings = [] + manager.on_failure = lambda operation, exc, **fields: warnings.append( + operation + ) + + manager.configure() + + assert "logging.destination_config_warning" in warnings + + +def test_manager_warns_on_double_ingestion_combo(monkeypatch) -> None: + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations import manager as manager_mod + + class StubGoogle: + name = "google_cloud_logging" + + def __init__(self, **kwargs) -> None: + pass + + def emit(self, payload, *, log_type, severity) -> None: + pass + + monkeypatch.setattr( + manager_mod, "GoogleCloudLoggingDestination", StubGoogle + ) + manager = _config_manager( + ObservabilityConfig( + stdout_format="google", + log_destinations=("stdout", "google_cloud_logging"), + ) + ) + warnings = [] + manager.on_failure = lambda operation, exc, **fields: warnings.append( + (operation, str(exc)) + ) + + manager.configure() + + assert any( + operation == "logging.destination_config_warning" + and "twice" in message + for operation, message in warnings + ) + + +def test_config_normalizes_string_knobs_programmatically() -> None: + from policyengine_observability.config import ObservabilityConfig + + config = ObservabilityConfig( + stdout_format=" Google ", log_emit_mode="ASYNC" + ) + + assert config.stdout_format == "google" + assert config.log_emit_mode == "async" + + +def test_configure_defers_reports_until_destinations_are_live( + monkeypatch, +) -> None: + """Failure reports fire only after the new destination set is live, + so a report's own emission cannot re-enter configuration.""" + import json + import logging + + from policyengine_observability.config import ObservabilityConfig + from policyengine_observability.destinations import manager as manager_mod + from policyengine_observability.destinations.manager import ( + LogDestinationManager, + ) + + events = [] + + def on_failure(operation, exc, **fields): + events.append(operation) + # Emulate the runtime: a report re-enters manager.emit. + manager.emit({"event": "report"}, log_type="event", severity="INFO") + + manager = LogDestinationManager( + config=ObservabilityConfig(log_destinations=("google_cloud_logging",)), + loggers={"event": logging.getLogger("test-deferred")}, + serializer=json.dumps, + on_failure=on_failure, + ) + monkeypatch.setattr( + manager_mod, + "GoogleCloudLoggingDestination", + lambda **kwargs: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + manager.emit({"event": "first"}, log_type="event", severity="INFO") + + # The build failure was reported after the fallback set went live, + # and the reentrant emit found a configured manager (no recursion). + assert "logging.destination_config" in events + assert manager.configured is True + assert len(manager.destinations) == 1