fix(bedrock): label converse metrics with their own model

packages/opentelemetry-instrumentation-bedrock/opentelemetry/instrumentation/bedrock/__init__.py

@@ -736,6 +736,13 @@ def read(self):
 @dont_throw
 def _handle_converse(span, kwargs, response, metric_params, event_logger):
     (provider, model_vendor, model) = _get_vendor_model(kwargs.get("modelId"))
+    # Usage recording reads vendor/model/is_stream off the shared `metric_params`, so they
+    # have to be set here the way the invoke_model handlers set them. Otherwise the
+    # converse metric points are labelled with "" on a fresh instrumentor, or with
+    # whatever model the last invoke_model call used.
+    metric_params.vendor = provider
+    metric_params.model = model
+    metric_params.is_stream = False
     guardrail_converse(span, response, provider, model, metric_params)
 
     set_converse_model_span_attributes(span, provider, model, kwargs)
@@ -754,6 +761,10 @@ def _handle_converse(span, kwargs, response, metric_params, event_logger):
 @dont_throw
 def _handle_converse_stream(span, kwargs, response, metric_params, event_logger):
     (provider, model_vendor, model) = _get_vendor_model(kwargs.get("modelId"))
+    # Keep the shared metric labels in sync, as in `_handle_converse`.
+    metric_params.vendor = provider
+    metric_params.model = model
+    metric_params.is_stream = True
 
     set_converse_model_span_attributes(span, provider, model, kwargs)
 
@@ -830,6 +841,10 @@ def _handle_async_converse_stream(span, kwargs, response, metric_params, event_l
     """Async variant of _handle_converse_stream — `_parse_event` is a coroutine
     in aiobotocore, so the wrapper must await it before inspecting the event."""
     (provider, model_vendor, model) = _get_vendor_model(kwargs.get("modelId"))
+    # Keep the shared metric labels in sync, as in `_handle_converse`.
+    metric_params.vendor = provider
+    metric_params.model = model
+    metric_params.is_stream = True
 
     set_converse_model_span_attributes(span, provider, model, kwargs)
 

packages/opentelemetry-instrumentation-bedrock/tests/metrics/test_converse_metric_labels.py

@@ -0,0 +1,116 @@
+from opentelemetry import trace
+from opentelemetry.instrumentation.bedrock import (
+    MetricParams,
+    _get_vendor_model,
+    _handle_converse,
+    _handle_converse_stream,
+)
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import InMemoryMetricReader
+from opentelemetry.semconv._incubating.attributes import (
+    gen_ai_attributes as GenAIAttributes,
+)
+from opentelemetry.semconv_ai import Meters
+
+MODEL_ID = "amazon.titan-text-express-v1"
+# The label the metric carries is whatever `_get_vendor_model` derives from the model id,
+# so derive the expectation the same way instead of duplicating the parsing rule here.
+_, _, EXPECTED_MODEL = _get_vendor_model(MODEL_ID)
+STALE_VENDOR = "anthropic"
+STALE_MODEL = "anthropic.claude-3-5-sonnet"
+
+CONVERSE_RESPONSE = {
+    "output": {"message": {"role": "assistant", "content": [{"text": "hi"}]}},
+    "stopReason": "end_turn",
+    "usage": {"inputTokens": 5, "outputTokens": 7, "totalTokens": 12},
+}
+
+CONVERSE_KWARGS = {
+    "modelId": MODEL_ID,
+    "messages": [{"role": "user", "content": [{"text": "hi"}]}],
+}
+
+
+def _metric_params():
+    """Real metric instruments backed by an in-memory reader."""
+    reader = InMemoryMetricReader()
+    provider = MeterProvider(metric_readers=[reader])
+    meter = provider.get_meter("test")
+
+    metric_params = MetricParams(
+        token_histogram=meter.create_histogram(Meters.LLM_TOKEN_USAGE),
+        choice_counter=meter.create_counter(Meters.LLM_GENERATION_CHOICES),
+        duration_histogram=meter.create_histogram(Meters.LLM_OPERATION_DURATION),
+        exception_counter=meter.create_counter("gen_ai.bedrock.completions.exceptions"),
+        guardrail_activation=meter.create_counter("guardrail.activation"),
+        guardrail_latency_histogram=meter.create_histogram("guardrail.latency"),
+        guardrail_coverage=meter.create_counter("guardrail.coverage"),
+        guardrail_sensitive_info=meter.create_counter("guardrail.sensitive_info"),
+        guardrail_topic=meter.create_counter("guardrail.topic"),
+        guardrail_content=meter.create_counter("guardrail.content"),
+        guardrail_words=meter.create_counter("guardrail.words"),
+        prompt_caching=meter.create_counter("prompt.caching"),
+    )
+
+    # What a previous invoke_model call would have left on the shared params.
+    metric_params.vendor = STALE_VENDOR
+    metric_params.model = STALE_MODEL
+    metric_params.is_stream = False
+
+    return metric_params, reader
+
+
+def _recorded_models(reader):
+    models = []
+    for resource_metrics in reader.get_metrics_data().resource_metrics:
+        for scope_metrics in resource_metrics.scope_metrics:
+            for metric in scope_metrics.metrics:
+                for data_point in metric.data.data_points:
+                    models.append(data_point.attributes.get(GenAIAttributes.GEN_AI_RESPONSE_MODEL))
+    return models
+
+
+def _span():
+    return trace.get_tracer(__name__).start_span("test")
+
+
+def test_converse_metrics_are_labelled_with_their_own_model():
+    """Converse metric points must name the converse model, not a previous call's.
+
+    `metric_params` is shared by every call on the instrumentor and the metric labels are
+    read from it, so it has to be updated per call. Without that, a fresh instrumentor
+    labelled points with "" and an invoke_model call left its model behind on subsequent
+    converse points.
+    """
+    metric_params, reader = _metric_params()
+
+    _handle_converse(_span(), CONVERSE_KWARGS, CONVERSE_RESPONSE, metric_params, None)
+
+    models = _recorded_models(reader)
+    assert models, "expected the converse call to record metric points"
+    assert set(models) == {EXPECTED_MODEL}
+    assert STALE_MODEL not in models
+
+
+def test_converse_stream_metrics_are_labelled_with_their_own_model():
+    metric_params, reader = _metric_params()
+
+    stream_response = {key: value for key, value in CONVERSE_RESPONSE.items() if key != "usage"}
+
+    def _parse_event(*args, **kwargs):
+        return {
+            "metadata": {
+                "usage": {"inputTokens": 5, "outputTokens": 7},
+            }
+        }
+
+    stream_response["stream"] = type("Stream", (), {"_parse_event": _parse_event})()
+
+    _handle_converse_stream(_span(), CONVERSE_KWARGS, stream_response, metric_params, None)
+    for _ in stream_response["stream"]._parse_event():
+        pass
+
+    models = _recorded_models(reader)
+    assert models, "expected the streamed converse call to record metric points"
+    assert set(models) == {EXPECTED_MODEL}
+    assert STALE_MODEL not in models