Fixes issue RuntimeTaskInstance does not contain log_url | added during taskrunner startup (#50376)

(cherry picked from commit 330789e)

Author: ayush3singh

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(

task-sdk/tests/task_sdk/execution_time/test_task_runner.py

@@ -106,6 +106,7 @@
     _push_xcom_if_needed,
     _xcom_push,
     finalize,
+    get_log_url_from_ti,
     parse,
     run,
     startup,
@@ -2206,6 +2207,8 @@ def execute(self, context):
         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