You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "wjddn279 (via GitHub)" <gi...@apache.org> on 2026/02/18 03:39:01 UTC
[PR] fix triggerer logger's file descriptor closed when it removed [airflow]
wjddn279 opened a new pull request, #62103:
URL: https://github.com/apache/airflow/pull/62103
closed: https://github.com/apache/airflow/issues/61916
An issue was reported where file descriptors were not being properly closed when the logger was destroyed in the existing triggerer.
As shown below, the number of open file descriptors was continuously increasing before this fix. After applying the logic in this PR, file descriptors are now properly closed and the count decreases as expected.
### AS-IS (3.1.7)
<img width="518" height="162" alt="image" src="https://github.com/user-attachments/assets/b3272105-0403-4365-a4aa-993d1f348017" />
<img width="517" height="82" alt="image" src="https://github.com/user-attachments/assets/3d404aef-e042-4089-aac8-5a374a98e470" />
### TO-BE (patched)
<img width="514" height="160" alt="image" src="https://github.com/user-attachments/assets/75f33c5c-12cf-4a7f-80bf-d507e383aaef" />
<!-- SPDX-License-Identifier: Apache-2.0 https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst` or `{issue_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2886721636
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -310,13 +312,20 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
pretty_logs = False
if pretty_logs:
- underlying_logger: WrappedLogger = structlog.WriteLogger(log_file.open("w", buffering=1))
+ self._filehandle = log_file.open("w", buffering=1)
+ underlying_logger: WrappedLogger = structlog.WriteLogger(self._filehandle)
else:
- underlying_logger = structlog.BytesLogger(log_file.open("wb"))
+ self._filehandle = log_file.open("wb")
+ underlying_logger = structlog.BytesLogger(self._filehandle)
logger = structlog.wrap_logger(underlying_logger, processors=processors).bind()
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if hasattr(self, "_filehandle") and self._filehandle:
+ self._filehandle.close()
+
def upload_to_remote(self):
from airflow.sdk.log import upload_to_remote
Review Comment:
Should also add a `not self._filehandle.closed` guard here — if `close()` is ever called explicitly before GC runs `__del__`, this would attempt to double-close. It's a no-op for CPython files but the guard is good practice.
##########
airflow-core/tests/unit/jobs/test_triggerer_job.py:
##########
@@ -303,6 +303,29 @@ def test_trigger_log(mock_monotonic, trigger, watcher_count, trigger_count, sess
trigger_runner_supervisor.kill(force=False)
+def test_trigger_logger_fd_closed_when_removed(session):
+
+ trigger = TimeDeltaTrigger(datetime.timedelta(seconds=0.5))
+
+ create_trigger_in_db(session, trigger)
+
+ mock_file = MagicMock()
+ mock_file.closed = False
+
+ with patch("airflow.sdk.log.init_log_file") as mock_init_log_file:
+ mock_init_log_file.return_value.open.return_value = mock_file
+
+ trigger_runner_supervisor = TriggerRunnerSupervisor.start(job=Job(id=123456), capacity=10)
+ trigger_runner_supervisor.load_triggers()
+
+ for _ in range(30):
+ trigger_runner_supervisor._service_subprocess(0.1)
+
+ mock_file.close.assert_called_once()
+
Review Comment:
This assertion depends on `__del__` being triggered by GC during the loop, which is non-deterministic across Python implementations. If you switch to an explicit `close()` method (called from `_handle_request`), this test becomes deterministic — you're verifying the cleanup lifecycle, not GC timing.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -310,13 +312,20 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
pretty_logs = False
if pretty_logs:
- underlying_logger: WrappedLogger = structlog.WriteLogger(log_file.open("w", buffering=1))
+ self._filehandle = log_file.open("w", buffering=1)
+ underlying_logger: WrappedLogger = structlog.WriteLogger(self._filehandle)
else:
- underlying_logger = structlog.BytesLogger(log_file.open("wb"))
+ self._filehandle = log_file.open("wb")
+ underlying_logger = structlog.BytesLogger(self._filehandle)
logger = structlog.wrap_logger(underlying_logger, processors=processors).bind()
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if hasattr(self, "_filehandle") and self._filehandle:
+ self._filehandle.close()
+
Review Comment:
`__del__` is not guaranteed to run promptly (or at all during interpreter shutdown / reference cycles), and exceptions inside it are silently swallowed. This is the exact same class of problem the DAG processor had, solved in #47574 with an explicit `close()` method + explicit call at the cleanup site.
I'd prefer we follow the same pattern here for consistency and reliability:
```suggestion
def close(self):
"""Explicitly close the underlying log file handle."""
if hasattr(self, "_filehandle") and self._filehandle and not self._filehandle.closed:
self._filehandle.close()
```
And then call `factory.close()` after `factory.upload_to_remote()` in `_handle_request`.
If you want to keep `__del__` as a safety net that's fine, but the primary cleanup path should be the explicit call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-3920777185
Can you add a test to avoid regression?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-3925660959
@eladkal Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2828337316
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -319,6 +318,17 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if raw_logger := getattr(self.bound_logger, "_logger", None):
+ file_handle = getattr(raw_logger, "_file", None)
+ else:
+ return
+
+ if file_handle and not file_handle.closed:
+ file_handle.flush()
+ file_handle.close()
Review Comment:
> In short something like this, where we store the handler and clear it to avoid diversion from that approach?
I agree of this and it looks better! Thanks!
I've thought about that situation and approach as well, but if we separate it with such an explicit method `close`, that method must always be called after `upload_to_remote` is invoked — which could lead to mistakes in future development.
What do you think about doing a try-catch in `__del__`? I actually considered that beforehand too, but I was worried it could cause a silent fd leak if it fails.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-3996500299
@amoghrajesh
Can I get more review about this?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2828337316
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -319,6 +318,17 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if raw_logger := getattr(self.bound_logger, "_logger", None):
+ file_handle = getattr(raw_logger, "_file", None)
+ else:
+ return
+
+ if file_handle and not file_handle.closed:
+ file_handle.flush()
+ file_handle.close()
Review Comment:
I've thought about that situation and approach as well, but if we separate it with such an explicit method `close`, that method must always be called after `upload_to_remote` is invoked — which could lead to mistakes in future development.
What do you think about doing a try-catch in `__del__`? I actually considered that beforehand too, but I was worried it could cause a silent fd leak if it fails.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil merged PR #62103:
URL: https://github.com/apache/airflow/pull/62103
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-4014916153
@tirkarthi Do you mind taking a look as well, please?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "amoghrajesh (via GitHub)" <gi...@apache.org>.
amoghrajesh commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2828166394
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -319,6 +318,17 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
self.bound_logger = logger
return logger
+ def __del__(self):
Review Comment:
The one concern I mainly have with using `del` is that if an exception occurs during cleanup, it will quietly exit
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -319,6 +318,17 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if raw_logger := getattr(self.bound_logger, "_logger", None):
+ file_handle = getattr(raw_logger, "_file", None)
+ else:
+ return
+
+ if file_handle and not file_handle.closed:
+ file_handle.flush()
+ file_handle.close()
Review Comment:
Can we handle it similar to how it is done for DAG processor: https://github.com/apache/airflow/pull/47574
In short something like this, where we store the handler and clear it to avoid diversion from that approach?
```diff
diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py
--- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py (revision 10cd08dff8916b93f8c3f94bc34265bb7544fde4)
+++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py (date 1771510381992)
@@ -30,7 +30,7 @@
from datetime import datetime
from socket import socket
from traceback import format_exception
-from typing import TYPE_CHECKING, Annotated, Any, ClassVar, Literal, TypedDict
+from typing import IO, TYPE_CHECKING, Annotated, Any, ClassVar, Literal, TypedDict
import anyio
import attrs
@@ -302,6 +302,8 @@
bound_logger: WrappedLogger = attrs.field(init=False, repr=False)
+ _filehandle: IO[Any] = attrs.field(init=False, repr=False)
+
def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedLogger:
if hasattr(self, "bound_logger"):
return self.bound_logger
@@ -312,13 +314,20 @@
pretty_logs = False
if pretty_logs:
- underlying_logger: WrappedLogger = structlog.WriteLogger(log_file.open("w", buffering=1))
+ self._filehandle = log_file.open("w", buffering=1)
+ underlying_logger: WrappedLogger = structlog.WriteLogger(self._filehandle)
else:
- underlying_logger = structlog.BytesLogger(log_file.open("wb"))
+ self._filehandle = log_file.open("wb")
+ underlying_logger = structlog.BytesLogger(self._filehandle)
logger = structlog.wrap_logger(underlying_logger, processors=processors).bind()
self.bound_logger = logger
return logger
+ def close(self):
+ """Explicitly close the underlying log file handle."""
+ if hasattr(self, "_filehandle"):
+ self._filehandle.close()
+
def upload_to_remote(self):
from airflow.sdk.log import upload_to_remote
@@ -421,10 +430,9 @@
for id in msg.finished or ():
self.running_triggers.discard(id)
self.cancelling_triggers.discard(id)
# Remove logger from the cache, and since structlog doesn't have an explicit close method, we
# only need to remove the last reference to it to close the open FH
if factory := self.logger_cache.pop(id, None):
factory.upload_to_remote()
+ factory.close()
response = messages.TriggerStateSync(
to_create=[],
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2886721618
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -310,13 +312,20 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
pretty_logs = False
if pretty_logs:
- underlying_logger: WrappedLogger = structlog.WriteLogger(log_file.open("w", buffering=1))
+ self._filehandle = log_file.open("w", buffering=1)
+ underlying_logger: WrappedLogger = structlog.WriteLogger(self._filehandle)
else:
- underlying_logger = structlog.BytesLogger(log_file.open("wb"))
+ self._filehandle = log_file.open("wb")
+ underlying_logger = structlog.BytesLogger(self._filehandle)
logger = structlog.wrap_logger(underlying_logger, processors=processors).bind()
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if hasattr(self, "_filehandle") and self._filehandle:
+ self._filehandle.close()
+
Review Comment:
`__del__` is not guaranteed to run promptly (or at all during interpreter shutdown / reference cycles), and exceptions inside it are silently swallowed. This is the exact same class of problem the DAG processor had, solved in #47574 (by @tirkarthi ) with an explicit `close()` method + explicit call at the cleanup site.
I'd prefer we follow the same pattern here for consistency and reliability:
```suggestion
def close(self):
"""Explicitly close the underlying log file handle."""
if hasattr(self, "_filehandle") and self._filehandle and not self._filehandle.closed:
self._filehandle.close()
```
And then call `factory.close()` after `factory.upload_to_remote()` in `_handle_request`.
If you want to keep `__del__` as a safety net that's fine, but the primary cleanup path should be the explicit call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on code in PR #62103:
URL: https://github.com/apache/airflow/pull/62103#discussion_r2828337316
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -319,6 +318,17 @@ def __call__(self, processors: Iterable[structlog.typing.Processor]) -> WrappedL
self.bound_logger = logger
return logger
+ def __del__(self):
+ # Explicitly close the file descriptor when the logger is garbage collected.
+ if raw_logger := getattr(self.bound_logger, "_logger", None):
+ file_handle = getattr(raw_logger, "_file", None)
+ else:
+ return
+
+ if file_handle and not file_handle.closed:
+ file_handle.flush()
+ file_handle.close()
Review Comment:
> In short something like this, where we store the handler and clear it to avoid diversion from that approach?
I agree of this and it looks better! Thanks!
But if we separate it with such an explicit method `close`, that method must always be called after `upload_to_remote` is invoked — which could lead to mistakes in future development.
What do you think about doing a try-catch in `__del__`? I actually considered that beforehand too, but I was worried it could cause a silent fd leak if it fails.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-4011860303
@kaxil @amoghrajesh
Thank you for the review! I now have a better understanding of the side effects of using `__del__`. I've updated the code to explicitly close the file instead.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "wjddn279 (via GitHub)" <gi...@apache.org>.
wjddn279 commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-3921155506
sure. I'll add it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix triggerer logger's file descriptor closed when it removed [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #62103:
URL: https://github.com/apache/airflow/pull/62103#issuecomment-4001011011
One more thing — the comment on [this line](https://github.com/apache/airflow/blob/505d5eb4cb932f8e71cf0a41e39b8b5503a1b64e/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L423-L424) 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"*. This PR exists because that assumption was wrong. Please update the comment to reflect the new behavior once you add the explicit `close()` call.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org