Update the botocore instrumentation to emit events via the logs API instead of the deprecated events API (open-telemetry#3624)

* Initial commit to replace deprecated events API/
SDK with logs.

* Use LogRecord in API instead of SDK

* Use LogRecord from APi instead of Events

* Add changelog, fix lint, fix ruff

* Respond to review comments

* Require v1.35 of otel python

* Specify the .0

* Fix changelog

* Update CHANGELOG with recent fixes

* respond to comments

* Update bedrock.py

* Update instrumentation/opentelemetry-instrumentation-botocore/pyproject.toml

---------

Co-authored-by: Riccardo Magliocchetti <[email protected]>
Co-authored-by: Emídio Neto <[email protected]>

Author: 3 people

CHANGELOG.md

@@ -12,10 +12,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## Unreleased
 
 ### Fixed
+
+- `opentelemetry-instrumentation-botocore`: migrate off the deprecated events API to use the logs API
+  ([#3624](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3624))
 - `opentelemetry-instrumentation-dbapi`: fix crash retrieving libpq version when enabling commenter with psycopg
   ([#3796](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3796))
 
 ### Added
+
 - `opentelemetry-instrumentation`: botocore: Add support for AWS Secrets Manager semantic convention attribute
   ([#3765](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3765))
 - Add `rstcheck` to pre-commit to stop introducing invalid RST

instrumentation/opentelemetry-instrumentation-botocore/pyproject.toml

@@ -25,7 +25,7 @@ classifiers = [
   "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
-  "opentelemetry-api ~= 1.30",
+  "opentelemetry-api ~= 1.37",
   "opentelemetry-instrumentation == 0.59b0.dev",
   "opentelemetry-semantic-conventions == 0.59b0.dev",
   "opentelemetry-propagator-aws-xray ~= 1.0",

instrumentation/opentelemetry-instrumentation-botocore/src/opentelemetry/instrumentation/botocore/__init__.py

@@ -88,7 +88,7 @@ def response_hook(span, service_name, operation_name, result):
 from botocore.exceptions import ClientError
 from wrapt import wrap_function_wrapper
 
-from opentelemetry._events import get_event_logger
+from opentelemetry._logs import get_logger
 from opentelemetry.instrumentation.botocore.extensions import (
     _find_extension,
     _has_extension,
@@ -139,8 +139,8 @@ def _instrument(self, **kwargs):
 
         # tracers are lazy initialized per-extension in _get_tracer
         self._tracers = {}
-        # event_loggers are lazy initialized per-extension in _get_event_logger
-        self._event_loggers = {}
+        # loggers are lazy initialized per-extension in _get_logger
+        self._loggers = {}
         # meters are lazy initialized per-extension in _get_meter
         self._meters = {}
         # metrics are lazy initialized per-extension in _get_metrics
@@ -154,7 +154,7 @@ def _instrument(self, **kwargs):
             self.propagator = propagator
 
         self.tracer_provider = kwargs.get("tracer_provider")
-        self.event_logger_provider = kwargs.get("event_logger_provider")
+        self.logger_provider = kwargs.get("logger_provider")
         self.meter_provider = kwargs.get("meter_provider")
 
         wrap_function_wrapper(
@@ -195,23 +195,23 @@ def _get_tracer(self, extension: _AwsSdkExtension):
         )
         return self._tracers[instrumentation_name]
 
-    def _get_event_logger(self, extension: _AwsSdkExtension):
-        """This is a multiplexer in order to have an event logger per extension"""
+    def _get_logger(self, extension: _AwsSdkExtension):
+        """This is a multiplexer in order to have a logger per extension"""
 
         instrumentation_name = self._get_instrumentation_name(extension)
-        event_logger = self._event_loggers.get(instrumentation_name)
-        if event_logger:
-            return event_logger
+        instrumentation_logger = self._loggers.get(instrumentation_name)
+        if instrumentation_logger:
+            return instrumentation_logger
 
         schema_version = extension.event_logger_schema_version()
-        self._event_loggers[instrumentation_name] = get_event_logger(
+        self._loggers[instrumentation_name] = get_logger(
             instrumentation_name,
             "",
             schema_url=f"https://opentelemetry.io/schemas/{schema_version}",
-            event_logger_provider=self.event_logger_provider,
+            logger_provider=self.logger_provider,
         )
 
-        return self._event_loggers[instrumentation_name]
+        return self._loggers[instrumentation_name]
 
     def _get_meter(self, extension: _AwsSdkExtension):
         """This is a multiplexer in order to have a meter per extension"""
@@ -287,11 +287,10 @@ def _patched_api_call(self, original_func, instance, args, kwargs):
         end_span_on_exit = extension.should_end_span_on_exit()
 
         tracer = self._get_tracer(extension)
-        event_logger = self._get_event_logger(extension)
         meter = self._get_meter(extension)
         metrics = self._get_metrics(extension, meter)
         instrumentor_ctx = _BotocoreInstrumentorContext(
-            event_logger=event_logger,
+            logger=self._get_logger(extension),
             metrics=metrics,
         )
         with tracer.start_as_current_span(

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

@@ -28,6 +28,7 @@
 from botocore.eventstream import EventStream
 from botocore.response import StreamingBody
 
+from opentelemetry.context import get_current
 from opentelemetry.instrumentation.botocore.extensions.bedrock_utils import (
     ConverseStreamWrapper,
     InvokeModelWithResponseStreamWrapper,
@@ -67,6 +68,7 @@
     GEN_AI_CLIENT_OPERATION_DURATION,
     GEN_AI_CLIENT_TOKEN_USAGE,
 )
+from opentelemetry.trace.propagation import set_span_in_context
 from opentelemetry.trace.span import Span
 from opentelemetry.trace.status import Status, StatusCode
 
@@ -457,9 +459,9 @@ def before_service_call(
 
         messages = self._get_request_messages()
         for message in messages:
-            event_logger = instrumentor_context.event_logger
+            logger = instrumentor_context.logger
             for event in message_to_event(message, capture_content):
-                event_logger.emit(event)
+                logger.emit(event)
 
         if span.is_recording():
             operation_name = span.attributes.get(GEN_AI_OPERATION_NAME, "")
@@ -501,18 +503,12 @@ def _converse_on_success(
 
         # In case of an early stream closure, the result may not contain outputs
         if self._stream_has_output_content(result):
-            event_logger = instrumentor_context.event_logger
+            logger = instrumentor_context.logger
             choice = _Choice.from_converse(result, capture_content)
             # this path is used by streaming apis, in that case we are already out of the span
             # context so need to add the span context manually
-            span_ctx = span.get_span_context()
-            event_logger.emit(
-                choice.to_choice_event(
-                    trace_id=span_ctx.trace_id,
-                    span_id=span_ctx.span_id,
-                    trace_flags=span_ctx.trace_flags,
-                )
-            )
+            context = set_span_in_context(span, get_current())
+            logger.emit(choice.to_choice_event(context=context))
 
         metrics = instrumentor_context.metrics
         metrics_attributes = self._extract_metrics_attributes()
@@ -729,11 +725,11 @@ def _handle_amazon_titan_response(
                     [result["completionReason"]],
                 )
 
-            event_logger = instrumentor_context.event_logger
+            logger = instrumentor_context.logger
             choice = _Choice.from_invoke_amazon_titan(
                 response_body, capture_content
             )
-            event_logger.emit(choice.to_choice_event())
+            logger.emit(choice.to_choice_event())
 
             metrics = instrumentor_context.metrics
             metrics_attributes = self._extract_metrics_attributes()
@@ -791,9 +787,9 @@ def _handle_amazon_nova_response(
 
         # In case of an early stream closure, the result may not contain outputs
         if self._stream_has_output_content(response_body):
-            event_logger = instrumentor_context.event_logger
+            logger = instrumentor_context.logger
             choice = _Choice.from_converse(response_body, capture_content)
-            event_logger.emit(choice.to_choice_event())
+            logger.emit(choice.to_choice_event())
 
         metrics = instrumentor_context.metrics
         metrics_attributes = self._extract_metrics_attributes()
@@ -848,11 +844,11 @@ def _handle_anthropic_claude_response(
                 GEN_AI_RESPONSE_FINISH_REASONS, [response_body["stop_reason"]]
             )
 
-        event_logger = instrumentor_context.event_logger
+        logger = instrumentor_context.logger
         choice = _Choice.from_invoke_anthropic_claude(
             response_body, capture_content
         )
-        event_logger.emit(choice.to_choice_event())
+        logger.emit(choice.to_choice_event())
 
         metrics = instrumentor_context.metrics
         metrics_attributes = self._extract_metrics_attributes()
@@ -903,11 +899,11 @@ def _handle_cohere_command_r_response(
                 [response_body["finish_reason"]],
             )
 
-        event_logger = instrumentor_context.event_logger
+        logger = instrumentor_context.logger
         choice = _Choice.from_invoke_cohere_command_r(
             response_body, capture_content
         )
-        event_logger.emit(choice.to_choice_event())
+        logger.emit(choice.to_choice_event())
 
     def _handle_cohere_command_response(
         self,
@@ -929,11 +925,11 @@ def _handle_cohere_command_response(
                     [generations["finish_reason"]],
                 )
 
-        event_logger = instrumentor_context.event_logger
+        logger = instrumentor_context.logger
         choice = _Choice.from_invoke_cohere_command(
             response_body, capture_content
         )
-        event_logger.emit(choice.to_choice_event())
+        logger.emit(choice.to_choice_event())
 
     def _handle_meta_llama_response(
         self,
@@ -956,9 +952,9 @@ def _handle_meta_llama_response(
                 GEN_AI_RESPONSE_FINISH_REASONS, [response_body["stop_reason"]]
             )
 
-        event_logger = instrumentor_context.event_logger
+        logger = instrumentor_context.logger
         choice = _Choice.from_invoke_meta_llama(response_body, capture_content)
-        event_logger.emit(choice.to_choice_event())
+        logger.emit(choice.to_choice_event())
 
     def _handle_mistral_ai_response(
         self,
@@ -979,11 +975,11 @@ def _handle_mistral_ai_response(
                     GEN_AI_RESPONSE_FINISH_REASONS, [outputs["stop_reason"]]
                 )
 
-        event_logger = instrumentor_context.event_logger
+        logger = instrumentor_context.logger
         choice = _Choice.from_invoke_mistral_mistral(
             response_body, capture_content
         )
-        event_logger.emit(choice.to_choice_event())
+        logger.emit(choice.to_choice_event())
 
     def on_error(
         self,

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

@@ -22,7 +22,8 @@
 from botocore.eventstream import EventStream, EventStreamError
 from wrapt import ObjectProxy
 
-from opentelemetry._events import Event
+from opentelemetry._logs import LogRecord
+from opentelemetry.context import get_current
 from opentelemetry.instrumentation.botocore.environment_variables import (
     OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT,
 )
@@ -492,7 +493,7 @@ def extract_tool_results(
 
 def message_to_event(
     message: dict[str, Any], capture_content: bool
-) -> Iterator[Event]:
+) -> Iterator[LogRecord]:
     attributes = {GEN_AI_SYSTEM: GenAiSystemValues.AWS_BEDROCK.value}
     role = message.get("role")
     content = message.get("content")
@@ -507,16 +508,18 @@ def message_to_event(
     elif role == "user":
         # in case of tool calls we send one tool event for tool call and one for the user event
         for tool_body in extract_tool_results(message, capture_content):
-            yield Event(
-                name="gen_ai.tool.message",
+            yield LogRecord(
+                event_name="gen_ai.tool.message",
                 attributes=attributes,
                 body=tool_body,
+                context=get_current(),
             )
 
-    yield Event(
-        name=f"gen_ai.{role}.message",
+    yield LogRecord(
+        event_name=f"gen_ai.{role}.message",
         attributes=attributes,
         body=body if body else None,
+        context=get_current(),
     )
 
 
@@ -617,10 +620,10 @@ def _to_body_dict(self) -> dict[str, Any]:
             "message": self.message,
         }
 
-    def to_choice_event(self, **event_kwargs) -> Event:
+    def to_choice_event(self, **event_kwargs) -> LogRecord:
         attributes = {GEN_AI_SYSTEM: GenAiSystemValues.AWS_BEDROCK.value}
-        return Event(
-            name="gen_ai.choice",
+        return LogRecord(
+            event_name="gen_ai.choice",
             attributes=attributes,
             body=self._to_body_dict(),
             **event_kwargs,

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

@@ -17,7 +17,7 @@
 import logging
 from typing import Any, Dict, Optional, Tuple
 
-from opentelemetry._events import EventLogger
+from opentelemetry._logs import Logger
 from opentelemetry.metrics import Instrument, Meter
 from opentelemetry.trace import SpanKind
 from opentelemetry.trace.span import Span
@@ -96,10 +96,10 @@ def _get_attr(obj, name: str, default=None):
 class _BotocoreInstrumentorContext:
     def __init__(
         self,
-        event_logger: EventLogger,
+        logger: Logger,
         metrics: Dict[str, Instrument] | None = None,
     ):
-        self.event_logger = event_logger
+        self.logger = logger
         self.metrics = metrics or {}
 
 

instrumentation/opentelemetry-instrumentation-botocore/tests/bedrock_utils.py

@@ -26,9 +26,6 @@
 )
 from opentelemetry.sdk.metrics._internal.point import ResourceMetrics
 from opentelemetry.sdk.trace import ReadableSpan
-from opentelemetry.semconv._incubating.attributes import (
-    event_attributes as EventAttributes,
-)
 from opentelemetry.semconv._incubating.attributes import (
     gen_ai_attributes as GenAIAttributes,
 )
@@ -282,17 +279,17 @@ def remove_none_values(body):
 
 def assert_log_parent(log, span):
     if span:
-        assert log.log_record.trace_id == span.get_span_context().trace_id
+        assert (
+            log.log_record.trace_id == span.get_span_context().trace_id
+        ), f"{span.get_span_context().trace_id} does not equal {log.log_record.trace_id}"
         assert log.log_record.span_id == span.get_span_context().span_id
         assert (
             log.log_record.trace_flags == span.get_span_context().trace_flags
         )
 
 
 def assert_message_in_logs(log, event_name, expected_content, parent_span):
-    assert (
-        log.log_record.attributes[EventAttributes.EVENT_NAME] == event_name
-    ), log.log_record.attributes[EventAttributes.EVENT_NAME]
+    assert log.log_record.event_name == event_name, log.log_record.event_name
     assert (
         log.log_record.attributes[GenAIAttributes.GEN_AI_SYSTEM]
         == GenAIAttributes.GenAiSystemValues.AWS_BEDROCK.value