nosami opened a new issue, #65560: URL: https://github.com/apache/airflow/issues/65560
### Under which category would you file this issue? Providers ### Apache Airflow version 3.1.8 ### What happened and how to reproduce it? ### Apache Airflow version 3.1.8 ### What happened? The triggerer supervisor process (parent process, not the TriggerRunner child) leaks file descriptors and memory when processing deferrable operators that use a re-defer polling pattern (e.g., `TimeDeltaTrigger` in a loop). Each trigger lifecycle event opens a log file via `TriggerLoggingFactory` but the underlying file handle is never explicitly closed. The comment at [line 411](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L411) says _"since structlog doesn't have an explicit close method, we only need to remove the last reference to it to close the open FH"_ — but this assumption is incorrect. The `structlog.BytesLogger` / `structlog.WriteLogger` wrapper does not close the file object passed to it when garbage collected. Additionally, the `TriggerLoggingFactory` stores the bound logger as an instance attribute (`self.bound_logger`), which keeps a reference chain alive. Furthermore, when a deferrable operator re-defers (e.g., defer → `TimeDeltaTrigger` fires → task resumes → checks condition → defers again), each cycle creates a **new trigger ID** in the Airflow metadata DB. The triggerer opens a new `TriggerLoggingFactory` for each new trigger ID ([line 666](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L666)), resulting in **multiple open file descriptors pointing to the same log file** — one per re-defer cycle. **Observed impact after ~3 hours of uptime with ~2-26 concurrent triggers:** | Metric | Value | |--------|-------| | Total open FDs (PID of supervisor) | 2,140 | | Trigger log FDs | 2,128 | | Unique trigger log files | 589 | | Average FDs per unique file | ~3.6 | | Process RSS | 1,688 MB | | Heap segment size | 2,197 MB | | FD growth rate | ~7/min (accelerating) | | Trigger log file sizes on disk | 6.5 KB each (3.3 MB total) | The process heap (`/proc/PID/smaps`) is 99% anonymous memory — Python file objects and their write buffers retained on the heap due to the unclosed file handles. ### Root cause [`TriggerLoggingFactory.__call__`](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L293) opens a file via `log_file.open("wb")` ([line 303](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L303)) and wraps it in a structlog logger. The resulting file object is stored in `self.bound_logger` ([line 305](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L305)). When a trigger completes, the cleanup path in `TriggererJobRunner._handle_runner_msg` ([line 412](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L412)) pops the factory from `self.logger_cache` and calls `factory.upload_to_remote()`, but **never closes the underlying file handle**. The `TriggerLoggingFactory` class has no `close()` or `__del__` method, and structlog's logger wrappers do not close their file objects on deletion. Additionally, [`cleanup_finished_triggers`](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L993) in the TriggerRunner child process removes triggers from `self.triggers` ([line 1011](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L1011)) but has no interaction with the log file handles — those are managed in the parent supervisor process. ### What you think should happen instead `TriggerLoggingFactory` should have an explicit `close()` method that closes the underlying file object: ```python def close(self): if hasattr(self, "bound_logger"): underlying = self.bound_logger._logger # structlog's wrapped logger if hasattr(underlying, '_file'): underlying._file.close() elif hasattr(underlying, '_write'): # BytesLogger stores the write method pass ``` And the cleanup path at [line 412](https://github.com/apache/airflow/blob/3.1.8/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L412) should call it: ```python if factory := self.logger_cache.pop(id, None): factory.upload_to_remote() factory.close() # Explicitly close the file handle ``` ### How to reproduce 1. Deploy Airflow 3.1.8 with the triggerer enabled (KubernetesExecutor, 1 triggerer replica) 2. Create a deferrable operator that uses a `TimeDeltaTrigger` re-defer polling loop (defer → fire → resume → check → defer again). Even a simple loop with 10-second intervals works. 3. Trigger several DAG runs concurrently 4. Monitor the triggerer supervisor process's open file descriptors: ```bash # From inside the triggerer container: ls /proc/<supervisor_pid>/fd | wc -l ls -la /proc/<supervisor_pid>/fd | grep "trigger" | wc -l ``` 5. Observe FDs growing monotonically with each trigger lifecycle event and never decreasing 6. Monitor RSS via `kubectl top pod` or `/proc/<pid>/status` — memory grows in step-function jumps correlated with batches of trigger completions ### Operating System Debian GNU/Linux 12 (bookworm) — AKS (Kubernetes 1.34.4) ### Versions of Apache Airflow Providers apache-airflow-providers-standard==1.10.0 ### Deployment Official Apache Airflow Helm Chart ### Deployment details - Helm chart version: 1.17.0 - Executor: KubernetesExecutor - Triggerer replicas: 1 - Triggerer resource limits: none (no memory limit set) - Platform: Azure Kubernetes Service (AKS) - Python: 3.12 ### Anything else? This bug is amplified by deferrable operators that use a re-defer polling pattern (e.g., polling a remote job status every N seconds via `TimeDeltaTrigger`). Each re-defer cycle creates a new trigger ID in the metadata DB, and the triggerer opens a new `TriggerLoggingFactory` for each ID. A single task that re-defers 100 times will leak ~100 file descriptors. In our production environment, we observed the triggerer process growing from ~300 MB to 8.7 GB over 5 days before node memory pressure caused alerts. The leak was initially misattributed to various other causes (kernel dentry leaks, liveness probes, concurrency settings) before being traced to the unclosed trigger log file handles. Related issues: - [#49887](https://github.com/apache/airflow/issues/49887) — dag-processor exits with too many open files (same pattern, different component, fixed in 3.0.2) - [#42195](https://github.com/apache/airflow/issues/42195) — Triggerer memory increasing continuously (different root cause — kernel dentry leak from liveness probes) ### Are you willing to submit PR? - [ ] 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) ### What you think should happen instead? _No response_ ### Operating System Debian GNU/Linux 12 (bookworm) ### Deployment Official Apache Airflow Helm Chart ### Apache Airflow Provider(s) _No response_ ### Versions of Apache Airflow Providers _No response_ ### Official Helm Chart version 1.17.0 ### Kubernetes Version _No response_ ### Helm Chart configuration _No response_ ### Docker Image customizations _No response_ ### Anything else? _No response_ ### Are you willing to submit PR? - [ ] 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]
