diff --git a/changelog/fixed/mislabeled-marmoset.md b/changelog/fixed/mislabeled-marmoset.md new file mode 100644 index 000000000..10a556b62 --- /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 32f17be9a..8a6e5bd69 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,8 @@ 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] + self._instruments: dict[ + tuple[str, temporalio.runtime.BufferedMetricKind], _Instrument ] = {} self._started = False self._shutdown_event = asyncio.Event() @@ -221,7 +220,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 +233,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 +243,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 +271,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 5b066c78d..0f228216a 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: 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}) + 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()