kaxil commented on code in PR #69523:
URL: https://github.com/apache/airflow/pull/69523#discussion_r3677710359
##########
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:
`_open_sockets` also holds the `requests` socket, so this drains that one
too. Its handler is `length_prefixed_frame_reader(self.handle_requests(log))`,
so any buffered frame gets dispatched into `_handle_request`, which calls out
to the API (`GetConnection`, `PutVariable`, `DeleteVariable`, ...) and then
`send_msg` writes the response into the socket of a process we just SIGKILLed.
A DAG file killed for timing out could still have its `Variable.set` land.
Only stdout/stderr/logs feed `logger_filehandle`, so the requests socket
only needs the unregister and the close. The channel type is already recorded
as the `_open_sockets` value if you want to filter on it.
##########
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:
Would this fit better on `DagFileProcessorProcess`, called from `close()`?
Right now core reaches into `processor._open_sockets` and re-derives the
`key.data == (handler, on_close)` contract that `_service_processor_sockets`
and `_service_subprocess` already own, and those live in a separately released
distribution (core pins `apache-airflow-task-sdk>=1.4.0,<1.5.0`).
Doing it inside `close()` would also make the drain-before-close ordering
structural, rather than something each kill path has to remember to pair up.
`_collect_results` would be unaffected, since `is_ready` already requires
`_open_sockets` to be empty.
##########
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:
The message text depends on the handle type: `open(path, "ab")` gives a
`BufferedWriter` whose message is `write to closed file`, but a text-mode or
unbuffered handle raises `I/O operation on closed file`. That couples the guard
to `_get_logger_for_dag_file` opening with `"ab"`. `"closed file" in str(e)`
covers all the CPython variants and keeps the narrowing intact.
A `debug` log would help too, right now the dropped lines leave no trace.
##########
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:
This only guards the `logs` channel. stdout and stderr are registered
through `_create_log_forwarder` -> `forward_to_log`, which does the same
unguarded `log.log(level, msg, logger=logger)` on the same `process_log`, so a
stale stdout event after `logger_filehandle.close()` raises the identical
`ValueError` and still takes the job down. Worth putting the guard somewhere
both paths share.
--
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]
