feat(propagation): add DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT to handle x-org propagation

.github/workflows/system-tests.yml

@@ -20,6 +20,7 @@ jobs:
         with:
           persist-credentials: false
           repository: 'DataDog/system-tests'
+          ref: 'zach.montoya/poc/propagation-behavior'
 
       - name: Build agent
         run: ./build.sh -i agent
@@ -65,6 +66,7 @@ jobs:
         with:
           persist-credentials: false
           repository: 'DataDog/system-tests'
+          ref: 'zach.montoya/poc/propagation-behavior'
 
       - name: Checkout dd-trace-py
         uses: actions/checkout@v4
@@ -117,6 +119,8 @@ jobs:
         with:
           persist-credentials: false
           repository: 'DataDog/system-tests'
+          ref: 'zach.montoya/poc/propagation-behavior'
+
 
       - name: Build runner
         uses: ./.github/actions/install_runner
@@ -290,6 +294,7 @@ jobs:
         with:
           persist-credentials: false
           repository: 'DataDog/system-tests'
+          ref: 'zach.montoya/poc/propagation-behavior'
       - name: Checkout dd-trace-py
         uses: actions/checkout@v4
         with:

ddtrace/_trace/tracer.py

@@ -775,8 +775,7 @@ def _start_span(
         service = config.service_mapping.get(service, service)
 
         links = context._span_links if not parent else []
-
-        if trace_id:
+        if trace_id or links:
             # child_of a non-empty context, so either a local child span or from a remote context
             span = Span(
                 name=name,

ddtrace/contrib/trace_utils.py

@@ -565,9 +565,9 @@ def activate_distributed_headers(tracer, int_config=None, request_headers=None,
         context = HTTPPropagator.extract(request_headers)
 
         # Only need to activate the new context if something was propagated
-        if not context.trace_id:
+        # The new context must have one of these values in order for it to be activated
+        if not context.trace_id and not context._baggage and not context._span_links:
             return None
-
         # Do not reactivate a context with the same trace id
         # DEV: An example could be nested web frameworks, when one layer already
         #      parsed request headers and activated them.
@@ -577,7 +577,10 @@ def activate_distributed_headers(tracer, int_config=None, request_headers=None,
         #     app = Flask(__name__)  # Traced via Flask instrumentation
         #     app = DDWSGIMiddleware(app)  # Extra layer on top for WSGI
         current_context = tracer.current_trace_context()
-        if current_context and current_context.trace_id == context.trace_id:
+
+        # We accept incoming contexts with only baggage or only span_links, however if we
+        # already have a current_context then a baggage only or span_link only context will be tossed out
+        if current_context and (not context.trace_id or current_context.trace_id == context.trace_id):
             log.debug(
                 "will not activate extracted Context(trace_id=%r, span_id=%r), a context with that trace id is already active",  # noqa: E501
                 context.trace_id,

ddtrace/internal/constants.py

@@ -19,6 +19,10 @@
     _PROPAGATION_STYLE_NONE,
     _PROPAGATION_STYLE_BAGGAGE,
 )
+_PROPAGATION_BEHAVIOR_CONTINUE = "continue"
+_PROPAGATION_BEHAVIOR_IGNORE = "ignore"
+_PROPAGATION_BEHAVIOR_RESTART = "restart"
+_PROPAGATION_BEHAVIOR_DEFAULT = _PROPAGATION_BEHAVIOR_CONTINUE
 W3C_TRACESTATE_KEY = "tracestate"
 W3C_TRACEPARENT_KEY = "traceparent"
 W3C_TRACESTATE_PARENT_ID_KEY = "p"

ddtrace/propagation/http.py

@@ -40,6 +40,8 @@
 from ..internal._tagset import decode_tagset_string
 from ..internal._tagset import encode_tagset_values
 from ..internal.compat import ensure_text
+from ..internal.constants import _PROPAGATION_BEHAVIOR_IGNORE
+from ..internal.constants import _PROPAGATION_BEHAVIOR_RESTART
 from ..internal.constants import _PROPAGATION_STYLE_BAGGAGE
 from ..internal.constants import _PROPAGATION_STYLE_NONE
 from ..internal.constants import _PROPAGATION_STYLE_W3C_TRACECONTEXT
@@ -974,12 +976,12 @@ class HTTPPropagator(object):
     """
 
     @staticmethod
-    def _extract_configured_contexts_avail(normalized_headers):
+    def _extract_configured_contexts_avail(normalized_headers: Dict[str, str]) -> Tuple[List[Context], List[str]]:
         contexts = []
         styles_w_ctx = []
         for prop_style in config._propagation_style_extract:
             propagator = _PROP_STYLES[prop_style]
-            context = propagator._extract(normalized_headers)
+            context = propagator._extract(normalized_headers)  # type: ignore
             # baggage is handled separately
             if prop_style == _PROPAGATION_STYLE_BAGGAGE:
                 continue
@@ -988,6 +990,24 @@ def _extract_configured_contexts_avail(normalized_headers):
                 styles_w_ctx.append(prop_style)
         return contexts, styles_w_ctx
 
+    @staticmethod
+    def _context_to_span_link(context: Context, style: str, reason: str) -> Optional[SpanLink]:
+        # encoding expects at least trace_id and span_id
+        if context.span_id and context.trace_id:
+            return SpanLink(
+                context.trace_id,
+                context.span_id,
+                flags=1 if context.sampling_priority and context.sampling_priority > 0 else 0,
+                tracestate=(
+                    context._meta.get(W3C_TRACESTATE_KEY, "") if style == _PROPAGATION_STYLE_W3C_TRACECONTEXT else None
+                ),
+                attributes={
+                    "reason": reason,
+                    "context_headers": style,
+                },
+            )
+        return None
+
     @staticmethod
     def _resolve_contexts(contexts, styles_w_ctx, normalized_headers):
         primary_context = contexts[0]
@@ -996,23 +1016,14 @@ def _resolve_contexts(contexts, styles_w_ctx, normalized_headers):
         for context in contexts[1:]:
             style_w_ctx = styles_w_ctx[contexts.index(context)]
             # encoding expects at least trace_id and span_id
-            if context.span_id and context.trace_id and context.trace_id != primary_context.trace_id:
-                links.append(
-                    SpanLink(
-                        context.trace_id,
-                        context.span_id,
-                        flags=1 if context.sampling_priority and context.sampling_priority > 0 else 0,
-                        tracestate=(
-                            context._meta.get(W3C_TRACESTATE_KEY, "")
-                            if style_w_ctx == _PROPAGATION_STYLE_W3C_TRACECONTEXT
-                            else None
-                        ),
-                        attributes={
-                            "reason": "terminated_context",
-                            "context_headers": style_w_ctx,
-                        },
-                    )
+            if context.trace_id and context.trace_id != primary_context.trace_id:
+                link = HTTPPropagator._context_to_span_link(
+                    context,
+                    style_w_ctx,
+                    "terminated_context",
                 )
+                if link:
+                    links.append(link)
             # if trace_id matches and the propagation style is tracecontext
             # add the tracestate to the primary context
             elif style_w_ctx == _PROPAGATION_STYLE_W3C_TRACECONTEXT:
@@ -1130,24 +1141,29 @@ def my_controller(url, headers):
         :param dict headers: HTTP headers to extract tracing attributes.
         :return: New `Context` with propagated attributes.
         """
-        if not headers:
-            return Context()
+        context = Context()
+        if not headers or config._propagation_behavior_extract == _PROPAGATION_BEHAVIOR_IGNORE:
+            return context
         try:
+            style = ""
             normalized_headers = {name.lower(): v for name, v in headers.items()}
-            context = Context()
             # tracer configured to extract first only
             if config._propagation_extract_first:
                 # loop through the extract propagation styles specified in order, return whatever context we get first
                 for prop_style in config._propagation_style_extract:
                     propagator = _PROP_STYLES[prop_style]
                     context = propagator._extract(normalized_headers)
+                    style = prop_style
                     if config.propagation_http_baggage_enabled is True:
                         _attach_baggage_to_context(normalized_headers, context)
                     break
 
             # loop through all extract propagation styles
             else:
                 contexts, styles_w_ctx = HTTPPropagator._extract_configured_contexts_avail(normalized_headers)
+                # check that styles_w_ctx is not empty
+                if styles_w_ctx:
+                    style = styles_w_ctx[0]
 
                 if contexts:
                     context = HTTPPropagator._resolve_contexts(contexts, styles_w_ctx, normalized_headers)
@@ -1159,9 +1175,12 @@ def my_controller(url, headers):
                 baggage_context = _BaggageHeader._extract(normalized_headers)
                 if baggage_context._baggage != {}:
                     if context:
-                        context._baggage = baggage_context._baggage
+                        context._baggage = baggage_context.get_all_baggage_items()
                     else:
                         context = baggage_context
+            if config._propagation_behavior_extract == _PROPAGATION_BEHAVIOR_RESTART:
+                link = HTTPPropagator._context_to_span_link(context, style, "propagation_behavior_extract")
+                context = Context(baggage=context.get_all_baggage_items(), span_links=[link] if link else [])
 
             return context
 

ddtrace/settings/config.py

@@ -20,6 +20,7 @@
 from ddtrace.vendor.debtcollector import deprecate
 
 from ..internal import gitmetadata
+from ..internal.constants import _PROPAGATION_BEHAVIOR_DEFAULT
 from ..internal.constants import _PROPAGATION_STYLE_DEFAULT
 from ..internal.constants import DEFAULT_BUFFER_SIZE
 from ..internal.constants import DEFAULT_MAX_PAYLOAD_SIZE
@@ -540,6 +541,10 @@ def __init__(self):
 
         self._propagation_extract_first = _get_config("DD_TRACE_PROPAGATION_EXTRACT_FIRST", False, asbool)
 
+        self._propagation_behavior_extract = _get_config(
+            ["DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT"], _PROPAGATION_BEHAVIOR_DEFAULT, self._lower
+        )
+
         # Datadog tracer tags propagation
         x_datadog_tags_max_length = _get_config("DD_TRACE_X_DATADOG_TAGS_MAX_LENGTH", 512, int)
         if x_datadog_tags_max_length < 0:
@@ -978,3 +983,6 @@ def convert_rc_trace_sampling_rules(self, rc_rules: List[Dict[str, Any]]) -> Opt
             return json.dumps(rc_rules)
         else:
             return None
+
+    def _lower(self, value):
+        return value.lower()

docs/configuration.rst

@@ -368,6 +368,26 @@ Trace Context propagation
      version_added:
        v1.7.0: The ``b3multi`` propagation style was added and ``b3`` was deprecated in favor it.
 
+   DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT:
+     default: |
+         ``continue``
+
+     description: |
+         String for how to handle incoming request headers that are extracted for propagation of trace info.
+
+         The supported values are ``continue``, ``restart``, and ``ignore``.
+
+         After extracting the headers for propagation, this configuration determines what is done with them.
+
+         The default value is ``continue`` which always propagates valid headers.
+         ``ignore`` ignores all incoming headers and ``restart`` turns the first extracted valid propagation header 
+         into a span link and propagates baggage if present.
+
+         Example: ``DD_TRACE_PROPAGATION_STYLE_EXTRACT="ignore"`` to ignore all incoming headers and to start a root span without a parent.
+
+     version_added:
+       v2.20.0:
+
    DD_TRACE_PROPAGATION_STYLE_INJECT:
      default: |
          ``tracecontext,datadog``

releasenotes/notes/propagation_behavior_extract-3d16765cfd07485b.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - |
+    propagation: This introduces the environment variable ``DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT`` 
+    to control the behavior of the extraction of distributed tracing headers. The values, ``continue`` (default),
+    ``ignore``, and ``restart``, are supported. The default value is ``continue`` which has no change from the current behavior of always propagating valid headers.
+    ``ignore`` ignores all incoming headers, never propagating the incoming trace information
+    and ``restart`` turns the first extracted propagation style into a span link and propagates baggage if extracted.

tests/telemetry/test_writer.py

@@ -268,7 +268,8 @@ def test_app_started_event_configuration_override(test_agent_session, run_python
     env["DD_SPAN_SAMPLING_RULES_FILE"] = str(file)
     env["DD_TRACE_PARTIAL_FLUSH_ENABLED"] = "false"
     env["DD_TRACE_PARTIAL_FLUSH_MIN_SPANS"] = "3"
-    env["DD_SITE"] = "datadoghq.com"
+    env["DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT"] = "restart"
+
     # By default telemetry collection is enabled after 10 seconds, so we either need to
     # to sleep for 10 seconds or manually call _app_started() to generate the app started event.
     # This delay allows us to collect start up errors and dynamic configurations
@@ -446,6 +447,7 @@ def test_app_started_event_configuration_override(test_agent_session, run_python
         {"name": "DD_TRACE_OTEL_ENABLED", "origin": "env_var", "value": True},
         {"name": "DD_TRACE_PARTIAL_FLUSH_ENABLED", "origin": "env_var", "value": False},
         {"name": "DD_TRACE_PARTIAL_FLUSH_MIN_SPANS", "origin": "env_var", "value": 3},
+        {"name": "DD_TRACE_PROPAGATION_BEHAVIOR_EXTRACT", "origin": "env_var", "value": "restart"},
         {"name": "DD_TRACE_PROPAGATION_EXTRACT_FIRST", "origin": "default", "value": False},
         {"name": "DD_TRACE_PROPAGATION_HTTP_BAGGAGE_ENABLED", "origin": "default", "value": False},
         {"name": "DD_TRACE_PROPAGATION_STYLE_EXTRACT", "origin": "env_var", "value": "tracecontext"},