This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-2-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-2-test by this push:
new b36783c2ba4 [v3-2-test] Bring back pre/post execute log groups for
task logs (#66037) (#66049)
b36783c2ba4 is described below
commit b36783c2ba456544c7d68a218c4ce77881c64086
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri May 1 17:22:40 2026 +0200
[v3-2-test] Bring back pre/post execute log groups for task logs (#66037)
(#66049)
In Airflow 2, task startup and finalization logs were wrapped in
collapsible ::group:: markers so the UI could hide them, letting users
focus on the actual task output. This was lost in the move to structlog.
Re-emit the markers in the task runner:
- ::group::Pre Execute before startup through pre_execute/callbacks
- ::endgroup:: just before execute()
- ::group::Post Execute / ::endgroup:: around finalize()
(cherry picked from commit 0b0271b74c52ad9cc5da5183495e3fb003a02faa)
Co-authored-by: Ash Berlin-Taylor <[email protected]>
---
.../src/airflow/sdk/execution_time/task_runner.py | 7 +++
.../task_sdk/execution_time/test_task_runner.py | 63 ++++++++++++++++++++++
2 files changed, 70 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 5a2624def02..0f89596cda2 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -1290,6 +1290,8 @@ def run(
# Send update only if value changed (e.g., user set context
variables during execution)
if ti.rendered_map_index and ti.rendered_map_index !=
previous_rendered_map_index:
SUPERVISOR_COMMS.send(msg=SetRenderedMapIndex(rendered_map_index=ti.rendered_map_index))
+ finally:
+ log.info("::group::Post Execute")
_push_xcom_if_needed(result, ti, log)
@@ -1679,6 +1681,8 @@ def _execute_task(context: Context, ti:
RuntimeTaskInstance, log: Logger):
_run_task_state_change_callbacks(task, "on_execute_callback", context, log)
+ log.info("::endgroup::")
+
if task.execution_timeout:
from airflow.sdk.execution_time.timeout import timeout
@@ -1846,6 +1850,8 @@ def finalize(
except Exception:
log.exception("error calling listener")
+ log.info("::endgroup::")
+
@contextmanager
def flush_spans():
@@ -1874,6 +1880,7 @@ def main():
with stack:
try:
try:
+ log.info("::group::Pre Execute")
startup_details = get_startup_details()
span = _make_task_span(msg=startup_details)
stack.enter_context(span)
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 9ceedc8ecaf..36084c21db6 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
@@ -802,6 +802,69 @@ def test_run_downstream_skipped(mocked_parse,
create_runtime_ti, mock_supervisor
)
+def test_run_emits_endgroup_before_execute(create_runtime_ti,
mock_supervisor_comms):
+ """::endgroup:: is emitted to close the pre-execute log group just before
execute() runs."""
+ call_order: list[str] = []
+
+ class TrackingOperator(BaseOperator):
+ def execute(self, context):
+ call_order.append("execute")
+
+ task = TrackingOperator(task_id="tracking_task")
+ ti = create_runtime_ti(task=task)
+ log = mock.MagicMock(spec=["info", "debug", "warning", "error",
"exception", "bind"])
+
+ def tracking_info(msg, *args, **kwargs):
+ if msg == "::endgroup::":
+ call_order.append("::endgroup::")
+
+ log.info.side_effect = tracking_info
+
+ run(ti, context=ti.get_template_context(), log=log)
+
+ assert "::endgroup::" in call_order, "::endgroup:: was never emitted"
+ assert "execute" in call_order, "execute() was never called"
+ assert call_order.index("::endgroup::") < call_order.index("execute")
+
+
+def test_run_emits_post_execute_group_before_xcom_push(create_runtime_ti,
mock_supervisor_comms):
+ """::group::Post Execute is opened before 'Pushing xcom' so the push
appears inside the group."""
+ call_order: list[str] = []
+
+ class ReturningOperator(BaseOperator):
+ def execute(self, context):
+ return "some_value"
+
+ task = ReturningOperator(task_id="returning_task")
+ ti = create_runtime_ti(task=task)
+ log = mock.MagicMock(spec=["info", "debug", "warning", "error",
"exception", "bind"])
+
+ def tracking_info(msg, *args, **kwargs):
+ if msg in ("::group::Post Execute", "Pushing xcom"):
+ call_order.append(msg)
+
+ log.info.side_effect = tracking_info
+
+ run(ti, context=ti.get_template_context(), log=log)
+
+ assert "::group::Post Execute" in call_order
+ assert "Pushing xcom" in call_order
+ assert call_order.index("::group::Post Execute") <
call_order.index("Pushing xcom")
+
+
+def test_finalize_emits_endgroup(create_runtime_ti, mock_supervisor_comms):
+ """finalize() closes the post-execute log group but does not open it."""
+ task = BaseOperator(task_id="some_task")
+ ti = create_runtime_ti(task=task)
+ log = mock.MagicMock()
+
+ finalize(ti, state=TaskInstanceState.SUCCESS,
context=ti.get_template_context(), log=log)
+
+ info_calls = log.info.call_args_list
+ assert call("::group::Post Execute") not in info_calls
+ assert info_calls[-1] == call("::endgroup::")
+
+
def test_resume_from_deferred(time_machine, create_runtime_ti,
mock_supervisor_comms, spy_agency: SpyAgency):
from airflow.providers.standard.sensors.date_time import
DateTimeSensorAsync
