add extract behavior feat

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
@@ -973,12 +975,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
@@ -987,6 +989,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]
@@ -995,23 +1015,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:
@@ -1129,24 +1140,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)
@@ -1161,6 +1177,9 @@ def my_controller(url, headers):
                         context._baggage = baggage_context._baggage
                     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()

releasenotes/notes/propagation_behavior_extract-3d16765cfd07485b.yaml

@@ -0,0 +1,7 @@
+---
+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 should have no change from the current behavior of always propagating valid headers.
+    With ``ignore`` ignoring all incoming headers and with ``restart`` turning incoming headers into a span links and propagating baggage items.

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
@@ -444,6 +445,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"},