diff --git a/README.md b/README.md index 5a2ca2b..82a68e9 100644 --- a/README.md +++ b/README.md @@ -103,6 +103,73 @@ 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 +transient errors retried only inside that budget, so a degraded Logging +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 +``` + +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. 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 +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 +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. 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 +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, 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/changelog.d/22.added.md b/changelog.d/22.added.md new file mode 100644 index 0000000..da5b0ce --- /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, 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 new file mode 100644 index 0000000..ce49579 --- /dev/null +++ b/changelog.d/22.changed.md @@ -0,0 +1 @@ +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/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= 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 2e37860..7139054 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", @@ -45,6 +51,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: @@ -87,6 +103,27 @@ 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 = DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS + stdout_format: str = "plain" + log_emit_mode: str = "sync" + 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 + + 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( @@ -166,6 +203,32 @@ 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, + ), + stdout_format=( + os.getenv("OBSERVABILITY_STDOUT_FORMAT") or cls.stdout_format + ), + log_emit_mode=( + os.getenv("OBSERVABILITY_LOG_EMIT_MODE") or cls.log_emit_mode + ), + 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/__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 new file mode 100644 index 0000000..0b812e8 --- /dev/null +++ b/policyengine_observability/destinations/background.py @@ -0,0 +1,391 @@ +from __future__ import annotations + +import atexit +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 ( + 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_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" + +# (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), 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). + """ + + def __init__( + self, + wrapped: LogDestination, + *, + on_failure: Callable[..., None], + 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 = 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.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() + 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, + payload: dict[str, Any], + *, + 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 " + 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. + # 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 ( + self._dropped == 1 + or self._dropped % DROP_REPORT_EVERY == 0 + ): + 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", + RuntimeError( + "Background emitter buffer is full; dropping oldest " + "log records." + ), + destination=self.name, + dropped_total=dropped_total, + ) + 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 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() + 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 + 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: + 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 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 _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(): + return + with self._start_lock: + worker = self._worker + 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( + 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: + try: + while not closed.is_set(): + # 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 + # 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, closed: threading.Event | None = None) -> str: + """Write one batch; returns a DRAIN_* outcome.""" + with self._lock: + if not self._buffer: + return DRAIN_EMPTY + batch = [ + self._buffer.popleft() + for _ in range(min(self.batch_size, len(self._buffer))) + ] + self._in_flight += len(batch) + try: + 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], + closed: threading.Event | None, + ) -> bool: + try: + if self._wrapped_emit_batch is not None: + self._wrapped_emit_batch(batch) + else: + for payload, log_type, severity, _timestamp in batch: + self.wrapped.emit( + payload, + log_type=log_type, + severity=severity, + ) + except Exception as exc: + 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, + **fields, + ) + return False + with self._lock: + self._consecutive_failures = 0 + return True diff --git a/policyengine_observability/destinations/base.py b/policyengine_observability/destinations/base.py index 72e798e..28cb3ce 100644 --- a/policyengine_observability/destinations/base.py +++ b/policyengine_observability/destinations/base.py @@ -35,3 +35,30 @@ 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 + + +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}" diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 8d8d918..99228cf 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -1,26 +1,31 @@ from __future__ import annotations -from collections.abc import Callable +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 normalize_payload +from .base import ( + bounded_labels, + normalize_payload, + trace_resource_name, +) class GoogleCloudLogger(Protocol): - def log_struct( - self, - payload: dict[str, Any], - **kwargs: Any, - ) -> None: ... + full_name: str + default_resource: Any class GoogleCloudLoggingClient(Protocol): project: str | None + logging_api: Any def logger(self, log_name: str) -> GoogleCloudLogger: ... @@ -32,6 +37,19 @@ def logger(self, log_name: str) -> GoogleCloudLogger: ... class GoogleCloudLoggingDestination: + """Writes structured payloads to Google Cloud Logging. + + 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``. 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" def __init__( @@ -40,9 +58,12 @@ def __init__( project: str | None, log_name: str, client_factory: GoogleCloudLoggingClientFactory | None = None, + timeout_seconds: float = DEFAULT_GOOGLE_LOG_TIMEOUT_SECONDS, + on_failure: Callable[..., None] | None = None, ) -> 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() @@ -61,6 +82,74 @@ 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._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 + ) -> 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: + 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: + # 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, + timeout=self.timeout_seconds, + predicate=if_exception_type( + api_exceptions.DeadlineExceeded, + api_exceptions.InternalServerError, + api_exceptions.ServiceUnavailable, + ), + ) + 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( + "Google Cloud Logging gapic transport unavailable; " + "writes fall back to the unbounded write_entries " + "call." + ), + destination=self.name, + ) + return None def emit( self, @@ -69,29 +158,84 @@ def emit( log_type: str, severity: str, ) -> None: - normalized = normalize_payload(payload) - kwargs: dict[str, Any] = { - "severity": severity, - "labels": _labels(normalized, log_type=log_type), + self._write( + [self._build_entry(payload, log_type=log_type, severity=severity)] + ) + + def emit_batch( + self, + records: Sequence[tuple[dict[str, Any], str, str, str | None]], + ) -> None: + """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, + 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 = payload if pre_normalized else normalize_payload(payload) + entry: dict[str, Any] = { + "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: - kwargs["trace"] = f"projects/{self.project}/traces/{trace_id}" + # 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")) + if trace: + entry["trace"] = trace span_id = normalized.get("span_id") if span_id: - kwargs["span_id"] = span_id - self.logger.log_struct(normalized, **kwargs) - - -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 + entry["spanId"] = str(span_id) + return entry + + def _write(self, entries: list[dict[str, Any]]) -> None: + if self._bounded_write is not None: + self._bounded_write(entries) + return + self.client.logging_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 diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 76b4725..f4480b9 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -1,21 +1,30 @@ from __future__ import annotations import logging +import threading +import time from collections.abc import Callable, Mapping 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 -# 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 +_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__( @@ -33,34 +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: - 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, @@ -96,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: @@ -107,12 +241,22 @@ def _ensure_destinations(self) -> list[LogDestination]: return self.destinations def _disable_destination(self, destination: LogDestination) -> None: - self.destinations = [ - existing - for existing in self.destinations - if existing is not destination - ] - self._consecutive_failures.pop(id(destination), None) + 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( @@ -121,24 +265,58 @@ 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() - if normalized in {"google", "google_cloud", "google_cloud_logging"}: + return self._maybe_background( + 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_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=report, ) raise ValueError( f"Unknown observability log destination: {destination_name}" ) + def _maybe_background(self, destination: LogDestination) -> LogDestination: + emit_mode = (self.config.log_emit_mode or "").strip().lower() + if 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 _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..50c31a7 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, trace_resource_name + +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]: + # 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: + normalized[GOOGLE_SPAN_ID_KEY] = str(span_id) + normalized[GOOGLE_LABELS_KEY] = bounded_labels( + normalized, + log_type=log_type, + ) + return normalized diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 8fbda49..123d97b 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)) @@ -1070,6 +1081,20 @@ 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 Exception 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: providers = [ ("trace", self.tracer_provider), @@ -1080,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 10c1941..6520503 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,98 @@ 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", "2026-07-08T00:00:00+00:00"), + ({"event": "b"}, "request", "WARNING", None), + ] + ) + 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 ────────────────────────────────────────── @@ -208,3 +310,966 @@ 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 "time" not in line + 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 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" + + +# ── 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)) + + +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() != "empty": + pass + + assert [len(batch) for batch in wrapped.batches] == [2, 1] + assert wrapped.single_emits == [] + 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: + 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() != "empty": + pass + + assert wrapped.payloads == [{"event": "a"}] + + +def test_background_overflow_drops_oldest_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 == [2, 3] + 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( + FlakyDestination(), batch_size=1, failure_limit=3 + ) + _inline(destination, monkeypatch) + + for index in range(5): + destination.emit({"event": index}, log_type="event", severity="INFO") + while not destination._tripped.is_set(): + destination._drain_once() + + reports = [ + fields + for operation, fields in failures + if operation == "logging.destination_emit_async" + ] + 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") + + +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( + FlakyDestination(), 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 destination._stopped.is_set() + 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( + FlakyDestination(), 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_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) + 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 False + + +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 + + 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 + first_worker.join(timeout=2.0) + assert not first_worker.is_alive() + # 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() + + +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): + 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", lambda *args, **kwargs: None + ) + + 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", lambda *args, **kwargs: None + ) + + 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", lambda *args, **kwargs: None + ) + + assert isinstance(destination, StdoutJsonDestination) + + +class FlushRecordingDestination: + name = "flush-recording" + + def __init__(self) -> None: + self.flushes = [] + + def emit(self, payload, *, log_type, severity) -> None: + pass + + def flush(self, deadline_seconds=None) -> None: + self.flushes.append(deadline_seconds) + + +def test_manager_flush_reaches_capable_destinations() -> None: + flushable = FlushRecordingDestination() + plain = RecordingDestination() + manager, failures = _manager([flushable, plain]) + + manager.flush(1.5) + + assert len(flushable.flushes) == 1 + assert 1.0 < flushable.flushes[0] <= 1.5 + 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: + """Synchronous writes rely on server receive time; batch records + carry the background emitter's enqueue timestamp.""" + 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_batch( + [({"event": "y"}, "event", "INFO", "2026-07-08T01:00:00+00:00")] + ) + + (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: + 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: + 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 + + +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 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" + )