diff --git a/runpod/serverless/modules/rp_fitness.py b/runpod/serverless/modules/rp_fitness.py index 369a84f0..77df97e7 100644 --- a/runpod/serverless/modules/rp_fitness.py +++ b/runpod/serverless/modules/rp_fitness.py @@ -110,6 +110,51 @@ def _reset_registration_state() -> None: _registration_state["system_checks"] = False +# Bound how long the best-effort unhealthy report may delay the exit. +_REPORT_TIMEOUT_SECONDS = 2 + + +def _report_unhealthy(check: str, reason: str) -> None: + """ + Best-effort report of a fitness-check failure to the host before exit. + + Sends a single GET to the ping URL (same URL/credentials the heartbeat + uses) with status=unhealthy plus the failing check name and reason, so the + host can emit a queryable worker.fitness_failed event. Any failure — no + ping URL, no API key, HTTP error, timeout — is swallowed, so this can never + prevent the os._exit that follows. It is synchronous, so it may delay that + exit by up to _REPORT_TIMEOUT_SECONDS (network phases only; it adds no + delay when there is no ping URL/API key to report to). + """ + ping_url = os.environ.get("RUNPOD_WEBHOOK_PING") + api_key = os.environ.get("RUNPOD_AI_API_KEY") + if not ping_url or ping_url == "PING_NOT_SET" or not api_key: + return + + try: + # Deferred imports: keep module import light and avoid import cycles. + from runpod.http_client import SyncClientSession + from runpod.serverless.modules.worker_state import WORKER_ID + from runpod.version import __version__ as runpod_version + + ping_url = ping_url.replace("$RUNPOD_POD_ID", WORKER_ID) + params = { + "status": "unhealthy", + "check": check, + "reason": reason[:256], + "runpod_version": runpod_version, + } + session = SyncClientSession() + try: + session.headers.update({"Authorization": api_key}) + session.get(ping_url, params=params, timeout=_REPORT_TIMEOUT_SECONDS) + finally: + session.close() + except Exception: + # Best-effort only; the exit is the guarantee, not this report. + pass + + def _ensure_gpu_check_registered() -> None: """ Ensure GPU fitness check is registered. @@ -234,6 +279,14 @@ async def run_fitness_checks() -> None: ) log.debug(f"Traceback:\n{full_traceback}") + # Best-effort report to the host so the failure is queryable. It is + # bounded (see _REPORT_TIMEOUT_SECONDS) and fully swallowed, so it + # can delay the force-exit below but can never prevent it. + try: + _report_unhealthy(check_name, f"{error_type}: {error_message}") + except Exception: # a report failure must never prevent the exit + pass + # Force-kill immediately; see _terminate_unhealthy for why this is # os._exit rather than sys.exit. log.error("Worker is unhealthy, exiting.") diff --git a/tests/test_serverless/test_modules/test_fitness/test_force_kill.py b/tests/test_serverless/test_modules/test_fitness/test_force_kill.py index 31255180..9bb4f213 100644 --- a/tests/test_serverless/test_modules/test_fitness/test_force_kill.py +++ b/tests/test_serverless/test_modules/test_fitness/test_force_kill.py @@ -7,11 +7,16 @@ raises ``SystemExit`` and then blocks in interpreter shutdown joining those threads, so the worker logs "unhealthy, exiting." but never terminates. The exit must go through ``os._exit`` to terminate unconditionally. + +Also covers SLS-380: the best-effort unhealthy report the worker sends to the +host before force-exiting. """ import subprocess import sys -from unittest.mock import patch +from unittest.mock import MagicMock, patch + +import pytest from runpod.serverless.modules import rp_fitness @@ -92,3 +97,78 @@ def test_terminate_unhealthy_exits_even_if_flush_raises(): rp_fitness._terminate_unhealthy(1) mock_exit.assert_called_once_with(1) + + +def test_report_unhealthy_posts_check_and_reason(monkeypatch): + monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping/$RUNPOD_POD_ID") + monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123") + + fake_session = MagicMock() + with patch("runpod.http_client.SyncClientSession", return_value=fake_session), \ + patch("runpod.serverless.modules.worker_state.WORKER_ID", "podABC"): + rp_fitness._report_unhealthy("_cuda_init_check", "RuntimeError: boom") + + assert fake_session.get.call_count == 1 + call = fake_session.get.call_args + url = call.args[0] if call.args else call.kwargs["url"] + params = call.kwargs["params"] + assert "podABC" in url # $RUNPOD_POD_ID substituted + assert params["status"] == "unhealthy" + assert params["check"] == "_cuda_init_check" + assert params["reason"] == "RuntimeError: boom" + fake_session.headers.update.assert_called_once_with({"Authorization": "key-123"}) + + +def test_report_unhealthy_skipped_without_ping_url(monkeypatch): + monkeypatch.delenv("RUNPOD_WEBHOOK_PING", raising=False) + monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123") + with patch("runpod.http_client.SyncClientSession") as session_cls: + rp_fitness._report_unhealthy("_memory_check", "RuntimeError: low") + session_cls.assert_not_called() + + +def test_report_unhealthy_truncates_long_reason(monkeypatch): + monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping") + monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123") + + fake_session = MagicMock() + with patch("runpod.http_client.SyncClientSession", return_value=fake_session): + rp_fitness._report_unhealthy("_disk_check", "x" * 300) + + params = fake_session.get.call_args.kwargs["params"] + assert len(params["reason"]) == 256 + + +def test_report_unhealthy_skipped_without_api_key(monkeypatch): + monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping") + monkeypatch.delenv("RUNPOD_AI_API_KEY", raising=False) + with patch("runpod.http_client.SyncClientSession") as session_cls: + rp_fitness._report_unhealthy("_memory_check", "RuntimeError: low") + session_cls.assert_not_called() + + +def test_report_unhealthy_swallows_errors(monkeypatch): + monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping") + monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123") + fake_session = MagicMock() + fake_session.get.side_effect = RuntimeError("network down") + with patch("runpod.http_client.SyncClientSession", return_value=fake_session): + rp_fitness._report_unhealthy("_disk_check", "RuntimeError: full") # must not raise + + +@pytest.mark.asyncio +async def test_failure_reports_before_exit(): + from runpod.serverless.modules.rp_fitness import register_fitness_check, run_fitness_checks + + @register_fitness_check + def _cuda_init_check(): + raise RuntimeError("device busy") + + with patch("runpod.serverless.modules.rp_fitness._report_unhealthy") as mock_report: + with pytest.raises(SystemExit): + await run_fitness_checks() + + mock_report.assert_called_once() + args = mock_report.call_args.args + assert args[0] == "_cuda_init_check" + assert "RuntimeError" in args[1] and "device busy" in args[1]