Skip to content

Commit f113a68

Browse files
Add opt-in OTel GenAI metrics to the observer
The OTel observer can now emit the metrics signal alongside its spans: two histogram instruments over provider calls, gated by a new enable_metrics flag (default off, independent of span emission). One records an LLM completion's input and output token counts; the other records the call duration, once per attempt under call-level retry and including a failed attempt (which carries error.type). Both draw from the per-attempt LlmRetryAttemptEvent, the LLM-span source, so metrics record even with spans disabled. The Meter comes from the configured MeterProvider (injectable; falls back to the OTel global no-op when none is set). Implements proposal 0067 (observability metrics), LLM path.
1 parent cd3a8ad commit f113a68

2 files changed

Lines changed: 315 additions & 0 deletions

File tree

src/openarmature/observability/otel/observer.py

Lines changed: 129 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,7 +81,9 @@
8181
from typing import Any, cast
8282

8383
from opentelemetry import context as otel_context
84+
from opentelemetry import metrics as otel_metrics
8485
from opentelemetry import trace as otel_trace
86+
from opentelemetry.metrics import Histogram, Meter, MeterProvider
8587
from opentelemetry.sdk.resources import Resource
8688
from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
8789
from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
@@ -142,6 +144,44 @@
142144
_PAYLOAD_DEFAULT_BYTES = 65536
143145

144146

