Bounded Google Cloud Logging writes, background log emitter, and agent-native stdout format

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

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.

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.

docs/operations/google-cloud-stage3-runbook.md

@@ -187,3 +187,14 @@ OBSERVABILITY_GOOGLE_SERVICE_ACCOUNT_EMAIL=<temporary-bridge-writer-service-acco
 
 Leave the central project and sink in place during rollback unless the sink
 itself is the source of the problem.
+
+
+## Kill-switches
+
+Both take effect on the next process start (or immediately via a Cloud Run
+`--update-env-vars` revision):
+
+- `OBSERVABILITY_LOG_DESTINATIONS=stdout` — turn off the direct Google Cloud
+  Logging path entirely; logs go to stdout only.
+- `OBSERVABILITY_LOG_EMIT_MODE=sync` — keep the Google path but revert from the
+  background emitter to bounded synchronous writes.

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",

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

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",