Log profiles, destination registry, and queued Cloud Logging transport

README.md

@@ -15,38 +15,110 @@ configured, spans and metrics stay in-process while logs still receive trace
 context. Set `OTEL_ENABLED=false` to opt out. Configure
 `OTEL_EXPORTER_OTLP_ENDPOINT` to export traces and metrics.
 
-Structured logs write to stdout by default:
+## Log routing profiles
+
+Log routing is owned by this package: consumers set one env var and the
+package expands it into destinations, formats, and transport.
 
 ```bash
-OBSERVABILITY_LOG_DESTINATIONS=stdout
+OBSERVABILITY_LOG_PROFILE=gcp-agent   # or gcp-direct | plain-sync | auto
 ```
 
-Applications can override that default in code when constructing
-`ObservabilityConfig`:
+- `gcp-agent` — google-format stdout only, for platforms whose logging
+  agent ingests stdout (Cloud Run, GKE). Fully synchronous, zero
+  threads; the agent ships lines to Cloud Logging with severity, trace,
+  span, and labels promoted to first-class LogEntry fields.
+- `gcp-direct` — plain stdout (the durable record) plus queued direct
+  Cloud Logging writes, for platforms with no ingesting agent (Modal).
+  Requires a resolvable Google Cloud project; downgrades to `plain-sync`
+  with a warning otherwise.
+- `plain-sync` — plain stdout only, guaranteed zero threads. Local
+  development and the kill switch: setting it disables all background
+  log machinery.
+- `auto` (default) — detects the platform via `OBSERVABILITY_PLATFORM`
+  (`google_cloud_run`/`modal`), then `K_SERVICE`, then Modal env
+  markers; when nothing matches, caller-supplied defaults apply.
+
+The granular controls still exist underneath and override the profile's
+expansion when set explicitly:
 
-```python
-ObservabilityConfig.from_env(
-    service_name="policyengine-api",
-    default_log_destinations=("google_cloud_logging",),
-)
+```bash
+OBSERVABILITY_LOG_DESTINATIONS=stdout,google_cloud_logging
+OBSERVABILITY_STDOUT_FORMAT=google
+OBSERVABILITY_GOOGLE_CLOUD_PROJECT=policyengine-api
+OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME=policyengine-observability
 ```
 
-`OBSERVABILITY_LOG_DESTINATIONS` still has precedence over application
-defaults. Cloud Run captures stdout and stderr into Google Cloud Logging
-automatically, but applications that need one consistent destination across
-Cloud Run and non-GCP runtimes can write directly to Google Cloud Logging by
-installing the `google` extra and enabling the Google destination:
+Destinations are named strategies: `stdout` is `inline` (synchronous on
+the caller's thread), and every `remote` strategy — `google_cloud_logging`
+today; future backends register the same way — is automatically wrapped
+in the queued transport below. Google Cloud Logging uses Application
+Default Credentials and requires permission to create log entries,
+typically through `roles/logging.logWriter`.
+
+Backends plug in through two top-level hooks, `register_destination`
+(with `transport="inline"|"remote"` and an optional `required_config`
+tuple naming the config fields the strategy needs — profiles that name
+the strategy downgrade gracefully when one is missing) and
+`register_stdout_formatter`. Registration happens at import time, so an
+external backend module must be imported before observability is
+configured. Backend-specific knobs are the strategy's own: the Google
+strategy reads `OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS` itself at
+construction (so it is re-read on `restart_observability()`) rather
+than through a core config field. Name lookups for destinations,
+formatters, and profiles all forgive case, whitespace, and
+hyphen/underscore variance; an unknown format name falls back to plain
+and is reported through the internal-error channel, and a registered
+formatter factory that raises degrades to the built-in plain formatter
+the same way rather than breaking configuration.
+
+## Log emission and delivery semantics
+
+Remote destinations never write on a request thread. The log call only
+snapshots the payload, stamps the enqueue time, and appends to a bounded
+in-memory queue (microseconds, never blocks, never raises); a stdlib
+`QueueListener` thread drains the queue and performs the writes, sending
+the enqueue time as the entry timestamp so delayed writes keep their
+event time.
+
+Delivery through the queue is best-effort by design — stdout is the
+durable sibling record. When the queue is full the newest record is
+dropped, and drops are counted and reported through the internal-error
+channel (first drop, then every 100th). Write failures are likewise
+reported and the record dropped; there is deliberately no breaker or
+retry queue in the transport. Each Google write carries an explicit
+budget that caps the call and its transient-error retries:
 
 ```bash
