refactor: remove ExecutionContext._inner_span (#19084)

This change removes the `_inner_span` attribute from `ddtrace.core.ExecutionContext`, replacing it with an `_inner_span` datum stored via existing mechanisms. This eliminates Spans as a special case in the Core API, treating them like any other piece of data stored in the context tree.

`span_bus` is introduced containing two of the span-related helper functions from `core`, indicating the separateness of Span management from Core. Note that while this separation does fix the import-time coupling between `_trace` and `core`, it does not address the functional dependence of certain products on `_trace`, specifically those that use Spans as context tracking.

Co-authored-by: emmett.butler <emmett.butler@datadoghq.com>

Author: emmettbutler

.gitlab/benchmarks/bp-runner.microbenchmarks.fail-on-breach.template.yml

@@ -123,13 +123,13 @@ experiments:
           # errortrackingflasksqli
           - name: errortrackingflasksqli-errortracking-enabled-all
             thresholds:
-              - execution_time < 2.32 ms
+              - execution_time < 2.34 ms
           - name: errortrackingflasksqli-errortracking-enabled-user
             thresholds:
               - execution_time < 2.85 ms
           - name: errortrackingflasksqli-tracer-enabled
             thresholds:
-              - execution_time < 2.32 ms
+              - execution_time < 2.33 ms
 
           # flask_simple
           - name: flasksimple-tracer

benchmarks/events_api/scenario.py

@@ -9,6 +9,7 @@
 from ddtrace._trace.trace_handlers import _start_span
 from ddtrace.constants import SPAN_KIND
 from ddtrace.internal import core
+from ddtrace.internal import span_bus
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.core.events import Event
 from ddtrace.internal.core.events import event_field
@@ -50,7 +51,7 @@ def benchmark_core_api(loops):
             # Register benchmark specific handlers
             def _context_started_handler(ctx: core.ExecutionContext) -> None:
                 _start_span(ctx, call_trace=True)
-                span = ctx.span
+                span = span_bus.span_from_context(ctx)
                 span._set_attribute("http.url", ctx.get_item("url"))
                 span._set_attribute("http.method", ctx.get_item("method"))
                 span._set_attribute("http.status_code", ctx.get_item("status_code"))
@@ -82,7 +83,7 @@ class SpanContextSubscriber(TracingSubscriber):
 
                 @classmethod
                 def on_started(cls, ctx: core.ExecutionContext) -> None:
-                    span = ctx.span
+                    span = span_bus.span_from_context(ctx)
                     event: BenchmarkTracingEvent = ctx.event
                     span._set_attribute("http.url", event.url)
                     span._set_attribute("http.method", event.method)

ddtrace/_trace/subscribers/_base.py

@@ -17,6 +17,8 @@
 from ddtrace.internal import core
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.core.subscriber import ContextSubscriber
+from ddtrace.internal.span_bus import span_from_context
+from ddtrace.internal.span_bus import store_span_on_context
 from ddtrace.trace import tracer
 
 
@@ -34,7 +36,7 @@ def _finish_span(
     Once every integration adopts the events API, trace_handlers._finish_span
     should be completely removed.
     """
-    span = ctx.span
+    span = span_from_context(ctx)
     if not span:
         return
 
@@ -98,7 +100,7 @@ def _start_span(ctx: core.ExecutionContext[TracingEventType]) -> Span:
         span._set_attribute(_SPAN_MEASURED_KEY, 1)
 
     set_service_and_source(span, ctx.get_item("service"), integration_config or dict())
-    ctx.span = span
+    store_span_on_context(ctx, span)
 
     if config._inferred_proxy_services_enabled:
         # TODO(IDM): Subscriber should be added for Inferred Proxy span handling
@@ -120,12 +122,12 @@ class MySpanSubscriber(SpanTracingSubscriber):
 
             @classmethod
             def on_started(cls, ctx):
-                ctx.span.set_tag("custom.tag", "value")
+                span_from_context(ctx).set_tag("custom.tag", "value")
 
             @classmethod
             def on_ended(cls, ctx, exc_info):
                 if exc_info[1]:
-                    ctx.span.set_tag("error", True)
+                    span_from_context(ctx).set_tag("error", True)
     """
 
     # Register here events that just create / finish spans

ddtrace/_trace/subscribers/aws_durable.py

@@ -8,6 +8,7 @@
 from ddtrace.contrib._events.aws_durable import AwsDurableOperationEvent
 from ddtrace.ext import aws_durable
 from ddtrace.internal import core
+from ddtrace.internal.span_bus import span_from_context
 
 
 class AwsDurableExecuteSubscriber(TracingSubscriber):
@@ -17,9 +18,11 @@ class AwsDurableExecuteSubscriber(TracingSubscriber):
     def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: AwsDurableExecuteEvent = ctx.event
         if event.execution_arn:
-            ctx.span._set_attribute(aws_durable.TAG_EXECUTION_ARN, event.execution_arn)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_EXECUTION_ARN, event.execution_arn)
         if event.is_replay_execution is not None:
-            ctx.span._set_attribute(aws_durable.TAG_REPLAYED, "true" if event.is_replay_execution else "false")
+            span_from_context(ctx)._set_attribute(
+                aws_durable.TAG_REPLAYED, "true" if event.is_replay_execution else "false"
+            )
 
     @classmethod
     def on_ended(
@@ -34,7 +37,7 @@ def on_ended(
             status = "failed"
         else:
             status = "succeeded"
-        ctx.span._set_attribute(aws_durable.TAG_INVOCATION_STATUS, status)
+        span_from_context(ctx)._set_attribute(aws_durable.TAG_INVOCATION_STATUS, status)
 
 
 class AwsDurableInvokeSubscriber(TracingSubscriber):
@@ -43,9 +46,9 @@ class AwsDurableInvokeSubscriber(TracingSubscriber):
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: AwsDurableInvokeEvent = ctx.event
-        ctx.span._set_attribute(aws_durable.TAG_INVOKE_FUNCTION_NAME, event.invoke_function_name)
+        span_from_context(ctx)._set_attribute(aws_durable.TAG_INVOKE_FUNCTION_NAME, event.invoke_function_name)
         if event.name is not None:
-            ctx.span._set_attribute(aws_durable.TAG_NAME, event.name)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_NAME, event.name)
 
     @classmethod
     def on_ended(
@@ -55,9 +58,9 @@ def on_ended(
     ) -> None:
         event: AwsDurableInvokeEvent = ctx.event
         if event.replayed is not None:
-            ctx.span._set_attribute(aws_durable.TAG_REPLAYED, "true" if event.replayed else "false")
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_REPLAYED, "true" if event.replayed else "false")
         if event.id is not None:
-            ctx.span._set_attribute(aws_durable.TAG_ID, event.id)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_ID, event.id)
 
 
 class AwsDurableOperationSubscriber(TracingSubscriber):
@@ -67,7 +70,7 @@ class AwsDurableOperationSubscriber(TracingSubscriber):
     def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: AwsDurableOperationEvent = ctx.event
         if event.name is not None:
-            ctx.span._set_attribute(aws_durable.TAG_NAME, event.name)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_NAME, event.name)
 
     @classmethod
     def on_ended(
@@ -77,8 +80,8 @@ def on_ended(
     ) -> None:
         event: AwsDurableOperationEvent = ctx.event
         if event.replayed is not None:
-            ctx.span._set_attribute(aws_durable.TAG_REPLAYED, "true" if event.replayed else "false")
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_REPLAYED, "true" if event.replayed else "false")
         if event.id is not None:
