From 8e4dd6a87c273bbffdad10323f26a31dded050a3 Mon Sep 17 00:00:00 2001 From: Sohom Chakraborty <16609933+sohom-cs@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:59:08 -0400 Subject: [PATCH 1/3] Fix MetricsExporter mixing up instruments and labels when object ids are reused --- changelog/fixed/mislabeled-marmoset.md | 1 + .../opentelemetry/_metrics_exporter.py | 40 ++++++++----------- .../opentelemetry/test_metrics_exporter.py | 28 +++++++++++++ 3 files changed, 45 insertions(+), 24 deletions(-) create mode 100644 changelog/fixed/mislabeled-marmoset.md diff --git a/changelog/fixed/mislabeled-marmoset.md b/changelog/fixed/mislabeled-marmoset.md new file mode 100644 index 0000000000..10a556b62d --- /dev/null +++ b/changelog/fixed/mislabeled-marmoset.md @@ -0,0 +1 @@ +Fix `contrib.opentelemetry` `MetricsExporter` recording a metric under another metric's instrument or another attribute set's labels when instruments or attributes are created per call. It cached them by `id()`, which Python reuses once an object is freed. diff --git a/temporalio/contrib/opentelemetry/_metrics_exporter.py b/temporalio/contrib/opentelemetry/_metrics_exporter.py index 32f17be9a3..93c5b1c53e 100644 --- a/temporalio/contrib/opentelemetry/_metrics_exporter.py +++ b/temporalio/contrib/opentelemetry/_metrics_exporter.py @@ -11,7 +11,6 @@ import opentelemetry.metrics import opentelemetry.util.types -import temporalio.common import temporalio.runtime logger = logging.getLogger(__name__) @@ -20,11 +19,12 @@ @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. + # Keyed by attribute values 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] + frozenset[tuple[str, opentelemetry.util.types.AttributeValue]], + tuple[Mapping[str, opentelemetry.util.types.AttributeValue], int | float], ] = field(default_factory=dict) @@ -136,9 +136,10 @@ def __init__( ).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] + # Not keyed by id(): the buffer frees a metric or attribute set once + # Core drops it, and a later one can then get the same id(). + self._instruments: dict[ + tuple[str, temporalio.runtime.BufferedMetricKind], _Instrument ] = {} self._started = False self._shutdown_event = asyncio.Event() @@ -221,7 +222,9 @@ def _handle_error(self, err: Exception) -> None: 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) + attributes: dict[str, opentelemetry.util.types.AttributeValue] = dict( + update.attributes + ) if instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_COUNTER: assert instrument.counter is not None @@ -232,7 +235,7 @@ def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None 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)] = ( + instrument.gauge_state.last_values[frozenset(attributes.items())] = ( attributes, update.value, ) @@ -242,7 +245,8 @@ def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None def _get_or_create_instrument( self, metric: temporalio.runtime.BufferedMetric ) -> _Instrument: - instrument = self._instruments.get(id(metric)) + key = (metric.name, metric.kind) + instrument = self._instruments.get(key) if instrument is not None: return instrument @@ -269,17 +273,5 @@ def _get_or_create_instrument( else: raise ValueError(f"Unrecognized buffered metric kind: {metric.kind}") - self._instruments[id(metric)] = instrument + self._instruments[key] = 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 index 5b066c78db..b4784e48da 100644 --- a/tests/contrib/opentelemetry/test_metrics_exporter.py +++ b/tests/contrib/opentelemetry/test_metrics_exporter.py @@ -132,6 +132,34 @@ async def test_metrics_exporter_attributes_passthrough(): assert dict(points[0].attributes) == {"foo": "bar", "baz": 123} +async def test_metrics_exporter_instruments_created_per_call(): + runtime, _, _, reader, exporter = _make_runtime_and_exporter() + + # Like an activity that creates its instruments on each call. Once drained, + # a call's metric and attribute objects are freed, so the next call's + # objects can reuse their ids. + calls = [("model-a", 100, 900), ("model-b", 7, 40), ("model-c", 55, 300)] + for model, tokens, latency_ms in calls: + meter = runtime.metric_meter.with_additional_attributes({"model": model}) + meter.create_counter("tokens").add(tokens) + meter.create_histogram("latency").record(latency_ms) + meter.create_gauge("last-tokens").set(tokens) + exporter._drain_once() + + metrics = _collect_metrics_by_name(reader) + tokens_by_model = {model: tokens for model, tokens, _ in calls} + assert { + p.attributes["model"]: p.value for p in metrics["tokens"].data.data_points + } == tokens_by_model + assert { + p.attributes["model"]: p.value for p in metrics["last-tokens"].data.data_points + } == tokens_by_model + assert { + p.attributes["model"]: (p.count, p.sum) + for p in metrics["latency"].data.data_points + } == {model: (1, latency_ms) for model, _, latency_ms in calls} + + async def test_metrics_exporter_buffer_not_attached_raises(): unattached_buffer = MetricBuffer(10_000) reader = InMemoryMetricReader() From a468e66291814e73c315c86c41c393e81e76d0be Mon Sep 17 00:00:00 2001 From: Sohom Chakraborty <16609933+sohom-cs@users.noreply.github.com> Date: Thu, 29 Oct 2026 20:46:58 -0400 Subject: [PATCH 2/3] Cache by id() again, holding each cached object Each entry keeps its object alive, so the id can't be reused while the entry is cached, and both caches are bounded: a full cache starts over. An attribute entry also keeps the gauge key once a gauge update has built it. Instruments are still shared by (name, kind) behind the metric cache. --- .../opentelemetry/_metrics_exporter.py | 75 ++++++++++++++++--- .../opentelemetry/test_metrics_exporter.py | 36 ++++++++- 2 files changed, 98 insertions(+), 13 deletions(-) diff --git a/temporalio/contrib/opentelemetry/_metrics_exporter.py b/temporalio/contrib/opentelemetry/_metrics_exporter.py index 93c5b1c53e..e20eaef99a 100644 --- a/temporalio/contrib/opentelemetry/_metrics_exporter.py +++ b/temporalio/contrib/opentelemetry/_metrics_exporter.py @@ -7,14 +7,43 @@ from collections.abc import Callable, Iterable, Mapping from dataclasses import dataclass, field from datetime import timedelta +from typing import Generic, TypeVar import opentelemetry.metrics import opentelemetry.util.types +import temporalio.common import temporalio.runtime logger = logging.getLogger(__name__) +_K = TypeVar("_K") +_V = TypeVar("_V") + + +class _IdCache(Generic[_K, _V]): + # Keyed by id() for speed. Each entry holds its object, so the id can't be + # reused by another object while the entry is cached. A full cache starts + # over, which bounds what it keeps alive. + def __init__(self, max_size: int) -> None: + self._max_size = max_size + self._entries: dict[int, tuple[_K, _V]] = {} + + def get(self, obj: _K) -> _V | None: + entry = self._entries.get(id(obj)) + return None if entry is None else entry[1] + + def put(self, obj: _K, value: _V) -> None: + if len(self._entries) >= self._max_size: + self._entries.clear() + self._entries[id(obj)] = (obj, value) + + def __len__(self) -> int: + return len(self._entries) + + +_ID_CACHE_SIZE = 1000 + @dataclass class _GaugeState: @@ -28,6 +57,15 @@ class _GaugeState: ] = field(default_factory=dict) +@dataclass(slots=True) +class _ConvertedAttributes: + attributes: Mapping[str, opentelemetry.util.types.AttributeValue] + # Only gauges key on it, so it is built on the first gauge update. + gauge_key: frozenset[tuple[str, opentelemetry.util.types.AttributeValue]] | None = ( + None + ) + + @dataclass class _Instrument: kind: temporalio.runtime.BufferedMetricKind @@ -136,11 +174,15 @@ def __init__( ).get_meter(meter_name, meter_version) self._poll_interval = poll_interval self._on_error = on_error - # Not keyed by id(): the buffer frees a metric or attribute set once - # Core drops it, and a later one can then get the same id(). self._instruments: dict[ tuple[str, temporalio.runtime.BufferedMetricKind], _Instrument ] = {} + self._instruments_by_metric: _IdCache[ + temporalio.runtime.BufferedMetric, _Instrument + ] = _IdCache(_ID_CACHE_SIZE) + self._attributes_cache: _IdCache[ + temporalio.common.MetricAttributes, _ConvertedAttributes + ] = _IdCache(_ID_CACHE_SIZE) self._started = False self._shutdown_event = asyncio.Event() self._run_complete_event = asyncio.Event() @@ -222,21 +264,21 @@ def _handle_error(self, err: Exception) -> None: def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None: instrument = self._get_or_create_instrument(update.metric) - attributes: dict[str, opentelemetry.util.types.AttributeValue] = dict( - update.attributes - ) + converted = 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) + instrument.counter.add(update.value, attributes=converted.attributes) elif instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_HISTOGRAM: assert instrument.histogram is not None - instrument.histogram.record(update.value, attributes=attributes) + instrument.histogram.record(update.value, attributes=converted.attributes) elif instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_GAUGE: assert instrument.gauge_state is not None + if converted.gauge_key is None: + converted.gauge_key = frozenset(converted.attributes.items()) with instrument.gauge_state.lock: - instrument.gauge_state.last_values[frozenset(attributes.items())] = ( - attributes, + instrument.gauge_state.last_values[converted.gauge_key] = ( + converted.attributes, update.value, ) else: @@ -245,9 +287,13 @@ def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None def _get_or_create_instrument( self, metric: temporalio.runtime.BufferedMetric ) -> _Instrument: + instrument = self._instruments_by_metric.get(metric) + if instrument is not None: + return instrument key = (metric.name, metric.kind) instrument = self._instruments.get(key) if instrument is not None: + self._instruments_by_metric.put(metric, instrument) return instrument unit = metric.unit or "" @@ -274,4 +320,15 @@ def _get_or_create_instrument( raise ValueError(f"Unrecognized buffered metric kind: {metric.kind}") self._instruments[key] = instrument + self._instruments_by_metric.put(metric, instrument) return instrument + + def _get_or_create_attributes( + self, attributes: temporalio.common.MetricAttributes + ) -> _ConvertedAttributes: + converted = self._attributes_cache.get(attributes) + if converted is not None: + return converted + converted = _ConvertedAttributes(dict(attributes)) + self._attributes_cache.put(attributes, converted) + return converted diff --git a/tests/contrib/opentelemetry/test_metrics_exporter.py b/tests/contrib/opentelemetry/test_metrics_exporter.py index b4784e48da..59e6ebf7e1 100644 --- a/tests/contrib/opentelemetry/test_metrics_exporter.py +++ b/tests/contrib/opentelemetry/test_metrics_exporter.py @@ -7,7 +7,7 @@ from opentelemetry.sdk.metrics import MeterProvider from opentelemetry.sdk.metrics.export import InMemoryMetricReader -from temporalio.contrib.opentelemetry import MetricsExporter +from temporalio.contrib.opentelemetry import MetricsExporter, _metrics_exporter from temporalio.runtime import MetricBuffer, Runtime, TelemetryConfig @@ -135,9 +135,9 @@ async def test_metrics_exporter_attributes_passthrough(): async def test_metrics_exporter_instruments_created_per_call(): runtime, _, _, reader, exporter = _make_runtime_and_exporter() - # Like an activity that creates its instruments on each call. Once drained, - # a call's metric and attribute objects are freed, so the next call's - # objects can reuse their ids. + # Like an activity that creates its instruments on each call: each call + # brings new metric and attribute objects, and a cache keyed by id() that + # doesn't hold them would see their ids reused. calls = [("model-a", 100, 900), ("model-b", 7, 40), ("model-c", 55, 300)] for model, tokens, latency_ms in calls: meter = runtime.metric_meter.with_additional_attributes({"model": model}) @@ -160,6 +160,34 @@ async def test_metrics_exporter_instruments_created_per_call(): } == {model: (1, latency_ms) for model, _, latency_ms in calls} +async def test_metrics_exporter_caches_are_bounded(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(_metrics_exporter, "_ID_CACHE_SIZE", 2) + runtime, _, _, reader, exporter = _make_runtime_and_exporter() + + meters = [ + runtime.metric_meter.with_additional_attributes({"n": n}) for n in range(5) + ] + counters = [meter.create_counter("bounded-counter") for meter in meters] + gauges = [meter.create_gauge("bounded-gauge") for meter in meters] + for value in (1, 2): + for counter, gauge in zip(counters, gauges): + counter.add(value) + gauge.set(value) + exporter._drain_once() + assert len(exporter._attributes_cache) <= 2 + assert len(exporter._instruments_by_metric) <= 2 + + # Every set was evicted before its second update, so this also checks + # that an evicted object is looked up again correctly. + metrics = _collect_metrics_by_name(reader) + assert { + p.attributes["n"]: p.value for p in metrics["bounded-counter"].data.data_points + } == {n: 3 for n in range(5)} + assert { + p.attributes["n"]: p.value for p in metrics["bounded-gauge"].data.data_points + } == {n: 2 for n in range(5)} + + async def test_metrics_exporter_buffer_not_attached_raises(): unattached_buffer = MetricBuffer(10_000) reader = InMemoryMetricReader() From fa6cb0143f39e86fec71743ae5b8bcf60bba616d Mon Sep 17 00:00:00 2001 From: Sohom Chakraborty <16609933+sohom-cs@users.noreply.github.com> Date: Fri, 30 Oct 2026 17:46:46 -0400 Subject: [PATCH 3/3] Drop the attribute cache Attributes are converted on each update again, and a gauge's last values are keyed by frozenset(attributes.items()). The id() cache stays only in front of the (name, kind) instrument map, and the bounded test now checks only that cache. --- .../opentelemetry/_metrics_exporter.py | 37 ++++--------------- .../opentelemetry/test_metrics_exporter.py | 9 +++-- 2 files changed, 12 insertions(+), 34 deletions(-) diff --git a/temporalio/contrib/opentelemetry/_metrics_exporter.py b/temporalio/contrib/opentelemetry/_metrics_exporter.py index e20eaef99a..ad52a331a6 100644 --- a/temporalio/contrib/opentelemetry/_metrics_exporter.py +++ b/temporalio/contrib/opentelemetry/_metrics_exporter.py @@ -12,7 +12,6 @@ import opentelemetry.metrics import opentelemetry.util.types -import temporalio.common import temporalio.runtime logger = logging.getLogger(__name__) @@ -57,15 +56,6 @@ class _GaugeState: ] = field(default_factory=dict) -@dataclass(slots=True) -class _ConvertedAttributes: - attributes: Mapping[str, opentelemetry.util.types.AttributeValue] - # Only gauges key on it, so it is built on the first gauge update. - gauge_key: frozenset[tuple[str, opentelemetry.util.types.AttributeValue]] | None = ( - None - ) - - @dataclass class _Instrument: kind: temporalio.runtime.BufferedMetricKind @@ -180,9 +170,6 @@ def __init__( self._instruments_by_metric: _IdCache[ temporalio.runtime.BufferedMetric, _Instrument ] = _IdCache(_ID_CACHE_SIZE) - self._attributes_cache: _IdCache[ - temporalio.common.MetricAttributes, _ConvertedAttributes - ] = _IdCache(_ID_CACHE_SIZE) self._started = False self._shutdown_event = asyncio.Event() self._run_complete_event = asyncio.Event() @@ -264,21 +251,21 @@ def _handle_error(self, err: Exception) -> None: def _apply_update(self, update: temporalio.runtime.BufferedMetricUpdate) -> None: instrument = self._get_or_create_instrument(update.metric) - converted = self._get_or_create_attributes(update.attributes) + attributes: dict[str, opentelemetry.util.types.AttributeValue] = dict( + update.attributes + ) if instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_COUNTER: assert instrument.counter is not None - instrument.counter.add(update.value, attributes=converted.attributes) + 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=converted.attributes) + instrument.histogram.record(update.value, attributes=attributes) elif instrument.kind == temporalio.runtime.BUFFERED_METRIC_KIND_GAUGE: assert instrument.gauge_state is not None - if converted.gauge_key is None: - converted.gauge_key = frozenset(converted.attributes.items()) with instrument.gauge_state.lock: - instrument.gauge_state.last_values[converted.gauge_key] = ( - converted.attributes, + instrument.gauge_state.last_values[frozenset(attributes.items())] = ( + attributes, update.value, ) else: @@ -322,13 +309,3 @@ def _get_or_create_instrument( self._instruments[key] = instrument self._instruments_by_metric.put(metric, instrument) return instrument - - def _get_or_create_attributes( - self, attributes: temporalio.common.MetricAttributes - ) -> _ConvertedAttributes: - converted = self._attributes_cache.get(attributes) - if converted is not None: - return converted - converted = _ConvertedAttributes(dict(attributes)) - self._attributes_cache.put(attributes, converted) - return converted diff --git a/tests/contrib/opentelemetry/test_metrics_exporter.py b/tests/contrib/opentelemetry/test_metrics_exporter.py index 59e6ebf7e1..7f1708e215 100644 --- a/tests/contrib/opentelemetry/test_metrics_exporter.py +++ b/tests/contrib/opentelemetry/test_metrics_exporter.py @@ -160,7 +160,9 @@ async def test_metrics_exporter_instruments_created_per_call(): } == {model: (1, latency_ms) for model, _, latency_ms in calls} -async def test_metrics_exporter_caches_are_bounded(monkeypatch: pytest.MonkeyPatch): +async def test_metrics_exporter_instrument_cache_is_bounded( + monkeypatch: pytest.MonkeyPatch, +): monkeypatch.setattr(_metrics_exporter, "_ID_CACHE_SIZE", 2) runtime, _, _, reader, exporter = _make_runtime_and_exporter() @@ -174,11 +176,10 @@ async def test_metrics_exporter_caches_are_bounded(monkeypatch: pytest.MonkeyPat counter.add(value) gauge.set(value) exporter._drain_once() - assert len(exporter._attributes_cache) <= 2 assert len(exporter._instruments_by_metric) <= 2 - # Every set was evicted before its second update, so this also checks - # that an evicted object is looked up again correctly. + # Every instrument was evicted before its second update, so this also + # checks that an evicted metric is looked up again correctly. metrics = _collect_metrics_by_name(reader) assert { p.attributes["n"]: p.value for p in metrics["bounded-counter"].data.data_points