-OBSERVABILITY_LOG_DESTINATIONS=google_cloud_logging
-OBSERVABILITY_GOOGLE_CLOUD_PROJECT=policyengine-api
-OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME=policyengine-observability
+OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS=10.0
+OBSERVABILITY_LOG_QUEUE_MAXSIZE=1000
+OBSERVABILITY_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS=2.0
 ```
 
-Multiple destinations can be enabled with a comma-separated list, for example
-`OBSERVABILITY_LOG_DESTINATIONS=stdout,google_cloud_logging`. Google Cloud
-Logging uses Application Default Credentials and requires permission to create
-log entries, typically through `roles/logging.logWriter`.
+(The bounded write rebinds the client's private gapic method at
+construction; when that handle is unavailable — HTTP transports,
+injected fakes — the library's ~60s default applies, which is harmless
+off the request path. All numeric knobs are clamped, so `0`, negative,
+or non-finite values can never disable or unbound a mechanism. The
+Google client library's one-time instrumentation diagnostic entry is
+suppressed at construction, so the stream carries only the records the
+service asked to write.)
+
+Shutdown closes log destinations inside the same bounded budget that
+flushes OpenTelemetry (`OBSERVABILITY_SHUTDOWN_TIMEOUT_SECONDS`); a
+queue that cannot drain before its deadline is abandoned with a report.
+A hard kill loses whatever was still queued. Note the google stdout
+format sets no `time` key: stdout emission is synchronous, so the
+agent's receive time is the correct event time.
+
+Processes that fork or restore from memory snapshots do not preserve
+threads or network clients. Call `restart_observability()` from the
+post-restore or post-fork hook (for example gunicorn `post_fork` when
+using `--preload`, or a Modal post-snapshot hook) — it closes and
+rebuilds destinations from configuration, and must only be called from
+single-threaded lifecycle moments, before serving traffic. It is a
+no-op when observability is disabled, so the kill switch holds across
+forks and restores.
 
 Request and operation logs include two timing views:
 

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 destination strategy registry and a generic queued transport: destinations register as `inline` (synchronous, e.g. stdout) or `remote`, and every remote destination is wrapped in a bounded queue drained by a stdlib QueueListener thread, so no remote write ever runs on a request thread. The queue drops-and-counts when full (throttled reports), and close is deadline-bounded (`OBSERVABILITY_LOG_QUEUE_MAXSIZE`, `OBSERVABILITY_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS`). `register_destination` and `register_stdout_formatter` are exported at the package top level so external backends can register strategies (with `required_config` declaring the config fields a strategy needs, checked generically during profile expansion), and the Google credential helpers moved to `policyengine_observability.destinations.google_credentials`.

changelog.d/22.changed.md

@@ -0,0 +1 @@
+Google Cloud Logging writes now carry an explicit per-call budget (default 10 seconds, `OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS`, clamped to sane bounds) that caps both the call and its transient-error retries, replacing the transport defaults that could hold a single write for up to 60 seconds when the Logging API degrades.

changelog.d/22.credentials.removed.md

@@ -0,0 +1 @@
+Removed the `policyengine_observability.google_credentials` module path; the credential helpers live at `policyengine_observability.destinations.google_credentials` and remain re-exported from the package top level (`load_google_credentials`, `configure_google_application_credentials`), which no known consumer's imports go beyond.

changelog.d/22.profiles.added.md

@@ -0,0 +1 @@
+Added log profiles (`OBSERVABILITY_LOG_PROFILE`): one env var expands to a full routing configuration — `gcp-agent` (agent-native google-format stdout only, for platforms whose agent ingests stdout), `gcp-direct` (plain stdout plus queued direct Cloud Logging writes, for platforms without an agent), `plain-sync` (plain stdout, guaranteed zero threads — the kill switch), and `auto` (default: detects the platform via `OBSERVABILITY_PLATFORM`, `K_SERVICE`, or Modal markers, and preserves caller defaults when none match). Explicit `OBSERVABILITY_LOG_DESTINATIONS`/`OBSERVABILITY_STDOUT_FORMAT` still override the expansion; `gcp-direct` downgrades to `plain-sync` with a warning when no Google Cloud project is resolvable.

changelog.d/22.restart.added.md

@@ -0,0 +1 @@
+Added `restart_observability()`: closes and rebuilds log destinations from configuration, for runtimes whose processes fork or restore from memory snapshots (threads and network clients survive neither). Documented for single-threaded lifecycle moments only — post-snapshot-restore hooks, post-fork hooks, before serving traffic. `restart_observability()` is a no-op when observability is disabled, so the kill switch holds across forks and restores.

changelog.d/22.shutdown.changed.md

@@ -0,0 +1 @@
+Runtime shutdown now closes log destinations first, inside the same bounded shutdown budget that bounds the OpenTelemetry flush; one deadline is shared across all destination closes. The Google Cloud Logging client library's one-time instrumentation diagnostic entry is suppressed, and unknown `OBSERVABILITY_STDOUT_FORMAT` names are reported through the internal-error channel instead of silently falling back to plain, and a registered stdout-formatter factory that raises degrades to the built-in plain formatter (with a report) instead of breaking configuration.

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

@@ -100,7 +100,12 @@ the services being deployed.
 Use these variables for deployed Cloud Run and Modal environments:
 
 ```bash
