ashb commented on code in PR #69633:
URL: https://github.com/apache/airflow/pull/69633#discussion_r3563873236
##########
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)}
Review Comment:
Shouldn't trace state be allowed to be a dict?
##########
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:
```suggestion
context=parent_ctx or context.Context(),
```
?
##########
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:
Same here ? Parameteize and combine?
##########
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:
Move to top level
##########
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:
I feel this needs to be in a try/except for safety
##########
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:
Might be better expressed as match/case?
##########
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
Review Comment:
Ditto
##########
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):
Review Comment:
Please add a trace parent sting (in and out of dict) that is almost a valid
trace id - like the 00- prefix changed to 99-, or an invalid flag at the end.
I think it will raise right now
##########
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:
Combine with previous test to have a single parameteized trace state test fn?
##########
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:
We've already validated the context carrier is set correctly; these two
tests feel like we are testing Otel library.
Remove them
--
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]
