feat: add DashServing HTTP adapter

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,13 +40,15 @@ client = [
   "tinker==0.16.1",
 ]
 server = [
+  "httpx>=0.25.0",
   "redis>=5.0",
   "psutil>=5.9.0",
   "pynvml>=11.0.0",
   "opentelemetry-api",
   "opentelemetry-sdk",
   "opentelemetry-exporter-otlp",
   "opentelemetry-instrumentation-logging",
+  "uvicorn>=0.24.0",
 ]
 test = [
   "hypothesis>=6.0",

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']

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

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,
+    }

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
+        }

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