-OBSERVABILITY_LOG_DESTINATIONS=google_cloud_logging
+# Cloud Run: the agent ingests stdout, no direct writes needed.
+OBSERVABILITY_LOG_PROFILE=gcp-agent
+OBSERVABILITY_GOOGLE_CLOUD_PROJECT=<observability-project-id>
+
+# Modal: stdout plus queued direct Cloud Logging writes.
+OBSERVABILITY_LOG_PROFILE=gcp-direct
 OBSERVABILITY_GOOGLE_CLOUD_PROJECT=<observability-project-id>
 OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME=<observability-log-name>
 OBSERVABILITY_GOOGLE_WORKLOAD_IDENTITY_PROVIDER=projects/<observability-project-number>/locations/global/workloadIdentityPools/<modal-pool-id>/providers/<modal-provider-id>
@@ -170,10 +175,11 @@ jsonPayload.event="observability_internal_error"
 
 ## Rollback
 
-For a failing runtime, set:
+For a failing runtime, set the kill switch — plain stdout only, with all
+background log machinery disabled:
 
 ```bash
-OBSERVABILITY_LOG_DESTINATIONS=stdout
+OBSERVABILITY_LOG_PROFILE=plain-sync
 ```
 
 To restore the temporary bridge destination, set:

policyengine_observability/__init__.py

@@ -4,7 +4,8 @@
 
 from .config import ObservabilityConfig
 from .context import OperationObservabilityContext, RequestObservabilityContext
-from .google_credentials import (
+from .destinations import register_destination, register_stdout_formatter
+from .destinations.google_credentials import (
     configure_google_application_credentials,
     load_google_credentials,
 )
@@ -109,6 +110,17 @@ def shutdown_tracing() -> None:
     shutdown_observability()
 
 
+def restart_observability() -> None:
+    """Close and rebuild log destinations from configuration.
+
+    For runtimes whose processes fork or restore from memory snapshots
+    (threads and network clients do not survive either). Call ONLY from
+    single-threaded lifecycle moments — a post-snapshot-restore hook, a
+    post-fork hook, before serving traffic.
+    """
+    observability_runtime().restart_log_destinations()
+
+
 def operation(name: str, *, flavor: str | None = None, **attrs: Any):
     return observability_runtime().operation(name, flavor=flavor, **attrs)
 
@@ -167,6 +179,9 @@ def collect_timings(name: str = "operation", **attrs: Any):
     "operation",
     "record_error",
     "record_event",
+    "register_destination",
+    "register_stdout_formatter",
+    "restart_observability",
     "segment",
     "set_attribute",
     "set_observability_runtime",