From 52176f917b9c3830e9dff6ad23f85ba66ceddc51 Mon Sep 17 00:00:00 2001 From: Yunnglin Date: Fri, 21 Aug 2026 10:30:58 +0800 Subject: [PATCH] feat: add DashServing HTTP adapter --- pyproject.toml | 3 + src/twinkle/server/dashserving/__init__.py | 6 + src/twinkle/server/dashserving/__main__.py | 24 ++ src/twinkle/server/dashserving/app.py | 162 +++++++++++ src/twinkle/server/dashserving/proxy.py | 91 +++++++ src/twinkle/server/dashserving/schemas.py | 29 ++ tests/dashserving/__init__.py | 0 tests/dashserving/test_mock_e2e.py | 300 +++++++++++++++++++++ 8 files changed, 615 insertions(+) create mode 100644 src/twinkle/server/dashserving/__init__.py create mode 100644 src/twinkle/server/dashserving/__main__.py create mode 100644 src/twinkle/server/dashserving/app.py create mode 100644 src/twinkle/server/dashserving/proxy.py create mode 100644 src/twinkle/server/dashserving/schemas.py create mode 100644 tests/dashserving/__init__.py create mode 100644 tests/dashserving/test_mock_e2e.py diff --git a/pyproject.toml b/pyproject.toml index 27a720456..09aa207b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -23,6 +23,7 @@ dependencies = [ [project.scripts] twinkle-server = "twinkle.server.cli:main" twinkle-auto = "twinkle_client.auto:main" +twinkle-dashserving-adapter = "twinkle.server.dashserving.__main__:main" [project.optional-dependencies] megatron = ["megatron-core>=0.12.0", "transformer-engine[pytorch]", "mcore_bridge"] @@ -39,6 +40,7 @@ client = [ "tinker==0.16.1", ] server = [ + "httpx>=0.25.0", "redis>=5.0", "psutil>=5.9.0", "pynvml>=11.0.0", @@ -46,6 +48,7 @@ server = [ "opentelemetry-sdk", "opentelemetry-exporter-otlp", "opentelemetry-instrumentation-logging", + "uvicorn>=0.24.0", ] test = [ "hypothesis>=6.0", diff --git a/src/twinkle/server/dashserving/__init__.py b/src/twinkle/server/dashserving/__init__.py new file mode 100644 index 000000000..5d88c2c5d --- /dev/null +++ b/src/twinkle/server/dashserving/__init__.py @@ -0,0 +1,6 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""DashServing Native HTTP adapter for the Twinkle server.""" + +from .app import create_dashserving_app + +__all__ = ['create_dashserving_app'] diff --git a/src/twinkle/server/dashserving/__main__.py b/src/twinkle/server/dashserving/__main__.py new file mode 100644 index 000000000..93f7d7967 --- /dev/null +++ b/src/twinkle/server/dashserving/__main__.py @@ -0,0 +1,24 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""Standalone launcher for the Twinkle DashServing adapter.""" +from __future__ import annotations + +import os + + +def main() -> None: + import uvicorn + + from .app import create_dashserving_app + + upstream_url = os.getenv('TWINKLE_INTERNAL_URL', 'http://127.0.0.1:8000') + port = int(os.getenv('PORT', '8091')) + timeout_seconds = float(os.getenv('TWINKLE_DS_TIMEOUT_SECONDS', '600')) + app = create_dashserving_app( + upstream_url=upstream_url, + timeout_seconds=timeout_seconds, + ) + uvicorn.run(app, host='0.0.0.0', port=port) + + +if __name__ == '__main__': + main() diff --git a/src/twinkle/server/dashserving/app.py b/src/twinkle/server/dashserving/app.py new file mode 100644 index 000000000..c3cccbb26 --- /dev/null +++ b/src/twinkle/server/dashserving/app.py @@ -0,0 +1,162 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""FastAPI application implementing DashServing Native HTTP for Twinkle.""" +from __future__ import annotations + +import httpx +import json +import uuid +from contextlib import asynccontextmanager +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse +from pydantic import ValidationError +from typing import AsyncIterator + +from twinkle.utils.logger import get_logger +from .proxy import RuntimeProxy +from .schemas import TunnelRequest + +logger = get_logger() + +_DS_REQUEST_ID = 'X-DashServing-Request-Id' +_DS_ATTRIBUTES = 'X-DashServing-Attributes' +_DS_USAGE = 'X-DashServing-Usage' +_DS_STATUS_CODE = 'X-DashServing-Status-Code' +_DS_STATUS_NAME = 'X-DashServing-Status-Name' +_DS_STATUS_MESSAGE = 'X-DashServing-Status-Message' + + +def create_dashserving_app( + *, + upstream_url: str = 'http://127.0.0.1:8000', + timeout_seconds: float = 600.0, + proxy: RuntimeProxy | None = None, +) -> FastAPI: + """Create the standalone DashServing adapter application. + + A caller-provided proxy is useful for tests and is owned by the caller. + Otherwise this application creates and closes its own proxy client. + """ + owns_proxy = proxy is None + tunnel_proxy = proxy or RuntimeProxy( + upstream_url=upstream_url, + timeout_seconds=timeout_seconds, + ) + + @asynccontextmanager + async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + yield + if owns_proxy: + await tunnel_proxy.close() + + app = FastAPI( + title='Twinkle DashServing Adapter', + description='DashServing Native HTTP adapter for the Twinkle server', + lifespan=lifespan, + ) + + @app.get('/health') + async def health() -> JSONResponse: + if await tunnel_proxy.health(): + return JSONResponse({ + 'status': 'healthy', + 'runtime_upstream': 'healthy', + }) + return JSONResponse( + { + 'status': 'unhealthy', + 'runtime_upstream': 'unhealthy', + }, + status_code=503, + ) + + @app.post('/api') + async def native_http(request: Request) -> JSONResponse: + request_id = request.headers.get(_DS_REQUEST_ID) or str(uuid.uuid4()) + + try: + body = await request.json() + tunnel_request = TunnelRequest.model_validate(body) + except (json.JSONDecodeError, ValidationError, ValueError): + return _error_response( + request_id=request_id, + status_code=400, + status_name='InvalidRequest', + message='Invalid tunnel request body.', + ) + + try: + tunnel_response = await tunnel_proxy.forward(tunnel_request) + except httpx.TimeoutException: + return _error_response( + request_id=request_id, + status_code=504, + status_name='UpstreamTimeout', + message='Runtime upstream timed out.', + ) + except httpx.HTTPError: + return _error_response( + request_id=request_id, + status_code=502, + status_name='UpstreamError', + message='Runtime upstream request failed.', + ) + except Exception: + logger.exception('Unhandled DashServing adapter error request_id=%s', request_id) + return _error_response( + request_id=request_id, + status_code=500, + status_name='InternalError', + message='DashServing adapter failed.', + ) + + return JSONResponse( + tunnel_response.model_dump(mode='json'), + status_code=200, + headers=_dashserving_headers( + request_id=request_id, + status_code=200, + status_name='Success', + message='Success.', + ), + ) + + return app + + +def _error_response( + *, + request_id: str, + status_code: int, + status_name: str, + message: str, +) -> JSONResponse: + return JSONResponse( + {'error': { + 'code': status_name, + 'message': message, + }}, + status_code=status_code, + headers=_dashserving_headers( + request_id=request_id, + status_code=status_code, + status_name=status_name, + message=message, + ), + ) + + +def _dashserving_headers( + *, + request_id: str, + status_code: int, + status_name: str, + message: str, +) -> dict[str, str]: + return { + _DS_REQUEST_ID: request_id, + _DS_ATTRIBUTES: '{}', + _DS_USAGE: '{}', + _DS_STATUS_CODE: str(status_code), + _DS_STATUS_NAME: status_name, + _DS_STATUS_MESSAGE: message, + } diff --git a/src/twinkle/server/dashserving/proxy.py b/src/twinkle/server/dashserving/proxy.py new file mode 100644 index 000000000..934c48ec7 --- /dev/null +++ b/src/twinkle/server/dashserving/proxy.py @@ -0,0 +1,91 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""Fixed-upstream proxy used by the DashServing adapter.""" +from __future__ import annotations + +import httpx +from typing import Any + +from .schemas import TunnelRequest, TunnelResponse + +_HOP_BY_HOP_HEADERS = { + 'connection', + 'content-length', + 'host', + 'transfer-encoding', +} + + +class RuntimeProxy: + """Proxy tunnel requests to one configured Twinkle server. + + The target origin is constructor configuration, never request data. This is + the security boundary that prevents the adapter from becoming an open proxy. + """ + + def __init__( + self, + upstream_url: str, + *, + timeout_seconds: float = 600.0, + client: httpx.AsyncClient | None = None, + ) -> None: + self._upstream_url = upstream_url.rstrip('/') + self._owns_client = client is None + self._client = client or httpx.AsyncClient( + timeout=httpx.Timeout(timeout_seconds), + trust_env=False, + ) + + async def close(self) -> None: + if self._owns_client: + await self._client.aclose() + + async def health(self) -> bool: + try: + response = await self._client.get( + f'{self._upstream_url}/api/v1/twinkle/healthz', + timeout=5.0, + ) + return response.status_code == 200 + except httpx.HTTPError: + return False + + async def forward(self, tunnel_request: TunnelRequest) -> TunnelResponse: + headers = self._build_headers(tunnel_request) + request_kwargs: dict[str, Any] = { + 'method': tunnel_request.method, + 'url': f'{self._upstream_url}{tunnel_request.path}', + 'params': tunnel_request.query, + 'headers': headers, + } + if tunnel_request.body is not None: + request_kwargs['json'] = tunnel_request.body + + response = await self._client.request(**request_kwargs) + if not response.content: + response_body = None + else: + try: + response_body = response.json() + except ValueError: + response_body = response.text + + response_headers = { + 'content-type': response.headers.get('content-type', 'application/json'), + } + replica_id = response.headers.get('x-twinkle-replica-id') + if replica_id: + response_headers['x-twinkle-replica-id'] = replica_id + + return TunnelResponse( + status_code=response.status_code, + headers=response_headers, + body=response_body, + ) + + @staticmethod + def _build_headers(tunnel_request: TunnelRequest) -> dict[str, str]: + return { + name: value + for name, value in tunnel_request.headers.items() if name.lower() not in _HOP_BY_HOP_HEADERS + } diff --git a/src/twinkle/server/dashserving/schemas.py b/src/twinkle/server/dashserving/schemas.py new file mode 100644 index 000000000..d65135995 --- /dev/null +++ b/src/twinkle/server/dashserving/schemas.py @@ -0,0 +1,29 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""Minimal private HTTP tunnel models used by ModelScope and DashServing.""" +from __future__ import annotations + +from pydantic import BaseModel, Field, field_validator +from typing import Any + + +class TunnelRequest(BaseModel): + """HTTP request to forward to the internal `/api/v1/` server.""" + + method: str = Field(min_length=1) + path: str + query: dict[str, str] = Field(default_factory=dict) + headers: dict[str, str] = Field(default_factory=dict) + body: Any = None + + @field_validator('path') + @classmethod + def validate_path(cls, path: str) -> str: + if not path.startswith('/api/v1/'): + raise ValueError('path must start with /api/v1/') + return path + + +class TunnelResponse(BaseModel): + status_code: int = Field(ge=100, le=599) + headers: dict[str, str] = Field(default_factory=dict) + body: Any = None diff --git a/tests/dashserving/__init__.py b/tests/dashserving/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/dashserving/test_mock_e2e.py b/tests/dashserving/test_mock_e2e.py new file mode 100644 index 000000000..c25aeb245 --- /dev/null +++ b/tests/dashserving/test_mock_e2e.py @@ -0,0 +1,300 @@ +# Copyright (c) ModelScope Contributors. All rights reserved. +"""Local HTTP tunnel mock using the production DashServing adapter code. + +This HTTP-level test lives outside ``tests/server`` because it does not +need that suite's session-scoped Ray cluster. +""" +from __future__ import annotations + +import json +import uuid +from contextlib import asynccontextmanager +from typing import AsyncIterator + +import httpx +import pytest +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse, Response + +from twinkle.server.dashserving import create_dashserving_app +from twinkle.server.dashserving.proxy import RuntimeProxy +from twinkle.server.dashserving.schemas import TunnelResponse + + +def _build_runtime_mock() -> FastAPI: + """Represent the existing server exposing Twinkle and Tinker APIs.""" + app = FastAPI() + + @app.api_route('/api/v1/{path:path}', methods=['GET', 'POST', 'DELETE']) + async def twinkle_endpoint(path: str, request: Request) -> JSONResponse: + if path == 'twinkle/healthz': + return JSONResponse({'status': 'ok'}) + if path.endswith('/expired'): + return JSONResponse({'detail': 'checkpoint expired'}, status_code=410) + + body = await request.json() if request.method == 'POST' else None + observed_headers = { + name: request.headers.get(name) + for name in ( + 'x-request-id', + 'x-ray-serve-request-id', + 'serve_multiplexed_model_id', + 'serve-multiplexed-model-id', + 'authorization', + 'twinkle-authorization', + 'x-twinkle-session-id', + ) + } + return JSONResponse({ + 'method': request.method, + 'path': request.url.path, + 'query': dict(request.query_params), + 'headers': observed_headers, + 'body': body, + }) + + return app + + +def _build_dashserving_mock(adapter_client: httpx.AsyncClient) -> FastAPI: + """Represent DS forwarding a Native HTTP request to runtime `/api`.""" + app = FastAPI() + + @app.post('/invoke') + async def invoke(request: Request) -> Response: + ds_request_id = request.headers.get('x-mock-ds-request-id') or str(uuid.uuid4()) + adapter_response = await adapter_client.post( + '/api', + content=await request.body(), + headers={ + 'Content-Type': 'application/json', + 'X-DashServing-Request-Id': ds_request_id, + }, + ) + response_headers = { + name: value + for name, value in adapter_response.headers.items() + if name.lower().startswith('x-dashserving-') + } + return Response( + content=adapter_response.content, + status_code=adapter_response.status_code, + headers=response_headers, + media_type='application/json', + ) + + return app + + +def _build_modelscope_mock(ds_client: httpx.AsyncClient) -> FastAPI: + """Represent the ModelScope public route and tunnel codec.""" + app = FastAPI() + + @app.api_route('/tinker/api/v1/{path:path}', methods=['GET', 'POST', 'DELETE']) + @app.api_route('/twinkle/api/v1/{path:path}', methods=['GET', 'POST', 'DELETE']) + async def runtime_route(path: str, request: Request) -> Response: + authorization = request.headers.get('authorization') + twinkle_request_id = request.headers.get('x-request-id') + if not authorization: + return JSONResponse({'detail': 'missing authorization'}, status_code=401) + if not twinkle_request_id: + return JSONResponse({'detail': 'missing x-request-id'}, status_code=400) + + forwarded_headers = { + 'authorization', + 'serve-multiplexed-model-id', + 'serve_multiplexed_model_id', + 'twinkle-authorization', + 'x-ray-serve-request-id', + 'x-request-id', + 'x-twinkle-session-id', + } + tunnel_request = { + 'method': request.method, + # The mounted route already consumed the public prefix. + 'path': f'/api/v1/{path}', + 'query': dict(request.query_params), + 'headers': { + name: value + for name, value in request.headers.items() + if name.lower() in forwarded_headers + }, + 'body': await request.json() if request.method == 'POST' else None, + } + ds_request_id = f'ds-{uuid.uuid4()}' + ds_response = await ds_client.post( + '/invoke', + json=tunnel_request, + headers={'X-Mock-DS-Request-Id': ds_request_id}, + ) + if ds_response.status_code == 504: + return JSONResponse({'detail': 'DashServing timeout'}, status_code=504) + if ds_response.status_code != 200: + return JSONResponse({'detail': 'DashServing invocation failed'}, status_code=502) + + required_ds_headers = { + 'x-dashserving-request-id', + 'x-dashserving-attributes', + 'x-dashserving-usage', + 'x-dashserving-status-code', + 'x-dashserving-status-name', + 'x-dashserving-status-message', + } + if not required_ds_headers.issubset(ds_response.headers): + return JSONResponse({'detail': 'DashServing response error'}, status_code=502) + if ds_response.headers['x-dashserving-request-id'] != ds_request_id: + return JSONResponse({'detail': 'DashServing request ID mismatch'}, status_code=502) + if ds_response.headers['x-dashserving-status-code'] != '200': + return JSONResponse({'detail': 'DashServing status error'}, status_code=502) + attributes = json.loads(ds_response.headers['x-dashserving-attributes']) + if attributes != {}: + return JSONResponse({'detail': 'Unexpected DashServing attributes'}, status_code=502) + + tunnel_response = TunnelResponse.model_validate(ds_response.json()) + response_headers = {} + content_type = tunnel_response.headers.get('content-type') + if content_type: + response_headers['Content-Type'] = content_type + return JSONResponse( + tunnel_response.body, + status_code=tunnel_response.status_code, + headers=response_headers, + ) + + return app + + +@asynccontextmanager +async def _mock_chain() -> AsyncIterator[tuple[httpx.AsyncClient, httpx.AsyncClient]]: + """Build Client -> ModelScope -> DS -> real Adapter -> Twinkle mock.""" + runtime_app = _build_runtime_mock() + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=runtime_app), + base_url='http://runtime-internal', + ) as runtime_client: + proxy = RuntimeProxy('http://runtime-internal', client=runtime_client) + adapter_app = create_dashserving_app(proxy=proxy) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=adapter_app), + base_url='http://adapter', + ) as adapter_client: + ds_app = _build_dashserving_mock(adapter_client) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=ds_app), + base_url='http://dashserving', + ) as ds_client: + modelscope_app = _build_modelscope_mock(ds_client) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=modelscope_app), + base_url='http://modelscope', + ) as user_client: + yield user_client, adapter_client + + +def _client_headers(*, session_id: str | None = None) -> dict[str, str]: + headers = { + 'Authorization': 'Bearer user-token', + 'Twinkle-Authorization': 'Bearer user-token', + 'x-request-id': 'client-sticky-01', + 'X-Ray-Serve-Request-Id': 'client-sticky-01', + 'serve_multiplexed_model_id': 'client-sticky-01', + 'Serve-Multiplexed-Model-Id': 'client-sticky-01', + } + if session_id: + headers['X-Twinkle-Session-Id'] = session_id + return headers + + +@pytest.mark.asyncio +async def test_post_crosses_the_complete_mock_chain() -> None: + async with _mock_chain() as (client, _adapter): + response = await client.post( + '/twinkle/api/v1/model/Qwen/Qwen3.6-27B/twinkle/forward_backward?timeout=20', + headers=_client_headers(session_id='session-123'), + json={'inputs': [], 'adapter_name': 'default'}, + ) + + assert response.status_code == 200 + assert 'x-dashserving-request-id' not in response.headers + observed = response.json() + assert observed['method'] == 'POST' + assert observed['path'] == '/api/v1/model/Qwen/Qwen3.6-27B/twinkle/forward_backward' + assert observed['query'] == {'timeout': '20'} + assert observed['body'] == {'inputs': [], 'adapter_name': 'default'} + assert observed['headers'] == { + 'x-request-id': 'client-sticky-01', + 'x-ray-serve-request-id': 'client-sticky-01', + 'serve_multiplexed_model_id': 'client-sticky-01', + 'serve-multiplexed-model-id': 'client-sticky-01', + 'authorization': 'Bearer user-token', + 'twinkle-authorization': 'Bearer user-token', + 'x-twinkle-session-id': 'session-123', + } + + +@pytest.mark.asyncio +async def test_get_query_and_inner_status_survive_the_chain() -> None: + async with _mock_chain() as (client, _adapter): + get_response = await client.get( + '/twinkle/api/v1/twinkle/training_runs?limit=20', + headers=_client_headers(), + ) + gone_response = await client.delete( + '/twinkle/api/v1/twinkle/checkpoints/expired', + headers=_client_headers(), + ) + + assert get_response.status_code == 200 + assert get_response.json()['method'] == 'GET' + assert get_response.json()['query'] == {'limit': '20'} + assert get_response.json()['body'] is None + assert gone_response.status_code == 410 + assert gone_response.json() == {'detail': 'checkpoint expired'} + + +@pytest.mark.asyncio +async def test_tinker_request_crosses_the_same_adapter() -> None: + async with _mock_chain() as (client, _adapter): + response = await client.post( + '/tinker/api/v1/create_model', + headers=_client_headers(session_id='session-123'), + json={'base_model': 'Qwen/Qwen3.6-27B'}, + ) + + assert response.status_code == 200 + observed = response.json() + assert observed['path'] == '/api/v1/create_model' + assert observed['body'] == {'base_model': 'Qwen/Qwen3.6-27B'} + assert observed['headers']['authorization'] == 'Bearer user-token' + assert observed['headers']['x-request-id'] == 'client-sticky-01' + + +@pytest.mark.asyncio +async def test_adapter_native_http_contract() -> None: + async with _mock_chain() as (_client, adapter): + tunnel_request = { + 'method': 'POST', + 'path': '/api/v1/create_model', + 'headers': _client_headers(), + 'body': {'base_model': 'Qwen/Qwen3.6-27B'}, + } + response = await adapter.post('/api', json=tunnel_request) + + assert response.status_code == 200 + assert response.headers['x-dashserving-request-id'] + assert response.headers['x-dashserving-status-code'] == '200' + assert response.headers['x-dashserving-status-name'] == 'Success' + assert response.headers['x-dashserving-attributes'] == '{}' + assert response.headers['x-dashserving-usage'] == '{}' + + +@pytest.mark.asyncio +async def test_adapter_health_checks_runtime_upstream() -> None: + async with _mock_chain() as (_client, adapter): + response = await adapter.get('/health') + + assert response.status_code == 200 + assert response.json() == { + 'status': 'healthy', + 'runtime_upstream': 'healthy', + }