[Core] Tracing updates (#39563)

This introduces native OpenTelemetry tracing to Azure Core. 

Signed-off-by: Paul Van Eck <paulvaneck@microsoft.com>

Author: pvaneck

Diff for: .vscode/cspell.json

@@ -354,6 +354,7 @@
     "onmicrosoft",
     "openai",
     "OPENAI",
+    "otel",
     "otlp",
     "OTLP",
     "owasp",

Diff for: sdk/core/azure-core/CHANGELOG.md

@@ -1,15 +1,34 @@
 # Release History
 
-## 1.32.1 (Unreleased)
+## 1.33.0 (Unreleased)
 
 ### Features Added
 
+- Added native OpenTelemetry tracing to Azure Core which enables users to use OpenTelemetry to trace Azure SDK operations without needing to install a plugin. #39563
+    - To enable native OpenTelemetry tracing, users need to:
+        1. Have `opentelemetry-api` installed.
+        2. Ensure that `settings.tracing_implementation` is not set.
+        3. Ensure that `settings.tracing_enabled` is set to `True`.
+    - If `setting.tracing_implementation` is set, the tracing plugin will be used instead of the native tracing.
+    - If `settings.tracing_enabled` is set to `False`, tracing will be disabled.
+    - The `OpenTelemetryTracer` class was added to the `azure.core.tracing.opentelemetry` module. This is a wrapper around the OpenTelemetry tracer that is used to create spans for Azure SDK operations.
+    - Added a `get_tracer` method to the new `azure.core.instrumentation` module. This method returns an instance of the `OpenTelemetryTracer` class if OpenTelemetry is available.
+    - A `TracingOptions` TypedDict class was added to define the options that SDK users can use to configure tracing per-operation. These options include the ability to enable or disable tracing and set additional attributes on spans.
+        - Example usage: `client.method(tracing_options={"enabled": True, "attributes": {"foo": "bar"}})`
+    - The `DistributedTracingPolicy` and `distributed_trace`/`distributed_trace_async` decorators now uses the OpenTelemetry tracer if it is available and native tracing is enabled.
+        - SDK clients can define an `_instrumentation_config` class variable to configure the OpenTelemetry tracer used in method span creation. Possible configuration options are `library_name`, `library_version`, `schema_url`, and `attributes`.
+        - `DistributedTracingPolicy` now accepts a `instrumentation_config` keyword argument to configure the OpenTelemetry tracer used in HTTP span creation.
+
 ### Breaking Changes
 
+- Removed automatic tracing enablement for the OpenTelemetry plugin if `opentelemetry` was imported. To enable tracing with the plugin, please import `azure.core.settings.settings` and set `settings.tracing_implementation` to `"opentelemetry"`. #39563
+
 ### Bugs Fixed
 
 ### Other Changes
 
+- Added `opentelemetry-api` as an optional dependency for tracing. This can be installed with `pip install azure-core[tracing]`. #39563
+
 ## 1.32.0 (2024-10-31)
 
 ### Features Added

Diff for: sdk/core/azure-core/azure/core/_version.py

@@ -9,4 +9,4 @@
 # regenerated.
 # --------------------------------------------------------------------------
 
-VERSION = "1.32.1"
+VERSION = "1.33.0"

Diff for: sdk/core/azure-core/azure/core/instrumentation.py

@@ -0,0 +1,67 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+from typing import Optional, Union, Mapping, TYPE_CHECKING
+from functools import lru_cache
+
+if TYPE_CHECKING:
+    from .tracing.opentelemetry import OpenTelemetryTracer
+
+
+def _get_tracer_impl():
+    # Check if OpenTelemetry is available/installed.
+    try:
+        from .tracing.opentelemetry import OpenTelemetryTracer
+
+        return OpenTelemetryTracer
+    except ImportError:
+        return None
+
+
+@lru_cache
+def _get_tracer_cached(
+    library_name: Optional[str],
+    library_version: Optional[str],
+    schema_url: Optional[str],
+    attributes_key: Optional[frozenset],
+) -> Optional["OpenTelemetryTracer"]:
+    tracer_impl = _get_tracer_impl()
+    if tracer_impl:
+        # Convert attributes_key back to dict if needed
+        attributes = dict(attributes_key) if attributes_key else None
+        return tracer_impl(
+            library_name=library_name,
+            library_version=library_version,
+            schema_url=schema_url,
+            attributes=attributes,
+        )
+    return None
+
+
+def get_tracer(
+    *,
+    library_name: Optional[str] = None,
+    library_version: Optional[str] = None,
+    schema_url: Optional[str] = None,
+    attributes: Optional[Mapping[str, Union[str, bool, int, float]]] = None,
+) -> Optional["OpenTelemetryTracer"]:
+    """Get the OpenTelemetry tracer instance if available.
+
+    If OpenTelemetry is not available, this method will return None. This method caches
+    the tracer instance for each unique set of parameters.
+
+    :keyword library_name: The name of the library to use in the tracer.
+    :paramtype library_name: str
+    :keyword library_version: The version of the library to use in the tracer.
+    :paramtype library_version: str
+    :keyword schema_url: Specifies the Schema URL of the emitted spans. Defaults to
+        "https://opentelemetry.io/schemas/1.23.1".
+    :paramtype schema_url: str
+    :keyword attributes: Attributes to add to the emitted spans.
+    :paramtype attributes: Mapping[str, Union[str, bool, int, float]]
+    :return: The OpenTelemetry tracer instance if available.
+    :rtype: Optional[~azure.core.tracing.opentelemetry.OpenTelemetryTracer]
+    """
+    attributes_key = frozenset(attributes.items()) if attributes else None
+    return _get_tracer_cached(library_name, library_version, schema_url, attributes_key)

Diff for: sdk/core/azure-core/azure/core/pipeline/_base.py

@@ -53,6 +53,7 @@
 
 def cleanup_kwargs_for_transport(kwargs: Dict[str, str]) -> None:
     """Remove kwargs that are not meant for the transport layer.
+
     :param kwargs: The keyword arguments.
     :type kwargs: dict
 
@@ -62,8 +63,9 @@ def cleanup_kwargs_for_transport(kwargs: Dict[str, str]) -> None:
       to the transport layer. This code is needed to handle the case that the
       SensitiveHeaderCleanupPolicy is not added into the pipeline and "insecure_domain_change" is not popped.
     "enable_cae" is added to the `get_token` method of the `TokenCredential` protocol.
+    "tracing_options" is used in the DistributedTracingPolicy and tracing decorators.
     """
-    kwargs_to_remove = ["insecure_domain_change", "enable_cae"]
+    kwargs_to_remove = ["insecure_domain_change", "enable_cae", "tracing_options"]
     if not kwargs:
         return
     for key in kwargs_to_remove:

Diff for: sdk/core/azure-core/azure/core/pipeline/policies/_distributed_tracing.py

@@ -27,7 +27,7 @@
 import logging
 import sys
 import urllib.parse
-from typing import TYPE_CHECKING, Optional, Tuple, TypeVar, Union, Any, Type
+from typing import TYPE_CHECKING, Optional, Tuple, TypeVar, Union, Any, Type, Mapping, Dict
 from types import TracebackType
 
 from azure.core.pipeline import PipelineRequest, PipelineResponse
@@ -39,11 +39,11 @@
 from azure.core.rest import HttpResponse, HttpRequest
 from azure.core.settings import settings
 from azure.core.tracing import SpanKind
+from azure.core.instrumentation import get_tracer
+from azure.core.tracing._models import TracingOptions
 
 if TYPE_CHECKING:
-    from azure.core.tracing._abstract_span import (
-        AbstractSpan,
-    )
+    from opentelemetry.trace import Span
 
 HTTPResponseType = TypeVar("HTTPResponseType", HttpResponse, LegacyHttpResponse)
 HTTPRequestType = TypeVar("HTTPRequestType", HttpRequest, LegacyHttpRequest)
@@ -74,37 +74,91 @@ class DistributedTracingPolicy(SansIOHTTPPolicy[HTTPRequestType, HTTPResponseTyp
     :type network_span_namer: callable[[~azure.core.pipeline.transport.HttpRequest], str]
     :keyword tracing_attributes: Attributes to set on all created spans
     :type tracing_attributes: dict[str, str]
+    :keyword instrumentation_config: Configuration for the instrumentation providers
+    :type instrumentation_config: dict[str, Any]
     """
 
     TRACING_CONTEXT = "TRACING_CONTEXT"
+    _SUPPRESSION_TOKEN = "SUPPRESSION_TOKEN"
+
+    # Current stable HTTP semantic conventions
+    _HTTP_RESEND_COUNT = "http.request.resend_count"
+    _USER_AGENT_ORIGINAL = "user_agent.original"
+    _HTTP_REQUEST_METHOD = "http.request.method"
+    _URL_FULL = "url.full"
+    _HTTP_RESPONSE_STATUS_CODE = "http.response.status_code"
+    _SERVER_ADDRESS = "server.address"
+    _SERVER_PORT = "server.port"
+    _ERROR_TYPE = "error.type"
+
+    # Azure attributes
     _REQUEST_ID = "x-ms-client-request-id"
     _RESPONSE_ID = "x-ms-request-id"
-    _HTTP_RESEND_COUNT = "http.request.resend_count"
 
-    def __init__(self, **kwargs: Any):
+    def __init__(self, *, instrumentation_config: Optional[Mapping[str, Any]] = None, **kwargs: Any):
         self._network_span_namer = kwargs.get("network_span_namer", _default_network_span_namer)
         self._tracing_attributes = kwargs.get("tracing_attributes", {})
+        self._instrumentation_config = instrumentation_config
 
     def on_request(self, request: PipelineRequest[HTTPRequestType]) -> None:
         ctxt = request.context.options
         try:
-            span_impl_type = settings.tracing_implementation()
-            if span_impl_type is None:
+            tracing_options: TracingOptions = ctxt.pop("tracing_options", {})
+            tracing_enabled = settings.tracing_enabled()
+
+            # User can explicitly disable tracing for this request.
+            user_enabled = tracing_options.get("enabled")
+            if user_enabled is False:
+                return
+
+            # If tracing is disabled globally and user didn't explicitly enable it, don't trace.
+            if not tracing_enabled and user_enabled is None:
                 return
 
+            span_impl_type = settings.tracing_implementation()
             namer = ctxt.pop("network_span_namer", self._network_span_namer)
             tracing_attributes = ctxt.pop("tracing_attributes", self._tracing_attributes)
             span_name = namer(request.http_request)
 
-            span = span_impl_type(name=span_name, kind=SpanKind.CLIENT)
-            for attr, value in tracing_attributes.items():
-                span.add_attribute(attr, value)
-            span.start()
+            span_attributes = {**tracing_attributes, **tracing_options.get("attributes", {})}
 
-            headers = span.to_header()
-            request.http_request.headers.update(headers)
+            if span_impl_type:
+                # If the plugin is enabled, prioritize it over the core tracing.
+                span = span_impl_type(name=span_name, kind=SpanKind.CLIENT)
+                for attr, value in span_attributes.items():
+                    span.add_attribute(attr, value)  # type: ignore
+
+                headers = span.to_header()
+                request.http_request.headers.update(headers)
+                request.context[self.TRACING_CONTEXT] = span
+            else:
+                # Otherwise, use the core tracing.
+                config = self._instrumentation_config or {}
+                tracer = get_tracer(
+                    library_name=config.get("library_name"),
+                    library_version=config.get("library_version"),
+                    attributes=config.get("attributes"),
+                )
+                if not tracer:
+                    _LOGGER.warning(
+                        "Tracing is enabled, but not able to get an OpenTelemetry tracer. "
+                        "Please ensure that `opentelemetry-api` is installed."
+                    )
+                    return
+
+                otel_span = tracer.start_span(
+                    name=span_name,
+                    kind=SpanKind.CLIENT,
+                    attributes=span_attributes,
+                )
+
+                trace_context_headers = tracer.get_trace_context()
+                request.http_request.headers.update(trace_context_headers)
+                request.context[self.TRACING_CONTEXT] = otel_span
+
+                token = tracer._suppress_auto_http_instrumentation()  # pylint: disable=protected-access
+                request.context[self._SUPPRESSION_TOKEN] = token
 
-            request.context[self.TRACING_CONTEXT] = span
         except Exception as err:  # pylint: disable=broad-except
             _LOGGER.warning("Unable to start network span: %s", err)
 
@@ -126,21 +180,52 @@ def end_span(
         if self.TRACING_CONTEXT not in request.context:
             return
 
-        span: "AbstractSpan" = request.context[self.TRACING_CONTEXT]
+        span = request.context[self.TRACING_CONTEXT]
+        if not span:
+            return
+
         http_request: Union[HttpRequest, LegacyHttpRequest] = request.http_request
-        if span is not None:
-            span.set_http_attributes(http_request, response=response)
-            if request.context.get("retry_count"):
-                span.add_attribute(self._HTTP_RESEND_COUNT, request.context["retry_count"])
-            request_id = http_request.headers.get(self._REQUEST_ID)
-            if request_id is not None:
-                span.add_attribute(self._REQUEST_ID, request_id)
-            if response and self._RESPONSE_ID in response.headers:
-                span.add_attribute(self._RESPONSE_ID, response.headers[self._RESPONSE_ID])
+
+        attributes: Dict[str, Any] = {}
+        if request.context.get("retry_count"):
+            attributes[self._HTTP_RESEND_COUNT] = request.context["retry_count"]
+        if http_request.headers.get(self._REQUEST_ID):
+            attributes[self._REQUEST_ID] = http_request.headers[self._REQUEST_ID]
+        if response and self._RESPONSE_ID in response.headers:
+            attributes[self._RESPONSE_ID] = response.headers[self._RESPONSE_ID]
+
+        # We'll determine if the span is from a plugin or the core tracing library based on the presence of the
+        # `set_http_attributes` method.
+        if hasattr(span, "set_http_attributes"):
+            # Plugin-based tracing
+            span.set_http_attributes(request=http_request, response=response)
+            for key, value in attributes.items():
+                span.add_attribute(key, value)
             if exc_info:
                 span.__exit__(*exc_info)
             else:
                 span.finish()
+        else:
+            # Native tracing
+            self._set_http_client_span_attributes(span, request=http_request, response=response)
+            span.set_attributes(attributes)
+            if exc_info:
+                # If there was an exception, set the error.type attribute.
+                exception_type = exc_info[0]
+                if exception_type:
+                    module = exception_type.__module__ if exception_type.__module__ != "builtins" else ""
+                    error_type = f"{module}.{exception_type.__qualname__}" if module else exception_type.__qualname__
+                    span.set_attribute(self._ERROR_TYPE, error_type)
+
+                span.__exit__(*exc_info)
+            else:
+                span.end()
+
+        suppression_token = request.context.get(self._SUPPRESSION_TOKEN)
+        if suppression_token:
+            tracer = get_tracer()
+            if tracer:
+                tracer._detach_from_context(suppression_token)  # pylint: disable=protected-access
 
     def on_response(
         self,
@@ -151,3 +236,39 @@ def on_response(
 
     def on_exception(self, request: PipelineRequest[HTTPRequestType]) -> None:
         self.end_span(request, exc_info=sys.exc_info())
+
+    def _set_http_client_span_attributes(
+        self,
+        span: "Span",
+        request: Union[HttpRequest, LegacyHttpRequest],
+        response: Optional[HTTPResponseType] = None,
+    ) -> None:
+        """Add attributes to an HTTP client span.
+
+        :param span: The span to add attributes to.
+        :type span: ~opentelemetry.trace.Span
+        :param request: The request made
+        :type request: ~azure.core.rest.HttpRequest
+        :param response: The response received from the server. Is None if no response received.
+        :type response: ~azure.core.rest.HTTPResponse or ~azure.core.pipeline.transport.HttpResponse
+        """
+        attributes: Dict[str, Any] = {
+            self._HTTP_REQUEST_METHOD: request.method,
+            self._URL_FULL: request.url,
+        }
+
+        parsed_url = urllib.parse.urlparse(request.url)
+        if parsed_url.hostname:
+            attributes[self._SERVER_ADDRESS] = parsed_url.hostname
+        if parsed_url.port:
+            attributes[self._SERVER_PORT] = parsed_url.port
+
+        user_agent = request.headers.get("User-Agent")
+        if user_agent:
+            attributes[self._USER_AGENT_ORIGINAL] = user_agent
+        if response and response.status_code:
+            attributes[self._HTTP_RESPONSE_STATUS_CODE] = response.status_code
+            if response.status_code >= 400:
+                attributes[self._ERROR_TYPE] = str(response.status_code)
+
+        span.set_attributes(attributes)