From 0dee5733108e1c0ce1bfe15963eca6f6f8ee9236 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Fri, 3 Jul 2026 17:24:43 +0200 Subject: [PATCH 01/11] First stab at reworked prototype for trace continuation strategy --- .../src/opentelemetry/sdk/trace/__init__.py | 44 ++++++- .../sdk/trace/trace_continuation.py | 110 ++++++++++++++++++ 2 files changed, 149 insertions(+), 5 deletions(-) create mode 100644 opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index d292fed8e55..9bc7e509be8 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 @@ -1120,6 +1120,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 +1131,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( @@ -1187,7 +1192,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 +1207,30 @@ 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 + continuation_result = self._continuation_decider.should_continue( + parent_context=context, + parent_span_context=parent_span_context, + direction=trace_continuation.ContinuationDirection.INGRESS, + kind=kind, + attributes=attributes, + 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) else: trace_id = parent_span_context.trace_id @@ -1216,7 +1241,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 +1289,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 +1346,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 +1368,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 +1466,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..3f7fe73bd3a --- /dev/null +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -0,0 +1,110 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import abc +import enum +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 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 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, + direction: ContinuationDirection | None = None, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: tuple[Link, ...] | None = None, + ) -> ContinuationResult: + pass + + @abc.abstractmethod + def get_description(self) -> str: + 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, + direction: ContinuationDirection | None = None, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: tuple[Link, ...] | None = None, + ) -> ContinuationResult: + result_links = links or () + if ( + self._decision is Decision.RESTART_WITH_LINK + and parent_span_context + ): + result_links += (Link(parent_span_context),) + return ContinuationResult( + decision=self._decision, + links=result_links, + ) + + def get_description(self) -> str: + if self._decision is Decision.RESTART_WITH_LINK: + return "AlwaysRestartWithLinkContinuationDecider" + elif self._decision is Decision.RESTART_WITHOUT_LINK: + return "AlwaysRestartWithoutLinkContinuationDecider" + return "AlwaysContinueContinuationDecider" + + +ALWAYS_CONTINUE = StaticTraceContinuationDecider(Decision.CONTINUE) +ALWAYS_RESTART_WITH_LINK = StaticTraceContinuationDecider( + Decision.RESTART_WITH_LINK +) +ALWAYS_RESTART_WITHOUT_LINK = StaticTraceContinuationDecider( + Decision.RESTART_WITHOUT_LINK +) From b853ae66ad2b168ed5c47f8d55148ce2df9565bd Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Mon, 6 Jul 2026 15:58:51 +0200 Subject: [PATCH 02/11] Add tests, please lint --- .../sdk/trace/trace_continuation.py | 2 +- opentelemetry-sdk/tests/trace/test_trace.py | 163 +++++++++++++++++- 2 files changed, 163 insertions(+), 2 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py index 3f7fe73bd3a..b77c77c42b8 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -96,7 +96,7 @@ def should_continue( def get_description(self) -> str: if self._decision is Decision.RESTART_WITH_LINK: return "AlwaysRestartWithLinkContinuationDecider" - elif self._decision is Decision.RESTART_WITHOUT_LINK: + if self._decision is Decision.RESTART_WITHOUT_LINK: return "AlwaysRestartWithoutLinkContinuationDecider" return "AlwaysContinueContinuationDecider" diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index 0b07908ec5b..2d52686a925 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,48 @@ get_tracer, set_tracer_provider, ) +from opentelemetry.util.types import Attributes + + +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 _RemoteOnlyRestartWithLinkDecider( + trace_continuation.TraceContinuationDecider +): + def should_continue( + self, + *, + parent_context: Context | None, + parent_span_context: trace_api.SpanContext | None, + direction: trace_continuation.ContinuationDirection | None = None, + kind: trace_api.SpanKind | None = None, + attributes: Attributes = None, + links: tuple[trace_api.Link, ...] | None = None, + ) -> trace_continuation.ContinuationResult: + result_links = links or () + if parent_span_context is not None and parent_span_context.is_remote: + return trace_continuation.ContinuationResult( + decision=trace_continuation.Decision.RESTART_WITH_LINK, + links=result_links + (trace_api.Link(parent_span_context),), + ) + return trace_continuation.ContinuationResult( + decision=trace_continuation.Decision.CONTINUE, + links=result_links, + ) + + def get_description(self) -> str: + return "RemoteOnlyRestartWithLinkDecider" class TestTracer(unittest.TestCase): @@ -323,7 +366,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 +589,124 @@ 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_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=_RemoteOnlyRestartWithLinkDecider() + ).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_start_as_current_span_implicit(self): tracer = new_tracer() From fd4000bf33c8aa48c759a430745f573430f43a94 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Mon, 6 Jul 2026 17:33:20 +0200 Subject: [PATCH 03/11] Add rule based continuation decider --- .../sdk/trace/trace_continuation.py | 141 +++++++++++++++++- opentelemetry-sdk/tests/trace/test_trace.py | 133 +++++++++++++++++ 2 files changed, 267 insertions(+), 7 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py index b77c77c42b8..bbba9bb115d 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -5,12 +5,15 @@ 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 Attributes +from opentelemetry.util.types import AnyValue, Attributes _logger = getLogger(__name__) @@ -83,14 +86,13 @@ def should_continue( links: tuple[Link, ...] | None = None, ) -> ContinuationResult: result_links = links or () - if ( - self._decision is Decision.RESTART_WITH_LINK - and parent_span_context - ): - result_links += (Link(parent_span_context),) return ContinuationResult( decision=self._decision, - links=result_links, + links=_maybe_add_restart_link( + decision=self._decision, + parent_span_context=parent_span_context, + links=result_links, + ), ) def get_description(self) -> str: @@ -108,3 +110,128 @@ def get_description(self) -> str: 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.""" + + strategy: Decision + attributes: Mapping[str, AnyValue] | None = None + direction: ContinuationDirection | 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 self.direction is not None and 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, + ) -> None: + self._rules = tuple(rules) + self._default_strategy = default_strategy + + def should_continue( + self, + *, + parent_context: Context | None, + parent_span_context: SpanContext | None, + direction: ContinuationDirection | None = 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=direction, + span_kind=kind, + attributes=attributes, + ): + return ContinuationResult( + decision=rule.strategy, + links=_maybe_add_restart_link( + decision=rule.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 _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 _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 2d52686a925..d55ad1b3def 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -707,6 +707,139 @@ def test_trace_continuation_restart_link_not_inherited_by_child(self): 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( + strategy=trace_continuation.Decision.RESTART_WITH_LINK, + attributes={"http.route": "/webhooks/*"}, + ), + trace_continuation.TraceContinuationRule( + 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( + 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( + 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_start_as_current_span_implicit(self): tracer = new_tracer() From 44727671877b6a531f86e531e76c6672fb619864 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 10:59:34 +0200 Subject: [PATCH 04/11] Use the same context on trace restart --- opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 9bc7e509be8..890f22b1a4d 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -1230,7 +1230,9 @@ def start_span( # pylint: disable=too-many-locals # 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 = trace_api.set_span_in_context( + trace_api.INVALID_SPAN, context + ) else: trace_id = parent_span_context.trace_id From e9e713d4f5486485e1d5d13d0eccb433b742a283 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 11:17:17 +0200 Subject: [PATCH 05/11] Only apply trace continuation decider to remote spans --- .../src/opentelemetry/sdk/trace/__init__.py | 29 +++++++++++------ opentelemetry-sdk/tests/trace/test_trace.py | 31 +------------------ 2 files changed, 21 insertions(+), 39 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 890f22b1a4d..05a7195b703 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -1208,15 +1208,26 @@ def start_span( # pylint: disable=too-many-locals return trace_api.NonRecordingSpan(context=parent_span_context) # the continuation result decides if we should restart the trace - # or not - continuation_result = self._continuation_decider.should_continue( - parent_context=context, - parent_span_context=parent_span_context, - direction=trace_continuation.ContinuationDirection.INGRESS, - kind=kind, - attributes=attributes, - links=links, - ) + # 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, + direction=trace_continuation.ContinuationDirection.INGRESS, + 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 ( diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index d55ad1b3def..0330775a254 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -63,7 +63,6 @@ get_tracer, set_tracer_provider, ) -from opentelemetry.util.types import Attributes def _remote_parent_context(): @@ -79,34 +78,6 @@ def _remote_parent_context(): return remote_parent, context -class _RemoteOnlyRestartWithLinkDecider( - trace_continuation.TraceContinuationDecider -): - def should_continue( - self, - *, - parent_context: Context | None, - parent_span_context: trace_api.SpanContext | None, - direction: trace_continuation.ContinuationDirection | None = None, - kind: trace_api.SpanKind | None = None, - attributes: Attributes = None, - links: tuple[trace_api.Link, ...] | None = None, - ) -> trace_continuation.ContinuationResult: - result_links = links or () - if parent_span_context is not None and parent_span_context.is_remote: - return trace_continuation.ContinuationResult( - decision=trace_continuation.Decision.RESTART_WITH_LINK, - links=result_links + (trace_api.Link(parent_span_context),), - ) - return trace_continuation.ContinuationResult( - decision=trace_continuation.Decision.CONTINUE, - links=result_links, - ) - - def get_description(self) -> str: - return "RemoteOnlyRestartWithLinkDecider" - - class TestTracer(unittest.TestCase): def test_no_deprecated_warning(self): with self.assertRaises(AssertionError): @@ -694,7 +665,7 @@ def test_trace_continuation_restart_updates_processor_parent_context(self): def test_trace_continuation_restart_link_not_inherited_by_child(self): remote_parent, context = _remote_parent_context() tracer = TracerProvider( - _continuation_decider=_RemoteOnlyRestartWithLinkDecider() + _continuation_decider=trace_continuation.ALWAYS_RESTART_WITH_LINK, ).get_tracer(__name__) with tracer.start_as_current_span("root", context) as root: From 72bc6fb45afa6db8e1599f48aeb820b6d2c113b9 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 15:12:10 +0200 Subject: [PATCH 06/11] Implement egress path --- .../trace/propagation/tracecontext.py | 26 ++++ .../test_tracecontexthttptextformat.py | 14 ++ .../src/opentelemetry/sdk/trace/__init__.py | 36 +++-- .../sdk/trace/trace_continuation.py | 80 +++++++++- opentelemetry-sdk/tests/trace/test_trace.py | 139 ++++++++++++++++++ 5 files changed, 279 insertions(+), 16 deletions(-) diff --git a/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py b/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py index 97c64d93737..2abd5b3f012 100644 --- a/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py +++ b/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py @@ -3,12 +3,35 @@ # 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 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 +107,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..eb54634852e 100644 --- a/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py +++ b/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py @@ -170,6 +170,20 @@ 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_tracestate_empty_header(self): """Test tracestate with an additional empty header (should be ignored)""" span = trace.get_current_span( diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index 05a7195b703..c5f009d6360 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -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 @@ -1173,13 +1174,32 @@ def start_as_current_span( record_exception=record_exception, set_status_on_exception=set_status_on_exception, ) - with trace_api.use_span( - span, - end_on_exit=end_on_exit, - record_exception=record_exception, - set_status_on_exception=set_status_on_exception, - ) as span: - yield span + token = None + if kind is trace_api.SpanKind.CLIENT: + egress_action = self._continuation_decider.should_inject( + context=context, + kind=kind, + attributes=attributes, + ) + if ( + egress_action + is trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT + ): + egress_context = tracecontext.suppress_trace_context_injection( + context + ) + token = context_api.attach(egress_context) + try: + with trace_api.use_span( + span, + end_on_exit=end_on_exit, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + ) as span: + yield span + finally: + if token is not None: + context_api.detach(token) def start_span( # pylint: disable=too-many-locals self, @@ -1209,7 +1229,6 @@ def start_span( # pylint: disable=too-many-locals # 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 @@ -1218,7 +1237,6 @@ def start_span( # pylint: disable=too-many-locals continuation_result = self._continuation_decider.should_continue( parent_context=context, parent_span_context=parent_span_context, - direction=trace_continuation.ContinuationDirection.INGRESS, kind=kind, attributes=attributes, links=links, diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py index bbba9bb115d..5c370beac09 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -29,6 +29,11 @@ class Decision(enum.Enum): 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)})" @@ -57,7 +62,6 @@ def should_continue( *, parent_context: Context | None, parent_span_context: SpanContext | None, - direction: ContinuationDirection | None = None, kind: SpanKind | None = None, attributes: Attributes = None, links: tuple[Link, ...] | None = None, @@ -68,6 +72,16 @@ def should_continue( 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.""" @@ -80,7 +94,6 @@ def should_continue( *, parent_context: Context | None, parent_span_context: SpanContext | None, - direction: ContinuationDirection | None = None, kind: SpanKind | None = None, attributes: Attributes = None, links: tuple[Link, ...] | None = None, @@ -102,6 +115,15 @@ def get_description(self) -> str: 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( @@ -116,7 +138,8 @@ def get_description(self) -> str: class TraceContinuationRule: """A rule that selects a trace continuation decision when all conditions match.""" - strategy: Decision + strategy: Decision | None = None + egress_action: EgressAction | None = None attributes: Mapping[str, AnyValue] | None = None direction: ContinuationDirection | None = None span_kind: SpanKind | None = None @@ -144,16 +167,17 @@ def __init__( *, 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, - direction: ContinuationDirection | None = None, kind: SpanKind | None = None, attributes: Attributes = None, links: tuple[Link, ...] | None = None, @@ -161,14 +185,15 @@ def should_continue( result_links = links or () for rule in self._rules: if rule.matches( - direction=direction, + direction=ContinuationDirection.INGRESS, span_kind=kind, attributes=attributes, ): + strategy = _ingress_strategy_from_rule(rule) return ContinuationResult( - decision=rule.strategy, + decision=strategy, links=_maybe_add_restart_link( - decision=rule.strategy, + decision=strategy, parent_span_context=parent_span_context, links=result_links, link_attributes=rule.link_attributes, @@ -187,6 +212,23 @@ def should_continue( 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( *, @@ -200,6 +242,30 @@ def _maybe_add_restart_link( 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, diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index 0330775a254..936e86127df 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -63,6 +63,7 @@ get_tracer, set_tracer_provider, ) +from opentelemetry.trace.propagation import tracecontext def _remote_parent_context(): @@ -603,6 +604,25 @@ def test_trace_continuation_restart_with_link_creates_root_with_link(self): 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, ): @@ -811,6 +831,125 @@ def test_trace_continuation_rule_based_adds_link_attributes(self): {"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 = {} + propagator.inject(carrier) + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + + 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() From c4813f2dd86a6aea73af687f7d7ecaa50039495d Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 16:28:37 +0200 Subject: [PATCH 07/11] Consider also PRODUCER span kinds in egress --- .../src/opentelemetry/sdk/trace/__init__.py | 2 +- opentelemetry-sdk/tests/trace/test_trace.py | 31 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py index c5f009d6360..f3bd3899906 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -1175,7 +1175,7 @@ def start_as_current_span( set_status_on_exception=set_status_on_exception, ) token = None - if kind is trace_api.SpanKind.CLIENT: + if kind in (trace_api.SpanKind.CLIENT, trace_api.SpanKind.PRODUCER): egress_action = self._continuation_decider.should_inject( context=context, kind=kind, diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index 936e86127df..cf4d9930b6b 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -877,6 +877,37 @@ def test_trace_continuation_egress_restart_suppresses_trace_context(self): 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 = {} + propagator.inject(carrier) + + self.assertNotIn("traceparent", carrier) + self.assertNotIn("tracestate", carrier) + def test_trace_continuation_egress_returns_propagation_action(self): decider = trace_continuation.RuleBasedTraceContinuationDecider( rules=( From dfffda1550043328c2b4a4c95ebdb790cf563bde Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 16:40:56 +0200 Subject: [PATCH 08/11] Make rules direction mandatory --- .../src/opentelemetry/sdk/trace/trace_continuation.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py index 5c370beac09..5c45801945d 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/trace_continuation.py @@ -138,10 +138,10 @@ def should_inject( 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 - direction: ContinuationDirection | None = None span_kind: SpanKind | None = None link_attributes: Attributes = None @@ -152,7 +152,7 @@ def matches( span_kind: SpanKind | None = None, attributes: Attributes = None, ) -> bool: - if self.direction is not None and direction != self.direction: + if direction != self.direction: return False if self.span_kind is not None and span_kind != self.span_kind: return False From 5148dd2cc4d3daa91732ef5b38e15f98c68e94fb Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 7 Jul 2026 16:42:16 +0200 Subject: [PATCH 09/11] Ooops --- opentelemetry-sdk/tests/trace/test_trace.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index cf4d9930b6b..1b20f0cfa85 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -703,10 +703,12 @@ def test_trace_continuation_rule_based_uses_first_matching_rule(self): 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"}, ), @@ -732,6 +734,7 @@ def test_trace_continuation_rule_based_uses_default_strategy(self): decider = trace_continuation.RuleBasedTraceContinuationDecider( rules=( trace_continuation.TraceContinuationRule( + direction=trace_continuation.ContinuationDirection.INGRESS, strategy=trace_continuation.Decision.CONTINUE, attributes={"http.route": "/internal/*"}, ), @@ -808,6 +811,7 @@ def test_trace_continuation_rule_based_adds_link_attributes(self): 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={ From e6131289f0ae4385fa703b314c87a7251a3bf493 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Thu, 9 Jul 2026 12:30:26 +0200 Subject: [PATCH 10/11] Rework egress path, move to tracer method instead of being Python specific --- .../src/opentelemetry/trace/__init__.py | 48 ++++++++ .../trace/propagation/tracecontext.py | 9 ++ .../test_tracecontexthttptextformat.py | 14 +++ opentelemetry-api/tests/trace/test_tracer.py | 5 + .../src/opentelemetry/sdk/trace/__init__.py | 57 ++++++---- opentelemetry-sdk/tests/trace/test_trace.py | 107 +++++++++++++++++- 6 files changed, 213 insertions(+), 27 deletions(-) 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 2abd5b3f012..ec647a9b205 100644 --- a/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py +++ b/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py @@ -24,6 +24,15 @@ def suppress_trace_context_injection( ) +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: diff --git a/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py b/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py index eb54634852e..3eec31c52ed 100644 --- a/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py +++ b/opentelemetry-api/tests/trace/propagation/test_tracecontexthttptextformat.py @@ -184,6 +184,20 @@ def test_suppress_trace_context_injection(self): 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..c8a91ba20dd 100644 --- a/opentelemetry-api/tests/trace/test_tracer.py +++ b/opentelemetry-api/tests/trace/test_tracer.py @@ -5,6 +5,7 @@ import asyncio from unittest import TestCase +from opentelemetry.context import get_current from opentelemetry.trace import ( INVALID_SPAN, NoOpTracer, @@ -41,6 +42,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 f3bd3899906..5fbed1907f7 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py @@ -1174,32 +1174,39 @@ def start_as_current_span( record_exception=record_exception, set_status_on_exception=set_status_on_exception, ) - token = None - if kind in (trace_api.SpanKind.CLIENT, trace_api.SpanKind.PRODUCER): - egress_action = self._continuation_decider.should_inject( - context=context, - kind=kind, - attributes=attributes, + with trace_api.use_span( + span, + end_on_exit=end_on_exit, + record_exception=record_exception, + set_status_on_exception=set_status_on_exception, + ) 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 ) - if ( - egress_action - is trace_continuation.EgressAction.SUPPRESS_TRACE_CONTEXT - ): - egress_context = tracecontext.suppress_trace_context_injection( - context - ) - token = context_api.attach(egress_context) - try: - with trace_api.use_span( - span, - end_on_exit=end_on_exit, - record_exception=record_exception, - set_status_on_exception=set_status_on_exception, - ) as span: - yield span - finally: - if token is not None: - context_api.detach(token) + else: + injection_context = tracecontext.enable_trace_context_injection( + injection_context + ) + yield injection_context def start_span( # pylint: disable=too-many-locals self, diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index 1b20f0cfa85..f3bd4a6e084 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -876,7 +876,50 @@ def test_trace_continuation_egress_restart_suppresses_trace_context(self): attributes={"server.address": "api.third-party.example"}, ): carrier = {} - propagator.inject(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) @@ -907,11 +950,71 @@ def test_trace_continuation_producer_egress_suppresses_trace_context( attributes={"messaging.destination.name": "jobs"}, ): carrier = {} - propagator.inject(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=( From bd9ebd8ba8a46f6a622baf1480fffd2d31017ef6 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Thu, 9 Jul 2026 12:38:55 +0200 Subject: [PATCH 11/11] Add tests requested in review by Tyler. --- opentelemetry-api/tests/trace/test_tracer.py | 44 ++++++++++++++++++++ opentelemetry-sdk/tests/trace/test_trace.py | 35 ++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/opentelemetry-api/tests/trace/test_tracer.py b/opentelemetry-api/tests/trace/test_tracer.py index c8a91ba20dd..c8095782e3e 100644 --- a/opentelemetry-api/tests/trace/test_tracer.py +++ b/opentelemetry-api/tests/trace/test_tracer.py @@ -8,12 +8,17 @@ 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): @@ -28,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 = [] diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index f3bd4a6e084..f0151fbeb4f 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/opentelemetry-sdk/tests/trace/test_trace.py @@ -924,6 +924,41 @@ def test_trace_continuation_egress_with_start_span_suppresses_trace_context( 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, ):