ferruzzi commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4497333610
##########
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:
Do we want to validate the upload before unlinking, or is that covered
elsewhere? It looks like there is a chance we lose files, but maybe it's an
edge case. I see you have `kept` down on/around line 1208 which seems like
it's doing that job in a different branch and we're not doing it here, but
maybe there's a reason.
##########
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:
This one comes from Claude so may be ignorable, but seems legit:
A dead triggerer's directory whose upload succeeded returns `kept=False` and
never reaches `_remove_old_trigger_log_dirs`, and since `delete_local_logs`
defaults to `False` (config.yml:841) the handler empties the file without
removing its parent, leaving one empty directory per trigger per dead job
forever on the default S3/GCS/WASB path, which also slows every later
base.glob("triggers/*/*") scan.
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3189,6 +3189,35 @@ triggerer:
type: integer
example: ~
default: "30"
Review Comment:
Nit: The naming is inconsistent for these three new values, should they
maybe standardize or is it intentional (watcher_logs* vs trigger_logs*)? It's a
pain to change config names once they go live.
--
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]
