From 0dee5733108e1c0ce1bfe15963eca6f6f8ee9236 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Fri, 3 Jul 2026 17:24:43 +0200 Subject: [PATCH 01/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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, ): From 4a7135109c9ed1d45d06659e89265ddeded44d73 Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 21 Jul 2026 14:51:40 +0200 Subject: [PATCH 12/13] Update schema with OTEP changes and generate models --- .../src/opentelemetry/configuration/models.py | 111 +++++++- .../opentelemetry/configuration/schema.json | 265 +++++++++++++++++- 2 files changed, 373 insertions(+), 3 deletions(-) diff --git a/opentelemetry-configuration/src/opentelemetry/configuration/models.py b/opentelemetry-configuration/src/opentelemetry/configuration/models.py index 4a0b9521dac..e0367875feb 100644 --- a/opentelemetry-configuration/src/opentelemetry/configuration/models.py +++ b/opentelemetry-configuration/src/opentelemetry/configuration/models.py @@ -1,6 +1,6 @@ # generated by datamodel-codegen: # filename: schema.json -# timestamp: 2026-06-23T15:17:37+00:00 +# timestamp: 2026-07-21T12:46:47+00:00 from __future__ import annotations @@ -88,6 +88,21 @@ class ExemplarFilter(Enum): trace_based = "trace_based" +ExperimentalAlwaysContinueTraceContinuationDecider: TypeAlias = ( + dict[str, Any] | None +) + + +ExperimentalAlwaysRestartWithLinkTraceContinuationDecider: TypeAlias = ( + dict[str, Any] | None +) + + +ExperimentalAlwaysRestartWithoutLinkTraceContinuationDecider: TypeAlias = ( + dict[str, Any] | None +) + + ExperimentalComposableAlwaysOffSampler: TypeAlias = dict[str, Any] | None @@ -177,6 +192,35 @@ class ExperimentalPrometheusTranslationStrategy(Enum): no_translation_development = "no_translation/development" +@dataclass +class ExperimentalRuleBasedTraceContinuationDeciderAttributePatterns: + key: str + included: list[str] | None = None + excluded: list[str] | None = None + + +@dataclass +class ExperimentalRuleBasedTraceContinuationDeciderAttributeValues: + key: str + values: list[str] + + +@dataclass +class ExperimentalRuleBasedTraceContinuationDeciderConditions: + """ + Match conditions for an ExperimentalRuleBasedTraceContinuationDeciderRule. When multiple conditions are specified, all must match. + """ + + attribute_values: ( + list[ExperimentalRuleBasedTraceContinuationDeciderAttributeValues] + | None + ) = None + attribute_patterns: ( + list[ExperimentalRuleBasedTraceContinuationDeciderAttributePatterns] + | None + ) = None + + @dataclass class ExperimentalSemconvConfig: version: int | None = None @@ -193,6 +237,22 @@ class ExperimentalSpanParent(Enum): local = "local" +class ExperimentalTraceContinuationDirection(Enum): + ingress = "ingress" + egress = "egress" + + +class ExperimentalTraceContinuationEgressAction(Enum): + inject_trace_context = "inject_trace_context" + suppress_trace_context = "suppress_trace_context" + + +class ExperimentalTraceContinuationStrategy(Enum): + continue_ = "continue" + restart_with_link = "restart_with_link" + restart_without_link = "restart_without_link" + + @dataclass class ExperimentalTracerConfig: enabled: bool | None = None @@ -527,6 +587,23 @@ class ExperimentalRpcInstrumentation: semconv: ExperimentalSemconvConfig | None = None +@dataclass +class ExperimentalRuleBasedTraceContinuationDeciderRule: + """ + A rule for ExperimentalRuleBasedTraceContinuationDecider. Rules are matched in order, and the first matching rule determines the trace continuation result. + + """ + + direction: ExperimentalTraceContinuationDirection + span_kind: SpanKind | None = None + conditions: ( + ExperimentalRuleBasedTraceContinuationDeciderConditions | None + ) = None + strategy: ExperimentalTraceContinuationStrategy | None = None + egress_action: ExperimentalTraceContinuationEgressAction | None = None + link_attributes: list[AttributeNameValue] | None = None + + @dataclass class ExperimentalSanitization: url: ExperimentalUrlSanitization | None = None @@ -672,6 +749,35 @@ class ExperimentalResourceDetection: detectors: list[ExperimentalResourceDetector] | None = None +@dataclass +class ExperimentalRuleBasedTraceContinuationDecider: + rules: list[ExperimentalRuleBasedTraceContinuationDeciderRule] + default_ingress_strategy: ExperimentalTraceContinuationStrategy | None = ( + None + ) + default_egress_action: ExperimentalTraceContinuationEgressAction | None = ( + None + ) + + +@_additional_properties +@dataclass +class ExperimentalTraceContinuationDecider: + always_continue_development: ( + ExperimentalAlwaysContinueTraceContinuationDecider | None + ) = None + always_restart_with_link_development: ( + ExperimentalAlwaysRestartWithLinkTraceContinuationDecider | None + ) = None + always_restart_without_link_development: ( + ExperimentalAlwaysRestartWithoutLinkTraceContinuationDecider | None + ) = None + rule_based_development: ( + ExperimentalRuleBasedTraceContinuationDecider | None + ) = None + additional_properties: ClassVar[dict[str, Any]] + + @_additional_properties @dataclass class LogRecordProcessor: @@ -829,6 +935,9 @@ class TracerProvider: processors: list[SpanProcessor] limits: SpanLimits | None = None sampler: Sampler | None = None + trace_continuation_decider_development: ( + ExperimentalTraceContinuationDecider | None + ) = None id_generator: IdGenerator | None = None tracer_configurator_development: ExperimentalTracerConfigurator | None = ( None diff --git a/opentelemetry-configuration/src/opentelemetry/configuration/schema.json b/opentelemetry-configuration/src/opentelemetry/configuration/schema.json index 1a30982ecb6..d5a1dfca665 100644 --- a/opentelemetry-configuration/src/opentelemetry/configuration/schema.json +++ b/opentelemetry-configuration/src/opentelemetry/configuration/schema.json @@ -466,6 +466,27 @@ "trace_based" ] }, + "ExperimentalAlwaysContinueTraceContinuationDecider": { + "type": [ + "object", + "null" + ], + "additionalProperties": false + }, + "ExperimentalAlwaysRestartWithLinkTraceContinuationDecider": { + "type": [ + "object", + "null" + ], + "additionalProperties": false + }, + "ExperimentalAlwaysRestartWithoutLinkTraceContinuationDecider": { + "type": [ + "object", + "null" + ], + "additionalProperties": false + }, "ExperimentalCodeInstrumentation": { "type": "object", "additionalProperties": false, @@ -1226,6 +1247,182 @@ } } }, + "ExperimentalRuleBasedTraceContinuationDecider": { + "type": [ + "object", + "null" + ], + "additionalProperties": false, + "properties": { + "default_ingress_strategy": { + "$ref": "#/$defs/ExperimentalTraceContinuationStrategy", + "description": "Configure the ingress strategy used when no ingress rule matches.\nValues include:\n* continue: continue, use the candidate span context as the parent according to existing parentage rules.\n* restart_with_link: restart_with_link, start a new trace and add a link to the candidate span context.\n* restart_without_link: restart_without_link, start a new trace without preserving the candidate span context as a link.\nIf omitted, continue is used.\n" + }, + "default_egress_action": { + "$ref": "#/$defs/ExperimentalTraceContinuationEgressAction", + "description": "Configure the egress action used when no egress rule matches.\nValues include:\n* inject_trace_context: inject_trace_context, preserve existing trace context injection behavior.\n* suppress_trace_context: suppress_trace_context, do not inject the selected trace context for this outgoing boundary.\nIf omitted, inject_trace_context is used.\n" + }, + "rules": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/ExperimentalRuleBasedTraceContinuationDeciderRule" + }, + "description": "The rules for the trace continuation decider, matched in order.\nEach rule declares a direction and can have multiple match conditions. All conditions must match for the rule to match.\nIf no optional conditions are specified, the rule matches all inputs for its direction.\nIf no rule matches, default_ingress_strategy or default_egress_action is used for the input direction.\nProperty is required and must be non-null.\n" + } + }, + "required": [ + "rules" + ] + }, + "ExperimentalRuleBasedTraceContinuationDeciderAttributePatterns": { + "type": "object", + "additionalProperties": false, + "properties": { + "key": { + "type": "string", + "description": "The attribute key to match against.\nProperty is required and must be non-null.\n" + }, + "included": { + "type": "array", + "minItems": 1, + "items": { + "type": "string" + }, + "description": "Configure list of value patterns to include.\nMatching is case-sensitive. Values are evaluated to match as follows:\n * If the value exactly matches.\n * If the value matches the wildcard pattern, where '?' matches any single character and '*' matches any number of characters including none.\nIf omitted, all values are included.\n" + }, + "excluded": { + "type": "array", + "minItems": 1, + "items": { + "type": "string" + }, + "description": "Configure list of value patterns to exclude. Applies after .included (i.e. excluded has higher priority than included).\nMatching is case-sensitive. Values are evaluated to match as follows:\n * If the value exactly matches.\n * If the value matches the wildcard pattern, where '?' matches any single character and '*' matches any number of characters including none.\nIf omitted, .included attributes are included.\n" + } + }, + "required": [ + "key" + ] + }, + "ExperimentalRuleBasedTraceContinuationDeciderAttributeValues": { + "type": "object", + "additionalProperties": false, + "properties": { + "key": { + "type": "string", + "description": "The attribute key to match against.\nProperty is required and must be non-null.\n" + }, + "values": { + "type": "array", + "minItems": 1, + "items": { + "type": "string" + }, + "description": "The attribute values to match against. Non-string attributes are matched using their string representation.\nFor array attributes, if any item matches, it is considered a match.\nProperty is required and must be non-null.\n" + } + }, + "required": [ + "key", + "values" + ] + }, + "ExperimentalRuleBasedTraceContinuationDeciderConditions": { + "type": "object", + "description": "Match conditions for an ExperimentalRuleBasedTraceContinuationDeciderRule. When multiple conditions are specified, all must match.", + "additionalProperties": false, + "properties": { + "attribute_values": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/ExperimentalRuleBasedTraceContinuationDeciderAttributeValues" + }, + "description": "Attribute value conditions. All listed conditions must match.\nIf omitted, ignore.\n" + }, + "attribute_patterns": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/ExperimentalRuleBasedTraceContinuationDeciderAttributePatterns" + }, + "description": "Attribute pattern conditions. All listed conditions must match.\nIf omitted, ignore.\n" + } + } + }, + "ExperimentalRuleBasedTraceContinuationDeciderRule": { + "type": "object", + "description": "A rule for ExperimentalRuleBasedTraceContinuationDecider. Rules are matched in order, and the first matching rule determines the trace continuation result.\n", + "additionalProperties": false, + "properties": { + "direction": { + "$ref": "#/$defs/ExperimentalTraceContinuationDirection", + "description": "The input direction this rule matches.\nValues include:\n* egress: egress, a context selected for propagation across an outgoing boundary.\n* ingress: ingress, a context extracted for an incoming boundary.\nProperty is required and must be non-null.\n" + }, + "span_kind": { + "$ref": "#/$defs/SpanKind", + "description": "The span kind to match.\nValues include:\n* client: client, a client span.\n* consumer: consumer, a consumer span.\n* internal: internal, an internal span.\n* producer: producer, a producer span.\n* server: server, a server span.\nIf omitted, ignore.\n" + }, + "conditions": { + "$ref": "#/$defs/ExperimentalRuleBasedTraceContinuationDeciderConditions", + "description": "Additional match conditions for this rule.\nIf omitted, ignore.\n" + }, + "strategy": { + "$ref": "#/$defs/ExperimentalTraceContinuationStrategy", + "description": "The ingress trace continuation strategy to use when this rule matches.\nValues include:\n* continue: continue, use the candidate span context as the parent according to existing parentage rules.\n* restart_with_link: restart_with_link, start a new trace and add a link to the candidate span context.\n* restart_without_link: restart_without_link, start a new trace without preserving the candidate span context as a link.\nIf omitted, ignore.\n" + }, + "egress_action": { + "$ref": "#/$defs/ExperimentalTraceContinuationEgressAction", + "description": "The egress propagation action to use when this rule matches.\nValues include:\n* inject_trace_context: inject_trace_context, preserve existing trace context injection behavior.\n* suppress_trace_context: suppress_trace_context, do not inject the selected trace context for this outgoing boundary.\nIf omitted, ignore.\n" + }, + "link_attributes": { + "type": "array", + "minItems": 1, + "items": { + "$ref": "#/$defs/AttributeNameValue" + }, + "description": "Link attributes to attach when strategy is restart_with_link.\nIf omitted, no link attributes are added.\n" + } + }, + "required": [ + "direction" + ], + "allOf": [ + { + "if": { + "properties": { + "direction": { + "const": "ingress" + } + }, + "required": [ + "direction" + ] + }, + "then": { + "required": [ + "strategy" + ] + } + }, + { + "if": { + "properties": { + "direction": { + "const": "egress" + } + }, + "required": [ + "direction" + ] + }, + "then": { + "required": [ + "egress_action" + ] + } + } + ] + }, "ExperimentalSanitization": { "type": "object", "additionalProperties": false, @@ -1282,6 +1479,63 @@ "local" ] }, + "ExperimentalTraceContinuationDecider": { + "type": "object", + "additionalProperties": { + "type": [ + "object", + "null" + ] + }, + "minProperties": 1, + "maxProperties": 1, + "properties": { + "always_continue/development": { + "$ref": "#/$defs/ExperimentalAlwaysContinueTraceContinuationDecider", + "description": "Configure the trace continuation decider to always continue the trace.\nIf omitted, ignore.\n" + }, + "always_restart_with_link/development": { + "$ref": "#/$defs/ExperimentalAlwaysRestartWithLinkTraceContinuationDecider", + "description": "Configure the trace continuation decider to always restart the trace linking previous parent.\nIf omitted, ignore.\n" + }, + "always_restart_without_link/development": { + "$ref": "#/$defs/ExperimentalAlwaysRestartWithoutLinkTraceContinuationDecider", + "description": "Configure the trace continuation decider to always restart without links.\nIf omitted, ignore.\n" + }, + "rule_based/development": { + "$ref": "#/$defs/ExperimentalRuleBasedTraceContinuationDecider", + "description": "Configure the trace continuation decider to use ordered rules.\nIf omitted, ignore.\n" + } + } + }, + "ExperimentalTraceContinuationDirection": { + "type": "string", + "enum": [ + "ingress", + "egress" + ] + }, + "ExperimentalTraceContinuationEgressAction": { + "type": [ + "string", + "null" + ], + "enum": [ + "inject_trace_context", + "suppress_trace_context" + ] + }, + "ExperimentalTraceContinuationStrategy": { + "type": [ + "string", + "null" + ], + "enum": [ + "continue", + "restart_with_link", + "restart_without_link" + ] + }, "ExperimentalTracerConfig": { "type": [ "object" @@ -1672,7 +1926,8 @@ "properties": { "opencensus": { "$ref": "#/$defs/OpenCensusMetricProducer", - "description": "Configure metric producer to be opencensus.\nIf omitted, ignore.\n" + "deprecated": true, + "description": "Configure metric producer to be opencensus.\n\n**Deprecated** as of v1.2.0, may be removed in v2.0.0. The OpenCensus\ncompatibility specification it relies on was deprecated in\nhttps://github.com/open-telemetry/opentelemetry-specification/pull/5138.\nSDKs MAY continue to support this entry for backwards compatibility;\nnew configurations SHOULD NOT use it.\nIf omitted, ignore.\n" } } }, @@ -1718,7 +1973,9 @@ "object", "null" ], - "additionalProperties": false + "additionalProperties": false, + "deprecated": true, + "description": "**Deprecated** as of v1.2.0, may be removed in v2.0.0. The OpenCensus\ncompatibility specification it relies on was deprecated in\nhttps://github.com/open-telemetry/opentelemetry-specification/pull/5138.\nSee also the\n[OpenCensus sunset announcement](https://opentelemetry.io/blog/2023/sunsetting-opencensus/).\n" }, "OtlpGrpcExporter": { "type": [ @@ -2460,6 +2717,10 @@ "$ref": "#/$defs/Sampler", "description": "Configure the sampler.\nIf omitted, parent based sampler with a root of always_on is used.\n" }, + "trace_continuation_decider/development": { + "$ref": "#/$defs/ExperimentalTraceContinuationDecider", + "description": "Configure the trace continuation decider, which determines whether trace context is continued or restarted across ingress and egress boundaries.\nIf omitted, ingress trace context is continued and egress trace context is injected.\n" + }, "id_generator": { "$ref": "#/$defs/IdGenerator", "description": "Configure the trace and span ID generator.\nIf omitted, RandomIdGenerator is used.\n" From e054577fe49dc64e1a02461e107fa1192041a81e Mon Sep 17 00:00:00 2001 From: Riccardo Magliocchetti Date: Tue, 21 Jul 2026 15:00:42 +0200 Subject: [PATCH 13/13] Implement parsing of ConfiguredRuleBasedTraceContinuationDecider --- .../configuration/_tracer_provider.py | 387 ++++++++++++++ .../tests/test_tracer_provider.py | 478 ++++++++++++++++++ 2 files changed, 865 insertions(+) diff --git a/opentelemetry-configuration/src/opentelemetry/configuration/_tracer_provider.py b/opentelemetry-configuration/src/opentelemetry/configuration/_tracer_provider.py index c65b659f4c2..9dc2c978e77 100644 --- a/opentelemetry-configuration/src/opentelemetry/configuration/_tracer_provider.py +++ b/opentelemetry-configuration/src/opentelemetry/configuration/_tracer_provider.py @@ -4,6 +4,9 @@ from __future__ import annotations import logging +from collections.abc import Sequence +from dataclasses import dataclass +from fnmatch import fnmatchcase from opentelemetry import trace from opentelemetry.configuration._common import ( @@ -16,6 +19,10 @@ ConfigurationError, MissingDependencyError, ) +from opentelemetry.configuration._resource import _coerce_attribute_value +from opentelemetry.configuration.models import ( + AttributeNameValue, +) from opentelemetry.configuration.models import ( ExperimentalComposableRuleBasedSampler as RuleBasedSamplerConfig, ) @@ -28,6 +35,27 @@ from opentelemetry.configuration.models import ( ExperimentalOtlpFileExporter as ExperimentalOtlpFileExporterConfig, ) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDecider as RuleBasedTraceContinuationDeciderConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderConditions as TraceContinuationConditionsConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderRule as TraceContinuationRuleConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationDecider as TraceContinuationDeciderConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationDirection as TraceContinuationDirectionConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationEgressAction as TraceContinuationEgressActionConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationStrategy as TraceContinuationStrategyConfig, +) from opentelemetry.configuration.models import ( IdGenerator as IdGeneratorConfig, ) @@ -98,6 +126,18 @@ Sampler, TraceIdRatioBased, ) +from opentelemetry.sdk.trace.trace_continuation import ( + ALWAYS_CONTINUE, + ALWAYS_RESTART_WITH_LINK, + ALWAYS_RESTART_WITHOUT_LINK, + ContinuationDirection, + ContinuationResult, + EgressAction, + TraceContinuationDecider, +) +from opentelemetry.sdk.trace.trace_continuation import ( + Decision as ContinuationDecision, +) from opentelemetry.trace import SpanKind as TraceSpanKind _logger = logging.getLogger(__name__) @@ -400,6 +440,343 @@ def _create_parent_based_sampler(config: ParentBasedSamplerConfig) -> Sampler: return ParentBased(**kwargs) +_STRATEGY_MAP = { + TraceContinuationStrategyConfig.continue_: ContinuationDecision.CONTINUE, + TraceContinuationStrategyConfig.restart_with_link: ( + ContinuationDecision.RESTART_WITH_LINK + ), + TraceContinuationStrategyConfig.restart_without_link: ( + ContinuationDecision.RESTART_WITHOUT_LINK + ), +} + +_EGRESS_ACTION_MAP = { + TraceContinuationEgressActionConfig.inject_trace_context: ( + EgressAction.INJECT_TRACE_CONTEXT + ), + TraceContinuationEgressActionConfig.suppress_trace_context: ( + EgressAction.SUPPRESS_TRACE_CONTEXT + ), +} + +_DIRECTION_MAP = { + TraceContinuationDirectionConfig.ingress: ContinuationDirection.INGRESS, + TraceContinuationDirectionConfig.egress: ContinuationDirection.EGRESS, +} + + +def _map_trace_continuation_strategy( + strategy: TraceContinuationStrategyConfig, +) -> ContinuationDecision: + return _STRATEGY_MAP[strategy] + + +def _map_trace_continuation_egress_action( + action: TraceContinuationEgressActionConfig, +) -> EgressAction: + return _EGRESS_ACTION_MAP[action] + + +def _map_trace_continuation_direction( + direction: TraceContinuationDirectionConfig, +) -> ContinuationDirection: + return _DIRECTION_MAP[direction] + + +def _coerce_link_attributes( + link_attributes: list[AttributeNameValue] | None, +) -> dict[str, object] | None: + if link_attributes is None: + return None + return { + attribute.name: _coerce_attribute_value(attribute) + for attribute in link_attributes + } + + +def _trace_continuation_attribute_values(value): + if isinstance(value, Sequence) and not isinstance( + value, (str, bytes, bytearray) + ): + return value + return (value,) + + +def _trace_continuation_attribute_values_match( + attributes, + attribute_values, +) -> bool: + if not attributes or attribute_values.key not in attributes: + return False + expected = frozenset(attribute_values.values) + return any( + str(value) in expected + for value in _trace_continuation_attribute_values( + attributes[attribute_values.key] + ) + ) + + +def _trace_continuation_attribute_patterns_match( + attributes, + attribute_patterns, +) -> bool: + if not attributes or attribute_patterns.key not in attributes: + return False + return any( + _trace_continuation_attribute_pattern_matches_value( + str(value), + attribute_patterns.included, + attribute_patterns.excluded, + ) + for value in _trace_continuation_attribute_values( + attributes[attribute_patterns.key] + ) + ) + + +def _trace_continuation_attribute_pattern_matches_value( + value: str, + included: list[str] | None, + excluded: list[str] | None, +) -> bool: + included_match = included is None or any( + fnmatchcase(value, pattern) for pattern in included + ) + excluded_match = excluded is not None and any( + fnmatchcase(value, pattern) for pattern in excluded + ) + return included_match and not excluded_match + + +def _trace_continuation_conditions_match( + conditions: TraceContinuationConditionsConfig | None, + attributes, +) -> bool: + if conditions is None: + return True + + return all( + _trace_continuation_attribute_values_match( + attributes, attribute_values + ) + for attribute_values in conditions.attribute_values or () + ) and all( + _trace_continuation_attribute_patterns_match( + attributes, attribute_patterns + ) + for attribute_patterns in conditions.attribute_patterns or () + ) + + +@dataclass(frozen=True) +class _ConfiguredTraceContinuationRule: + direction: ContinuationDirection + strategy: ContinuationDecision | None = None + egress_action: EgressAction | None = None + conditions: TraceContinuationConditionsConfig | None = None + span_kind: TraceSpanKind | None = None + link_attributes: dict[str, object] | None = None + + def matches( + self, + *, + direction: ContinuationDirection, + span_kind: TraceSpanKind | None, + attributes, + ) -> bool: + if direction != self.direction: + return False + if self.span_kind is not None and span_kind != self.span_kind: + return False + return _trace_continuation_conditions_match( + self.conditions, attributes + ) + + +class _ConfiguredRuleBasedTraceContinuationDecider(TraceContinuationDecider): + def __init__( + self, + *, + rules: Sequence[_ConfiguredTraceContinuationRule], + default_strategy: ContinuationDecision = ContinuationDecision.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, + parent_span_context, + kind: TraceSpanKind | None = None, + attributes=None, + links: tuple[trace.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_trace_continuation_rule(rule) + return ContinuationResult( + decision=strategy, + links=_maybe_add_trace_continuation_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_trace_continuation_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=None, + kind: TraceSpanKind | None = None, + attributes=None, + ) -> EgressAction: + for rule in self._rules: + if rule.matches( + direction=ContinuationDirection.EGRESS, + span_kind=kind, + attributes=attributes, + ): + return _egress_action_from_trace_continuation_rule(rule) + + return self._default_egress_action + + +def _maybe_add_trace_continuation_restart_link( + *, + decision: ContinuationDecision, + parent_span_context, + links: tuple[trace.Link, ...], + link_attributes: dict[str, object] | None = None, +) -> tuple[trace.Link, ...]: + if ( + decision is ContinuationDecision.RESTART_WITH_LINK + and parent_span_context + ): + return links + (trace.Link(parent_span_context, link_attributes),) + return links + + +def _ingress_strategy_from_trace_continuation_rule( + rule: _ConfiguredTraceContinuationRule, +) -> ContinuationDecision: + if rule.strategy is None: + raise ValueError( + "Ingress trace continuation rules must define strategy" + ) + return rule.strategy + + +def _egress_action_from_trace_continuation_rule( + rule: _ConfiguredTraceContinuationRule, +) -> EgressAction: + if rule.egress_action is not None: + return rule.egress_action + if rule.strategy is ContinuationDecision.CONTINUE: + return EgressAction.INJECT_TRACE_CONTEXT + if rule.strategy is not None: + return EgressAction.SUPPRESS_TRACE_CONTEXT + raise ValueError( + "Egress trace continuation rules must define egress_action or strategy" + ) + + +def _create_trace_continuation_rules( + config: TraceContinuationRuleConfig, +) -> _ConfiguredTraceContinuationRule: + span_kind = ( + TraceSpanKind[config.span_kind.value.upper()] + if config.span_kind is not None + else None + ) + strategy = ( + _map_trace_continuation_strategy(config.strategy) + if config.strategy is not None + else None + ) + egress_action = ( + _map_trace_continuation_egress_action(config.egress_action) + if config.egress_action is not None + else None + ) + link_attributes = _coerce_link_attributes(config.link_attributes) + direction = _map_trace_continuation_direction(config.direction) + + return _ConfiguredTraceContinuationRule( + direction=direction, + strategy=strategy, + egress_action=egress_action, + conditions=config.conditions, + span_kind=span_kind, + link_attributes=link_attributes, + ) + + +def _create_rule_based_trace_continuation_decider( + config: RuleBasedTraceContinuationDeciderConfig, +) -> _ConfiguredRuleBasedTraceContinuationDecider: + rules = [ + _create_trace_continuation_rules(rule_config) + for rule_config in config.rules + ] + kwargs: dict = {} + if config.default_ingress_strategy is not None: + kwargs["default_strategy"] = _map_trace_continuation_strategy( + config.default_ingress_strategy + ) + if config.default_egress_action is not None: + kwargs["default_egress_action"] = ( + _map_trace_continuation_egress_action(config.default_egress_action) + ) + return _ConfiguredRuleBasedTraceContinuationDecider(rules=rules, **kwargs) + + +def _create_trace_continuation_decider( + config: TraceContinuationDeciderConfig, +) -> TraceContinuationDecider: + if config.always_continue_development is not None: + return ALWAYS_CONTINUE + if config.always_restart_with_link_development is not None: + return ALWAYS_RESTART_WITH_LINK + if config.always_restart_without_link_development is not None: + return ALWAYS_RESTART_WITHOUT_LINK + if config.rule_based_development is not None: + return _create_rule_based_trace_continuation_decider( + config.rule_based_development + ) + if config.additional_properties: + name = next(iter(config.additional_properties)) + return load_entry_point( + "opentelemetry_trace_continuation_decider", name + )() + raise ConfigurationError( + f"Unknown or unsupported trace continuation decider type in config: {config!r}. " + "Supported types: always_continue/development, always_restart_with_link/development, " + "always_restart_without_link/development, rule_based/development." + ) + + def _create_span_limits(config: SpanLimitsConfig) -> SpanLimits: """Create SpanLimits from config. @@ -475,11 +852,21 @@ def create_tracer_provider( ) ) + continuation_decider = ( + _create_trace_continuation_decider( + config.trace_continuation_decider_development + ) + if config is not None + and config.trace_continuation_decider_development is not None + else None + ) + provider = TracerProvider( resource=resource, sampler=sampler, span_limits=span_limits, id_generator=id_generator, + _continuation_decider=continuation_decider, ) if config is not None: diff --git a/opentelemetry-configuration/tests/test_tracer_provider.py b/opentelemetry-configuration/tests/test_tracer_provider.py index b8fe78f82a7..61d39a1a925 100644 --- a/opentelemetry-configuration/tests/test_tracer_provider.py +++ b/opentelemetry-configuration/tests/test_tracer_provider.py @@ -15,6 +15,9 @@ create_tracer_provider, ) from opentelemetry.configuration.file._loader import ConfigurationError +from opentelemetry.configuration.models import ( + AttributeNameValue, +) from opentelemetry.configuration.models import ( BatchSpanProcessor as BatchSpanProcessorConfig, ) @@ -39,9 +42,36 @@ from opentelemetry.configuration.models import ( ExperimentalOtlpFileExporter as ExperimentalOtlpFileExporterConfig, ) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDecider as RuleBasedTraceContinuationDeciderConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderAttributePatterns as TraceContinuationAttributePatternsConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderAttributeValues as TraceContinuationAttributeValuesConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderConditions as TraceContinuationConditionsConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalRuleBasedTraceContinuationDeciderRule as TraceContinuationRuleConfig, +) from opentelemetry.configuration.models import ( ExperimentalSpanParent as SpanParentConfig, ) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationDecider as TraceContinuationDeciderConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationDirection as TraceContinuationDirectionConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationEgressAction as TraceContinuationEgressActionConfig, +) +from opentelemetry.configuration.models import ( + ExperimentalTraceContinuationStrategy as TraceContinuationStrategyConfig, +) from opentelemetry.configuration.models import ( IdGenerator as IdGeneratorConfig, ) @@ -94,6 +124,12 @@ Sampler, TraceIdRatioBased, ) +from opentelemetry.sdk.trace.trace_continuation import ( + ALWAYS_CONTINUE, + ALWAYS_RESTART_WITH_LINK, + ALWAYS_RESTART_WITHOUT_LINK, + EgressAction, +) from opentelemetry.trace import SpanContext, TraceFlags from opentelemetry.trace import SpanKind as TraceSpanKind @@ -907,3 +943,445 @@ def test_empty_id_generator_raises_configuration_error(self): """Empty IdGenerator config (no type specified) raises ConfigurationError.""" with self.assertRaises(ConfigurationError): self._make_provider(IdGeneratorConfig()) + + +class TestCreateTraceContinuationDecider(unittest.TestCase): + @staticmethod + def _make_provider(trace_continuation_decider_config): + return create_tracer_provider( + TracerProviderConfig( + processors=[], + trace_continuation_decider_development=( + trace_continuation_decider_config + ), + ) + ) + + @staticmethod + def _remote_parent_context(): + remote_parent = SpanContext( + TRACE_ID, + SPAN_ID, + is_remote=True, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + ) + return remote_parent, trace_api.set_span_in_context( + trace_api.NonRecordingSpan(remote_parent) + ) + + @staticmethod + def _rule(**kwargs): + return TraceContinuationRuleConfig(**kwargs) + + @staticmethod + def _conditions(**kwargs): + return TraceContinuationConditionsConfig(**kwargs) + + @staticmethod + def _attribute_values(key, values): + return TraceContinuationAttributeValuesConfig(key=key, values=values) + + @staticmethod + def _attribute_patterns(key, included=None, excluded=None): + return TraceContinuationAttributePatternsConfig( + key=key, + included=included, + excluded=excluded, + ) + + def test_always_continue(self): + provider = self._make_provider( + TraceContinuationDeciderConfig(always_continue_development={}) + ) + self.assertIs(provider._continuation_decider, ALWAYS_CONTINUE) + + def test_always_restart_with_link(self): + provider = self._make_provider( + TraceContinuationDeciderConfig( + always_restart_with_link_development={} + ) + ) + self.assertIs(provider._continuation_decider, ALWAYS_RESTART_WITH_LINK) + + def test_always_restart_without_link(self): + provider = self._make_provider( + TraceContinuationDeciderConfig( + always_restart_without_link_development={} + ) + ) + self.assertIs( + provider._continuation_decider, ALWAYS_RESTART_WITHOUT_LINK + ) + + def test_absent_decider_uses_sdk_default(self): + provider = create_tracer_provider(TracerProviderConfig(processors=[])) + self.assertIsNone(provider._continuation_decider) + + def test_rule_based_first_match_wins(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.restart_with_link, + conditions=self._conditions( + attribute_patterns=[ + self._attribute_patterns( + "http.route", ["/webhooks/*"] + ) + ] + ), + ), + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_values=[ + self._attribute_values( + "http.route", ["/webhooks/partner"] + ) + ] + ), + ), + ] + ) + ) + ) + tracer = provider.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_rule_based_default_ingress_strategy(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_patterns=[ + self._attribute_patterns( + "http.route", ["/internal/*"] + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.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_rule_based_matches_all_conditions(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + span_kind=SpanKindConfig.server, + conditions=self._conditions( + attribute_patterns=[ + self._attribute_patterns( + "http.route", ["/internal/*"] + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + child = tracer.start_span( + "child", + context, + kind=TraceSpanKind.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_rule_based_egress_action(self): + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.egress, + egress_action=( + TraceContinuationEgressActionConfig.suppress_trace_context + ), + span_kind=SpanKindConfig.client, + ), + ] + ) + ) + ) + decider = provider._continuation_decider + self.assertEqual( + decider.get_description(), "RuleBasedTraceContinuationDecider" + ) + self.assertEqual( + decider.should_inject(kind=TraceSpanKind.CLIENT), + EgressAction.SUPPRESS_TRACE_CONTEXT, + ) + + def test_rule_based_adds_link_attributes(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.restart_with_link, + conditions=self._conditions( + attribute_patterns=[ + self._attribute_patterns( + "http.route", ["/webhooks/*"] + ) + ] + ), + link_attributes=[ + AttributeNameValue( + name="otel.trace_continuation.reason", + value="external_webhook", + ) + ], + ), + ] + ) + ) + ) + tracer = provider.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].attributes, + {"otel.trace_continuation.reason": "external_webhook"}, + ) + + def test_rule_based_expands_multiple_attribute_values(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_values=[ + self._attribute_values( + "http.route", ["/health", "/ready"] + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + health = tracer.start_span( + "health", context, attributes={"http.route": "/health"} + ) + ready = tracer.start_span( + "ready", context, attributes={"http.route": "/ready"} + ) + other = tracer.start_span( + "other", context, attributes={"http.route": "/other"} + ) + + self.assertEqual(health.parent, remote_parent) + self.assertEqual(ready.parent, remote_parent) + self.assertIsNone(other.parent) + + def test_no_decider_type_raises_configuration_error(self): + with self.assertRaises(ConfigurationError): + self._make_provider(TraceContinuationDeciderConfig()) + + def test_rule_based_attribute_values_stringifies_values(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_values=[ + self._attribute_values( + "http.response.status_code", ["404"] + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + child = tracer.start_span( + "child", + context, + attributes={"http.response.status_code": 404}, + ) + + self.assertEqual(child.parent, remote_parent) + + def test_rule_based_attribute_values_match_exactly(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_values=[ + self._attribute_values( + "http.response.status_code", ["4*"] + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + root = tracer.start_span( + "root", + context, + attributes={"http.response.status_code": "404"}, + ) + + self.assertIsNone(root.parent) + self.assertNotEqual( + root.get_span_context().trace_id, remote_parent.trace_id + ) + + def test_rule_based_duplicate_attribute_conditions_are_anded(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_values=[ + self._attribute_values( + "http.route", ["/health", "/ready"] + ), + self._attribute_values( + "http.route", ["/ready", "/live"] + ), + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + ready = tracer.start_span( + "ready", context, attributes={"http.route": "/ready"} + ) + health = tracer.start_span( + "health", context, attributes={"http.route": "/health"} + ) + + self.assertEqual(ready.parent, remote_parent) + self.assertIsNone(health.parent) + + def test_rule_based_attribute_patterns_exclude_values(self): + remote_parent, context = self._remote_parent_context() + provider = self._make_provider( + TraceContinuationDeciderConfig( + rule_based_development=RuleBasedTraceContinuationDeciderConfig( + rules=[ + self._rule( + direction=TraceContinuationDirectionConfig.ingress, + strategy=TraceContinuationStrategyConfig.continue_, + conditions=self._conditions( + attribute_patterns=[ + self._attribute_patterns( + "http.route", + included=["/internal/*"], + excluded=["/internal/health"], + ) + ] + ), + ), + ], + default_ingress_strategy=( + TraceContinuationStrategyConfig.restart_without_link + ), + ) + ) + ) + tracer = provider.get_tracer(__name__) + users = tracer.start_span( + "users", context, attributes={"http.route": "/internal/users"} + ) + health = tracer.start_span( + "health", context, attributes={"http.route": "/internal/health"} + ) + + self.assertEqual(users.parent, remote_parent) + self.assertIsNone(health.parent)