seanghaeli commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4556328203
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -835,6 +1076,308 @@ def _handle_request(self, msg: ToTriggerSupervisor, log:
FilteringBoundLogger, r
self.send_msg(resp, request_id=req_id, error=None, **dump_opts)
+ @staticmethod
+ def get_watcher_log_max_open_files() -> int:
+ """
+ Return ``[triggerer] watcher_log_max_open_files`` (0: no limit), at
most a quarter of ``ulimit -n``.
+
+ The task and callback trigger logs, the database and the sockets need
descriptors too.
+ """
+ setting = conf.getint("triggerer", "watcher_log_max_open_files")
+ try:
+ import resource # Unix only
+
+ soft, _ = resource.getrlimit(resource.RLIMIT_NOFILE)
+ except (ImportError, OSError, ValueError):
+ return setting
+ if soft == resource.RLIM_INFINITY:
+ return setting
+ limit = max(1, soft // 4)
+ return limit if setting <= 0 else min(setting, limit)
+
+ def _note_watcher_log_write(self, trigger_id: int) -> None:
+ """Keep at most ``watcher_log_max_open_files`` watcher files open,
closing the least recent."""
+ self._open_watcher_logs.pop(trigger_id, None)
+ self._open_watcher_logs[trigger_id] = None
+ while 0 < self.watcher_log_max_open_files <
len(self._open_watcher_logs):
+ idle = next(iter(self._open_watcher_logs))
+ del self._open_watcher_logs[idle]
+ if (factory := self.logger_cache.get(idle)) is not None:
+ factory.close()
+
+ def _close_trigger_logger(self, trigger_id: int) -> None:
+ self._open_watcher_logs.pop(trigger_id, None)
+ if not (factory := self.logger_cache.pop(trigger_id, None)):
+ return
+ if not isinstance(factory, WatcherLoggingFactory) or self.stop:
+ self._upload_and_close(trigger_id, factory)
+ return
+ # In the background: a triggerer that lost its watchers (a DB
partition) releases them all at once.
+ if self._upload_pool is None:
+ self._upload_pool = ThreadPoolExecutor(max_workers=4,
thread_name_prefix="trigger-log-release")
+ upload = self._upload_pool.submit(self._upload_and_close, trigger_id,
factory)
+ self._pending_uploads[trigger_id] = upload
+ self._released_factories[trigger_id] = factory
+
+ def forget(done: Future) -> None:
+ if self._pending_uploads.get(trigger_id) is done:
+ self._pending_uploads.pop(trigger_id, None)
+ self._released_factories.pop(trigger_id, None)
+
+ upload.add_done_callback(forget)
+
+ def _rotated_upload_done(self, trigger_id: int) -> bool:
+ upload = self._pending_rotated_uploads.get(trigger_id)
+ if upload is not None and not upload.done():
+ return False
+ if self._rotated_pool is not None and (path :=
self._rotated_unconfirmed.pop(trigger_id, None)):
+ # Now, while it is the only file in its directory: with
delete_local_logs, a handler deletes the
+ # directory of the file it uploaded. Replaced by the next rotation
whether it goes through or not.
+ self._pending_rotated_uploads[trigger_id] =
self._rotated_pool.submit(
+ self._upload_rotated_again, trigger_id, path
+ )
+ return False
+ return True
+
+ @staticmethod
+ def _upload_rotated_again(trigger_id: int, path: str) -> None:
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ try:
+ if (handler := load_remote_log_handler()) is not None:
+ handler.upload(path, None)
+ except Exception:
+ log.exception("Failed to upload a rotated trigger log",
trigger_id=trigger_id, path=path)
+
+ def _upload_rotated(self, trigger_id: int, log_path: str) -> None:
+ # Called by the line that rotated the file: the rest runs in the pool,
not to hold up the runner's logs.
+ if self._rotated_pool is None:
+ self._rotated_pool = ThreadPoolExecutor(max_workers=4,
thread_name_prefix="trigger-log-rotated")
+ rotated = trigger_log_rotated_path(log_path)
+ # Numbered here: a watcher's next rotation waits for this upload (see
_rotated_upload_done).
+ part = self._rotated_parts.get(trigger_id, 0) + 1
+
+ def upload() -> None:
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ path = rotated
+ base = Path(conf.get("logging", "base_log_folder"))
+ try:
+ if (handler := load_remote_log_handler()) is None:
+ return
+ if not _handler_keeps_no_copy(handler):
+ # A handler that stores files gets each rotated file as
its own object (see
+ # trigger_log_part_path), numbered by this triggerer, the
only one writing to the directory.
+ # Counted, not read from the disk: a watcher that comes
back gets a new file, and the parts
+ # it uploaded before are gone locally.
+ for done in (base / rotated).parent.glob("*.log"):
+ if done.name != TRIGGER_LOG_FILE_NAME:
+ with suppress(OSError):
+ done.unlink() # the previous one, uploaded or
tried twice: one at a time
+ path = trigger_log_part_path(log_path, part)
+ # Named only now: until then the triggerer's log server
serves it as the rotated file.
+ os.replace(base / rotated, base / path)
+ handler.upload(path, None)
+ raised = False
+ except Exception:
+ log.exception("Failed to upload a rotated trigger log",
trigger_id=trigger_id, path=path)
+ raised = True
+ if path != rotated and _has_lines(base / path):
+ # Sent again if the upload raised, or if S3, GCS or WASB left
lines (others may keep them).
+ if (raised and not _handler_is(handler, _APPENDING_HANDLERS))
or (
+ not raised and _handler_is(handler,
_EMPTIES_STORED_FILE_HANDLERS)
+ ):
+ self._rotated_unconfirmed[trigger_id] = path
+
+ self._pending_rotated_uploads[trigger_id] =
self._rotated_pool.submit(upload)
+ self._rotated_parts[trigger_id] = part
+
+ def _upload_and_close(self, trigger_id: int, factory:
TriggerLoggingFactory) -> None:
+ if (rotated_upload := self._pending_rotated_uploads.pop(trigger_id,
None)) is not None:
+ rotated_upload.result() # else the rotated file could be read
again before it is emptied
+ self._rotated_unconfirmed.pop(trigger_id, None) # sent again below,
with the live file
+ try:
+ factory.upload_to_remote()
+ except Exception:
+ log.exception("Failed to upload trigger logs to remote",
trigger_id=trigger_id)
+ finally:
+ # Close even if the upload failed, not to leak the descriptor.
Closing flushes, which can fail.
+ try:
+ factory.close()
+ except OSError as e:
+ self._warn_trigger_log_error(trigger_id, e)
+
+ def _warn_trigger_log_error(self, trigger_id: int | None, error: OSError)
-> None:
+ now = time.monotonic()
+ if now - self._trigger_log_errors_at.get(trigger_id or 0, -math.inf)
>= 60:
+ self._trigger_log_errors_at[trigger_id or 0] = now
+ log.warning(
+ "Could not write a trigger's log file; its lines go to the
triggerer's output meanwhile",
+ trigger_id=trigger_id,
+ error=str(error),
+ )
+
+ def handle_orphaned_trigger_logs(self) -> None:
+ """
+ Upload, or remove, the asset watcher files left on this disk by
triggerers that are no longer alive.
+
+ A killed triggerer uploads nothing, so its files are uploaded now,
then their directories removed.
+ Files remote logging doesn't take (it is off, or the handler keeps no
copy, like Elasticsearch) are
+ removed instead: all but the most recently written for each watcher,
and all for triggers that no
+ longer exist. Files of live triggerers, this one included, of
triggerers without a job (no job id
+ directory), and those a failed upload left stay.
+ """
+ from airflow.jobs.job import Job
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ handler = load_remote_log_handler()
+ base = Path(conf.get("logging", "base_log_folder"))
+ own = str(self.job.id) if self.job is not None else None
+ dirs_by_job: dict[int, list[Path]] = {}
+ for producer_dir in base.glob(f"{TRIGGER_LOG_ROOT}/*/*"):
+ if producer_dir.name.isdigit() and producer_dir.name != own and
producer_dir.is_dir():
Review Comment:
Good idea, added. I also named the directories
`trigger_id=<id>/triggerer_job_id=<id>`, like the default task log template, so
the scan can't match a task's log directory even with a custom template.
--
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]
