This is an automated email from the ASF dual-hosted git repository. kaxilnaik pushed a commit to branch v3-0-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 4be0475ce4f05b2d1528f906b5ab59d24c844c51 Author: ayush3singh <[email protected]> AuthorDate: Fri May 16 14:13:06 2025 +0530 Fixes issue RuntimeTaskInstance does not contain log_url | added during taskrunner startup (#50376) (cherry picked from commit 330789ecd8bd3df4c5760f5b6172a10c77b6b0ef) --- .../src/airflow/sdk/execution_time/task_runner.py | 20 ++++++++++++++++++++ .../task_sdk/execution_time/test_task_runner.py | 3 +++ 2 files changed, 23 insertions(+) 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 5fa868b9b88..6c6e597f65e 100644 --- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py +++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py @@ -144,6 +144,8 @@ class RuntimeTaskInstance(TaskInstance): rendered_map_index: str | None = None + log_url: str | None = None + def __rich_repr__(self): yield "id", self.id yield "task_id", self.task_id @@ -549,6 +551,23 @@ def _xcom_push_to_db(ti: RuntimeTaskInstance, key: str, value: Any) -> None: ) +def get_log_url_from_ti(ti: RuntimeTaskInstance) -> str: + from urllib.parse import quote + + from airflow.configuration import conf + + run_id = quote(ti.run_id) + base_url = conf.get("api", "base_url", fallback="http://localhost:8080/";) + map_index_value = getattr(ti, "map_index", -1) + map_index = f"/mapped/{map_index_value}" if map_index_value is not None and map_index_value >= 0 else "" + try_number_value = getattr(ti, "try_number", 0) + 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}dags/{ti.dag_id}/runs/{run_id}/tasks/{ti.task_id}{map_index}{try_number}" + return _log_uri + + def parse(what: StartupDetails, log: Logger) -> RuntimeTaskInstance: # TODO: Task-SDK: # Using DagBag here is about 98% wrong, but it'll do for now @@ -677,6 +696,7 @@ def startup() -> tuple[RuntimeTaskInstance, Context, Logger]: with _airflow_parsing_context_manager(dag_id=msg.ti.dag_id, task_id=msg.ti.task_id): ti = parse(msg, log) + ti.log_url = get_log_url_from_ti(ti) log.debug("DAG file parsed", file=msg.dag_rel_path) run_as_user = getattr(ti.task, "run_as_user", None) or conf.get( 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 180068f801b..f2ff8967081 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 @@ -106,6 +106,7 @@ from airflow.sdk.execution_time.task_runner import ( _push_xcom_if_needed, _xcom_push, finalize, + get_log_url_from_ti, parse, run, startup, @@ -2206,6 +2207,8 @@ class TestTaskRunnerCallsListeners: mocked_parse(what, "basic_dag", task) runtime_ti, context, log = startup() + assert runtime_ti is not None + assert runtime_ti.log_url == get_log_url_from_ti(runtime_ti) assert isinstance(listener.component, TaskRunnerMarker) del listener.component
