From 3a9444522618931a1286563e943f377342fdb98f Mon Sep 17 00:00:00 2001 From: harsh543 Date: Wed, 29 Jul 2026 08:21:13 -0700 Subject: [PATCH 1/2] Add OpenTelemetry metrics exporter built on MetricBuffer Resolves #1049 as scoped by @Sushisource: drain a MetricBuffer into a real OTel MeterProvider (views, resources, exemplars all come for free) rather than wiring MetricMeter across a multiprocessing queue. MetricsExporter polls MetricBuffer.retrieve_updates() on a fixed interval and maps buffered updates onto the OTel metrics API: counters via Counter.add() (buffered counter values are deltas), histograms via Histogram.record(), and gauges via create_observable_gauge() backed by a lock-protected last-value cache (OTel invokes gauge callbacks from its own export thread, concurrently with the drain loop). Instrument and attribute objects are cached keyed by id(), exploiting the identity-stability guarantee BufferedMetric/attributes already document. Lifecycle (run/shutdown/async context manager) mirrors Worker's existing asyncio.Event-based shutdown pattern rather than using threads, since that's the only such pattern already in this codebase. Bumps the opentelemetry extra's floor from >=1.11.1 to >=1.12.0 for both api and sdk -- the public opentelemetry.metrics module simply doesn't exist at 1.11.1, confirmed by direct import against the wheel. uv.lock is left untouched; CI resolves it fresh via `uv sync --all-extras` with no --frozen/--locked gate on PR checks. Co-Authored-By: Claude Sonnet 5 --- README.md | 33 ++ pyproject.toml | 2 +- temporalio/contrib/opentelemetry/README.md | 25 +- temporalio/contrib/opentelemetry/__init__.py | 10 +- .../opentelemetry/_metrics_exporter.py | 282 ++++++++++++++++++ .../opentelemetry/test_metrics_exporter.py | 194 ++++++++++++ 6 files changed, 541 insertions(+), 5 deletions(-) create mode 100644 temporalio/contrib/opentelemetry/_metrics_exporter.py create mode 100644 tests/contrib/opentelemetry/test_metrics_exporter.py diff --git a/README.md b/README.md index 7a1caedd9..7eb26db4c 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,7 @@ informal introduction to the features and their implementation. - [Observability](#observability) - [Metrics](#metrics) - [OpenTelemetry Tracing](#opentelemetry-tracing) + - [OpenTelemetry Metrics](#opentelemetry-metrics) - [Protobuf 3.x vs 4.x](#protobuf-3x-vs-4x) - [Known Compatibility Issues](#known-compatibility-issues) - [gevent Patching](#gevent-patching) @@ -1990,6 +1991,38 @@ as an interceptor on the `interceptors` argument of `Client.connect`. When set, calls and for all activity and workflow invocations on the worker, spans will be created and properly serialized through the server to give one proper trace for a workflow execution. +#### OpenTelemetry Metrics + +Metrics support also requires the `opentelemetry` extra (see above). Rather than using +`temporalio.runtime.PrometheusConfig` or `temporalio.runtime.OpenTelemetryConfig`, set a +`temporalio.runtime.MetricBuffer` as the `metrics` on `TelemetryConfig`, then drain it into a real OpenTelemetry +`MeterProvider` using `temporalio.contrib.opentelemetry.MetricsExporter`: + +```python +from datetime import timedelta + +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader + +from temporalio.client import Client +from temporalio.contrib.opentelemetry import MetricsExporter +from temporalio.runtime import MetricBuffer, Runtime, TelemetryConfig + +buffer = MetricBuffer(10_000) +runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer)) +meter_provider = MeterProvider( + metric_readers=[PeriodicExportingMetricReader(ConsoleMetricExporter(), export_interval_millis=5000)] +) + +async with MetricsExporter(buffer, meter_provider): + client = await Client.connect("localhost:7233", runtime=runtime) + # ... run workers/workflows while the exporter drains the buffer in the background +``` + +`MetricsExporter` must be running (via `async with` or manual `run()`/`shutdown()`) for as long as metrics should be +exported, since it works by draining the buffer on a fixed interval (`poll_interval`, default one second) -- per the +warning on `MetricBuffer`, updates are dropped if the buffer isn't drained regularly. + ### Protobuf 3.x vs 4.x Python currently has two somewhat-incompatible protobuf library versions - the 3.x series and the 4.x series. Python diff --git a/pyproject.toml b/pyproject.toml index d5de4c077..c646bd62f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ 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.12.0,<2", "opentelemetry-sdk>=1.12.0,<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", "mcp>=1.24,<2"] diff --git a/temporalio/contrib/opentelemetry/README.md b/temporalio/contrib/opentelemetry/README.md index 2c6e39817..d242c2625 100644 --- a/temporalio/contrib/opentelemetry/README.md +++ b/temporalio/contrib/opentelemetry/README.md @@ -1,6 +1,29 @@ # OpenTelemetry Integration for Temporal Python SDK -This package provides OpenTelemetry tracing integration for Temporal workflows, activities, and other operations. It includes automatic span creation and propagation for distributed tracing across your Temporal applications. +This package provides OpenTelemetry tracing and metrics integration for Temporal workflows, activities, and other operations. It includes automatic span creation and propagation for distributed tracing, and a `MetricsExporter` for exporting Temporal SDK/Core metrics through the standard OpenTelemetry metrics API, across your Temporal applications. + +## Metrics + +`MetricsExporter` drains a `temporalio.runtime.MetricBuffer` into an OpenTelemetry `MeterProvider`, so Temporal's own SDK/Core metrics (and any custom metrics recorded via `activity.metric_meter()`/`workflow.metric_meter()`) can be exported through the standard OpenTelemetry metrics pipeline (views, resources, any OTel-compatible backend) instead of only through `PrometheusConfig`/`OpenTelemetryConfig`. + +```python +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader +from temporalio.contrib.opentelemetry import MetricsExporter +from temporalio.runtime import MetricBuffer, Runtime, TelemetryConfig + +buffer = MetricBuffer(10_000) +runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer)) +meter_provider = MeterProvider( + metric_readers=[PeriodicExportingMetricReader(ConsoleMetricExporter())] +) + +async with MetricsExporter(buffer, meter_provider): + client = await Client.connect("localhost:7233", runtime=runtime) + ... +``` + +**Note:** the `Runtime` must be constructed with the buffer attached *before* `MetricsExporter` is started, and the exporter must keep running (it polls on a fixed interval) for as long as metrics should be exported. ## Overview diff --git a/temporalio/contrib/opentelemetry/__init__.py b/temporalio/contrib/opentelemetry/__init__.py index 74f069322..e2d11c523 100644 --- a/temporalio/contrib/opentelemetry/__init__.py +++ b/temporalio/contrib/opentelemetry/__init__.py @@ -1,14 +1,17 @@ """OpenTelemetry v2 integration for Temporal SDK. -This package provides OpenTelemetry tracing integration for Temporal workflows, -activities, and other operations. It includes automatic span creation and -propagation for distributed tracing. +This package provides OpenTelemetry tracing and metrics integration for +Temporal workflows, activities, and other operations. It includes automatic +span creation and propagation for distributed tracing, and a +:py:class:`MetricsExporter` for exporting Temporal SDK/Core metrics through +the standard OpenTelemetry metrics API. """ from temporalio.contrib.opentelemetry._interceptor import ( TracingInterceptor, TracingWorkflowInboundInterceptor, ) +from temporalio.contrib.opentelemetry._metrics_exporter import MetricsExporter 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 @@ -18,5 +21,6 @@ "TracingWorkflowInboundInterceptor", "OpenTelemetryInterceptor", "OpenTelemetryPlugin", + "MetricsExporter", "create_tracer_provider", ] diff --git a/temporalio/contrib/opentelemetry/_metrics_exporter.py b/temporalio/contrib/opentelemetry/_metrics_exporter.py new file mode 100644 index 000000000..7566bb6cc --- /dev/null +++ b/temporalio/contrib/opentelemetry/_metrics_exporter.py @@ -0,0 +1,282 @@ +"""OpenTelemetry metrics exporter built on :py:class:`temporalio.runtime.MetricBuffer`.""" + +import asyncio +import contextlib +import logging +import threading +from collections.abc import Callable, Iterable, Mapping +from dataclasses import dataclass, field +from datetime import timedelta + +import opentelemetry.metrics +import opentelemetry.util.types + +import temporalio.common +import temporalio.runtime + +logger = logging.getLogger(__name__) + + +@dataclass +class _GaugeState: + lock: threading.Lock = field(default_factory=threading.Lock) + # Keyed by id(BufferedMetricUpdate.attributes) so repeated updates for the + # same attribute set overwrite rather than accumulate -- gauges are + # "latest value wins", unlike counters. + last_values: dict[ + int, tuple[Mapping[str, opentelemetry.util.types.AttributeValue], int | float] + ] = field(default_factory=dict) + + +@dataclass +class _Instrument: + kind: temporalio.runtime.BufferedMetricKind + counter: opentelemetry.metrics.Counter | None = None + histogram: opentelemetry.metrics.Histogram | None = None + gauge_state: "_GaugeState | None" = None + + +def _make_gauge_callback( + state: _GaugeState, +) -> Callable[ + [opentelemetry.metrics.CallbackOptions], + Iterable[opentelemetry.metrics.Observation], +]: + # OTel invokes observable-gauge callbacks from its own export thread, which + # runs concurrently with the drain loop mutating last_values -- hence the + # lock on both sides. + def callback( + options: opentelemetry.metrics.CallbackOptions, + ) -> Iterable[opentelemetry.metrics.Observation]: + with state.lock: + return [ + opentelemetry.metrics.Observation(value, attributes=dict(attrs)) + for attrs, value in state.last_values.values() + ] + + return callback + + +class MetricsExporter: + """Exports metrics recorded via a :py:class:`temporalio.runtime.MetricBuffer` + to an OpenTelemetry :py:class:`opentelemetry.metrics.MeterProvider`. + + .. warning:: + This class is experimental and may change in future versions. + Use with caution in production environments. + + This must be started (via :py:meth:`run` or ``async with``) after the + :py:class:`temporalio.runtime.Runtime` referencing ``buffer`` has been + constructed, and it must be kept running for as long as metrics should be + exported. It drains the buffer on a fixed interval; per + :py:class:`temporalio.runtime.MetricBuffer`, updates are dropped (with an + error logged by Core) if the buffer is not drained regularly. + + Example: + + .. code-block:: python + + from datetime import timedelta + + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import ( + ConsoleMetricExporter, + PeriodicExportingMetricReader, + ) + + from temporalio.client import Client + from temporalio.contrib.opentelemetry import MetricsExporter + from temporalio.runtime import MetricBuffer, Runtime, TelemetryConfig + + buffer = MetricBuffer(10_000) + runtime = Runtime(telemetry=TelemetryConfig(metrics=buffer)) + meter_provider = MeterProvider( + metric_readers=[ + PeriodicExportingMetricReader( + ConsoleMetricExporter(), export_interval_millis=5000 + ) + ] + ) + + async with MetricsExporter(buffer, meter_provider): + client = await Client.connect("localhost:7233", runtime=runtime) + ... + """ + + def __init__( + self, + buffer: temporalio.runtime.MetricBuffer, + meter_provider: opentelemetry.metrics.MeterProvider | None = None, + *, + meter_name: str = "temporalio", + meter_version: str | None = None, + poll_interval: timedelta = timedelta(seconds=1), + on_error: Callable[[Exception], None] | None = None, + ) -> None: + """Create an exporter that drains ``buffer`` into ``meter_provider``. + + Args: + buffer: The buffer to drain. Must already be (or about to be) set + as the ``metrics`` of a + :py:class:`temporalio.runtime.TelemetryConfig` on a + constructed :py:class:`temporalio.runtime.Runtime`. + meter_provider: The provider to create instruments on. Defaults to + :py:func:`opentelemetry.metrics.get_meter_provider` (the + global provider) if not given. + meter_name: Name passed to ``get_meter`` on the provider. + meter_version: Version passed to ``get_meter`` on the provider. + poll_interval: How often to drain the buffer. Must be reasonably + frequent -- see the warning on + :py:class:`temporalio.runtime.MetricBuffer`. + on_error: Optional callback invoked (in addition to logging) when + draining the buffer or applying an individual update fails. + If this callback itself raises, that is logged and ignored. + """ + self._buffer = buffer + self._meter = ( + meter_provider or opentelemetry.metrics.get_meter_provider() + ).get_meter(meter_name, meter_version) + self._poll_interval = poll_interval + self._on_error = on_error + self._instruments: dict[int, _Instrument] = {} + self._attributes_cache: dict[ + int, Mapping[str, opentelemetry.util.types.AttributeValue] + ] = {} + self._started = False + self._shutdown_event = asyncio.Event() + self._run_complete_event = asyncio.Event() + self._run_task: asyncio.Task | None = None + + async def run(self) -> None: + """Drain and export on ``poll_interval`` until :py:meth:`shutdown` is called. + + Raises: + RuntimeError: If ``buffer`` was never attached to a constructed + :py:class:`temporalio.runtime.Runtime`, or if this is called + while already running. + """ + if self._started: + raise RuntimeError("MetricsExporter is already running") + self._started = True + self._shutdown_event.clear() + self._run_complete_event.clear() + try: + while not self._shutdown_event.is_set(): + self._drain_once() + with contextlib.suppress(asyncio.TimeoutError): + await asyncio.wait_for( + self._shutdown_event.wait(), + timeout=self._poll_interval.total_seconds(), + ) + # Final drain so nothing recorded right before shutdown is lost. + self._drain_once() + finally: + self._started = False + self._run_complete_event.set() + + async def shutdown(self) -> None: + """Stop :py:meth:`run` (including a final drain) and wait for it to finish.""" + self._shutdown_event.set() + await self._run_complete_event.wait() + + async def __aenter__(self) -> "MetricsExporter": + """Start :py:meth:`run` as a background task.""" + self._run_task = asyncio.ensure_future(self.run()) + return self + + async def __aexit__(self, *exc_info: object) -> None: + """Call :py:meth:`shutdown` and await the background task.""" + await self.shutdown() + if self._run_task: + await self._run_task + + def _drain_once(self) -> None: + try: + updates = self._buffer.retrieve_updates() + except RuntimeError: + # Buffer was never attached to a constructed Runtime -- this is a + # setup error, not a transient failure, so it should propagate + # rather than spin silently and drop metrics forever. + raise + except Exception as err: + self._handle_error(err) + return + + for update in updates: + try: + self._apply_update(update) + except Exception as err: + self._handle_error(err) + + def _handle_error(self, err: Exception) -> None: + logger.exception("Error exporting buffered Temporal metrics") + if self._on_error: + try: + self._on_error(err) + except Exception: + logger.exception("on_error handler failed") + + def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None: + instrument = self._get_or_create_instrument(update.metric) + attributes = self._get_or_create_attributes(update.attributes) + + if instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_COUNTER: + assert instrument.counter is not None + instrument.counter.add(update.value, attributes=attributes) + elif instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_HISTOGRAM: + assert instrument.histogram is not None + instrument.histogram.record(update.value, attributes=attributes) + elif instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_GAUGE: + assert instrument.gauge_state is not None + with instrument.gauge_state.lock: + instrument.gauge_state.last_values[id(update.attributes)] = ( + attributes, + update.value, + ) + else: + raise ValueError(f"Unrecognized buffered metric kind: {instrument.kind}") + + def _get_or_create_instrument( + self, metric: temporalio.runtime.BufferedMetric + ) -> _Instrument: + instrument = self._instruments.get(id(metric)) + if instrument is not None: + return instrument + + unit = metric.unit or "" + description = metric.description or "" + instrument = _Instrument(kind=metric.kind) + if metric.kind == temporalio.runtime.BUFFERED_METRIC_KIND_COUNTER: + instrument.counter = self._meter.create_counter( + metric.name, unit=unit, description=description + ) + elif metric.kind == temporalio.runtime.BUFFERED_METRIC_KIND_HISTOGRAM: + instrument.histogram = self._meter.create_histogram( + metric.name, unit=unit, description=description + ) + elif metric.kind == temporalio.runtime.BUFFERED_METRIC_KIND_GAUGE: + gauge_state = _GaugeState() + instrument.gauge_state = gauge_state + self._meter.create_observable_gauge( + metric.name, + callbacks=[_make_gauge_callback(gauge_state)], + unit=unit, + description=description, + ) + else: + raise ValueError(f"Unrecognized buffered metric kind: {metric.kind}") + + self._instruments[id(metric)] = instrument + return instrument + + def _get_or_create_attributes( + self, attributes: temporalio.common.MetricAttributes + ) -> Mapping[str, opentelemetry.util.types.AttributeValue]: + cached = self._attributes_cache.get(id(attributes)) + if cached is not None: + return cached + converted: Mapping[str, opentelemetry.util.types.AttributeValue] = dict( + attributes + ) + self._attributes_cache[id(attributes)] = converted + return converted diff --git a/tests/contrib/opentelemetry/test_metrics_exporter.py b/tests/contrib/opentelemetry/test_metrics_exporter.py new file mode 100644 index 000000000..1d045e185 --- /dev/null +++ b/tests/contrib/opentelemetry/test_metrics_exporter.py @@ -0,0 +1,194 @@ +from __future__ import annotations + +import asyncio +from typing import Any + +import pytest +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics.export import InMemoryMetricReader + +from temporalio.contrib.opentelemetry import MetricsExporter +from temporalio.runtime import MetricBuffer, Runtime, TelemetryConfig + + +def _make_runtime_and_exporter( + **telemetry_kwargs: Any, +) -> tuple[Runtime, MetricBuffer, MeterProvider, InMemoryMetricReader, MetricsExporter]: + buffer = MetricBuffer(10_000) + runtime = Runtime( + telemetry=TelemetryConfig( + metrics=buffer, attach_service_name=False, **telemetry_kwargs + ) + ) + reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[reader]) + exporter = MetricsExporter(buffer, meter_provider) + return runtime, buffer, meter_provider, reader, exporter + + +def _collect_metrics_by_name(reader: InMemoryMetricReader) -> dict[str, Any]: + data = reader.get_metrics_data() + result: dict[str, Any] = {} + if not data: + return result + for resource_metrics in data.resource_metrics: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + result[metric.name] = metric + return result + + +async def test_metrics_exporter_counter(): + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + + counter = runtime.metric_meter.create_counter( + "my-counter", "my-counter-desc", "my-counter-unit" + ) + counter.add(100) + counter.with_additional_attributes({"foo": "bar"}).add(200) + + exporter._drain_once() + + metrics = _collect_metrics_by_name(reader) + assert "my-counter" in metrics + metric = metrics["my-counter"] + assert metric.description == "my-counter-desc" + assert metric.unit == "my-counter-unit" + points = { + tuple(sorted(p.attributes.items())): p.value for p in metric.data.data_points + } + assert points[()] == 100 + assert points[(("foo", "bar"),)] == 200 + + # Draining again with no new adds should not change values (counters + # accumulate deltas across drains, not reset). + exporter._drain_once() + metrics = _collect_metrics_by_name(reader) + points = { + tuple(sorted(p.attributes.items())): p.value + for p in metrics["my-counter"].data.data_points + } + assert points[()] == 100 + assert points[(("foo", "bar"),)] == 200 + + # More adds accumulate on top. + counter.add(50) + exporter._drain_once() + metrics = _collect_metrics_by_name(reader) + points = { + tuple(sorted(p.attributes.items())): p.value + for p in metrics["my-counter"].data.data_points + } + assert points[()] == 150 + + +async def test_metrics_exporter_gauge_reflects_latest_value(): + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + + gauge = runtime.metric_meter.create_gauge_float("my-gauge") + gauge.set(1.5) + exporter._drain_once() + + metrics = _collect_metrics_by_name(reader) + points = list(metrics["my-gauge"].data.data_points) + assert len(points) == 1 + assert points[0].value == 1.5 + + # Setting a new value should replace, not add to, the prior one. + gauge.set(9.5) + exporter._drain_once() + metrics = _collect_metrics_by_name(reader) + points = list(metrics["my-gauge"].data.data_points) + assert len(points) == 1 + assert points[0].value == 9.5 + + +async def test_metrics_exporter_histogram(): + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + + histogram = runtime.metric_meter.create_histogram_float("my-histogram") + histogram.record(1.0) + histogram.record(2.0) + histogram.record(3.0) + exporter._drain_once() + + metrics = _collect_metrics_by_name(reader) + points = list(metrics["my-histogram"].data.data_points) + assert len(points) == 1 + assert points[0].count == 3 + assert points[0].sum == 6.0 + + +async def test_metrics_exporter_attributes_passthrough(): + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + + counter = runtime.metric_meter.create_counter("attrs-counter") + counter.with_additional_attributes({"foo": "bar", "baz": 123}).add(1) + exporter._drain_once() + + metrics = _collect_metrics_by_name(reader) + points = list(metrics["attrs-counter"].data.data_points) + assert len(points) == 1 + assert dict(points[0].attributes) == {"foo": "bar", "baz": 123} + + +async def test_metrics_exporter_buffer_not_attached_raises(): + unattached_buffer = MetricBuffer(10_000) + reader = InMemoryMetricReader() + meter_provider = MeterProvider(metric_readers=[reader]) + exporter = MetricsExporter(unattached_buffer, meter_provider) + + with pytest.raises(RuntimeError): + exporter._drain_once() + + with pytest.raises(RuntimeError): + await exporter.run() + + +async def test_metrics_exporter_run_shutdown_lifecycle(): + from datetime import timedelta + + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + exporter._poll_interval = timedelta(milliseconds=10) + + counter = runtime.metric_meter.create_counter("lifecycle-counter") + + async with exporter: + counter.add(42) + # Give the background poller a couple of cycles to drain. + await asyncio.sleep(0.2) + + metrics = _collect_metrics_by_name(reader) + points = list(metrics["lifecycle-counter"].data.data_points) + assert points[0].value == 42 + + # After exit, the background task must actually be done (not leaked). + assert exporter._run_task is not None + assert exporter._run_task.done() + + +async def test_metrics_exporter_error_in_one_update_does_not_block_others(): + runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + + good_counter = runtime.metric_meter.create_counter("good-counter") + bad_counter = runtime.metric_meter.create_counter("bad-counter") + good_counter.add(1) + bad_counter.add(1) + + errors: list[Exception] = [] + exporter._on_error = errors.append + + original_apply_update = exporter._apply_update + + def flaky_apply_update(update: Any) -> None: + if update.metric.name == "bad-counter": + raise ValueError("simulated failure applying bad-counter") + original_apply_update(update) + + exporter._apply_update = flaky_apply_update # type: ignore[method-assign] + exporter._drain_once() + + assert len(errors) == 1 + metrics = _collect_metrics_by_name(reader) + assert "good-counter" in metrics + assert "bad-counter" not in metrics From 4dd89cca22830dc54a6d6b61725156e0c1e91de6 Mon Sep 17 00:00:00 2001 From: harsh543 Date: Fri, 14 Aug 2026 23:50:38 -0700 Subject: [PATCH 2/2] Fix P1 review findings: shutdown race, lambda-otel floor, changelog - run() cleared the shutdown-request event at startup, which could erase a shutdown() call made before the background task got a chance to run, causing async with to hang forever on immediate enter/exit. Only clear it after the loop exits instead. - Raise the lambda-worker-otel extra's opentelemetry-api/-sdk floor to >=1.12.0 to match the main opentelemetry extra, since 1.11.1 lacks the public opentelemetry.metrics module this now unconditionally imports. - Add a changelog entry for the new MetricsExporter. --- CHANGELOG.md | 4 +++ README.md | 10 ++++--- pyproject.toml | 4 +-- .../opentelemetry/_metrics_exporter.py | 13 ++++---- .../opentelemetry/test_metrics_exporter.py | 30 +++++++++++++++---- 5 files changed, 44 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f97062e39..6bb9ad2b5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -90,6 +90,10 @@ to include examples, links to docs, or any other relevant information. This lets types with transfer type converters delegate their wire representation to the configured payload converter, preserving SDK behavior such as serialization contexts. +- Added `temporalio.contrib.opentelemetry.MetricsExporter`, which drains a + `temporalio.runtime.MetricBuffer` on a fixed interval and exports through a + real OpenTelemetry `MeterProvider`, giving SDK/Core metrics access to + standard OTel features (views, resource, exemplars). Experimental. - Added `TLSConfig.verification_server_name` to verify the server certificate against a fixed name instead of the connection's server name. Unlike `domain`, it does not change the TLS SNI or HTTP/2 authority values, which keep following the connected host, so it can be used when the diff --git a/README.md b/README.md index 7eb26db4c..fc1faf83b 100644 --- a/README.md +++ b/README.md @@ -1993,10 +1993,12 @@ the server to give one proper trace for a workflow execution. #### OpenTelemetry Metrics -Metrics support also requires the `opentelemetry` extra (see above). Rather than using -`temporalio.runtime.PrometheusConfig` or `temporalio.runtime.OpenTelemetryConfig`, set a -`temporalio.runtime.MetricBuffer` as the `metrics` on `TelemetryConfig`, then drain it into a real OpenTelemetry -`MeterProvider` using `temporalio.contrib.opentelemetry.MetricsExporter`: +Metrics support also requires the `opentelemetry` extra (see above). If you want your Temporal SDK/Core metrics (and +any custom metrics recorded via `activity.metric_meter()`/`workflow.metric_meter()`) to flow through the standard +OpenTelemetry metrics pipeline (views, resources, any OTel-compatible backend) rather than only through +`PrometheusConfig`/`OpenTelemetryConfig`, set a `temporalio.runtime.MetricBuffer` as the `metrics` on +`TelemetryConfig`, then drain it into a real OpenTelemetry `MeterProvider` using +`temporalio.contrib.opentelemetry.MetricsExporter`: ```python from datetime import timedelta diff --git a/pyproject.toml b/pyproject.toml index c646bd62f..d980a5f46 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,8 +38,8 @@ deepagents = [ "langchain-core>=1.4.8,<2; python_version >= '3.11'", ] lambda-worker-otel = [ - "opentelemetry-api>=1.11.1,<2", - "opentelemetry-sdk>=1.11.1,<2", + "opentelemetry-api>=1.12.0,<2", + "opentelemetry-sdk>=1.12.0,<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/opentelemetry/_metrics_exporter.py b/temporalio/contrib/opentelemetry/_metrics_exporter.py index 7566bb6cc..32f17be9a 100644 --- a/temporalio/contrib/opentelemetry/_metrics_exporter.py +++ b/temporalio/contrib/opentelemetry/_metrics_exporter.py @@ -46,7 +46,7 @@ def _make_gauge_callback( # runs concurrently with the drain loop mutating last_values -- hence the # lock on both sides. def callback( - options: opentelemetry.metrics.CallbackOptions, + _options: opentelemetry.metrics.CallbackOptions, ) -> Iterable[opentelemetry.metrics.Observation]: with state.lock: return [ @@ -72,9 +72,7 @@ class MetricsExporter: :py:class:`temporalio.runtime.MetricBuffer`, updates are dropped (with an error logged by Core) if the buffer is not drained regularly. - Example: - - .. code-block:: python + Example:: from datetime import timedelta @@ -158,7 +156,6 @@ async def run(self) -> None: if self._started: raise RuntimeError("MetricsExporter is already running") self._started = True - self._shutdown_event.clear() self._run_complete_event.clear() try: while not self._shutdown_event.is_set(): @@ -172,6 +169,12 @@ async def run(self) -> None: self._drain_once() finally: self._started = False + # Clear here (rather than at the top of this method) so a + # shutdown() requested before this run() got a chance to start + # is still observed on the loop's first check above -- clearing + # up front would silently erase that pending request and poll + # forever, since shutdown() only sets the event once. + self._shutdown_event.clear() self._run_complete_event.set() async def shutdown(self) -> None: diff --git a/tests/contrib/opentelemetry/test_metrics_exporter.py b/tests/contrib/opentelemetry/test_metrics_exporter.py index 1d045e185..5b066c78d 100644 --- a/tests/contrib/opentelemetry/test_metrics_exporter.py +++ b/tests/contrib/opentelemetry/test_metrics_exporter.py @@ -39,7 +39,7 @@ def _collect_metrics_by_name(reader: InMemoryMetricReader) -> dict[str, Any]: async def test_metrics_exporter_counter(): - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() counter = runtime.metric_meter.create_counter( "my-counter", "my-counter-desc", "my-counter-unit" @@ -83,7 +83,7 @@ async def test_metrics_exporter_counter(): async def test_metrics_exporter_gauge_reflects_latest_value(): - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() gauge = runtime.metric_meter.create_gauge_float("my-gauge") gauge.set(1.5) @@ -104,7 +104,7 @@ async def test_metrics_exporter_gauge_reflects_latest_value(): async def test_metrics_exporter_histogram(): - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() histogram = runtime.metric_meter.create_histogram_float("my-histogram") histogram.record(1.0) @@ -120,7 +120,7 @@ async def test_metrics_exporter_histogram(): async def test_metrics_exporter_attributes_passthrough(): - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() counter = runtime.metric_meter.create_counter("attrs-counter") counter.with_additional_attributes({"foo": "bar", "baz": 123}).add(1) @@ -148,7 +148,7 @@ async def test_metrics_exporter_buffer_not_attached_raises(): async def test_metrics_exporter_run_shutdown_lifecycle(): from datetime import timedelta - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() exporter._poll_interval = timedelta(milliseconds=10) counter = runtime.metric_meter.create_counter("lifecycle-counter") @@ -167,8 +167,26 @@ async def test_metrics_exporter_run_shutdown_lifecycle(): assert exporter._run_task.done() +async def test_metrics_exporter_shutdown_before_run_starts_does_not_hang(): + # __aenter__ only schedules run() via ensure_future; it doesn't run it. + # If the `async with` body never suspends, __aexit__ can call shutdown() + # (setting the shutdown event) before the run() task has executed at + # all. run() must still notice that pending shutdown on its first loop + # check instead of clearing the event and polling forever. + _, _, _, _, exporter = _make_runtime_and_exporter() + + async def enter_and_exit_immediately() -> None: + async with exporter: + pass + + await asyncio.wait_for(enter_and_exit_immediately(), timeout=5) + + assert exporter._run_task is not None + assert exporter._run_task.done() + + async def test_metrics_exporter_error_in_one_update_does_not_block_others(): - runtime, buffer, _, reader, exporter = _make_runtime_and_exporter() + runtime, _, _, reader, exporter = _make_runtime_and_exporter() good_counter = runtime.metric_meter.create_counter("good-counter") bad_counter = runtime.metric_meter.create_counter("bad-counter")