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")