diff --git a/README.md b/README.md index 5a2ca2b..15ebeb2 100644 --- a/README.md +++ b/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: diff --git a/changelog.d/22.added.md b/changelog.d/22.added.md new file mode 100644 index 0000000..e1e0f77 --- /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 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`. diff --git a/changelog.d/22.changed.md b/changelog.d/22.changed.md new file mode 100644 index 0000000..63043d7 --- /dev/null +++ b/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. diff --git a/changelog.d/22.credentials.removed.md b/changelog.d/22.credentials.removed.md new file mode 100644 index 0000000..eeddf4c --- /dev/null +++ b/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. diff --git a/changelog.d/22.profiles.added.md b/changelog.d/22.profiles.added.md new file mode 100644 index 0000000..15dfff4 --- /dev/null +++ b/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. diff --git a/changelog.d/22.restart.added.md b/changelog.d/22.restart.added.md new file mode 100644 index 0000000..9328093 --- /dev/null +++ b/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. diff --git a/changelog.d/22.shutdown.changed.md b/changelog.d/22.shutdown.changed.md new file mode 100644 index 0000000..8b2fe4a --- /dev/null +++ b/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. diff --git a/docs/operations/google-cloud-stage3-runbook.md b/docs/operations/google-cloud-stage3-runbook.md index 6f0ea5d..c6c25b2 100644 --- a/docs/operations/google-cloud-stage3-runbook.md +++ b/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= + +# Modal: stdout plus queued direct Cloud Logging writes. +OBSERVABILITY_LOG_PROFILE=gcp-direct OBSERVABILITY_GOOGLE_CLOUD_PROJECT= OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME= OBSERVABILITY_GOOGLE_WORKLOAD_IDENTITY_PROVIDER=projects//locations/global/workloadIdentityPools//providers/ @@ -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: diff --git a/policyengine_observability/__init__.py b/policyengine_observability/__init__.py index 88a8fdf..538d1f0 100644 --- a/policyengine_observability/__init__.py +++ b/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", diff --git a/policyengine_observability/config.py b/policyengine_observability/config.py index 2e37860..5d58448 100644 --- a/policyengine_observability/config.py +++ b/policyengine_observability/config.py @@ -2,8 +2,15 @@ import logging import os -from collections.abc import Sequence +from collections.abc import Mapping, Sequence from dataclasses import dataclass +from typing import Any + +# Defaults for the generic queued-transport knobs; the queued destination +# imports these so a constructor call and an env-configured build can +# never disagree about what "default" means. +DEFAULT_LOG_QUEUE_MAXSIZE = 1000 +DEFAULT_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS = 2.0 DEFAULT_METRIC_ATTRIBUTE_KEYS = ( "service.name", @@ -55,6 +62,16 @@ def float_from_env(name: str, default: float) -> float: return default +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 default_environment() -> str: return ( os.getenv("OBSERVABILITY_ENVIRONMENT") @@ -65,6 +82,103 @@ def default_environment() -> str: ) +# A log profile is a named preset expanding to generic routing primitives +# (a destination-name tuple and a stdout-formatter name). Presets may name +# strategies; the expansion mechanism knows nothing about any backend. +LOG_PROFILE_PRESETS: dict[str, tuple[tuple[str, ...], str]] = { + # Platforms whose logging agent ingests stdout (Cloud Run, GKE): + # agent-native stdout only, fully synchronous, zero threads. + "gcp-agent": (("stdout",), "google"), + # Platforms with no ingesting agent (Modal): plain stdout as the + # durable record plus queued direct Cloud Logging writes. + "gcp-direct": (("stdout", "google_cloud_logging"), "plain"), + # Local development and the kill switch: plain stdout, zero threads. + "plain-sync": (("stdout",), "plain"), +} + + +def _detect_log_profile() -> str | None: + platform = (os.getenv("OBSERVABILITY_PLATFORM") or "").strip().lower() + if platform == "google_cloud_run": + return "gcp-agent" + if platform == "modal": + return "gcp-direct" + if os.getenv("K_SERVICE"): + return "gcp-agent" + if os.getenv("MODAL_ENVIRONMENT") or os.getenv("MODAL_TASK_ID"): + return "gcp-direct" + return None + + +def _missing_strategy_requirements( + destination_names: Sequence[str], + resolved_config: Mapping[str, Any], +) -> list[tuple[str, str]]: + """(destination, config field) pairs a preset needs but lacks. + + Strategies declare their requirements at registration + (``register_destination(required_config=...)``); this check knows + nothing about any backend. + """ + # Imported lazily: the destinations package imports this module, so + # a module-level import here would be circular. By the time a config + # is resolved the package (and its strategy registrations) is loaded. + from .destinations.registry import destination_strategy + + missing: list[tuple[str, str]] = [] + for name in destination_names: + strategy = destination_strategy(name) + if strategy is None: + continue + for field in strategy.required_config: + if not resolved_config.get(field): + missing.append((name, field)) + return missing + + +def _resolve_log_profile( + raw_profile: str, + *, + resolved_config: Mapping[str, Any], +) -> tuple[str, tuple[tuple[str, ...], str] | None, list[str]]: + """Resolve a profile name to (name, preset-or-None, warnings). + + ``auto`` without a recognized platform marker resolves to no preset, + so caller-supplied defaults keep applying. ``resolved_config`` + carries the already-resolved config values that registered + strategies may declare as requirements. + """ + warnings: list[str] = [] + # Canonical profile names are hyphenated; accept the same case, + # whitespace, and hyphen/underscore variance as destination and + # formatter names. + profile = raw_profile.strip().lower().replace("_", "-") + if profile == "auto": + detected = _detect_log_profile() + if detected is None: + return "auto", None, warnings + profile = detected + preset = LOG_PROFILE_PRESETS.get(profile) + if preset is None: + warnings.append( + f"Unknown OBSERVABILITY_LOG_PROFILE {raw_profile!r}; " + "using plain-sync." + ) + profile = "plain-sync" + preset = LOG_PROFILE_PRESETS[profile] + missing = _missing_strategy_requirements(preset[0], resolved_config) + if missing: + requirements = ", ".join( + f"{field} (destination {name})" for name, field in missing + ) + warnings.append( + f"Log profile {profile} requires {requirements}; using plain-sync." + ) + profile = "plain-sync" + preset = LOG_PROFILE_PRESETS[profile] + return profile, preset, warnings + + @dataclass(frozen=True) class ObservabilityConfig: service_name: str = "policyengine-service" @@ -87,6 +201,13 @@ class ObservabilityConfig: log_destinations: tuple[str, ...] = ("stdout",) google_cloud_project: str | None = None google_cloud_log_name: str = "policyengine-observability" + stdout_format: str = "plain" + log_queue_maxsize: int = DEFAULT_LOG_QUEUE_MAXSIZE + log_queue_close_timeout_seconds: float = ( + DEFAULT_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS + ) + log_profile: str = "auto" + config_warnings: tuple[str, ...] = () @classmethod def from_env( @@ -120,6 +241,32 @@ def from_env( or DEFAULT_METRIC_ATTRIBUTE_KEYS, (*extra_metric_attribute_keys, *env_extra_metric_keys), ) + google_cloud_project = ( + os.getenv("OBSERVABILITY_GOOGLE_CLOUD_PROJECT") + or os.getenv("GOOGLE_CLOUD_PROJECT") + or os.getenv("GCP_PROJECT") + or os.getenv("GCLOUD_PROJECT") + or None + ) + log_profile, preset, profile_warnings = _resolve_log_profile( + os.getenv("OBSERVABILITY_LOG_PROFILE") or cls.log_profile, + # The values strategies may declare via required_config; + # extend as future fields become requirement candidates. + resolved_config={"google_cloud_project": google_cloud_project}, + ) + profile_destinations, profile_stdout_format = preset or (None, None) + # Explicit granular env vars override the profile's expansion; + # the profile overrides caller-supplied defaults. + resolved_log_destinations = _dedupe( + env_log_destinations + or profile_destinations + or default_log_destinations + ) + resolved_stdout_format = ( + os.getenv("OBSERVABILITY_STDOUT_FORMAT") + or profile_stdout_format + or cls.stdout_format + ) return cls( service_name=os.getenv("OBSERVABILITY_SERVICE_NAME") or os.getenv("OTEL_SERVICE_NAME") @@ -141,7 +288,7 @@ def from_env( meter_name=os.getenv("OBSERVABILITY_METER_NAME"), shutdown_timeout_seconds=float_from_env( "OBSERVABILITY_SHUTDOWN_TIMEOUT_SECONDS", - 3.0, + cls.shutdown_timeout_seconds, ), instrument_fastapi=bool_from_env( "OBSERVABILITY_INSTRUMENT_FASTAPI", @@ -152,20 +299,23 @@ def from_env( instrument_httpx, ), metric_attribute_keys=resolved_metric_keys, - log_destinations=_dedupe( - env_log_destinations or default_log_destinations - ), - google_cloud_project=( - os.getenv("OBSERVABILITY_GOOGLE_CLOUD_PROJECT") - or os.getenv("GOOGLE_CLOUD_PROJECT") - or os.getenv("GCP_PROJECT") - or os.getenv("GCLOUD_PROJECT") - or None - ), + log_destinations=resolved_log_destinations, + google_cloud_project=google_cloud_project, google_cloud_log_name=( os.getenv("OBSERVABILITY_GOOGLE_CLOUD_LOG_NAME") or cls.google_cloud_log_name ), + stdout_format=resolved_stdout_format, + log_queue_maxsize=int_from_env( + "OBSERVABILITY_LOG_QUEUE_MAXSIZE", + cls.log_queue_maxsize, + ), + log_queue_close_timeout_seconds=float_from_env( + "OBSERVABILITY_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS", + cls.log_queue_close_timeout_seconds, + ), + log_profile=log_profile, + config_warnings=tuple(profile_warnings), ) diff --git a/policyengine_observability/destinations/__init__.py b/policyengine_observability/destinations/__init__.py index 5703e23..e37847b 100644 --- a/policyengine_observability/destinations/__init__.py +++ b/policyengine_observability/destinations/__init__.py @@ -3,12 +3,17 @@ from .base import LogDestination, normalize_payload from .google_cloud_logging import GoogleCloudLoggingDestination from .manager import LogDestinationManager -from .stdout import StdoutJsonDestination +from .queued import QueuedLogDestination +from .registry import register_destination +from .stdout import StdoutJsonDestination, register_stdout_formatter __all__ = [ "GoogleCloudLoggingDestination", "LogDestination", "LogDestinationManager", + "QueuedLogDestination", "StdoutJsonDestination", "normalize_payload", + "register_destination", + "register_stdout_formatter", ] diff --git a/policyengine_observability/destinations/base.py b/policyengine_observability/destinations/base.py index 72e798e..8c93e09 100644 --- a/policyengine_observability/destinations/base.py +++ b/policyengine_observability/destinations/base.py @@ -1,6 +1,8 @@ from __future__ import annotations -from collections.abc import Mapping, Sequence +import inspect +import math +from collections.abc import Callable, Mapping, Sequence from typing import Any, Protocol @@ -17,6 +19,88 @@ def emit( """Write one structured observability payload.""" +def normalize_name(name: str) -> str: + """Canonical lookup key for registered names. + + Destination, formatter, and profile lookups all forgive case, + surrounding whitespace, and hyphen/underscore variance the same way, + so a spelling that works for one registry works for every registry. + """ + return name.strip().lower().replace("-", "_") + + +def accepts_keyword(func: Callable[..., Any], name: str) -> bool: + """Whether ``func`` can safely be called with keyword ``name``.""" + try: + parameters = inspect.signature(func).parameters + except (TypeError, ValueError): + return False + if name in parameters: + return True + return any( + parameter.kind is inspect.Parameter.VAR_KEYWORD + for parameter in parameters.values() + ) + + +def safe_report( + on_failure: Callable[..., None], + operation: str, + exc: BaseException, + **fields: Any, +) -> None: + """Report through the internal-error channel; never raises.""" + try: + on_failure(operation, exc, **fields) + except Exception: + pass + + +def close_destination( + destination: LogDestination, + *, + on_failure: Callable[..., None], + deadline_seconds: float | None = None, +) -> None: + """Close a destination if it supports closing; never raises. + + ``close`` is duck-typed with one calling convention everywhere: the + deadline is passed only when the signature accepts it, so both + ``close(self)`` and ``close(self, deadline_seconds=None)`` work under + every close path (manager shutdown, reconfigure, queued drain). + """ + close = getattr(destination, "close", None) + if not callable(close): + return + try: + if accepts_keyword(close, "deadline_seconds"): + close(deadline_seconds=deadline_seconds) + else: + close() + except Exception as exc: + safe_report( + on_failure, + "logging.destination_close", + exc, + destination=getattr(destination, "name", None), + ) + + +def clamped(value: Any, *, low: float, high: float, default: float) -> float: + """Coerce a config knob to a finite float within [low, high]. + + Anything unparseable or non-finite falls back to the default, so a + stray env value can never disable or unbound the mechanism it tunes. + """ + try: + number = float(value) + except (TypeError, ValueError): + return default + if not math.isfinite(number): + return default + return min(max(number, low), high) + + def normalize_payload(value: Any) -> Any: if value is None or isinstance(value, str | bool | int | float): return value diff --git a/policyengine_observability/destinations/google_cloud_logging.py b/policyengine_observability/destinations/google_cloud_logging.py index 8d8d918..2e1c96e 100644 --- a/policyengine_observability/destinations/google_cloud_logging.py +++ b/policyengine_observability/destinations/google_cloud_logging.py @@ -1,14 +1,28 @@ from __future__ import annotations +import functools from collections.abc import Callable +from datetime import datetime from typing import Any, Protocol -from policyengine_observability.google_credentials import ( +from ..config import float_from_env +from .base import clamped, normalize_payload +from .google_credentials import ( configure_google_application_credentials, load_google_credentials, ) +from .registry import register_destination +from .stdout import StdoutFormatter, register_stdout_formatter -from .base import normalize_payload +DEFAULT_WRITE_TIMEOUT_SECONDS = 10.0 +MIN_WRITE_TIMEOUT_SECONDS = 0.5 +MAX_WRITE_TIMEOUT_SECONDS = 60.0 + +# Structured-JSON keys the Cloud Run/GKE logging agent promotes to +# first-class LogEntry fields when it ingests a stdout line. +GOOGLE_TRACE_KEY = "logging.googleapis.com/trace" +GOOGLE_SPAN_ID_KEY = "logging.googleapis.com/spanId" +GOOGLE_LABELS_KEY = "logging.googleapis.com/labels" class GoogleCloudLogger(Protocol): @@ -40,9 +54,16 @@ def __init__( project: str | None, log_name: str, client_factory: GoogleCloudLoggingClientFactory | None = None, + write_timeout_seconds: float | None = None, ) -> None: self.project = project self.log_name = log_name + self.write_timeout_seconds = clamped( + write_timeout_seconds, + low=MIN_WRITE_TIMEOUT_SECONDS, + high=MAX_WRITE_TIMEOUT_SECONDS, + default=DEFAULT_WRITE_TIMEOUT_SECONDS, + ) credentials = load_google_credentials(prefer_workload_identity=True) if credentials is None: configure_google_application_credentials() @@ -61,6 +82,61 @@ 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._bound_write_timeout() + self._suppress_instrumentation_entry() + + def _suppress_instrumentation_entry(self) -> None: + """Keep the library's diagnostic entry out of the log stream. + + ``log_struct`` prepends a one-time instrumentation diagnostic + entry to the first write per process; observability entries + should be only the records we were asked to write. + """ + try: + from google.cloud import logging_v2 + + logging_v2._instrumentation_emitted = True + except ImportError: # pragma: no cover - google extra has it + pass + + def _bound_write_timeout(self) -> None: + """Bound every write on this destination's own client. + + ``log_struct`` exposes no call options, and the transport default + lets a degraded Logging API hold one write for up to 60 seconds. + The gapic method is the single choke point underneath + ``log_struct``, so rebind it on this client instance with a retry + whose transient-error budget and per-call timeout are both capped + by ``write_timeout_seconds``. When the private handle is absent + (HTTP transport, injected fakes), the library default applies — + acceptable because no write on this destination ever runs on a + request thread. + """ + api = getattr(self.client, "logging_api", None) + gapic = getattr(api, "_gapic_api", None) + if gapic is None: + return + try: + from google.api_core import exceptions as api_exceptions + from google.api_core.retry import Retry, if_exception_type + except ImportError: # pragma: no cover - google extra always has it + return + retry = Retry( + initial=0.1, + maximum=1.0, + multiplier=1.3, + timeout=self.write_timeout_seconds, + predicate=if_exception_type( + api_exceptions.DeadlineExceeded, + api_exceptions.InternalServerError, + api_exceptions.ServiceUnavailable, + ), + ) + gapic.write_log_entries = functools.partial( + gapic.write_log_entries, + retry=retry, + timeout=self.write_timeout_seconds, + ) def emit( self, @@ -68,20 +144,38 @@ def emit( *, log_type: str, severity: str, + timestamp: datetime | None = None, ) -> None: normalized = normalize_payload(payload) kwargs: dict[str, Any] = { "severity": severity, "labels": _labels(normalized, log_type=log_type), } + if timestamp is not None: + # Supplied by queued transports so delayed writes keep the + # record's event time instead of the delivery time. + kwargs["timestamp"] = timestamp trace_id = normalized.get("trace_id") if trace_id and self.project: - kwargs["trace"] = f"projects/{self.project}/traces/{trace_id}" + kwargs["trace"] = _trace_resource(self.project, trace_id) span_id = normalized.get("span_id") if span_id: kwargs["span_id"] = span_id self.logger.log_struct(normalized, **kwargs) + def close(self) -> None: + """Release the owned client's transport, if it supports it.""" + close = getattr(self.client, "close", None) + if callable(close): + close() + + +def _trace_resource(project: str, trace_id: str) -> str: + # The LogEntry trace resource name; the direct write path and the + # agent-native stdout formatter must build it identically for trace + # correlation to work. + return f"projects/{project}/traces/{trace_id}" + def _labels(payload: dict[str, Any], *, log_type: str) -> dict[str, str]: labels = {"log_type": log_type} @@ -95,3 +189,53 @@ def _labels(payload: dict[str, Any], *, log_type: str) -> dict[str, str]: if value is not None: labels[key] = str(value) return labels + + +def _google_stdout_formatter_factory(config: Any) -> StdoutFormatter: + """Shape stdout lines with the agent-native Cloud Logging keys. + + Emission stays synchronous, so no ``time`` key is set — the agent's + receive time is the event time. + """ + project = getattr(config, "google_cloud_project", None) + + def format_google( + payload: dict[str, Any], *, log_type: str, severity: str + ) -> dict[str, Any]: + payload["severity"] = str(severity).upper() + payload[GOOGLE_LABELS_KEY] = _labels(payload, log_type=log_type) + trace_id = payload.get("trace_id") + if trace_id and project: + payload[GOOGLE_TRACE_KEY] = _trace_resource(project, trace_id) + span_id = payload.get("span_id") + if span_id: + payload[GOOGLE_SPAN_ID_KEY] = str(span_id) + return payload + + return format_google + + +register_stdout_formatter("google", _google_stdout_formatter_factory) + + +def _google_destination_factory(*, config: Any, **_: Any): + return GoogleCloudLoggingDestination( + project=config.google_cloud_project, + log_name=config.google_cloud_log_name, + # Backend knobs belong to the strategy: parsed here at + # construction (so re-read on restart_observability()), keeping + # per-backend fields off the core config dataclass. + write_timeout_seconds=float_from_env( + "OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS", + DEFAULT_WRITE_TIMEOUT_SECONDS, + ), + ) + + +register_destination( + "google_cloud_logging", + _google_destination_factory, + transport="remote", + aliases=("google", "google_cloud"), + required_config=("google_cloud_project",), +) diff --git a/policyengine_observability/google_credentials.py b/policyengine_observability/destinations/google_credentials.py similarity index 100% rename from policyengine_observability/google_credentials.py rename to policyengine_observability/destinations/google_credentials.py diff --git a/policyengine_observability/destinations/manager.py b/policyengine_observability/destinations/manager.py index 76b4725..086e377 100644 --- a/policyengine_observability/destinations/manager.py +++ b/policyengine_observability/destinations/manager.py @@ -1,19 +1,23 @@ from __future__ import annotations import logging +import time from collections.abc import Callable, Mapping from typing import Any from ..config import ObservabilityConfig -from .base import LogDestination -from .google_cloud_logging import GoogleCloudLoggingDestination -from .stdout import StdoutJsonDestination +from .base import LogDestination, close_destination +from .queued import QueuedLogDestination +from .registry import destination_strategy +from .stdout import StdoutJsonDestination, build_stdout_destination # 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. +# rest of the process. Inline destinations emit synchronously on the +# caller's (request) path, so a persistently failing one — e.g. a broken +# serializer — must not keep charging every request; an observability sink +# can never be allowed to degrade the host service. Queued destinations +# never raise from emit (drops are counted internally), so this breaker +# only ever governs inline strategies. DESTINATION_FAILURE_LIMIT = 3 @@ -35,32 +39,57 @@ def __init__( self._consecutive_failures: dict[int, int] = {} def configure(self) -> None: - failures: list[tuple[str, BaseException]] = [] + # Reconfigure (restart_observability) runs only from + # single-threaded lifecycle moments by documented contract. + # Construction-time reports are deferred until the new + # destinations are installed: reporting routes through emit, so + # firing mid-build would recurse into configure. + previous = self.destinations + deferred: list[tuple[str, BaseException, dict[str, Any]]] = [] + + def deferred_report( + operation: str, exc: BaseException, **fields: Any + ) -> None: + deferred.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, + build_on_failure=deferred_report, + ) + ) except BaseException as exc: - failures.append((destination_name, exc)) + deferred_report( + "logging.destination_config", + exc, + destination=destination_name, + ) if not destinations: destinations.append(self._stdout_destination()) - failures.append( - ( - "stdout_fallback", - RuntimeError( - "No configured observability log destination " - "initialized; falling back to stdout." - ), - ) + deferred_report( + "logging.destination_config", + RuntimeError( + "No configured observability log destination " + "initialized; falling back to stdout." + ), + destination="stdout_fallback", ) self.destinations = destinations + # The failure ledger is keyed by id(); clear it with the swap so + # stale entries cannot attach to a new destination via id reuse. + self._consecutive_failures.clear() self.configured = True - for destination_name, exc in failures: - self.on_failure( - "logging.destination_config", - exc, - destination=destination_name, - ) + # Close the replaced destinations only after the new ones are + # installed, so their close-phase failure reports (which route + # through emit) still have a sink. + self._close_destinations(previous) + for operation, exc, fields in deferred: + self.on_failure(operation, exc, **fields) + for warning in getattr(self.config, "config_warnings", ()): + self.on_failure("logging.profile_config", ValueError(warning)) def emit( self, @@ -96,6 +125,36 @@ def emit( for destination in tripped: self._disable_destination(destination) + def close(self, deadline_seconds: float | None = None) -> None: + """Close every destination that supports closing, best-effort.""" + self._close_destinations(self.destinations, deadline_seconds) + + def _close_destinations( + self, + destinations: list[LogDestination], + deadline_seconds: float | None = None, + ) -> None: + # One deadline covers the whole batch: each close gets whatever + # budget the earlier ones left, so N stuck destinations cannot + # take N times the budget. None means each destination applies + # its own default (a reconfigure, not a bounded shutdown). + deadline = ( + None + if deadline_seconds is None + else time.monotonic() + max(0.0, deadline_seconds) + ) + for destination in destinations: + remaining = ( + None + if deadline is None + else max(0.0, deadline - time.monotonic()) + ) + close_destination( + destination, + on_failure=self.on_failure, + deadline_seconds=remaining, + ) + def _ensure_destinations(self) -> list[LogDestination]: if not self.configured: try: @@ -124,21 +183,41 @@ def _disable_destination(self, destination: LogDestination) -> None: if not self.destinations: self.destinations.append(self._stdout_destination()) - def _build_destination(self, destination_name: str) -> LogDestination: - normalized = destination_name.strip().lower().replace("-", "_") - if normalized == "stdout": - return self._stdout_destination() - if normalized in {"google", "google_cloud", "google_cloud_logging"}: - return GoogleCloudLoggingDestination( - project=self.config.google_cloud_project, - log_name=self.config.google_cloud_log_name, + def _build_destination( + self, + destination_name: str, + *, + build_on_failure: Callable[..., None], + ) -> LogDestination: + strategy = destination_strategy(destination_name) + if strategy is None: + raise ValueError( + f"Unknown observability log destination: {destination_name}" ) - raise ValueError( - f"Unknown observability log destination: {destination_name}" + destination = strategy.factory( + config=self.config, + loggers=self.loggers, + serializer=self.serializer, + on_failure=build_on_failure, ) + if strategy.transport == "remote": + # No remote strategy may ever write on a request thread. + destination = QueuedLogDestination( + inner=destination, + on_failure=self.on_failure, + maxsize=self.config.log_queue_maxsize, + close_timeout_seconds=( + self.config.log_queue_close_timeout_seconds + ), + ) + return destination def _stdout_destination(self) -> StdoutJsonDestination: - return StdoutJsonDestination( + # The fail-open fallback: built directly (registry-free) and + # with reporting suppressed, because this can run from inside a + # failure-reporting path where another report would recurse. + return build_stdout_destination( + config=self.config, loggers=self.loggers, serializer=self.serializer, ) diff --git a/policyengine_observability/destinations/queued.py b/policyengine_observability/destinations/queued.py new file mode 100644 index 0000000..1b5f30f --- /dev/null +++ b/policyengine_observability/destinations/queued.py @@ -0,0 +1,300 @@ +"""Bounded, best-effort background delivery for remote log destinations. + +``QueuedLogDestination`` wraps any ``LogDestination`` so the caller's +thread only ever enqueues: a stdlib ``logging.handlers.QueueListener`` +thread performs the actual writes, so a degraded sink can never stall a +request. Delivery is best-effort by contract — the stdout sibling +destination is the durable record — so a full queue drops the newest +record and counts it, and there is deliberately no circuit breaker, +retry queue, or recovery machinery in this component. + +Mutable state census (any addition needs design review): + +1. ``_queue`` — thread-safe by construction (``queue.Queue``). +2. ``_listener``— started once in ``__init__``, stopped once in ``close``. +3. ``_drops`` — best-effort drop counter with its throttle state. +4. ``_closed`` — one-way flag flipped by ``close``. + +(The handler's write-failure counter is confined to the listener +thread, so it is not shared mutable state.) + +Accepted races, all bounded and within the best-effort contract: + +- An ``emit`` that passes the ``_closed`` check while ``close`` runs can + lose that one record uncounted. +- When ``close`` times out, the daemon listener thread is abandoned; it + keeps draining in the background until process exit. +- A process that forks after construction (e.g. gunicorn ``--preload``) + inherits a dead listener; records drop with ``reason="full"`` until + the child calls ``restart_observability()`` from a post-fork hook. + There are deliberately no fork hooks or pid checks here. +""" + +from __future__ import annotations + +import atexit +import queue as queue_module +import time +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime +from logging.handlers import QueueListener +from typing import Any + +from ..config import ( + DEFAULT_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS, + DEFAULT_LOG_QUEUE_MAXSIZE, +) +from .base import ( + LogDestination, + accepts_keyword, + clamped, + close_destination, + normalize_payload, + safe_report, +) + +MIN_QUEUE_MAXSIZE = 10 +MAX_QUEUE_MAXSIZE = 100_000 +MIN_CLOSE_TIMEOUT_SECONDS = 0.0 +MAX_CLOSE_TIMEOUT_SECONDS = 30.0 +DROP_REPORT_INTERVAL = 100 +WRITE_FAILURE_REPORT_INTERVAL = 100 + + +@dataclass(frozen=True, slots=True) +class _QueuedRecord: + payload: Any + log_type: str + severity: str + enqueued_at: datetime + + +class _ThrottledCounter: + """Count occurrences; say when one should be reported. + + The shared throttle policy for failure-path reporting: the first + occurrence always reports, then every ``interval``-th, so a + persistent problem stays visible without flooding the internal-error + channel. + """ + + __slots__ = ("count", "interval") + + def __init__(self, interval: int) -> None: + self.interval = max(1, int(interval)) + self.count = 0 + + def tick(self) -> int | None: + """Increment; return the count when this occurrence reports.""" + self.count += 1 + if self.count == 1 or self.count % self.interval == 0: + return self.count + return None + + +class _QueuedRecordHandler: + """Duck-typed QueueListener handler: only ``handle`` is ever called. + + The failure counter is confined to the listener thread. A write + failure must never kill the listener, so everything below the emit + is guarded; ``BaseException`` is deliberately not caught (swallowing + ``SystemExit`` on a worker thread is worse than losing the queue — + the consequences of a dead listener are bounded to counted drops). + """ + + def __init__( + self, + inner: LogDestination, + on_failure: Callable[..., None], + *, + forward_timestamp: bool, + report_interval: int = WRITE_FAILURE_REPORT_INTERVAL, + ) -> None: + self.inner = inner + self.on_failure = on_failure + self.forward_timestamp = forward_timestamp + self.failures = _ThrottledCounter(report_interval) + + def handle(self, record: _QueuedRecord) -> None: + try: + if self.forward_timestamp: + self.inner.emit( + record.payload, + log_type=record.log_type, + severity=record.severity, + timestamp=record.enqueued_at, + ) + else: + self.inner.emit( + record.payload, + log_type=record.log_type, + severity=record.severity, + ) + except Exception as exc: + count = self.failures.tick() + if count is None: + return + safe_report( + self.on_failure, + "logging.queue_write", + exc, + destination=getattr(self.inner, "name", None), + log_type=record.log_type, + write_failures_total=count, + ) + + +class _BoundedQueueListener(QueueListener): + """QueueListener whose stop can be given a hard deadline. + + The stdlib ``stop()`` enqueues its sentinel with ``put_nowait`` + (which raises on a jammed bounded queue) and then joins the worker + thread without a timeout. Here one monotonic deadline covers both + the blocking sentinel put and the join; on expiry the daemon thread + is abandoned and ``False`` is returned. + """ + + def stop(self, timeout: float | None = None) -> bool: + thread = self._thread + if thread is None: + return True + if timeout is None: + super().stop() + return True + deadline = time.monotonic() + max(0.0, timeout) + try: + self.queue.put( + self._sentinel, + timeout=max(0.0, deadline - time.monotonic()), + ) + except queue_module.Full: + pass + thread.join(max(0.0, deadline - time.monotonic())) + stopped = not thread.is_alive() + self._thread = None + return stopped + + +class QueuedLogDestination: + def __init__( + self, + *, + inner: LogDestination, + on_failure: Callable[..., None], + maxsize: float = DEFAULT_LOG_QUEUE_MAXSIZE, + close_timeout_seconds: float = DEFAULT_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS, + drop_report_interval: int = DROP_REPORT_INTERVAL, + ) -> None: + self.inner = inner + self.on_failure = on_failure + self.name = f"queued_{getattr(inner, 'name', 'destination')}" + self.maxsize = int( + clamped( + maxsize, + low=MIN_QUEUE_MAXSIZE, + high=MAX_QUEUE_MAXSIZE, + default=DEFAULT_LOG_QUEUE_MAXSIZE, + ) + ) + self.close_timeout_seconds = clamped( + close_timeout_seconds, + low=MIN_CLOSE_TIMEOUT_SECONDS, + high=MAX_CLOSE_TIMEOUT_SECONDS, + default=DEFAULT_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS, + ) + self._queue: queue_module.Queue[_QueuedRecord | None] = ( + queue_module.Queue(self.maxsize) + ) + self._listener = _BoundedQueueListener( + self._queue, + _QueuedRecordHandler( + inner, + on_failure, + forward_timestamp=accepts_keyword(inner.emit, "timestamp"), + ), + ) + self._drops = _ThrottledCounter(drop_report_interval) + self._closed = False + # Construction happens at configure time on the startup thread, + # never lazily on a request thread. + self._listener.start() + atexit.register(self.close) + + def emit( + self, + payload: dict[str, Any], + *, + log_type: str, + severity: str, + ) -> None: + try: + if self._closed: + self._record_drop("closed", log_type) + return + record = _QueuedRecord( + # Snapshot now: callers keep mutating nested structures + # after emit returns, and the write happens later on the + # listener thread. The enqueue time becomes the entry + # timestamp so delayed writes keep event time. + payload=normalize_payload(payload), + log_type=log_type, + severity=severity, + enqueued_at=datetime.now(UTC), + ) + try: + self._queue.put_nowait(record) + except queue_module.Full: + self._record_drop("full", log_type) + except Exception as exc: + self._record_drop("exception", log_type, exc=exc) + + def close(self, deadline_seconds: float | None = None) -> None: + if self._closed: + return + self._closed = True + atexit.unregister(self.close) + deadline = clamped( + deadline_seconds, + low=MIN_CLOSE_TIMEOUT_SECONDS, + high=MAX_CLOSE_TIMEOUT_SECONDS, + default=self.close_timeout_seconds, + ) + drained = self._listener.stop(timeout=deadline) + if not drained: + safe_report( + self.on_failure, + "logging.queue_close_timeout", + TimeoutError( + "Observability log queue did not drain before " + "the close deadline; remaining records are lost." + ), + destination=self.name, + deadline_seconds=deadline, + pending_records=self._queue.qsize(), + ) + # The abandoned listener may still be mid-write; leave the + # inner destination alone rather than closing it underneath + # an active write. + return + close_destination(self.inner, on_failure=self.on_failure) + + def _record_drop( + self, + reason: str, + log_type: str, + exc: BaseException | None = None, + ) -> None: + count = self._drops.tick() + if count is None: + return + safe_report( + self.on_failure, + "logging.queue_drop", + exc or RuntimeError("Observability log queue dropped a record."), + destination=self.name, + log_type=log_type, + reason=reason, + dropped_total=count, + queue_maxsize=self.maxsize, + ) diff --git a/policyengine_observability/destinations/registry.py b/policyengine_observability/destinations/registry.py new file mode 100644 index 0000000..25b9521 --- /dev/null +++ b/policyengine_observability/destinations/registry.py @@ -0,0 +1,57 @@ +"""Named destination strategies. + +Backend modules register themselves here so the manager can build +destinations without knowing any backend: an ``inline`` strategy writes +synchronously on the caller's thread (stdout — the durable, dependency- +free record), while every ``remote`` strategy is wrapped in the bounded +queue transport, so no remote write can ever run on a request thread. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Literal + +from .base import LogDestination, normalize_name + +# Factories are called with keyword arguments (config, loggers, +# serializer, on_failure) and may ignore what they do not need. The +# on_failure callable is for construction-time reporting only; reports +# made through it may be deferred until the build completes. +DestinationFactory = Callable[..., LogDestination] + + +@dataclass(frozen=True) +class DestinationStrategy: + factory: DestinationFactory + transport: Literal["inline", "remote"] + # Config attribute names that must resolve truthy for the strategy + # to be usable. Profiles naming this strategy downgrade gracefully + # (with a warning) when a requirement is missing, instead of failing + # at construction and falling back with a config-failure report. + required_config: tuple[str, ...] = () + + +_STRATEGIES: dict[str, DestinationStrategy] = {} + + +def register_destination( + name: str, + factory: DestinationFactory, + *, + transport: Literal["inline", "remote"], + aliases: tuple[str, ...] = (), + required_config: tuple[str, ...] = (), +) -> None: + strategy = DestinationStrategy( + factory=factory, + transport=transport, + required_config=required_config, + ) + for key in (name, *aliases): + _STRATEGIES[normalize_name(key)] = strategy + + +def destination_strategy(name: str) -> DestinationStrategy | None: + return _STRATEGIES.get(normalize_name(name)) diff --git a/policyengine_observability/destinations/stdout.py b/policyengine_observability/destinations/stdout.py index 5bc01b1..2f562f5 100644 --- a/policyengine_observability/destinations/stdout.py +++ b/policyengine_observability/destinations/stdout.py @@ -4,7 +4,72 @@ from collections.abc import Callable, Mapping from typing import Any -from .base import normalize_payload +from .base import normalize_name, normalize_payload, safe_report +from .registry import register_destination + +# A stdout formatter shapes the normalized payload into the JSON line a +# platform's log agent expects. Formatters are registered by name so +# backend modules can contribute agent-native shapes without the core +# knowing about any backend; "plain" is the built-in default. Factories +# receive the ObservabilityConfig so a formatter can close over settings +# it needs (duck-typed to keep this module config-agnostic). +StdoutFormatter = Callable[..., dict[str, Any]] +StdoutFormatterFactory = Callable[[Any], StdoutFormatter] + +_FORMATTER_FACTORIES: dict[str, StdoutFormatterFactory] = {} + + +def register_stdout_formatter( + name: str, factory: StdoutFormatterFactory +) -> None: + _FORMATTER_FACTORIES[normalize_name(name)] = factory + + +def resolve_stdout_formatter( + config: Any, + on_failure: Callable[..., None] | None = None, +) -> StdoutFormatter: + raw = getattr(config, "stdout_format", None) or "plain" + factory = _FORMATTER_FACTORIES.get(normalize_name(raw)) + if factory is None: + # Falling back must not lose the record, but a typo'd format + # name should not pass silently either — the agent-native shape + # it named would just quietly never appear. + if on_failure is not None: + safe_report( + on_failure, + "logging.stdout_format", + ValueError(f"Unknown stdout format {raw!r}; using plain."), + ) + factory = _FORMATTER_FACTORIES["plain"] + try: + return factory(config) + except Exception as exc: + # Stdout is the fail-open record, and this resolver also runs on + # the manager's last-resort fallback path: a registered factory + # that raises must degrade to the built-in plain formatter (not + # the registry entry, which could be the broken one), never + # break configure or startup. + if on_failure is not None: + safe_report( + on_failure, + "logging.stdout_format", + exc, + stdout_format=raw, + ) + return _plain_formatter_factory(config) + + +def _plain_formatter_factory(config: Any) -> StdoutFormatter: + def format_plain( + payload: dict[str, Any], *, log_type: str, severity: str + ) -> dict[str, Any]: + return payload + + return format_plain + + +register_stdout_formatter("plain", _plain_formatter_factory) class StdoutJsonDestination: @@ -15,9 +80,11 @@ def __init__( *, loggers: Mapping[str, logging.Logger], serializer: Callable[[dict[str, Any]], str], + formatter: StdoutFormatter | None = None, ) -> None: self.loggers = loggers self.serializer = serializer + self.formatter = formatter or _plain_formatter_factory(None) def emit( self, @@ -26,11 +93,46 @@ def emit( log_type: str, severity: str, ) -> None: + normalized = normalize_payload(payload) + try: + # Formatters receive (and may mutate) the private normalized + # copy. Stdout is the fallback sink, so a broken formatter + # degrades to the unformatted line rather than losing the + # record or tripping the breaker. + formatted = self.formatter( + normalized, log_type=log_type, severity=severity + ) + except Exception: + formatted = normalized + message = self.serializer(formatted) logger = self.loggers.get(log_type) or self.loggers["event"] - message = self.serializer(normalize_payload(payload)) if severity in {"ERROR", "CRITICAL"}: logger.error(message) elif severity == "WARNING": logger.warning(message) else: logger.info(message) + + +def build_stdout_destination( + *, + config: Any, + loggers: Any, + serializer: Any, + on_failure: Callable[..., None] | None = None, + **_: Any, +) -> StdoutJsonDestination: + """The one place a configured stdout destination is assembled. + + Used both as the registered ``stdout`` strategy factory and by the + manager's fail-open fallback, so formatter resolution can never + diverge between the two paths. + """ + return StdoutJsonDestination( + loggers=loggers, + serializer=serializer, + formatter=resolve_stdout_formatter(config, on_failure=on_failure), + ) + + +register_destination("stdout", build_stdout_destination, transport="inline") diff --git a/policyengine_observability/runtime.py b/policyengine_observability/runtime.py index 8fbda49..051f79a 100644 --- a/policyengine_observability/runtime.py +++ b/policyengine_observability/runtime.py @@ -24,6 +24,7 @@ _metric_attrs, ) from .destinations import LogDestinationManager +from .destinations.base import clamped from .logging import configure_plain_logger from .segments import coerce_segment_name @@ -1071,6 +1072,12 @@ def instrument_httpx(self) -> None: self.log_observability_failure("httpx.auto_instrument", exc) def shutdown(self) -> None: + budget = clamped( + self.config.shutdown_timeout_seconds, + low=0.0, + high=60.0, + default=ObservabilityConfig.shutdown_timeout_seconds, + ) providers = [ ("trace", self.tracer_provider), ("metrics", self.meter_provider), @@ -1080,8 +1087,20 @@ def shutdown(self) -> None: for name, provider in providers if provider is not None ] + # Destination close is inherently deadline-bounded, so it runs + # inline and first, with a deadline that leaves room for the + # provider flush when there is one. Everything below fits inside + # the one shutdown budget by construction. + started = time.monotonic() + try: + self.log_destination_manager.close( + budget / 2 if providers else budget + ) + except BaseException as exc: + self.log_observability_failure("logging.destination_close", exc) if not providers: return + remaining = max(0.0, budget - (time.monotonic() - started)) def flush() -> None: for name, provider in providers: @@ -1099,17 +1118,33 @@ def flush() -> None: daemon=True, ) thread.start() - thread.join(timeout=self.config.shutdown_timeout_seconds) + thread.join(timeout=remaining) if thread.is_alive(): self.log_observability_failure( "otel.shutdown_timeout", TimeoutError("OpenTelemetry shutdown timed out."), - timeout_seconds=self.config.shutdown_timeout_seconds, + timeout_seconds=remaining, ) def shutdown_tracing(self) -> None: self.shutdown() + def restart_log_destinations(self) -> None: + """Close and rebuild log destinations from configuration. + + Call ONLY from single-threaded lifecycle moments — a + post-snapshot-restore hook, a post-fork hook, before serving + traffic. There is deliberately no locking here: under that + contract there is no concurrency, and a violated contract costs + at most a counted drop into a closing destination. + + A no-op when observability is disabled, mirroring configure(): + the kill switch must hold across forks and snapshot restores. + """ + if not self.enabled: + return + self.log_destination_manager.configure() + def log_observability_failure( self, operation: str, diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..dbc84a1 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import pytest +from fakes import RecordingDestination + +from policyengine_observability.destinations.registry import ( + _STRATEGIES, + register_destination, +) + + +@pytest.fixture +def fake_remote_strategy(): + """Register a fake remote strategy; yields its destination name.""" + register_destination( + "fake-remote", + lambda **kwargs: RecordingDestination(), + transport="remote", + ) + yield "fake_remote" + _STRATEGIES.pop("fake_remote", None) diff --git a/tests/fakes.py b/tests/fakes.py new file mode 100644 index 0000000..b4ea394 --- /dev/null +++ b/tests/fakes.py @@ -0,0 +1,121 @@ +"""Shared destination fakes and manager builders for the test suite.""" + +from __future__ import annotations + +import json +import logging +import threading + +from policyengine_observability.config import ObservabilityConfig +from policyengine_observability.destinations.manager import ( + LogDestinationManager, +) + + +class RecordingLogger: + """Captures level-routed serialized lines like a stdlib logger.""" + + 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)) + + +class RecordingDestination: + name = "recording" + + def __init__(self) -> None: + self.calls = [] + + @property + def payloads(self): + return [payload for payload, *_ in self.calls] + + def emit(self, payload, *, log_type, severity, timestamp=None) -> None: + self.calls.append((payload, log_type, severity, timestamp)) + + +class ClosableRecordingDestination(RecordingDestination): + """A recording destination with a zero-argument duck-typed close.""" + + def __init__(self) -> None: + super().__init__() + self.closed = 0 + + def close(self) -> None: + self.closed += 1 + + +class TimestampBlindDestination: + name = "timestamp-blind" + + def __init__(self) -> None: + self.calls = [] + + def emit(self, payload, *, log_type, severity) -> None: + self.calls.append((payload, log_type, severity)) + + +class BlockingDestination: + """Blocks inside emit until released; sequenced via Events.""" + + name = "blocking" + + def __init__(self) -> None: + self.entered = threading.Event() + self.release = threading.Event() + self.calls = 0 + self.closed = 0 + + def emit(self, payload, *, log_type, severity, timestamp=None) -> None: + self.calls += 1 + self.entered.set() + self.release.wait(10) + + def close(self) -> None: + self.closed += 1 + + +class FailingDestination: + """Raises from emit — always, or only for the first ``fail_first``.""" + + name = "failing" + + def __init__(self, fail_first: int | None = None) -> None: + self.calls = 0 + self.fail_first = fail_first + self.delivered = [] + + def emit(self, payload, *, log_type, severity, timestamp=None) -> None: + self.calls += 1 + if self.fail_first is None or self.calls <= self.fail_first: + raise RuntimeError("emit failed") + self.delivered.append(payload) + + +def make_manager(config=None, *, loggers=None, destinations=None): + """A manager with a failure-capturing on_failure. + + Returns ``(manager, failures)`` where each failure is the + ``(operation, exception, fields)`` triple the manager reported. + """ + failures = [] + manager = LogDestinationManager( + config=config or ObservabilityConfig(), + loggers=loggers or {"event": logging.getLogger("test-manager")}, + serializer=json.dumps, + on_failure=lambda operation, exc, **fields: failures.append( + (operation, exc, fields) + ), + ) + if destinations is not None: + manager.destinations = list(destinations) + manager.configured = True + return manager, failures diff --git a/tests/test_destinations.py b/tests/test_destinations.py index 10c1941..ae51fb5 100644 --- a/tests/test_destinations.py +++ b/tests/test_destinations.py @@ -1,10 +1,31 @@ from __future__ import annotations +import json +import math + +import pytest +from fakes import ( + ClosableRecordingDestination, + FailingDestination, + RecordingDestination, + RecordingLogger, + make_manager, +) + +from policyengine_observability.config import ObservabilityConfig from policyengine_observability.destinations import ( GoogleCloudLoggingDestination, google_cloud_logging, normalize_payload, ) +from policyengine_observability.destinations.base import ( + accepts_keyword, + clamped, +) +from policyengine_observability.destinations.stdout import ( + StdoutJsonDestination, + resolve_stdout_formatter, +) class Unprintable: @@ -20,17 +41,90 @@ def log_struct(self, payload, **kwargs) -> None: self.calls.append((payload, kwargs)) -class FakeClient: +class FakeGapicApi: def __init__(self) -> None: + self.calls = [] + + def write_log_entries(self, *args, **kwargs) -> None: + self.calls.append((args, kwargs)) + + +class FakeLoggingApi: + def __init__(self) -> None: + self._gapic_api = FakeGapicApi() + + +class FakeClient: + def __init__(self, *, gapic: bool = False) -> None: self.project = "resolved-project" self.fake_logger = FakeLogger() self.log_names = [] + if gapic: + self.logging_api = FakeLoggingApi() def logger(self, log_name: str) -> FakeLogger: self.log_names.append(log_name) return self.fake_logger +def _google_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, + ) + + +@pytest.mark.parametrize( + ("value", "expected"), + [ + (5.0, 5.0), + ("5", 5.0), + (0, 0.5), + (-3, 0.5), + (1000, 60.0), + (float("inf"), 10.0), + (float("nan"), 10.0), + (None, 10.0), + ("garbage", 10.0), + ], +) +def test_clamped_bounds_and_rejects_non_finite(value, expected) -> None: + result = clamped(value, low=0.5, high=60.0, default=10.0) + + assert result == expected + assert math.isfinite(result) + + +def test_accepts_keyword_covers_named_var_keyword_and_uninspectable() -> None: + def named(payload, *, timestamp=None): + pass + + def var_keyword(payload, **kwargs): + pass + + def blind(payload): + pass + + assert accepts_keyword(named, "timestamp") is True + assert accepts_keyword(var_keyword, "timestamp") is True + assert accepts_keyword(blind, "timestamp") is False + # Builtins without introspectable signatures degrade to False + # instead of raising at construction time. + assert accepts_keyword(min, "timestamp") is False + + def test_normalize_payload_recursively_stringifies_unsafe_values() -> None: normalized = normalize_payload( { @@ -102,53 +196,355 @@ def test_google_destination_writes_structured_log_with_bounded_labels( assert "path" not in kwargs["labels"] -# ── Destination circuit breaker ────────────────────────────────────────── +def test_google_destination_bounds_gapic_writes(monkeypatch) -> None: + client = FakeClient(gapic=True) + destination = _google_destination( + monkeypatch, client, write_timeout_seconds=5.0 + ) + # The write path under log_struct funnels through this method; the + # rebinding must inject the bounded retry and per-call timeout. + client.logging_api._gapic_api.write_log_entries(request="sentinel") -class FlakyDestination: - name = "flaky" + assert destination.write_timeout_seconds == 5.0 + ((args, kwargs),) = client.logging_api._gapic_api.calls + assert kwargs["request"] == "sentinel" + assert kwargs["timeout"] == 5.0 + assert kwargs["retry"].timeout == 5.0 - def __init__(self, fail_first: int | None = None) -> None: - self.calls = 0 - self.fail_first = fail_first - def emit(self, payload, *, log_type, severity) -> None: - self.calls += 1 - if self.fail_first is None or self.calls <= self.fail_first: - raise RuntimeError("emit failed") +def test_google_destination_clamps_write_timeout(monkeypatch) -> None: + destination = _google_destination( + monkeypatch, FakeClient(gapic=True), write_timeout_seconds=0.0 + ) + assert destination.write_timeout_seconds == 0.5 -class RecordingDestination: - name = "recording" - def __init__(self) -> None: - self.payloads = [] +def test_google_destination_without_gapic_transport_still_works( + monkeypatch, +) -> None: + client = FakeClient() - def emit(self, payload, *, log_type, severity) -> None: - self.payloads.append(payload) + destination = _google_destination(monkeypatch, client) + destination.emit({"event": "x"}, log_type="event", severity="INFO") + assert len(client.fake_logger.calls) == 1 -def _manager(destinations): - import json - import logging - from policyengine_observability.config import ObservabilityConfig - from policyengine_observability.destinations.manager import ( - LogDestinationManager, +def test_google_destination_forwards_enqueue_timestamp(monkeypatch) -> None: + from datetime import UTC, datetime + + client = FakeClient() + destination = _google_destination(monkeypatch, client) + stamp = datetime(2026, 7, 8, 12, 0, 0, tzinfo=UTC) + + destination.emit( + {"event": "x"}, log_type="event", severity="INFO", timestamp=stamp ) + destination.emit({"event": "y"}, log_type="event", severity="INFO") - failures = [] - manager = LogDestinationManager( - config=ObservabilityConfig(), - loggers={"event": logging.getLogger("test-destinations")}, + (_, stamped_kwargs), (_, plain_kwargs) = client.fake_logger.calls + assert stamped_kwargs["timestamp"] is stamp + assert "timestamp" not in plain_kwargs + + +def test_google_destination_close_closes_client(monkeypatch) -> None: + class ClosableFakeClient(FakeClient): + def __init__(self) -> None: + super().__init__() + self.closed = 0 + + def close(self) -> None: + self.closed += 1 + + client = ClosableFakeClient() + destination = _google_destination(monkeypatch, client) + + destination.close() + + assert client.closed == 1 + + +def test_google_destination_close_tolerates_closeless_client( + monkeypatch, +) -> None: + destination = _google_destination(monkeypatch, FakeClient()) + + destination.close() # FakeClient has no close; must be a no-op + + +def test_google_destination_suppresses_instrumentation_entry( + monkeypatch, +) -> None: + logging_v2 = pytest.importorskip("google.cloud.logging_v2") + monkeypatch.setattr( + logging_v2, "_instrumentation_emitted", False, raising=False + ) + + _google_destination(monkeypatch, FakeClient()) + + assert logging_v2._instrumentation_emitted is True + + +def test_google_factory_reads_write_timeout_env(monkeypatch) -> None: + captured = {} + + class StubDestination: + def __init__(self, **kwargs) -> None: + captured.update(kwargs) + + monkeypatch.setattr( + google_cloud_logging, "GoogleCloudLoggingDestination", StubDestination + ) + monkeypatch.setenv("OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS", "2.5") + + from policyengine_observability.destinations.registry import ( + destination_strategy, + ) + + destination_strategy("google_cloud_logging").factory( + config=ObservabilityConfig(google_cloud_project="proj"), + loggers={}, serializer=json.dumps, + ) + + assert captured["project"] == "proj" + assert captured["write_timeout_seconds"] == 2.5 + + +def test_google_factory_write_timeout_defaults_without_env( + monkeypatch, +) -> None: + captured = {} + + class StubDestination: + def __init__(self, **kwargs) -> None: + captured.update(kwargs) + + monkeypatch.setattr( + google_cloud_logging, "GoogleCloudLoggingDestination", StubDestination + ) + monkeypatch.delenv( + "OBSERVABILITY_GOOGLE_WRITE_TIMEOUT_SECONDS", raising=False + ) + + from policyengine_observability.destinations.registry import ( + destination_strategy, + ) + + destination_strategy("google_cloud_logging").factory( + config=ObservabilityConfig(google_cloud_project="proj"), + loggers={}, + serializer=json.dumps, + ) + + assert captured["write_timeout_seconds"] == 10.0 + + +# ── Stdout formatters ──────────────────────────────────────────────────── + + +def _stdout_destination(config=None, formatter=None): + logger = RecordingLogger() + if formatter is None and config is not None: + formatter = resolve_stdout_formatter(config) + destination = StdoutJsonDestination( + loggers={"event": logger}, + serializer=json.dumps, + formatter=formatter, + ) + return destination, logger + + +def _emitted_line(logger): + ((_, message),) = logger.lines + return json.loads(message) + + +def test_stdout_google_formatter_maps_agent_native_keys() -> None: + config = ObservabilityConfig( + stdout_format="google", google_cloud_project="proj" + ) + destination, logger = _stdout_destination(config) + + destination.emit( + { + "schema_version": "policyengine.observability.event.v1", + "service_name": "svc", + "event": "x", + "trace_id": "abc123", + "span_id": 456, + }, + log_type="event", + severity="error", + ) + + line = _emitted_line(logger) + assert line["severity"] == "ERROR" + assert line["logging.googleapis.com/trace"] == ( + "projects/proj/traces/abc123" + ) + assert line["logging.googleapis.com/spanId"] == "456" + assert line["logging.googleapis.com/labels"] == { + "log_type": "event", + "service_name": "svc", + "schema_version": "policyengine.observability.event.v1", + } + assert "time" not in line + assert line["event"] == "x" + + +def test_stdout_google_formatter_omits_trace_without_project() -> None: + config = ObservabilityConfig(stdout_format="google") + destination, logger = _stdout_destination(config) + + destination.emit( + {"event": "x", "trace_id": "abc"}, log_type="event", severity="INFO" + ) + + line = _emitted_line(logger) + assert "logging.googleapis.com/trace" not in line + + +def test_stdout_unknown_format_falls_back_to_plain() -> None: + config = ObservabilityConfig(stdout_format=" GoOgLeX ") + destination, logger = _stdout_destination(config) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + assert _emitted_line(logger) == {"event": "x"} + + +def test_stdout_unknown_format_reports_when_channel_available() -> None: + failures = [] + formatter = resolve_stdout_formatter( + ObservabilityConfig(stdout_format="agent-natve"), on_failure=lambda operation, exc, **fields: failures.append( - (operation, fields) + (operation, str(exc)) ), ) - manager.destinations = list(destinations) - manager.configured = True - return manager, failures + + formatted = formatter({"event": "x"}, log_type="event", severity="INFO") + + assert formatted == {"event": "x"} + assert len(failures) == 1 + assert failures[0][0] == "logging.stdout_format" + assert "agent-natve" in failures[0][1] + + +def test_stdout_format_name_is_normalized() -> None: + config = ObservabilityConfig(stdout_format=" GOOGLE ") + destination, logger = _stdout_destination(config) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + assert _emitted_line(logger)["severity"] == "INFO" + + +def test_stdout_broken_formatter_degrades_to_unformatted() -> None: + def broken(payload, *, log_type, severity): + raise RuntimeError("formatter bug") + + destination, logger = _stdout_destination(formatter=broken) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + assert _emitted_line(logger) == {"event": "x"} + + +def test_stdout_broken_formatter_factory_degrades_to_plain() -> None: + from policyengine_observability.destinations.stdout import ( + _FORMATTER_FACTORIES, + register_stdout_formatter, + ) + + def broken_factory(config): + raise RuntimeError("factory bug") + + register_stdout_formatter("broken-test", broken_factory) + try: + failures = [] + formatter = resolve_stdout_formatter( + ObservabilityConfig(stdout_format="broken-test"), + on_failure=lambda operation, exc, **fields: failures.append( + (operation, fields) + ), + ) + finally: + _FORMATTER_FACTORIES.pop("broken_test", None) + + formatted = formatter({"event": "x"}, log_type="event", severity="INFO") + + assert formatted == {"event": "x"} + assert len(failures) == 1 + assert failures[0][0] == "logging.stdout_format" + assert failures[0][1]["stdout_format"] == "broken-test" + + +def test_configure_fallback_survives_broken_formatter_factory() -> None: + """The crash path: no destination builds, so the manager's + last-resort stdout fallback resolves the same broken formatter — + configure must stay fail-open and emit must still write plain.""" + from policyengine_observability.destinations.stdout import ( + _FORMATTER_FACTORIES, + register_stdout_formatter, + ) + + def broken_factory(config): + raise RuntimeError("factory bug") + + register_stdout_formatter("broken-test", broken_factory) + try: + config = ObservabilityConfig( + log_destinations=("nonexistent",), stdout_format="broken-test" + ) + logger = RecordingLogger() + manager, _failures = make_manager(config, loggers={"event": logger}) + + manager.configure() # raised before the resolver guard existed + manager.emit({"event": "x"}, log_type="event", severity="INFO") + finally: + _FORMATTER_FACTORIES.pop("broken_test", None) + + assert _emitted_line(logger) == {"event": "x", "severity": "INFO"} + + +def test_stdout_google_formatter_never_mutates_caller_payload() -> None: + config = ObservabilityConfig( + stdout_format="google", google_cloud_project="proj" + ) + destination, logger = _stdout_destination(config) + payload = {"event": "x", "trace_id": "abc"} + + destination.emit(payload, log_type="event", severity="INFO") + + assert payload == {"event": "x", "trace_id": "abc"} + + +def test_custom_stdout_formatter_registers_and_resolves() -> None: + from policyengine_observability.destinations.stdout import ( + _FORMATTER_FACTORIES, + register_stdout_formatter, + ) + + def factory(config): + return lambda payload, *, log_type, severity: {"wrapped": payload} + + register_stdout_formatter("custom-test", factory) + try: + # Hyphen/underscore variance is forgiven the same way it is for + # destination names. + config = ObservabilityConfig(stdout_format=" Custom_Test ") + destination, logger = _stdout_destination(config) + destination.emit({"event": "x"}, log_type="event", severity="INFO") + finally: + _FORMATTER_FACTORIES.pop("custom_test", None) + + assert _emitted_line(logger) == {"wrapped": {"event": "x"}} + + +# ── Destination circuit breaker and manager lifecycle ─────────────────── def test_destination_disabled_after_consecutive_emit_failures() -> None: @@ -156,9 +552,9 @@ def test_destination_disabled_after_consecutive_emit_failures() -> None: DESTINATION_FAILURE_LIMIT, ) - flaky = FlakyDestination() + flaky = FailingDestination() healthy = RecordingDestination() - manager, failures = _manager([flaky, healthy]) + manager, failures = make_manager(destinations=[flaky, healthy]) for _ in range(DESTINATION_FAILURE_LIMIT + 2): manager.emit({"event": "x"}, log_type="event", severity="INFO") @@ -166,7 +562,7 @@ def test_destination_disabled_after_consecutive_emit_failures() -> None: assert flaky.calls == DESTINATION_FAILURE_LIMIT assert flaky not in manager.destinations assert len(healthy.payloads) == DESTINATION_FAILURE_LIMIT + 2 - assert any(op == "logging.destination_disabled" for op, _ in failures) + assert any(op == "logging.destination_disabled" for op, *_ in failures) def test_emit_success_resets_the_failure_counter() -> None: @@ -174,8 +570,8 @@ def test_emit_success_resets_the_failure_counter() -> None: DESTINATION_FAILURE_LIMIT, ) - flaky = FlakyDestination(fail_first=DESTINATION_FAILURE_LIMIT - 1) - manager, failures = _manager([flaky]) + flaky = FailingDestination(fail_first=DESTINATION_FAILURE_LIMIT - 1) + manager, failures = make_manager(destinations=[flaky]) for _ in range(DESTINATION_FAILURE_LIMIT + 2): manager.emit({"event": "x"}, log_type="event", severity="INFO") @@ -183,7 +579,7 @@ def test_emit_success_resets_the_failure_counter() -> None: assert flaky in manager.destinations counts = [ fields["consecutive_failures"] - for op, fields in failures + for op, _exc, fields in failures if op == "logging.destination_emit" ] assert max(counts) == DESTINATION_FAILURE_LIMIT - 1 @@ -193,12 +589,9 @@ def test_sole_disabled_destination_falls_back_to_stdout() -> None: from policyengine_observability.destinations.manager import ( DESTINATION_FAILURE_LIMIT, ) - from policyengine_observability.destinations.stdout import ( - StdoutJsonDestination, - ) - flaky = FlakyDestination() - manager, failures = _manager([flaky]) + flaky = FailingDestination() + manager, _failures = make_manager(destinations=[flaky]) for _ in range(DESTINATION_FAILURE_LIMIT + 1): manager.emit({"event": "x"}, log_type="event", severity="INFO") @@ -208,3 +601,90 @@ def test_sole_disabled_destination_falls_back_to_stdout() -> None: isinstance(destination, StdoutJsonDestination) for destination in manager.destinations ) + + +def test_manager_close_reports_failures_and_closes_the_rest() -> None: + class ExplodingClose(RecordingDestination): + def close(self) -> None: + raise RuntimeError("close failed") + + exploding = ExplodingClose() + closable = ClosableRecordingDestination() + manager, failures = make_manager(destinations=[exploding, closable]) + + manager.close(1.0) # must not raise + + assert closable.closed == 1 + assert any(op == "logging.destination_close" for op, *_ in failures) + + +def test_manager_close_accepts_zero_argument_close() -> None: + """A duck-typed close(self) works under every close path — the + deadline is passed only when the signature accepts it.""" + closable = ClosableRecordingDestination() + manager, failures = make_manager(destinations=[closable]) + + manager.close(1.0) + + assert closable.closed == 1 + assert failures == [] + + +def test_manager_close_shares_one_deadline_across_destinations() -> None: + import time + + deadlines = [] + + class SlowClose(RecordingDestination): + def close(self, deadline_seconds=None) -> None: + deadlines.append(deadline_seconds) + time.sleep(0.05) + + manager, _failures = make_manager(destinations=[SlowClose(), SlowClose()]) + + manager.close(1.0) + + first, second = deadlines + # The second destination only gets what the first one left, so N + # stuck destinations cannot take N times the budget. + assert first <= 1.0 + assert second <= first - 0.04 + + +def test_emit_falls_back_to_stdout_when_configure_crashes( + monkeypatch, +) -> None: + logger = RecordingLogger() + manager, failures = make_manager(loggers={"event": logger}) + + def broken_configure() -> None: + raise RuntimeError("configure exploded") + + monkeypatch.setattr(manager, "configure", broken_configure) + + manager.emit({"event": "x"}, log_type="event", severity="INFO") + + assert manager.configured is True + assert isinstance(manager.destinations[0], StdoutJsonDestination) + assert _emitted_line(logger) == {"event": "x", "severity": "INFO"} + assert any(op == "logging.destination_config" for op, *_ in failures) + + +def test_reconfigure_closes_previous_after_installing_new() -> None: + """Close-phase failure reports route through emit; the replaced + destinations must be closed only after the new ones are installed + so those reports still have a sink.""" + sink_states = [] + manager, _failures = make_manager() + + class CloseProbe(RecordingDestination): + def close(self) -> None: + sink_states.append(list(manager.destinations)) + + manager.destinations = [CloseProbe()] + manager.configured = True + + manager.configure() + + assert len(sink_states) == 1 + assert sink_states[0], "previous destination closed before new install" diff --git a/tests/test_google_credentials.py b/tests/test_google_credentials.py index c112ab0..4870c26 100644 --- a/tests/test_google_credentials.py +++ b/tests/test_google_credentials.py @@ -7,9 +7,11 @@ import pytest -from policyengine_observability import google_credentials -from policyengine_observability.destinations import google_cloud_logging -from policyengine_observability.google_credentials import ( +from policyengine_observability.destinations import ( + google_cloud_logging, + google_credentials, +) +from policyengine_observability.destinations.google_credentials import ( configure_google_application_credentials, load_google_credentials, ) @@ -96,7 +98,7 @@ def test_configure_google_application_credentials_fails_open_on_unexpected_error monkeypatch.delenv("GOOGLE_APPLICATION_CREDENTIALS", raising=False) monkeypatch.setenv("GCP_CREDENTIALS_JSON", '{"project_id":"test"}') monkeypatch.setattr( - "policyengine_observability.google_credentials.json.loads", + "policyengine_observability.destinations.google_credentials.json.loads", lambda _value: (_ for _ in ()).throw(RuntimeError("boom")), ) diff --git a/tests/test_log_profiles.py b/tests/test_log_profiles.py new file mode 100644 index 0000000..ddade7a --- /dev/null +++ b/tests/test_log_profiles.py @@ -0,0 +1,292 @@ +from __future__ import annotations + +import json + +import pytest +from fakes import RecordingLogger, make_manager + +from policyengine_observability.config import ObservabilityConfig +from policyengine_observability.destinations.queued import ( + QueuedLogDestination, +) + +PLATFORM_MARKERS = ( + "OBSERVABILITY_LOG_PROFILE", + "OBSERVABILITY_PLATFORM", + "K_SERVICE", + "MODAL_ENVIRONMENT", + "MODAL_TASK_ID", + "OBSERVABILITY_LOG_DESTINATIONS", + "OBSERVABILITY_STDOUT_FORMAT", + "OBSERVABILITY_GOOGLE_CLOUD_PROJECT", + "GOOGLE_CLOUD_PROJECT", + "GCP_PROJECT", + "GCLOUD_PROJECT", +) + + +@pytest.fixture(autouse=True) +def _clean_environment(monkeypatch): + for name in PLATFORM_MARKERS: + monkeypatch.delenv(name, raising=False) + + +def _from_env(monkeypatch, **env): + for name, value in env.items(): + monkeypatch.setenv(name, value) + return ObservabilityConfig.from_env(service_name="svc") + + +def test_explicit_gcp_agent_profile(monkeypatch) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE="gcp-agent") + + assert config.log_profile == "gcp-agent" + assert config.log_destinations == ("stdout",) + assert config.stdout_format == "google" + assert config.config_warnings == () + + +def test_profile_name_accepts_underscore_variant(monkeypatch) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE=" GCP_Agent ") + + assert config.log_profile == "gcp-agent" + assert config.config_warnings == () + + +def test_explicit_gcp_direct_profile_with_project(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="gcp-direct", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_profile == "gcp-direct" + assert config.log_destinations == ("stdout", "google_cloud_logging") + assert config.stdout_format == "plain" + assert config.config_warnings == () + + +def test_gcp_direct_without_project_downgrades_with_warning( + monkeypatch, +) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE="gcp-direct") + + assert config.log_profile == "plain-sync" + assert config.log_destinations == ("stdout",) + assert config.stdout_format == "plain" + assert len(config.config_warnings) == 1 + # The requirement comes from the strategy registration, and the + # warning names the actual profile — no hard-coded backend text. + assert "gcp-direct" in config.config_warnings[0] + assert "google_cloud_project" in config.config_warnings[0] + + +def test_explicit_plain_sync_profile_is_kill_switch(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="plain-sync", + OBSERVABILITY_PLATFORM="modal", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_profile == "plain-sync" + assert config.log_destinations == ("stdout",) + assert config.stdout_format == "plain" + + +def test_unknown_profile_falls_back_to_plain_sync_with_warning( + monkeypatch, +) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE="gcp-agnet") + + assert config.log_profile == "plain-sync" + assert config.log_destinations == ("stdout",) + assert any("gcp-agnet" in w for w in config.config_warnings) + + +def test_auto_detects_cloud_run_via_observability_platform( + monkeypatch, +) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_PLATFORM="google_cloud_run") + + assert config.log_profile == "gcp-agent" + assert config.stdout_format == "google" + + +def test_auto_detects_modal_via_observability_platform(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_PLATFORM="modal", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_profile == "gcp-direct" + assert config.log_destinations == ("stdout", "google_cloud_logging") + + +def test_auto_detects_cloud_run_via_k_service(monkeypatch) -> None: + config = _from_env(monkeypatch, K_SERVICE="household-api") + + assert config.log_profile == "gcp-agent" + + +def test_auto_detects_modal_via_task_marker(monkeypatch) -> None: + config = _from_env( + monkeypatch, + MODAL_TASK_ID="ta-123", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_profile == "gcp-direct" + + +def test_auto_detects_modal_via_environment_marker(monkeypatch) -> None: + config = _from_env( + monkeypatch, + MODAL_ENVIRONMENT="main", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_profile == "gcp-direct" + + +def test_auto_detected_modal_without_project_downgrades(monkeypatch) -> None: + """The deployment-realistic failure: Modal markers present but no + resolvable project must land on plain-sync with a warning, exactly + like the explicit profile.""" + config = _from_env(monkeypatch, MODAL_TASK_ID="ta-123") + + assert config.log_profile == "plain-sync" + assert config.log_destinations == ("stdout",) + assert any("google_cloud_project" in w for w in config.config_warnings) + + +def test_observability_platform_beats_generic_markers(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_PLATFORM="google_cloud_run", + MODAL_TASK_ID="ta-123", + ) + + assert config.log_profile == "gcp-agent" + + +def test_auto_without_markers_preserves_caller_defaults(monkeypatch) -> None: + config = ObservabilityConfig.from_env( + service_name="svc", + default_log_destinations=("stdout", "custom"), + ) + + assert config.log_profile == "auto" + assert config.log_destinations == ("stdout", "custom") + assert config.stdout_format == "plain" + assert config.config_warnings == () + + +def test_explicit_destination_env_overrides_profile(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="gcp-agent", + OBSERVABILITY_LOG_DESTINATIONS="stdout,google_cloud_logging", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + + assert config.log_destinations == ("stdout", "google_cloud_logging") + # The profile's other half still applies. + assert config.stdout_format == "google" + + +def test_explicit_stdout_format_env_overrides_profile(monkeypatch) -> None: + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="gcp-agent", + OBSERVABILITY_STDOUT_FORMAT="plain", + ) + + assert config.stdout_format == "plain" + assert config.log_destinations == ("stdout",) + + +def test_sync_profiles_build_no_queued_destinations(monkeypatch) -> None: + for profile in ("gcp-agent", "plain-sync"): + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE=profile) + manager, failures = make_manager(config) + manager.configure() + + assert not any( + isinstance(destination, QueuedLogDestination) + for destination in manager.destinations + ) + assert failures == [] + + +def test_gcp_agent_profile_formats_stdout_through_manager( + monkeypatch, +) -> None: + """End-to-end wiring: the profile's formatter half must survive the + manager's build path, not just direct construction.""" + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="gcp-agent", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + logger = RecordingLogger() + manager, failures = make_manager(config, loggers={"event": logger}) + manager.configure() + + manager.emit( + {"event": "x", "trace_id": "abc"}, log_type="event", severity="INFO" + ) + + line = json.loads(logger.lines[0][1]) + assert line["logging.googleapis.com/labels"]["log_type"] == "event" + assert line["logging.googleapis.com/trace"] == "projects/proj/traces/abc" + assert failures == [] + + +def test_manager_reports_profile_warnings_once(monkeypatch) -> None: + config = _from_env(monkeypatch, OBSERVABILITY_LOG_PROFILE="bogus") + manager, failures = make_manager(config) + manager.configure() + + warnings = [ + str(exc) + for operation, exc, _fields in failures + if operation == "logging.profile_config" + ] + assert len(warnings) == 1 + assert "bogus" in warnings[0] + manager.close() + + +def test_gcp_direct_profile_builds_queued_google(monkeypatch) -> None: + from policyengine_observability.destinations import ( + google_cloud_logging as google_module, + ) + + class StubGoogle: + name = "google_cloud_logging" + + def __init__(self, **kwargs) -> None: + self.kwargs = kwargs + + def emit(self, payload, *, log_type, severity, timestamp=None): + pass + + monkeypatch.setattr( + google_module, "GoogleCloudLoggingDestination", StubGoogle + ) + config = _from_env( + monkeypatch, + OBSERVABILITY_LOG_PROFILE="gcp-direct", + OBSERVABILITY_GOOGLE_CLOUD_PROJECT="proj", + ) + manager, failures = make_manager(config) + manager.configure() + + stdout_destination, queued = manager.destinations + assert isinstance(queued, QueuedLogDestination) + assert isinstance(queued.inner, StubGoogle) + assert queued.inner.kwargs["project"] == "proj" + assert failures == [] + manager.close() diff --git a/tests/test_public_api.py b/tests/test_public_api.py index 26bfbaa..a24f474 100644 --- a/tests/test_public_api.py +++ b/tests/test_public_api.py @@ -84,9 +84,11 @@ async def async_step() -> str: observability.instrument_httpx() observability.shutdown_tracing() observability.shutdown_observability() + observability.restart_observability() assert runtime.operation_duration.calls assert runtime.segment_duration.calls + assert runtime.log_destination_manager.configured is True def test_setting_public_runtime_clears_stale_context() -> None: @@ -117,3 +119,8 @@ async def run() -> dict[str, float]: return timings assert "load_ms" in asyncio.run(run()) + + +def test_registration_hooks_are_exported_at_top_level() -> None: + assert callable(observability.register_destination) + assert callable(observability.register_stdout_formatter) diff --git a/tests/test_queued_destination.py b/tests/test_queued_destination.py new file mode 100644 index 0000000..b5b5364 --- /dev/null +++ b/tests/test_queued_destination.py @@ -0,0 +1,472 @@ +"""QueuedLogDestination tests. + +Determinism strategy: real listener threads are used, but sequencing is +via threading.Event only — close()/stop() drain to a sentinel enqueued +behind the records, so "close then assert delivered" is deterministic. +The only real-time waits are bounded joins that are themselves the +behavior under test, with generous ceilings. +""" + +from __future__ import annotations + +import time +from datetime import UTC, datetime + +import pytest +from fakes import ( + BlockingDestination, + FailingDestination, + RecordingDestination, + TimestampBlindDestination, + make_manager, +) + +from policyengine_observability.config import ObservabilityConfig +from policyengine_observability.destinations.queued import ( + QueuedLogDestination, +) +from policyengine_observability.destinations.registry import ( + destination_strategy, +) + + +def _queued(inner, **kwargs): + failures = [] + destination = QueuedLogDestination( + inner=inner, + on_failure=lambda operation, exc, **fields: failures.append( + (operation, fields) + ), + **kwargs, + ) + return destination, failures + + +def _raise_on_failure(*args, **kwargs): + raise RuntimeError("reporting channel is broken") + + +def _release_abandoned_listener(destination, blocking_inner) -> None: + """Let an abandoned listener thread exit so tests do not leak it.""" + blocking_inner.release.set() + try: + destination._queue.put_nowait(None) # the stdlib sentinel + except Exception: + pass + + +def test_close_drains_all_records_in_order_with_enqueue_timestamps() -> None: + inner = RecordingDestination() + destination, failures = _queued(inner) + before = datetime.now(UTC) + + for index in range(5): + destination.emit({"index": index}, log_type="event", severity="INFO") + destination.close(5.0) + + assert [payload["index"] for payload, *_ in inner.calls] == [ + 0, + 1, + 2, + 3, + 4, + ] + for _payload, log_type, severity, timestamp in inner.calls: + assert log_type == "event" + assert severity == "INFO" + assert timestamp.tzinfo is not None + assert before <= timestamp <= datetime.now(UTC) + assert failures == [] + + +def test_timestamp_not_forwarded_to_blind_inner() -> None: + inner = TimestampBlindDestination() + destination, failures = _queued(inner) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + destination.close(5.0) + + assert inner.calls == [({"event": "x"}, "event", "INFO")] + assert failures == [] + + +def test_timestamp_forwarded_to_var_keyword_inner() -> None: + class KwargsDestination: + name = "kwargs" + + def __init__(self) -> None: + self.kwargs = [] + + def emit(self, payload, **kwargs) -> None: + self.kwargs.append(kwargs) + + inner = KwargsDestination() + destination, failures = _queued(inner) + + destination.emit({"event": "x"}, log_type="event", severity="INFO") + destination.close(5.0) + + assert inner.kwargs[0]["timestamp"].tzinfo is not None + assert failures == [] + + +def test_uninspectable_inner_emit_degrades_to_timestamp_blind() -> None: + class BuiltinEmitDestination: + # inspect.signature(min) raises; construction must survive and + # fall back to timestamp-blind delivery. + name = "builtin-emit" + emit = min + + destination, _failures = _queued(BuiltinEmitDestination()) + + handler = destination._listener.handlers[0] + assert handler.forward_timestamp is False + destination.close(5.0) + + +def test_payload_snapshot_taken_at_enqueue() -> None: + inner = RecordingDestination() + destination, _failures = _queued(inner) + payload = {"nested": {"value": 1}} + + destination.emit(payload, log_type="event", severity="INFO") + payload["nested"]["value"] = 2 + destination.close(5.0) + + ((delivered, *_),) = inner.calls + assert delivered["nested"]["value"] == 1 + + +def test_overflow_drops_newest_and_throttles_reports() -> None: + inner = BlockingDestination() + destination, failures = _queued(inner, maxsize=10, drop_report_interval=3) + + destination.emit({"index": 0}, log_type="event", severity="INFO") + assert inner.entered.wait(5) + # The listener holds record 0; fill the 10-slot queue, then drop 4. + started = time.monotonic() + for index in range(1, 15): + destination.emit({"index": index}, log_type="event", severity="INFO") + assert time.monotonic() - started < 1.0 + + drops = [fields for op, fields in failures if op == "logging.queue_drop"] + assert [d["dropped_total"] for d in drops] == [1, 3] + assert all(d["reason"] == "full" for d in drops) + assert all(d["queue_maxsize"] == 10 for d in drops) + assert destination._drops.count == 4 + + inner.release.set() + destination.close(5.0) + + +def test_close_with_stuck_write_is_bounded_and_terminal() -> None: + inner = BlockingDestination() + destination, failures = _queued(inner) + destination.emit({"index": 0}, log_type="event", severity="INFO") + assert inner.entered.wait(5) + destination.emit({"index": 1}, log_type="event", severity="INFO") + + started = time.monotonic() + destination.close(0.05) + elapsed = time.monotonic() - started + + assert elapsed < 1.0 + timeouts = [ + fields + for op, fields in failures + if op == "logging.queue_close_timeout" + ] + assert len(timeouts) == 1 + assert timeouts[0]["pending_records"] >= 1 + + destination.emit({"index": 2}, log_type="event", severity="INFO") + drops = [fields for op, fields in failures if op == "logging.queue_drop"] + assert drops[-1]["reason"] == "closed" + + _release_abandoned_listener(destination, inner) + + +def test_timed_out_close_leaves_inner_unclosed() -> None: + """An abandoned listener may be mid-write; the inner destination + must not be closed underneath it.""" + inner = BlockingDestination() + destination, _failures = _queued(inner) + destination.emit({"index": 0}, log_type="event", severity="INFO") + assert inner.entered.wait(5) + + destination.close(0.05) + + assert inner.closed == 0 + _release_abandoned_listener(destination, inner) + + +def test_close_with_sentinel_blocked_by_full_queue_is_bounded() -> None: + inner = BlockingDestination() + destination, failures = _queued(inner, maxsize=10) + destination.emit({"index": 0}, log_type="event", severity="INFO") + assert inner.entered.wait(5) + for index in range(1, 11): + destination.emit({"index": index}, log_type="event", severity="INFO") + + started = time.monotonic() + destination.close(0.05) + elapsed = time.monotonic() - started + + assert elapsed < 1.0 + assert any(op == "logging.queue_close_timeout" for op, _fields in failures) + + _release_abandoned_listener(destination, inner) + + +def test_listener_survives_write_failures_and_throttles_reports() -> None: + inner = FailingDestination(fail_first=3) + destination, failures = _queued(inner) + + for index in range(5): + destination.emit({"index": index}, log_type="event", severity="INFO") + destination.close(5.0) + + assert [p["index"] for p in inner.delivered] == [3, 4] + writes = [fields for op, fields in failures if op == "logging.queue_write"] + # Throttle: failure 1 reports, failures 2-3 are under the interval. + assert [w["write_failures_total"] for w in writes] == [1] + assert writes[0]["destination"] == "failing" + + +def test_listener_survives_broken_reporting_channel() -> None: + """The write-failure report itself is guarded on the listener + thread: a raising on_failure must not kill delivery.""" + inner = FailingDestination(fail_first=1) + destination = QueuedLogDestination( + inner=inner, on_failure=_raise_on_failure + ) + + destination.emit({"index": 0}, log_type="event", severity="INFO") + destination.emit({"index": 1}, log_type="event", severity="INFO") + destination.close(5.0) + + assert [p["index"] for p in inner.delivered] == [1] + + +def test_emit_never_raises() -> None: + inner = RecordingDestination() + destination = QueuedLogDestination( + inner=inner, on_failure=_raise_on_failure + ) + + poisoned = {} + poisoned["self"] = poisoned # RecursionError inside normalize + destination.emit(poisoned, log_type="event", severity="INFO") + + destination.close(5.0) + destination.emit({"event": "x"}, log_type="event", severity="INFO") + + blocked = BlockingDestination() + full_destination = QueuedLogDestination( + inner=blocked, on_failure=_raise_on_failure, maxsize=10 + ) + for index in range(12): + full_destination.emit( + {"index": index}, log_type="event", severity="INFO" + ) + + blocked.release.set() + full_destination.close(5.0) + + +def test_atexit_registered_on_construction_unregistered_on_close( + monkeypatch, +) -> None: + registered = [] + unregistered = [] + monkeypatch.setattr( + "policyengine_observability.destinations.queued.atexit.register", + lambda func: registered.append(func), + ) + monkeypatch.setattr( + "policyengine_observability.destinations.queued.atexit.unregister", + lambda func: unregistered.append(func), + ) + inner = RecordingDestination() + + destination, failures = _queued(inner) + destination.close(5.0) + destination.close(5.0) # idempotent: no second stop, no reports + + assert registered == [destination.close] + assert unregistered == [destination.close] + assert failures == [] + + +def test_close_forwards_to_inner_close_only_when_drained() -> None: + class ClosableInner(RecordingDestination): + def __init__(self) -> None: + super().__init__() + self.closed = 0 + + def close(self) -> None: + self.closed += 1 + + inner = ClosableInner() + destination, _failures = _queued(inner) + destination.emit({"event": "x"}, log_type="event", severity="INFO") + destination.close(5.0) + + assert inner.closed == 1 + + +def test_inner_close_failure_is_reported_not_raised() -> None: + class ExplodingClose(RecordingDestination): + def close(self) -> None: + raise RuntimeError("client teardown failed") + + destination, failures = _queued(ExplodingClose()) + + destination.close(5.0) # must not raise (atexit calls this) + + assert any(op == "logging.destination_close" for op, _ in failures) + + +def test_knobs_are_clamped() -> None: + inner = RecordingDestination() + destination, _failures = _queued( + inner, maxsize=float("inf"), close_timeout_seconds=-5 + ) + + assert destination.maxsize == 1000 + assert destination.close_timeout_seconds == 0.0 + destination.close(5.0) + + +@pytest.mark.parametrize( + ("kwargs", "attribute", "expected"), + [ + ({"maxsize": 3}, "maxsize", 10), + ({"maxsize": 10**6}, "maxsize", 100_000), + ({"close_timeout_seconds": 100}, "close_timeout_seconds", 30.0), + ], +) +def test_knob_boundaries_are_pinned(kwargs, attribute, expected) -> None: + """Pin the clamp bounds so a low/high swap cannot pass silently.""" + destination, _failures = _queued(RecordingDestination(), **kwargs) + + assert getattr(destination, attribute) == expected + destination.close(5.0) + + +# ── Destination registry ───────────────────────────────────────────────── + + +def test_registry_builds_remote_wrapped_and_inline_bare( + fake_remote_strategy, +) -> None: + manager, _failures = make_manager( + ObservabilityConfig(log_destinations=(fake_remote_strategy, "stdout")) + ) + manager.configure() + + remote, inline = manager.destinations + assert isinstance(remote, QueuedLogDestination) + assert remote.name == "queued_recording" + assert not isinstance(inline, QueuedLogDestination) + manager.close() + + +def test_registry_unknown_name_reports_config_failure() -> None: + manager, failures = make_manager( + ObservabilityConfig(log_destinations=("nonexistent",)) + ) + manager.configure() + + assert any( + op == "logging.destination_config" + and fields.get("destination") == "nonexistent" + for op, _exc, fields in failures + ) + + +def test_google_strategy_registered_as_remote_with_aliases() -> None: + canonical = destination_strategy("google_cloud_logging") + assert canonical is not None + assert canonical.transport == "remote" + assert canonical.required_config == ("google_cloud_project",) + assert destination_strategy("google") is canonical + assert destination_strategy(" Google-Cloud ") is canonical + + +# ── Manager integration ────────────────────────────────────────────────── + + +def test_manager_breaker_never_disables_queued_destination() -> None: + inner = FailingDestination() + manager, failures = make_manager() + destination = QueuedLogDestination( + inner=inner, on_failure=manager.on_failure + ) + manager.destinations = [destination] + manager.configured = True + + for _ in range(5): + manager.emit({"event": "x"}, log_type="event", severity="INFO") + destination.close(5.0) + + operations = [op for op, *_ in failures] + assert destination in manager.destinations + assert "logging.destination_disabled" not in operations + assert "logging.destination_emit" not in operations + + +def test_manager_passes_queue_knobs_to_queued_destination( + fake_remote_strategy, +) -> None: + manager, _failures = make_manager( + ObservabilityConfig( + log_destinations=(fake_remote_strategy,), + log_queue_maxsize=50, + log_queue_close_timeout_seconds=7.5, + ) + ) + manager.configure() + + queued = manager.destinations[0] + assert queued.maxsize == 50 + assert queued.close_timeout_seconds == 7.5 + manager.close() + + +def test_reconfigure_closes_previous_destinations( + fake_remote_strategy, +) -> None: + manager, _failures = make_manager( + ObservabilityConfig(log_destinations=(fake_remote_strategy,)) + ) + manager.configure() + first = manager.destinations[0] + manager._consecutive_failures[id(first)] = 2 + + manager.configure() + + assert first._closed is True + assert manager._consecutive_failures == {} + assert manager.destinations[0] is not first + manager.close() + + +def test_reconfigure_after_close_resumes_delivery( + fake_remote_strategy, +) -> None: + """The fork/snapshot story: close → drops → configure() → a fresh + listener delivers again.""" + manager, _failures = make_manager( + ObservabilityConfig(log_destinations=(fake_remote_strategy,)) + ) + manager.configure() + manager.close() + manager.emit({"event": "dropped"}, log_type="event", severity="INFO") + + manager.configure() + manager.emit({"event": "delivered"}, log_type="event", severity="INFO") + manager.close(5.0) + + rebuilt = manager.destinations[0] + assert [p["event"] for p in rebuilt.inner.payloads] == ["delivered"] diff --git a/tests/test_runtime.py b/tests/test_runtime.py index 6dfaf7c..e940594 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -19,7 +19,7 @@ from policyengine_observability import runtime as runtime_module from policyengine_observability.config import DEFAULT_METRIC_ATTRIBUTE_KEYS from policyengine_observability.destinations import ( - manager as destination_manager_module, + google_cloud_logging as google_cloud_logging_module, ) @@ -1166,6 +1166,34 @@ def test_from_env_invalid_shutdown_timeout_falls_back(monkeypatch) -> None: assert config.shutdown_timeout_seconds == 3.0 +def test_from_env_reads_stdout_format(monkeypatch) -> None: + monkeypatch.setenv("OBSERVABILITY_STDOUT_FORMAT", "google") + + config = ObservabilityConfig.from_env(service_name="svc") + + assert config.stdout_format == "google" + + +def test_from_env_reads_queue_knobs(monkeypatch) -> None: + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_MAXSIZE", "50") + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS", "1.5") + + config = ObservabilityConfig.from_env(service_name="svc") + + assert config.log_queue_maxsize == 50 + assert config.log_queue_close_timeout_seconds == 1.5 + + +def test_from_env_queue_knobs_fall_back_on_garbage(monkeypatch) -> None: + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_MAXSIZE", "many") + monkeypatch.setenv("OBSERVABILITY_LOG_QUEUE_CLOSE_TIMEOUT_SECONDS", "soon") + + config = ObservabilityConfig.from_env(service_name="svc") + + assert config.log_queue_maxsize == 1000 + assert config.log_queue_close_timeout_seconds == 2.0 + + def test_from_env_enables_otel_by_default() -> None: config = ObservabilityConfig.from_env(service_name="svc") @@ -1411,6 +1439,151 @@ def shutdown(self) -> None: assert "otel.shutdown_timeout" in failures +def test_shutdown_closes_destinations_with_full_budget_and_no_thread( + monkeypatch, +) -> None: + observed = runtime(shutdown_timeout_seconds=2.0) + close_calls = [] + monkeypatch.setattr( + observed.log_destination_manager, + "close", + lambda deadline=None: close_calls.append(deadline), + ) + + def fail_thread(*args, **kwargs): + raise AssertionError( + "no watchdog thread should exist without providers" + ) + + monkeypatch.setattr(runtime_module.threading, "Thread", fail_thread) + + observed.shutdown() + + assert close_calls == [2.0] + + +def test_shutdown_destination_deadline_fits_inside_provider_budget( + monkeypatch, +) -> None: + """The prior design gave the log flush a deadline larger than the + join bounding it, starving provider shutdown; the deadline must be + derived from (and smaller than) the shutdown budget.""" + + class Provider: + def __init__(self) -> None: + self.shutdown_called = False + + def shutdown(self) -> None: + self.shutdown_called = True + + observed = runtime(shutdown_timeout_seconds=2.0) + provider = Provider() + observed.tracer_provider = provider + close_calls = [] + monkeypatch.setattr( + observed.log_destination_manager, + "close", + lambda deadline=None: close_calls.append(deadline), + ) + + observed.shutdown() + + assert close_calls == [1.0] + assert provider.shutdown_called + + +def test_shutdown_slow_destination_close_still_runs_providers() -> None: + class Provider: + def __init__(self) -> None: + self.shutdown_called = False + + def shutdown(self) -> None: + self.shutdown_called = True + + observed = runtime(shutdown_timeout_seconds=0.2) + provider = Provider() + observed.tracer_provider = provider + observed.log_destination_manager.close = lambda deadline=None: time.sleep( + 0.05 + ) + + observed.shutdown() + + assert provider.shutdown_called + + +def test_shutdown_clamps_pathological_budget(monkeypatch) -> None: + observed = runtime(shutdown_timeout_seconds=float("inf")) + close_calls = [] + monkeypatch.setattr( + observed.log_destination_manager, + "close", + lambda deadline=None: close_calls.append(deadline), + ) + + observed.shutdown() + + assert close_calls == [3.0] + + +def test_restart_log_destinations_rebuilds_from_config() -> None: + observed = runtime(otel_enabled=False) + observed.configure() + first = observed.log_destination_manager.destinations[0] + + observed.restart_log_destinations() + + rebuilt = observed.log_destination_manager.destinations + assert len(rebuilt) == 1 + assert rebuilt[0] is not first + assert observed.log_destination_manager.configured is True + + +def test_restart_log_destinations_noops_when_disabled(monkeypatch) -> None: + """The kill switch must hold across forks and snapshot restores: + a disabled runtime's restart must not build destinations.""" + observed = runtime(enabled=False) + configure_calls = [] + monkeypatch.setattr( + observed.log_destination_manager, + "configure", + lambda: configure_calls.append(True), + ) + + observed.restart_log_destinations() + + assert configure_calls == [] + + +def test_shutdown_survives_destination_close_failure(monkeypatch) -> None: + class Provider: + def __init__(self) -> None: + self.shutdown_called = False + + def shutdown(self) -> None: + self.shutdown_called = True + + observed = runtime(shutdown_timeout_seconds=1.0) + provider = Provider() + observed.tracer_provider = provider + failures = [] + observed.log_observability_failure = lambda operation, exc, **fields: ( + failures.append(operation) + ) + + def broken_close(deadline=None): + raise RuntimeError("close exploded") + + monkeypatch.setattr( + observed.log_destination_manager, "close", broken_close + ) + + observed.shutdown() + + assert provider.shutdown_called + assert "logging.destination_close" in failures + + def test_configure_otel_creates_real_providers_and_instruments() -> None: observed = runtime(otel_enabled=True) @@ -1901,7 +2074,7 @@ def fail_google_destination(**_kwargs): raise ImportError("google-cloud-logging missing") monkeypatch.setattr( - destination_manager_module, + google_cloud_logging_module, "GoogleCloudLoggingDestination", fail_google_destination, ) @@ -1928,7 +2101,7 @@ def fail_google_destination(**_kwargs): raise AssertionError("google destination should not initialize") monkeypatch.setattr( - destination_manager_module, + google_cloud_logging_module, "GoogleCloudLoggingDestination", fail_google_destination, )