diff --git a/airflow-core/src/airflow/models/taskinstance.py b/airflow-core/src/airflow/models/taskinstance.py index 4690e8adf4667..d1854c2cc549f 100644 --- a/airflow-core/src/airflow/models/taskinstance.py +++ b/airflow-core/src/airflow/models/taskinstance.py @@ -27,7 +27,7 @@ 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 @@ def log_url(self) -> str: """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 0d92f4e321030..c02af00a0d56f 100644 --- a/airflow-core/tests/unit/models/test_taskinstance.py +++ b/airflow-core/tests/unit/models/test_taskinstance.py @@ -1974,6 +1974,30 @@ def test_log_url(self, create_task_instance): 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 6651a7785e939..97b374a6dea95 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 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 @@ def log_url(self) -> str: 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 bef7f005c5619..a7946f3a5c35d 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 @@ def test_runtime_task_instance_log_url_property(self, create_runtime_ti, base_ur 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."""