diff --git a/skyrl/backends/skyrl_train/inference_engines/ray_wrapped_inference_engine.py b/skyrl/backends/skyrl_train/inference_engines/ray_wrapped_inference_engine.py index e3f0e43517..e1285b0c0d 100644 --- a/skyrl/backends/skyrl_train/inference_engines/ray_wrapped_inference_engine.py +++ b/skyrl/backends/skyrl_train/inference_engines/ray_wrapped_inference_engine.py @@ -3,6 +3,7 @@ import ray from packaging import version from ray.actor import ActorHandle +from ray.exceptions import ActorDiedError if TYPE_CHECKING: from skyrl.backends.skyrl_train.weight_sync.transfer_strategy import ( @@ -322,6 +323,13 @@ def create_ray_wrapped_inference_engines( # NOTE(shu): set to 1 for LoRA sleep_level = 1 if enable_lora else sleep_level sleep_refs = [engine.inference_engine_actor.sleep.remote(level=sleep_level) for engine in engines] - ray.get(sleep_refs) + try: + ray.get(sleep_refs) + except ActorDiedError as e: + from skyrl.train.utils.ray_logging import reraise_with_actor_diagnostics + + reraise_with_actor_diagnostics( + e, inference_engine_actors, "Inference engine actor(s) died during engine initialization." + ) return engines diff --git a/skyrl/backends/skyrl_train/inference_servers/setup.py b/skyrl/backends/skyrl_train/inference_servers/setup.py index 2d5e2d9092..e0ba1bc109 100644 --- a/skyrl/backends/skyrl_train/inference_servers/setup.py +++ b/skyrl/backends/skyrl_train/inference_servers/setup.py @@ -5,10 +5,12 @@ import ray from loguru import logger +from ray.exceptions import RayTaskError from ray.util.placement_group import placement_group as ray_placement_group from skyrl.env_vars import SKYRL_RAY_PG_TIMEOUT_IN_S from skyrl.train.config import InferenceEngineConfig, SkyRLTrainConfig +from skyrl.train.utils.ray_logging import reraise_with_actor_diagnostics from skyrl.train.utils.utils import ( ResolvedPlacementGroup, get_ray_pg_ready_with_timeout, @@ -29,6 +31,23 @@ VLLM_START_PORT = 8000 +def _start_server_groups(server_groups: List[ServerGroup]) -> None: + """Start all groups and block until every server actor reports ready.""" + # Start all server groups in parallel (non-blocking) + all_refs = [ref for g in server_groups for ref in g.start(blocking=False)] + try: + # Wait for all servers to be ready in one shot + ray.get(all_refs) + except RayTaskError as e: + # Engine init runs inside the actor's start(), so its failure arrives as a + # RayTaskError from the still-alive actor + reraise_with_actor_diagnostics( + e, + [actor for g in server_groups for actor in g.get_actors()], + "Inference server actor(s) failed during startup.", + ) + + @dataclass class InferenceServerSetup: """Inference server setup result with optional router and groups. @@ -136,16 +155,7 @@ def create_inference_servers( for i in range(num_decode) ] - # Start all prefill and decode groups in parallel (non-blocking) - all_refs = [] - for g in prefill_server_groups: - all_refs.extend(g.start(blocking=False)) - - for g in decode_server_groups: - all_refs.extend(g.start(blocking=False)) - - # Wait for all servers to be ready in one shot - ray.get(all_refs) + _start_server_groups(prefill_server_groups + decode_server_groups) # Collect URLs — refs are already resolved so lazy property returns immediately prefill_urls = [info.url for g in prefill_server_groups for info in g.server_infos] @@ -192,13 +202,7 @@ def create_inference_servers( for i in range(ie_cfg.num_engines) ] - # Start all engine groups in parallel (non-blocking) - all_refs = [] - for g in server_groups: - all_refs.extend(g.start(blocking=False)) - - # Wait for all servers to be ready in one shot - ray.get(all_refs) + _start_server_groups(server_groups) # Collect URLs — refs are already resolved so lazy property returns immediately server_urls = [info.url for g in server_groups for info in g.server_infos] diff --git a/skyrl/train/utils/ray_logging.py b/skyrl/train/utils/ray_logging.py index 887936b503..a8f12f739a 100644 --- a/skyrl/train/utils/ray_logging.py +++ b/skyrl/train/utils/ray_logging.py @@ -1,12 +1,22 @@ """ -Helper to redirect Ray actor stdout/stderr to log file. +Helpers for Ray actor log handling. -This prevents infrastructure logs from polluting the driver's stdout, -allowing only training progress to be displayed to the user. +Redirects actor stdout/stderr to a log file so infrastructure logs don't pollute the +driver's stdout, and recovers those logs' tails for diagnostics when an actor fails. """ +import contextlib import os import sys +import time +from typing import TYPE_CHECKING, NoReturn, Sequence + +if TYPE_CHECKING: + from ray.actor import ActorHandle + +# Caps total diagnostics size, keeping the tail, sized so the whole cap survives +# SkyRL Tracking.log_exception's truncation of the formatted exception +_MAX_DIAGNOSTICS_CHARS = 10_000 def redirect_actor_output_to_file(): @@ -35,3 +45,137 @@ def redirect_actor_output_to_file(): with open(log_file, "a", buffering=1) as log_f: os.dup2(log_f.fileno(), sys.stdout.fileno()) os.dup2(log_f.fileno(), sys.stderr.fileno()) + + +def _tail_file(path: str, max_lines: int, max_bytes: int = 256 * 1024) -> list[str]: + """ + Return up to the last ``max_lines`` lines of ``path``, reading at most ``max_bytes``. + + The byte bound matters if the input file is unbounded in size, where reading it + could take nontrivial time on slow filesystems. + """ + with open(path, "rb") as f: + file_size = f.seek(0, os.SEEK_END) + f.seek(max(file_size - max_bytes, 0)) + return ( + f.read() + .decode( + # The seek can split a multibyte character; + # don't let one bad byte kill the diagnostics + errors="replace" + ) + .splitlines()[-max_lines:] + ) + + +def get_actor_logs_tail( + actor_ids: Sequence[str], *, max_lines_per_actor: int = 100, state_api_timeout_s: int = 10 +) -> str | None: + """ + Best-effort collection of log tails for failure diagnostics. Never raises. + + Looks in the two places actor stderr can land: + 1. The shared `SKYRL_LOG_FILE`, when set: actors calling `redirect_actor_output_to_file` + redirect their stdout/stderr, and that of any subprocess they spawn (like vLLM's + engine core), into this file. + 2. Each given actor's Ray worker stderr file (`worker-*.err` in the session logs dir), + which this function fetches via the Ray state API, covering actors on remote nodes. + + Returns: + Joined log sections, or None if nothing could be collected. + """ + sections: list[str] = [] + + # Set by SkyRL `initialize_ray` util, in both its calling process + # and every Ray worker's runtime env + log_file = os.getenv("SKYRL_LOG_FILE") + if log_file: + with contextlib.suppress(Exception): + tail = _tail_file(log_file, max_lines=max_lines_per_actor) + if tail: + sections.append( + f"--- tail of SKYRL_LOG_FILE {log_file} " + f"(last {len(tail)} lines; infra actors redirect stdout/stderr here) ---\n" + "\n".join(tail) + ) + + with contextlib.suppress(Exception): # Never raise out of diagnostics collection + from ray._private.worker import get_dashboard_url + from ray.util.state import get_log + + # The state API is served by Ray's dashboard HTTP server; without it, resolving the + # server URL blocks for 20 x 2-s retries before failing, per internal_kv_get_with_retry: + # https://github.com/ray-project/ray/blob/ray-2.51.1/python/ray/_private/utils.py#L1126-L1147 + if get_dashboard_url(): + for actor_id in actor_ids: + with contextlib.suppress(Exception): + stderr_tail = "".join( + get_log( + actor_id=actor_id, + suffix="err", + tail=max_lines_per_actor, + timeout=state_api_timeout_s, + errors="replace", + ) + ).rstrip() + if stderr_tail: + sections.append( + f"--- stderr tail of actor {actor_id} (full log: " + f"`ray logs actor --id {actor_id} --err`) ---\n" + stderr_tail + ) + + if not sections: + return None + diagnostics = "\n\n".join(sections) + if len(diagnostics) > _MAX_DIAGNOSTICS_CHARS: + prefix = "...(truncated)...\n" + diagnostics = prefix + diagnostics[len(prefix) - _MAX_DIAGNOSTICS_CHARS :] + return diagnostics + + +def reraise_with_actor_diagnostics( + e: BaseException, actors: "Sequence[ActorHandle]", context_message: str, log_flush_grace_s: float = 2 +) -> NoReturn: + """ + Re-raise a Ray actor failure as a RuntimeError carrying the actors' log tails. + + Ray surfaces actor-side failures without the actor's stderr, which is where the root + cause actually lives when a subprocess of the actor (e.g. a vLLM engine-core child) dies; + the driver-side exception bottoms out at vLLM's `wait_for_engine_startup` with just: + > RuntimeError: Engine core initialization failed. See root cause above. Failed core proc(s): {} + + Args: + e: The original exception, chained via `__cause__`. + actors: Actor handles to fall back to when `e` doesn't name the failed actor. + context_message: Lead text describing what was being attempted. + log_flush_grace_s: Seconds to wait before snapshotting the logs. + """ + # ActorDiedError: publicly exposes the failed actor's id + # RayTaskError (a method failed on a still-alive actor): has a private _actor_id + failed_id: str | None = getattr(e, "actor_id", None) or getattr(e, "_actor_id", None) + diagnostics = None + with contextlib.suppress(Exception): # Don't mask the original failure + # vLLM relays the engine-core child's output to the actor's log via a pipe-reader + # thread that can lag the failure reaching the driver; give it a moment to drain + time.sleep(log_flush_grace_s) + diagnostics = get_actor_logs_tail( + actor_ids=[failed_id] if failed_id else [actor._actor_id.hex() for actor in actors] + ) + if diagnostics is None: + log_file = os.getenv("SKYRL_LOG_FILE") + if log_file: + location = f"SKYRL_LOG_FILE ({log_file})" + else: + logs_dir = "/session_latest/logs" + with contextlib.suppress(Exception): # Tolerate Ray being uninitialized + import ray._private.worker + + logs_dir = ray._private.worker._global_node.get_logs_dir_path() + location = f"{logs_dir}/worker-*.err on the failed actor's node" + diagnostics = ( + f"(could not fetch actor logs; check {location}, " + f"or run: ray logs actor --id {failed_id or ''} --err)" + ) + raise RuntimeError( + f"{context_message} The root-cause traceback usually lives in the failed actor's " + f"logs, not in the current Ray exception.\n\n{diagnostics}" + ) from e diff --git a/tests/train/utils/test_ray_logging.py b/tests/train/utils/test_ray_logging.py index 4db175d401..28207a0040 100644 --- a/tests/train/utils/test_ray_logging.py +++ b/tests/train/utils/test_ray_logging.py @@ -1,57 +1,77 @@ """ -Unit tests for ray_logging module (actor stdout/stderr redirection). +Unit tests for ray_logging module. uv run --isolated --extra fsdp pytest tests/train/utils/test_ray_logging.py """ +import contextlib import os import sys import tempfile +import uuid +from collections.abc import Iterator +from pathlib import Path +from unittest.mock import patch + +import pytest +import ray +from ray._private.worker import get_dashboard_url +from ray.exceptions import ActorDiedError, RayTaskError import skyrl.env_vars as env_vars_mod -from skyrl.train.utils.ray_logging import redirect_actor_output_to_file +from skyrl.train.utils.ray_logging import ( + get_actor_logs_tail, + redirect_actor_output_to_file, + reraise_with_actor_diagnostics, +) -def _set_dump_infra(monkeypatch, enabled: bool): +def _set_dump_infra(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None: """Patch both the env var and the cached module-level constant.""" monkeypatch.setenv("SKYRL_DUMP_INFRA_LOG_TO_STDOUT", "1" if enabled else "0") monkeypatch.setattr(env_vars_mod, "SKYRL_DUMP_INFRA_LOG_TO_STDOUT", enabled) +@contextlib.contextmanager +def _preserved_fd(fd: int) -> Iterator[int]: + """Yield a duplicate of ``fd``; on exit, restore ``fd`` from it and close it.""" + saved = os.dup(fd) + try: + yield saved + finally: + os.dup2(saved, fd) + os.close(saved) + + class TestRedirectActorOutputToFile: """Tests for redirect_actor_output_to_file().""" - def test_dump_to_std_skips_redirection(self, monkeypatch): + def test_dump_to_std_skips_redirection(self, monkeypatch: pytest.MonkeyPatch) -> None: """When SKYRL_DUMP_INFRA_LOG_TO_STDOUT=1, no redirection should happen.""" _set_dump_infra(monkeypatch, True) monkeypatch.setenv("SKYRL_LOG_FILE", "/tmp/should-not-be-opened.log") - original_stdout_fd = os.dup(sys.stdout.fileno()) - original_stderr_fd = os.dup(sys.stderr.fileno()) - try: + with ( + _preserved_fd(sys.stdout.fileno()) as original_stdout_fd, + _preserved_fd(sys.stderr.fileno()) as original_stderr_fd, + ): redirect_actor_output_to_file() # stdout/stderr should still point to original fds (not redirected) assert os.fstat(sys.stdout.fileno()).st_ino == os.fstat(original_stdout_fd).st_ino assert os.fstat(sys.stderr.fileno()).st_ino == os.fstat(original_stderr_fd).st_ino - finally: - os.close(original_stdout_fd) - os.close(original_stderr_fd) - def test_no_log_file_set_is_noop(self, monkeypatch): + def test_no_log_file_set_is_noop(self, monkeypatch: pytest.MonkeyPatch) -> None: """When SKYRL_LOG_FILE is not set, no redirection should happen.""" _set_dump_infra(monkeypatch, False) monkeypatch.delenv("SKYRL_LOG_FILE", raising=False) - original_stdout_fd = os.dup(sys.stdout.fileno()) - try: + with _preserved_fd(sys.stdout.fileno()) as original_stdout_fd: redirect_actor_output_to_file() assert os.fstat(sys.stdout.fileno()).st_ino == os.fstat(original_stdout_fd).st_ino - finally: - os.close(original_stdout_fd) - def test_redirects_stdout_and_stderr_to_file(self, monkeypatch): + def test_redirects_stdout_and_stderr_to_file(self, monkeypatch: pytest.MonkeyPatch) -> None: """With dump disabled and SKYRL_LOG_FILE set, stdout/stderr should write to the log file.""" _set_dump_infra(monkeypatch, False) @@ -59,10 +79,8 @@ def test_redirects_stdout_and_stderr_to_file(self, monkeypatch): log_path = os.path.join(tmpdir, "test-infra.log") monkeypatch.setenv("SKYRL_LOG_FILE", log_path) - # Save original fds so we can restore them after the test - saved_stdout_fd = os.dup(sys.stdout.fileno()) - saved_stderr_fd = os.dup(sys.stderr.fileno()) - try: + # Restore original stdout/stderr after so subsequent tests aren't affected + with _preserved_fd(sys.stdout.fileno()), _preserved_fd(sys.stderr.fileno()): redirect_actor_output_to_file() # Write to stdout/stderr — should go to the log file @@ -74,27 +92,22 @@ def test_redirects_stdout_and_stderr_to_file(self, monkeypatch): assert "stdout-test-line" in contents assert "stderr-test-line" in contents - finally: - # Restore original stdout/stderr so subsequent tests aren't affected - os.dup2(saved_stdout_fd, sys.stdout.fileno()) - os.dup2(saved_stderr_fd, sys.stderr.fileno()) - os.close(saved_stdout_fd) - os.close(saved_stderr_fd) - def test_actors_append_to_log_file(self, monkeypatch): + def test_actors_append_to_log_file(self, monkeypatch: pytest.MonkeyPatch) -> None: """Multiple actor redirections should append to the same log file.""" _set_dump_infra(monkeypatch, False) with tempfile.TemporaryDirectory() as tmpdir: log_path = os.path.join(tmpdir, "test-infra.log") - # Simulate driver truncation (initialize_ray opens with "w") + # Simulate driver truncation (SkyRL initialize_ray util opens with "w") open(log_path, "w").close() monkeypatch.setenv("SKYRL_LOG_FILE", log_path) - saved_stdout_fd = os.dup(sys.stdout.fileno()) - saved_stderr_fd = os.dup(sys.stderr.fileno()) - try: + with ( + _preserved_fd(sys.stdout.fileno()) as saved_stdout_fd, + _preserved_fd(sys.stderr.fileno()) as saved_stderr_fd, + ): # First actor redirects redirect_actor_output_to_file() os.write(sys.stdout.fileno(), b"actor-1-output\n") @@ -112,13 +125,8 @@ def test_actors_append_to_log_file(self, monkeypatch): assert "actor-1-output" in contents assert "actor-2-output" in contents - finally: - os.dup2(saved_stdout_fd, sys.stdout.fileno()) - os.dup2(saved_stderr_fd, sys.stderr.fileno()) - os.close(saved_stdout_fd) - os.close(saved_stderr_fd) - def test_dump_to_std_skips_log_file_creation(self, monkeypatch): + def test_dump_to_std_skips_log_file_creation(self, monkeypatch: pytest.MonkeyPatch) -> None: """When dump enabled, initialize_ray should not create log dir or set SKYRL_LOG_FILE.""" _set_dump_infra(monkeypatch, True) monkeypatch.delenv("SKYRL_LOG_FILE", raising=False) @@ -132,4 +140,93 @@ def test_dump_to_std_skips_log_file_creation(self, monkeypatch): with tempfile.TemporaryDirectory() as tmpdir: expected_log_dir = os.path.join(tmpdir, "skyrl-logs", "test-run") assert not os.path.exists(expected_log_dir) - assert os.environ.get("SKYRL_LOG_FILE") is None + assert not os.environ.get("SKYRL_LOG_FILE") + + +@ray.remote(num_cpus=0) +class _FailingInitActor: + """Actor whose constructor writes a marker to stderr and then dies.""" + + def __init__(self, marker: str) -> None: + redirect_actor_output_to_file() # no-op unless SKYRL_LOG_FILE is set + # Write at the fd level: survives the dup2 redirect and bypasses Python buffering + os.write(sys.stderr.fileno(), f"{marker}\n".encode()) + raise RuntimeError("intentional init failure") + + def ping(self) -> None: + """No-op for ray.get to surface the constructor's failure.""" + return None + + +@ray.remote(num_cpus=0) +class _FailingMethodActor: + """Actor that stays alive but writes a marker to stderr and raises from a method.""" + + def fail(self, marker: str) -> None: + os.write(sys.stderr.fileno(), f"{marker}\n".encode()) + raise RuntimeError("intentional method failure") + + +class TestActorLogDiagnostics: + """Tests for get_actor_logs_tail() and reraise_with_actor_diagnostics().""" + + @pytest.mark.parametrize( + ("use_skyrl_log_file", "dies_in_init"), + [ + pytest.param(True, True, id="dead_actor_skyrl_log_file_route"), + pytest.param(False, True, id="dead_actor_state_api_route"), + pytest.param(False, False, id="live_actor_state_api_route"), + ], + ) + def test_reraise_surfaces_actor_stderr(self, use_skyrl_log_file: bool, dies_in_init: bool, tmp_path: Path) -> None: + if not use_skyrl_log_file and not get_dashboard_url(): + pytest.skip("Ray dashboard unavailable, state API cannot fetch actor logs") + marker = f"SKYRL-TEST-MARKER-{uuid.uuid4().hex}" + with patch.dict(os.environ): + # Drop any inherited SKYRL_LOG_FILE so only the parametrized route is active + os.environ.pop("SKYRL_LOG_FILE", None) + if use_skyrl_log_file: + log_file = tmp_path / "infra.log" + os.environ["SKYRL_LOG_FILE"] = str(log_file) + actor = _FailingInitActor.options(runtime_env={"env_vars": {"SKYRL_LOG_FILE": str(log_file)}}).remote( + marker + ) + elif dies_in_init: + actor = _FailingInitActor.remote(marker) + else: + actor = _FailingMethodActor.remote() + + with pytest.raises(RuntimeError, match="engine init failed") as exc_info: + try: + ray.get(actor.ping.remote() if dies_in_init else actor.fail.remote(marker)) + except (ActorDiedError, RayTaskError) as e: + reraise_with_actor_diagnostics(e, [actor], "engine init failed") + + assert marker in str(exc_info.value), "the actor's stderr tail should be attached" + assert isinstance(exc_info.value.__cause__, ActorDiedError if dies_in_init else RayTaskError) + + def test_get_actor_logs_tail_without_actors(self, tmp_path: Path) -> None: + log_file = tmp_path / "infra.log" + log_file.write_text("\n".join(f"line-{i}" for i in range(300)) + "\n") + with patch.dict(os.environ, {"SKYRL_LOG_FILE": str(log_file)}): + diagnostics = get_actor_logs_tail([], max_lines_per_actor=5) + + assert diagnostics + assert str(log_file) in diagnostics, "the section header should name the log file" + assert "line-299" in diagnostics + assert "line-294" not in diagnostics, "the tail should be capped at max_lines_per_actor" + + with patch.dict(os.environ): + # Drop any inherited SKYRL_LOG_FILE so the log-file route stays off + os.environ.pop("SKYRL_LOG_FILE", None) + assert not get_actor_logs_tail([]), "no SKYRL_LOG_FILE and no actors leaves nothing to collect" + + def test_reraise_falls_back_to_pointer_text(self) -> None: + cause = ValueError("boom") + with patch.dict(os.environ): + # Drop any inherited SKYRL_LOG_FILE so the log-file route stays off + os.environ.pop("SKYRL_LOG_FILE", None) + with pytest.raises(RuntimeError, match="could not fetch actor logs") as exc_info: + reraise_with_actor_diagnostics(cause, [], "engines died") + + assert exc_info.value.__cause__ is cause