kaxil opened a new pull request, #68800:
URL: https://github.com/apache/airflow/pull/68800

   When the API server serves logs for a **RUNNING** task instance, 
`FileTaskHandler` constructs the configured executor to call `get_task_log()` 
and never starts or shuts it down. `KubernetesExecutor.__init__` eagerly 
created a `multiprocessing.Manager()`, which forks a `serve_forever` child 
process. Because the executor instance is cached per process, this leaks **one 
orphaned `Manager` process (~350-400 MB resident) per API-server worker**. With 
gunicorn/uvicorn worker recycling, each refreshed worker's `Manager` reparents 
to PID 1 and is never reaped, so they accumulate and push the API server toward 
OOM.
   
   `get_task_log()` only needs the kube client and the pod namespace; it never 
touches `task_queue`/`result_queue`/`_manager`. The Manager is purely a 
scheduling-loop resource.
   
   Closes #68693.
   
   ## Fix
   
   Create the `multiprocessing.Manager()` and its queues lazily in `start()` 
instead of `__init__()`. The scheduling loop 
(`start`/`execute_async`/`sync`/`end`) is their only consumer, and the 
scheduler always calls `start()` first. `end()` now no-ops when the executor 
was never started. Constructing the executor purely to read logs no longer 
spawns a Manager.
   
   ## Design rationale
   
   - **Why `start()` and not a lazy property?** The queues are only used after 
the scheduler calls `start()`; nothing touches them between construction and 
`start()`. Tying creation to `start()` keeps the lifecycle explicit and pairs 
cleanly with `end()`.
   - **Consistent with `LocalExecutor`.** `LocalExecutor` already defers its 
queue/process creation to `start()`, so it never leaked on this path. This 
change brings `KubernetesExecutor` in line with that pattern, which also covers 
`LocalKubernetesExecutor` (its embedded `KubernetesExecutor` is constructed in 
`__init__` and only started via `LocalKubernetesExecutor.start()`).
   
   ## Affected path
   
   - `FileTaskHandler._read()` calls the executor's `get_task_log` for running 
tasks: 
[file_task_handler.py#L637](https://github.com/apache/airflow/blob/be44ad1ac6/airflow-core/src/airflow/utils/log/file_task_handler.py#L637)
   - which loads/instantiates the default executor: 
[executor_loader.py#L292](https://github.com/apache/airflow/blob/be44ad1ac6/airflow-core/src/airflow/executors/executor_loader.py#L292)
   
   ## Backward compatibility
   
   No public API change. `task_queue`/`result_queue`/`_manager` are `None` 
until `start()` is called; all internal consumers run after `start()`. Behavior 
for a started executor (the scheduler) is unchanged.
   


-- 
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]

Reply via email to