147+
# §11.2 (proposal 0067) explicit bucket-boundary advisories for the two
148+
# GenAI metric histograms, mirroring the upstream gen_ai.client.* bucket
149+
# advisories so a future stable cutover is a mechanical prefix-strip.
150+
# Token usage in {token}; operation duration in seconds.
151+
_TOKEN_USAGE_BUCKETS: list[float] = [
152+
1,
153+
4,
154+
16,
155+
64,
156+
256,
157+
1024,
158+
4096,
159+
16384,
160+
65536,
161+
262144,
162+
1048576,
163+
4194304,
164+
16777216,
165+
67108864,
166+
]
167+
_DURATION_BUCKETS: list[float] = [
168+
0.01,
169+
0.02,
170+
0.04,
171+
0.08,
172+
0.16,
173+
0.32,
174+
0.64,
175+
1.28,
176+
2.56,
177+
5.12,
178+
10.24,
179+
20.48,
180+
40.96,
181+
81.92,
182+
]
183+
184+
145185
def _read_spec_version() -> str:
146186
"""Read the spec version pinned at package level. Lazy import
147187
avoids a circular at module-load time (the package's ``__init__``
@@ -469,6 +509,21 @@ class OTelObserver:
469509
# LLM-aware backends (Langfuse, Phoenix, Honeycomb's LLM lens)
470510
# render correctly out of the box, which keys off gen_ai.*.
471511
disable_genai_semconv: bool = False
512+
# Proposal 0067 (observability §11): opt-in GenAI metrics. Default
513+
# off; the flag name is normative. When True, __post_init__ creates
514+
# the two histogram instruments and provider-call events record token
515+
# + duration measurements. Independent of span emission (§11.1) — the
516+
# disable_llm_spans / disable_provider_payload / disable_genai_semconv
517+
# flags govern spans only.
518+
enable_metrics: bool = False
519+
# The MeterProvider the metric instruments are obtained from. None
520+
# (default) falls back to the configured OTel global MeterProvider,
521+
# which is the no-op meter when none is set up (§11.1: recording is a
522+
# silent no-op, never raises). Injectable so tests (and callers who
523+
# keep a private MeterProvider) can capture measurements without
524+
# mutating the OTel global. Unlike the span side's private
525+
# TracerProvider, metrics use the *configured* provider per §11.1.
526+
meter_provider: MeterProvider | None = None
472527
# Per-attribute byte cap for the §5.5.1 payload attributes. Default
473528
# 64 KiB; minimum 256 bytes (§5.5.5), validated in __post_init__.
474529
payload_max_bytes: int = _PAYLOAD_DEFAULT_BYTES
@@ -498,6 +553,11 @@ class OTelObserver:
498553
# Internal state, populated in __post_init__ and during invocation.
499554
_provider: TracerProvider = field(init=False, repr=False)
500555
_tracer: otel_trace.Tracer = field(init=False, repr=False)
556+
# §11 metric instruments — created in __post_init__ only when
557+
# enable_metrics is True; None otherwise (and recording is skipped).
558+
_meter: Meter | None = field(init=False, repr=False, default=None)
559+
_token_histogram: Histogram | None = field(init=False, repr=False, default=None)
560+
_duration_histogram: Histogram | None = field(init=False, repr=False, default=None)
501561
# Per-invocation_id span state — concurrent invocations on a
502562
# shared observer each get their own ``_InvState`` so internal
503563
# maps never collide.
@@ -539,6 +599,26 @@ def __post_init__(self) -> None:
539599
for proc in processors:
540600
self._provider.add_span_processor(proc)
541601
self._tracer = self._provider.get_tracer("openarmature")
602+
# §11.1 metrics: opt-in. When enabled, obtain a Meter from the
603+
# configured MeterProvider (injected, else the OTel global, which
604+
# is the no-op meter when none is set up) and create the two
605+
# histograms with the §11.2 explicit bucket advisories. When
606+
# disabled, no instrument is created and nothing records.
607+
if self.enable_metrics:
608+
meter_provider = (
609+
self.meter_provider if self.meter_provider is not None else otel_metrics.get_meter_provider()
610+
)
611+
self._meter = meter_provider.get_meter("openarmature")
612+
self._token_histogram = self._meter.create_histogram(
613+
"openarmature.gen_ai.client.token.usage",
614+
unit="{token}",
615+
explicit_bucket_boundaries_advisory=_TOKEN_USAGE_BUCKETS,
616+
)
617+
self._duration_histogram = self._meter.create_histogram(
618+
"openarmature.gen_ai.client.operation.duration",
619+
unit="s",
620+
explicit_bucket_boundaries_advisory=_DURATION_BUCKETS,
621+
)
542622

543623
# ------------------------------------------------------------------
544624
# Enricher invocation (friction-roundup #7p2)
@@ -612,6 +692,10 @@ async def __call__(
612692
# LlmRetryAttemptEvent — one span per attempt under call-level
613693
# retry (attempt_index 0..N-1), one for a no-retry call.
614694
if isinstance(event, LlmRetryAttemptEvent):
695+
# §11 metrics record per attempt, independent of span emission
696+
# (§11.1) — so this runs regardless of disable_llm_spans.
697+
if self.enable_metrics:
698+
self._record_llm_metrics(event)
615699
if not self.disable_llm_spans:
616700
self._handle_typed_llm_retry_attempt(event)
617701
return
@@ -1214,6 +1298,51 @@ def _emit_checkpoint_save_span(self, event: NodeEvent) -> None:
12141298
# active_prompt_group snapshots taken at dispatch time — NOT the
12151299
# ContextVar. The dispatch worker's task-local Context doesn't see
12161300
# node-body ContextVar writes.
1301+
def _record_llm_metrics(self, event: LlmRetryAttemptEvent) -> None:
1302+
"""Record the §11 GenAI metric observations for one LLM-call
1303+
attempt.
1304+
1305+
Duration is recorded for every attempt (including a failed one,
1306+
carrying ``error.type``); token usage only when the attempt
1307+
returned a usage record (a failed attempt has none). Sourced from
1308+
the per-attempt ``LlmRetryAttemptEvent`` — one duration sample per
1309+
attempt under call-level retry, matching the per-attempt span
1310+
model. The attempt index is deliberately not a dimension (§11.2
1311+
cardinality).
1312+
"""
1313+
if self._duration_histogram is None or self._token_histogram is None:
1314+
return
1315+
# §11.3 dimensions shared by both instruments. operation is "chat"
1316+
# for an LLM completion; gen_ai.request.model / gen_ai.system are
1317+
# the recognized-core de-facto-standard keys used directly.
1318+
base_dims: dict[str, str] = {
1319+
"openarmature.gen_ai.operation": "chat",
1320+
"gen_ai.request.model": event.model,
1321+
"gen_ai.system": event.provider,
1322+
}
1323+
# Duration (seconds): every attempt. error.type carries the §7
1324+
# category on a failed attempt; absent on success.
1325+
if event.latency_ms is not None:
1326+
duration_dims = dict(base_dims)
1327+
if event.error_category is not None:
1328+
duration_dims["error.type"] = event.error_category
1329+
self._duration_histogram.record(event.latency_ms / 1000.0, duration_dims)
1330+
# Token usage: only when a usage record is present (a failed
1331+
# attempt has none). Two observations for an LLM completion —
1332+
# input + output — each only when its count was reported.
1333+
usage = event.usage
1334+
if usage is not None:
1335+
if usage.prompt_tokens is not None:
1336+
self._token_histogram.record(
1337+
usage.prompt_tokens,
1338+
{**base_dims, "openarmature.gen_ai.token.type": "input"},
1339+
)
1340+
if usage.completion_tokens is not None:
1341+
self._token_histogram.record(
1342+
usage.completion_tokens,
1343+
{**base_dims, "openarmature.gen_ai.token.type": "output"},
1344+
)
1345+
12171346
def _handle_typed_llm_retry_attempt(self, event: LlmRetryAttemptEvent) -> None:
12181347
"""Open + close one ``openarmature.llm.complete`` span from a
12191348
per-attempt LlmRetryAttemptEvent.

