diff --git a/opentelemetry-api/src/opentelemetry/trace/__init__.py b/opentelemetry-api/src/opentelemetry/trace/__init__.py index 996576c3ee0..45001310414 100644 --- a/opentelemetry-api/src/opentelemetry/trace/__init__.py +++ b/opentelemetry-api/src/opentelemetry/trace/__init__.py @@ -73,6 +73,7 @@ from opentelemetry import context as context_api from opentelemetry.attributes import BoundedAttributes +from opentelemetry.context import get_current from opentelemetry.context.context import Context from opentelemetry.environment_variables import OTEL_PYTHON_TRACER_PROVIDER from opentelemetry.trace.propagation import ( @@ -402,6 +403,31 @@ def function(): The newly-created span. """ + @_agnosticcontextmanager + @abstractmethod + def apply_egress_continuation( + self, + kind: SpanKind, + context: Context | None = None, + attributes: types.Attributes = None, + ) -> Iterator[Context]: + """Context manager for getting a Context informed with continuation hints + for egress propagation based on span kind and attributes. + + Example:: + + with tracer.apply_egress_continuation(kind=Span.CLIENT, attributes=attributes) as injection_context: + propagate.inject(carrier, context=injection_context, setter=setter) + + Args: + context: An optional Context. Defaults to the global context. + kind: The span's kind. + attributes: The span's attributes. + + Yields: + A context that contains egress continuation hints for propagators + """ + class ProxyTracer(Tracer): # pylint: disable=W0222,signature-differs @@ -442,6 +468,18 @@ def start_as_current_span(self, *args, **kwargs) -> Iterator[Span]: with self._tracer.start_as_current_span(*args, **kwargs) as span: # type: ignore yield span + @_agnosticcontextmanager + def apply_egress_continuation( + self, + kind: SpanKind, + context: Context | None = None, + attributes: types.Attributes = None, + ) -> Iterator[Context]: + with self._tracer.apply_egress_continuation( + kind, context, attributes + ) as injection_context: + yield injection_context + class NoOpTracer(Tracer): """The default Tracer, used when no Tracer implementation is available. @@ -507,6 +545,16 @@ def start_as_current_span( ) as span: yield span + @_agnosticcontextmanager + def apply_egress_continuation( + self, + kind: SpanKind, + context: Context | None = None, + attributes: types.Attributes = None, + ) -> Iterator[Context]: + injected_context = context if context else get_current() + yield injected_context + @deprecated("You should use NoOpTracer. Deprecated since version 1.9.0.") class _DefaultTracer(NoOpTracer): diff --git a/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py b/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py index 97c64d93737..ec647a9b205 100644 --- a/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py +++ b/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py @@ -3,12 +3,44 @@ # import re +from opentelemetry import context as context_api from opentelemetry import trace from opentelemetry.context.context import Context from opentelemetry.propagators import textmap from opentelemetry.trace import format_span_id, format_trace_id from opentelemetry.trace.span import TraceState +_SUPPRESS_TRACE_CONTEXT_INJECTION_KEY = context_api.create_key( + "suppress-trace-context-injection" +) + + +def suppress_trace_context_injection( + context: Context | None = None, +) -> Context: + """Returns a context that suppresses W3C Trace Context injection.""" + return context_api.set_value( + _SUPPRESS_TRACE_CONTEXT_INJECTION_KEY, True, context + ) + + +def enable_trace_context_injection( + context: Context | None = None, +) -> Context: + """Returns a context that allows W3C Trace Context injection.""" + return context_api.set_value( + _SUPPRESS_TRACE_CONTEXT_INJECTION_KEY, False, context + ) + + +def is_trace_context_injection_suppressed( + context: Context | None = None, +) -> bool: + """Returns whether W3C Trace Context injection is suppressed.""" + return bool( + context_api.get_value(_SUPPRESS_TRACE_CONTEXT_INJECTION_KEY, context) + ) + class TraceContextTextMapPropagator(textmap.TextMapPropagator): """Extracts and injects using w3c TraceContext's headers.""" @@ -84,6 +116,9 @@ def inject( See `opentelemetry.propagators.textmap.TextMapPropagator.inject` """ + if is_trace_context_injection_suppressed(context): + return + span = trace.get_current_span(context) span_context = span.get_span_context() if span_context == trace.INVALID_SPAN_CONTEXT: diff --git a/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py b/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py index 87d410c5bee..3eec31c52ed 100644 --- a/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py +++ b/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py @@ -170,6 +170,34 @@ def test_propagate_invalid_context(self): FORMAT.inject(output, context=ctx) self.assertFalse("traceparent" in output) + def test_suppress_trace_context_injection(self): + """Do not propagate trace context when injection is suppressed.""" + output: dict[str, str] = {} + span = trace.NonRecordingSpan( + trace.SpanContext(self.TRACE_ID, self.SPAN_ID, is_remote=False) + ) + ctx = trace.set_span_in_context(span) + ctx = tracecontext.suppress_trace_context_injection(ctx) + + FORMAT.inject(output, context=ctx) + + self.assertFalse("traceparent" in output) + self.assertFalse("tracestate" in output) + + def test_enable_trace_context_injection(self): + """Propagate trace context after inherited suppression is cleared.""" + output: dict[str, str] = {} + span = trace.NonRecordingSpan( + trace.SpanContext(self.TRACE_ID, self.SPAN_ID, is_remote=False) + ) + ctx = trace.set_span_in_context(span) + ctx = tracecontext.suppress_trace_context_injection(ctx) + ctx = tracecontext.enable_trace_context_injection(ctx) + + FORMAT.inject(output, context=ctx) + + self.assertTrue("traceparent" in output) + def test_tracestate_empty_header(self): """Test tracestate with an additional empty header (should be ignored)""" span = trace.get_current_span( diff --git a/opentelemetry-api/tests/trace/test_tracer.py b/opentelemetry-api/tests/trace/test_tracer.py index 68bec0647de..c8095782e3e 100644 --- a/opentelemetry-api/tests/trace/test_tracer.py +++ b/opentelemetry-api/tests/trace/test_tracer.py @@ -5,14 +5,20 @@ import asyncio from unittest import TestCase +from opentelemetry.context import get_current from opentelemetry.trace import ( INVALID_SPAN, + NonRecordingSpan, NoOpTracer, Span, + SpanContext, + SpanKind, Tracer, _agnosticcontextmanager, get_current_span, + set_span_in_context, ) +from opentelemetry.trace.propagation import tracecontext class TestTracer(TestCase): @@ -27,6 +33,45 @@ def test_start_as_current_span_context_manager(self): with self.tracer.start_as_current_span("") as span: self.assertIsInstance(span, Span) + def test_apply_egress_continuation_noop_tracer_preserves_context(self): + propagator = tracecontext.TraceContextTextMapPropagator() + span_context = SpanContext( + trace_id=0x000000000000000000000000DEADBEEF, + span_id=0x00000000DEADBEF0, + is_remote=False, + ) + context = set_span_in_context(NonRecordingSpan(span_context)) + + with self.tracer.apply_egress_continuation( + kind=SpanKind.CLIENT, + context=context, + attributes={"server.address": "api.example.com"}, + ) as injection_context: + carrier: dict[str, str] = {} + propagator.inject(carrier, context=injection_context) + + self.assertIn("traceparent", carrier) + + def test_apply_egress_continuation_noop_tracer_preserves_suppression(self): + propagator = tracecontext.TraceContextTextMapPropagator() + span_context = SpanContext( + trace_id=0x000000000000000000000000DEADBEEF, + span_id=0x00000000DEADBEF0, + is_remote=False, + ) + context = set_span_in_context(NonRecordingSpan(span_context)) + context = tracecontext.suppress_trace_context_injection(context) + + with self.tracer.apply_egress_continuation( + kind=SpanKind.CLIENT, + context=context, + attributes={"server.address": "api.example.com"}, + ) as injection_context: + carrier: dict[str, str] = {} + propagator.inject(carrier, context=injection_context) + + self.assertNotIn("traceparent", carrier) + def test_start_as_current_span_decorator(self): # using a list to track the mock call order calls = [] @@ -41,6 +86,10 @@ def start_as_current_span(self, *args, **kwargs): # type: ignore yield INVALID_SPAN calls.append(9) + @_agnosticcontextmanager # pylint: disable=protected-access + def apply_egress_continuation(self, *args, **kwargs): # type: ignore + yield get_current() + mock_tracer = MockTracer() # test 1 : sync function diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index d292fed8e55..5fbed1907f7 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -51,7 +51,7 @@ parse_boolean_environment_variable, ) from opentelemetry.sdk.resources import Resource -from opentelemetry.sdk.trace import sampling +from opentelemetry.sdk.trace import sampling, trace_continuation from opentelemetry.sdk.trace._tracer_metrics import create_tracer_metrics from opentelemetry.sdk.trace.id_generator import IdGenerator, RandomIdGenerator from opentelemetry.sdk.util import BoundedList @@ -67,6 +67,7 @@ EXCEPTION_TYPE, ) from opentelemetry.trace import NoOpTracer, SpanContext +from opentelemetry.trace.propagation import tracecontext from opentelemetry.trace.status import Status, StatusCode from opentelemetry.util import types from opentelemetry.util._decorator import _agnosticcontextmanager @@ -1120,6 +1121,8 @@ def __init__( *, meter_provider: metrics_api.MeterProvider | None = None, _tracer_config: _TracerConfig | None = None, + _continuation_decider: trace_continuation.TraceContinuationDecider + | None, ) -> None: self.sampler = sampler self.resource = resource @@ -1129,6 +1132,9 @@ def __init__( self._span_limits = span_limits self._instrumentation_scope = instrumentation_scope self._tracer_config = _tracer_config or _TracerConfig.default() + self._continuation_decider = ( + _continuation_decider or trace_continuation.ALWAYS_CONTINUE + ) meter_provider = meter_provider or metrics_api.get_meter_provider() self._tracer_metrics = create_tracer_metrics( @@ -1176,6 +1182,32 @@ def start_as_current_span( ) as span: yield span + @_agnosticcontextmanager # pylint: disable=protected-access + def apply_egress_continuation( + self, + kind: trace_api.SpanKind, + context: context_api.Context | None = None, + attributes: types.Attributes = None, + ) -> Iterator[context_api.Context]: + injection_context = context or context_api.get_current() + egress_action = self._continuation_decider.should_inject( + context=injection_context, + kind=kind, + attributes=attributes, + ) + if ( + egress_action + is trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ): + injection_context = tracecontext.suppress_trace_context_injection( + injection_context + ) + else: + injection_context = tracecontext.enable_trace_context_injection( + injection_context + ) + yield injection_context + def start_span( # pylint: disable=too-many-locals self, name: str, @@ -1187,7 +1219,7 @@ def start_span( # pylint: disable=too-many-locals record_exception: bool = True, set_status_on_exception: bool = True, ) -> trace_api.Span: - links = links or () + links = tuple(links) if links else () parent_span_context = trace_api.get_current_span( context ).get_span_context() @@ -1202,10 +1234,41 @@ def start_span( # pylint: disable=too-many-locals if not self._is_enabled(): return trace_api.NonRecordingSpan(context=parent_span_context) + # the continuation result decides if we should restart the trace + # or not, consider only remote parent spans + if ( + parent_span_context is not None + and parent_span_context.is_valid + and parent_span_context.is_remote + ): + continuation_result = self._continuation_decider.should_continue( + parent_context=context, + parent_span_context=parent_span_context, + kind=kind, + attributes=attributes, + links=links, + ) + else: + continuation_result = trace_continuation.ContinuationResult( + decision=trace_continuation.Decision.CONTINUE, + links=links, + ) + # is_valid determines root span - if parent_span_context is None or not parent_span_context.is_valid: + if ( + parent_span_context is None + or not parent_span_context.is_valid + or continuation_result.should_restart + ): parent_span_context = None trace_id = self.id_generator.generate_trace_id() + + # if we need to restart remove the previous span from the context + # so that the samplers do not see it + if continuation_result.should_restart: + context = trace_api.set_span_in_context( + trace_api.INVALID_SPAN, context + ) else: trace_id = parent_span_context.trace_id @@ -1216,7 +1279,12 @@ def start_span( # pylint: disable=too-many-locals # to include information about the sampling result. # The sampler may also modify the parent span context's tracestate sampling_result = self.sampler.should_sample( - context, trace_id, name, kind, attributes, links + context, + trace_id, + name, + kind, + attributes, + continuation_result.links, ) trace_flags = ( @@ -1259,7 +1327,7 @@ def start_span( # pylint: disable=too-many-locals attributes=sampling_result.attributes, span_processor=self.span_processor, kind=kind, - links=links, + links=continuation_result.links, instrumentation_info=self.instrumentation_info, record_exception=record_exception, set_status_on_exception=set_status_on_exception, @@ -1316,6 +1384,8 @@ def __init__( *, meter_provider: metrics_api.MeterProvider | None = None, _tracer_configurator: _TracerConfiguratorT | None = None, + _continuation_decider: trace_continuation.TraceContinuationDecider + | None = None, ) -> None: self._active_span_processor = ( active_span_processor or SynchronousMultiSpanProcessor() @@ -1336,6 +1406,7 @@ def __init__( self._disabled = disabled.lower().strip() == "true" self._atexit_handler = None self._meter_provider = meter_provider + self._continuation_decider = _continuation_decider if shutdown_on_exit: self._atexit_handler = atexit.register(self.shutdown) @@ -1433,6 +1504,7 @@ def get_tracer( instrumentation_scope, meter_provider=self._meter_provider, _tracer_config=tracer_config, + _continuation_decider=self._continuation_decider, ) self._tracers[instrumentation_scope] = tracer diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py new file mode 100644 index 00000000000..5c45801945d --- /dev/null +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -0,0 +1,303 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import abc +import enum +from collections.abc import Mapping, Sequence +from dataclasses import dataclass +from fnmatch import fnmatchcase +from logging import getLogger + +from opentelemetry.context import Context +from opentelemetry.trace import Link, SpanKind +from opentelemetry.trace.span import SpanContext +from opentelemetry.util.types import AnyValue, Attributes + +_logger = getLogger(__name__) + + +class ContinuationDirection(enum.Enum): + INGRESS = 0 + EGRESS = 1 + + +class Decision(enum.Enum): + CONTINUE = 0 + RESTART_WITH_LINK = 1 + RESTART_WITHOUT_LINK = 2 + + +class EgressAction(enum.Enum): + INJECT_TRACE_CONTEXT = 0 + SUPPRESS_TRACE_CONTEXT = 1 + + +class ContinuationResult: + def __repr__(self) -> str: + return f"{type(self).__name__}({str(self.decision)}, links={str(self.links)})" + + def __init__( + self, + *, + decision: Decision, + links: tuple[Link, ...] = (), + ) -> None: + self.decision = decision + self.links = links + + @property + def should_restart(self): + return self.decision in ( + Decision.RESTART_WITH_LINK, + Decision.RESTART_WITHOUT_LINK, + ) + + +class TraceContinuationDecider(abc.ABC): + @abc.abstractmethod + def should_continue( + self, + *, + parent_context: Context | None, + parent_span_context: SpanContext | None, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: tuple[Link, ...] | None = None, + ) -> ContinuationResult: + pass + + @abc.abstractmethod + def get_description(self) -> str: + pass + + @abc.abstractmethod + def should_inject( + self, + *, + context: Context | None = None, + kind: SpanKind | None = None, + attributes: Attributes = None, + ) -> EgressAction: + pass + + +class StaticTraceContinuationDecider(TraceContinuationDecider): + """Continuation decider that always returns the same decision.""" + + def __init__(self, decision: Decision) -> None: + self._decision = decision + + def should_continue( + self, + *, + parent_context: Context | None, + parent_span_context: SpanContext | None, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: tuple[Link, ...] | None = None, + ) -> ContinuationResult: + result_links = links or () + return ContinuationResult( + decision=self._decision, + links=_maybe_add_restart_link( + decision=self._decision, + parent_span_context=parent_span_context, + links=result_links, + ), + ) + + def get_description(self) -> str: + if self._decision is Decision.RESTART_WITH_LINK: + return "AlwaysRestartWithLinkContinuationDecider" + if self._decision is Decision.RESTART_WITHOUT_LINK: + return "AlwaysRestartWithoutLinkContinuationDecider" + return "AlwaysContinueContinuationDecider" + + def should_inject( + self, + *, + context: Context | None = None, + kind: SpanKind | None = None, + attributes: Attributes = None, + ) -> EgressAction: + return _egress_action_from_decision(self._decision) + + +ALWAYS_CONTINUE = StaticTraceContinuationDecider(Decision.CONTINUE) +ALWAYS_RESTART_WITH_LINK = StaticTraceContinuationDecider( + Decision.RESTART_WITH_LINK +) +ALWAYS_RESTART_WITHOUT_LINK = StaticTraceContinuationDecider( + Decision.RESTART_WITHOUT_LINK +) + + +@dataclass(frozen=True) +class TraceContinuationRule: + """A rule that selects a trace continuation decision when all conditions match.""" + + direction: ContinuationDirection + strategy: Decision | None = None + egress_action: EgressAction | None = None + attributes: Mapping[str, AnyValue] | None = None + span_kind: SpanKind | None = None + link_attributes: Attributes = None + + def matches( + self, + *, + direction: ContinuationDirection | None = None, + span_kind: SpanKind | None = None, + attributes: Attributes = None, + ) -> bool: + if direction != self.direction: + return False + if self.span_kind is not None and span_kind != self.span_kind: + return False + return _attributes_match(attributes, self.attributes) + + +class RuleBasedTraceContinuationDecider(TraceContinuationDecider): + """Trace continuation decider with ordered first-match-wins rules.""" + + def __init__( + self, + *, + rules: Sequence[TraceContinuationRule], + default_strategy: Decision = Decision.CONTINUE, + default_egress_action: EgressAction = EgressAction.INJECT_TRACE_CONTEXT, + ) -> None: + self._rules = tuple(rules) + self._default_strategy = default_strategy + self._default_egress_action = default_egress_action + + def should_continue( + self, + *, + parent_context: Context | None, + parent_span_context: SpanContext | None, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: tuple[Link, ...] | None = None, + ) -> ContinuationResult: + result_links = links or () + for rule in self._rules: + if rule.matches( + direction=ContinuationDirection.INGRESS, + span_kind=kind, + attributes=attributes, + ): + strategy = _ingress_strategy_from_rule(rule) + return ContinuationResult( + decision=strategy, + links=_maybe_add_restart_link( + decision=strategy, + parent_span_context=parent_span_context, + links=result_links, + link_attributes=rule.link_attributes, + ), + ) + + return ContinuationResult( + decision=self._default_strategy, + links=_maybe_add_restart_link( + decision=self._default_strategy, + parent_span_context=parent_span_context, + links=result_links, + ), + ) + + def get_description(self) -> str: + return "RuleBasedTraceContinuationDecider" + + def should_inject( + self, + *, + context: Context | None = None, + kind: SpanKind | None = None, + attributes: Attributes = None, + ) -> EgressAction: + for rule in self._rules: + if rule.matches( + direction=ContinuationDirection.EGRESS, + span_kind=kind, + attributes=attributes, + ): + return _egress_action_from_rule(rule) + + return self._default_egress_action + + +def _maybe_add_restart_link( + *, + decision: Decision, + parent_span_context: SpanContext | None, + links: tuple[Link, ...], + link_attributes: Attributes = None, +) -> tuple[Link, ...]: + if decision is Decision.RESTART_WITH_LINK and parent_span_context: + return links + (Link(parent_span_context, link_attributes),) + return links + + +def _egress_action_from_decision(decision: Decision) -> EgressAction: + if decision is Decision.CONTINUE: + return EgressAction.INJECT_TRACE_CONTEXT + return EgressAction.SUPPRESS_TRACE_CONTEXT + + +def _ingress_strategy_from_rule(rule: TraceContinuationRule) -> Decision: + if rule.strategy is None: + raise ValueError( + "Ingress trace continuation rules must define strategy" + ) + return rule.strategy + + +def _egress_action_from_rule(rule: TraceContinuationRule) -> EgressAction: + if rule.egress_action is not None: + return rule.egress_action + if rule.strategy is not None: + return _egress_action_from_decision(rule.strategy) + raise ValueError( + "Egress trace continuation rules must define egress_action or strategy" + ) + + +def _attributes_match( + attributes: Attributes, + expected_attributes: Mapping[str, AnyValue] | None, +) -> bool: + if not expected_attributes: + return True + if not attributes: + return False + return all( + key in attributes and _attribute_matches(attributes[key], expected) + for key, expected in expected_attributes.items() + ) + + +def _attribute_matches(value: AnyValue, expected: AnyValue) -> bool: + return any( + _single_attribute_value_matches(actual, expected) + for actual in _attribute_values(value) + ) + + +def _single_attribute_value_matches( + value: AnyValue, expected: AnyValue +) -> bool: + if isinstance(expected, str) and isinstance(value, str): + return fnmatchcase(value, expected) + return value == expected + + +def _attribute_values(value: AnyValue): + if isinstance(value, Sequence) and not isinstance( + value, (str, bytes, bytearray) + ): + return value + return (value,) diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index 0b07908ec5b..f0151fbeb4f 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -38,6 +38,7 @@ TracerProvider, _RuleBasedTracerConfigurator, _TracerConfig, + trace_continuation, ) from opentelemetry.sdk.trace.id_generator import RandomIdGenerator from opentelemetry.sdk.trace.sampling import ( @@ -62,6 +63,20 @@ get_tracer, set_tracer_provider, ) +from opentelemetry.trace.propagation import tracecontext + + +def _remote_parent_context(): + remote_parent = trace_api.SpanContext( + trace_id=0x000000000000000000000000DEADBEEF, + span_id=0x00000000DEADBEF0, + is_remote=True, + trace_flags=trace_api.TraceFlags(trace_api.TraceFlags.SAMPLED), + ) + context = trace_api.set_span_in_context( + trace_api.NonRecordingSpan(remote_parent) + ) + return remote_parent, context class TestTracer(unittest.TestCase): @@ -323,7 +338,7 @@ def verify_default_sampler(self, tracer_provider): self.assertEqual(tracer_provider.sampler._root, ALWAYS_ON) -class TestSpanCreation(unittest.TestCase): +class TestSpanCreation(unittest.TestCase): # pylint: disable=too-many-public-methods def test_start_span_invalid_spancontext(self): """If an invalid span context is passed as the parent, the created span should use a new span id. @@ -546,6 +561,568 @@ def test_start_span_preserves_parent_random_trace_id_flag(self): child_trace_flags.random_trace_id, ) + def test_trace_continuation_continue_uses_remote_parent(self): + remote_parent, context = _remote_parent_context() + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_CONTINUE + ).get_tracer(__name__) + + child = tracer.start_span("child", context) + + self.assertEqual(child.parent, remote_parent) + self.assertEqual( + child.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(child.links, ()) + + def test_trace_continuation_restart_without_link_creates_root(self): + remote_parent, context = _remote_parent_context() + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITHOUT_LINK + ).get_tracer(__name__) + + root = tracer.start_span("root", context) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(root.links, ()) + + def test_trace_continuation_restart_with_link_creates_root_with_link(self): + remote_parent, context = _remote_parent_context() + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITH_LINK + ).get_tracer(__name__) + + root = tracer.start_span("root", context) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(len(root.links), 1) + self.assertEqual(root.links[0].context, remote_parent) + + def test_trace_continuation_client_span_with_remote_parent_can_restart( + self, + ): + remote_parent, context = _remote_parent_context() + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITH_LINK + ).get_tracer(__name__) + + root = tracer.start_span( + "client", context, kind=trace_api.SpanKind.CLIENT + ) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(len(root.links), 1) + self.assertEqual(root.links[0].context, remote_parent) + + def test_trace_continuation_restart_with_link_preserves_explicit_links( + self, + ): + remote_parent, context = _remote_parent_context() + explicit_link_context = trace_api.SpanContext( + trace_id=0x000000000000000000000000FEEDBEEF, + span_id=0x00000000FEEDBEF0, + is_remote=False, + ) + explicit_link = trace_api.Link(explicit_link_context) + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITH_LINK + ).get_tracer(__name__) + + root = tracer.start_span("root", context, links=(explicit_link,)) + + self.assertEqual(len(root.links), 2) + self.assertEqual(root.links[0].context, explicit_link_context) + self.assertEqual(root.links[1].context, remote_parent) + + def test_trace_continuation_restart_sampler_sees_root_context(self): + _, context = _remote_parent_context() + sampler = ParentBased( + root=ALWAYS_OFF, + remote_parent_sampled=ALWAYS_ON, + remote_parent_not_sampled=ALWAYS_OFF, + local_parent_sampled=ALWAYS_OFF, + local_parent_not_sampled=ALWAYS_OFF, + ) + tracer = TracerProvider( + sampler=sampler, + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITHOUT_LINK, + ).get_tracer(__name__) + + span = tracer.start_span("root", context) + + self.assertIsInstance(span, trace_api.NonRecordingSpan) + + def test_trace_continuation_restart_updates_processor_parent_context(self): + _, context = _remote_parent_context() + span_processor = mock.Mock(spec=trace.SpanProcessor) + tracer_provider = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITHOUT_LINK + ) + tracer_provider.add_span_processor(span_processor) + tracer = tracer_provider.get_tracer(__name__) + + root = tracer.start_span("root", context) + + span_processor.on_start.assert_called_once() + parent_context = span_processor.on_start.call_args.kwargs[ + "parent_context" + ] + self.assertEqual( + trace_api.get_current_span(parent_context), + trace_api.INVALID_SPAN, + ) + root.end() + + def test_trace_continuation_restart_link_not_inherited_by_child(self): + remote_parent, context = _remote_parent_context() + tracer = TracerProvider( + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITH_LINK, + ).get_tracer(__name__) + + with tracer.start_as_current_span("root", context) as root: + child = tracer.start_span("child") + child.end() + + self.assertIsNone(root.parent) + self.assertEqual(len(root.links), 1) + self.assertEqual(root.links[0].context, remote_parent) + self.assertEqual(child.parent, root.get_span_context()) + self.assertEqual(child.links, ()) + + def test_trace_continuation_rule_based_uses_first_matching_rule(self): + remote_parent, context = _remote_parent_context() + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + direction=trace_continuation.ContinuationDirection.INGRESS, + strategy=trace_continuation.Decision.RESTART_WITH_LINK, + attributes={"http.route": "/webhooks/*"}, + ), + trace_continuation.TraceContinuationRule( + direction=trace_continuation.ContinuationDirection.INGRESS, + strategy=trace_continuation.Decision.CONTINUE, + attributes={"http.route": "/webhooks/partner"}, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + + root = tracer.start_span( + "root", context, attributes={"http.route": "/webhooks/partner"} + ) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(len(root.links), 1) + self.assertEqual(root.links[0].context, remote_parent) + + def test_trace_continuation_rule_based_uses_default_strategy(self): + remote_parent, context = _remote_parent_context() + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + direction=trace_continuation.ContinuationDirection.INGRESS, + strategy=trace_continuation.Decision.CONTINUE, + attributes={"http.route": "/internal/*"}, + ), + ), + default_strategy=trace_continuation.Decision.RESTART_WITHOUT_LINK, + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + + root = tracer.start_span( + "root", context, attributes={"http.route": "/webhooks/partner"} + ) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(root.links, ()) + + def test_trace_continuation_rule_based_matches_all_conditions(self): + remote_parent, context = _remote_parent_context() + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + strategy=trace_continuation.Decision.CONTINUE, + attributes={"http.route": "/internal/*"}, + direction=trace_continuation.ContinuationDirection.INGRESS, + span_kind=trace_api.SpanKind.SERVER, + ), + ), + default_strategy=trace_continuation.Decision.RESTART_WITHOUT_LINK, + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + + child = tracer.start_span( + "child", + context, + kind=trace_api.SpanKind.SERVER, + attributes={"http.route": "/internal/users"}, + ) + + self.assertEqual(child.parent, remote_parent) + self.assertEqual( + child.get_span_context().trace_id, remote_parent.trace_id + ) + self.assertEqual(child.links, ()) + + def test_trace_continuation_rule_based_is_direction_aware(self): + remote_parent, context = _remote_parent_context() + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + strategy=trace_continuation.Decision.RESTART_WITHOUT_LINK, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + + child = tracer.start_span("child", context) + + self.assertEqual(child.parent, remote_parent) + self.assertEqual( + child.get_span_context().trace_id, remote_parent.trace_id + ) + + def test_trace_continuation_rule_based_adds_link_attributes(self): + remote_parent, context = _remote_parent_context() + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + direction=trace_continuation.ContinuationDirection.INGRESS, + strategy=trace_continuation.Decision.RESTART_WITH_LINK, + attributes={"http.route": "/webhooks/*"}, + link_attributes={ + "otel.trace_continuation.reason": "external_webhook" + }, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + + root = tracer.start_span( + "root", context, attributes={"http.route": "/webhooks/partner"} + ) + + self.assertEqual(len(root.links), 1) + self.assertEqual(root.links[0].context, remote_parent) + self.assertEqual( + root.links[0].attributes, + {"otel.trace_continuation.reason": "external_webhook"}, + ) + + def test_trace_continuation_egress_continue_preserves_injection(self): + tracer = TracerProvider().get_tracer(__name__) + propagator = tracecontext.TraceContextTextMapPropagator() + + with tracer.start_as_current_span( + "client", + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "internal.example.com"}, + ) as span: + carrier = {} + propagator.inject(carrier) + + self.assertIn("traceparent", carrier) + self.assertIn( + format(span.get_span_context().trace_id, "032x"), + carrier["traceparent"], + ) + + def test_trace_continuation_egress_restart_suppresses_trace_context(self): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ), + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + + with tracer.start_as_current_span( + "client", + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ): + carrier = {} + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) as injection_context: + propagator.inject(carrier, context=injection_context) + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + + def test_trace_continuation_egress_with_start_span_suppresses_trace_context( + self, + ): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ), + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + span = tracer.start_span( + "client", + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) + context = trace_api.set_span_in_context(span) + + try: + carrier = {} + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.CLIENT, + context=context, + attributes={"server.address": "api.third-party.example"}, + ) as injection_context: + propagator.inject(carrier, context=injection_context) + finally: + span.end() + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + + def test_trace_continuation_egress_with_separate_activation_suppresses_trace_context( + self, + ): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ), + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + span = tracer.start_span( + "client", + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) + + with trace_api.use_span(span, end_on_exit=True): + carrier = {} + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) as injection_context: + propagator.inject(carrier, context=injection_context) + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + + def test_trace_continuation_producer_egress_suppresses_trace_context( + self, + ): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ), + attributes={"messaging.destination.name": "jobs"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + span_kind=trace_api.SpanKind.PRODUCER, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + + with tracer.start_as_current_span( + "publish", + kind=trace_api.SpanKind.PRODUCER, + attributes={"messaging.destination.name": "jobs"}, + ): + carrier = {} + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.PRODUCER, + attributes={"messaging.destination.name": "jobs"}, + ) as injection_context: + propagator.inject(carrier, context=injection_context) + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + + def test_trace_continuation_nested_egress_inject_overrides_suppression( + self, + ): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ), + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + trace_continuation.TraceContinuationRule( + egress_action=( + trace_continuation.EgressAction.INJECT_TRACE_CONTEXT + ), + attributes={"server.address": "internal.example.com"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + + with tracer.start_as_current_span( + "client", + kind=trace_api.SpanKind.CLIENT, + ) as span: + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) as suppressed_context: + suppressed_carrier = {} + propagator.inject( + suppressed_carrier, context=suppressed_context + ) + + injected_carrier = {} + with tracer.apply_egress_continuation( + kind=trace_api.SpanKind.CLIENT, + context=suppressed_context, + attributes={"server.address": "internal.example.com"}, + ) as injection_context: + propagator.inject( + injected_carrier, context=injection_context + ) + + self.assertNotIn("traceparent", suppressed_carrier) + self.assertIn("traceparent", injected_carrier) + self.assertIn( + format(span.get_span_context().trace_id, "032x"), + injected_carrier["traceparent"], + ) + + def test_trace_continuation_egress_returns_propagation_action(self): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + strategy=trace_continuation.Decision.RESTART_WITH_LINK, + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + + action = decider.should_inject( + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) + + self.assertIs( + action, + trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT, + ) + + def test_trace_continuation_egress_strategy_maps_to_propagation_action( + self, + ): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + strategy=trace_continuation.Decision.CONTINUE, + attributes={"server.address": "internal.example.com"}, + direction=trace_continuation.ContinuationDirection.EGRESS, + ), + ) + ) + + action = decider.should_inject( + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "internal.example.com"}, + ) + + self.assertIs( + action, + trace_continuation.EgressAction.INJECT_TRACE_CONTEXT, + ) + + def test_trace_continuation_egress_rule_is_direction_aware(self): + decider = trace_continuation.RuleBasedTraceContinuationDecider( + rules=( + trace_continuation.TraceContinuationRule( + strategy=trace_continuation.Decision.RESTART_WITHOUT_LINK, + attributes={"server.address": "api.third-party.example"}, + direction=trace_continuation.ContinuationDirection.INGRESS, + ), + ) + ) + tracer = TracerProvider(_continuation_decider=decider).get_tracer( + __name__ + ) + propagator = tracecontext.TraceContextTextMapPropagator() + + with tracer.start_as_current_span( + "client", + kind=trace_api.SpanKind.CLIENT, + attributes={"server.address": "api.third-party.example"}, + ) as span: + carrier = {} + propagator.inject(carrier) + + self.assertIn("traceparent", carrier) + self.assertIn( + format(span.get_span_context().trace_id, "032x"), + carrier["traceparent"], + ) + def test_start_as_current_span_implicit(self): tracer = new_tracer()