apache / apache/airflow

CeleryExecutor: no way to forward task logs to worker stdout (subprocess_logs_to_stdout hard-coded False)

Open
#69,554 0 comments 0 reactions 0 assignees View on GitHub
area:Executors-core kind:feature
Dominant language
Python
Stars
46.9k
Forks
17.8k
Avg merge
2d 9h
Merged PRs (30d)
472

Description

### Apache Airflow version

3.2.0 (also present on `main`)

### What happened

When running the **CeleryExecutor**, task subprocess stdout/stderr is **not** forwarded to the Celery worker's own stdout, so it never reaches a container-level log collector (e.g. Kubernetes/AKS `ContainerLogV2`, Loki, etc.). The **LocalExecutor** and the **KubernetesExecutor** per-task pod both forward task logs to stdout, so migrating a deployment from KubernetesExecutor to CeleryExecutor silently drops task-level logs from any stdout-based collector. Task logs still reach the remote/base task-log handler (files/blob) and the UI — only the container-stdout stream is affected.

This is a companion asymmetry to the JSON-rendering gap fixed in #68912 / #68916, but it is a **separate** flag and is **not** addressed by that fix.

### What you think should happen instead

The CeleryExecutor worker should have a supported way (config option, ideally `[celery]`/`[logging]`) to also emit task logs to the worker stdout — i.e. parity with LocalExecutor / the Kubernetes pod path — instead of the value being hard-coded per call site.

### Root cause

Whether task logs are forwarded to the worker's stdout logger is controlled by the keyword-only `subprocess_logs_to_stdout` argument of `supervise()`, which **defaults to `False`**:

- Default `False`: https://github.com/apache/airflow/blob/3.2.0/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L479 and https://github.com/apache/airflow/blob/3.2.0/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L2012
- The stdout logger is only added to the log forwarder's `target_loggers` when the flag is `True`: https://github.com/apache/airflow/blob/3.2.0/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L569-L572 and https://github.com/apache/airflow/blob/3.2.0/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L1505-L1507

The call sites are inconsistent:

- **LocalExecutor** passes `subprocess_logs_to_stdout=True`: https://github.com/apache/airflow/blob/3.2.0/airflow-core/src/airflow/executors/local_executor.py#L144-L152
- **Kubernetes pod path** (`execute_workload`) passes `subprocess_logs_to_stdout=True`: https://github.com/apache/airflow/blob/3.2.0/task-sdk/src/airflow/sdk/execution_time/execute_workload.py#L66-L77
- **CeleryExecutor** calls `supervise(...)` **without** the argument, so it defaults to `False`: https://github.com/apache/airflow/blob/providers-celery/3.17.1/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py#L202-L210

There is currently no `[celery]`/`[logging]` config option to toggle this for the Celery worker, so operators must monkeypatch `supervise`'s default to restore the KubernetesExecutor behaviour.

### How to reproduce

1. Run a deployment with `[core] executor = CeleryExecutor` (Airflow 3.2.0, `apache-airflow-providers-celery==3.17.1`).
2. Run any task that writes to stdout/stderr using a subprocess (e.g. a `BashOperator` echo, or an ssh subprocess).
3. Observe the **Celery worker container stdout** (e.g. `kubectl logs ` / `docker logs`): the task's stdout/stderr lines are absent (they appear in the task log file / UI only).
4. Repeat with LocalExecutor or the KubernetesExecutor pod path: the same lines **do** appear on the process stdout.

### Proposed solution

Add a config option honored by the Celery worker's task execution path — consistent with the `[celery] json_logs` / `[logging] json_logs` fallback pattern added in #68916 — and pass it through to `supervise(...)`:

```python
# providers/celery/.../executors/celery_executor_utils.py, in execute_workload()
subprocess_logs_to_stdout = conf.getboolean("celery", "logs_to_stdout", fallback=None)
if subprocess_logs_to_stdout is None:
subprocess_logs_to_stdout = conf.getboolean("logging", "worker_logs_to_stdout", fallback=False)
supervise(
...,
subprocess_logs_to_stdout=subprocess_logs_to_stdout,
)
```

(Exact config names open to maintainer preference; the key point is Celery parity + a supported toggle rather than a hard-coded default.)

### Are you willing to submit a PR?

Happy to test and open a PR if maintainers agree on the config surface.

### Anything else

Related: #68912 / #68916 fixed the *rendering* (`json_output`) half of this Kubernetes→Celery asymmetry; this issue tracks the *forwarding* (`subprocess_logs_to_stdout`) half, which remains hard-coded.

Contributor guide

Open the contributing guide

Research direction

Start in providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py at execute_workload(), then read the supervise() signature and logging behavior in task-sdk/src/airflow/sdk/execution_time/supervisor.py. Check the existing json_logs configuration pattern. Done means the Celery worker has a supported configuration toggle and passes its value through so task subprocess output can reach worker stdout without changing task-log storage or UI behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
python
Domain
backend, observability
Issue type
Feature
Difficulty
4/5
Estimated time
3-5 days
Activity status
Quiet
Clarity
Mostly clear
Newbie friendliness
56/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.