Add support for forwarding CeleryExecutor task logs to stdout

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.

airflow-core/src/airflow/config_templates/config.yml

@@ -921,6 +921,17 @@ 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 the core default consulted by executors that supervise task subprocesses
+        (e.g. the Celery executor's ``[celery] task_logs_to_stdout`` acts as a worker-level override
+        of this setting). Disabled by default to preserve existing behaviour.
+      version_added: 3.4.0
+      type: boolean
+      example: "True"
+      default: "False"
     uvicorn_logging_level:
       description: |
         Logging level for uvicorn (API server and serve-logs).

airflow-core/src/airflow/executors/base_executor.py

@@ -682,7 +682,7 @@ def run_workload(
         *,
         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:
         """
@@ -695,10 +695,15 @@ def run_workload(
         :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

airflow-core/tests/unit/executors/test_base_executor.py

@@ -556,6 +556,78 @@ def test_run_workload_passes_team_name_to_connection_test_supervisor(mock_superv
     )
 
 
+@mock.patch("airflow.sdk.execution_time.supervisor.supervise_task")
+@pytest.mark.parametrize(
+    ("config_overrides", "expected"),
+    [
+        pytest.param({}, False, id="unset-defaults-to-false"),
+        pytest.param({("logging", "task_logs_to_stdout"): "True"}, True, id="logging-conf-true"),
+        pytest.param({("logging", "task_logs_to_stdout"): "False"}, False, id="logging-conf-false"),
+    ],
+)
+def test_run_workload_defaults_subprocess_logs_to_stdout_from_logging_conf(
+    mock_supervise_task, config_overrides, expected
+):
+    """When the caller omits subprocess_logs_to_stdout, it falls back to [logging] task_logs_to_stdout."""
+    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",
+    )
+
+    with conf_vars(config_overrides):
+        BaseExecutor.run_workload(wl, server="http://localhost:8080/execution/")
+
+    assert mock_supervise_task.call_args.kwargs["subprocess_logs_to_stdout"] is expected
+
+
+def test_run_workload_explicit_subprocess_logs_to_stdout_overrides_logging_conf():
+    """An explicit subprocess_logs_to_stdout argument is not overridden by [logging] task_logs_to_stdout."""
+    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",
+    )
+
+    with (
+        conf_vars({("logging", "task_logs_to_stdout"): "True"}),
+        mock.patch("airflow.sdk.execution_time.supervisor.supervise_task") as mock_supervise_task,
+    ):
+        mock_supervise_task.return_value = 0
+        BaseExecutor.run_workload(
+            wl, server="http://localhost:8080/execution/", subprocess_logs_to_stdout=False
+        )
+
+    assert mock_supervise_task.call_args.kwargs["subprocess_logs_to_stdout"] is False
+
+
 def test_trigger_connection_tests_skipped_when_not_supported():
     """trigger_connection_tests is a no-op when supports_connection_test is False."""
     executor = BaseExecutor()

providers/celery/docs/celery_executor.rst

@@ -260,6 +260,44 @@ 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 (applies to all executors that
+supervise task subprocesses) or override it for the Celery worker alone with the
+``[celery]`` section:
+
+.. code-block:: ini
+
+    # Global — affects any executor that supervises task subprocesses:
+    [logging]
+    task_logs_to_stdout = True
+
+    # Or override for the Celery worker only, leaving other components unchanged:
+    [celery]
+    task_logs_to_stdout = True
+
+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

providers/celery/provider.yaml

@@ -229,6 +229,22 @@ 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. This gives the Celery worker parity
+          with the LocalExecutor and the KubernetesExecutor per-task pod, which always
+          forward task logs to stdout. 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:

providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py

@@ -98,6 +98,20 @@
 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)}")

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. This gives the Celery worker parity\nwith the LocalExecutor and the KubernetesExecutor per-task pod, which always\nforward task logs to stdout. 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 defaults to ``False``. Airflow 3+\nonly. On apache-airflow<3.0.0 tasks are still routed through ``execute_command``\nrather than ``supervise``.\n",
+                        "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,