diff --git a/astrbot/core/utils/event_loop_diagnostics.py b/astrbot/core/utils/event_loop_diagnostics.py index 487fee061..d11df16f8 100644 --- a/astrbot/core/utils/event_loop_diagnostics.py +++ b/astrbot/core/utils/event_loop_diagnostics.py @@ -1,11 +1,12 @@ import asyncio import faulthandler import os -import sys from dataclasses import dataclass +from pathlib import Path from typing import TextIO from astrbot import logger +from astrbot.core.utils.astrbot_path import get_astrbot_data_path LAG_MONITOR_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_LAG_MONITOR" LAG_MONITOR_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_LAG_INTERVAL" @@ -13,11 +14,15 @@ LAG_MONITOR_THRESHOLD_ENV = "ASTRBOT_EVENT_LOOP_LAG_THRESHOLD" WATCHDOG_ENABLED_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG" WATCHDOG_INTERVAL_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_INTERVAL" WATCHDOG_TIMEOUT_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_TIMEOUT" +WATCHDOG_LOG_PATH_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_PATH" +WATCHDOG_LOG_MAX_BYTES_ENV = "ASTRBOT_EVENT_LOOP_WATCHDOG_LOG_MAX_BYTES" DEFAULT_LAG_MONITOR_INTERVAL = 1.0 DEFAULT_LAG_MONITOR_THRESHOLD = 5.0 DEFAULT_WATCHDOG_INTERVAL = 1.0 DEFAULT_WATCHDOG_TIMEOUT = 15.0 +DEFAULT_WATCHDOG_LOG_RELATIVE_PATH = Path("logs") / "event_loop_watchdog.log" +DEFAULT_WATCHDOG_LOG_MAX_BYTES = 1024 * 1024 @dataclass(frozen=True) @@ -31,6 +36,8 @@ class EventLoopDiagnosticSettings: watchdog_enabled: Whether to arm the faulthandler watchdog. watchdog_interval: Seconds between faulthandler watchdog refreshes. watchdog_timeout: Seconds without event loop progress before dumping stacks. + watchdog_log_path: File that receives faulthandler watchdog output. + watchdog_log_max_bytes: Maximum watchdog log bytes before rotation. """ lag_monitor_enabled: bool @@ -39,6 +46,8 @@ class EventLoopDiagnosticSettings: watchdog_enabled: bool watchdog_interval: float watchdog_timeout: float + watchdog_log_path: Path + watchdog_log_max_bytes: int def _env_flag(name: str, default: bool) -> bool: @@ -93,6 +102,59 @@ def _env_float(name: str, default: float, minimum: float) -> float: return parsed +def _env_int(name: str, default: int, minimum: int) -> int: + """Read a bounded integer from the environment. + + Args: + name: Environment variable name. + default: Value to use when parsing fails or the value is too small. + minimum: Smallest accepted value. + + Returns: + Parsed integer value or the default. + """ + value = os.environ.get(name) + if value is None or value.strip() == "": + return default + try: + parsed = int(value) + except ValueError: + logger.warning( + "Invalid %s=%r, fallback to %d.", + name, + value, + default, + ) + return default + if parsed < minimum: + logger.warning( + "Invalid %s=%r, expected at least %d; fallback to %d.", + name, + value, + minimum, + default, + ) + return default + return parsed + + +def _watchdog_log_path() -> Path: + """Resolve the watchdog stack dump log path. + + Returns: + Absolute path for watchdog stack dump output. + """ + configured = os.environ.get(WATCHDOG_LOG_PATH_ENV, "").strip() + path = ( + Path(configured).expanduser() + if configured + else DEFAULT_WATCHDOG_LOG_RELATIVE_PATH + ) + if not path.is_absolute(): + path = Path(get_astrbot_data_path()) / path + return path + + def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: """Load event loop diagnostic settings from environment variables. @@ -122,6 +184,12 @@ def load_event_loop_diagnostic_settings() -> EventLoopDiagnosticSettings: DEFAULT_WATCHDOG_TIMEOUT, 1.0, ), + watchdog_log_path=_watchdog_log_path(), + watchdog_log_max_bytes=_env_int( + WATCHDOG_LOG_MAX_BYTES_ENV, + DEFAULT_WATCHDOG_LOG_MAX_BYTES, + 1024, + ), ) @@ -151,11 +219,46 @@ async def monitor_event_loop_lag( expected = now + interval +def _rotate_watchdog_log_file(log_path: Path, max_bytes: int) -> None: + """Rotate the watchdog log when it reaches the configured size limit. + + Args: + log_path: Current watchdog log path. + max_bytes: Maximum current log size before rotation. + """ + try: + if not log_path.exists() or log_path.stat().st_size < max_bytes: + return + rotated_path = log_path.with_name(f"{log_path.name}.1") + if rotated_path.exists(): + rotated_path.unlink() + log_path.replace(rotated_path) + except OSError as e: + logger.warning("Failed to rotate event loop watchdog log %s: %s", log_path, e) + + +def _open_watchdog_log_file(log_path: Path, max_bytes: int) -> TextIO: + """Open the watchdog log file after applying size-based rotation. + + Args: + log_path: Current watchdog log path. + max_bytes: Maximum current log size before rotation. + + Returns: + Writable text file object for faulthandler output. + """ + log_path.parent.mkdir(parents=True, exist_ok=True) + _rotate_watchdog_log_file(log_path, max_bytes) + return log_path.open("a", encoding="utf-8") + + async def faulthandler_event_loop_watchdog( *, timeout: float = DEFAULT_WATCHDOG_TIMEOUT, interval: float = DEFAULT_WATCHDOG_INTERVAL, dump_file: TextIO | None = None, + dump_path: Path | None = None, + max_bytes: int = DEFAULT_WATCHDOG_LOG_MAX_BYTES, ) -> None: """Dump all thread stacks if the event loop is blocked for too long. @@ -163,20 +266,26 @@ async def faulthandler_event_loop_watchdog( timeout: Seconds without watchdog refresh before faulthandler dumps stacks. interval: Seconds between watchdog refreshes while the event loop is healthy. dump_file: File object that receives faulthandler output. + dump_path: Path that receives faulthandler output when dump_file is unset. + max_bytes: Maximum current log size before rotation. """ - output = dump_file or sys.stderr - if not faulthandler.is_enabled(): - faulthandler.enable(file=output) - + log_path = dump_path or _watchdog_log_path() try: while True: faulthandler.cancel_dump_traceback_later() - faulthandler.dump_traceback_later( - timeout, - repeat=False, - file=output, - ) - await asyncio.sleep(interval) + output = dump_file or _open_watchdog_log_file(log_path, max_bytes) + should_close = dump_file is None + try: + faulthandler.dump_traceback_later( + timeout, + repeat=False, + file=output, + ) + await asyncio.sleep(interval) + finally: + faulthandler.cancel_dump_traceback_later() + if should_close: + output.close() finally: faulthandler.cancel_dump_traceback_later() @@ -204,15 +313,20 @@ def create_event_loop_diagnostic_tasks() -> list[asyncio.Task]: if settings.watchdog_enabled: logger.warning( "Event loop faulthandler watchdog enabled: timeout=%.3fs interval=%.3fs. " - "If the loop is blocked, Python thread stacks will be written to stderr.", + "If the loop is blocked, Python thread stacks will be written to %s " + "(rotates at %d bytes).", settings.watchdog_timeout, settings.watchdog_interval, + settings.watchdog_log_path, + settings.watchdog_log_max_bytes, ) tasks.append( asyncio.create_task( faulthandler_event_loop_watchdog( timeout=settings.watchdog_timeout, interval=settings.watchdog_interval, + dump_path=settings.watchdog_log_path, + max_bytes=settings.watchdog_log_max_bytes, ), name="event_loop_faulthandler_watchdog", ) diff --git a/tests/unit/test_event_loop_diagnostics.py b/tests/unit/test_event_loop_diagnostics.py index a1e4d7144..20af5d425 100644 --- a/tests/unit/test_event_loop_diagnostics.py +++ b/tests/unit/test_event_loop_diagnostics.py @@ -18,6 +18,8 @@ def _clear_diagnostic_env(monkeypatch): diagnostics.WATCHDOG_ENABLED_ENV, diagnostics.WATCHDOG_INTERVAL_ENV, diagnostics.WATCHDOG_TIMEOUT_ENV, + diagnostics.WATCHDOG_LOG_PATH_ENV, + diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, ): monkeypatch.delenv(name, raising=False) @@ -32,6 +34,7 @@ def test_load_event_loop_diagnostic_settings_defaults(monkeypatch): assert settings.lag_monitor_interval == diagnostics.DEFAULT_LAG_MONITOR_INTERVAL assert settings.lag_monitor_threshold == diagnostics.DEFAULT_LAG_MONITOR_THRESHOLD assert settings.watchdog_enabled is False + assert settings.watchdog_log_max_bytes == diagnostics.DEFAULT_WATCHDOG_LOG_MAX_BYTES def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): @@ -40,12 +43,16 @@ def test_load_event_loop_diagnostic_settings_from_env(monkeypatch): monkeypatch.setenv(diagnostics.LAG_MONITOR_ENABLED_ENV, "0") monkeypatch.setenv(diagnostics.WATCHDOG_ENABLED_ENV, "yes") monkeypatch.setenv(diagnostics.WATCHDOG_TIMEOUT_ENV, "30") + monkeypatch.setenv(diagnostics.WATCHDOG_LOG_PATH_ENV, "/tmp/astrbot-watchdog.log") + monkeypatch.setenv(diagnostics.WATCHDOG_LOG_MAX_BYTES_ENV, "2048") settings = diagnostics.load_event_loop_diagnostic_settings() assert settings.lag_monitor_enabled is False assert settings.watchdog_enabled is True assert settings.watchdog_timeout == 30 + assert settings.watchdog_log_path.as_posix() == "/tmp/astrbot-watchdog.log" + assert settings.watchdog_log_max_bytes == 2048 @pytest.mark.asyncio @@ -81,13 +88,6 @@ async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch): calls = [] class FakeFaultHandler: - def is_enabled(self): - calls.append("is_enabled") - return False - - def enable(self, file): - calls.append(("enable", file)) - def cancel_dump_traceback_later(self): calls.append("cancel") @@ -104,6 +104,44 @@ async def test_faulthandler_watchdog_cancels_pending_dump(monkeypatch): task.cancel() await asyncio.gather(task, return_exceptions=True) - assert "is_enabled" in calls assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls) assert calls[-1] == "cancel" + + +@pytest.mark.asyncio +async def test_faulthandler_watchdog_writes_rotating_log(tmp_path, monkeypatch): + """The faulthandler watchdog should write to and rotate its log file.""" + log_path = tmp_path / "logs" / "event_loop_watchdog.log" + log_path.parent.mkdir() + log_path.write_text("x" * 8, encoding="utf-8") + calls = [] + + class FakeFaultHandler: + def cancel_dump_traceback_later(self): + calls.append("cancel") + + def dump_traceback_later(self, timeout, repeat, file): + calls.append(("dump", timeout, repeat, file.name)) + file.write("watchdog dump\n") + file.flush() + + fake_faulthandler = FakeFaultHandler() + monkeypatch.setattr(diagnostics, "faulthandler", fake_faulthandler) + + task = asyncio.create_task( + diagnostics.faulthandler_event_loop_watchdog( + timeout=10, + interval=1, + dump_path=log_path, + max_bytes=4, + ) + ) + await asyncio.sleep(0) + task.cancel() + await asyncio.gather(task, return_exceptions=True) + + assert log_path.read_text(encoding="utf-8") == "watchdog dump\n" + assert log_path.with_name("event_loop_watchdog.log.1").read_text( + encoding="utf-8" + ) == "x" * 8 + assert any(isinstance(call, tuple) and call[0] == "dump" for call in calls)