seanghaeli commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4556284095
##########
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():
+ 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 = {
+ job_id: [d for d in dirs if
self._upload_orphaned_trigger_log_dir(handler, base, d, job_id)]
+ for job_id, dirs in dead_dirs.items()
+ }
+ gone = {str(trigger_id) for trigger_id in trigger_ids if trigger_id
not in existing}
+ self._remove_old_trigger_log_dirs(dead_dirs, gone=gone)
+
+ @staticmethod
+ def _upload_orphaned_trigger_log_dir(handler, base: Path, producer_dir:
Path, job_id: int) -> bool:
+ """Upload a dead triggerer's files unless another triggerer is; True
if the handler left any."""
+ # Claimed first: triggerers starting together on a shared log folder
see the same files, and remote
+ # handlers append, so a file uploaded twice would be in the remote
copy twice.
+ claim = producer_dir / _ORPHAN_UPLOAD_CLAIM
+ with suppress(OSError):
+ if time.time() - claim.stat().st_mtime >=
_ORPHAN_UPLOAD_CLAIM_MAX_AGE:
+ claim.unlink() # left by a triggerer that died while uploading
+ try:
+ os.close(os.open(claim, os.O_CREAT | os.O_EXCL | os.O_WRONLY))
+ except OSError:
+ return False
+ kept = left = False
+ try:
+ # The rotated files first: remote handlers append in upload order.
+ for path in trigger_log_files(producer_dir):
+ # Remote handlers empty a file they uploaded unless they
delete it; skip those.
+ if not path.is_file() or not path.stat().st_size:
+ continue
+ relative_path = path.relative_to(base).as_posix()
+ try:
+ handler.upload(relative_path, None)
+ except Exception:
+ log.exception("Failed to upload an orphaned trigger log",
path=relative_path)
+ left = True
+ continue
+ if path.is_file() and path.stat().st_size:
+ # Neither emptied nor deleted: a handler that keeps no
copy of it, or a failed upload, whose
+ # files stay for a later start.
+ kept = kept or _handler_keeps_no_copy(handler)
Review Comment:
Good catch, it's the same gap Dennis found for rotated files
([above](https://github.com/apache/airflow/pull/76527#discussion_r4499015135)),
so the orphan scan now uses the same rule: only S3, GCS and WASB count leftover
lines as a failed upload, so HDFS and OSS get each orphaned file once. #76640
is the real fix, so the triggerer won't have to guess.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -317,6 +329,8 @@ 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)
+ # After kill(): the runner's last lines have been drained into
the files by now.
Review Comment:
Yes, dropped on purpose: draining here hangs while a trigger's child keeps
writing. I'm going to leave the drain to #76514, and closing #76867 as a
follow-up will be the long-term fix. The comment now says the last lines aren't
read yet.
--
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]
