o-nikolas commented on code in PR #76527:
URL: https://github.com/apache/airflow/pull/76527#discussion_r4532326353
##########
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:
The last force push dropped d558328993, and this comment now says the
runner's last lines are already drained. As far as I can tell `kill()` returns
as soon as it sees the exit code, so lines still sitting in the log socket
aren't read yet and would miss the upload. Was dropping it on purpose, to leave
that to #76514? If so, could the comment say that instead?
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -3189,6 +3189,37 @@ 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: |
+ Target 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
+ normally 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, and a rotation waits
+ for the previous upload to finish (and one retry if it failed), so
while the remote store is slow or
Review Comment:
Small accuracy thing: "each rotated file is uploaded before it is replaced"
isn't quite true during a longer outage. Once the upload and its one retry
fail, the next rotation deletes that part (line 1176 of the runner, and
`test_a_store_that_stays_down_keeps_one_rotated_file` checks it on purpose).
I'm fine with that trade-off for the disk cap, but can we say the part is tried
twice and then dropped, so operators know they can lose lines while the store
is down? The same sentence is in `event-scheduling.rst` line 511. The "so
nothing is lost" in the PR description could get the same tweak.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
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:
With HDFS and OSS (or a custom store that keeps the local file), a
successful upload here looks the same as a failed one, since they never empty
the file unless `delete_local_logs` is on. So a dead triggerer's directories
never get removed, and they're uploaded again on every scan, twice per
triggerer start. For OSS that appends the same lines again each time, and the
local disk isn't capped the way the docs say. Could this use the same rule as
`_upload_rotated`, where leftover lines only count as a failure for S3, GCS and
WASB? `test_files_a_failed_upload_left_are_kept_for_a_later_start` would then
need an S3-named fake, because its bare handler also matches the HDFS success
case.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
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():
Review Comment:
Small guard idea: the scan checks that the job id part of the path is
digits, but not the trigger id part. With a custom `log_filename_template` and
a Dag called `triggers`, a task log directory could match `triggers/*/<digits>`
and get rmtree'd here. Also checking `producer_dir.parent.name.isdigit()` would
rule that out.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
airflow-core/src/airflow/utils/log/trigger_log.py:
##########
@@ -0,0 +1,69 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Where an asset watcher's own log lives, relative to the local and remote
log folders."""
+
+from __future__ import annotations
+
+import posixpath
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+ from pathlib import Path
+ from uuid import UUID
+
+TRIGGER_LOG_ROOT = "triggers"
+TRIGGER_LOG_FILE_NAME = "trigger.log"
+
+# Older lines moved aside by the local size cap. Their own directory, because
remote handlers with
+# ``delete_local_logs`` delete the uploaded file's parent directory.
+TRIGGER_LOG_ROTATED_DIR = "rotated"
+
+
+def trigger_log_relative_path(trigger_id: int, producer_id: int | UUID) -> str:
Review Comment:
AGENTS.md asks for verb names on functions. Maybe
`get_trigger_log_relative_path`, `get_trigger_log_rotated_path`,
`get_trigger_log_part_path` and `list_trigger_log_files`?
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
##########
airflow-core/tests/unit/jobs/test_triggerer_job.py:
##########
@@ -1545,6 +1555,1146 @@ def
test_trigger_logger_fd_closed_when_upload_to_remote_raises(jobless_superviso
assert 42 not in jobless_supervisor.running_triggers
+class TestWatcherLogFile:
+ @staticmethod
+ def _watcher_trigger(mocker, trigger_id):
+ asset = mocker.Mock(uri="s3://bucket/orders")
+ asset.name = "orders"
+ watcher = mocker.Mock(asset=asset)
+ watcher.name = "new_files"
+ return mocker.Mock(
+ spec=Trigger,
+ id=trigger_id,
+ classpath="some.path.Trigger",
+ encrypted_kwargs="",
+ task_instance=None,
+ callback=None,
+ assets=[asset],
+ asset_watchers=[watcher],
+ )
+
+ @staticmethod
+ def _create(supervisor, trigger, mocker):
+ supervisor._create_workload(
+ trigger=trigger, dag_bag=mocker.Mock(),
render_log_fname=mocker.Mock(), session=mocker.Mock()
+ )
+
+ @conf_vars({("triggerer", "watcher_log_max_bytes"): "1234"})
+ def test_one_file_per_assignment_and_triggerer(self, jobless_supervisor,
mocker):
+ trigger = self._watcher_trigger(mocker, 42)
+
+ self._create(jobless_supervisor, trigger, mocker)
+ factory = jobless_supervisor.logger_cache[42]
+ # A restarted instance (e.g. an SQS-style watcher after each event)
keeps writing the same file.
+ self._create(jobless_supervisor, trigger, mocker)
+
+ assert isinstance(factory, WatcherLoggingFactory)
+ assert factory.log_path ==
f"triggers/42/{jobless_supervisor.id}/trigger.log"
+ assert (factory.trigger_id, factory.max_bytes) == (42, 1234)
+ assert jobless_supervisor.logger_cache[42] is factory
+
+ def test_a_callback_trigger_logs_to_the_callback_log_only(self,
jobless_supervisor, mocker):
+ trigger = self._watcher_trigger(mocker, 42)
+ trigger.assets = []
+ trigger.callback = mocker.Mock(data={"dag_id": "d"}, id=uuid.uuid4())
+
+ self._create(jobless_supervisor, trigger, mocker)
+
+ factory = jobless_supervisor.logger_cache[42]
+ assert not isinstance(factory, WatcherLoggingFactory)
+ assert factory.log_path ==
f"triggerer_callbacks/d/unknown/{trigger.callback.id}/callback.log"
+
+ def test_file_is_kept_across_instance_restarts(self, jobless_supervisor,
mocker):
+ factory = mocker.MagicMock(spec=WatcherLoggingFactory)
+ jobless_supervisor.logger_cache[42] = factory
+ jobless_supervisor.running_triggers.add(42)
+ mocker.patch.object(TriggerRunnerSupervisor,
"build_trigger_workloads", return_value=[])
+
+ jobless_supervisor._handle_request(
+ messages.TriggerStateChanges(finished=[42]),
log=mocker.Mock(spec=FilteringBoundLogger), req_id=0
+ )
+ jobless_supervisor.update_triggers({42})
+
+ factory.close.assert_not_called()
+ assert jobless_supervisor.logger_cache[42] is factory
+
+ @pytest.mark.parametrize(
+ ("assignments", "queue"),
+ [
+ pytest.param({42: 7}, "releasing_triggers", id="moved"),
+ pytest.param({}, "cancelling_triggers", id="gone"),
+ ],
+ )
+ def test_file_is_closed_one_loop_after_the_trigger_left(
+ self, jobless_supervisor, mocker, assignments, queue
+ ):
+ factory = mocker.MagicMock(spec=WatcherLoggingFactory)
+ jobless_supervisor.logger_cache[42] = factory
+ jobless_supervisor.running_triggers.add(42)
+ mocker.patch.object(TriggerRunnerSupervisor,
"fetch_trigger_assignments", return_value=assignments)
+ log = mocker.Mock(spec=FilteringBoundLogger)
+
+ jobless_supervisor.update_triggers(set())
+ assert 42 in getattr(jobless_supervisor, queue)
+ # Still open: the instance's last lines (on_kill, cleanup) are still
to come on the log socket.
+
jobless_supervisor._handle_request(messages.TriggerStateChanges(finished=[42]),
log=log, req_id=0)
+ factory.close.assert_not_called()
+
+ jobless_supervisor.update_triggers(set())
+ jobless_supervisor._upload_pool.shutdown(wait=True)
+
+ factory.upload_to_remote.assert_called_once()
+ factory.close.assert_called()
+ assert 42 not in jobless_supervisor.logger_cache
+
+ def test_close_trigger_log_files_uploads_every_open_trigger_log(self,
jobless_supervisor, mocker):
+ own_file = mocker.MagicMock(spec=WatcherLoggingFactory)
+ task_log = mocker.MagicMock(spec=TriggerLoggingFactory)
+ jobless_supervisor.logger_cache.update({2: own_file, 3: task_log})
+
+ jobless_supervisor.close_trigger_log_files()
+
+ own_file.upload_to_remote.assert_called_once()
+ task_log.upload_to_remote.assert_called_once()
+ task_log.close.assert_called_once()
+ assert jobless_supervisor.logger_cache == {}
+
+ @staticmethod
+ def _processors():
+ return [JSONRenderer(serializer=lambda obj, **_:
json.dumps(obj).encode())]
+
+ @staticmethod
+ def _factory(max_bytes=0, **kwargs):
+ return WatcherLoggingFactory(
+ log_path="triggers/1/5/trigger.log", trigger_id=1,
max_bytes=max_bytes, to_stdout=False, **kwargs
+ )
+
+ @pytest.mark.parametrize("max_bytes", [0, 400])
+ def test_file_is_appended_to_when_reopened(self, tmp_path, max_bytes):
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ for run in range(2):
+ factory = self._factory(max_bytes)
+ factory(self._processors()).log(logging.INFO, "line", run=run)
+ factory.close()
+
+ lines = (tmp_path /
"triggers/1/5/trigger.log").read_text().splitlines()
+ assert [json.loads(line)["run"] for line in lines] == [0, 1]
+ assert {json.loads(line)["trigger_id"] for line in lines} == {1}
+
+ def test_a_closed_file_is_reopened_by_the_next_line(self, tmp_path):
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ factory = self._factory()
+ factory(self._processors()).log(logging.INFO, "line", n=1)
+ factory.close()
+ factory(self._processors()).log(logging.INFO, "line", n=2)
+ factory.close()
+
+ lines = (tmp_path /
"triggers/1/5/trigger.log").read_text().splitlines()
+ assert [json.loads(line)["n"] for line in lines] == [1, 2]
+
+ def
test_a_closed_file_is_open_again_before_the_processors_see_the_next_line(self,
tmp_path):
+ # A remote handler that streams each line (CloudWatch) names its
stream after the open file.
+ seen_closed = []
+
+ def record(logger, method_name, event_dict):
+ seen_closed.append(logger._file.closed)
+ return event_dict
+
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ factory = self._factory()
+ factory([record, *self._processors()]).log(logging.INFO, "line",
n=1)
+ factory.close()
+ factory([record, *self._processors()]).log(logging.INFO, "line",
n=2)
+ factory.close()
+
+ assert seen_closed == [False, False]
+
+ @pytest.mark.parametrize("max_bytes", [0, 80], ids=["no-cap", "rotates"])
+ @pytest.mark.parametrize("closed", [True, False], ids=["closed", "open"])
+ def test_a_removed_directory_is_recreated_by_the_next_line(self, tmp_path,
max_bytes, closed):
+ # Open: on a shared log folder, another triggerer's orphan scan can
remove a running one's files.
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ factory = self._factory(max_bytes)
+ logger = factory(self._processors())
+ logger.log(logging.INFO, "line", n=1)
+ if closed:
+ factory.close()
+ shutil.rmtree(tmp_path / "triggers/1")
+ logger.log(logging.INFO, "line", n=2)
+ logger.log(logging.INFO, "line", n=3)
+ factory.close()
+
+ files = [tmp_path / "triggers/1/5/rotated/trigger.log", tmp_path /
"triggers/1/5/trigger.log"]
+ lines = [line for f in files if f.exists() for line in
f.read_text().splitlines()]
+ assert [json.loads(line)["n"] for line in lines] == [2, 3]
+
+ def
test_local_disk_is_capped_with_one_rotated_file_in_its_own_directory(self,
tmp_path):
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ factory = self._factory(1000)
+ logger = factory(self._processors())
+ for n in range(500):
+ logger.log(logging.INFO, "line", n=n)
+ factory.close()
+
+ live = tmp_path / "triggers/1/5/trigger.log"
+ rotated = tmp_path / "triggers/1/5/rotated/trigger.log"
+ numbers = [json.loads(line)["n"] for p in (rotated, live) for line in
p.read_text().splitlines()]
+ assert live.stat().st_size < 1000
+ assert rotated.stat().st_size < 1000 + 100
+ assert numbers == list(range(numbers[0], 500))
+
+ def
test_upload_closes_first_and_sends_the_rotated_file_before_the_live_file(self,
tmp_path, mocker):
+ handler = mocker.Mock()
+ open_at_upload = []
+ handler.upload.side_effect = lambda path, ti: open_at_upload.append(
+ not factory.bound_logger._logger._file.closed
+ )
+ mocker.patch("airflow.sdk.log.load_remote_log_handler",
return_value=handler)
+ with conf_vars({("logging", "base_log_folder"): str(tmp_path)}):
+ factory = self._factory(200)
+ logger = factory(self._processors())
+ for n in range(10):
+ logger.log(logging.INFO, "line", n=n)
+ factory.upload_to_remote()
+
+ assert [c.args for c in handler.upload.call_args_list] == [
+ ("triggers/1/5/rotated/trigger.log", None),
+ ("triggers/1/5/trigger.log", None),
+ ]
+ assert open_at_upload == [False, False]
+
+ def test_with_remote_logging_every_rotated_file_is_uploaded(self,
jobless_supervisor, mocker, tmp_path):
+ # A watcher stays on one triggerer for days: a rotated file replaced
before it is uploaded is lost.
+ remote: dict[str, str] = {}
+
+ def upload(path, ti): # appends, then empties the local file, as the
S3 handler does
+ local = tmp_path / path
+ remote[path] = remote.get(path, "") + local.read_text()
+ local.write_text("")
+
+ mocker.patch("airflow.sdk.log.load_remote_log_handler",
return_value=mocker.Mock(upload=upload))
Review Comment:
A bare `Mock` handler has a truthy `processors` attribute, so this test goes
down the streaming-handler path (no part renaming), even though the comment
says it acts like S3. Giving these handler mocks a spec, or `processors=()`,
would keep each test on the path it means to cover. Same for the handler mocks
at lines 1747, 2191 and 2587.
---
Drafted-by: Kiro (claude-opus-5.5); reviewed by @o-nikolas before posting
--
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]
