hkc-8010 commented on code in PR #69523:
URL: https://github.com/apache/airflow/pull/69523#discussion_r3699989706
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
Review Comment:
Moved the closed-handle guard into shared `_log_to_target(...)` in
`da59ecaea6` and reused it from both structured log processing and
`forward_to_log()`, so stdout/stderr now go through the same suppression path
as the `logs` channel.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
+ except ValueError as e:
+ if "write to closed file" not in str(e):
Review Comment:
Updated the check to `"closed file" in str(e)` and added a debug breadcrumb
when a line is dropped in `da59ecaea6`. I also added coverage for both
`ValueError("write to closed file")` and `ValueError("I/O operation on closed
file")` variants.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1163,6 +1164,42 @@ def remove_orphaned_file_stats(self, present:
set[DagFileInfo]):
for file in stats_to_remove:
del self._file_stats[file]
+ def _deregister_processor_sockets(self, processor:
DagFileProcessorProcess) -> None:
+ """
+ Drain buffered log data from all sockets of a killed processor, then
close them.
+
+ After SIGKILL the OS closes the subprocess write ends; kernel buffers
may still hold
+ unread data. Reading and processing that data here prevents log lines
from being lost.
+ """
+ for sock in list(processor._open_sockets.keys()):
Review Comment:
Addressed in `da59ecaea6`: post-kill cleanup now lives in task-sdk, and
`cleanup_sockets_after_kill()` only drains `stdout` / `stderr` / `logs`. The
`requests` socket is now only unregistered/closed, so buffered request frames
are no longer dispatched after SIGKILL. I also added a regression covering that
path.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1163,6 +1164,42 @@ def remove_orphaned_file_stats(self, present:
set[DagFileInfo]):
for file in stats_to_remove:
del self._file_stats[file]
+ def _deregister_processor_sockets(self, processor:
DagFileProcessorProcess) -> None:
+ """
+ Drain buffered log data from all sockets of a killed processor, then
close them.
+
+ After SIGKILL the OS closes the subprocess write ends; kernel buffers
may still hold
+ unread data. Reading and processing that data here prevents log lines
from being lost.
+ """
+ for sock in list(processor._open_sockets.keys()):
Review Comment:
Addressed in `da59ecaea6`: post-kill cleanup now lives in task-sdk, and
`cleanup_sockets_after_kill()` only drains `stdout` / `stderr` / `logs`. The
`requests` socket is now only unregistered/closed, so buffered request frames
are no longer dispatched after SIGKILL. I also added a regression covering that
path.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
+ except ValueError as e:
+ if "write to closed file" not in str(e):
Review Comment:
Updated the check to `"closed file" in str(e)` and added a debug breadcrumb
when a line is dropped in `da59ecaea6`. I also added coverage for both
`ValueError("write to closed file")` and `ValueError("I/O operation on closed
file")` variants.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1163,6 +1164,42 @@ def remove_orphaned_file_stats(self, present:
set[DagFileInfo]):
for file in stats_to_remove:
del self._file_stats[file]
+ def _deregister_processor_sockets(self, processor:
DagFileProcessorProcess) -> None:
+ """
+ Drain buffered log data from all sockets of a killed processor, then
close them.
+
+ After SIGKILL the OS closes the subprocess write ends; kernel buffers
may still hold
+ unread data. Reading and processing that data here prevents log lines
from being lost.
+ """
+ for sock in list(processor._open_sockets.keys()):
Review Comment:
Addressed in `da59ecaea6`: post-kill cleanup now lives in task-sdk, and
`cleanup_sockets_after_kill()` only drains `stdout` / `stderr` / `logs`. The
`requests` socket is now only unregistered/closed, so buffered request frames
are no longer dispatched after SIGKILL. I also added a regression covering that
path.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
+ except ValueError as e:
+ if "write to closed file" not in str(e):
Review Comment:
Updated the check to `"closed file" in str(e)` and added a debug breadcrumb
when a line is dropped in `da59ecaea6`. I also added coverage for both
`ValueError("write to closed file")` and `ValueError("I/O operation on closed
file")` variants.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
Review Comment:
Moved the closed-handle guard into shared `_log_to_target(...)` in
`da59ecaea6` and reused it from both structured log processing and
`forward_to_log()`, so stdout/stderr now go through the same suppression path
as the `logs` channel.
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2350,7 +2350,12 @@ def process_log_messages_from_subprocess(
if level := NAME_TO_LEVEL.get(event.pop("level")):
msg = event.pop("event", None)
for target in loggers:
- target.log(level, msg, **event)
+ try:
+ target.log(level, msg, **event)
Review Comment:
Moved the closed-handle guard into shared `_log_to_target(...)` in
`da59ecaea6` and reused it from both structured log processing and
`forward_to_log()`, so stdout/stderr now go through the same suppression path
as the `logs` channel.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1163,6 +1164,42 @@ def remove_orphaned_file_stats(self, present:
set[DagFileInfo]):
for file in stats_to_remove:
del self._file_stats[file]
+ def _deregister_processor_sockets(self, processor:
DagFileProcessorProcess) -> None:
Review Comment:
Moved this into `DagFileProcessorProcess.close()` on the task-sdk side in
`da59ecaea6`, so core now just does `kill(...); close()` and no longer reaches
into `_open_sockets` or re-derives the selector callback contract.
`_collect_results` behavior stays 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]
