Harden replay-safe metrics against old opentelemetry-api

- Guard the private _Gauge import (added in opentelemetry-api 1.23) and
  import ReplaySafeMeterProvider into the package lazily so
  temporalio.contrib.opentelemetry stays importable for tracing-only
  users on opentelemetry-api < 1.12, with an actionable error on access.
- Classify the global meter provider without a module-level private
  import; when the private proxy class cannot be imported the provider
  is unclassifiable and no warning is issued.
- Forward the context argument to sync instrument wrappers only when
  set; the parameter was added in opentelemetry-api 1.28 and older
  instruments reject it.
- Add explicit UserWarning category and stacklevel so provider warnings
  point at the user's Worker(...) call.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: DABH

temporalio/contrib/google_adk_agents/_plugin.py

@@ -10,8 +10,7 @@
 
 import opentelemetry.metrics
 import opentelemetry.trace
-from opentelemetry.metrics import NoOpMeterProvider
-from opentelemetry.metrics._internal import _ProxyMeterProvider
+from opentelemetry.metrics import MeterProvider, NoOpMeterProvider
 from opentelemetry.trace import NoOpTracerProvider, ProxyTracerProvider
 
 from temporalio import workflow
@@ -37,23 +36,47 @@
 from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner
 
 
+def _meter_provider_replay_safe(provider: MeterProvider) -> bool | None:
+    """Classify the global meter provider for replay safety.
+
+    Returns True when replay-safe (including providers that drop all
+    recordings), False when positively identified as replay-unsafe, and None
+    when it cannot be classified, in which case no warning should be issued
+    since a false positive is worse than a missed warning.
+    """
+    if isinstance(provider, (ReplaySafeMeterProvider, NoOpMeterProvider)):
+        return True
+    try:
+        # Unlike tracing's public ProxyTracerProvider, the proxy (unset) meter
+        # provider has no public counterpart. Import the private class lazily
+        # so a moved or removed symbol cannot break module import; it is
+        # present in opentelemetry-api 1.12 through at least 1.42.
+        from opentelemetry.metrics._internal import _ProxyMeterProvider
+    except ImportError:
+        return None
+    return isinstance(provider, _ProxyMeterProvider)
+
+
 def _warn_if_global_otel_providers_not_replay_safe() -> None:
     # ADK records metrics and spans 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. Unset (proxy)
     # and no-op providers drop recordings and are fine.
+    #
+    # stacklevel=4 attributes the warnings to the user's Worker(...) call:
+    # warn -> this helper -> GoogleAdkPlugin.configure_worker ->
+    # Worker.__init__ -> user code.
     meter_provider = opentelemetry.metrics.get_meter_provider()
-    if not isinstance(
-        meter_provider,
-        (ReplaySafeMeterProvider, NoOpMeterProvider, _ProxyMeterProvider),
-    ):
+    if _meter_provider_replay_safe(meter_provider) is False:
         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))"
+            "opentelemetry.metrics.set_meter_provider(ReplaySafeMeterProvider(provider))",
+            UserWarning,
+            stacklevel=4,
         )
     tracer_provider = opentelemetry.trace.get_tracer_provider()
     if not isinstance(
@@ -65,7 +88,9 @@ def _warn_if_global_otel_providers_not_replay_safe() -> None:
             "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())"
+            "temporalio.contrib.opentelemetry.create_tracer_provider())",
+            UserWarning,
+            stacklevel=4,
         )
 
 

temporalio/contrib/opentelemetry/__init__.py

