dstandish commented on code in PR #69633:
URL: https://github.com/apache/airflow/pull/69633#discussion_r3573928543
##########
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/dagrun_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(DAGRUN_PARENT_TRACE_CONTEXT_KEY)
+ if isinstance(raw, str):
Review Comment:
Done — rewrote the dispatch as `match`/`case` in 07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
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/dagrun_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(DAGRUN_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)}
+ else:
+ return None
+ ctx = TraceContextTextMapPropagator().extract(carrier)
Review Comment:
Added — the `extract()` call is now wrapped, so any parsing failure degrades
to a root trace instead of failing run creation. 07c19ac7a6
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
airflow-core/src/airflow/models/dagrun.py:
##########
@@ -1143,18 +1172,21 @@ def _emit_dagrun_span(self, state: DagRunState):
if self.partition_date:
attributes["airflow.dag_run.partition_date"] =
self.partition_date.isoformat()
- # TODO: make the empty parent context optional. Default should be
to
- # nest the dag run span under the currently active parent span (by
- # omitting `context` here); only use the empty
`context.Context()` to
- # force a root span when Airflow itself initiates the run (e.g.
dag
- # triggered via API, scheduler, or backfill). Today this forces a
- # root span unconditionally.
- # Tracked at https://github.com/apache/airflow/issues/67210
+ # Root by default: an Airflow-initiated run owns its own trace, so
we do not
+ # attach to whatever span is incidentally active in the scheduler.
When the run
+ # opted into an external trace via
airflow/dagrun_parent_trace_context, start under that
+ # parent so the run nests inside it. The carrier was already built
riding the same
+ # external trace_id (both read the same conf), so the override_ids
trace_id is a
+ # no-op here — only the span_id override takes effect and the
parent link comes
+ # from parent_ctx. The emitted span's sampling then follows the
external parent, so
+ # an unsampled external trace drops this dag_run span even when
airflow/trace_sampled
+ # forced the head decision that the task/worker spans ride.
+ parent_ctx = parent_trace_context(self.conf)
span = tracer.start_span(
name=f"dag_run.{self.dag_id}",
start_time=int((self.queued_at or self.start_date or
timezone.utcnow()).timestamp() * 1e9),
attributes=attributes,
- context=context.Context(),
+ context=parent_ctx if parent_ctx is not None else
context.Context(),
Review Comment:
Done in 07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -4442,6 +4445,149 @@ def
test_context_carrier_honors_trace_sampled_conf(self, dag_maker, flag):
span_ctx = trace.get_current_span(ctx).get_span_context()
assert span_ctx.trace_flags.sampled is flag
+ _EXTERNAL_TRACE_ID = "11111111111111111111111111111111"
+ _EXTERNAL_SPAN_ID = "2222222222222222"
+
+ @pytest.mark.parametrize(
+ ("conf", "expected_trace_id"),
+ [
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"},
+ _EXTERNAL_TRACE_ID,
+ id="traceparent-string",
+ ),
+ pytest.param(
+ {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"
+ }
+ },
+ _EXTERNAL_TRACE_ID,
+ id="carrier-dict",
+ ),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY:
"not-a-traceparent"}, None, id="malformed"),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY: 123}, None,
id="non-str"),
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY: {"nope": "x"}}, None,
id="dict-without-traceparent"
+ ),
+ pytest.param({}, None, id="empty-conf"),
+ pytest.param(None, None, id="no-conf"),
+ pytest.param({"other": "x"}, None, id="unrelated-key"),
+ ],
+ )
+ def test_parent_trace_context(self, conf, expected_trace_id):
+ """Only a valid W3C traceparent (string or carrier dict) yields a
parent context; else None."""
+ from airflow.models.dagrun import parent_trace_context
Review Comment:
Done — moved the pure `parent_trace_context` tests and their shared
constants out of the DB-oriented class to module level in 07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -4442,6 +4445,149 @@ def
test_context_carrier_honors_trace_sampled_conf(self, dag_maker, flag):
span_ctx = trace.get_current_span(ctx).get_span_context()
assert span_ctx.trace_flags.sampled is flag
+ _EXTERNAL_TRACE_ID = "11111111111111111111111111111111"
+ _EXTERNAL_SPAN_ID = "2222222222222222"
+
+ @pytest.mark.parametrize(
+ ("conf", "expected_trace_id"),
+ [
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"},
+ _EXTERNAL_TRACE_ID,
+ id="traceparent-string",
+ ),
+ pytest.param(
+ {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"
+ }
+ },
+ _EXTERNAL_TRACE_ID,
+ id="carrier-dict",
+ ),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY:
"not-a-traceparent"}, None, id="malformed"),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY: 123}, None,
id="non-str"),
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY: {"nope": "x"}}, None,
id="dict-without-traceparent"
+ ),
+ pytest.param({}, None, id="empty-conf"),
+ pytest.param(None, None, id="no-conf"),
+ pytest.param({"other": "x"}, None, id="unrelated-key"),
+ ],
+ )
+ def test_parent_trace_context(self, conf, expected_trace_id):
+ """Only a valid W3C traceparent (string or carrier dict) yields a
parent context; else None."""
+ from airflow.models.dagrun import parent_trace_context
+
+ ctx = parent_trace_context(conf)
+ if expected_trace_id is None:
+ assert ctx is None
+ else:
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert span_ctx.is_valid
+ assert format(span_ctx.trace_id, "032x") == expected_trace_id
+
+ def test_parent_trace_context_carrier_dict_preserves_tracestate(self):
+ """A carrier dict may also carry tracestate, which is extracted
alongside traceparent."""
+ from airflow.models.dagrun import parent_trace_context
+
+ conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01",
+ "tracestate": "foo=bar",
+ }
+ }
+ span_ctx =
otel_trace.get_current_span(parent_trace_context(conf)).get_span_context()
+ assert span_ctx.trace_state.get("foo") == "bar"
+
+ def test_parent_trace_context_drops_non_str_tracestate(self):
Review Comment:
Done — merged into a single parametrized
`test_parent_trace_context_tracestate` in 07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -4442,6 +4445,149 @@ def
test_context_carrier_honors_trace_sampled_conf(self, dag_maker, flag):
span_ctx = trace.get_current_span(ctx).get_span_context()
assert span_ctx.trace_flags.sampled is flag
+ _EXTERNAL_TRACE_ID = "11111111111111111111111111111111"
+ _EXTERNAL_SPAN_ID = "2222222222222222"
+
+ @pytest.mark.parametrize(
+ ("conf", "expected_trace_id"),
+ [
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"},
+ _EXTERNAL_TRACE_ID,
+ id="traceparent-string",
+ ),
+ pytest.param(
+ {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"
+ }
+ },
+ _EXTERNAL_TRACE_ID,
+ id="carrier-dict",
+ ),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY:
"not-a-traceparent"}, None, id="malformed"),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY: 123}, None,
id="non-str"),
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY: {"nope": "x"}}, None,
id="dict-without-traceparent"
+ ),
+ pytest.param({}, None, id="empty-conf"),
+ pytest.param(None, None, id="no-conf"),
+ pytest.param({"other": "x"}, None, id="unrelated-key"),
+ ],
+ )
+ def test_parent_trace_context(self, conf, expected_trace_id):
+ """Only a valid W3C traceparent (string or carrier dict) yields a
parent context; else None."""
+ from airflow.models.dagrun import parent_trace_context
+
+ ctx = parent_trace_context(conf)
+ if expected_trace_id is None:
+ assert ctx is None
+ else:
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert span_ctx.is_valid
+ assert format(span_ctx.trace_id, "032x") == expected_trace_id
+
+ def test_parent_trace_context_carrier_dict_preserves_tracestate(self):
+ """A carrier dict may also carry tracestate, which is extracted
alongside traceparent."""
+ from airflow.models.dagrun import parent_trace_context
+
+ conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01",
+ "tracestate": "foo=bar",
+ }
+ }
+ span_ctx =
otel_trace.get_current_span(parent_trace_context(conf)).get_span_context()
+ assert span_ctx.trace_state.get("foo") == "bar"
+
+ def test_parent_trace_context_drops_non_str_tracestate(self):
+ """A non-str tracestate is dropped instead of being passed to the
propagator, which would raise."""
+ from airflow.models.dagrun import parent_trace_context
+
+ conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01",
+ "tracestate": 123,
+ }
+ }
+ span_ctx =
otel_trace.get_current_span(parent_trace_context(conf)).get_span_context()
+ assert span_ctx.is_valid
+ assert format(span_ctx.trace_id, "032x") == self._EXTERNAL_TRACE_ID
+ assert len(span_ctx.trace_state) == 0
+
+ def test_context_carrier_embeds_external_parent_from_conf(self, dag_maker):
+ """A run with airflow/dagrun_parent_trace_context rides the external
trace_id."""
+ with dag_maker("test_tracing_embed_conf"):
+ EmptyOperator(task_id="t1")
+ dr = dag_maker.create_dagrun(
+ conf={
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01"
+ }
+ )
+
+ ctx = TraceContextTextMapPropagator().extract(dr.context_carrier)
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert format(span_ctx.trace_id, "032x") == self._EXTERNAL_TRACE_ID
+
+ def test_context_carrier_is_root_without_parent_conf(self, dag_maker):
Review Comment:
Done — merged into a parametrized `test_context_carrier_parent_conf` in
07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
##########
airflow-core/tests/unit/models/test_dagrun.py:
##########
@@ -4442,6 +4445,149 @@ def
test_context_carrier_honors_trace_sampled_conf(self, dag_maker, flag):
span_ctx = trace.get_current_span(ctx).get_span_context()
assert span_ctx.trace_flags.sampled is flag
+ _EXTERNAL_TRACE_ID = "11111111111111111111111111111111"
+ _EXTERNAL_SPAN_ID = "2222222222222222"
+
+ @pytest.mark.parametrize(
+ ("conf", "expected_trace_id"),
+ [
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"},
+ _EXTERNAL_TRACE_ID,
+ id="traceparent-string",
+ ),
+ pytest.param(
+ {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{_EXTERNAL_TRACE_ID}-{_EXTERNAL_SPAN_ID}-01"
+ }
+ },
+ _EXTERNAL_TRACE_ID,
+ id="carrier-dict",
+ ),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY:
"not-a-traceparent"}, None, id="malformed"),
+ pytest.param({DAGRUN_PARENT_TRACE_CONTEXT_KEY: 123}, None,
id="non-str"),
+ pytest.param(
+ {DAGRUN_PARENT_TRACE_CONTEXT_KEY: {"nope": "x"}}, None,
id="dict-without-traceparent"
+ ),
+ pytest.param({}, None, id="empty-conf"),
+ pytest.param(None, None, id="no-conf"),
+ pytest.param({"other": "x"}, None, id="unrelated-key"),
+ ],
+ )
+ def test_parent_trace_context(self, conf, expected_trace_id):
+ """Only a valid W3C traceparent (string or carrier dict) yields a
parent context; else None."""
+ from airflow.models.dagrun import parent_trace_context
+
+ ctx = parent_trace_context(conf)
+ if expected_trace_id is None:
+ assert ctx is None
+ else:
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert span_ctx.is_valid
+ assert format(span_ctx.trace_id, "032x") == expected_trace_id
+
+ def test_parent_trace_context_carrier_dict_preserves_tracestate(self):
+ """A carrier dict may also carry tracestate, which is extracted
alongside traceparent."""
+ from airflow.models.dagrun import parent_trace_context
+
+ conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01",
+ "tracestate": "foo=bar",
+ }
+ }
+ span_ctx =
otel_trace.get_current_span(parent_trace_context(conf)).get_span_context()
+ assert span_ctx.trace_state.get("foo") == "bar"
+
+ def test_parent_trace_context_drops_non_str_tracestate(self):
+ """A non-str tracestate is dropped instead of being passed to the
propagator, which would raise."""
+ from airflow.models.dagrun import parent_trace_context
+
+ conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY: {
+ "traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01",
+ "tracestate": 123,
+ }
+ }
+ span_ctx =
otel_trace.get_current_span(parent_trace_context(conf)).get_span_context()
+ assert span_ctx.is_valid
+ assert format(span_ctx.trace_id, "032x") == self._EXTERNAL_TRACE_ID
+ assert len(span_ctx.trace_state) == 0
+
+ def test_context_carrier_embeds_external_parent_from_conf(self, dag_maker):
+ """A run with airflow/dagrun_parent_trace_context rides the external
trace_id."""
+ with dag_maker("test_tracing_embed_conf"):
+ EmptyOperator(task_id="t1")
+ dr = dag_maker.create_dagrun(
+ conf={
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01"
+ }
+ )
+
+ ctx = TraceContextTextMapPropagator().extract(dr.context_carrier)
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert format(span_ctx.trace_id, "032x") == self._EXTERNAL_TRACE_ID
+
+ def test_context_carrier_is_root_without_parent_conf(self, dag_maker):
+ """A run with no parent conf mints its own root trace, not the
external one."""
+ with dag_maker("test_tracing_root_default"):
+ EmptyOperator(task_id="t1")
+ dr = dag_maker.create_dagrun()
+
+ ctx = TraceContextTextMapPropagator().extract(dr.context_carrier)
+ span_ctx = otel_trace.get_current_span(ctx).get_span_context()
+ assert format(span_ctx.trace_id, "032x") != self._EXTERNAL_TRACE_ID
+
+ def test_emit_dagrun_span_nests_under_external_parent(self, dag_maker,
session):
+ """With the parent conf key set, the dag_run span is a child of the
external span."""
+ in_mem_exporter = InMemorySpanExporter()
+ provider = TracerProvider(id_generator=OverrideableRandomIdGenerator())
+ provider.add_span_processor(SimpleSpanProcessor(in_mem_exporter))
+ test_tracer = provider.get_tracer("test")
+
+ with dag_maker("test_tracing_embed_emit", session=session) as dag:
+ EmptyOperator(task_id="t1")
+ dr = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ dr.dag = dag
+ dr.conf = {
+ DAGRUN_PARENT_TRACE_CONTEXT_KEY:
f"00-{self._EXTERNAL_TRACE_ID}-{self._EXTERNAL_SPAN_ID}-01"
+ }
+ # Carrier already rides the external trace with its own child span id,
sampled.
+ dr.context_carrier = {"traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-3333333333333333-01"}
+
+ with mock.patch("airflow.models.dagrun.tracer", test_tracer):
+ dr._emit_dagrun_span(state=DagRunState.SUCCESS)
+
+ spans = in_mem_exporter.get_finished_spans()
+ assert len(spans) == 1
+ span = spans[0]
+ assert span.context.trace_id == int(self._EXTERNAL_TRACE_ID, 16)
+ assert span.context.span_id == int("3333333333333333", 16)
+ assert span.parent is not None
+ assert span.parent.trace_id == int(self._EXTERNAL_TRACE_ID, 16)
+ assert span.parent.span_id == int(self._EXTERNAL_SPAN_ID, 16)
+
+ def test_emit_dagrun_span_is_root_by_default(self, dag_maker, session):
+ """Without the parent conf key, the dag_run span is a root span (no
parent)."""
+ in_mem_exporter = InMemorySpanExporter()
+ provider = TracerProvider(id_generator=OverrideableRandomIdGenerator())
+ provider.add_span_processor(SimpleSpanProcessor(in_mem_exporter))
+ test_tracer = provider.get_tracer("test")
+
+ with dag_maker("test_tracing_root_emit", session=session) as dag:
+ EmptyOperator(task_id="t1")
+ dr = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ dr.dag = dag
+ dr.context_carrier = {"traceparent":
f"00-{self._EXTERNAL_TRACE_ID}-3333333333333333-01"}
+
+ with mock.patch("airflow.models.dagrun.tracer", test_tracer):
+ dr._emit_dagrun_span(state=DagRunState.SUCCESS)
+
+ spans = in_mem_exporter.get_finished_spans()
+ assert len(spans) == 1
+ assert spans[0].parent is None
+
Review Comment:
Removed both in 07c19ac7a6.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @dstandish before posting
--
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]
