seanghaeli commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4499015135
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -835,6 +1025,269 @@ 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] trigger_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", "trigger_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)
+ return upload is None or upload.done()
+
+ def _upload_rotated(self, trigger_id: int, log_path: str) -> None:
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ if (handler := load_remote_log_handler()) is None:
+ return
+ if self._rotated_pool is None:
+ self._rotated_pool = ThreadPoolExecutor(max_workers=4,
thread_name_prefix="trigger-log-rotated")
+ path = rotated = trigger_log_rotated_path(log_path)
+ base = Path(conf.get("logging", "base_log_folder"))
+ 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 here, not 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"):
Review Comment:
You're right, that could lose a part: if its upload failed, nothing retried
it, and the next rotation deleted it. Now it gets one more try first, and the
disk stays capped either way. Remote loggers HDFS and OSS can't tell us whether
an upload worked, so they skip the retry for now (#76640). The `kept` in the
orphan scan is a different thing: it flags handlers that keep no copy, like
Elasticsearch.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -835,6 +1025,269 @@ 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] trigger_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", "trigger_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)
+ return upload is None or upload.done()
+
+ def _upload_rotated(self, trigger_id: int, log_path: str) -> None:
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ if (handler := load_remote_log_handler()) is None:
+ return
+ if self._rotated_pool is None:
+ self._rotated_pool = ThreadPoolExecutor(max_workers=4,
thread_name_prefix="trigger-log-rotated")
+ path = rotated = trigger_log_rotated_path(log_path)
+ base = Path(conf.get("logging", "base_log_folder"))
+ 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 here, not 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, its upload done:
one rotated file at a time
+ self._rotated_parts[trigger_id] = part =
self._rotated_parts.get(trigger_id, 0) + 1
+ path = trigger_log_part_path(log_path, part)
+
+ def upload() -> None:
+ try:
+ if path != rotated:
+ # 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)
+ except Exception:
+ log.exception("Failed to upload a rotated trigger log",
trigger_id=trigger_id, path=path)
+
+ self._pending_rotated_uploads[trigger_id] =
self._rotated_pool.submit(upload)
+
+ 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
+ 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.
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():
+ dirs_by_job.setdefault(int(producer_dir.name),
[]).append(producer_dir)
+ if not dirs_by_job:
+ return
+ trigger_ids = sorted(
+ {int(d.parent.name) for ds in dirs_by_job.values() for d in ds if
d.parent.name.isdigit()}
+ )
+ with create_session() as session:
+ jobs = session.scalars(select(Job).where(Job.id.in_(dirs_by_job)))
+ dead = dirs_by_job.keys() - {job.id for job in jobs if
job.is_alive()}
+ # A watcher whose kwargs changed got a new trigger; the old id's
files have nothing left to keep.
+ existing = {
+ trigger_id
+ for ids in chunks(trigger_ids, _TRIGGER_ID_QUERY_CHUNK_SIZE)
+ for trigger_id in
session.scalars(select(Trigger.id).where(Trigger.id.in_(ids)))
+ }
+ dead_dirs = {job_id: dirs_by_job[job_id] for job_id in dead}
+ if handler is not None:
+ # What the handler left as it was is then removed as without
remote logging.
+ dead_dirs = {
Review Comment:
Ah, good catch! After a dead triggerer's logs are uploaded, their emptied
folders were left hanging around. They're now removed once the upload is 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
