This is an automated email from the ASF dual-hosted git repository.
ashb pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new e5ba7beb7e1 [v3-4-test] Link task instances of task loops to their own
page (#75458) (#75494)
e5ba7beb7e1 is described below
commit e5ba7beb7e1bd597d60b728310a3a7ae8e483783
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 23 18:12:29 2026 +0100
[v3-4-test] Link task instances of task loops to their own page (#75458)
(#75494)
A task instance's log link ignored its loop iteration: the one Airflow
builds put the iteration where the UI expects a map index, pointing at a
page that doesn't exist, and the one the Task SDK builds couldn't tell the
iterations of a loop apart. Failure emails, notifiers and OpenLineage
events carried these links.
(cherry picked from commit a0be43d3b009dbe81cd0c148f5944ab64b93e9f8)
Co-authored-by: Kacper Muda <[email protected]>
Co-authored-by: Claude <[email protected]>
---
airflow-core/src/airflow/models/taskinstance.py | 21 +++++++++++++------
.../tests/unit/models/test_taskinstance.py | 24 ++++++++++++++++++++++
.../src/airflow/sdk/execution_time/task_runner.py | 16 ++++++++-------
.../task_sdk/execution_time/test_task_runner.py | 16 +++++++++++++++
4 files changed, 64 insertions(+), 13 deletions(-)
diff --git a/airflow-core/src/airflow/models/taskinstance.py
b/airflow-core/src/airflow/models/taskinstance.py
index 3b5d9750b67..eeb548e5881 100644
--- a/airflow-core/src/airflow/models/taskinstance.py
+++ b/airflow-core/src/airflow/models/taskinstance.py
@@ -27,7 +27,7 @@ from collections import defaultdict
from collections.abc import Callable, Collection, Iterable, Mapping, Sequence
from datetime import datetime, timedelta
from typing import TYPE_CHECKING, Any, NamedTuple
-from urllib.parse import quote
+from urllib.parse import quote, urlencode
from uuid import UUID
import attrs
@@ -1428,11 +1428,20 @@ class TaskInstance(Base, LoggingMixin, BaseWorkload):
"""Log URL for TaskInstance."""
run_id = quote(self.run_id)
base_url = conf.get("api", "base_url",
fallback="http://localhost:8080/";)
- map_index = f"/mapped/{self.region_index}" if self.region_index >= 0
else ""
- try_number = f"?try_number={self.try_number}" if self.try_number > 0
else ""
- _log_uri =
f"{base_url.rstrip('/')}/dags/{self.dag_id}/runs/{run_id}/tasks/{self.task_id}{map_index}{try_number}"
-
- return _log_uri
+ _log_uri =
f"{base_url.rstrip('/')}/dags/{self.dag_id}/runs/{run_id}/tasks/{self.task_id}"
+ query: dict[str, Any] = {}
+ if self.region_id == SENTINEL_REGION_ID:
+ if self.region_index >= 0:
+ _log_uri += f"/mapped/{self.region_index}"
+ else:
+ # In a region, region_index is a map index only for a mapped task
(in a loop it is the iteration
+ # otherwise), and the UI selects the task instance by its region,
which tells loop iterations apart
+ if (task := getattr(self, "task", None)) is not None and
task.get_needs_expansion():
+ _log_uri += f"/mapped/{self.region_index}"
+ query.update(region_id=self.region_id,
region_index=self.region_index)
+ if self.try_number > 0:
+ query["try_number"] = self.try_number
+ return f"{_log_uri}?{urlencode(query)}" if query else _log_uri
@property
def mark_success_url(self) -> str:
diff --git a/airflow-core/tests/unit/models/test_taskinstance.py
b/airflow-core/tests/unit/models/test_taskinstance.py
index 0d92f4e3210..c02af00a0d5 100644
--- a/airflow-core/tests/unit/models/test_taskinstance.py
+++ b/airflow-core/tests/unit/models/test_taskinstance.py
@@ -1974,6 +1974,30 @@ class TestTaskInstance:
expected_url =
"http://localhost:8080/dags/my_dag/runs/test/tasks/op/mapped/1?try_number=2";
assert ti.log_url == expected_url
+ @pytest.mark.parametrize(
+ ("mapped", "mapped_path"),
+ [
+ pytest.param(False, "", id="loop_iteration"),
+ pytest.param(True, "/mapped/1", id="mapped"),
+ pytest.param(None, "", id="task_not_attached"),
+ ],
+ )
+ def test_log_url_selects_a_task_instance_in_a_region_by_its_region(self,
dag_maker, mapped, mapped_path):
+ with dag_maker("my_dag", serialized=True):
+ if mapped:
+ BashOperator.partial(task_id="op").expand(bash_command=["echo
1", "echo 2"])
+ else:
+ EmptyOperator(task_id="op")
+ ti = dag_maker.create_dagrun(run_id="test").task_instances[0]
+ ti.region_id, ti.region_index, ti.try_number = uuid6.uuid7(), 1, 2
+ if mapped is None:
+ ti.task = None
+
+ assert ti.log_url == (
+
f"http://localhost:8080/dags/my_dag/runs/test/tasks/op{mapped_path}";
+ f"?region_id={ti.region_id}®ion_index=1&try_number=2"
+ )
+
@pytest.mark.parametrize(
"kwargs",
[
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index ee78312e95e..ed1fa449e84 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -32,7 +32,7 @@ from datetime import UTC, datetime, timedelta
from itertools import product
from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any, Literal, NoReturn, cast
-from urllib.parse import quote
+from urllib.parse import quote, urlencode
from uuid import UUID
import attrs
@@ -964,12 +964,14 @@ class RuntimeTaskInstance(TaskInstance):
map_index = (
f"/mapped/{map_index_value}" if map_index_value is not None and
map_index_value >= 0 else ""
)
- try_number_value = self.try_number
- try_number = (
- f"?try_number={try_number_value}" if try_number_value is not None
and try_number_value > 0 else ""
- )
- _log_uri =
f"{base_url.rstrip('/')}/dags/{self.dag_id}/runs/{run_id}/tasks/{self.task_id}{map_index}{try_number}"
- return _log_uri
+ query: dict[str, Any] = {}
+ # The UI selects a task instance in a region by its region, which
tells loop iterations apart
+ if self.region_id is not None and self.region_id.int != 0:
+ query.update(region_id=self.region_id,
region_index=self.region_index)
+ if self.try_number is not None and self.try_number > 0:
+ query["try_number"] = self.try_number
+ _log_uri =
f"{base_url.rstrip('/')}/dags/{self.dag_id}/runs/{run_id}/tasks/{self.task_id}{map_index}"
+ return f"{_log_uri}?{urlencode(query)}" if query else _log_uri
@property
def mark_success_url(self) -> str:
diff --git a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
index 14e1ec4801e..c3f1b9e72e5 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_task_runner.py
@@ -5895,6 +5895,22 @@ class TestTaskRunnerCallsCallbacks:
log_url = runtime_ti.log_url
assert log_url == expected_url
+ @pytest.mark.parametrize(("map_index", "mapped_path"), [(-1, ""), (1,
"/mapped/1")])
+ def
test_runtime_task_instance_log_url_selects_a_task_instance_in_a_region_by_its_region(
+ self, create_runtime_ti, map_index, mapped_path
+ ):
+ task = BaseOperator(task_id="test_task")
+ runtime_ti = create_runtime_ti(
+ task=task, dag_id="test_dag", run_id="test_run", try_number=2,
map_index=map_index
+ )
+ runtime_ti.region_id, runtime_ti.region_index = uuid7(), 1
+
+ with conf_vars({("api", "base_url"): "http://localhost:8080/"}):
+ assert runtime_ti.log_url == (
+
f"http://localhost:8080/dags/test_dag/runs/test_run/tasks/test_task{mapped_path}";
+
f"?region_id={runtime_ti.region_id}®ion_index=1&try_number=2"
+ )
+
def test_task_runner_on_failure_callback_context(self, create_runtime_ti):
"""Test that on_failure_callback context has end_date and duration."""
