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