tests/unit/test_observability_otel.py

Lines changed: 186 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -865,6 +865,192 @@ async def test_llm_span_output_tool_calls_payload_gating() -> None:
865865
]
866866

867867

868+
# ---------------------------------------------------------------------------
869+
# Proposal 0067 — GenAI metrics (observability §11)
870+
# ---------------------------------------------------------------------------
871+
872+
873+
def _collect_metric_points(reader: Any) -> list[tuple[str, float, int, dict[str, Any]]]:
874+
"""Flatten an InMemoryMetricReader's collected data into
875+
``(instrument_name, point_sum, point_count, point_attributes)``
876+
tuples. Histogram observations with identical attribute sets
877+
aggregate into one data point (sum + count), so per-attempt tests
878+
assert on ``count``, not point cardinality."""
879+
data = reader.get_metrics_data()
880+
points: list[tuple[str, float, int, dict[str, Any]]] = []
881+
if data is None:
882+
return points
883+
for resource_metric in data.resource_metrics:
884+
for scope_metric in resource_metric.scope_metrics:
885+
for metric in scope_metric.metrics:
886+
for pt in metric.data.data_points:
887+
points.append((metric.name, pt.sum, pt.count, dict(pt.attributes)))
888+
return points
889+
890+
891+
async def _drive_metrics_events(
892+
events: list[Any],
893+
*,
894+
enable_metrics: bool = True,
895+
disable_llm_spans: bool = False,
896+
) -> tuple[list[tuple[str, float, int, dict[str, Any]]], list[Any]]:
897+
"""Feed LlmRetryAttemptEvents through an OTelObserver wired to a
898+
private MeterProvider + InMemoryMetricReader; return the captured
899+
``(metric_points, llm_complete_spans)``."""
900+
from opentelemetry.sdk.metrics import MeterProvider as SdkMeterProvider
901+
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
902+
903+
from openarmature.observability.correlation import (
904+
_reset_invocation_id,
905+
_set_invocation_id,
906+
)
907+
908+
reader = InMemoryMetricReader()
909+
meter_provider = SdkMeterProvider(metric_readers=[reader])
910+
exporter = InMemorySpanExporter()
911+
observer = OTelObserver(
912+
span_processor=SimpleSpanProcessor(exporter),
913+
enable_metrics=enable_metrics,
914+
disable_llm_spans=disable_llm_spans,
915+
meter_provider=meter_provider,
916+
)
917+
token = _set_invocation_id("inv-metrics")
918+
try:
919+
for event in events:
920+
await observer(event)
921+
finally:
922+
_reset_invocation_id(token)
923+
observer.shutdown()
924+
llm_spans = [s for s in exporter.get_finished_spans() if s.name == "openarmature.llm.complete"]
925+
return _collect_metric_points(reader), llm_spans
926+
927+
928+
async def test_metrics_records_token_and_duration() -> None:
929+
# Proposal 0067 §11 (mirrors fixture 088): a successful LLM attempt
930+
# with usage {input 5, output 1} records two token.usage observations
931+
# (input + output) and one duration observation, with the §11.3
932+
# dimensions. Duration value is not asserted (§11.4).
933+
from openarmature.llm.response import Usage
934+
from tests._helpers.typed_event import make_retry_attempt_event
935+
936+
event = make_retry_attempt_event(
937+
model="test-model",
938+
provider="openai",
939+
latency_ms=12.0,
940+
usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
941+
)
942+
points, _ = await _drive_metrics_events([event])
943+
token_points = [p for p in points if p[0] == "openarmature.gen_ai.client.token.usage"]
944+
duration_points = [p for p in points if p[0] == "openarmature.gen_ai.client.operation.duration"]
945+
by_type = {p[3]["openarmature.gen_ai.token.type"]: p for p in token_points}
946+
assert by_type["input"][1] == 5
947+
assert by_type["output"][1] == 1
948+
for ttype in ("input", "output"):
949+
dims = by_type[ttype][3]
950+
assert dims["openarmature.gen_ai.operation"] == "chat"
951+
assert dims["gen_ai.request.model"] == "test-model"
952+
assert dims["gen_ai.system"] == "openai"
953+
assert len(duration_points) == 1
954+
ddims = duration_points[0][3]
955+
assert ddims["openarmature.gen_ai.operation"] == "chat"
956+
assert ddims["gen_ai.request.model"] == "test-model"
957+
assert ddims["gen_ai.system"] == "openai"
958+
assert "error.type" not in ddims
959+
960+
961+
async def test_metrics_records_duration_with_error_type_on_failure() -> None:
962+
# Proposal 0067 §11.2 / §11.3 (mirrors fixture 090): a failed attempt
963+
# records a duration observation carrying error.type, and NO
964+
# token.usage observation (a failed attempt returned no usage).
965+
from tests._helpers.typed_event import make_retry_attempt_event
966+
967+
event = make_retry_attempt_event(
968+
model="test-model",
969+
provider="openai",
970+
latency_ms=8.0,
971+
finish_reason=None,
972+
usage=None,
973+
error_category="provider_unavailable",
974+
error_type="ProviderUnavailable",
975+
error_message="down",
976+
)
977+
points, _ = await _drive_metrics_events([event])
978+
token_points = [p for p in points if p[0] == "openarmature.gen_ai.client.token.usage"]
979+
duration_points = [p for p in points if p[0] == "openarmature.gen_ai.client.operation.duration"]
980+
assert token_points == []
981+
assert len(duration_points) == 1
982+
assert duration_points[0][3]["error.type"] == "provider_unavailable"
983+
assert duration_points[0][3]["openarmature.gen_ai.operation"] == "chat"
984+
985+
986+
async def test_metrics_disabled_records_nothing() -> None:
987+
# Proposal 0067 §11.1 (mirrors fixture 091): enable_metrics off (the
988+
# default) creates no instrument and records nothing.
989+
from openarmature.llm.response import Usage
990+
from tests._helpers.typed_event import make_retry_attempt_event
991+
992+
event = make_retry_attempt_event(usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6))
993+
points, _ = await _drive_metrics_events([event], enable_metrics=False)
994+
assert points == []
995+
996+
997+
async def test_metrics_independent_of_disable_llm_spans() -> None:
998+
# Proposal 0067 §11.1: metrics record even with spans disabled — the
999+
# disable_llm_spans flag governs span emission only.
1000+
from openarmature.llm.response import Usage
1001+
from tests._helpers.typed_event import make_retry_attempt_event
1002+
1003+
event = make_retry_attempt_event(usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6))
1004+
points, llm_spans = await _drive_metrics_events([event], disable_llm_spans=True)
1005+
assert llm_spans == []
1006+
assert any(p[0] == "openarmature.gen_ai.client.operation.duration" for p in points)
1007+
assert any(p[0] == "openarmature.gen_ai.client.token.usage" for p in points)
1008+
1009+
1010+
async def test_metrics_record_once_per_attempt_under_retry() -> None:
1011+
# Proposal 0067 §11.2 "Call-level retry": the duration histogram
1012+
# records once per attempt (failed attempts carry error.type), and
1013+
# token.usage only for an attempt that returned usage. Two failed
1014+
# attempts + one success -> 3 duration observations (2 with
1015+
# error.type), 2 token.usage observations (the success's input +
1016+
# output). Observations with identical dimensions aggregate into one
1017+
# data point, so this asserts on histogram counts.
1018+
from openarmature.llm.response import Usage
1019+
from tests._helpers.typed_event import make_retry_attempt_event
1020+
1021+
failed = [
1022+
make_retry_attempt_event(
1023+
llm_attempt_index=i,
1024+
latency_ms=5.0,
1025+
finish_reason=None,
1026+
usage=None,
1027+
error_category="provider_unavailable",
1028+
error_type="ProviderUnavailable",
1029+
error_message="down",
1030+
)
1031+
for i in range(2)
1032+
]
1033+
success = make_retry_attempt_event(
1034+
llm_attempt_index=2,
1035+
latency_ms=7.0,
1036+
usage=Usage(prompt_tokens=5, completion_tokens=1, total_tokens=6),
1037+
)
1038+
points, _ = await _drive_metrics_events([*failed, success])
1039+
duration_points = [p for p in points if p[0] == "openarmature.gen_ai.client.operation.duration"]
1040+
token_points = [p for p in points if p[0] == "openarmature.gen_ai.client.token.usage"]
1041+
# 3 duration observations total: 2 share the error dims (one
1042+
# aggregated point, count 2), the success is a separate point.
1043+
assert sum(p[2] for p in duration_points) == 3
1044+
error_duration = [p for p in duration_points if p[3].get("error.type") == "provider_unavailable"]
1045+
assert sum(p[2] for p in error_duration) == 2
1046+
success_duration = [p for p in duration_points if "error.type" not in p[3]]
1047+
assert sum(p[2] for p in success_duration) == 1
1048+
# token.usage only from the success attempt: one input, one output.
1049+
by_type = {p[3]["openarmature.gen_ai.token.type"]: p for p in token_points}
1050+
assert by_type["input"][1] == 5 and by_type["input"][2] == 1
1051+
assert by_type["output"][1] == 1 and by_type["output"][2] == 1
1052+
1053+
8681054
async def test_llm_span_zero_duration_when_latency_missing() -> None:
8691055
# When the typed event omits latency_ms (None), the handler falls
8701056
# back to a zero-duration span at end_time rather than guessing

0 commit comments

Comments
 (0)