anukratb opened a new issue, #75275:
URL: https://github.com/apache/airflow/issues/75275
### Under which category would you file this issue?
Task SDK
### Apache Airflow version
3.3.2 (apache-airflow-task-sdk 1.3.2)
### What happened and how to reproduce it?
If the filesystem that holds the task log fills up while a task is running,
the supervisor's write to the task log file raises `OSError: [Errno 28] No
space left on device`. Nothing in the supervisor handles it, so it propagates
out of `supervise_task()`. The executor reports the task instance as failed,
but the task process and everything it started are never signalled. The task
keeps running unsupervised while the UI shows it as failed. A retry (or a
manual clear) then starts a second copy next to the orphaned one, which is a
real problem for tasks that must not run twice.
Code Path:
1. `_log_to_target()` only handles `ValueError("... closed file")`, so an
`OSError` from the file write propagates:
https://github.com/apache/airflow/blob/3.3.2/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L2453-L2459
2. `_service_subprocess()` only catches `BrokenPipeError` and
`ConnectionResetError` from the socket handlers:
https://github.com/apache/airflow/blob/3.3.2/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L1181-L1184
3. The `finally` in `ActivitySubprocess.wait()` closes the selector and
restores the signal handlers, but never kills the child when
`_monitor_subprocess()` raised:
https://github.com/apache/airflow/blob/3.3.2/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L1515-L1520
4. `execute_task()` in the coordinator has no cleanup around
`process.wait()` either:
https://github.com/apache/airflow/blob/3.3.2/task-sdk/src/airflow/sdk/execution_time/coordinator.py#L144-L153
5. The `finally` in `supervise_task()` then raises a second ENOSPC from
`log_file_descriptor.close()`:
https://github.com/apache/airflow/blob/3.3.2/task-sdk/src/airflow/sdk/execution_time/supervisor.py#L2667-L2669
Steps to Reproduce:
Stock `apache/airflow:3.3.2` image running `airflow standalone`
(LocalExecutor), with the task log folder on a 1 MiB tmpfs so it fills in a few
seconds. The same happens with CeleryExecutor, where the worker logs `Task
execute_workload[...] raised unexpected: OSError(28, 'No space left on
device')`.
1. Save this DAG as `dags/enospc_repro.py`. It writes a heartbeat file every
second and logs a line every 2 ms:
```python
from datetime import datetime
from airflow.providers.standard.operators.bash import BashOperator
from airflow.sdk import dag
@dag(schedule=None, start_date=datetime(2026, 1, 1), catchup=False)
def enospc_repro():
BashOperator(
task_id="publisher",
retries=0,
bash_command=(
"(while true; do date +%s > /tmp/publisher.beat; sleep 1; done)
& "
'while true; do echo "publishing $(date +%s.%N)
padding-padding-padding-padding-padding-padding"; sleep 0.002; done'
),
)
enospc_repro()
```
2. From the directory that contains `dags/`, start Airflow with the task
logs on a 1 MiB tmpfs:
```bash
docker run -d --name enospc-repro \
--tmpfs /tasklogs:size=1m,mode=1777 \
-v "$PWD/dags:/opt/airflow/dags:ro" \
-e AIRFLOW__LOGGING__BASE_LOG_FOLDER=/tasklogs \
-e
AIRFLOW__LOGGING__DAG_PROCESSOR_CHILD_PROCESS_LOG_DIRECTORY=/opt/airflow/logs/dag_processor
\
-e AIRFLOW__CORE__LOAD_EXAMPLES=False \
-e AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=False \
apache/airflow:3.3.2 standalone
```
3. After about a minute, once the DAG is parsed, trigger it:
```bash
docker exec enospc-repro airflow dags trigger enospc_repro
```
4. Within about 15 seconds `/tasklogs` is full. The executor logs `Workload
execution failed.` with this traceback (trimmed), and the scheduler marks the
task instance failed:
```
[error ] Workload execution failed.
[airflow.executors.local_executor.LocalExecutor]
Traceback (most recent call last):
File ".../airflow/sdk/execution_time/supervisor.py", line 2648, in
supervise_task
result = coordinator.execute_task(
File ".../airflow/sdk/execution_time/coordinator.py", line 152, in
execute_task
exit_code = process.wait()
File ".../airflow/sdk/execution_time/supervisor.py", line 1516, in wait
self._monitor_subprocess()
File ".../airflow/sdk/execution_time/supervisor.py", line 1658, in
_monitor_subprocess
alive = self._service_subprocess(max_wait_time=max_wait_time) is None
File ".../airflow/sdk/execution_time/supervisor.py", line 1182, in
_service_subprocess
need_more = socket_handler(key.fileobj)
File ".../airflow/sdk/execution_time/supervisor.py", line 2351, in cb
buffer = process(buffer)
File ".../airflow/sdk/execution_time/supervisor.py", line 2329, in process
gen.send(line)
File ".../airflow/sdk/execution_time/supervisor.py", line 2450, in
process_log_messages_from_subprocess
_log_to_target(target, level, msg, **event)
File ".../airflow/sdk/execution_time/supervisor.py", line 2455, in
_log_to_target
target.log(level, msg, **event)
File ".../structlog/_output.py", line 348, in msg
self._flush()
OSError: [Errno 28] No space left on device
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File ".../airflow/executors/local_executor.py", line 102, in _run_worker
BaseExecutor.run_workload(
File ".../airflow/executors/base_executor.py", line 706, in run_workload
return supervise_task(
File ".../airflow/sdk/execution_time/supervisor.py", line 2669, in
supervise_task
log_file_descriptor.close()
OSError: [Errno 28] No space left on device
```
5. Check the task a minute after it was marked failed:
```bash
docker exec enospc-repro bash -c 'for p in /proc/[0-9]*; do c=$(tr "\0" " "
< $p/cmdline 2>/dev/null); case "$c" in "airflow worker --
"[0-9a-f]*|"/usr/bin/bash -c (while"*) echo "$(basename $p) ${c:0:70}";; esac;
done; echo "heartbeat file age: $(( $(date +%s) - $(cat /tmp/publisher.beat)
))s"'
```
```
97 airflow worker -- 01a15dc7-b2f2-713a-98fe-553ea29dffe3
99 /usr/bin/bash -c (while true; do date +%s > /tmp/publisher.beat; sleep
100 /usr/bin/bash -c (while true; do date +%s > /tmp/publisher.beat; sleep
heartbeat file age: 0s
```
The task runner (the same `pid=97` the scheduler printed in its
`TaskInstance Finished` line) and its bash command are still running, and
nothing will ever stop them. The LocalExecutor worker that owned the task is
back to `<idle>` and picks up new work.
### What you think should happen instead?
The supervisor should never return or raise while the task process is still
alive.
1. If `_monitor_subprocess()` raises, `wait()` should kill the task before
re-raising, using the existing `kill()` (which already signals the task's
process group since #65738): SIGTERM first so the task runner's handler calls
`on_kill()`, then SIGKILL after a timeout.
2. A failure to write the task log should end the task cleanly instead of
crashing the supervisor: kill the task and report it failed with an error
saying the task log could not be written. Dropping log lines and carrying on
would also avoid the orphan, but silently losing the task log seems worse than
failing the task.
### Operating System
Debian GNU/Linux 12 (bookworm), from the official apache/airflow:3.3.2 image
### Deployment
Other Docker-based deployment
### Apache Airflow Provider(s)
_No response_
### Versions of Apache Airflow Providers
Not provider specific. The repro uses the providers bundled in
apache/airflow:3.3.2 (apache-airflow-providers-standard 1.19.0 for
BashOperator). Also reproduced with CeleryExecutor and
apache-airflow-providers-celery 3.24.1.
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
- Happens every time the task log filesystem fills up while a task is
writing logs. Any task on that worker that logs while the disk is full goes
down the same path, not only the task that filled it.
- Related but different: #65505 (fixed by #65738) covered the heartbeat 409
path, where `kill()` was called but did not reach the task's subprocesses. Here
`kill()` is never called at all, because the supervisor raises out of its loop.
- SIGTERM before SIGKILL matters here: `SubprocessHook` (used by
BashOperator) starts the command with `start_new_session=True`, so the command
is outside the task runner's process group and is only stopped through
`on_kill()`. Sending SIGKILL straight away would kill the task runner and still
leave the bash command running.
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
