Merge branch 'open-telemetry:main' into xray_traces

Author: liustve

CHANGELOG.md

@@ -47,6 +47,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3200](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3200))
 - `opentelemetry-opentelemetry-botocore` Add basic support for GenAI attributes for AWS Bedrock ConverseStream API
   ([#3204](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3204))
+- `opentelemetry-opentelemetry-botocore` Add basic support for GenAI attributes for AWS Bedrock InvokeModelWithStreamResponse API
+  ([#3206](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3206))
 - `opentelemetry-instrumentation-pymssql` Add pymssql instrumentation
   ([#394](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/394))
 
@@ -65,14 +67,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `opentelemetry-instrumentation-sqlalchemy` including sqlcomment in `db.statement` span attribute value is now opt-in
   ([#3112](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3112))
-
-### Breaking changes
-
-- `opentelemetry-instrumentation-dbapi` including sqlcomment in `db.statement` span attribute value is now opt-in
-  ([#3115](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3115))
-
-### Breaking changes
-
 - `opentelemetry-instrumentation-dbapi` including sqlcomment in `db.statement` span attribute value is now opt-in
   ([#3115](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3115))
 - `opentelemetry-instrumentation-psycopg2`, `opentelemetry-instrumentation-psycopg`, `opentelemetry-instrumentation-mysqlclient`, `opentelemetry-instrumentation-pymysql`: including sqlcomment in `db.statement` span attribute value is now opt-in

instrumentation-genai/opentelemetry-instrumentation-vertexai/CHANGELOG.md

@@ -11,3 +11,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3192](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3192))
 - Initial VertexAI instrumentation
   ([#3123](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3123))
+- Add server attributes to Vertex AI spans
+  ([#3208](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3208))

instrumentation-genai/opentelemetry-instrumentation-vertexai/src/opentelemetry/instrumentation/vertexai/patch.py

@@ -25,6 +25,7 @@
 from opentelemetry.instrumentation.vertexai.utils import (
     GenerateContentParams,
     get_genai_request_attributes,
+    get_server_attributes,
     get_span_name,
 )
 from opentelemetry.trace import SpanKind, Tracer
@@ -100,7 +101,11 @@ def traced_method(
         kwargs: Any,
     ):
         params = _extract_params(*args, **kwargs)
-        span_attributes = get_genai_request_attributes(params)
+        api_endpoint: str = instance.api_endpoint  # type: ignore[reportUnknownMemberType]
+        span_attributes = {
+            **get_genai_request_attributes(params),
+            **get_server_attributes(api_endpoint),
+        }
 
         span_name = get_span_name(span_attributes)
         with tracer.start_as_current_span(

instrumentation-genai/opentelemetry-instrumentation-vertexai/src/opentelemetry/instrumentation/vertexai/utils.py

@@ -22,10 +22,12 @@
     Mapping,
     Sequence,
 )
+from urllib.parse import urlparse
 
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAIAttributes,
 )
+from opentelemetry.semconv.attributes import server_attributes
 from opentelemetry.util.types import AttributeValue
 
 if TYPE_CHECKING:
@@ -58,6 +60,24 @@ class GenerateContentParams:
     ) = None
 
 
+def get_server_attributes(
+    endpoint: str,
+) -> dict[str, AttributeValue]:
+    """Get server.* attributes from the endpoint, which is a hostname with optional port e.g.
+    - ``us-central1-aiplatform.googleapis.com``
+    - ``us-central1-aiplatform.googleapis.com:5431``
+    """
+    parsed = urlparse(f"scheme://{endpoint}")
+
+    if not parsed.hostname:
+        return {}
+
+    return {
+        server_attributes.SERVER_ADDRESS: parsed.hostname,
+        server_attributes.SERVER_PORT: parsed.port or 443,
+    }
+
+
 def get_genai_request_attributes(
     params: GenerateContentParams,
     operation_name: GenAIAttributes.GenAiOperationNameValues = GenAIAttributes.GenAiOperationNameValues.CHAT,

instrumentation-genai/opentelemetry-instrumentation-vertexai/tests/test_chat_completions.py

@@ -34,6 +34,8 @@ def test_generate_content(
         "gen_ai.operation.name": "chat",
         "gen_ai.request.model": "gemini-1.5-flash-002",
         "gen_ai.system": "vertex_ai",
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
     }
 
 
@@ -62,6 +64,8 @@ def test_generate_content_empty_model(
         "gen_ai.operation.name": "chat",
         "gen_ai.request.model": "",
         "gen_ai.system": "vertex_ai",
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
     }
     assert_span_error(spans[0])
 
@@ -91,6 +95,8 @@ def test_generate_content_missing_model(
         "gen_ai.operation.name": "chat",
         "gen_ai.request.model": "gemini-does-not-exist",
         "gen_ai.system": "vertex_ai",
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
     }
     assert_span_error(spans[0])
 
@@ -122,6 +128,8 @@ def test_generate_content_invalid_temperature(
         "gen_ai.request.model": "gemini-1.5-flash-002",
         "gen_ai.request.temperature": 1000.0,
         "gen_ai.system": "vertex_ai",
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
     }
     assert_span_error(spans[0])
 
@@ -158,6 +166,8 @@ def test_generate_content_extra_params(span_exporter, instrument_no_content):
         "gen_ai.request.temperature": 0.20000000298023224,
         "gen_ai.request.top_p": 0.949999988079071,
         "gen_ai.system": "vertex_ai",
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
     }
 
 

instrumentation-genai/opentelemetry-instrumentation-vertexai/tests/test_utils.py

@@ -0,0 +1,32 @@
+# Copyright The OpenTelemetry Authors
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from opentelemetry.instrumentation.vertexai.utils import get_server_attributes
+
+
+def test_get_server_attributes() -> None:
+    # without port
+    assert get_server_attributes("us-central1-aiplatform.googleapis.com") == {
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 443,
+    }
+
+    # with port
+    assert get_server_attributes(
+        "us-central1-aiplatform.googleapis.com:5432"
+    ) == {
+        "server.address": "us-central1-aiplatform.googleapis.com",
+        "server.port": 5432,
+    }

instrumentation/opentelemetry-instrumentation-botocore/examples/bedrock-runtime/zero-code/README.rst

@@ -20,6 +20,7 @@ Available examples
 - `converse.py` uses `bedrock-runtime` `Converse API <https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html>_`.
 - `converse_stream.py` uses `bedrock-runtime` `ConverseStream API <https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ConverseStream.html>_`.
 - `invoke_model.py` uses `bedrock-runtime` `InvokeModel API <https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModel.html>_`.
+- `invoke_model_stream.py` uses `bedrock-runtime` `InvokeModelWithResponseStrea API <https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_InvokeModelWithResponseStream.html>_`.
 
 Setup
 -----

instrumentation/opentelemetry-instrumentation-botocore/examples/bedrock-runtime/zero-code/invoke_model_stream.py

@@ -0,0 +1,51 @@
+import json
+import os
+
+import boto3
+
+
+def main():
+    chat_model = os.getenv("CHAT_MODEL", "amazon.titan-text-lite-v1")
+    prompt = "Write a short poem on OpenTelemetry."
+    if "amazon.titan" in chat_model:
+        body = {
+            "inputText": prompt,
+            "textGenerationConfig": {},
+        }
+    elif "amazon.nova" in chat_model:
+        body = {
+            "messages": [{"role": "user", "content": [{"text": prompt}]}],
+            "schemaVersion": "messages-v1",
+        }
+    elif "anthropic.claude" in chat_model:
+        body = {
+            "messages": [
+                {"role": "user", "content": [{"text": prompt, "type": "text"}]}
+            ],
+            "anthropic_version": "bedrock-2023-05-31",
+            "max_tokens": 200,
+        }
+    else:
+        raise ValueError()
+    client = boto3.client("bedrock-runtime")
+    response = client.invoke_model_with_response_stream(
+        modelId=chat_model,
+        body=json.dumps(body),
+    )
+
+    answer = ""
+    for event in response["body"]:
+        json_bytes = event.get("chunk", {}).get("bytes", b"")
+        decoded = json_bytes.decode("utf-8")
+        chunk = json.loads(decoded)
+        if "outputText" in chunk:
+            answer += chunk["outputText"]
+        elif "completion" in chunk:
+            answer += chunk["completion"]
+        elif "contentBlockDelta" in chunk:
+            answer += chunk["contentBlockDelta"]["delta"]["text"]
+    print(answer)
+
+
+if __name__ == "__main__":
+    main()

instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/extensions/bedrock.py

@@ -28,6 +28,7 @@
 
 from opentelemetry.instrumentation.botocore.extensions.bedrock_utils import (
     ConverseStreamWrapper,
+    InvokeModelWithResponseStreamWrapper,
 )
 from opentelemetry.instrumentation.botocore.extensions.types import (
     _AttributeMapT,
@@ -66,8 +67,16 @@ class _BedrockRuntimeExtension(_AwsSdkExtension):
     Amazon Bedrock Runtime</a>.
     """
 
-    _HANDLED_OPERATIONS = {"Converse", "ConverseStream", "InvokeModel"}
-    _DONT_CLOSE_SPAN_ON_END_OPERATIONS = {"ConverseStream"}
+    _HANDLED_OPERATIONS = {
+        "Converse",
+        "ConverseStream",
+        "InvokeModel",
+        "InvokeModelWithResponseStream",
+    }
+    _DONT_CLOSE_SPAN_ON_END_OPERATIONS = {
+        "ConverseStream",
+        "InvokeModelWithResponseStream",
+    }
 
     def should_end_span_on_exit(self):
         return (
@@ -257,6 +266,12 @@ def _invoke_model_on_success(
             if original_body is not None:
                 original_body.close()
 
+    def _on_stream_error_callback(self, span: Span, exception):
+        span.set_status(Status(StatusCode.ERROR, str(exception)))
+        if span.is_recording():
+            span.set_attribute(ERROR_TYPE, type(exception).__qualname__)
+        span.end()
+
     def on_success(self, span: Span, result: dict[str, Any]):
         if self._call_context.operation not in self._HANDLED_OPERATIONS:
             return
@@ -273,8 +288,11 @@ def stream_done_callback(response):
                 self._converse_on_success(span, response)
                 span.end()
 
+            def stream_error_callback(exception):
+                self._on_stream_error_callback(span, exception)
+
             result["stream"] = ConverseStreamWrapper(
-                result["stream"], stream_done_callback
+                result["stream"], stream_done_callback, stream_error_callback
             )
             return
 
@@ -288,6 +306,26 @@ def stream_done_callback(response):
         # InvokeModel
         if "body" in result and isinstance(result["body"], StreamingBody):
             self._invoke_model_on_success(span, result, model_id)
+            return
+
+        # InvokeModelWithResponseStream
+        if "body" in result and isinstance(result["body"], EventStream):
+
+            def invoke_model_stream_done_callback(response):
+                # the callback gets data formatted as the simpler converse API
+                self._converse_on_success(span, response)
+                span.end()
+
+            def invoke_model_stream_error_callback(exception):
+                self._on_stream_error_callback(span, exception)
+
+            result["body"] = InvokeModelWithResponseStreamWrapper(
+                result["body"],
+                invoke_model_stream_done_callback,
+                invoke_model_stream_error_callback,
+                model_id,
+            )
+            return
 
     # pylint: disable=no-self-use
     def _handle_amazon_titan_response(