seanghaeli commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4522240606
##########
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:
Agreed, it's definitely just a stop-gap that #76640 will address properly.
For now, I added a log message
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3189,6 +3189,35 @@ triggerer:
type: integer
example: ~
default: "30"
+ watcher_log_to_stdout:
+ description: |
+ Also write an asset watcher's lines to the triggerer's standard
output, in addition to the watcher's
+ own log file. Platform log shippers that read the triggerer's output
rely on this. A deferred task's
+ and a callback's trigger lines go to their task instance's and
callback's logs, as before.
+ version_added: 3.5.0
+ type: boolean
+ example: ~
+ default: "True"
+ watcher_log_max_open_files:
+ description: |
+ Maximum number of asset watcher log files a triggerer keeps open at
once. When another watcher logs,
+ the least recently written file is closed, and it is reopened,
appending, when its watcher logs
+ again. Set to 0 for no limit. Either way it is at most a quarter of
the triggerer's open-files limit
+ (``ulimit -n``).
+ version_added: 3.5.0
+ type: integer
+ example: ~
+ default: "256"
+ watcher_log_max_bytes:
+ description: |
+ Maximum size in bytes of an asset watcher's local log file. When it is
reached the older lines are
+ moved to a ``rotated`` directory next to it, replacing the previous
rotated file, so a watcher uses at
+ most twice this much local disk on each triggerer: 4 MiB by default,
about 13,000 lines. With remote
+ logging, each rotated file is uploaded before it is replaced. Set to 0
to disable the cap.
Review Comment:
I agree with softening the docs language. Updated.
--
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]