@@ -5,11 +5,24 @@
 propagation for distributed tracing.
 """
 
+from typing import Any
+
 from temporalio.contrib.opentelemetry._interceptor import (
     TracingInterceptor,
     TracingWorkflowInboundInterceptor,
 )
-from temporalio.contrib.opentelemetry._meter_provider import ReplaySafeMeterProvider
+
+_meter_provider_import_error: ImportError | None = None
+try:
+    from temporalio.contrib.opentelemetry._meter_provider import (
+        ReplaySafeMeterProvider,
+    )
+except ImportError as err:
+    # opentelemetry-api < 1.12 has no opentelemetry.metrics module. Keep the
+    # tracing integration importable and raise a clear error only when
+    # ReplaySafeMeterProvider is actually accessed (see __getattr__ below).
+    _meter_provider_import_error = err
+
 from temporalio.contrib.opentelemetry._otel_interceptor import OpenTelemetryInterceptor
 from temporalio.contrib.opentelemetry._plugin import OpenTelemetryPlugin
 from temporalio.contrib.opentelemetry._tracer_provider import (
@@ -26,3 +39,16 @@
     "ReplaySafeTracerProvider",
     "create_tracer_provider",
 ]
+
+
+def __getattr__(name: str) -> Any:
+    # Only reachable for ReplaySafeMeterProvider when the guarded import above
+    # failed; otherwise the module attribute exists and this is never called.
+    if name == "ReplaySafeMeterProvider":
+        raise ImportError(
+            "ReplaySafeMeterProvider requires the OpenTelemetry metrics API "
+            "(opentelemetry.metrics), which the installed opentelemetry-api "
+            "version does not provide. Install opentelemetry-api >= 1.12 "
+            "(>= 1.23 for synchronous gauge support)."
+        ) from _meter_provider_import_error
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

temporalio/contrib/opentelemetry/_meter_provider.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from opentelemetry.context import Context
 from opentelemetry.metrics import (
@@ -12,13 +12,32 @@
     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}
+
+
+if TYPE_CHECKING:
+    from opentelemetry.metrics import _Gauge
+else:
+    try:
+        from opentelemetry.metrics import _Gauge
+    except ImportError:
+        # Synchronous gauges (and Meter.create_gauge) were added in
+        # opentelemetry-api 1.23. Fall back to a plain base class so this
+        # module stays importable on older versions in the supported range;
+        # meters there have no create_gauge to wrap anyway.
+        _Gauge = object
+
+
 def _skip_recording() -> bool:
     # in_workflow() must be evaluated first: is_replaying() requires an active
     # workflow context.
@@ -41,7 +60,7 @@ def add(
         if _skip_recording():
             # Skip recording metrics during workflow replay to avoid duplicate telemetry
             return
-        self._counter.add(amount, attributes, context)
+        self._counter.add(amount, attributes, **_forward_context_kwarg(context))
 
 
 class _ReplaySafeUpDownCounter(UpDownCounter):
@@ -60,7 +79,7 @@ def add(
         if _skip_recording():
             # Skip recording metrics during workflow replay to avoid duplicate telemetry
             return
-        self._counter.add(amount, attributes, context)
+        self._counter.add(amount, attributes, **_forward_context_kwarg(context))
 
 
 class _ReplaySafeHistogram(Histogram):
@@ -79,7 +98,7 @@ def record(
         if _skip_recording():
             # Skip recording metrics during workflow replay to avoid duplicate telemetry
             return
-        self._histogram.record(amount, attributes, context)
+        self._histogram.record(amount, attributes, **_forward_context_kwarg(context))
 
 
 class _ReplaySafeGauge(_Gauge):
@@ -98,7 +117,7 @@ def set(
         if _skip_recording():
             # Skip recording metrics during workflow replay to avoid duplicate telemetry
             return
-        self._gauge.set(amount, attributes, context)
+        self._gauge.set(amount, attributes, **_forward_context_kwarg(context))
 
 
 class _ReplaySafeMeter(Meter):
@@ -154,6 +173,8 @@ def create_gauge(
         unit: str = "",
         description: str = "",
     ) -> _Gauge:
+        # On opentelemetry-api < 1.23 the underlying meter has no create_gauge
+        # and this raises AttributeError, same as calling it on the raw meter.
         return _ReplaySafeGauge(self._meter.create_gauge(name, unit, description))
 
     # Observable instruments pass through unwrapped: their callbacks run on the

tests/contrib/google_adk_agents/test_replay_metrics.py

@@ -9,6 +9,7 @@
 recordings while leaving first-execution recordings intact.
 """
 
+import sys
 import uuid
 import warnings
 from collections.abc import AsyncGenerator
@@ -23,6 +24,8 @@
 from google.adk.runners import InMemoryRunner
 from google.adk.utils.context_utils import Aclosing
 from google.genai import types
+from opentelemetry.metrics import NoOpMeterProvider
+from opentelemetry.metrics._internal import _ProxyMeterProvider
 from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.metrics.export import InMemoryMetricReader
 from opentelemetry.sdk.trace import TracerProvider
@@ -287,3 +290,63 @@ def test_plugin_does_not_warn_with_unset_providers(
         warnings.simplefilter("always")
         GoogleAdkPlugin().configure_worker(_worker_config())
     assert not [w for w in recorded if "replay-safe" in str(w.message)]
+
+
+def test_meter_provider_replay_safety_classification(monkeypatch: pytest.MonkeyPatch):
+    from temporalio.contrib.google_adk_agents._plugin import (
+        _meter_provider_replay_safe,
+    )
+
+    assert _meter_provider_replay_safe(ReplaySafeMeterProvider(MeterProvider())) is True
+    assert _meter_provider_replay_safe(NoOpMeterProvider()) is True
+    assert _meter_provider_replay_safe(_ProxyMeterProvider()) is True
+    assert _meter_provider_replay_safe(MeterProvider()) is False
+
+    # When the private proxy class cannot be imported, providers other than
+    # the replay-safe/no-op ones are unclassifiable and must not warn.
+    monkeypatch.setitem(sys.modules, "opentelemetry.metrics._internal", None)
+    assert _meter_provider_replay_safe(MeterProvider()) is None
+    assert _meter_provider_replay_safe(NoOpMeterProvider()) is True
+
+
+def test_plugin_does_not_warn_when_proxy_detection_unavailable(
+    monkeypatch: pytest.MonkeyPatch,
+    reset_otel_meter_provider,  # type: ignore[reportUnusedParameter]
+    reset_otel_tracer_provider,  # type: ignore[reportUnusedParameter]
+):
+    # A replay-unsafe provider is set, but with the private proxy class
+    # unimportable it cannot be classified, so the plugin stays silent rather
+    # than risking a false positive.
+    opentelemetry.metrics.set_meter_provider(MeterProvider())
+    monkeypatch.setitem(sys.modules, "opentelemetry.metrics._internal", None)
+    with warnings.catch_warnings(record=True) as recorded:
+        warnings.simplefilter("always")
+        GoogleAdkPlugin().configure_worker(_worker_config())
+    assert not [
+        w for w in recorded if "MeterProvider is not 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__