diff --git a/airflow-core/newsfragments/69597.feature.rst b/airflow-core/newsfragments/69597.feature.rst new file mode 100644 index 0000000000000..4c8e5f3bd9247 --- /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. diff --git a/airflow-core/src/airflow/config_templates/config.yml b/airflow-core/src/airflow/config_templates/config.yml index eac6b83174f18..640e9e6d3b989 100644 --- a/airflow-core/src/airflow/config_templates/config.yml +++ b/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). diff --git a/airflow-core/src/airflow/executors/base_executor.py b/airflow-core/src/airflow/executors/base_executor.py index ef5fafded8b39..e336804a94e88 100644 --- a/airflow-core/src/airflow/executors/base_executor.py +++ b/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: "``. :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 1d8a2dfa32936..6b396d7a40ee4 100644 --- a/airflow-core/tests/unit/executors/test_base_executor.py +++ b/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() diff --git a/providers/celery/docs/celery_executor.rst b/providers/celery/docs/celery_executor.rst index 045cec72c9f83..1de19c2cd3efe 100644 --- a/providers/celery/docs/celery_executor.rst +++ b/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 diff --git a/providers/celery/provider.yaml b/providers/celery/provider.yaml index bdde5389f8dd4..6ed15178389c3 100644 --- a/providers/celery/provider.yaml +++ b/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: 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 39d52deb1dd13..c1355c2ac559f 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 @@ 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 02c71ab36ac47..f19e678f3a309 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. 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, 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 62c74c56ef67a..647908ded03ce 100644 --- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py +++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py @@ -23,7 +23,9 @@ import signal import sys from datetime import timedelta +from pathlib import Path from unittest import mock +from uuid import UUID # Leave this it is used by the test worker. import celery.contrib.testing.tasks # noqa: F401 @@ -1238,16 +1240,15 @@ def test_process_workloads_routes_execute_callback(mock_send_workloads, callback mock_send_workloads.assert_called_once_with([(workload.callback.key, workload, expected_queue, None)]) -@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only used for Airflow 3+") -@pytest.mark.skipif(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.""" +@pytest.fixture +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", @@ -1257,21 +1258,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() + + +@pytest.fixture +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()) + +@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only used for Airflow 3+") +@pytest.mark.skipif(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" @@ -1280,39 +1291,71 @@ def test_execute_workload_runs_execute_task_before_airflow_3_3(): assert mock_supervise.call_args.kwargs["log_path"] == "test.log" -@pytest.mark.skipif(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, +@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only used for Airflow 3+") +@pytest.mark.skipif(AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path only applies before Airflow 3.3") +@pytest.mark.parametrize( + ("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 + + +@pytest.mark.skipif(not AIRFLOW_V_3_3_PLUS, reason="3.3+ path uses BaseExecutor.run_workload") +@pytest.mark.parametrize( + ("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 + + +@pytest.mark.skipif(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]