This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 22a339029e0 Fix ExternalTaskSensor link pointing to the wrong external
Dag run (#74528)
22a339029e0 is described below
commit 22a339029e0d3656452d1845416bdfae444278b7
Author: Kevin Yang <[email protected]>
AuthorDate: Mon Nov 9 11:35:59 2026 -0500
Fix ExternalTaskSensor link pointing to the wrong external Dag run (#74528)
The "External DAG" link reused the sensor's own run ID under the external
Dag, so it usually opened a run that does not exist: whenever
execution_delta or execution_date_fn targets a different logical date, and
whenever the two runs have different run types or run_after values, such as
a
backfilled or manually triggered sensor waiting on a scheduled run.
On Airflow 3 the link is built on the worker, which cannot query the
metadata database, so the sensor records the run IDs it matched through the
Execution API and the link reads them back. Until a run is matched the link
opens the external Dag instead of a guessed run.
* Remove redundant paragraph from ExternalDagLink docstring
* Keep ExternalTaskSensor outcome when storing matched run IDs fails
The run IDs for the External DAG link are stored before the sensor
raises or returns its result, so an Execution API error at that point
could turn a soft-fail skip or a success into a task failure. The link
is best effort and already falls back to the external Dag page.
The single-run link no longer depends on a core-internal URL helper.
---
.../providers/standard/sensors/external_task.py | 53 +++-
.../unit/standard/sensors/test_external_task.py | 281 +++++++++++++++++++--
2 files changed, 308 insertions(+), 26 deletions(-)
diff --git
a/providers/standard/src/airflow/providers/standard/sensors/external_task.py
b/providers/standard/src/airflow/providers/standard/sensors/external_task.py
index 034e86ca261..8f2b8bd58e4 100644
--- a/providers/standard/src/airflow/providers/standard/sensors/external_task.py
+++ b/providers/standard/src/airflow/providers/standard/sensors/external_task.py
@@ -22,6 +22,7 @@ import typing
import warnings
from collections.abc import Callable, Collection, Iterable, Sequence
from typing import TYPE_CHECKING, ClassVar
+from urllib.parse import urljoin
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models.dag import DagModel
@@ -29,6 +30,7 @@ from airflow.providers.common.compat.sdk import (
AirflowSkipException,
BaseOperatorLink,
BaseSensorOperator,
+ XCom,
conf,
)
from airflow.providers.standard.exceptions import (
@@ -65,6 +67,13 @@ if TYPE_CHECKING:
from airflow.providers.common.compat.sdk import Context, TaskInstanceKey
+XCOM_EXTERNAL_RUN_IDS = "external_run_ids"
+
+
+def _build_airflow_ui_url(path: str) -> str:
+ base_url = conf.get("api", "base_url", fallback="/")
+ return urljoin(base_url.rstrip("/") + "/", path)
+
class ExternalDagLink(BaseOperatorLink):
"""
@@ -97,9 +106,12 @@ class ExternalDagLink(BaseOperatorLink):
external_dag_id: str = template_fields.get("external_dag_id",
operator.external_dag_id) # type: ignore[no-redef]
if AIRFLOW_V_3_0_PLUS:
- from airflow.utils.helpers import build_airflow_dagrun_url
-
- return build_airflow_dagrun_url(dag_id=external_dag_id,
run_id=ti_key.run_id)
+ external_run_ids = XCom.get_value(ti_key=ti_key,
key=XCOM_EXTERNAL_RUN_IDS)
+ if not external_run_ids:
+ return _build_airflow_ui_url(f"dags/{external_dag_id}")
+ if len(external_run_ids) > 1:
+ return _build_airflow_ui_url(f"dags/{external_dag_id}/runs")
+ return
_build_airflow_ui_url(f"dags/{external_dag_id}/runs/{external_run_ids[0]}")
from airflow.utils.helpers import build_airflow_url_with_query #
type:ignore[attr-defined]
query = {"dag_id": external_dag_id, "run_id": ti_key.run_id}
@@ -378,16 +390,43 @@ class ExternalTaskSensor(BaseSensorOperator):
if self.failed_states:
count = _get_count(self.failed_states)
count_failed = self._calculate_count(count, dttm_filter)
+ if count_failed > 0:
+ self._push_external_run_ids(context, dttm_filter)
self._handle_failed_states(count_failed)
if self.skipped_states:
count = _get_count(self.skipped_states)
count_skipped = self._calculate_count(count, dttm_filter)
+ if count_skipped > 0:
+ self._push_external_run_ids(context, dttm_filter)
self._handle_skipped_states(count_skipped)
count = _get_count(self.allowed_states)
count_allowed = self._calculate_count(count, dttm_filter)
- return count_allowed == len(dttm_filter)
+ if count_allowed != len(dttm_filter):
+ return False
+ self._push_external_run_ids(context, dttm_filter)
+ return True
+
+ def _push_external_run_ids(self, context: Context, dttm_filter:
Sequence[datetime.datetime]) -> None:
+ """Store the run IDs of the external Dag runs matched by
``dttm_filter`` for ExternalDagLink."""
+ # Runs before the sensor raises or returns its result, so a failure
here must not replace that
+ # result. Without the XCom, ExternalDagLink falls back to the external
Dag page.
+ try:
+ # The Execution API has no lookup from logical date to run ID; the
task states map is keyed by run ID.
+ task_states = context["ti"].get_task_states(
+ dag_id=self.external_dag_id,
+ task_ids=list(self.external_task_ids) if
self.external_task_ids else None,
+ task_group_id=self.external_task_group_id,
+ logical_dates=list(dttm_filter),
+ )
+ context["ti"].xcom_push(key=XCOM_EXTERNAL_RUN_IDS,
value=sorted(task_states))
+ except Exception:
+ self.log.warning(
+ "Could not store the matched run IDs of external Dag %s for
the External DAG link",
+ self.external_dag_id,
+ exc_info=True,
+ )
def _calculate_count(self, count: int, dttm_filter:
Sequence[datetime.datetime]) -> float | int:
"""Calculate the normalized count based on the type of check."""
@@ -523,7 +562,11 @@ class ExternalTaskSensor(BaseSensorOperator):
raise ExternalTaskNotFoundError("No event received from trigger")
# Re-set as attribute after coming back from deferral - to be used by
listeners
- self.external_dates_filter =
self._serialize_dttm_filter(self._get_dttm_filter(context))
+ dttm_filter = self._get_dttm_filter(context)
+ self.external_dates_filter = self._serialize_dttm_filter(dttm_filter)
+
+ if AIRFLOW_V_3_0_PLUS and event["status"] in ("success", "skipped",
"failed"):
+ self._push_external_run_ids(context, dttm_filter)
if event["status"] == "success":
self.log.info("External tasks %s has executed successfully.",
self.external_task_ids)
diff --git
a/providers/standard/tests/unit/standard/sensors/test_external_task.py
b/providers/standard/tests/unit/standard/sensors/test_external_task.py
index 45c6c8329e2..949ca5012e2 100644
--- a/providers/standard/tests/unit/standard/sensors/test_external_task.py
+++ b/providers/standard/tests/unit/standard/sensors/test_external_task.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import contextlib
import itertools
import logging
import re
@@ -37,6 +38,7 @@ from airflow.providers.common.compat.sdk import (
AirflowSensorTimeout,
AirflowSkipException,
TaskDeferred,
+ XCom,
timezone,
)
from airflow.providers.standard.exceptions import (
@@ -52,7 +54,12 @@ from airflow.providers.standard.exceptions import (
from airflow.providers.standard.operators.bash import BashOperator
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.python import PythonOperator
-from airflow.providers.standard.sensors.external_task import
ExternalTaskMarker, ExternalTaskSensor
+from airflow.providers.standard.sensors.external_task import (
+ XCOM_EXTERNAL_RUN_IDS,
+ ExternalDagLink,
+ ExternalTaskMarker,
+ ExternalTaskSensor,
+)
from airflow.providers.standard.sensors.time import TimeSensor
from airflow.providers.standard.triggers.external_task import WorkflowTrigger
from airflow.timetables.base import DataInterval
@@ -1357,11 +1364,15 @@ class TestExternalTaskSensorV3:
self.context["ti"].get_task_states.return_value = {"run_id":
{"test_group.task_id": State.SUCCESS}}
op.execute(context=self.context)
- self.context["ti"].get_task_states.assert_called_once_with(
- dag_id="test_dag_parent",
- logical_dates=[DEFAULT_DATE],
- task_group_id="test_group",
- )
+ assert self.context["ti"].get_task_states.call_args_list == [
+ mock.call(dag_id="test_dag_parent", logical_dates=[DEFAULT_DATE],
task_group_id="test_group"),
+ mock.call(
+ dag_id="test_dag_parent",
+ task_ids=None,
+ task_group_id="test_group",
+ logical_dates=[DEFAULT_DATE],
+ ),
+ ]
assert op.external_dates_filter == DEFAULT_DATE.isoformat()
@pytest.mark.execution_timeout(10)
@@ -1596,13 +1607,64 @@ class TestExternalTaskSensorV3:
with pytest.raises(ExternalTaskGroupFailedError):
op.execute(context=self.context)
- self.context["ti"].get_task_states.assert_called_once_with(
- dag_id="test_dag_parent",
- logical_dates=[DEFAULT_DATE],
- task_group_id="test_group",
- )
+ assert self.context["ti"].get_task_states.call_args_list == [
+ mock.call(dag_id="test_dag_parent", logical_dates=[DEFAULT_DATE],
task_group_id="test_group"),
+ mock.call(
+ dag_id="test_dag_parent",
+ task_ids=None,
+ task_group_id="test_group",
+ logical_dates=[DEFAULT_DATE],
+ ),
+ ]
assert op.external_dates_filter == DEFAULT_DATE.isoformat()
+ def test_poke_does_not_store_external_run_ids_while_waiting(self):
+ op = ExternalTaskSensor(
+ task_id=TASK_ID,
+ external_dag_id=EXTERNAL_DAG_ID,
+ external_task_id=EXTERNAL_TASK_ID,
+ failed_states=[State.FAILED],
+ skipped_states=[State.SKIPPED],
+ )
+ ti = mock.MagicMock(spec=["get_ti_count", "get_task_states",
"xcom_push"])
+ ti.get_ti_count.return_value = 0
+
+ assert op.poke(context={"logical_date": DEFAULT_DATE, "ti": ti}) is
False
+
+ ti.get_task_states.assert_not_called()
+ ti.xcom_push.assert_not_called()
+
+ @pytest.mark.parametrize(
+ ("sensor_kwargs", "expectation"),
+ [
+ pytest.param(
+ {"failed_states": [State.FAILED], "soft_fail": True},
+ pytest.raises(AirflowSkipException),
+ id="failed-soft-fail",
+ ),
+ pytest.param(
+ {"skipped_states": [State.SKIPPED]},
pytest.raises(AirflowSkipException), id="skipped"
+ ),
+ pytest.param({}, contextlib.nullcontext(), id="success"),
+ ],
+ )
+ def test_poke_keeps_outcome_when_storing_external_run_ids_fails(self,
sensor_kwargs, expectation):
+ op = ExternalTaskSensor(
+ task_id=TASK_ID,
+ external_dag_id=EXTERNAL_DAG_ID,
+ external_task_id=EXTERNAL_TASK_ID,
+ **sensor_kwargs,
+ )
+ ti = mock.MagicMock(spec=["get_ti_count", "get_task_states",
"xcom_push"])
+ ti.get_ti_count.return_value = 1
+ ti.get_task_states.side_effect = RuntimeError("Execution API
unavailable")
+
+ with expectation:
+ assert op.poke(context={"logical_date": DEFAULT_DATE, "ti": ti})
is True
+
+ ti.get_task_states.assert_called_once()
+ ti.xcom_push.assert_not_called()
+
def test_get_logical_date(self):
"""For AF 3, we check for logical date or dag_run.run_after in
context."""
@@ -1721,7 +1783,11 @@ class TestExternalTaskAsyncSensor:
deferrable=True,
)
- context = {"execution_date": DEFAULT_DATE, "logical_date":
DEFAULT_DATE}
+ context = {
+ "execution_date": DEFAULT_DATE,
+ "logical_date": DEFAULT_DATE,
+ "ti": mock.MagicMock(spec=["get_task_states", "xcom_push"]),
+ }
with mock.patch.object(sensor.log, "info") as mock_log_info:
sensor.execute_complete(
context=context,
@@ -1738,7 +1804,11 @@ class TestExternalTaskAsyncSensor:
deferrable=True,
)
- context = {"execution_date": DEFAULT_DATE, "logical_date":
DEFAULT_DATE}
+ context = {
+ "execution_date": DEFAULT_DATE,
+ "logical_date": DEFAULT_DATE,
+ "ti": mock.MagicMock(spec=["get_task_states", "xcom_push"]),
+ }
with pytest.raises(ExternalDagFailedError, match="External job has
failed."):
sensor.execute_complete(
context=context,
@@ -1755,7 +1825,11 @@ class TestExternalTaskAsyncSensor:
soft_fail=True,
)
- context = {"execution_date": DEFAULT_DATE, "logical_date":
DEFAULT_DATE}
+ context = {
+ "execution_date": DEFAULT_DATE,
+ "logical_date": DEFAULT_DATE,
+ "ti": mock.MagicMock(spec=["get_task_states", "xcom_push"]),
+ }
with pytest.raises(AirflowSkipException, match="External job has
failed skipping."):
sensor.execute_complete(
context=context,
@@ -1793,11 +1867,74 @@ class TestExternalTaskAsyncSensor:
)
assert sensor.external_dates_filter is None
- context = {"execution_date": DEFAULT_DATE, "logical_date":
DEFAULT_DATE}
+ context = {
+ "execution_date": DEFAULT_DATE,
+ "logical_date": DEFAULT_DATE,
+ "ti": mock.MagicMock(spec=["get_task_states", "xcom_push"]),
+ }
sensor.execute_complete(context=context, event={"status": "success"})
assert sensor.external_dates_filter == DEFAULT_DATE.isoformat()
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Run IDs are looked up
through the Task SDK")
+ @pytest.mark.parametrize(
+ ("status", "expectation"),
+ [
+ pytest.param("success", contextlib.nullcontext(), id="success"),
+ pytest.param("failed", pytest.raises(ExternalDagFailedError),
id="failed"),
+ pytest.param("skipped", pytest.raises(AirflowSkipException),
id="skipped"),
+ ],
+ )
+ def test_defer_execute_complete_pushes_matched_external_run_ids(self,
status, expectation):
+ sensor = ExternalTaskSensor(
+ task_id=TASK_ID,
+ external_task_id=EXTERNAL_TASK_ID,
+ external_dag_id=EXTERNAL_DAG_ID,
+ deferrable=True,
+ )
+ ti = mock.MagicMock(spec=["get_task_states", "xcom_push"])
+ ti.get_task_states.return_value = {"run_b": {EXTERNAL_TASK_ID:
status}, "run_a": {}}
+ context = {"logical_date": DEFAULT_DATE, "ti": ti}
+
+ with expectation:
+ sensor.execute_complete(context=context, event={"status": status})
+
+ ti.get_task_states.assert_called_once_with(
+ dag_id=EXTERNAL_DAG_ID,
+ task_ids=[EXTERNAL_TASK_ID],
+ task_group_id=None,
+ logical_dates=[DEFAULT_DATE],
+ )
+ ti.xcom_push.assert_called_once_with(key=XCOM_EXTERNAL_RUN_IDS,
value=["run_a", "run_b"])
+
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Run IDs are looked up
through the Task SDK")
+ @pytest.mark.parametrize(
+ ("status", "expectation"),
+ [
+ pytest.param("success", contextlib.nullcontext(), id="success"),
+ pytest.param("failed", pytest.raises(ExternalDagFailedError),
id="failed"),
+ pytest.param("skipped", pytest.raises(AirflowSkipException),
id="skipped"),
+ ],
+ )
+ def
test_defer_execute_complete_keeps_outcome_when_storing_external_run_ids_fails(
+ self, status, expectation
+ ):
+ sensor = ExternalTaskSensor(
+ task_id=TASK_ID,
+ external_task_id=EXTERNAL_TASK_ID,
+ external_dag_id=EXTERNAL_DAG_ID,
+ deferrable=True,
+ )
+ ti = mock.MagicMock(spec=["get_task_states", "xcom_push"])
+ ti.get_task_states.side_effect = RuntimeError("Execution API
unavailable")
+ context = {"logical_date": DEFAULT_DATE, "ti": ti}
+
+ with expectation:
+ sensor.execute_complete(context=context, event={"status": status})
+
+ ti.get_task_states.assert_called_once()
+ ti.xcom_push.assert_not_called()
+
def test_poke_interval_set_on_init(self):
"""Test that poke_interval is set on init and the deprecated
poll_interval attribute mirrors it."""
sensor = ExternalTaskSensor(
@@ -1821,7 +1958,9 @@ class TestExternalTaskAsyncSensor:
],
ids=["not_templated", "templated"],
)
[email protected](XCom, "get_value", autospec=True, return_value=None)
def test_external_task_sensor_extra_link(
+ mock_get_value,
external_dag_id,
external_task_id,
expected_external_dag_id,
@@ -1844,7 +1983,7 @@ def test_external_task_sensor_extra_link(
url = task.operator_extra_links[0].get_link(operator=task, ti_key=ti.key)
- assert f"/dags/{expected_external_dag_id}/runs" in url
+ assert url.endswith(f"/dags/{expected_external_dag_id}")
class TestExternalTaskMarker:
@@ -2420,9 +2559,8 @@ class TestExternalDagLink:
pytest.param(ExternalTaskMarker, id="marker"),
],
)
- def test_link_points_to_external_dag_run(self, operator_class, dag_maker):
- from airflow.configuration import conf
-
+ @mock.patch.object(XCom, "get_value", autospec=True, return_value=None)
+ def test_link_points_to_external_dag_without_matched_run(self,
mock_get_value, operator_class, dag_maker):
with dag_maker("test_external_dag_link", serialized=True):
task = operator_class(
task_id="task_with_link",
@@ -2439,5 +2577,106 @@ class TestExternalDagLink:
url = link.get_link(operator=task, ti_key=ti.key)
- base_url = conf.get("api", "base_url", fallback="/").lower()
- assert url == f"{base_url}dags/external_dag/runs/{dr.run_id}"
+ assert url.endswith("/dags/external_dag")
+
+ @pytest.mark.parametrize(
+ ("sensor_kwargs", "external_logical_date"),
+ [
+ pytest.param(
+ {"execution_delta": timedelta(minutes=5)},
+ DEFAULT_DATE - timedelta(minutes=5),
+ id="execution_delta",
+ ),
+ pytest.param(
+ {"execution_date_fn": lambda logical_date: logical_date -
timedelta(minutes=5)},
+ DEFAULT_DATE - timedelta(minutes=5),
+ id="execution_date_fn",
+ ),
+ pytest.param({}, DEFAULT_DATE,
id="same_logical_date_different_run_type"),
+ ],
+ )
+ def test_sensor_link_points_to_waited_external_dag_run(
+ self, sensor_kwargs, external_logical_date, dag_maker
+ ):
+ with dag_maker(
+ "external_dag", schedule=None, start_date=DEFAULT_DATE -
timedelta(days=1), serialized=True
+ ):
+ EmptyOperator(task_id="external_task")
+ external_dr = dag_maker.create_dagrun(
+ run_type=DagRunType.SCHEDULED,
+ logical_date=external_logical_date,
+ state=DagRunState.SUCCESS,
+ )
+
+ with dag_maker("sensor_dag", schedule=None, serialized=True):
+ ExternalTaskSensor(task_id="wait", external_dag_id="external_dag",
**sensor_kwargs)
+ sensor_dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
logical_date=DEFAULT_DATE)
+
+ ti = dag_maker.run_ti("wait", sensor_dr)
+ assert ti.state == State.SUCCESS
+
+ url = dag_maker.serialized_dag.get_task("wait").get_extra_links(ti,
ExternalDagLink.name)
+ assert url.endswith(f"/dags/external_dag/runs/{external_dr.run_id}")
+
+ def
test_sensor_link_points_to_external_dag_runs_when_several_runs_matched(self,
dag_maker):
+ external_logical_dates = [DEFAULT_DATE - timedelta(days=1),
DEFAULT_DATE]
+ with dag_maker(
+ "external_dag", schedule=None, start_date=DEFAULT_DATE -
timedelta(days=1), serialized=True
+ ):
+ EmptyOperator(task_id="external_task")
+ for logical_date in external_logical_dates:
+ dag_maker.create_dagrun(
+ run_type=DagRunType.SCHEDULED, logical_date=logical_date,
state=DagRunState.SUCCESS
+ )
+
+ with dag_maker("sensor_dag", schedule=None, serialized=True):
+ ExternalTaskSensor(
+ task_id="wait",
+ external_dag_id="external_dag",
+ execution_date_fn=lambda _: external_logical_dates,
+ )
+ sensor_dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
logical_date=DEFAULT_DATE)
+
+ ti = dag_maker.run_ti("wait", sensor_dr)
+ assert ti.state == State.SUCCESS
+
+ url = dag_maker.serialized_dag.get_task("wait").get_extra_links(ti,
ExternalDagLink.name)
+ assert url.endswith("/dags/external_dag/runs")
+
+ @pytest.mark.parametrize(
+ ("sensor_kwargs", "expectation", "expected_state"),
+ [
+ pytest.param(
+ {"failed_states": [DagRunState.FAILED]},
+ pytest.raises(ExternalDagFailedError),
+ State.FAILED,
+ id="failed_states",
+ ),
+ pytest.param(
+ {"skipped_states": [DagRunState.FAILED]},
+ contextlib.nullcontext(),
+ State.SKIPPED,
+ id="skipped_states",
+ ),
+ ],
+ )
+ def test_sensor_link_points_to_external_dag_run_that_ended_the_wait(
+ self, sensor_kwargs, expectation, expected_state, dag_maker
+ ):
+ with dag_maker("external_dag", schedule=None, serialized=True):
+ EmptyOperator(task_id="external_task")
+ external_dr = dag_maker.create_dagrun(
+ run_type=DagRunType.SCHEDULED, logical_date=DEFAULT_DATE,
state=DagRunState.FAILED
+ )
+
+ with dag_maker("sensor_dag", schedule=None, serialized=True):
+ ExternalTaskSensor(task_id="wait", external_dag_id="external_dag",
**sensor_kwargs)
+ sensor_dr = dag_maker.create_dagrun(run_type=DagRunType.MANUAL,
logical_date=DEFAULT_DATE)
+
+ with expectation:
+ dag_maker.run_ti("wait", sensor_dr)
+ ti = sensor_dr.get_task_instance("wait")
+ assert ti.state == expected_state
+
+ url = dag_maker.serialized_dag.get_task("wait").get_extra_links(ti,
ExternalDagLink.name)
+ assert url.endswith(f"/dags/external_dag/runs/{external_dr.run_id}")
