From 81b8ed7ca2a86251d609231ce5e85379067ff314 Mon Sep 17 00:00:00 2001 From: Dmitry Meyer Date: Mon, 12 Oct 2026 14:24:32 +0000 Subject: [PATCH] Detect HTTP worker runtime via /v1/models Probe GET /v1/models first and dispatch on the first model's owned_by, the same way SMG detects the runtime. Until the runtime is known, what a gRPC worker, a starting worker, or an unrelated HTTP server produces (request errors, 404, bodies that are not a model list) is logged at debug level. Other status codes, vLLM HTTP workers (not supported, use gRPC), and unknown runtimes are logged as warnings. For SGLang, fall back to the deprecated /get_server_info on 404 (SGLang < 0.5.6), and validate /server_info and gRPC server_args with pydantic models. Once the runtime is detected, every failure is a warning. --- .../services/runs/router_worker_sync.py | 253 ++++++++++++++---- .../services/runs/test_router_worker_sync.py | 234 ++++++++++++++-- 2 files changed, 409 insertions(+), 78 deletions(-) diff --git a/src/dstack/_internal/server/services/runs/router_worker_sync.py b/src/dstack/_internal/server/services/runs/router_worker_sync.py index 83e78e582..bd40ae9e7 100644 --- a/src/dstack/_internal/server/services/runs/router_worker_sync.py +++ b/src/dstack/_internal/server/services/runs/router_worker_sync.py @@ -1,7 +1,6 @@ """Reconcile SGLang router /workers with dstack's ready worker replicas (async, SSH-tunneled).""" import asyncio -import json import logging from collections.abc import Awaitable from contextlib import AsyncExitStack @@ -327,6 +326,7 @@ def _set_target_worker_job_id(worker: "_TargetWorker", job_id: UUID) -> None: _WORKER_BASE_URL = "http://worker" _WORKER_PROBE_TIMEOUT = 30.0 +_WORKER_MAX_MODELS_RESPONSE_BYTES = 256 * 1024 _WORKER_MAX_SERVER_INFO_RESPONSE_BYTES = 256 * 1024 _WORKER_LABEL_DSTACK_JOB_ID = "dstack.ai/job-id" @@ -392,6 +392,39 @@ class _TargetWorker(TypedDict): _RuntimeType = Literal["sglang", "vllm"] +# https://github.com/sgl-project/sglang/blob/e00930c5489053f26d86b179cee0d087f846acbb/python/sglang/srt/entrypoints/openai/protocol.py#L81 +# https://github.com/vllm-project/vllm/blob/db9527a46873454610df6dbedf79a36d6bf1a7f6/vllm/entrypoints/serve/engine/protocol.py#L105 +# Only fields used to detect the runtime are included +class _ModelCard(TypedDict): + owned_by: str + + +# https://github.com/sgl-project/sglang/blob/e00930c5489053f26d86b179cee0d087f846acbb/python/sglang/srt/entrypoints/openai/protocol.py#L93 +# https://github.com/vllm-project/vllm/blob/db9527a46873454610df6dbedf79a36d6bf1a7f6/vllm/entrypoints/serve/engine/protocol.py#L116 +class _ModelList(TypedDict): + data: Annotated[list[_ModelCard], Field(min_length=1)] + + +_model_list_ta = TypeAdapter(_ModelList) + + +# https://github.com/smg-project/smg/blob/3be823a700fabaff3add8a390cf78f163479d686/grpc_servicer/smg_grpc_servicer/sglang/servicer.py#L484 +# Only fields used to build the target worker are included +class _SGLangServerArgs(TypedDict): + disaggregation_mode: NotRequired[Literal["null", "prefill", "decode"]] + disaggregation_bootstrap_port: NotRequired[int] + + +# https://github.com/sgl-project/sglang/blob/e00930c5489053f26d86b179cee0d087f846acbb/python/sglang/srt/entrypoints/http_server.py#L841 +# Server args merged with scheduler info +class _SGLangServerInfo(_SGLangServerArgs): + status: str + + +_sglang_server_args_ta = TypeAdapter(_SGLangServerArgs) +_sglang_server_info_ta = TypeAdapter(_SGLangServerInfo) + + class _ResponseTooLargeError(DstackError): pass @@ -599,66 +632,148 @@ async def _probe_http_worker( worker_url = f"http://{address}" logger.debug("%s: %s: probing", log_prefix, worker_url) + # Until the runtime is detected, failures that are expected from a gRPC worker, a worker that + # is still starting, or an HTTP server that is not a worker at all are logged at debug level: + # they repeat every sync. try: status_code, raw_response = await _request_json_limited( client, "GET", - f"{_WORKER_BASE_URL}/server_info", - max_response_bytes=_WORKER_MAX_SERVER_INFO_RESPONSE_BYTES, + f"{_WORKER_BASE_URL}/v1/models", + max_response_bytes=_WORKER_MAX_MODELS_RESPONSE_BYTES, timeout=_WORKER_PROBE_TIMEOUT, ) except RequestError as e: - logger.debug("%s: %s: GET /server_info: request failed: %r", log_prefix, worker_url, e) + logger.debug("%s: %s: GET /v1/models: request failed: %r", log_prefix, worker_url, e) return None except _ResponseTooLargeError: - logger.warning("%s: %s: GET /server_info: response too large", log_prefix, worker_url) + logger.debug("%s: %s: GET /v1/models: response too large", log_prefix, worker_url) + return None + + if status_code == 404: + logger.debug( + "%s: %s: GET /v1/models: status code: %s", log_prefix, worker_url, status_code + ) return None + # E.g., 401 from a worker that requires an API key if status_code != 200: logger.warning( - "%s: %s: GET /server_info: unexpected status code: %s", + "%s: %s: GET /v1/models: unexpected status code: %s", log_prefix, worker_url, status_code, ) return None - # ValueError, not JSONDecodeError: a body that is not valid UTF-8 raises UnicodeDecodeError try: - data = json.loads(raw_response) - except ValueError as e: - logger.warning( - "%s: %s: GET /server_info: response parsing failed: %s", log_prefix, worker_url, e + response = _model_list_ta.validate_json(raw_response) + except ValidationError as e: + logger.debug( + "%s: %s: GET /v1/models: response parsing failed: %s", log_prefix, worker_url, e ) return None - if isinstance(data, dict): - if data.get("status") != "ready": + + # The same way as SMG detects it, see + # https://github.com/smg-project/smg/blob/3be823a700fabaff3add8a390cf78f163479d686/model_gateway/src/workflow/steps/local/detect_backend.rs#L121 + match owned_by := response["data"][0]["owned_by"]: + case "sglang": + return await _probe_sglang_http_worker(client, log_prefix=log_prefix, address=address) + case "vllm": + logger.warning( + "%s: %s: %s: HTTP workers are not supported, use gRPC", + log_prefix, + worker_url, + owned_by, + ) + return None + case _: + logger.warning( + "%s: %s: GET /v1/models: unsupported runtime: %s", log_prefix, worker_url, owned_by + ) return None - mode = data.get("disaggregation_mode", "") - if mode == "prefill": - bootstrap_port = data.get("disaggregation_bootstrap_port") - worker: _TargetWorker = { - "url": worker_url, - "worker_type": "prefill", - "connection_mode": "http", - "runtime_type": "sglang", - } - if bootstrap_port is not None: - worker["bootstrap_port"] = bootstrap_port - return worker - if mode == "decode": - return { - "url": worker_url, - "worker_type": "decode", - "connection_mode": "http", - "runtime_type": "sglang", - } - return { - "url": worker_url, - "worker_type": "regular", - "connection_mode": "http", - "runtime_type": "sglang", - } + + +async def _probe_sglang_http_worker( + client: AsyncClient, *, log_prefix: str, address: str +) -> Optional[_TargetWorker]: + runtime_type: _RuntimeType = "sglang" + worker_url = f"http://{address}" + + # The runtime is already detected, so any failure is unexpected + path = "/server_info" + try: + status_code, raw_response = await _request_sglang_server_info(client, path=path) + # SGLang < 0.5.6 only serves the deprecated endpoint + if status_code == 404: + path = "/get_server_info" + status_code, raw_response = await _request_sglang_server_info(client, path=path) + except RequestError as e: + logger.warning( + "%s: %s: %s: GET %s: request failed: %r", + log_prefix, + worker_url, + runtime_type, + path, + e, + ) + return None + except _ResponseTooLargeError: + logger.warning( + "%s: %s: %s: GET %s: response too large", + log_prefix, + worker_url, + runtime_type, + path, + ) + return None + + if status_code != 200: + logger.warning( + "%s: %s: %s: GET %s: unexpected status code: %s", + log_prefix, + worker_url, + runtime_type, + path, + status_code, + ) + return None + + try: + response = _sglang_server_info_ta.validate_json(raw_response) + except ValidationError as e: + logger.warning( + "%s: %s: %s: GET %s: response parsing failed: %s", + log_prefix, + worker_url, + runtime_type, + path, + e, + ) + return None + + if response["status"] != "ready": + logger.warning( + "%s: %s: %s: GET %s: status is not ready: %s", + log_prefix, + worker_url, + runtime_type, + path, + response["status"], + ) + return None + + return _get_sglang_target_worker(response, url=worker_url, connection_mode="http") + + +async def _request_sglang_server_info(client: AsyncClient, *, path: str) -> tuple[int, bytes]: + return await _request_json_limited( + client, + "GET", + f"{_WORKER_BASE_URL}{path}", + max_response_bytes=_WORKER_MAX_SERVER_INFO_RESPONSE_BYTES, + timeout=_WORKER_PROBE_TIMEOUT, + ) async def _probe_sglang_grpc_worker( @@ -667,7 +782,7 @@ async def _probe_sglang_grpc_worker( runtime_type: _RuntimeType = "sglang" # The RPC goes over the tunnel, `worker_url` is the address the router itself dials. worker_url = f"grpc://{address}" - logger.debug("%s: %s: probing %s", log_prefix, worker_url, runtime_type) + logger.debug("%s: %s: %s: probing", log_prefix, worker_url, runtime_type) stub = sglang_scheduler_pb2_grpc.SglangSchedulerStub(channel) request = sglang_scheduler_pb2.GetServerInfoRequest() @@ -675,12 +790,37 @@ async def _probe_sglang_grpc_worker( try: response = await stub.GetServerInfo(request, timeout=_WORKER_PROBE_TIMEOUT) except grpc.aio.AioRpcError as e: - _log_failed_grpc_worker_probe( - log_prefix=log_prefix, worker_url=worker_url, runtime_type=runtime_type, error=e + _log_failed_grpc_worker_request( + log_prefix=log_prefix, + worker_url=worker_url, + runtime_type=runtime_type, + method="GetServerInfo", + error=e, + ) + return None + + # MessageToDict raises ValueError, validate_python raises ValidationError, which is a subclass + # of ValueError + try: + server_args = _sglang_server_args_ta.validate_python( + MessageToDict(response.server_args, preserving_proto_field_name=True) + ) + except ValueError as e: + logger.warning( + "%s: %s: %s: GetServerInfo: server_args parsing failed: %s", + log_prefix, + worker_url, + runtime_type, + e, ) return None - server_args = MessageToDict(response.server_args, preserving_proto_field_name=True) + return _get_sglang_target_worker(server_args, url=worker_url, connection_mode="grpc") + + +def _get_sglang_target_worker( + server_args: _SGLangServerArgs, *, url: str, connection_mode: _ConnectionMode +) -> _TargetWorker: worker_type: _WorkerType match disaggregation_mode := server_args.get("disaggregation_mode"): case "prefill" | "decode": @@ -688,15 +828,15 @@ async def _probe_sglang_grpc_worker( case _: worker_type = "regular" worker: _TargetWorker = { - "url": worker_url, + "url": url, "worker_type": worker_type, - "connection_mode": "grpc", - "runtime_type": runtime_type, + "connection_mode": connection_mode, + "runtime_type": "sglang", } if worker_type == "prefill": bootstrap_port = server_args.get("disaggregation_bootstrap_port") if bootstrap_port is not None: - worker["bootstrap_port"] = int(bootstrap_port) + worker["bootstrap_port"] = bootstrap_port return worker @@ -706,7 +846,7 @@ async def _probe_vllm_grpc_worker( runtime_type: _RuntimeType = "vllm" # The RPC goes over the tunnel, `worker_url` is the address the router itself dials. worker_url = f"grpc://{address}" - logger.debug("%s: %s: probing %s", log_prefix, worker_url, runtime_type) + logger.debug("%s: %s: %s: probing", log_prefix, worker_url, runtime_type) stub = vllm_engine_pb2_grpc.VllmEngineStub(channel) request = vllm_engine_pb2.GetServerInfoRequest() @@ -714,8 +854,12 @@ async def _probe_vllm_grpc_worker( try: response = await stub.GetServerInfo(request, timeout=_WORKER_PROBE_TIMEOUT) except grpc.aio.AioRpcError as e: - _log_failed_grpc_worker_probe( - log_prefix=log_prefix, worker_url=worker_url, runtime_type=runtime_type, error=e + _log_failed_grpc_worker_request( + log_prefix=log_prefix, + worker_url=worker_url, + runtime_type=runtime_type, + method="GetServerInfo", + error=e, ) return None @@ -740,16 +884,21 @@ async def _probe_vllm_grpc_worker( return worker -def _log_failed_grpc_worker_probe( - log_prefix: str, worker_url: str, runtime_type: _RuntimeType, error: grpc.aio.AioRpcError +def _log_failed_grpc_worker_request( + log_prefix: str, + worker_url: str, + runtime_type: _RuntimeType, + method: str, + error: grpc.aio.AioRpcError, ): log_level = logging.DEBUG if _is_expected_grpc_error(error) else logging.WARNING logger.log( log_level, - "%s: %s: %s probe failed: %s: %s", + "%s: %s: %s: %s: request failed: %s: %s", log_prefix, worker_url, runtime_type, + method, error.code(), error.details(), ) diff --git a/src/tests/_internal/server/services/runs/test_router_worker_sync.py b/src/tests/_internal/server/services/runs/test_router_worker_sync.py index 9e4c58320..0bdd03ffa 100644 --- a/src/tests/_internal/server/services/runs/test_router_worker_sync.py +++ b/src/tests/_internal/server/services/runs/test_router_worker_sync.py @@ -33,6 +33,7 @@ _get_worker_jobs_with_addresses, _probe_http_worker, _probe_sglang_grpc_worker, + _probe_sglang_http_worker, _probe_vllm_grpc_worker, _probe_worker_replica, _remove_worker_from_router, @@ -277,7 +278,7 @@ async def test_skips_probing_when_no_router_returns_workers( class TestProbeWorkerReplica: async def test_detects_http_sglang_worker(self): with _fake_replica_transports( - http_handler=lambda _: httpx.Response(200, json={"status": "ready"}), + http_handler=_sglang_http_handler({"status": "ready"}), # The HTTP/1.1 server answers the HTTP/2 preface with an error grpc_channel=_FakeGrpcChannel(unknown_method_code=grpc.StatusCode.UNAVAILABLE), ): @@ -324,7 +325,7 @@ async def test_detects_vllm_grpc_worker(self): async def test_returns_none_when_no_probe_succeeds(self): with _fake_replica_transports( - http_handler=lambda _: httpx.Response(200, json={"status": "starting"}), + http_handler=_sglang_http_handler({"status": "starting"}), grpc_channel=_FakeGrpcChannel(unknown_method_code=grpc.StatusCode.UNAVAILABLE), ): assert await _probe_worker_replica(_job(), address="10.0.0.1:8000") is None @@ -332,7 +333,7 @@ async def test_returns_none_when_no_probe_succeeds(self): async def test_cancels_pending_probes_after_success(self): grpc_channel = _FakeGrpcChannel(hang=True) with _fake_replica_transports( - http_handler=lambda _: httpx.Response(200, json={"status": "ready"}), + http_handler=_sglang_http_handler({"status": "ready"}), grpc_channel=grpc_channel, ): worker = await _probe_worker_replica(_job(), address="10.0.0.1:8000") @@ -690,6 +691,87 @@ async def test_logs_truncated_body_on_unexpected_status( @pytest.mark.asyncio class TestProbeHttpWorker: + async def test_detects_sglang_worker(self): + async with _mock_client(_sglang_http_handler({"status": "ready"})) as client: + worker = await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + assert worker == { + "url": "http://10.0.0.1:8000", + "worker_type": "regular", + "connection_mode": "http", + "runtime_type": "sglang", + } + + @pytest.mark.parametrize( + "response", + [ + pytest.param(httpx.Response(404), id="not-found"), + pytest.param(httpx.Response(200, content=b""), id="not-json"), + pytest.param(httpx.Response(200, json={"status": "ok"}), id="not-model-list"), + pytest.param(httpx.Response(200, json={"data": []}), id="no-models"), + pytest.param(httpx.Response(200, json={"data": [{"id": "m"}]}), id="no-owned-by"), + pytest.param(httpx.Response(200, content=b" " * (256 * 1024 + 1)), id="too-large"), + ], + ) + async def test_returns_none_quietly_when_not_worker( + self, caplog: pytest.LogCaptureFixture, response: httpx.Response + ): + caplog.set_level(level=logging.DEBUG, logger=router_worker_sync.__name__) + async with _mock_client(lambda _: response) as client: + assert ( + await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + is None + ) + # Repeats every sync if the service port is served by an HTTP server that is not a + # worker, so it must not be logged as an error. + assert "Worker: http://10.0.0.1:8000: GET /v1/models: " in caplog.text + assert not [r for r in caplog.records if r.levelno > logging.DEBUG] + + async def test_returns_none_quietly_on_request_error(self, caplog: pytest.LogCaptureFixture): + caplog.set_level(level=logging.DEBUG, logger=router_worker_sync.__name__) + async with _mock_client(_grpc_server_http_handler) as client: + assert ( + await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + is None + ) + # Repeats every sync until the worker is up, or forever if it is a gRPC worker, so it + # must not be logged as an error, see https://github.com/dstackai/dstack/issues/4300. + assert "GET /v1/models: request failed" in caplog.text + assert not [r for r in caplog.records if r.levelno > logging.DEBUG] + + @pytest.mark.parametrize( + ["response", "message"], + [ + pytest.param( + # A worker that requires an API key + httpx.Response(401), + "Worker: http://10.0.0.1:8000: GET /v1/models: unexpected status code: 401", + id="unauthorized", + ), + pytest.param( + httpx.Response(200, json={"data": [{"owned_by": "vllm"}]}), + "Worker: http://10.0.0.1:8000: vllm: HTTP workers are not supported, use gRPC", + id="vllm", + ), + pytest.param( + httpx.Response(200, json={"data": [{"owned_by": "tgi"}]}), + "Worker: http://10.0.0.1:8000: GET /v1/models: unsupported runtime: tgi", + id="unsupported-runtime", + ), + ], + ) + async def test_warns_when_worker_not_supported( + self, caplog: pytest.LogCaptureFixture, response: httpx.Response, message: str + ): + async with _mock_client(lambda _: response) as client: + assert ( + await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + is None + ) + assert message in caplog.text + + +@pytest.mark.asyncio +class TestProbeSglangHttpWorker: """ The probe talks to the replica over the tunnel, but the `url` it reports is the address the router dials. @@ -721,7 +803,9 @@ class TestProbeHttpWorker: ) async def test_reports_worker(self, server_info: dict, expected: dict): async with _mock_client(lambda _: httpx.Response(200, json=server_info)) as client: - worker = await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + worker = await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) assert worker == { "url": "http://10.0.0.1:8000", "connection_mode": "http", @@ -729,54 +813,109 @@ async def test_reports_worker(self, server_info: dict, expected: dict): **expected, } - async def test_returns_none_when_not_ready(self): + async def test_falls_back_to_deprecated_endpoint(self): + # SGLang < 0.5.6 + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/get_server_info": + return httpx.Response(200, json={"status": "ready"}) + return httpx.Response(404) + + async with _mock_client(handler) as client: + worker = await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) + assert worker == { + "url": "http://10.0.0.1:8000", + "worker_type": "regular", + "connection_mode": "http", + "runtime_type": "sglang", + } + + async def test_returns_none_when_not_ready(self, caplog: pytest.LogCaptureFixture): response = httpx.Response(200, json={"status": "starting"}) async with _mock_client(lambda _: response) as client: assert ( - await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) is None ) + assert ( + "Worker: http://10.0.0.1:8000: sglang: GET /server_info: status is not ready: starting" + in caplog.text + ) @pytest.mark.parametrize( "body", [ pytest.param(b"not json", id="not-json"), - # Detected as UTF-16 by its BOM, but not valid UTF-16 - pytest.param(b"\xff\xfe\xfa", id="not-text"), + pytest.param(b"\xff\xfe\xfa", id="not-utf-8"), + pytest.param(b"[]", id="not-object"), + pytest.param(b"{}", id="no-status"), + pytest.param( + json.dumps({"status": "ready", "disaggregation_mode": "encode"}).encode(), + id="unknown-disaggregation-mode", + ), + pytest.param( + json.dumps( + { + "status": "ready", + "disaggregation_mode": "prefill", + "disaggregation_bootstrap_port": "auto", + } + ).encode(), + id="non-integer-bootstrap-port", + ), ], ) - async def test_returns_none_on_unparsable_body( + async def test_returns_none_on_invalid_body( self, caplog: pytest.LogCaptureFixture, body: bytes ): async with _mock_client(lambda _: httpx.Response(200, content=body)) as client: assert ( - await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) is None ) assert ( - "Worker: http://10.0.0.1:8000: GET /server_info: response parsing failed" + "Worker: http://10.0.0.1:8000: sglang: GET /server_info: response parsing failed" in caplog.text ) - async def test_returns_none_on_unexpected_status(self, caplog: pytest.LogCaptureFixture): - async with _mock_client(lambda _: httpx.Response(404)) as client: + @pytest.mark.parametrize( + ["status_code", "message"], + [ + pytest.param(500, "GET /server_info: unexpected status code: 500", id="error"), + pytest.param(404, "GET /get_server_info: unexpected status code: 404", id="not-found"), + ], + ) + async def test_returns_none_on_unexpected_status( + self, caplog: pytest.LogCaptureFixture, status_code: int, message: str + ): + async with _mock_client(lambda _: httpx.Response(status_code)) as client: assert ( - await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) is None ) - assert "GET /server_info: unexpected status code: 404" in caplog.text + assert f"Worker: http://10.0.0.1:8000: sglang: {message}" in caplog.text async def test_returns_none_on_request_error(self, caplog: pytest.LogCaptureFixture): - caplog.set_level(level=logging.DEBUG, logger=router_worker_sync.__name__) - async with _mock_client(_grpc_server_http_handler) as client: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadError("connection reset") + + async with _mock_client(handler) as client: assert ( - await _probe_http_worker(client, log_prefix="Worker", address="10.0.0.1:8000") + await _probe_sglang_http_worker( + client, log_prefix="Worker", address="10.0.0.1:8000" + ) is None ) - # Repeats every sync until the worker is up, or forever if it is a gRPC worker, so it - # must not be logged as an error, see https://github.com/dstackai/dstack/issues/4300. - assert "GET /server_info: request failed" in caplog.text - assert not [r for r in caplog.records if r.levelno > logging.DEBUG] + assert ( + "Worker: http://10.0.0.1:8000: sglang: GET /server_info: request failed" in caplog.text + ) @pytest.mark.asyncio @@ -850,7 +989,37 @@ async def test_warns_on_unexpected_error(self, caplog: pytest.LogCaptureFixture) is None ) assert ( - "Worker: grpc://10.0.0.1:8000: sglang probe failed: StatusCode.PERMISSION_DENIED" + "Worker: grpc://10.0.0.1:8000: sglang: GetServerInfo: request failed:" + " StatusCode.PERMISSION_DENIED" + ) in caplog.text + + @pytest.mark.parametrize( + "server_args", + [ + pytest.param({"disaggregation_mode": "encode"}, id="unknown-disaggregation-mode"), + pytest.param( + {"disaggregation_mode": "prefill", "disaggregation_bootstrap_port": 8998.5}, + id="non-integer-bootstrap-port", + ), + # `MessageToDict` fails on any non-finite number, including in fields the probe + # doesn't use + pytest.param( + {"disaggregation_mode": "null", "max_total_tokens": float("inf")}, + id="non-finite-number", + ), + ], + ) + async def test_returns_none_on_invalid_server_args( + self, caplog: pytest.LogCaptureFixture, server_args: dict + ): + response = sglang_scheduler_pb2.GetServerInfoResponse(server_args=server_args) + channel = _FakeGrpcChannel({_SGLANG_GET_SERVER_INFO: response}) + assert ( + await _probe_sglang_grpc_worker(channel, log_prefix="Worker", address="10.0.0.1:8000") + is None + ) + assert ( + "Worker: grpc://10.0.0.1:8000: sglang: GetServerInfo: server_args parsing failed" in caplog.text ) @@ -937,9 +1106,9 @@ async def test_warns_on_unexpected_error(self, caplog: pytest.LogCaptureFixture) is None ) assert ( - "Worker: grpc://10.0.0.1:8000: vllm probe failed: StatusCode.PERMISSION_DENIED" - in caplog.text - ) + "Worker: grpc://10.0.0.1:8000: vllm: GetServerInfo: request failed:" + " StatusCode.PERMISSION_DENIED" + ) in caplog.text def _service_configuration() -> ServiceConfiguration: @@ -1148,6 +1317,19 @@ async def get_grpc_channel(uds_path: Path) -> AsyncIterator["_FakeGrpcChannel"]: yield +def _sglang_http_handler(server_info: dict) -> Callable[[httpx.Request], httpx.Response]: + """An SGLang HTTP worker replica that reports the given server info""" + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/v1/models": + return httpx.Response(200, json={"data": [{"id": "model", "owned_by": "sglang"}]}) + if request.url.path == "/server_info": + return httpx.Response(200, json=server_info) + return httpx.Response(404) + + return handler + + def _grpc_server_http_handler(request: httpx.Request) -> httpx.Response: # What an HTTP/1.1 request to an HTTP/2-only gRPC server produces raise httpx.RemoteProtocolError("illegal request line")