Add support for forwarding CeleryExecutor task logs to stdout - #69597
Add support for forwarding CeleryExecutor task logs to stdout#69597jayachandrakasarla wants to merge 7 commits into
Conversation
|
Note that this has been attempted before, so worth checking that your PR satisfies all the review here: #64481 |
356a922 to
45d1584
Compare
44f9200 to
f60d4b3
Compare
|
Quickest fix: git fetch upstream main && git rebase upstream/main
rm uv.lock && uv lock
git add uv.lock && git rebase --continue
git push --force-with-leaseAutomated nudge — ignore if you're not ready to rebase. This comment is updated in place on future |
719b65c to
550b2d3
Compare
| BaseExecutor.run_workload(workload) | ||
| BaseExecutor.run_workload( | ||
| workload, | ||
| subprocess_logs_to_stdout=conf.getboolean("celery", "task_logs_to_stdout", fallback=False), |
There was a problem hiding this comment.
The edge3 worker has the same gap (providers/edge3/src/airflow/providers/edge3/cli/worker.py:438 calls run_workload without this kwarg, and supervise at :451 on the pre-3.3 branch), so a [celery]-scoped key means every executor provider ends up carrying its own copy of the same option. That is the reason #64481 put it in [logging], which is what multimeric's comment above is pointing at. Would it work to have run_workload default subprocess_logs_to_stdout from a core [logging] key when the caller does not pass one, and keep [celery] task_logs_to_stdout as the worker-level override? That is the same two-level lookup this provider already uses for [celery] json_logs falling back to [logging] json_logs.
| 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. Disabled by default to preserve existing behaviour. |
There was a problem hiding this comment.
celery_executor.rst has a "Worker logging" section that walks through [celery] json_logs with an ini example and a version note, and that is where someone hunting for "how do I get task logs into my container log collector" will look, so this option is worth covering there rather than only in the generated config ref. Separately, the provider still supports apache-airflow>=2.11.0, where tasks go through execute_command and never reach supervise, so setting this on Airflow 2 does nothing at all -- saying "Airflow 3+ only" here would match how the json_logs note calls out 3.2.
| ("config_value", "expected"), | ||
| [("True", True), ("False", False)], | ||
| ) | ||
| def test_execute_workload_forwards_task_logs_to_stdout_before_airflow_3_3(config_value, expected): |
There was a problem hiding this comment.
These two new tests are near-verbatim copies of test_execute_workload_runs_execute_task_before_airflow_3_3 and test_execute_workload_runs_base_executor_workload_on_airflow_3_3_plus sitting right beside them: same workload construction, same mock_app setup, differing only in the final assert. Could the conf_vars parametrize be added to those two existing tests instead, or the workload/mock setup pulled into a fixture? 85 lines for two assertions is a lot to keep in sync.
3 participants
Closes #69554
Currently, CeleryExecutor task logs are not forwarded to stdout because this behavior is hardcoded to false in
BaseExecutor.run_workload().This PR allows users to enable log forwarding by setting
AIRFLOW__CELERY__TASK_LOGS_TO_STDOUT=True. By default, this setting remains false to preserve the existing behavior. When enabled, task logs are forwarded to stdout, allowing container-level log collectors to capture them.Below is a screenshot of the Celery worker logs showing the task's stdout output.
Was generative AI tooling used to co-author this PR?