seanghaeli commented on code in PR #76514:
URL: https://github.com/apache/airflow/pull/76514#discussion_r4552595700
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -317,6 +317,10 @@ def _execute(self) -> int | None:
# to force-kill here. trigger_runner may be None if start() raised.
if self.trigger_runner is not None:
self.trigger_runner.kill(escalation_delay=10, force=True)
+ # kill() can return with log lines still unread, e.g. the
"trigger completed" lines the
+ # runner writes as it cancels its triggers, so drain the
sockets before uploading.
+ self.trigger_runner.cleanup_sockets_after_kill()
Review Comment:
Found this in testing: if a trigger started a process that keeps writing to
stdout (a CLI run without capturing its output, say), this drain never ends.
The child inherits the runner's stdout and outlives it, so the socket never
runs empty. With a `yes` child in Breeze, a SIGTERM never finished and nothing
was uploaded, and after a kill -9 of the runner the triggerer never exited and
its heartbeat went stale. main exits in under a second in both. Shutting the
sockets for reading first makes the drain stop at what's already queued:
```suggestion
self.trigger_runner.stop_taking_output()
self.trigger_runner.cleanup_sockets_after_kill()
```
```python
def stop_taking_output(self) -> None:
for sock in list(self._open_sockets):
with suppress(OSError):
sock.shutdown(SHUT_RD)
```
With that, both cases exit in under half a second, and the uploaded file has
every line the runner wrote, including "trigger completed" on a stop. #76867
has the details.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -835,6 +832,31 @@ def _handle_request(self, msg: ToTriggerSupervisor, log:
FilteringBoundLogger, r
self.send_msg(resp, request_id=req_id, error=None, **dump_opts)
+ @staticmethod
+ def _upload_trigger_log(
+ trigger_id: int, factory: TriggerLoggingFactory, log:
FilteringBoundLogger
+ ) -> None:
+ try:
+ factory.upload_to_remote()
+ except Exception:
+ log.exception("Failed to upload trigger logs to remote",
trigger_id=trigger_id)
+ finally:
+ # Close the FD explicitly even if upload raised, otherwise the file
+ # handle leaks for every failed upload.
+ factory.close()
+
+ def upload_remaining_trigger_logs(self) -> None:
+ """
+ Upload the logs of triggers that never finished here, for use once the
runner has exited.
+
+ A trigger's log is otherwise only uploaded when the runner reports it
finished. Triggers that
+ are still running when the triggerer shuts down never report that, so
without this what they
+ logged here would be lost with the local disk.
+ """
+ while self.logger_cache:
Review Comment:
On the thread pool question in the description, I think it's worth it. With
1000 triggers (the default capacity) and a remote handler that takes 50 ms per
upload, the stop took 51 s here, against the chart's 60 s
`terminationGracePeriodSeconds`. A slower store or a few more triggers and the
SIGKILL cuts the rest off. #76527 has a 16-thread version of this loop you
could reuse.
--
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]