-            ctx.span._set_attribute(aws_durable.TAG_ID, event.id)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_ID, event.id)
         if event.operation_attempt is not None:
-            ctx.span._set_attribute(aws_durable.TAG_OPERATION_ATTEMPT, event.operation_attempt)
+            span_from_context(ctx)._set_attribute(aws_durable.TAG_OPERATION_ATTEMPT, event.operation_attempt)

ddtrace/_trace/subscribers/http_client.py

@@ -9,6 +9,7 @@
 from ddtrace.contrib._events.http_client import HttpClientRequestEvent
 from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.span_bus import span_from_context
 from ddtrace.propagation.http import HTTPPropagator
 
 
@@ -54,7 +55,7 @@ def on_started(cls, ctx: core.ExecutionContext) -> None:
             return
 
         if trace_utils.distributed_tracing_enabled(event.integration_config) and event.request_headers is not None:
-            HTTPPropagator.inject(ctx.span.context, cast(dict[str, str], event.request_headers))
+            HTTPPropagator.inject(span_from_context(ctx).context, cast(dict[str, str], event.request_headers))
 
     @classmethod
     def on_ended(
@@ -66,7 +67,7 @@ def on_ended(
 
         try:
             trace_utils.set_http_meta(
-                ctx.span,
+                span_from_context(ctx),
                 event.integration_config,
                 method=event.request_method,
                 url=event.request_url,

ddtrace/_trace/subscribers/llm.py

@@ -7,6 +7,7 @@
 from ddtrace.internal import core
 from ddtrace.internal.constants import COMPONENT
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.span_bus import span_from_context
 
 
 # Duplicated from ddtrace.llmobs._constants to avoid importing
@@ -30,7 +31,7 @@ class LlmTracingSubscriber(TracingSubscriber["LlmRequestEvent"]):
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext["LlmRequestEvent"]) -> None:
         event: LlmRequestEvent = ctx.event
-        span = ctx.span
+        span = span_from_context(ctx)
 
         # Remove component/span.kind tags set by _start_span — the old
         # BaseLLMIntegration.trace() never set these, so existing snapshot
@@ -65,7 +66,7 @@ def on_ended(
         """
         event: LlmRequestEvent = ctx.event
         event.llmobs_integration.llmobs_set_tags(
-            ctx.span,
+            span_from_context(ctx),
             args=[],
             kwargs=event.request_kwargs,
             response=event.response,

ddtrace/_trace/subscribers/ray.py

@@ -47,6 +47,7 @@
 from ddtrace.internal import core
 from ddtrace.internal.core.subscriber import Subscriber
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.span_bus import span_from_context
 from ddtrace.propagation.http import _TraceContext
 from ddtrace.trace import tracer
 
@@ -64,7 +65,7 @@ def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: RayJobEvent = ctx.event
         submission_id = event.submission_id
 
-        job_span = ctx.span
+        job_span = span_from_context(ctx)
         _set_dist_ai_metrics(job_span)
         _set_runtime_context_attributes(job_span, submission_id)
         start_long_running_job(job_span)
@@ -92,7 +93,7 @@ def on_ended(
         exc_info: tuple[Optional[type], Optional[BaseException], Optional[TracebackType]],
     ) -> None:
         event: RayJobEvent = ctx.event
-        job_span = ctx.span
+        job_span = span_from_context(ctx)
 
         # a simple if event.submit_failed does not work on event_field
         if event.submit_failed is True:
@@ -113,7 +114,7 @@ class RayExecutionSubscriber(TracingSubscriber):
 
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext) -> None:
-        span = ctx.span
+        span = span_from_context(ctx)
         _set_dist_ai_metrics(span)
         _set_runtime_context_attributes(span)
 
@@ -147,7 +148,7 @@ def on_ended(
         ctx: core.ExecutionContext,
         exc_info: tuple[Optional[type], Optional[BaseException], Optional[TracebackType]],
     ) -> None:
-        span = ctx.span
+        span = span_from_context(ctx)
 
         exc_type, exc_val, exc_tb = exc_info
         if exc_type is not None and exc_val is not None:
@@ -164,7 +165,7 @@ class RayCoreAPITracingSubscriber(TracingSubscriber):
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: RayCoreAPIEvent = ctx.event
-        span = ctx.span
+        span = span_from_context(ctx)
 
         _set_dist_ai_metrics(span)
         _set_runtime_context_attributes(span)
@@ -179,7 +180,7 @@ def on_ended(
         exc_info: tuple[Optional[type], Optional[BaseException], Optional[TracebackType]],
     ) -> None:
         event: RayCoreAPIEvent = ctx.event
-        span = ctx.span
+        span = span_from_context(ctx)
 
         exc_type, exc_val, exc_tb = exc_info
         if exc_type is not None and exc_val is not None:
@@ -200,7 +201,7 @@ class RaySubmissionSubscriber(TracingSubscriber):
     @classmethod
     def on_started(cls, ctx: core.ExecutionContext) -> None:
         event: RaySubmissionEvent = ctx.event
-        span = ctx.span
+        span = span_from_context(ctx)
 
         _set_dist_ai_metrics(span)
         _set_runtime_context_attributes(span)
@@ -250,7 +251,7 @@ def on_ended(
         exc_info: tuple[Optional[type], Optional[BaseException], Optional[TracebackType]],
     ) -> None:
         event: RaySubmissionEvent = ctx.event
-        span = ctx.span
+        span = span_from_context(ctx)
 
         status_tag = RAY_TASK_SUBMIT_STATUS if event.is_task_submission else RAY_ACTOR_METHOD_SUBMIT_STATUS
         span._set_attribute(status_tag, RAY_STATUS_ERROR if exc_info[1] is not None else RAY_STATUS_SUCCESS)

ddtrace/_trace/subscribers/web_framework.py

@@ -9,6 +9,7 @@
 from ddtrace.ext import http
 from ddtrace.internal import core
 from ddtrace.internal.logger import get_logger
+from ddtrace.internal.span_bus import span_from_context
 
 
 log = get_logger(__name__)
@@ -37,7 +38,7 @@ def on_ended(
         method = event.request_method
         res_headers = event.response_headers
 
-        span: Span = ctx.span
+        span: Span = span_from_context(ctx)
 
         # event.resource can be updated at span finish time
         if event.resource: