This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new cd9f233f6d4 Add support for forwarding CeleryExecutor task logs to
stdout (#69597)
cd9f233f6d4 is described below
commit cd9f233f6d47374bcab853986b1588123c47318f
Author: Jayachandra Kasarla <[email protected]>
AuthorDate: Wed Oct 14 05:54:31 2026 +0530
Add support for forwarding CeleryExecutor task logs to stdout (#69597)
---
airflow-core/newsfragments/69597.feature.rst | 1 +
.../src/airflow/config_templates/config.yml | 14 +++
.../src/airflow/executors/base_executor.py | 7 +-
.../tests/unit/executors/test_base_executor.py | 46 ++++++++
providers/celery/docs/celery_executor.rst | 51 +++++++++
providers/celery/provider.yaml | 14 +++
.../celery/executors/celery_executor_utils.py | 27 ++++-
.../airflow/providers/celery/get_provider_info.py | 7 ++
.../unit/celery/executors/test_celery_executor.py | 124 ++++++++++++++-------
9 files changed, 248 insertions(+), 43 deletions(-)
diff --git a/airflow-core/newsfragments/69597.feature.rst
b/airflow-core/newsfragments/69597.feature.rst
new file mode 100644
index 00000000000..f1457ab2dfc
--- /dev/null
+++ b/airflow-core/newsfragments/69597.feature.rst
@@ -0,0 +1 @@
+Add ``[logging] task_logs_to_stdout`` core config option to forward task
subprocess stdout/stderr to the worker's own stdout, so a container-level log
collector (e.g. Kubernetes/Loki) captures task logs in addition to the task-log
handler and the UI. It is only consulted by components that do not already
decide this themselves (currently the Celery worker and the Edge3 worker's fork
path); executors that always forward task logs to stdout, such as the
LocalExecutor and the Kubernetes, EC [...]
diff --git a/airflow-core/src/airflow/config_templates/config.yml
b/airflow-core/src/airflow/config_templates/config.yml
index 12089e4c869..2ef0ebdd916 100644
--- a/airflow-core/src/airflow/config_templates/config.yml
+++ b/airflow-core/src/airflow/config_templates/config.yml
@@ -939,6 +939,20 @@ logging:
type: boolean
example: "True"
default: "False"
+ task_logs_to_stdout:
+ description: |
+ Also forward task subprocess stdout/stderr to the worker's own stdout,
so task logs reach
+ a container-level log collector (e.g. Kubernetes/Loki) in addition to
the task-log handler
+ and the UI. This is not a global switch: it is only consulted when a
caller leaves
+ ``subprocess_logs_to_stdout`` out of ``BaseExecutor.run_workload``.
Executors that already pass
+ ``True`` explicitly (the LocalExecutor and the per-task entrypoint
used by the Kubernetes, ECS,
+ Batch and Lambda executors) always forward task logs to stdout and
ignore this setting. Currently
+ it is read by the Celery worker (which resolves ``[celery]
task_logs_to_stdout`` first, then this
+ setting) and by the Edge3 worker's fork path on Airflow 3.4+.
+ version_added: 3.4.0
+ type: boolean
+ example: "True"
+ default: "False"
uvicorn_logging_level:
description: |
Logging level for uvicorn (API server and serve-logs).
diff --git a/airflow-core/src/airflow/executors/base_executor.py
b/airflow-core/src/airflow/executors/base_executor.py
index c00c8f70e79..0e86013e0a7 100644
--- a/airflow-core/src/airflow/executors/base_executor.py
+++ b/airflow-core/src/airflow/executors/base_executor.py
@@ -860,7 +860,7 @@ class BaseExecutor(LoggingMixin):
*,
server: str | None = None,
dry_run: bool = False,
- subprocess_logs_to_stdout: bool = False,
+ subprocess_logs_to_stdout: bool | None = None,
proctitle: str | None = None,
) -> int:
"""
@@ -873,10 +873,15 @@ class BaseExecutor(LoggingMixin):
:param server: Base URL of the API server (used by task workloads).
:param dry_run: If True, execute without actual task execution
(simulate run).
:param subprocess_logs_to_stdout: Should task logs also be sent to
stdout via the main logger.
+ When not passed (``None``), falls back to the core ``[logging]
task_logs_to_stdout``
+ setting, so executors that do not need a worker-level override can
rely on this default.
:param proctitle: Process title to set for this workload. If not
provided, defaults to
``"airflow supervisor: <workload.display_name>"``.
:return: Exit code of the process.
"""
+ if subprocess_logs_to_stdout is None:
+ subprocess_logs_to_stdout = conf.getboolean("logging",
"task_logs_to_stdout", fallback=False)
+
try:
if sys.platform != "darwin":
from setproctitle import setproctitle
diff --git a/airflow-core/tests/unit/executors/test_base_executor.py
b/airflow-core/tests/unit/executors/test_base_executor.py
index 7b2d1051eb2..a44cde67846 100644
--- a/airflow-core/tests/unit/executors/test_base_executor.py
+++ b/airflow-core/tests/unit/executors/test_base_executor.py
@@ -941,6 +941,52 @@ def
test_run_workload_passes_team_name_to_connection_test_supervisor(mock_superv
)
[email protected]("airflow.sdk.execution_time.supervisor.supervise_task")
[email protected](
+ ("config_overrides", "explicit_arg", "expected"),
+ [
+ pytest.param({}, None, False, id="unset-defaults-to-false"),
+ pytest.param({("logging", "task_logs_to_stdout"): "True"}, None, True,
id="logging-conf-true"),
+ pytest.param({("logging", "task_logs_to_stdout"): "False"}, None,
False, id="logging-conf-false"),
+ pytest.param(
+ {("logging", "task_logs_to_stdout"): "True"}, False, False,
id="explicit-false-overrides-conf"
+ ),
+ pytest.param(
+ {("logging", "task_logs_to_stdout"): "False"}, True, True,
id="explicit-true-overrides-conf"
+ ),
+ ],
+)
+def test_run_workload_subprocess_logs_to_stdout(
+ mock_supervise_task, config_overrides, explicit_arg, expected
+):
+ """An omitted subprocess_logs_to_stdout falls back to [logging]
task_logs_to_stdout; an explicit one wins."""
+ mock_supervise_task.return_value = 0
+ wl = workloads.ExecuteTask(
+ ti=workloads.TaskInstance(
+ id="00000000-0000-0000-0000-000000000001",
+ dag_version_id="00000000-0000-0000-0000-000000000002",
+ task_id="test_task",
+ dag_id="test_dag",
+ run_id="test_run",
+ try_number=1,
+ map_index=-1,
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ ),
+ dag_rel_path="test_dag.py",
+ bundle_info=BundleInfo(name="test-bundle", version=None),
+ token="test-token",
+ log_path="test.log",
+ )
+ kwargs = {} if explicit_arg is None else {"subprocess_logs_to_stdout":
explicit_arg}
+
+ with conf_vars(config_overrides):
+ BaseExecutor.run_workload(wl,
server="http://localhost:8080/execution/";, **kwargs)
+
+ assert mock_supervise_task.call_args.kwargs["subprocess_logs_to_stdout"]
is expected
+
+
@mock.patch.dict("os.environ", {}, clear=True)
class TestExecutorConf:
"""Test ExecutorConf shim class that provides team-specific configuration
access."""
diff --git a/providers/celery/docs/celery_executor.rst
b/providers/celery/docs/celery_executor.rst
index 18637f47e55..31a1619ffb0 100644
--- a/providers/celery/docs/celery_executor.rst
+++ b/providers/celery/docs/celery_executor.rst
@@ -236,6 +236,11 @@ components) or override it for the worker alone with the
``[celery]`` section:
[celery]
json_logs = True
+A Celery worker runs up to ``worker_concurrency`` tasks at once, so lines from
concurrent
+tasks interleave on the worker's stdout. Raw ``print``/stderr lines are
forwarded with
+only ``logger="task.stdout"`` and carry no task instance ID, so a log
collector cannot
+attribute them to a task.
+
The lookup order is:
1. ``[celery] json_logs`` — if set, takes precedence.
@@ -253,6 +258,52 @@ code.
``[celery] json_logs = True`` is the safe way to enable JSON logs
regardless
of the core version.
+Task log forwarding to stdout
+------------------------------
+
+Task logs are normally only written to the task-log handler (and surfaced in
the
+UI). To also forward each task's subprocess stdout/stderr to the Celery
worker's
+own stdout — so a container-level log collector (e.g. Kubernetes/Loki) captures
+them too — enable it via the ``[logging]`` section or override it for the
Celery
+worker alone with the ``[celery]`` section. ``[logging] task_logs_to_stdout``
is not a
+global switch: it only applies to components that read it, or that leave
+``subprocess_logs_to_stdout`` unset when calling ``run_workload`` (currently
the Celery
+worker and the Edge3 worker's fork path). Executors that always forward task
logs to stdout, such as the
+LocalExecutor and the Kubernetes, ECS, Batch and Lambda executors, ignore it:
+
+.. code-block:: ini
+
+ # Default for components that don't set the behaviour themselves (Celery,
Edge3):
+ [logging]
+ task_logs_to_stdout = True
+
+ # Or override for the Celery worker only, leaving other components
unchanged:
+ [celery]
+ task_logs_to_stdout = True
+
+A Celery worker runs up to ``worker_concurrency`` tasks at once, so lines from
concurrent
+tasks interleave on the worker's stdout. Raw ``print``/stderr lines are
forwarded with
+only ``logger="task.stdout"`` and carry no task instance ID, so a log
collector cannot
+attribute them to a task.
+
+The lookup order is:
+
+1. ``[celery] task_logs_to_stdout`` — if set, takes precedence.
+2. ``[logging] task_logs_to_stdout`` — used when the celery-specific key is
absent.
+3. ``False`` — the default when neither key is configured.
+
+.. note::
+
+ This is Airflow 3+ only. On ``apache-airflow<3.0.0``, Celery tasks are
routed
+ through ``execute_command`` and never reach ``supervise``, so setting
either key
+ has no effect there.
+
+ ``[logging] task_logs_to_stdout`` is consulted natively by Airflow core
starting
+ in 3.4.0. On earlier 3.x versions this provider resolves the ``[celery]``/
+ ``[logging]`` fallback itself, so the global key still takes effect on
+ Airflow 3.0–3.3 as long as this provider version is installed — it is not
+ limited to 3.4+.
+
.. _celery_executor:queue:
Queues
diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml
index e26ecc6971e..80277f56fc1 100644
--- a/providers/celery/provider.yaml
+++ b/providers/celery/provider.yaml
@@ -232,6 +232,20 @@ config:
type: boolean
example: ~
default: ~
+ task_logs_to_stdout:
+ description: |
+ Also forward task subprocess stdout/stderr to the Celery worker's
own stdout,
+ so task logs reach a container-level log collector (e.g.
Kubernetes/Loki) in
+ addition to the task-log handler and the UI. When set, this takes
precedence over the global
+ ``[logging] task_logs_to_stdout`` setting, allowing the Celery
worker to override
+ it independently. When unset (the default), the value falls back to
+ ``[logging] task_logs_to_stdout``, which itself defaults to
``False``. Airflow 3+
+ only. On apache-airflow<3.0.0 tasks are still routed through
``execute_command``
+ rather than ``supervise``.
+ version_added: ~
+ type: boolean
+ example: ~
+ default: ~
broker_url:
description: |
The Celery broker URL. Celery supports multiple broker types. See:
diff --git
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 39d52deb1dd..c1355c2ac55 100644
---
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -98,6 +98,20 @@ OPERATION_TIMEOUT = conf.getfloat("celery",
"operation_timeout")
CELERY_FETCH_ERR_MSG_HEADER = "Error fetching Celery task state"
+def _celery_task_logs_to_stdout_override() -> bool | None:
+ """
+ Resolve the ``[celery] task_logs_to_stdout`` worker-level override.
+
+ Returns ``None`` when unset so callers can fall back to the core
+ ``[logging] task_logs_to_stdout`` default, mirroring the ``[celery]
json_logs`` /
+ ``[logging] json_logs`` two-level lookup.
+ """
+ value = conf.get("celery", "task_logs_to_stdout", fallback="")
+ if value and value.lower() != "none":
+ return conf.getboolean("celery", "task_logs_to_stdout")
+ return None
+
+
@cache
def get_celery_configuration() -> dict[str, Any]:
"""Get the Celery configuration dictionary."""
@@ -240,7 +254,13 @@ def execute_workload(input: str) -> None:
log.info("[%s] Executing workload in Celery: %s", celery_task_id, workload)
try:
- BaseExecutor.run_workload(workload)
+ subprocess_logs_to_stdout = _celery_task_logs_to_stdout_override()
+ if subprocess_logs_to_stdout is None:
+ subprocess_logs_to_stdout = conf.getboolean("logging",
"task_logs_to_stdout", fallback=False)
+ BaseExecutor.run_workload(
+ workload,
+ subprocess_logs_to_stdout=subprocess_logs_to_stdout,
+ )
except Exception as e:
from airflow.sdk.exceptions import TaskAlreadyRunningError
@@ -277,6 +297,10 @@ def _execute_workload_pre_3_3(input: str) -> None:
try:
if isinstance(workload, workloads.ExecuteTask):
+ subprocess_logs_to_stdout = _celery_task_logs_to_stdout_override()
+ if subprocess_logs_to_stdout is None:
+ # No run_workload() to apply the core default on this pre-3.3
path, so resolve it here.
+ subprocess_logs_to_stdout = conf.getboolean("logging",
"task_logs_to_stdout", fallback=False)
supervise(
# This is the "wrong" ti type, but it duck types the same.
TODO: Create a protocol for this.
ti=workload.ti, # type: ignore[arg-type]
@@ -285,6 +309,7 @@ def _execute_workload_pre_3_3(input: str) -> None:
token=workload.token,
server=conf.get("core", "execution_api_server_url",
fallback=default_execution_api_server),
log_path=workload.log_path,
+ subprocess_logs_to_stdout=subprocess_logs_to_stdout,
)
else:
raise ValueError(f"CeleryExecutor does not know how to handle
{type(workload)}")
diff --git a/providers/celery/src/airflow/providers/celery/get_provider_info.py
b/providers/celery/src/airflow/providers/celery/get_provider_info.py
index 7fd3ba29bde..ec9be555b00 100644
--- a/providers/celery/src/airflow/providers/celery/get_provider_info.py
+++ b/providers/celery/src/airflow/providers/celery/get_provider_info.py
@@ -117,6 +117,13 @@ def get_provider_info():
"example": None,
"default": None,
},
+ "task_logs_to_stdout": {
+ "description": "Also forward task subprocess
stdout/stderr to the Celery worker's own stdout,\nso task logs reach a
container-level log collector (e.g. Kubernetes/Loki) in\naddition to the
task-log handler and the UI. When set, this takes precedence over the
global\n``[logging] task_logs_to_stdout`` setting, allowing the Celery worker
to override\nit independently. When unset (the default), the value falls back
to\n``[logging] task_logs_to_stdout``, which itself d [...]
+ "version_added": None,
+ "type": "boolean",
+ "example": None,
+ "default": None,
+ },
"broker_url": {
"description": "The Celery broker URL. Celery supports
multiple broker types.
See:\nhttps://docs.celeryq.dev/en/stable/getting-started/backends-and-brokers/index.html#broker-overview\n";,
"version_added": None,
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index bbd1493b32f..39d7bf0175b 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -23,8 +23,9 @@ import os
import signal
import sys
from datetime import timedelta
+from pathlib import Path
from unittest import mock
-from uuid import uuid4
+from uuid import UUID, uuid4
# Leave this it is used by the test worker.
import celery.contrib.testing.tasks # noqa: F401
@@ -1270,16 +1271,15 @@ def test_process_workloads_routes_execute_callback(
mock_send_workloads.assert_called_once_with([(workload.callback.key,
workload, expected_queue, None)])
[email protected](not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
[email protected](AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
-def test_execute_workload_runs_execute_task_before_airflow_3_3():
- """execute_workload routes serialized ExecuteTask payloads to supervise on
Airflow 3.0-3.2."""
[email protected]
+def execute_task_workload_json() -> str:
+ """A serialized ExecuteTask workload usable by execute_workload tests."""
from airflow.executors import workloads
workload = workloads.ExecuteTask(
ti=workloads.TaskInstance(
- id="00000000-0000-0000-0000-000000000001",
- dag_version_id="00000000-0000-0000-0000-000000000002",
+ id=UUID("00000000-0000-0000-0000-000000000001"),
+ dag_version_id=UUID("00000000-0000-0000-0000-000000000002"),
task_id="test_task",
dag_id="test_dag",
run_id="test_run",
@@ -1289,21 +1289,31 @@ def
test_execute_workload_runs_execute_task_before_airflow_3_3():
queue="default",
priority_weight=1,
),
- dag_rel_path="test_dag.py",
+ dag_rel_path=Path("test_dag.py"),
bundle_info=workloads.BundleInfo(name="test-bundle", version=None),
token="test-token",
log_path="test.log",
)
+ return workload.model_dump_json()
+
+
[email protected]
+def mock_celery_app():
+ """Patch celery_executor_utils.app with a worker whose current_task id is
fixed."""
mock_current_task = mock.MagicMock()
mock_current_task.request.id = "test-celery-task-id"
mock_app = mock.MagicMock()
mock_app.current_task = mock_current_task
+ with mock.patch.object(celery_executor_utils, "app", mock_app):
+ yield mock_app
- with (
- mock.patch.object(celery_executor_utils, "app", mock_app),
- mock.patch("airflow.sdk.execution_time.supervisor.supervise") as
mock_supervise,
- ):
-
celery_executor_utils.execute_workload.__wrapped__(workload.model_dump_json())
+
[email protected](not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
[email protected](AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
+def
test_execute_workload_runs_execute_task_before_airflow_3_3(execute_task_workload_json,
mock_celery_app):
+ """execute_workload routes serialized ExecuteTask payloads to supervise on
Airflow 3.0-3.2."""
+ with mock.patch("airflow.sdk.execution_time.supervisor.supervise") as
mock_supervise:
+
celery_executor_utils.execute_workload.__wrapped__(execute_task_workload_json)
mock_supervise.assert_called_once()
assert mock_supervise.call_args.kwargs["ti"].task_id == "test_task"
@@ -1312,39 +1322,71 @@ def
test_execute_workload_runs_execute_task_before_airflow_3_3():
assert mock_supervise.call_args.kwargs["log_path"] == "test.log"
[email protected](not AIRFLOW_V_3_3_PLUS, reason="3.3+ path uses
BaseExecutor.run_workload")
-def test_execute_workload_runs_base_executor_workload_on_airflow_3_3_plus():
- """execute_workload routes serialized ExecuteTask payloads to BaseExecutor
on Airflow 3.3+."""
- from airflow.executors import workloads
-
- workload = workloads.ExecuteTask(
- ti=workloads.TaskInstance(
- id="00000000-0000-0000-0000-000000000001",
- dag_version_id="00000000-0000-0000-0000-000000000002",
- task_id="test_task",
- dag_id="test_dag",
- run_id="test_run",
- try_number=1,
- map_index=-1,
- pool_slots=1,
- queue="default",
- priority_weight=1,
[email protected](not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
[email protected](AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
[email protected](
+ ("config_overrides", "expected"),
+ [
+ pytest.param({}, False, id="unset-defaults-to-false"),
+ pytest.param({("logging", "task_logs_to_stdout"): "True"}, True,
id="global-fallback"),
+ pytest.param({("celery", "task_logs_to_stdout"): "True"}, True,
id="celery-override-only"),
+ pytest.param(
+ {("logging", "task_logs_to_stdout"): "True", ("celery",
"task_logs_to_stdout"): "False"},
+ False,
+ id="celery-overrides-global",
),
- dag_rel_path="test_dag.py",
- bundle_info=workloads.BundleInfo(name="test-bundle", version=None),
- token="test-token",
- log_path="test.log",
- )
- mock_current_task = mock.MagicMock()
- mock_current_task.request.id = "test-celery-task-id"
- mock_app = mock.MagicMock()
- mock_app.current_task = mock_current_task
+ ],
+)
+def test_execute_workload_forwards_task_logs_to_stdout_before_airflow_3_3(
+ execute_task_workload_json, mock_celery_app, config_overrides, expected
+):
+ """Before Airflow 3.3, supervise() receives the resolved
[celery]/[logging] task_logs_to_stdout value."""
+ with (
+ conf_vars(config_overrides),
+ mock.patch("airflow.sdk.execution_time.supervisor.supervise") as
mock_supervise,
+ ):
+
celery_executor_utils.execute_workload.__wrapped__(execute_task_workload_json)
+
+ assert mock_supervise.call_args.kwargs["subprocess_logs_to_stdout"] is
expected
+
[email protected](not AIRFLOW_V_3_3_PLUS, reason="3.3+ path uses
BaseExecutor.run_workload")
[email protected](
+ ("config_overrides", "expected"),
+ [
+ pytest.param({}, False, id="unset-defaults-to-false"),
+ pytest.param({("logging", "task_logs_to_stdout"): "True"}, True,
id="global-fallback"),
+ pytest.param({("celery", "task_logs_to_stdout"): "True"}, True,
id="celery-override-only"),
+ pytest.param({("celery", "task_logs_to_stdout"): "False"}, False,
id="celery-override-false"),
+ pytest.param(
+ {("logging", "task_logs_to_stdout"): "True", ("celery",
"task_logs_to_stdout"): "False"},
+ False,
+ id="celery-overrides-global",
+ ),
+ ],
+)
+def test_execute_workload_forwards_task_logs_to_stdout_on_airflow_3_3_plus(
+ execute_task_workload_json, mock_celery_app, config_overrides, expected
+):
+ """On Airflow 3.3+, run_workload() receives the resolved
[celery]/[logging] task_logs_to_stdout
+ value, since older 3.3.x releases predate core's own [logging] fallback in
run_workload().
+ """
with (
- mock.patch.object(celery_executor_utils, "app", mock_app),
+ conf_vars(config_overrides),
mock.patch("airflow.executors.base_executor.BaseExecutor.run_workload") as
mock_run_workload,
):
-
celery_executor_utils.execute_workload.__wrapped__(workload.model_dump_json())
+
celery_executor_utils.execute_workload.__wrapped__(execute_task_workload_json)
+
+ assert mock_run_workload.call_args.kwargs["subprocess_logs_to_stdout"] is
expected
+
+
[email protected](not AIRFLOW_V_3_3_PLUS, reason="3.3+ path uses
BaseExecutor.run_workload")
+def test_execute_workload_runs_base_executor_workload_on_airflow_3_3_plus(
+ execute_task_workload_json, mock_celery_app
+):
+ """execute_workload routes serialized ExecuteTask payloads to BaseExecutor
on Airflow 3.3+."""
+ with
mock.patch("airflow.executors.base_executor.BaseExecutor.run_workload") as
mock_run_workload:
+
celery_executor_utils.execute_workload.__wrapped__(execute_task_workload_json)
mock_run_workload.assert_called_once()
decoded_workload = mock_run_workload.call_args.args[0]
