diff --git a/ddtrace/_trace/context.py b/ddtrace/_trace/context.py index bf1c277f82e..8e593056620 100644 --- a/ddtrace/_trace/context.py +++ b/ddtrace/_trace/context.py @@ -8,6 +8,7 @@ from ddtrace.constants import _ORIGIN_KEY from ddtrace.constants import _SAMPLING_PRIORITY_KEY from ddtrace.constants import _USER_ID_KEY +from ddtrace.internal import core from ddtrace.internal.compat import NumericType from ddtrace.internal.constants import MAX_UINT_64BITS as _MAX_UINT_64BITS from ddtrace.internal.constants import W3C_TRACEPARENT_KEY @@ -31,6 +32,9 @@ _DD_ORIGIN_INVALID_CHARS_REGEX = re.compile(r"[^\x20-\x7E]+") +# For listeners that mirror the W3C trace-flags byte outside the tracer. +SAMPLING_DECISION_EVENT = "ddtrace.trace.sampling_decision" + log = get_logger(__name__) @@ -136,8 +140,12 @@ def sampling_priority(self, value: Optional[NumericType]) -> None: if value is None: if _SAMPLING_PRIORITY_KEY in self._metrics: del self._metrics[_SAMPLING_PRIORITY_KEY] - return - self._metrics[_SAMPLING_PRIORITY_KEY] = value + else: + self._metrics[_SAMPLING_PRIORITY_KEY] = value + # Trace sampling usually decides at trace-chunk finish, long after anything + # mirroring the trace-flags byte published it. Dispatched outside the lock + # because listeners run arbitrary code; runs once per trace, so not hot. + core.dispatch(SAMPLING_DECISION_EVENT) @property def _traceparent(self) -> str: diff --git a/ddtrace/internal/ci_visibility/context.py b/ddtrace/internal/ci_visibility/context.py index e6b198780cb..eaeb28a1cd6 100644 --- a/ddtrace/internal/ci_visibility/context.py +++ b/ddtrace/internal/ci_visibility/context.py @@ -44,3 +44,9 @@ def active(self) -> ContextTypeValue: if isinstance(item, Span): return self._update_active(item) return item + + def _peek_active(self) -> ContextTypeValue: + # Required because this provider has its own storage: the inherited + # implementation reads the default contextvar. Not read-only, unlike the + # inherited one -- nothing needs that of this provider yet. + return self.active() diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi index 2c5855d60d7..495aef222da 100644 --- a/ddtrace/internal/native/_native.pyi +++ b/ddtrace/internal/native/_native.pyi @@ -191,6 +191,18 @@ if sys.platform == "linux": :param trace_flags: W3C Trace Context trace-flags byte (bit 0 = sampled). """ ... + def update_otel_thread_context_ids( + trace_id: int, span_id: int, trace_flags: int, local_root_span_id: int + ) -> None: + """ + Update the OTel thread context from raw ids, for an active ``Context``. + + :param trace_id: 128-bit trace id. + :param span_id: The span this execution is attributable to. + :param trace_flags: W3C Trace Context trace-flags byte (bit 0 = sampled). + :param local_root_span_id: Local root span id. + """ + ... def detach_otel_thread_context() -> None: """Detach the OTel thread context from the current thread.""" ... @@ -1627,6 +1639,13 @@ class BaseContextProvider(abc.ABC): def activate(self, ctx: Optional[ActiveTrace]) -> None: ... @abc.abstractmethod def active(self) -> Optional[ActiveTrace]: ... + def _peek_active(self) -> Optional[ActiveTrace]: + """``active()`` without its repair: no contextvar write, no activate dispatch. + + For observers that must not perturb the active trace, such as anything + running from a CPython context-switch watcher. + """ + ... def __call__(self, *args: Any, **kwargs: Any) -> Optional[ActiveTrace]: ... class DefaultContextProvider(BaseContextProvider): diff --git a/ddtrace/internal/opentelemetry/thread_context.py b/ddtrace/internal/opentelemetry/thread_context.py index 0a4af7c5431..0fd6c9e3537 100644 --- a/ddtrace/internal/opentelemetry/thread_context.py +++ b/ddtrace/internal/opentelemetry/thread_context.py @@ -4,10 +4,12 @@ from typing import Protocol from typing import Union +from ddtrace._trace.context import SAMPLING_DECISION_EVENT from ddtrace._trace.context import Context from ddtrace._trace.provider import BaseContextProvider from ddtrace._trace.span import Span from ddtrace.internal import core +from ddtrace.internal.compat import NumericType from ddtrace.internal.settings._config import config @@ -16,36 +18,53 @@ class TracerProtocol(Protocol): def context_provider(self) -> BaseContextProvider: ... -_ContextActivationListener = Callable[[BaseContextProvider, Optional[Union[Context, Span]]], None] -_ContextSwitchListener = Callable[[], None] -_ThreadContextListeners = tuple[_ContextActivationListener, _ContextSwitchListener] +_ActiveTrace = Optional[Union[Context, Span]] +_ContextActivationListener = Callable[[BaseContextProvider, _ActiveTrace], None] +_ResyncListener = Callable[[], None] +_ThreadContextListeners = tuple[_ContextActivationListener, _ResyncListener] + + +def _w3c_trace_flags(sampling_priority: Optional[NumericType]) -> int: + return 1 if sampling_priority is not None and sampling_priority > 0 else 0 if sys.platform == "linux": from ddtrace.internal.native._native import detach_otel_thread_context from ddtrace.internal.native._native import update_otel_thread_context + from ddtrace.internal.native._native import update_otel_thread_context_ids def register_otel_thread_context_listener(tracer: TracerProtocol) -> Optional[_ThreadContextListeners]: if not config._otel_thread_context_enabled: return None - def _sync_otel_thread_context(ctx: Optional[Union[Context, Span]]) -> None: + def _sync_otel_thread_context(ctx: _ActiveTrace) -> None: if type(ctx) is Span: - sampling_priority = ctx._local_root.context.sampling_priority - trace_flags = 1 if sampling_priority is not None and sampling_priority > 0 else 0 - update_otel_thread_context(ctx, ctx._local_root_value, trace_flags) + update_otel_thread_context( + ctx, ctx._local_root_value, _w3c_trace_flags(ctx._local_root.context.sampling_priority) + ) + elif type(ctx) is Context and ctx.trace_id is not None and ctx.span_id is not None: + # A Context is a span this execution runs inside but does not own -- a + # remote parent, or the submitter of work handed to this thread. The + # execution is still attributable to it. Its local root is not knowable + # from here, so the span stands in for it, as it does for a root span. + update_otel_thread_context_ids( + ctx.trace_id, ctx.span_id, _w3c_trace_flags(ctx.sampling_priority), ctx.span_id + ) else: detach_otel_thread_context() def _sync_active_otel_thread_context() -> None: - _sync_otel_thread_context(tracer.context_provider.active()) + _sync_otel_thread_context(tracer.context_provider._peek_active()) - def _on_context_provider_activate(provider: BaseContextProvider, ctx: Optional[Union[Context, Span]]) -> None: + def _on_context_provider_activate(provider: BaseContextProvider, ctx: _ActiveTrace) -> None: if provider is tracer.context_provider: _sync_otel_thread_context(ctx) core.on("ddtrace.context_provider.activate", _on_context_provider_activate) core.on("python.context.switch", _sync_active_otel_thread_context) + # Sampling usually decides at trace-chunk finish, after the record was published + # with trace_flags=0. + core.on(SAMPLING_DECISION_EVENT, _sync_active_otel_thread_context) if sys.implementation.name == "cpython" and sys.version_info >= (3, 14): from ddtrace.internal.native._native import register_context_watcher diff --git a/ddtrace/llmobs/_context.py b/ddtrace/llmobs/_context.py index 89b9f368241..9d59d84a5b6 100644 --- a/ddtrace/llmobs/_context.py +++ b/ddtrace/llmobs/_context.py @@ -58,3 +58,9 @@ def active(self) -> ContextTypeValue: if isinstance(item, Span): return self._update_active(item) return item + + def _peek_active(self) -> ContextTypeValue: + # Required because this provider has its own storage: the inherited + # implementation reads the default contextvar. Not read-only, unlike the + # inherited one -- nothing needs that of this provider yet. + return self.active() diff --git a/releasenotes/notes/fix-otel-thread-context-correctness-0fb65fa39ac88d25.yaml b/releasenotes/notes/fix-otel-thread-context-correctness-0fb65fa39ac88d25.yaml new file mode 100644 index 00000000000..e15077dc207 --- /dev/null +++ b/releasenotes/notes/fix-otel-thread-context-correctness-0fb65fa39ac88d25.yaml @@ -0,0 +1,16 @@ +--- +fixes: + - | + tracing: Fixes an issue where the OpenTelemetry thread context reported no active + trace whenever a trace was propagated without a local span, so compatible profiling + and security tools could not correlate samples taken in executor worker threads, or + on a distributed request before its first span started. + - | + tracing: Fixes an issue where the OpenTelemetry thread context kept reporting a trace + as unsampled for the whole life of the trace, because trace sampling usually decides + after the record is published. The record is now updated when the sampling decision + is made. + - | + tracing: Fixes an issue where publishing the OpenTelemetry thread context on an + asynchronous context switch could change which span is active in that context, and + publish the record twice for a single switch. diff --git a/src/native/context_provider.rs b/src/native/context_provider.rs index 006ff059e20..e47e3bc2850 100644 --- a/src/native/context_provider.rs +++ b/src/native/context_provider.rs @@ -77,6 +77,57 @@ fn dispatch_activate( event_hub::dispatch(py, ACTIVATE_EVENT, Some(args.into_any().unbind()), false) } +enum Resolved<'py> { + Unchanged, + /// May hold Python `None`, when the whole ancestor chain is finished. + Ancestor(Bound<'py, PyAny>), + ReactivatableContext(Bound<'py, PyAny>), +} + +/// Where the active trace *should* be, walking past finished ancestors of `span`. +/// +/// Shared by `_update_active`, which applies the result, and `_peek_active`, which +/// only reports it, so the two cannot drift apart. Borrows `span` so that +/// `Resolved::Unchanged` costs the caller nothing -- `_update_active` runs on every +/// span finish, where an extra incref shows up. +#[inline] +fn resolve_active<'py>(py: Python<'py>, span: &Bound<'py, PyAny>) -> PyResult> { + let mut current = span.clone(); + loop { + // PERF: read `duration`, `_parent`, and `_parent_context` straight off the + // native SpanData fields in one borrow -- avoids three Python attribute + // lookups per ancestor hop. + let (parent, parent_context) = { + let Ok(sd) = current.cast::() else { + break; // not a Span (e.g. None) -- stop walking parents + }; + let sd = sd.borrow(); + if sd.duration.is_none() { + break; // unfinished span -- stop walking parents + } + ( + sd._parent.as_ref().map(|p| p.bind(py).clone()), + sd._parent_context.as_ref().map(|c| c.bind(py).clone()), + ) + }; + if parent.is_none() { + if let Some(parent_context) = parent_context { + // `_reactivate` lives on the pure-Python Context -- still a getattr. + if parent_context.getattr("_reactivate")?.is_truthy()? { + return Ok(Resolved::ReactivatableContext(parent_context)); + } + } + } + // Advance to the parent; `None` ends the walk on the next iteration. + current = parent.unwrap_or_else(|| py.None().into_bound(py)); + } + if current.is(span) { + Ok(Resolved::Unchanged) + } else { + Ok(Resolved::Ancestor(current)) + } +} + /// `Some(v)` unless `v` is Python `None`, matching the `Optional[...]` return /// convention used throughout this module. Consumes `v` to avoid an incref. #[inline] @@ -169,6 +220,12 @@ impl BaseContextProvider { Err(PyNotImplementedError::new_err(())) } + /// See `DefaultContextProvider::_peek_active`. A provider with its own storage + /// keeps today's behavior until it opts into a read-only path. + fn _peek_active(slf: &Bound<'_, Self>) -> PyResult>> { + slf.call_method0("active").map(none_or_unbind) + } + /// Method available for backward-compatibility. It proxies the call to /// ``self.active()`` and must not do anything more. #[pyo3(signature = (*_args, **_kwargs))] @@ -251,41 +308,43 @@ impl DefaultContextProvider { py: Python<'py>, span: Bound<'py, PyAny>, ) -> PyResult>> { - let original = span.clone(); - let mut current = span; - loop { - // PERF: read `duration`, `_parent`, and `_parent_context` straight - // off the native SpanData fields in one borrow -- avoids three - // Python attribute lookups per ancestor hop. - let (parent, parent_context) = { - let Ok(sd) = current.cast::() else { - break; // not a Span (e.g. None) -- stop walking parents - }; - let sd = sd.borrow(); - if sd.duration.is_none() { - break; // unfinished span -- stop walking parents - } - ( - sd._parent.as_ref().map(|p| p.bind(py).clone()), - sd._parent_context.as_ref().map(|c| c.bind(py).clone()), - ) - }; - if parent.is_none() { - if let Some(parent_context) = parent_context { - // `_reactivate` lives on the pure-Python Context -- still a getattr. - if parent_context.getattr("_reactivate")?.is_truthy()? { - call_activate(slf, py, Some(parent_context.clone()))?; - return Ok(Some(parent_context.unbind())); - } - } + match resolve_active(py, &span)? { + Resolved::Unchanged => Ok(none_or_unbind(span)), + Resolved::ReactivatableContext(parent_context) => { + call_activate(slf, py, Some(parent_context.clone()))?; + Ok(Some(parent_context.unbind())) } - // Advance to the parent; `None` ends the walk on the next iteration. - current = parent.unwrap_or_else(|| py.None().into_bound(py)); + Resolved::Ancestor(current) => { + call_activate(slf, py, none_or_clone(¤t))?; + Ok(none_or_unbind(current)) + } + } + } + + /// `active` without its repair: no `activate`, so no contextvar write and no + /// `ddtrace.context_provider.activate` dispatch. + /// + /// For observers that must not perturb the active trace. A CPython + /// context-switch watcher is one: `activate` there would write to whichever + /// context the switch just made current, and re-enter the event hub mid-switch. + /// + /// Reads this contextvar and resolves the way `_update_active` does, so **a subclass + /// that overrides `active` or `_update_active` must override this too** -- otherwise + /// it reports from storage it does not use. Subclasses that only add behaviour + /// elsewhere inherit it safely. + fn _peek_active<'py>(_slf: &Bound<'py, Self>, py: Python<'py>) -> PyResult>> { + let item = contextvar_get(py, contextvar(py)?)?; + if item.is_none() { + return Ok(None); + } + if item.cast::().is_err() { + return Ok(Some(item.unbind())); } - if !current.is(&original) { - call_activate(slf, py, none_or_clone(¤t))?; + match resolve_active(py, &item)? { + Resolved::Unchanged => Ok(none_or_unbind(item)), + Resolved::ReactivatableContext(parent_context) => Ok(Some(parent_context.unbind())), + Resolved::Ancestor(current) => Ok(none_or_unbind(current)), } - Ok(none_or_unbind(current)) } } diff --git a/src/native/lib.rs b/src/native/lib.rs index 9b16a4d07ae..5945d2a758e 100644 --- a/src/native/lib.rs +++ b/src/native/lib.rs @@ -71,6 +71,9 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_wrapped(wrap_pyfunction!( otel_thread_ctx::update_otel_thread_context ))?; + m.add_wrapped(wrap_pyfunction!( + otel_thread_ctx::update_otel_thread_context_ids + ))?; m.add_wrapped(wrap_pyfunction!( otel_thread_ctx::detach_otel_thread_context ))?; diff --git a/src/native/otel_thread_ctx.rs b/src/native/otel_thread_ctx.rs index f1a8c2da426..c552b3712cd 100644 --- a/src/native/otel_thread_ctx.rs +++ b/src/native/otel_thread_ctx.rs @@ -23,6 +23,23 @@ pub fn update_otel_thread_context( ); } +/// Publish from raw ids, for an active `Context` -- there is no `SpanData` to read. +#[pyfunction] +pub fn update_otel_thread_context_ids( + trace_id: u128, + span_id: u64, + trace_flags: u8, + local_root_span_id: u64, +) { + ThreadContext::update( + trace_id.to_be_bytes(), + span_id.to_be_bytes(), + trace_flags, + local_root_span_id.to_be_bytes(), + &[], + ); +} + #[pyfunction] pub fn detach_otel_thread_context() { ThreadContext::detach(); diff --git a/tests/tracer/test_otel_thread_context.py b/tests/tracer/test_otel_thread_context.py index 4a80560e46a..b3ea46d7592 100644 --- a/tests/tracer/test_otel_thread_context.py +++ b/tests/tracer/test_otel_thread_context.py @@ -7,6 +7,8 @@ import pytest +from ddtrace._trace.context import SAMPLING_DECISION_EVENT +from ddtrace._trace.context import Context as DDContext from ddtrace._trace.provider import DefaultContextProvider from ddtrace._trace.tracer import Tracer from ddtrace.internal import core @@ -20,18 +22,23 @@ from ddtrace.internal.native import _native from ddtrace.internal.native._native import detach_otel_thread_context + # 28-byte header, then packed (key_index, len, value) attribute entries. + _ROOT_SPAN_KEY_INDEX = 0 + class _ThreadContextRecord(ctypes.Structure): _fields_ = [ ("trace_id", ctypes.c_ubyte * 16), ("span_id", ctypes.c_ubyte * 8), ("valid", ctypes.c_ubyte), ("trace_flags", ctypes.c_ubyte), + ("attrs_data_size", ctypes.c_uint16), + ("attrs_data", ctypes.c_ubyte * 612), ] _NATIVE_LIBRARY = ctypes.CDLL(_native.__file__) -def _published_span_id(): +def _record(): slot = ctypes.c_void_p.in_dll(_NATIVE_LIBRARY, "otel_thread_ctx_v1") if slot.value is None: return None @@ -39,28 +46,56 @@ def _published_span_id(): record = _ThreadContextRecord.from_address(slot.value) if not record.valid: return None + return record + + +def _published_span_id(): + record = _record() + if record is None: + return None return int.from_bytes(record.span_id, byteorder="big") def _published_trace_flags(): - slot = ctypes.c_void_p.in_dll(_NATIVE_LIBRARY, "otel_thread_ctx_v1") - if slot.value is None: + record = _record() + if record is None: return None + return record.trace_flags - record = _ThreadContextRecord.from_address(slot.value) - if not record.valid: + +def _published_trace_id(): + record = _record() + if record is None: return None - return record.trace_flags + return int.from_bytes(record.trace_id, byteorder="big") + + +def _published_local_root_span_id(): + record = _record() + if record is None: + return None + blob = bytes(record.attrs_data[: record.attrs_data_size]) + offset = 0 + while offset + 2 <= len(blob): + key, length = blob[offset], blob[offset + 1] + value = blob[offset + 2 : offset + 2 + length] + if len(value) < length: + break + if key == _ROOT_SPAN_KEY_INDEX: + return int.from_bytes(value, byteorder="big") + offset += 2 + length + return None @pytest.fixture(autouse=True) def _register_otel_thread_context_listener(tracer): listeners = register_otel_thread_context_listener(tracer) assert listeners is not None - activation_listener, context_switch_listener = listeners + activation_listener, resync_listener = listeners yield core.reset_listeners("ddtrace.context_provider.activate", activation_listener) - core.reset_listeners("python.context.switch", context_switch_listener) + core.reset_listeners("python.context.switch", resync_listener) + core.reset_listeners(SAMPLING_DECISION_EVENT, resync_listener) def test_span_context_is_published_and_detached(tracer: Tracer): @@ -81,6 +116,80 @@ def test_span_context_publishes_trace_flags(tracer: Tracer): assert _published_trace_flags() == 0 +def test_span_publishes_local_root_span_id(tracer: Tracer): + with tracer.trace("root") as root: + assert _published_local_root_span_id() == root.span_id + with tracer.trace("child") as child: + assert _published_span_id() == child.span_id + assert _published_local_root_span_id() == root.span_id + + +def test_active_context_is_published(tracer: Tracer): + ctx = DDContext(trace_id=12345, span_id=678) + with tracer._activate_context(ctx): + assert _published_trace_id() == 12345 + assert _published_span_id() == 678 + # A Context's local root is not knowable, so the span stands in for it. + assert _published_local_root_span_id() == 678 + + assert _published_span_id() is None + + +def test_offloaded_work_publishes_the_submitting_span(tracer: Tracer): + """A worker thread is attributable to the span that handed it the work. + + This is the handoff futures/threading.py performs on submit. + """ + published = {} + + with tracer.trace("submitter") as submitter: + handoff = submitter.context.copy(submitter.trace_id, submitter.span_id) + + def worker(): + with tracer._activate_context(handoff): + published["trace_id"] = _published_trace_id() + published["span_id"] = _published_span_id() + + thread = threading.Thread(target=worker) + thread.start() + thread.join() + + assert published["trace_id"] == submitter.trace_id + assert published["span_id"] == submitter.span_id + + +def test_context_switch_does_not_repair_the_contextvar(tracer: Tracer): + from ddtrace._trace.provider import _DD_CONTEXTVAR + + parent = DDContext(trace_id=1, span_id=2) + parent._reactivate = True + span = tracer.start_span("child", child_of=parent) + span.finish() + tracer.context_provider.activate(span) + + core.dispatch("python.context.switch") + + assert _published_span_id() == 2 + assert _DD_CONTEXTVAR.get() is span + + +def test_sampling_decision_republishes_trace_flags(tracer: Tracer): + """Nothing re-activates the span here: the decision alone has to republish.""" + with tracer.trace("test") as span: + assert _published_trace_flags() == 0 + + span.context.sampling_priority = 2 + assert _published_trace_flags() == 1 + + span.context.sampling_priority = -1 + assert _published_trace_flags() == 0 + + span.context.sampling_priority = 1 + assert _published_trace_flags() == 1 + span.context.sampling_priority = None + assert _published_trace_flags() == 0 + + def test_span_context_is_thread_local(tracer: Tracer): barrier = threading.Barrier(2) @@ -110,12 +219,14 @@ def test_thread_context_listeners_can_be_disabled(): assert "ddtrace" not in sys.modules + from ddtrace._trace.context import SAMPLING_DECISION_EVENT from ddtrace.internal import core from ddtrace.internal.settings._config import config assert config._otel_thread_context_enabled is False assert core.has_listeners("ddtrace.context_provider.activate") is False assert core.has_listeners("python.context.switch") is False + assert core.has_listeners(SAMPLING_DECISION_EVENT) is False if sys.implementation.name == "cpython" and sys.version_info >= (3, 14): from ddtrace.internal.native._native import is_context_watcher_registered diff --git a/tests/tracer/test_tracer.py b/tests/tracer/test_tracer.py index c708da3b740..d878e91958a 100644 --- a/tests/tracer/test_tracer.py +++ b/tests/tracer/test_tracer.py @@ -30,6 +30,7 @@ from ddtrace.constants import VERSION_KEY from ddtrace.contrib.internal.trace_utils import set_user from ddtrace.ext import user +from ddtrace.internal import core from ddtrace.internal.settings._config import Config from ddtrace.internal.writer import AgentWriterInterface from ddtrace.trace import Context @@ -2161,3 +2162,94 @@ def test_activate_context_nesting_and_restoration(tracer): assert active.span_id == 1 assert tracer.context_provider.active() is None + + +def _finished_span_over_reactivatable_context(tracer): + """The state `active()` repairs by activating the parent context. + + `_on_span_finish` already repairs the finishing execution's contextvar, so + reaching this state means activating the finished span explicitly -- the position + an async task holding a span that finished elsewhere is in. + """ + parent = Context(trace_id=1, span_id=2) + parent._reactivate = True + span = tracer.start_span("child", child_of=parent) + span.finish() + tracer.context_provider.activate(span) + return parent, span + + +def test_peek_active_resolves_without_activating(tracer): + from ddtrace._trace.provider import _DD_CONTEXTVAR + + parent, span = _finished_span_over_reactivatable_context(tracer) + + activations = [] + + def record(provider, ctx): + activations.append(ctx) + + core.on("ddtrace.context_provider.activate", record) + try: + peeked = tracer.context_provider._peek_active() + finally: + # Pass the callback: a bare reset would drop every other listener too. + core.reset_listeners("ddtrace.context_provider.activate", record) + + assert peeked is parent + assert _DD_CONTEXTVAR.get() is span + assert activations == [] + + +def test_active_still_repairs_the_contextvar(tracer): + from ddtrace._trace.provider import _DD_CONTEXTVAR + + parent, _ = _finished_span_over_reactivatable_context(tracer) + + assert tracer.context_provider.active() is parent + assert _DD_CONTEXTVAR.get() is parent + + +def test_peek_active_is_inherited_by_behaviour_preserving_subclasses(tracer): + """A subclass that does not override active()/_update_active() uses the same storage, + so it must inherit the read-only peek rather than falling back to active(). + """ + from ddtrace._trace.provider import DefaultContextProvider + + class Subclass(DefaultContextProvider): + def __enter__(self): + pass + + def __exit__(self, *exc): + pass + + provider = Subclass() + parent = Context(trace_id=1, span_id=2) + parent._reactivate = True + span = tracer.start_span("child", child_of=parent) + span.finish() + provider.activate(span) + + activations = [] + + def record(prov, ctx): + activations.append(ctx) + + core.on("ddtrace.context_provider.activate", record) + try: + assert provider._peek_active() is parent + finally: + core.reset_listeners("ddtrace.context_provider.activate", record) + + assert activations == [] + + +def test_peek_active_matches_active_for_simple_states(tracer): + assert tracer.context_provider._peek_active() is None + + ctx = Context(trace_id=1, span_id=1) + with tracer._activate_context(ctx): + assert tracer.context_provider._peek_active() is ctx + + with tracer.trace("root") as span: + assert tracer.context_provider._peek_active() is span