GenAI Utils | Adding Embedding metrics

util/opentelemetry-util-genai/CHANGELOG.md

@@ -7,6 +7,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add metrics support for EmbeddingInvocation
+  ([#4377](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4377))
 - Add support for workflow in genAI utils handler.
   ([https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4366](#4366))
 - Enrich ToolCall type, breaking change: usage of ToolCall class renamed to ToolCallRequest

util/opentelemetry-util-genai/src/opentelemetry/util/genai/_embedding_invocation.py

@@ -120,5 +120,4 @@ def _apply_finish(self, error: Error | None = None) -> None:
             self._apply_error_attributes(error)
         attributes.update(self.attributes)
         self.span.set_attributes(attributes)
-        # Metrics recorder currently supports InferenceInvocation fields only.
-        # No-op until dedicated embedding metric support is added.
+        self._metrics_recorder.record(self)

util/opentelemetry-util-genai/tests/test_handler_metrics.py

@@ -181,3 +181,145 @@ def _assert_metric_scope_schema_urls(
                 self.assertEqual(
                     scope_metric.scope.schema_url, expected_schema_url
                 )
+
+    def test_stop_embedding_records_duration_and_tokens(self) -> None:
+        """Verify embedding invocations record duration and input token metrics."""
+        handler = TelemetryHandler(
+            tracer_provider=self.tracer_provider,
+            meter_provider=self.meter_provider,
+        )
+        # Patch default_timer during start to ensure monotonic_start_s
+        with patch("timeit.default_timer", return_value=1000.0):
+            invocation = handler.start_embedding(
+                "embed-prov", request_model="embed-model"
+            )
+        invocation.input_tokens = 100
+
+        # Simulate 1.5 seconds of elapsed monotonic time
+        with patch("timeit.default_timer", return_value=1001.5):
+            invocation.stop()
+
+        self._assert_metric_scope_schema_urls(_DEFAULT_SCHEMA_URL)
+        metrics = self._harvest_metrics()
+
+        # Duration should be recorded
+        self.assertIn("gen_ai.client.operation.duration", metrics)
+        duration_points = metrics["gen_ai.client.operation.duration"]
+        self.assertEqual(len(duration_points), 1)
+        duration_point = duration_points[0]
+        self.assertEqual(
+            duration_point.attributes[GenAI.GEN_AI_OPERATION_NAME],
+            GenAI.GenAiOperationNameValues.EMBEDDINGS.value,
+        )
+        self.assertEqual(
+            duration_point.attributes[GenAI.GEN_AI_REQUEST_MODEL],
+            "embed-model",
+        )
+        self.assertEqual(
+            duration_point.attributes[GenAI.GEN_AI_PROVIDER_NAME], "embed-prov"
+        )
+        self.assertAlmostEqual(duration_point.sum, 1.5, places=3)
+
+        # Token metrics should be recorded for embedding (input only)
+        self.assertIn("gen_ai.client.token.usage", metrics)
+        token_points = metrics["gen_ai.client.token.usage"]
+        self.assertEqual(len(token_points), 1)  # Only input tokens
+        token_point = token_points[0]
+        self.assertEqual(
+            token_point.attributes[GenAI.GEN_AI_TOKEN_TYPE],
+            GenAI.GenAiTokenTypeValues.INPUT.value,
+        )
+        self.assertAlmostEqual(token_point.sum, 100.0, places=3)
+
+    def test_stop_embedding_records_duration_with_additional_attributes(
+        self,
+    ) -> None:
+        """Verify embedding metrics include server and custom attributes."""
+        handler = TelemetryHandler(
+            tracer_provider=self.tracer_provider,
+            meter_provider=self.meter_provider,
+        )
+        invocation = handler.start_embedding(
+            "embed-prov",
+            request_model="embed-model",
+            server_address="embed.server.com",
+            server_port=8080,
+        )
+        invocation.metric_attributes = {"custom.embed.attr": "embed_value"}
+        invocation.response_model_name = "embed-response-model"
+        invocation.stop()
+
+        self._assert_metric_scope_schema_urls(_DEFAULT_SCHEMA_URL)
+        metrics = self._harvest_metrics()
+
+        self.assertIn("gen_ai.client.operation.duration", metrics)
+        duration_points = metrics["gen_ai.client.operation.duration"]
+        self.assertEqual(len(duration_points), 1)
+        duration_point = duration_points[0]
+
+        self.assertEqual(
+            duration_point.attributes["server.address"], "embed.server.com"
+        )
+        self.assertEqual(duration_point.attributes["server.port"], 8080)
+        self.assertEqual(
+            duration_point.attributes["custom.embed.attr"], "embed_value"
+        )
+        self.assertEqual(
+            duration_point.attributes[GenAI.GEN_AI_RESPONSE_MODEL],
+            "embed-response-model",
+        )
+
+    def test_fail_embedding_records_error_and_duration(self) -> None:
+        """Verify embedding failure records error type and duration."""
+        handler = TelemetryHandler(
+            tracer_provider=self.tracer_provider,
+            meter_provider=self.meter_provider,
+        )
+        with patch("timeit.default_timer", return_value=3000.0):
+            invocation = handler.start_embedding(
+                "embed-prov", request_model="embed-err-model"
+            )
+
+        error = Error(message="embedding failed", type=RuntimeError)
+        with patch("timeit.default_timer", return_value=3002.5):
+            invocation.fail(error)
+
+        self._assert_metric_scope_schema_urls(_DEFAULT_SCHEMA_URL)
+        metrics = self._harvest_metrics()
+
+        self.assertIn("gen_ai.client.operation.duration", metrics)
+        duration_points = metrics["gen_ai.client.operation.duration"]
+        self.assertEqual(len(duration_points), 1)
+        duration_point = duration_points[0]
+
+        self.assertEqual(
+            duration_point.attributes.get("error.type"), "RuntimeError"
+        )
+        self.assertEqual(
+            duration_point.attributes.get(GenAI.GEN_AI_REQUEST_MODEL),
+            "embed-err-model",
+        )
+        self.assertAlmostEqual(duration_point.sum, 2.5, places=3)
+
+        # Token metrics should NOT be recorded when input_tokens is not set
+        self.assertNotIn("gen_ai.client.token.usage", metrics)
+
+    def test_stop_embedding_without_tokens(self) -> None:
+        """Verify embedding without input_tokens does not record token metrics."""
+        handler = TelemetryHandler(
+            tracer_provider=self.tracer_provider,
+            meter_provider=self.meter_provider,
+        )
+        invocation = handler.start_embedding(
+            "embed-prov", request_model="embed-model"
+        )
+        # input_tokens is not set
+        invocation.stop()
+
+        metrics = self._harvest_metrics()
+
+        # Duration should be recorded
+        self.assertIn("gen_ai.client.operation.duration", metrics)
+
+        # Token metrics should NOT be recorded when input_tokens is not set
+        self.assertNotIn("gen_ai.client.token.usage", metrics)