iHore/vLLM
8
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[Logs] DP Supervisor Log Improvement (#48278)

Browse Source 
Signed-off-by: Robert Shaw <robertgshaw2-redhat@ip-172-31-18-125.us-east-2.compute.internal>
Co-authored-by: Robert Shaw <robertgshaw2-redhat@ip-172-31-18-125.us-east-2.compute.internal>
This commit is contained in:
Robert Shaw
2026-07-11 12:07:00 +08:00
committed by GitHub
co-authored by Robert Shaw
parent 04d553f390
commit 3d99b0499a
2 changed files with 96 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/entrypoints/openai/test_dp_supervisor.py
+50 -30
View File
@@ -301,9 +301,20 @@ async def test_shutdown_if_supervisor_server_error_on_startup(
async def fake_shutdown_children(self):
return None
def fake_start_children(self):
return None
async def fake_monitor_children(self):
# Mark ready so the supervisor server is started, then block until
# shutdown (triggered when the failing server task exits).
self._is_ready = True
await self._shutdown_event.wait()
monkeypatch.setattr(dp_sup.asyncio, "get_running_loop", lambda: FakeLoop())
monkeypatch.setattr(dp_sup.uvicorn, "Server", FakeServer)
monkeypatch.setattr(DPSupervisor, "_shutdown_children", fake_shutdown_children)
monkeypatch.setattr(DPSupervisor, "_start_children", fake_start_children)
monkeypatch.setattr(DPSupervisor, "_monitor_children", fake_monitor_children)
supervisor = DPSupervisor(_make_unit_args())
@@ -437,8 +448,11 @@ def launch_mock_vllm_with_drain(
async def _poll_supervisor_health(expected_status: int, use_ssl: bool = False) -> bool:
"""
Poll GET /health on the supervisor until expected_status is seen.
A connection error is treated as 503-equivalent when expected_status != 200.
GET /health on the supervisor once and check for expected_status.
Pass expected_status=-1 to assert the supervisor is not listening yet
(a connection error is expected). The supervisor only starts its HTTP
server once every child is ready, so /health is refused until then.
"""
scheme = "https" if use_ssl else "http"
url = f"{scheme}://127.0.0.1:{_SUPERVISOR_PORT}/health"
@@ -456,6 +470,17 @@ async def _poll_supervisor_health(expected_status: int, use_ssl: bool = False) -
return True
async def _await_supervisor_health(
expected_status: int, use_ssl: bool = False, retries: int = 20
) -> bool:
"""Retry _poll_supervisor_health, tolerating supervisor server startup."""
for _ in range(retries):
if await _poll_supervisor_health(expected_status, use_ssl=use_ssl):
return True
await asyncio.sleep(0.5)
return False
async def _poll_until_api_server_running(
port: int, retries: int = 10, use_ssl: bool = False
) -> None:
@@ -534,7 +559,7 @@ async def _run_supervisor(
@pytest.mark.asyncio
async def test_basic_lifecycle(monkeypatch):
"""
A) Supervisor /health returns 503 while children are unhealthy.
A) Supervisor is not listening while children are unhealthy.
B) /health returns 200 once every child reports healthy.
C) SIGTERM and shutdown
"""
@@ -543,24 +568,23 @@ async def test_basic_lifecycle(monkeypatch):
vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
for port in vllm_server_ports:
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
await _poll_until_api_server_running(port)
await _set_healthy(vllm_server_ports[0])
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
print("/health is 503 --- expected!")
print("supervisor not listening --- expected!")
for port in vllm_server_ports:
await _set_healthy(port)
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(200)
assert await _await_supervisor_health(200)
assert supervisor.is_ready
print("/health is 200 --- expected!")
@@ -589,19 +613,18 @@ async def test_basic_lifecycle_with_ssl(monkeypatch):
vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
assert await _poll_supervisor_health(503, use_ssl=True)
assert await _poll_supervisor_health(-1, use_ssl=True)
assert not supervisor.is_ready
for port in vllm_server_ports:
assert await _poll_supervisor_health(503, use_ssl=True)
assert await _poll_supervisor_health(-1, use_ssl=True)
assert not supervisor.is_ready
await _poll_until_api_server_running(port, use_ssl=True)
for port in vllm_server_ports:
await _set_healthy(port, use_ssl=True)
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(200, use_ssl=True)
assert await _await_supervisor_health(200, use_ssl=True)
assert supervisor.is_ready
@@ -616,7 +639,7 @@ async def test_failed_startup(monkeypatch):
vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
for port in vllm_server_ports:
@@ -632,7 +655,7 @@ async def test_failed_startup(monkeypatch):
@pytest.mark.asyncio
async def test_becomes_unhealthy(monkeypatch):
"""
A) Supervisor /health returns 503 while children are unhealthy.
A) Supervisor is not listening while children are unhealthy.
B) /health returns 200 once every child reports healthy.
C) Child process becomes unhealthy.
D) Detected and shutdown.
@@ -642,24 +665,23 @@ async def test_becomes_unhealthy(monkeypatch):
vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
for port in vllm_server_ports:
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
await _poll_until_api_server_running(port)
await _set_healthy(vllm_server_ports[0])
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
print("/health is 503 --- expected!")
print("supervisor not listening --- expected!")
for port in vllm_server_ports:
await _set_healthy(port)
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(200)
assert await _await_supervisor_health(200)
assert supervisor.is_ready
print("/health is 200 --- expected!")
@@ -674,7 +696,7 @@ async def test_becomes_unhealthy(monkeypatch):
@pytest.mark.asyncio
async def test_dp_server_fails(monkeypatch):
"""
A) Supervisor /health returns 503 while children are unhealthy.
A) Supervisor is not listening while children are unhealthy.
B) /health returns 200 once every child reports healthy.
C) Child process fails.
D) Detected and shutdown.
@@ -684,24 +706,23 @@ async def test_dp_server_fails(monkeypatch):
vllm_server_ports = [_CHILD_PORT_BASE + i for i in range(_N_CHILDREN)]
async with _run_supervisor(args, monkeypatch) as (supervisor, _task):
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
for port in vllm_server_ports:
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
await _poll_until_api_server_running(port)
await _set_healthy(vllm_server_ports[0])
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(503)
assert await _poll_supervisor_health(-1)
assert not supervisor.is_ready
print("/health is 503 --- expected!")
print("supervisor not listening --- expected!")
for port in vllm_server_ports:
await _set_healthy(port)
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(200)
assert await _await_supervisor_health(200)
assert supervisor.is_ready
print("/health is 200 --- expected!")
@@ -739,8 +760,7 @@ async def test_shutdown_timeout(monkeypatch: pytest.MonkeyPatch):
for port in vllm_server_ports:
await _set_healthy(port)
await asyncio.sleep(1.0)
assert await _poll_supervisor_health(200)
assert await _await_supervisor_health(200)
assert supervisor.is_ready
start_t = time.perf_counter()
vllm/entrypoints/openai/dp_supervisor.py
+46 -12
View File
@@ -283,14 +283,46 @@ class DPSupervisor:
async def run(self) -> None:
loop = asyncio.get_running_loop()
decorate_logs("DPSupervisor")
# K8s sends SIGTERM for shutdown - begin graceful termination.
for sig in (signal.SIGTERM, signal.SIGINT):
loop.add_signal_handler(sig, partial(self._handle_signal, sig))
# Launch DPSupervisor Server.
supervisor_server: uvicorn.Server | None = None
supervisor_server_task: asyncio.Task[None] | None = None
try:
# Launch vLLM DP Servers and begin monitoring them.
self._start_children()
monitor_task = asyncio.create_task(
self._monitor_children(), name="dp-monitor"
)
# Only start the DPSupervisor server once the vLLM DP Servers are
# ready. This avoids the supervisor /health endpoint returning 503
# to external load balancer probes while the engines initialize.
await self._wait_until_ready(monitor_task)
if self.is_ready and not monitor_task.done():
supervisor_server, supervisor_server_task = await self._start_server()
await monitor_task
finally:
self._is_ready = False
await self._shutdown_children()
# Shutdown the DP Supervisor server.
if supervisor_server is not None and supervisor_server_task is not None:
supervisor_server.should_exit = True
await supervisor_server_task
async def _start_server(self) -> tuple[uvicorn.Server, asyncio.Task[None]]:
"""
Launch the DPSupervisor HTTP server.
Called only after the vLLM DP Servers are ready so that /health does
not return 503 to external probes while the engines are initializing.
"""
app = _build_dp_supervisor_app(self)
decorate_logs("DPSupervisor")
host = self.args.host or "0.0.0.0"
config = uvicorn.Config(
app,
@@ -319,18 +351,20 @@ class DPSupervisor:
raise RuntimeError("DPSupervisor exited before startup.")
await asyncio.sleep(0.05)
logger.info("Started DPSupervisor on %s:%d", host, self.supervisor_port)
return supervisor_server, supervisor_server_task
# Launch and Monitor vLLM Server Processes.
try:
self._start_children()
await self._monitor_children()
finally:
self._is_ready = False
await self._shutdown_children()
async def _wait_until_ready(self, monitor_task: asyncio.Task[None]) -> None:
"""
Block until the vLLM DP Servers are ready or shutdown is triggered.
# Shutdown the DP Supervisor server.
supervisor_server.should_exit = True
await supervisor_server_task
Returns early if monitoring stops (e.g. a DP Server dies during
startup), in which case the supervisor server is never started.
"""
logger.info("Waiting for vLLM DP Servers to become ready.")
while not self._is_ready and not self._shutdown_event.is_set():
if monitor_task.done():
return
await asyncio.sleep(0.05)
def _handle_signal(self, signum: int) -> None:
"""