ferruzzi commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4518288386
##########
airflow-core/docs/authoring-and-scheduling/event-scheduling.rst:
##########
@@ -493,6 +493,32 @@ If subscribers repeatedly overflow, there are two ways to
address this:
Both reduce the mismatch between producer throughput and per-subscriber
consume rate.
+.. _asset-watcher-logs:
+
+Watcher logs
+~~~~~~~~~~~~
+
+A deferred task's trigger logs to that task try's log, and a callback's
trigger to the callback's log. An asset
+watcher has no such log, so its lines go to a file of its own on the triggerer
that runs it,
+``triggers/<trigger id>/<triggerer job id>/trigger.log`` under ``[logging]
base_log_folder``. The file lasts as
+long as the trigger stays on that triggerer, across every restart of the
watcher, and each line keeps the
+watcher's ``trigger_id``. With ``[logging] remote_logging`` on, the remote log
handler that uploads task and
+callback logs also uploads the file when the trigger leaves the triggerer or
the triggerer stops, and a triggerer
+that starts uploads the files left on its disk by triggerers that died. A
triggerer that stops also uploads the
+task and callback logs of the triggers it was still running.
+
+``[triggerer] watcher_log_max_bytes`` caps the file, with one rotated file in
a ``rotated`` directory next to
+it (with remote logging on, each rotated file is uploaded before it is
replaced, as its own
+``rotated/trigger.<n>.log`` with a handler that stores files, such as S3),
+``[triggerer] watcher_log_max_open_files`` caps how many such files are open
at once, and
+``[triggerer] watcher_log_to_stdout`` keeps the watcher's lines in the
triggerer's output too, for platforms
+that collect it. The files of triggerers that are gone are uploaded by the
next triggerer on the same disk with
+remote logging on. Without it, or with a handler that keeps no copy of them
(Elasticsearch, OpenSearch), only the
+most recently written of them is kept for each watcher, and none for a trigger
that no longer exists, so a watcher
+uses at most four times ``watcher_log_max_bytes`` of one disk. A full or
read-only log disk doesn't stop the triggerer:
Review Comment:
Same as above. We can't really promise that anymore, and maybe need to
soften this so it doesn't sound like a hard rule?
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -578,6 +614,177 @@ def upload_to_remote(self):
upload_to_remote(self.bound_logger, self.ti)
+class _CappedBytesLogger(structlog.BytesLogger):
+ """
+ ``BytesLogger`` appending to ``path``, moved to the rotated directory once
it reaches ``max_bytes``.
+
+ ``may_rotate`` can hold a rotation off, ``on_rotate`` is called after one;
``max_bytes <= 0`` is no
+ cap. A closed or removed file (a log cleaner, ``delete_local_logs``, or on
a shared log folder another
+ triggerer's orphan scan) is reopened by the next line.
+ """
+
+ __slots__ = ("_max_bytes", "_may_rotate", "_on_rotate", "_path", "_size")
+
+ def __init__(
+ self,
+ path: Path,
+ max_bytes: int,
+ may_rotate: Callable[[], bool] | None = None,
+ on_rotate: Callable[[], None] | None = None,
+ ):
+ self._path, self._max_bytes = path, max_bytes
+ self._may_rotate, self._on_rotate = may_rotate, on_rotate
+ super().__init__(path.open("ab"))
+ self._size = self._file.tell()
+
+ @property
+ def rotated_path(self) -> Path:
+ return self._path.parent / TRIGGER_LOG_ROTATED_DIR / self._path.name
+
+ def ensure_open(self) -> None:
+ if not self._file.closed and self._unlinked():
+ # Lines written to a removed file are lost.
+ with suppress(OSError):
+ self._file.close()
+ if self._file.closed:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ self._file = self._path.open("ab")
+ self._write = self._file.write
+ self._flush = self._file.flush
+ self._size = self._file.tell()
+
+ def _unlinked(self) -> bool:
+ try:
+ return os.fstat(self._file.fileno()).st_nlink == 0
+ except (OSError, ValueError):
+ return False
+
+ def msg(self, message: bytes) -> None:
+ with self._lock:
+ try:
+ self.ensure_open()
+ self._write(message + b"\n")
+ self._flush()
+ self._size += len(message) + 1
+ if 0 < self._max_bytes <= self._size and (self._may_rotate is
None or self._may_rotate()):
+ self._file.close()
+ self.rotated_path.parent.mkdir(parents=True, exist_ok=True)
+ with suppress(FileNotFoundError):
+ os.replace(self._path, self.rotated_path)
+ self.ensure_open()
+ if self._on_rotate is not None:
+ try:
+ self._on_rotate()
+ except Exception:
+ # The line is written: the caller, reading the
runner's logs, must not see this.
+ log.exception(
+ "Failed to upload a rotated trigger log",
path=str(self.rotated_path)
+ )
+ except OSError:
+ # Reopened by the next line: a file removed meanwhile (to free
a full disk) frees its space.
+ with suppress(OSError):
+ self._file.close()
+ raise
+
+ def close(self) -> None:
+ # Closing flushes; on a full disk those lines are lost either way.
+ with suppress(OSError):
+ self._file.close()
+
+ log = debug = info = warn = warning = msg
+ fatal = failure = err = error = critical = exception = msg
+
+
[email protected](kw_only=True)
+class WatcherLoggingFactory(TriggerLoggingFactory):
+ """
+ An asset watcher's own log file on this triggerer, as its owner has no log.
+
+ It is kept while the trigger is assigned here, across restarts of the
watcher (SQS-style ones restart
+ after every event), so it is appended to, and capped on local disk. It is
uploaded with the remote log
+ handler task logs use, and every line keeps its ``trigger_id``.
+ """
+
+ max_bytes: int = 0
+ to_stdout: bool = True
+ # Told about every line, to cap how many of these files are open at once.
+ on_write: Callable[[int], None] | None = attrs.field(default=None,
repr=False)
+ may_rotate: Callable[[], bool] | None = attrs.field(default=None,
repr=False)
+ on_rotate: Callable[[], None] | None = attrs.field(default=None,
repr=False)
+
+ def _open(self, processors: Iterable[structlog.typing.Processor]) -> None:
+ if hasattr(self, "bound_logger"):
+ # Before the processors run: a handler streaming each line
(CloudWatch) names its stream after the
+ # open file, and would drop the line of a file the open-files cap
closed.
+ self.bound_logger._logger.ensure_open()
+ else:
+ from airflow.sdk.log import init_log_file # noqa: SDK001
+
+ underlying_logger = _CappedBytesLogger(
+ init_log_file(self.log_path), self.max_bytes, self.may_rotate,
self.on_rotate
+ )
+ self.bound_logger = structlog.wrap_logger(underlying_logger,
processors=processors).bind(
+ trigger_id=self.trigger_id
+ )
+ if self.on_write is not None and self.trigger_id is not None:
+ self.on_write(self.trigger_id)
+
+ def close(self):
+ if hasattr(self, "bound_logger"):
+ self.bound_logger._logger.close()
+
+ def upload_to_remote(self):
+ from airflow.sdk.log import load_remote_log_handler # noqa: SDK001
+
+ if not hasattr(self, "bound_logger") or (handler :=
load_remote_log_handler()) is None:
+ return
+ # Never upload an open file: remote handlers empty or delete it
afterwards. The rotated files first, as
+ # remote handlers append in upload order. Nor an empty one to a
handler that stores files: it would
+ # download and upload the remote copy again for nothing.
+ self.close()
+ live = self.bound_logger._logger._path
+ for path in trigger_log_files(live.parent):
+ if _handler_keeps_no_copy(handler) or (path.is_file() and
path.stat().st_size):
+ handler.upload((Path(self.log_path).parent /
path.relative_to(live.parent)).as_posix(), None)
+
+
+# Marks a dead triggerer's log directory that one triggerer is uploading (see
handle_orphaned_trigger_logs).
+_ORPHAN_UPLOAD_CLAIM = ".uploading"
+_ORPHAN_UPLOAD_CLAIM_MAX_AGE = 600.0
+_TRIGGER_ID_QUERY_CHUNK_SIZE = 500
+# Remote log handlers that take only task logs: a watcher's file is left as it
was however often it is sent.
+_TASK_LOGS_ONLY_HANDLERS = frozenset({"ElasticsearchRemoteLogIO",
"OpensearchRemoteLogIO"})
+# Remote log handlers that empty a file they stored and only log a failed
write: lines left mean it failed.
+_EMPTIES_STORED_FILE_HANDLERS = frozenset({"S3RemoteLogIO", "GCSRemoteLogIO",
"WasbRemoteLogIO"})
+# Remote log handlers that append to the stored object: a file sent again
would be stored twice (#76640).
+_APPENDING_HANDLERS = frozenset({"OSSRemoteLogIO"})
Review Comment:
This has an "magic string" code smell. I know #76640 is fixing most of
this, but if a handler gets renamed, this becomes a silent breaking change.
May be worth adding some kind of log message when the handler doesn't match any
of these name lists so at least it leaves an audit trail if something breaks?
It might save future-us some troubleshooting time.
--
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]
