diff --git a/CHANGELOG.md b/CHANGELOG.md index 483c73d1d..d3ba12252 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,8 +20,27 @@ to include examples, links to docs, or any other relevant information. ### Added +- Added experimental `temporalio.contrib.opentelemetry.ReplaySafeMeterProvider` and + `ReplaySafeLoggerProvider`, which wrap a user-supplied OpenTelemetry provider and drop + synchronous instrument recordings and emitted log records made from workflow code during + replay. Install them as the + process-global providers when libraries record OpenTelemetry metrics or emit log events from + workflow code (e.g. Google ADK) so that workflow replays (cache eviction, worker restarts, + redeploys) do not duplicate telemetry; recordings are first-execution-only, matching + `temporalio.workflow.metric_meter()`. + `temporalio.contrib.opentelemetry.ReplaySafeTracerProvider` is now also exported. + `GoogleAdkPlugin` now warns at worker and replayer configuration time when the global + OpenTelemetry meter or tracer provider is positively identified as not replay-safe + (an OpenTelemetry SDK provider used directly). + ### Changed +- The `opentelemetry` and `lambda-worker-otel` extras now require + `opentelemetry-api`/`opentelemetry-sdk` `>= 1.24`, aligning the declared floor with what + `temporalio.contrib.opentelemetry` already required in practice (it has depended on an API + added in `opentelemetry-api` 1.24 since the tracing integration was introduced, and the + lambda worker builds on it). + ### Deprecated ### :boom: Breaking Changes diff --git a/pyproject.toml b/pyproject.toml index 2303bdf00..2c2c03436 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,15 +26,15 @@ classifiers = [ [project.optional-dependencies] grpc = ["grpcio>=1.48.2,<2"] -opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"] +opentelemetry = ["opentelemetry-api>=1.24,<2", "opentelemetry-sdk>=1.24,<2"] pydantic = ["pydantic>=2.0.0,<3"] openai-agents = ["openai-agents>=0.17.5", "mcp>=1.9.4, <2"] google-adk = ["google-adk>=2.2.0,<3"] langgraph = ["langgraph>=1.1.0"] langsmith = ["langsmith>=0.7.34,<0.9"] lambda-worker-otel = [ - "opentelemetry-api>=1.11.1,<2", - "opentelemetry-sdk>=1.11.1,<2", + "opentelemetry-api>=1.24,<2", + "opentelemetry-sdk>=1.24,<2", "opentelemetry-exporter-otlp-proto-grpc>=1.11.1,<2", "opentelemetry-semantic-conventions>=0.40b0,<1", "opentelemetry-sdk-extension-aws>=2.0.0,<3", diff --git a/temporalio/contrib/google_adk_agents/README.md b/temporalio/contrib/google_adk_agents/README.md index 92fd1cea7..ec813ef03 100644 --- a/temporalio/contrib/google_adk_agents/README.md +++ b/temporalio/contrib/google_adk_agents/README.md @@ -199,6 +199,66 @@ agent = Agent( ) ``` +## Telemetry and Workflow Replay + +ADK records OpenTelemetry metrics (scope `gcp.vertex.agent`, e.g. +`gen_ai.client.token.usage`), spans, and log events (e.g. `gen_ai.choice`) +through the process-global OpenTelemetry providers from code that runs +inside the workflow. Workflow code re-executes on every replay, so with a +plain global provider each replay re-records all of that telemetry even +though no model or tool actually ran again — for example, 1 real execution +followed by 3 replays yields 4x the observations on every instrument and 4 +copies of every log event. Replays happen routinely in production: workflow +cache eviction, worker restarts, redeploys, or running with +`max_cached_workflows=0`. + +To avoid this, install Temporal's replay-safe providers as the global +OpenTelemetry providers. They pass recordings through on first execution and +drop them during replay: + +```python +import opentelemetry._logs +import opentelemetry.metrics +import opentelemetry.trace +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader +from opentelemetry.sdk.trace.export import BatchSpanProcessor + +from temporalio.contrib.opentelemetry import ( + ReplaySafeLoggerProvider, + ReplaySafeMeterProvider, + create_tracer_provider, +) + +# The global set_*_provider functions only take effect once per process, so +# these wrappers must be the first and only global providers set. +opentelemetry.metrics.set_meter_provider( + ReplaySafeMeterProvider( + MeterProvider(metric_readers=[PeriodicExportingMetricReader(my_exporter)]) + ) +) +tracer_provider = create_tracer_provider() +tracer_provider.add_span_processor(BatchSpanProcessor(my_span_exporter)) +opentelemetry.trace.set_tracer_provider(tracer_provider) +logger_provider = LoggerProvider() +logger_provider.add_log_record_processor(BatchLogRecordProcessor(my_log_exporter)) +opentelemetry._logs.set_logger_provider(ReplaySafeLoggerProvider(logger_provider)) +``` + +`GoogleAdkPlugin` warns at worker and replayer configuration time when the +global meter or tracer provider is positively identified as not replay-safe +(an OpenTelemetry SDK provider used directly). The global logger provider is +not checked because the OpenTelemetry logs SDK has no public import path yet, +but the same replay duplication applies to it. + +Recordings are first-execution-only, matching +`temporalio.workflow.metric_meter()`: a retried workflow task re-executes +live and can record again, and tokens consumed by failed activity attempts +are not counted. Telemetry recorded from activities (worker-side) is +unaffected. + ## Integration Points This integration provides comprehensive support for running Google ADK Agents within Temporal workflows while maintaining: diff --git a/temporalio/contrib/google_adk_agents/_plugin.py b/temporalio/contrib/google_adk_agents/_plugin.py index 7344485c8..9fc3c4d19 100644 --- a/temporalio/contrib/google_adk_agents/_plugin.py +++ b/temporalio/contrib/google_adk_agents/_plugin.py @@ -1,12 +1,18 @@ from __future__ import annotations import dataclasses +import inspect import time import uuid +import warnings from collections.abc import AsyncIterator, Callable from contextlib import asynccontextmanager +from types import FrameType from typing import Any +import opentelemetry.metrics +import opentelemetry.trace + from temporalio import workflow from temporalio.contrib.google_adk_agents._mcp import TemporalMcpToolSetProvider from temporalio.contrib.google_adk_agents._model import ( @@ -20,11 +26,72 @@ from temporalio.converter import DataConverter, DefaultPayloadConverter from temporalio.plugin import SimplePlugin from temporalio.worker import ( + ReplayerConfig, + WorkerConfig, WorkflowRunner, ) from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner +def _stacklevel_outside_temporalio() -> int: + # Attribute provider warnings to the nearest frame outside temporalio, + # e.g. the user's Worker(...)/Replayer(...) call or a user plugin that + # delegates here, however many plugin frames sit in between. + level = 1 + own_frame: FrameType | None = inspect.currentframe() + frame = own_frame.f_back if own_frame is not None else None + while frame is not None: + module = frame.f_globals.get("__name__", "") + if module != "temporalio" and not module.startswith("temporalio."): + return level + frame = frame.f_back + level += 1 + return 1 + + +def _warn_if_global_otel_providers_not_replay_safe() -> None: + # ADK records metrics, spans, and log events through the process-global + # OpenTelemetry providers from code that runs workflow-side, so a + # non-replay-safe global provider re-emits that telemetry on every + # workflow replay. Warn only on providers positively identified as + # replay-unsafe: an OpenTelemetry SDK provider used directly as the + # global. Anything else stays silent -- unset (proxy) and no-op providers + # drop recordings, and unknown provider types (e.g. a custom provider + # delegating to a replay-safe one) cannot be classified, where a false + # positive is worse than a missed warning. The SDK logger provider is not + # checked because its class is only importable from the underscore + # namespace opentelemetry.sdk._logs while OpenTelemetry logs are pre-GA. + try: + from opentelemetry.sdk.metrics import MeterProvider as SdkMeterProvider + from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider + except ImportError: + # Without the opentelemetry-sdk package installed no SDK provider can + # exist, so there is nothing replay-unsafe to warn about. + return + stacklevel = _stacklevel_outside_temporalio() + if isinstance(opentelemetry.metrics.get_meter_provider(), SdkMeterProvider): + warnings.warn( + "The global OpenTelemetry MeterProvider is not replay-safe: Google ADK " + "records metrics from workflow code, so every workflow replay will " + "re-record them. Wrap your provider in " + "temporalio.contrib.opentelemetry.ReplaySafeMeterProvider and make it " + "the first and only global provider set: " + "opentelemetry.metrics.set_meter_provider(ReplaySafeMeterProvider(provider))", + UserWarning, + stacklevel=stacklevel, + ) + if isinstance(opentelemetry.trace.get_tracer_provider(), SdkTracerProvider): + warnings.warn( + "The global OpenTelemetry TracerProvider is not replay-safe: Google ADK " + "creates spans from workflow code, so every workflow replay will " + "re-emit them. Install a replay-safe provider: " + "opentelemetry.trace.set_tracer_provider(" + "temporalio.contrib.opentelemetry.create_tracer_provider())", + UserWarning, + stacklevel=stacklevel, + ) + + def setup_deterministic_runtime(): """Configures ADK runtime for Temporal determinism. @@ -68,6 +135,10 @@ class GoogleAdkPlugin(SimplePlugin): This plugin configures: - Pydantic Payload Converter (required for ADK objects). - Sandbox Passthrough for google.adk and google.genai modules. + + At worker and replayer configuration time it also warns when the global + OpenTelemetry meter or tracer provider is not replay-safe, since ADK + telemetry recorded from workflow code would duplicate on replay. """ def __init__( @@ -118,6 +189,22 @@ def workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner: workflow_runner=workflow_runner, ) + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + """See base class. Also warns when the global OpenTelemetry meter or + tracer provider is not replay-safe, since ADK telemetry would + duplicate on replay. + """ + _warn_if_global_otel_providers_not_replay_safe() + return super().configure_worker(config) + + def configure_replayer(self, config: ReplayerConfig) -> ReplayerConfig: + """See base class. Also warns when the global OpenTelemetry meter or + tracer provider is not replay-safe, since every replayed workflow + would re-emit ADK telemetry. + """ + _warn_if_global_otel_providers_not_replay_safe() + return super().configure_replayer(config) + def _configure_data_converter( self, converter: DataConverter | None ) -> DataConverter: diff --git a/temporalio/contrib/opentelemetry/README.md b/temporalio/contrib/opentelemetry/README.md index 2c6e39817..a81c40bbf 100644 --- a/temporalio/contrib/opentelemetry/README.md +++ b/temporalio/contrib/opentelemetry/README.md @@ -236,6 +236,62 @@ with tracer.start_as_current_span("my-operation") as span: }) ``` +## Replay-Safe Metrics + +For Temporal SDK metrics inside workflows, use `temporalio.workflow.metric_meter()`, +which is already replay-safe. However, third-party libraries (e.g. Google ADK) may +record OpenTelemetry metrics through the process-global meter provider from code +that runs inside workflows. Workflow code re-executes on every replay (cache +eviction, worker restart, redeploy), so a plain global meter provider re-records +those metrics on each replay, inflating counts. + +`ReplaySafeMeterProvider` wraps your meter provider so synchronous instrument +recordings made from workflow code are dropped during replay, mirroring what +`create_tracer_provider()` does for spans: + +```python +import opentelemetry.metrics +from opentelemetry.sdk.metrics import MeterProvider +from temporalio.contrib.opentelemetry import ReplaySafeMeterProvider + +# set_meter_provider only takes effect once per process, so this wrapper must +# be the first and only global meter provider set, installed before any +# library records metrics. +opentelemetry.metrics.set_meter_provider( + ReplaySafeMeterProvider(MeterProvider(metric_readers=[my_reader])) +) +``` + +Recordings are first-execution-only, matching `workflow.metric_meter()`: a +retried workflow task re-executes live and can record again. Observable +(asynchronous) instruments and recordings made outside workflows pass through +untouched. + +## Replay-Safe Log Events + +Libraries may also emit OpenTelemetry log records through the process-global +logger provider from workflow code (e.g. Google ADK's `gen_ai.*` events), +which duplicate on every replay the same way. `ReplaySafeLoggerProvider` +wraps your logger provider so records emitted from workflow code are dropped +during replay: + +```python +import opentelemetry._logs +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import BatchLogRecordProcessor +from temporalio.contrib.opentelemetry import ReplaySafeLoggerProvider + +# set_logger_provider only takes effect once per process, so this wrapper +# must be the first and only global logger provider set, installed before +# any library emits log records. +logger_provider = LoggerProvider() +logger_provider.add_log_record_processor(BatchLogRecordProcessor(my_log_exporter)) +opentelemetry._logs.set_logger_provider(ReplaySafeLoggerProvider(logger_provider)) +``` + +Emissions are first-execution-only: a retried workflow task re-executes live +and can emit again. Emissions outside workflows pass through untouched. + ## Best Practices 1. **Register on Client**: Always register plugins/interceptors on the client, not the worker, to ensure proper context propagation diff --git a/temporalio/contrib/opentelemetry/__init__.py b/temporalio/contrib/opentelemetry/__init__.py index 74f069322..feafe91ae 100644 --- a/temporalio/contrib/opentelemetry/__init__.py +++ b/temporalio/contrib/opentelemetry/__init__.py @@ -2,21 +2,34 @@ This package provides OpenTelemetry tracing integration for Temporal workflows, activities, and other operations. It includes automatic span creation and -propagation for distributed tracing. +propagation for distributed tracing. It also provides replay-safe wrappers for +the global OpenTelemetry tracer, meter, and logger providers. """ from temporalio.contrib.opentelemetry._interceptor import ( TracingInterceptor, TracingWorkflowInboundInterceptor, ) +from temporalio.contrib.opentelemetry._logger_provider import ( + ReplaySafeLoggerProvider, +) +from temporalio.contrib.opentelemetry._meter_provider import ( + ReplaySafeMeterProvider, +) from temporalio.contrib.opentelemetry._otel_interceptor import OpenTelemetryInterceptor from temporalio.contrib.opentelemetry._plugin import OpenTelemetryPlugin -from temporalio.contrib.opentelemetry._tracer_provider import create_tracer_provider +from temporalio.contrib.opentelemetry._tracer_provider import ( + ReplaySafeTracerProvider, + create_tracer_provider, +) __all__ = [ "TracingInterceptor", "TracingWorkflowInboundInterceptor", "OpenTelemetryInterceptor", "OpenTelemetryPlugin", + "ReplaySafeLoggerProvider", + "ReplaySafeMeterProvider", + "ReplaySafeTracerProvider", "create_tracer_provider", ] diff --git a/temporalio/contrib/opentelemetry/_logger_provider.py b/temporalio/contrib/opentelemetry/_logger_provider.py new file mode 100644 index 000000000..2a9cac28f --- /dev/null +++ b/temporalio/contrib/opentelemetry/_logger_provider.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +# opentelemetry._logs is the import path OpenTelemetry itself documents for +# the logs bridge API while it is pre-GA (there is no non-underscore +# counterpart); the only alternative would be not gating ADK-style log +# emission at all. +from opentelemetry._logs import Logger, LoggerProvider + +from temporalio import workflow + +if TYPE_CHECKING: + # _ExtendedAttributes is the annotation OpenTelemetry's own public + # get_logger signature uses; there is no public alias. + from opentelemetry.util.types import _ExtendedAttributes + + +def _skip_emitting() -> bool: + # in_workflow() must be evaluated first: is_replaying() requires an active + # workflow context. + return workflow.in_workflow() and workflow.unsafe.is_replaying() + + +class _ReplaySafeLogger(Logger): + def __init__( + self, + logger: Logger, + name: str, + version: str | None = None, + schema_url: str | None = None, + ) -> None: + super().__init__(name, version=version, schema_url=schema_url) + self._logger = logger + + def __getattr__(self, name: str) -> object: + return getattr(self._logger, name) + + def emit(self, *args: Any, **kwargs: Any) -> None: + # emit's signature differs across the supported opentelemetry-api + # range (a single positional LogRecord through 1.37, keyword fields + # from 1.38), so forward arguments verbatim rather than pinning one + # shape; the wrapped logger comes from the same installed API. + if _skip_emitting(): + # Skip emitting log records during workflow replay to avoid duplicate telemetry + return + self._logger.emit(*args, **kwargs) + + +class ReplaySafeLoggerProvider(LoggerProvider): + """A logger provider that is safe for use during workflow replay. + + .. warning:: + This class is experimental and may change in future versions. + Use with caution in production environments. + + This logger provider wraps an OpenTelemetry LoggerProvider and drops log + records emitted (``Logger.emit()``) from workflow code while the workflow + is replaying. Without this, libraries that emit OpenTelemetry log records + from workflow code (e.g. ``google-adk``'s ``gen_ai.*`` events) re-emit + every record on each replay, duplicating telemetry. + + Emissions are therefore first-execution-only: a workflow task retry + re-executes live and can emit again. Emissions outside workflows are + unaffected. + + Install this as the process-global logger provider before any library + (e.g. ``google-adk``) obtains loggers:: + + opentelemetry._logs.set_logger_provider( + ReplaySafeLoggerProvider(my_logger_provider) + ) + + OpenTelemetry proxy loggers late-bind, so calling ``set_logger_provider`` + after such libraries are imported still routes their loggers through this + wrapper. However, ``set_logger_provider`` only takes effect once per + process, so this wrapper must be the one and only global logger provider + ever set. + """ + + def __init__(self, logger_provider: LoggerProvider) -> None: + """Initialize the replay-safe logger provider. + + Args: + logger_provider: The underlying OpenTelemetry LoggerProvider to wrap. + """ + self._logger_provider = logger_provider + + def __getattr__(self, name: str) -> Any: + """Delegate all other attributes (e.g. ``shutdown``, ``force_flush``) + to the underlying logger provider. + """ + return getattr(self._logger_provider, name) + + def get_logger( + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: _ExtendedAttributes | None = None, + ) -> Logger: + """Get a replay-safe logger from the underlying provider. + + Args: + name: The name of the instrumenting module. + version: The version string of the instrumenting library. + schema_url: The schema URL for the logger. + attributes: Instrumentation scope attributes for the logger. + + Returns: + A replay-safe logger instance. + """ + # Forward attributes only when set: the parameter was added in + # opentelemetry 1.26 and passing it to older providers raises + # TypeError. + if attributes is None: + inner = self._logger_provider.get_logger(name, version, schema_url) + else: + inner = self._logger_provider.get_logger( + name, version, schema_url, attributes + ) + return _ReplaySafeLogger(inner, name, version=version, schema_url=schema_url) diff --git a/temporalio/contrib/opentelemetry/_meter_provider.py b/temporalio/contrib/opentelemetry/_meter_provider.py new file mode 100644 index 000000000..164c63cfd --- /dev/null +++ b/temporalio/contrib/opentelemetry/_meter_provider.py @@ -0,0 +1,278 @@ +from collections.abc import Sequence +from typing import Any + +from opentelemetry.context import Context + +# _Gauge is OpenTelemetry's canonical exported name for the spec-experimental +# synchronous gauge, re-exported by opentelemetry.metrics since 1.23. +from opentelemetry.metrics import ( + CallbackT, + Counter, + Histogram, + Meter, + MeterProvider, + ObservableCounter, + ObservableGauge, + ObservableUpDownCounter, + UpDownCounter, + _Gauge, +) +from opentelemetry.util.types import Attributes + +from temporalio import workflow + + +def _forward_context_kwarg(context: Context | None) -> dict[str, Any]: + # Forward the context only when set: the parameter was added to the + # synchronous instrument methods in opentelemetry 1.28 and older + # instruments raise TypeError when it is passed. + return {} if context is None else {"context": context} + + +def _skip_recording() -> bool: + # in_workflow() must be evaluated first: is_replaying() requires an active + # workflow context. + return workflow.in_workflow() and workflow.unsafe.is_replaying() + + +class _ReplaySafeCounter(Counter): + def __init__(self, counter: Counter) -> None: + self._counter = counter + + def __getattr__(self, name: str) -> object: + return getattr(self._counter, name) + + def add( + self, + amount: int | float, + attributes: Attributes | None = None, + context: Context | None = None, + ) -> None: + if _skip_recording(): + # Skip recording metrics during workflow replay to avoid duplicate telemetry + return + self._counter.add(amount, attributes, **_forward_context_kwarg(context)) + + +class _ReplaySafeUpDownCounter(UpDownCounter): + def __init__(self, counter: UpDownCounter) -> None: + self._counter = counter + + def __getattr__(self, name: str) -> object: + return getattr(self._counter, name) + + def add( + self, + amount: int | float, + attributes: Attributes | None = None, + context: Context | None = None, + ) -> None: + if _skip_recording(): + # Skip recording metrics during workflow replay to avoid duplicate telemetry + return + self._counter.add(amount, attributes, **_forward_context_kwarg(context)) + + +class _ReplaySafeHistogram(Histogram): + def __init__(self, histogram: Histogram) -> None: + self._histogram = histogram + + def __getattr__(self, name: str) -> object: + return getattr(self._histogram, name) + + def record( + self, + amount: int | float, + attributes: Attributes | None = None, + context: Context | None = None, + ) -> None: + if _skip_recording(): + # Skip recording metrics during workflow replay to avoid duplicate telemetry + return + self._histogram.record(amount, attributes, **_forward_context_kwarg(context)) + + +class _ReplaySafeGauge(_Gauge): + def __init__(self, gauge: _Gauge) -> None: + self._gauge = gauge + + def __getattr__(self, name: str) -> object: + return getattr(self._gauge, name) + + def set( + self, + amount: int | float, + attributes: Attributes | None = None, + context: Context | None = None, + ) -> None: + if _skip_recording(): + # Skip recording metrics during workflow replay to avoid duplicate telemetry + return + self._gauge.set(amount, attributes, **_forward_context_kwarg(context)) + + +class _ReplaySafeMeter(Meter): + # Overrides every Meter method as of opentelemetry-api 1.44. OTel adds new + # instrument kinds as non-abstract no-op defaults on the Meter ABC (e.g. + # create_gauge in 1.23), which __getattr__ cannot intercept, so new Meter + # methods must be audited and overridden here on opentelemetry upgrades. + def __init__(self, meter: Meter) -> None: + super().__init__(meter.name, version=meter.version, schema_url=meter.schema_url) + self._meter = meter + + def create_counter( + self, + name: str, + unit: str = "", + description: str = "", + ) -> Counter: + return _ReplaySafeCounter(self._meter.create_counter(name, unit, description)) + + def create_up_down_counter( + self, + name: str, + unit: str = "", + description: str = "", + ) -> UpDownCounter: + return _ReplaySafeUpDownCounter( + self._meter.create_up_down_counter(name, unit, description) + ) + + def create_histogram( + self, + name: str, + unit: str = "", + description: str = "", + *, + explicit_bucket_boundaries_advisory: Sequence[float] | None = None, + ) -> Histogram: + # Forward the advisory only when set: the parameter was added in + # opentelemetry 1.30 and unconditionally forwarding it raises TypeError + # on older APIs still within the supported version range. + kwargs: dict[str, Any] = {} + if explicit_bucket_boundaries_advisory is not None: + kwargs["explicit_bucket_boundaries_advisory"] = ( + explicit_bucket_boundaries_advisory + ) + return _ReplaySafeHistogram( + self._meter.create_histogram(name, unit, description, **kwargs) + ) + + def create_gauge( + self, + name: str, + unit: str = "", + description: str = "", + ) -> _Gauge: + return _ReplaySafeGauge(self._meter.create_gauge(name, unit, description)) + + # Observable instruments pass through unwrapped: their callbacks run on the + # metric reader's collect thread, never inside workflow code. + + def create_observable_counter( + self, + name: str, + callbacks: Sequence[CallbackT] | None = None, + unit: str = "", + description: str = "", + ) -> ObservableCounter: + return self._meter.create_observable_counter(name, callbacks, unit, description) + + def create_observable_gauge( + self, + name: str, + callbacks: Sequence[CallbackT] | None = None, + unit: str = "", + description: str = "", + ) -> ObservableGauge: + return self._meter.create_observable_gauge(name, callbacks, unit, description) + + def create_observable_up_down_counter( + self, + name: str, + callbacks: Sequence[CallbackT] | None = None, + unit: str = "", + description: str = "", + ) -> ObservableUpDownCounter: + return self._meter.create_observable_up_down_counter( + name, callbacks, unit, description + ) + + +class ReplaySafeMeterProvider(MeterProvider): + """A meter provider that is safe for use during workflow replay. + + .. warning:: + This class is experimental and may change in future versions. + Use with caution in production environments. + + This meter provider wraps an OpenTelemetry MeterProvider and drops + synchronous instrument recordings (counter ``add()``, up-down counter + ``add()``, histogram ``record()``, and gauge ``set()``) made from workflow + code while the workflow is replaying. Without this, libraries that record + metrics from workflow code (e.g. ``google-adk``) re-record every + measurement on each replay, inflating counts. + + Recordings are therefore first-execution-only, matching + :py:meth:`temporalio.workflow.metric_meter`: a workflow task retry + re-executes live and can record again. Observable (asynchronous) + instruments pass through untouched since their callbacks run on the metric + reader's collect thread, never inside workflow code. Recordings outside + workflows are unaffected. + + Install this as the process-global meter provider before any library + (e.g. ``google-adk``) creates instruments:: + + opentelemetry.metrics.set_meter_provider( + ReplaySafeMeterProvider(my_meter_provider) + ) + + OpenTelemetry proxy meters late-bind, so calling ``set_meter_provider`` + after such libraries are imported still routes their instruments through + this wrapper. However, ``set_meter_provider`` only takes effect once per + process, so this wrapper must be the one and only global meter provider + ever set. + """ + + def __init__(self, meter_provider: MeterProvider) -> None: + """Initialize the replay-safe meter provider. + + Args: + meter_provider: The underlying OpenTelemetry MeterProvider to wrap. + """ + self._meter_provider = meter_provider + + def __getattr__(self, name: str) -> Any: + """Delegate all other attributes (e.g. ``shutdown``, ``force_flush``) + to the underlying meter provider. + """ + return getattr(self._meter_provider, name) + + def get_meter( + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: Attributes | None = None, + ) -> Meter: + """Get a replay-safe meter from the underlying provider. + + Args: + name: The name of the instrumenting module. + version: The version string of the instrumenting library. + schema_url: The schema URL for the meter. + attributes: Additional attributes for the meter. + + Returns: + A replay-safe meter instance. + """ + # Forward attributes only when set: the parameter was added in + # opentelemetry 1.26 and passing it to older providers raises + # TypeError. + if attributes is None: + inner = self._meter_provider.get_meter(name, version, schema_url) + else: + inner = self._meter_provider.get_meter( + name, version, schema_url, attributes + ) + return _ReplaySafeMeter(inner) diff --git a/temporalio/contrib/opentelemetry/_tracer_provider.py b/temporalio/contrib/opentelemetry/_tracer_provider.py index 929f8bf27..24adeb780 100644 --- a/temporalio/contrib/opentelemetry/_tracer_provider.py +++ b/temporalio/contrib/opentelemetry/_tracer_provider.py @@ -174,16 +174,20 @@ def __init__( ): """Initialize the replay-safe tracer provider. + Prefer :py:func:`create_tracer_provider`, which constructs a provider + that satisfies these requirements. + Args: tracer_provider: The underlying OpenTelemetry TracerProvider to wrap. - Must use a _TemporalIdGenerator for replay safety. + Must use a ``TemporalIdGenerator`` for replay safety. + id_generator: The ``TemporalIdGenerator`` used by ``tracer_provider``. Raises: - ValueError: If the tracer provider doesn't use a _TemporalIdGenerator. + ValueError: If the tracer provider doesn't use a ``TemporalIdGenerator``. """ if not isinstance(tracer_provider.id_generator, TemporalIdGenerator): raise ValueError( - "ReplaySafeTracerProvider should only be used with a TemporalIdGenerator for replay safety. The given TracerProvider doesnt use one." + "ReplaySafeTracerProvider should only be used with a TemporalIdGenerator for replay safety. The given TracerProvider doesn't use one." ) self._id_generator = id_generator self._tracer_provider = tracer_provider diff --git a/tests/conftest.py b/tests/conftest.py index 9c57bc0d1..533bcf0f6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -4,6 +4,8 @@ import sys from collections.abc import AsyncGenerator, Iterator +import opentelemetry._logs._internal +import opentelemetry.metrics._internal import opentelemetry.trace import pytest import pytest_asyncio @@ -260,11 +262,49 @@ def continue_as_new_suggest_history_count() -> int: return CONTINUE_AS_NEW_SUGGEST_HISTORY_COUNT +# OpenTelemetry's global providers are set-once per process with no public +# way to unset them, so tests needing their own provider must reset the +# globals directly -- the same isolation pattern OpenTelemetry's own test +# suite uses (opentelemetry.test.globals_test). + + @pytest.fixture def reset_otel_tracer_provider(): - """Reset global OpenTelemetry tracer provider state around tests.""" + """Isolate global OpenTelemetry tracer provider state around a test.""" opentelemetry.trace._TRACER_PROVIDER_SET_ONCE = Once() opentelemetry.trace._TRACER_PROVIDER = None yield opentelemetry.trace._TRACER_PROVIDER_SET_ONCE = Once() opentelemetry.trace._TRACER_PROVIDER = None + + +@pytest.fixture +def reset_otel_meter_provider(): + """Isolate global OpenTelemetry meter provider state around a test. + + Proxy meters/instruments already bound to a real provider stay bound after + this reset; only the next set_meter_provider call rebinds them. Tests must + not assume an unset global provider drops recordings from instruments + created in earlier tests. + """ + opentelemetry.metrics._internal._METER_PROVIDER_SET_ONCE = Once() + opentelemetry.metrics._internal._METER_PROVIDER = None + yield + opentelemetry.metrics._internal._METER_PROVIDER_SET_ONCE = Once() + opentelemetry.metrics._internal._METER_PROVIDER = None + + +@pytest.fixture +def reset_otel_logger_provider(): + """Isolate global OpenTelemetry logger provider state around a test. + + Unlike proxy meters, proxy loggers cache their real logger on first use + and never rebind, even across a later set_logger_provider call. Tests + exercising a library's module-level logger must clear that cache + themselves (e.g. Google ADK's telemetry logger). + """ + opentelemetry._logs._internal._LOGGER_PROVIDER_SET_ONCE = Once() + opentelemetry._logs._internal._LOGGER_PROVIDER = None + yield + opentelemetry._logs._internal._LOGGER_PROVIDER_SET_ONCE = Once() + opentelemetry._logs._internal._LOGGER_PROVIDER = None diff --git a/tests/contrib/google_adk_agents/test_replay_metrics.py b/tests/contrib/google_adk_agents/test_replay_metrics.py new file mode 100644 index 000000000..96281eafc --- /dev/null +++ b/tests/contrib/google_adk_agents/test_replay_metrics.py @@ -0,0 +1,505 @@ +"""Tests for replay-safe handling of ADK's OpenTelemetry metrics and log events. + +Google ADK records token-usage/latency metrics and emits gen_ai.* log events +through the process-global OpenTelemetry providers (scope "gcp.vertex.agent") +from code that runs workflow-side under the Temporal adapter. Without gating, +1 real execution + N replays produces (1 + N) observations per instrument and +(1 + N) copies of every log event, while the activity (the model / tool call) +runs exactly once. Installing ReplaySafeMeterProvider / +ReplaySafeLoggerProvider as the global providers suppresses the replay +recordings while leaving first-execution recordings intact. +""" + +import inspect +import uuid +import warnings +from collections.abc import AsyncGenerator +from datetime import timedelta + +import google.adk.telemetry.tracing +import opentelemetry._logs +import opentelemetry.metrics +import pytest +from google.adk import Agent +from google.adk.models import BaseLlm, LLMRegistry +from google.adk.models.llm_request import LlmRequest +from google.adk.models.llm_response import LlmResponse +from google.adk.runners import InMemoryRunner +from google.adk.utils.context_utils import Aclosing +from google.genai import types +from opentelemetry.metrics import Meter +from opentelemetry.metrics import MeterProvider as ApiMeterProvider +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, +) +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import InMemoryMetricReader +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.trace import set_tracer_provider +from opentelemetry.util.types import Attributes + +import temporalio.contrib.google_adk_agents.workflow +from temporalio import activity, workflow +from temporalio.api.enums.v1 import EventType +from temporalio.client import Client +from temporalio.contrib.google_adk_agents import GoogleAdkPlugin, TemporalModel +from temporalio.contrib.opentelemetry import ( + ReplaySafeLoggerProvider, + ReplaySafeMeterProvider, + create_tracer_provider, +) +from temporalio.worker import Replayer, Worker, WorkerConfig +from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner + +MODEL_NAME = "replay-metrics-model" +ADK_METER_SCOPE = "gcp.vertex.agent" + +# One agent invocation, one tool call, two model calls (tool call + final +# answer), token usage recorded twice per model call (input + output). +EXPECTED_BASELINE = { + "gen_ai.invoke_agent.duration": 1, + "gen_ai.invoke_agent.inference_calls": 1, + "gen_ai.invoke_agent.tool_calls": 1, + "gen_ai.execute_tool.duration": 1, + "gen_ai.client.operation.duration": 2, + "gen_ai.client.token.usage": 4, +} + +# Log events per model call: one gen_ai.system.message, one +# gen_ai.user.message per request content (1 for the first call; 3 for the +# second: prompt + tool call + tool response), one gen_ai.choice per result. +EXPECTED_LOG_BASELINE = { + "gen_ai.system.message": 2, + "gen_ai.user.message": 4, + "gen_ai.choice": 2, +} + +# Counts real (worker-side) activity executions; replays must not add to it. +activity_executions = 0 + + +@activity.defn +async def replay_metrics_get_weather(city: str) -> str: # type: ignore[reportUnusedParameter] + global activity_executions + activity_executions += 1 + return "Warm and sunny. 17 degrees." + + +class ReplayMetricsModel(BaseLlm): + """Scripted model: one tool call, then a final text answer. + + Both responses carry usage_metadata so gen_ai.client.token.usage records. + """ + + @classmethod + def supported_models(cls) -> list[str]: + return [MODEL_NAME] + + def _responses(self) -> list[LlmResponse]: + return [ + LlmResponse( + content=types.Content( + role="model", + parts=[ + types.Part( + function_call=types.FunctionCall( + args={"city": "New York"}, + name="replay_metrics_get_weather", + ) + ) + ], + ), + usage_metadata=types.GenerateContentResponseUsageMetadata( + prompt_token_count=100, + candidates_token_count=25, + total_token_count=125, + ), + ), + LlmResponse( + content=types.Content( + role="model", + parts=[types.Part(text="warm and sunny")], + ), + usage_metadata=types.GenerateContentResponseUsageMetadata( + prompt_token_count=200, + candidates_token_count=10, + total_token_count=210, + ), + ), + ] + + async def generate_content_async( + self, llm_request: LlmRequest, stream: bool = False + ) -> AsyncGenerator[LlmResponse, None]: + for response in self._responses(): + if any(content == response.content for content in llm_request.contents): + continue + yield response + return + + +@workflow.defn +class ReplayMetricsAgent: + @workflow.run + async def run(self, prompt: str, model_name: str) -> str | None: + weather_tool = temporalio.contrib.google_adk_agents.workflow.activity_tool( + replay_metrics_get_weather, + start_to_close_timeout=timedelta(seconds=60), + ) + agent = Agent( + name="replay_metrics_agent", + model=TemporalModel(model_name), + tools=[weather_tool], + ) + runner = InMemoryRunner(agent=agent, app_name="replay_metrics_app") + session = await runner.session_service.create_session( + app_name="replay_metrics_app", user_id="test" + ) + last_text = None + async with Aclosing( + runner.run_async( + user_id="test", + session_id=session.id, + new_message=types.Content(role="user", parts=[types.Part(text=prompt)]), + ) + ) as agen: + async for event in agen: + if ( + event.content + and event.content.parts + and event.content.parts[0].text + ): + last_text = event.content.parts[0].text + return last_text + + +def adk_metric_counts(reader: InMemoryMetricReader) -> dict[str, int]: + """Total observation count per ADK instrument (sum of data-point counts).""" + counts: dict[str, int] = {} + data = reader.get_metrics_data() + if data is None: + return counts + for rm in data.resource_metrics: + for sm in rm.scope_metrics: + if sm.scope.name != ADK_METER_SCOPE: + continue + for metric in sm.metrics: + for dp in getattr(metric.data, "data_points", []): + counts[metric.name] = counts.get(metric.name, 0) + getattr( + dp, "count", 1 + ) + return counts + + +def adk_log_event_counts(exporter: InMemoryLogRecordExporter) -> dict[str, int]: + """Emission count per ADK log event name.""" + counts: dict[str, int] = {} + for log in exporter.get_finished_logs(): + if log.instrumentation_scope is None or ( + log.instrumentation_scope.name != ADK_METER_SCOPE + ): + continue + event_name = log.log_record.event_name + if event_name: + counts[event_name] = counts.get(event_name, 0) + 1 + return counts + + +@pytest.fixture +def reset_adk_proxy_logger(): + """Clear ADK's cached proxy logger binding around tests. + + ADK's module-level otel_logger is a proxy that caches its real logger on + first emit and never rebinds, even across a later set_logger_provider + call, so each test must clear the cache for its own provider to receive + the events. + """ + + def clear() -> None: + proxy = google.adk.telemetry.tracing.otel_logger + if hasattr(proxy, "_real_logger"): + proxy._real_logger = None # type: ignore[attr-defined] + + clear() + yield + clear() + + +async def _run_once_and_replay(client: Client, num_replays: int) -> int: + """Run the agent workflow once for real, then replay it num_replays times. + + Returns the number of real activity executions observed for this run. + Skips the calling test if the live run's history shows a workflow task + retry: a retried task legitimately re-records live telemetry + (at-least-once semantics), which would break the exact-count assertions + the callers make. + """ + LLMRegistry.register(ReplayMetricsModel) + + new_config = client.config() + new_config["plugins"] = [GoogleAdkPlugin()] + client = Client(**new_config) + + activity_executions_before = activity_executions + task_queue = f"replay-metrics-{uuid.uuid4()}" + # Deliberately not setting max_cached_workflows=0 so the live run is + # exactly one real execution with no forced replay per workflow task. + async with Worker( + client, + task_queue=task_queue, + activities=[replay_metrics_get_weather], + workflows=[ReplayMetricsAgent], + ): + handle = await client.start_workflow( + ReplayMetricsAgent.run, + args=["What is the weather in New York?", MODEL_NAME], + id=f"replay-metrics-{uuid.uuid4()}", + task_queue=task_queue, + execution_timeout=timedelta(seconds=60), + ) + result = await handle.result() + assert result == "warm and sunny" + history = await handle.fetch_history() + + if any( + event.event_type + in ( + EventType.EVENT_TYPE_WORKFLOW_TASK_FAILED, + EventType.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT, + ) + for event in history.events + ): + pytest.skip( + "Workflow task retried during the live run; exact telemetry " + "counts require a retry-free history" + ) + + for _ in range(num_replays): + await Replayer( + workflows=[ReplayMetricsAgent], + plugins=[GoogleAdkPlugin()], + ).replay_workflow(history) + + return activity_executions - activity_executions_before + + +async def test_replay_safe_meter_provider_suppresses_replay_metrics( + client: Client, + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] +): + reader = InMemoryMetricReader() + opentelemetry.metrics.set_meter_provider( + ReplaySafeMeterProvider(MeterProvider(metric_readers=[reader])) + ) + + real_executions = await _run_once_and_replay(client, num_replays=3) + + # First execution recorded exactly once (not suppressed), replays added + # zero observations, and the activity never re-executed. + assert real_executions == 1 + assert adk_metric_counts(reader) == EXPECTED_BASELINE + + +async def test_replay_metrics_duplicate_without_replay_safe_meter_provider( + client: Client, + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] +): + # Control: without the wrapper, every replay re-records every + # workflow-side ADK metric even though nothing really re-executed. + reader = InMemoryMetricReader() + opentelemetry.metrics.set_meter_provider(MeterProvider(metric_readers=[reader])) + + real_executions = await _run_once_and_replay(client, num_replays=3) + + assert real_executions == 1 + assert adk_metric_counts(reader) == { + name: count * (1 + 3) for name, count in EXPECTED_BASELINE.items() + } + + +def _in_memory_logger_provider() -> tuple[LoggerProvider, InMemoryLogRecordExporter]: + exporter = InMemoryLogRecordExporter() + provider = LoggerProvider() + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + return provider, exporter + + +async def test_replay_safe_logger_provider_suppresses_replay_log_events( + client: Client, + reset_otel_logger_provider, # type: ignore[reportUnusedParameter] + reset_adk_proxy_logger, # type: ignore[reportUnusedParameter] +): + provider, exporter = _in_memory_logger_provider() + opentelemetry._logs.set_logger_provider(ReplaySafeLoggerProvider(provider)) + + real_executions = await _run_once_and_replay(client, num_replays=3) + + # First execution emitted exactly once (not suppressed), replays added + # zero log events, and the activity never re-executed. + assert real_executions == 1 + assert adk_log_event_counts(exporter) == EXPECTED_LOG_BASELINE + + +async def test_replay_log_events_duplicate_without_replay_safe_logger_provider( + client: Client, + reset_otel_logger_provider, # type: ignore[reportUnusedParameter] + reset_adk_proxy_logger, # type: ignore[reportUnusedParameter] +): + # Control: without the wrapper, every replay re-emits every workflow-side + # ADK log event even though nothing really re-executed. + provider, exporter = _in_memory_logger_provider() + opentelemetry._logs.set_logger_provider(provider) + + real_executions = await _run_once_and_replay(client, num_replays=3) + + assert real_executions == 1 + assert adk_log_event_counts(exporter) == { + name: count * (1 + 3) for name, count in EXPECTED_LOG_BASELINE.items() + } + + +def _worker_config() -> WorkerConfig: + return WorkerConfig(workflow_runner=SandboxedWorkflowRunner()) + + +def test_plugin_warns_on_non_replay_safe_meter_provider( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] +): + opentelemetry.metrics.set_meter_provider(MeterProvider()) + with pytest.warns(UserWarning, match="MeterProvider is not replay-safe"): + GoogleAdkPlugin().configure_worker(_worker_config()) + + +def test_plugin_warns_on_non_replay_safe_tracer_provider( + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] +): + set_tracer_provider(TracerProvider()) + with pytest.warns(UserWarning, match="TracerProvider is not replay-safe"): + GoogleAdkPlugin().configure_worker(_worker_config()) + + +def test_plugin_warns_on_replayer_construction( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] +): + # Replayer replays are exactly where an unsafe global provider re-records + # telemetry, so the warning must fire there too. + opentelemetry.metrics.set_meter_provider(MeterProvider()) + with pytest.warns(UserWarning, match="MeterProvider is not replay-safe"): + Replayer(workflows=[ReplayMetricsAgent], plugins=[GoogleAdkPlugin()]) + + +def test_plugin_does_not_warn_with_replay_safe_providers( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] + reset_otel_logger_provider, # type: ignore[reportUnusedParameter] +): + opentelemetry.metrics.set_meter_provider(ReplaySafeMeterProvider(MeterProvider())) + set_tracer_provider(create_tracer_provider()) + opentelemetry._logs.set_logger_provider(ReplaySafeLoggerProvider(LoggerProvider())) + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + GoogleAdkPlugin().configure_worker(_worker_config()) + assert not [w for w in recorded if "replay-safe" in str(w.message)] + + +def test_plugin_does_not_warn_with_unset_providers( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] + reset_otel_logger_provider, # type: ignore[reportUnusedParameter] +): + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + GoogleAdkPlugin().configure_worker(_worker_config()) + assert not [w for w in recorded if "replay-safe" in str(w.message)] + + +class _DelegatingMeterProvider(ApiMeterProvider): + """Unknown custom provider delegating to a replay-safe one.""" + + def __init__(self) -> None: + self._inner = ReplaySafeMeterProvider(MeterProvider()) + + def get_meter( + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: Attributes | None = None, + ) -> Meter: + return self._inner.get_meter(name, version, schema_url) + + +def test_plugin_does_not_warn_on_unknown_custom_provider( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] +): + # A custom provider delegating to a replay-safe one is a fully replay-safe + # configuration; an unclassifiable provider must not trigger a false + # positive. + opentelemetry.metrics.set_meter_provider(_DelegatingMeterProvider()) + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + GoogleAdkPlugin().configure_worker(_worker_config()) + assert not [w for w in recorded if "replay-safe" in str(w.message)] + + +async def test_plugin_warning_points_at_worker_construction( + client: Client, + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] +): + # stacklevel on the warning must attribute it to the user's Worker(...) + # call, i.e. this file, not SDK internals. + opentelemetry.metrics.set_meter_provider(MeterProvider()) + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + Worker( + client, + task_queue=f"replay-metrics-{uuid.uuid4()}", + activities=[replay_metrics_get_weather], + workflows=[ReplayMetricsAgent], + plugins=[GoogleAdkPlugin()], + ) + warned = [ + w for w in recorded if "MeterProvider is not replay-safe" in str(w.message) + ] + assert len(warned) == 1 + assert warned[0].category is UserWarning + assert warned[0].filename == __file__ + + +class _WrappingPlugin: + """User plugin that delegates to GoogleAdkPlugin through an extra frame.""" + + def __init__(self) -> None: + self._inner = GoogleAdkPlugin() + + def configure_worker(self, config: WorkerConfig) -> WorkerConfig: + return self._inner.configure_worker(config) + + +def test_plugin_warning_points_at_wrapping_plugin_caller( + reset_otel_meter_provider, # type: ignore[reportUnusedParameter] + reset_otel_tracer_provider, # type: ignore[reportUnusedParameter] +): + # When another plugin wraps GoogleAdkPlugin, the warning must attribute + # to the nearest user frame (the wrapper's delegation line), not SDK + # internals or a fixed frame depth. + opentelemetry.metrics.set_meter_provider(MeterProvider()) + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + _WrappingPlugin().configure_worker(_worker_config()) + warned = [ + w for w in recorded if "MeterProvider is not replay-safe" in str(w.message) + ] + assert len(warned) == 1 + assert warned[0].filename == __file__ + source_lines, start = inspect.getsourcelines(_WrappingPlugin.configure_worker) + delegation_line = start + next( + offset + for offset, line in enumerate(source_lines) + if "self._inner.configure_worker" in line + ) + assert warned[0].lineno == delegation_line diff --git a/tests/contrib/opentelemetry/test_logger_provider.py b/tests/contrib/opentelemetry/test_logger_provider.py new file mode 100644 index 000000000..94c2a5d0a --- /dev/null +++ b/tests/contrib/opentelemetry/test_logger_provider.py @@ -0,0 +1,127 @@ +"""Unit tests for ReplaySafeLoggerProvider outside workflows.""" + +from typing import Any + +from opentelemetry._logs import ( + Logger, + LoggerProvider, + LogRecord, + NoOpLogger, + NoOpLoggerProvider, +) +from opentelemetry.sdk._logs import LoggerProvider as SdkLoggerProvider +from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, +) +from opentelemetry.util.types import _ExtendedAttributes + +from temporalio.contrib.opentelemetry import ReplaySafeLoggerProvider + + +def _sdk_provider() -> tuple[ReplaySafeLoggerProvider, InMemoryLogRecordExporter]: + exporter = InMemoryLogRecordExporter() + inner = SdkLoggerProvider() + inner.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + return ReplaySafeLoggerProvider(inner), exporter + + +def test_replay_safe_logger_provider_emit_passes_through_outside_workflow(): + provider, exporter = _sdk_provider() + logger = provider.get_logger("test-logger") + + logger.emit(LogRecord(event_name="record-form", body="hello")) + logger.emit(event_name="kwargs-form", body="world", attributes={"attr": "val"}) + + records = [log.log_record for log in exporter.get_finished_logs()] + assert [(r.event_name, r.body) for r in records] == [ + ("record-form", "hello"), + ("kwargs-form", "world"), + ] + assert records[1].attributes and dict(records[1].attributes) == {"attr": "val"} + + +def test_replay_safe_logger_provider_delegates_get_logger_arguments(): + class RecordingLoggerProvider(LoggerProvider): + def __init__(self) -> None: + self.calls: list[tuple] = [] + self._inner = SdkLoggerProvider() + + def get_logger( + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: _ExtendedAttributes | None = None, + ) -> Logger: + self.calls.append((name, version, schema_url, attributes)) + return self._inner.get_logger(name, version, schema_url, attributes) + + inner_provider = RecordingLoggerProvider() + provider = ReplaySafeLoggerProvider(inner_provider) + provider.get_logger( + "test-logger", + version="1.2.3", + schema_url="https://example.com/schema", + attributes={"attr": "val"}, + ) + + assert inner_provider.calls == [ + ("test-logger", "1.2.3", "https://example.com/schema", {"attr": "val"}) + ] + + +def test_replay_safe_logger_provider_supports_older_otel_signatures(): + """Newer opentelemetry-api parameters (get_logger attributes, 1.26; emit + keyword fields, 1.38) must only be forwarded when the caller passes them, + so loggers and providers with older signatures keep working.""" + + class Pre138Logger(NoOpLogger): + def __init__(self) -> None: + super().__init__("pre-1.38-logger") + self.records: list[LogRecord] = [] + + def emit(self, record: LogRecord) -> None: # type: ignore[override] + self.records.append(record) + + class Pre126LoggerProvider(NoOpLoggerProvider): + def __init__(self) -> None: + self.logger = Pre138Logger() + self.get_logger_calls: list[tuple[str, str | None, str | None]] = [] + + def get_logger( # type: ignore[override] + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + ) -> Logger: + self.get_logger_calls.append((name, version, schema_url)) + return self.logger + + inner_provider = Pre126LoggerProvider() + provider = ReplaySafeLoggerProvider(inner_provider) + record = LogRecord(event_name="event", body="hello") + provider.get_logger("test-logger").emit(record) + + assert inner_provider.get_logger_calls == [("test-logger", None, None)] + assert inner_provider.logger.records == [record] + + +def test_replay_safe_logger_provider_delegates_other_attributes(): + provider, _ = _sdk_provider() + assert provider.force_flush() + provider.shutdown() + + +def test_replay_safe_logger_provider_delegates_other_logger_attributes(): + class AttributedLogger(NoOpLogger): + def __init__(self) -> None: + super().__init__("attributed-logger") + self.custom = "custom-value" + + class AttributedLoggerProvider(NoOpLoggerProvider): + def get_logger(self, *args: Any, **kwargs: Any) -> Logger: # type: ignore[override] + return AttributedLogger() + + logger = ReplaySafeLoggerProvider(AttributedLoggerProvider()).get_logger("test") + assert logger.custom == "custom-value" # type: ignore[attr-defined] diff --git a/tests/contrib/opentelemetry/test_meter_provider.py b/tests/contrib/opentelemetry/test_meter_provider.py new file mode 100644 index 000000000..cddd88b9a --- /dev/null +++ b/tests/contrib/opentelemetry/test_meter_provider.py @@ -0,0 +1,224 @@ +"""Unit tests for ReplaySafeMeterProvider outside workflows.""" + +from collections.abc import Iterable + +from opentelemetry.context import Context +from opentelemetry.metrics import ( + CallbackOptions, + Counter, + Histogram, + Meter, + MeterProvider, + NoOpMeter, + NoOpMeterProvider, + Observation, +) +from opentelemetry.sdk.metrics import MeterProvider as SdkMeterProvider +from opentelemetry.sdk.metrics.export import InMemoryMetricReader +from opentelemetry.util.types import Attributes + +from temporalio.contrib.opentelemetry import ReplaySafeMeterProvider + + +def _metric_data_points(reader: InMemoryMetricReader) -> dict[str, list]: + points: dict[str, list] = {} + data = reader.get_metrics_data() + assert data is not None + for rm in data.resource_metrics: + for sm in rm.scope_metrics: + for metric in sm.metrics: + points.setdefault(metric.name, []).extend(metric.data.data_points) + return points + + +def test_replay_safe_meter_provider_sync_instruments_pass_through_outside_workflow(): + reader = InMemoryMetricReader() + provider = ReplaySafeMeterProvider(SdkMeterProvider(metric_readers=[reader])) + meter = provider.get_meter("test-meter") + + meter.create_counter("counter").add(2, {"attr": "val"}) + meter.create_up_down_counter("up_down_counter").add(-3) + meter.create_histogram("histogram").record(4) + meter.create_gauge("gauge").set(5) + + points = _metric_data_points(reader) + assert points["counter"][0].value == 2 + assert dict(points["counter"][0].attributes) == {"attr": "val"} + assert points["up_down_counter"][0].value == -3 + assert points["histogram"][0].count == 1 + assert points["histogram"][0].sum == 4 + assert points["gauge"][0].value == 5 + + +def test_replay_safe_meter_provider_observable_instruments_pass_through(): + def callback(options: CallbackOptions) -> Iterable[Observation]: # type: ignore[reportUnusedParameter] + return [Observation(10)] + + reader = InMemoryMetricReader() + provider = ReplaySafeMeterProvider(SdkMeterProvider(metric_readers=[reader])) + meter = provider.get_meter("test-meter") + + meter.create_observable_counter("observable_counter", callbacks=[callback]) + meter.create_observable_gauge("observable_gauge", callbacks=[callback]) + meter.create_observable_up_down_counter( + "observable_up_down_counter", callbacks=[callback] + ) + + points = _metric_data_points(reader) + assert points["observable_counter"][0].value == 10 + assert points["observable_gauge"][0].value == 10 + assert points["observable_up_down_counter"][0].value == 10 + + +def test_replay_safe_meter_provider_delegates_get_meter_arguments(): + class RecordingMeterProvider(MeterProvider): + def __init__(self) -> None: + self.calls: list[tuple] = [] + self._inner = SdkMeterProvider() + + def get_meter( + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: Attributes | None = None, + ) -> Meter: + self.calls.append((name, version, schema_url, attributes)) + return self._inner.get_meter(name, version, schema_url, attributes) + + inner_provider = RecordingMeterProvider() + provider = ReplaySafeMeterProvider(inner_provider) + meter = provider.get_meter( + "test-meter", + version="1.2.3", + schema_url="https://example.com/schema", + attributes={"attr": "val"}, + ) + + assert inner_provider.calls == [ + ("test-meter", "1.2.3", "https://example.com/schema", {"attr": "val"}) + ] + assert meter.name == "test-meter" + assert meter.version == "1.2.3" + assert meter.schema_url == "https://example.com/schema" + + +def test_replay_safe_meter_provider_supports_older_otel_signatures(): + """Newer opentelemetry-api parameters (get_meter attributes, 1.26; sync + instrument context, 1.28; create_histogram + explicit_bucket_boundaries_advisory, 1.30) must only be forwarded when + set, so providers with older signatures keep working.""" + + class Pre128Counter(Counter): + def __init__(self) -> None: + self.calls: list[tuple[int | float, Attributes | None]] = [] + + def add( # type: ignore[override] + self, + amount: int | float, + attributes: Attributes | None = None, + ) -> None: + self.calls.append((amount, attributes)) + + class Pre130Meter(NoOpMeter): + def __init__(self) -> None: + super().__init__("pre-1.30-meter") + self.counter = Pre128Counter() + self.histogram_calls: list[tuple[str, str, str]] = [] + + def create_counter( # type: ignore[override] + self, + name: str, + unit: str = "", + description: str = "", + ) -> Counter: + return self.counter + + def create_histogram( # type: ignore[override] + self, + name: str, + unit: str = "", + description: str = "", + ) -> Histogram: + self.histogram_calls.append((name, unit, description)) + return super().create_histogram(name, unit, description) + + class Pre126MeterProvider(NoOpMeterProvider): + def __init__(self) -> None: + self.meter = Pre130Meter() + self.get_meter_calls: list[tuple[str, str | None, str | None]] = [] + + def get_meter( # type: ignore[override] + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + ) -> Meter: + self.get_meter_calls.append((name, version, schema_url)) + return self.meter + + inner_provider = Pre126MeterProvider() + provider = ReplaySafeMeterProvider(inner_provider) + meter = provider.get_meter("test-meter") + meter.create_histogram("histogram").record(1) + meter.create_counter("counter").add(2, {"attr": "val"}) + + assert inner_provider.get_meter_calls == [("test-meter", None, None)] + assert inner_provider.meter.histogram_calls == [("histogram", "", "")] + assert inner_provider.meter.counter.calls == [(2, {"attr": "val"})] + + +def test_replay_safe_meter_provider_forwards_context_when_set(): + class RecordingCounter(Counter): + def __init__(self) -> None: + self.calls: list[tuple[int | float, Attributes | None, Context | None]] = [] + + def add( + self, + amount: int | float, + attributes: Attributes | None = None, + context: Context | None = None, + ) -> None: + self.calls.append((amount, attributes, context)) + + class RecordingMeter(NoOpMeter): + def __init__(self) -> None: + super().__init__("recording-meter") + self.counter = RecordingCounter() + + def create_counter( # type: ignore[override] + self, + name: str, + unit: str = "", + description: str = "", + ) -> Counter: + return self.counter + + class RecordingProvider(NoOpMeterProvider): + def __init__(self) -> None: + self.meter = RecordingMeter() + + def get_meter( # type: ignore[override] + self, + name: str, + version: str | None = None, + schema_url: str | None = None, + attributes: Attributes | None = None, + ) -> Meter: + return self.meter + + inner_provider = RecordingProvider() + provider = ReplaySafeMeterProvider(inner_provider) + context = Context() + provider.get_meter("test-meter").create_counter("counter").add( + 3, {"attr": "val"}, context + ) + + assert inner_provider.meter.counter.calls == [(3, {"attr": "val"}, context)] + + +def test_replay_safe_meter_provider_delegates_other_attributes(): + inner_provider = SdkMeterProvider() + provider = ReplaySafeMeterProvider(inner_provider) + assert provider.force_flush() + provider.shutdown() diff --git a/uv.lock b/uv.lock index c825d5488..9deca81af 100644 --- a/uv.lock +++ b/uv.lock @@ -4715,11 +4715,11 @@ requires-dist = [ { name = "mcp", marker = "extra == 'openai-agents'", specifier = ">=1.9.4,<2" }, { name = "nexus-rpc", specifier = "==1.4.0" }, { name = "openai-agents", marker = "extra == 'openai-agents'", specifier = ">=0.17.5" }, - { name = "opentelemetry-api", marker = "extra == 'lambda-worker-otel'", specifier = ">=1.11.1,<2" }, - { name = "opentelemetry-api", marker = "extra == 'opentelemetry'", specifier = ">=1.11.1,<2" }, + { name = "opentelemetry-api", marker = "extra == 'lambda-worker-otel'", specifier = ">=1.24,<2" }, + { name = "opentelemetry-api", marker = "extra == 'opentelemetry'", specifier = ">=1.24,<2" }, { name = "opentelemetry-exporter-otlp-proto-grpc", marker = "extra == 'lambda-worker-otel'", specifier = ">=1.11.1,<2" }, - { name = "opentelemetry-sdk", marker = "extra == 'lambda-worker-otel'", specifier = ">=1.11.1,<2" }, - { name = "opentelemetry-sdk", marker = "extra == 'opentelemetry'", specifier = ">=1.11.1,<2" }, + { name = "opentelemetry-sdk", marker = "extra == 'lambda-worker-otel'", specifier = ">=1.24,<2" }, + { name = "opentelemetry-sdk", marker = "extra == 'opentelemetry'", specifier = ">=1.24,<2" }, { name = "opentelemetry-sdk-extension-aws", marker = "extra == 'lambda-worker-otel'", specifier = ">=2.0.0,<3" }, { name = "opentelemetry-semantic-conventions", marker = "extra == 'lambda-worker-otel'", specifier = ">=0.40b0,<1" }, { name = "protobuf", specifier = ">=3.20,<8.0.0" },