xBis7 commented on code in PR #69633:
URL: https://github.com/apache/airflow/pull/69633#discussion_r3552096474
##########
shared/observability/src/airflow_shared/observability/traces/__init__.py:
##########
@@ -60,10 +60,11 @@ def generate_trace_id(self):
TASK_SPAN_DETAIL_LEVEL_KEY = "airflow/task_span_detail_level"
DEFAULT_TASK_SPAN_DETAIL_LEVEL = 1
TRACE_SAMPLED_KEY = "airflow/trace_sampled"
+PARENT_TRACE_CONTEXT_KEY = "airflow/parent_trace_context"
Review Comment:
```suggestion
PARENT_DR_TRACE_CONTEXT_KEY = "airflow/parent_dr_trace_context"
```
For clarity, in case we add traces that aren't related to a dag run. E.g.
the scheduler debug spans that we discussed.
##########
shared/observability/tests/observability/test_traces.py:
##########
@@ -250,6 +267,62 @@ def test_force_sampled_preserves_detail_level(self,
with_sampler):
assert _carrier_is_sampled(carrier) is True
+class TestNewDagrunTraceCarrierParentContext:
+ """parent_context embeds the run in an external trace instead of a fresh
root."""
+
+ @pytest.fixture
+ def with_sampler(self, monkeypatch):
+ def _install(sampler):
+ provider = TracerProvider(sampler=sampler)
+ monkeypatch.setattr(
+
"airflow_shared.observability.traces.trace.get_tracer_provider",
+ lambda: provider,
+ )
+
+ return _install
+
+ def test_embeds_in_parent_trace_with_own_span(self):
+ parent = _parent_context(trace_id=0xABC123, span_id=0xDEF456)
+ span_ctx =
_carrier_span_context(new_dagrun_trace_carrier(parent_context=parent))
+ assert span_ctx.trace_id == 0xABC123 # rides the external trace
+ assert span_ctx.span_id != 0xDEF456 # but is its own child span
+
+ def test_without_parent_context_mints_fresh_root(self):
+ parent = _parent_context(trace_id=0xABC123)
+ embedded =
_carrier_span_context(new_dagrun_trace_carrier(parent_context=parent))
+ root = _carrier_span_context(new_dagrun_trace_carrier())
+ assert root.trace_id != embedded.trace_id
+
+ def test_empty_context_treated_as_no_parent(self):
+ # An empty Context carries an invalid span -> root trace, not an embed.
+ span_ctx =
_carrier_span_context(new_dagrun_trace_carrier(parent_context=context.Context()))
+ assert span_ctx.is_valid
+
+ def test_parent_based_sampler_inherits_sampled_parent(self, with_sampler):
Review Comment:
`test_parent_based_sampler_inherits_sampled_parent`,
`test_parent_based_sampler_inherits_unsampled_parent` and
`test_force_sampled_overrides_parent` could be refactored into 1 parameterized
test.
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -194,6 +195,33 @@ def trace_sampled_override(conf) -> bool | None:
return None
+def parent_trace_context(conf) -> context.Context | None:
+ """
+ External parent trace context from the ``airflow/parent_trace_context``
run conf key.
+
+ Lets a run be embedded in a trace owned by an external system (an upstream
+ orchestrator, event pipeline, CI job, another Airflow) instead of being a
root
+ trace. The value is a W3C ``traceparent`` string, or a carrier dict
carrying
+ ``traceparent`` (and optionally ``tracestate``); anything else --
including a
+ malformed traceparent -- is ignored so it can neither silently mis-parent
the
+ run nor fail run creation. Returns an OpenTelemetry ``Context`` when a
valid
+ parent is present, else None (root trace, the default).
+ """
+ if not conf:
+ return None
+ raw = conf.get(PARENT_TRACE_CONTEXT_KEY)
+ if isinstance(raw, str):
+ carrier = {"traceparent": raw}
+ elif isinstance(raw, dict) and isinstance(raw.get("traceparent"), str):
+ carrier = {k: raw[k] for k in ("traceparent", "tracestate") if
isinstance(raw.get(k), str)}
Review Comment:
Checking specifically for `traceparent` and `tracestate` here seems
redundant because `TraceContextTextMapPropagator().extract(carrier)` already
does a bunch of validations and if something is off, then it's not going to set
the context and the following check
```
if not trace.get_current_span(ctx).get_span_context().is_valid:
```
is going to return `False`.
https://github.com/open-telemetry/opentelemetry-python/blob/main/opentelemetry-api/src/opentelemetry/trace/propagation/tracecontext.py#L24-L64
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
