Fixes issue RuntimeTaskInstance context does not contain log_url

Author: Ayush

providers/openlineage/src/airflow/providers/openlineage/utils/utils.py

@@ -444,6 +444,7 @@ class TaskInstanceInfo(InfoJsonEncodable):
 
     includes = ["duration", "try_number", "pool", "queued_dttm", "log_url"]
     casts = {
+        "log_url": lambda ti: ti.log_url if getattr(ti, "log_url", -1) != -1 else None,
         "map_index": lambda ti: ti.map_index if getattr(ti, "map_index", -1) != -1 else None,
         "dag_bundle_version": lambda ti: (
             ti.bundle_instance.version if hasattr(ti, "bundle_instance") else None

providers/openlineage/tests/unit/openlineage/utils/test_utils.py

@@ -1499,6 +1499,7 @@ def test_taskinstance_info_af3():
     runtime_ti.bundle_instance = bundle_instance
 
     assert dict(TaskInstanceInfo(runtime_ti)) == {
+        "log_url": None,
         "map_index": 2,
         "try_number": 1,
         "dag_bundle_version": "bundle_version",

task-sdk/src/airflow/sdk/execution_time/task_runner.py

@@ -137,6 +137,10 @@ class RuntimeTaskInstance(TaskInstance):
 
     rendered_map_index: str | None = None
 
+    log_url: str | None = None
+
+    mark_success_url: str | None = None
+
     def __rich_repr__(self):
         yield "id", self.id
         yield "task_id", self.task_id
@@ -148,6 +152,27 @@ def __rich_repr__(self):
 
     __rich_repr__.angular = True  # type: ignore[attr-defined]
 
+    def get_mark_success_url(self):
+        return self.log_url
+
+    def get_log_url_from_ti(self):
+        from urllib.parse import quote
+
+        from airflow.configuration import conf
+
+        run_id = quote(self.run_id)
+        base_url = conf.get("api", "base_url", fallback="http://localhost:8080/")
+        map_index_value = getattr(self, "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(self, "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/{self.dag_id}/runs/{run_id}/tasks/{self.task_id}{map_index}{try_number}"
+        return _log_uri
+
     def get_template_context(self) -> Context:
         # TODO: Move this to `airflow.sdk.execution_time.context`
         #   once we port the entire context logic from airflow/utils/context.py ?
@@ -158,6 +183,9 @@ def get_template_context(self) -> Context:
 
         validated_params = process_params(self.task.dag, self.task, dag_run_conf, suppress_exception=False)
 
+        self.log_url = self.get_log_url_from_ti()
+        self.mark_success_url = self.get_mark_success_url()
+
         context: Context = {
             # From the Task Execution interface
             "dag": self.task.dag,

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

@@ -1028,6 +1028,8 @@ def test_get_context_without_ti_context_from_server(self, mocked_parse, make_ti_
             _ti_context_from_server=None,
             start_date=start_date,
         )
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
+
         context = runtime_ti.get_template_context()
 
         # Verify the context keys and values
@@ -1061,6 +1063,7 @@ def test_get_context_with_ti_context_from_server(self, create_runtime_ti, mock_s
         # `task_sdk/tests/api/test_client.py::test_task_instance_start` checks the context is received
         # from the API server
         runtime_ti = create_runtime_ti(task=task, dag_id="basic_task")
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
 
         dr = runtime_ti._ti_context_from_server.dag_run
 
@@ -1113,6 +1116,7 @@ def test_lazy_loading_not_triggered_until_accessed(self, create_runtime_ti, mock
         """Ensure lazy-loaded attributes are not resolved until accessed."""
         task = BaseOperator(task_id="hello")
         runtime_ti = create_runtime_ti(task=task, dag_id="basic_task")
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
 
         mock_supervisor_comms.get_message.return_value = PrevSuccessfulDagRunResult(
             data_interval_end=timezone.datetime(2025, 1, 1, 2, 0, 0),
@@ -1149,6 +1153,7 @@ def test_get_connection_from_context(self, create_runtime_ti, mock_supervisor_co
         )
 
         runtime_ti = create_runtime_ti(task=task, dag_id="test_get_connection_from_context")
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
         mock_supervisor_comms.get_message.return_value = conn
 
         context = runtime_ti.get_template_context()
@@ -1185,6 +1190,7 @@ def test_template_render(self, create_runtime_ti):
         task = BaseOperator(task_id="test_template_render_task")
 
         runtime_ti = create_runtime_ti(task=task, dag_id="test_template_render")
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
         template_context = runtime_ti.get_template_context()
         result = runtime_ti.task.render_template(
             "Task: {{ dag.dag_id }} -> {{ task.task_id }}", template_context
@@ -1212,6 +1218,7 @@ def test_template_with_connection(
         """
         task = BaseOperator(task_id="hello")
         runtime_ti = create_runtime_ti(task=task, dag_id="test_template_with_connection")
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
 
         conn = ConnectionResult(
             conn_id="a_connection",
@@ -1247,6 +1254,7 @@ def test_get_variable_from_context(
 
         task = BaseOperator(task_id="hello")
         runtime_ti = create_runtime_ti(task=task)
+        runtime_ti.log_url = runtime_ti.get_log_url_from_ti()
 
         var = VariableResult(key="test_key", value=var_value)