Skip to content
Merged
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 38 additions & 3 deletions miles/router/router.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
import argparse
import asyncio
import json
import logging

import httpx
import uvicorn
Expand All @@ -9,6 +11,8 @@

from miles.utils.misc import load_function

logger = logging.getLogger(__name__)


def run_router(args):
"""
Expand All @@ -28,6 +32,7 @@ def __init__(self, args, verbose=False):
self.verbose = verbose

self.app = FastAPI()
self.app.add_event_handler("startup", self._start_background_health_check)

# Worker information
self.worker_urls: dict[str, int] = {}
Expand Down Expand Up @@ -63,9 +68,39 @@ def _setup_routes(self):
# Catch-all route for proxying to SGLang - must be registered LAST
self.app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])(self.proxy)

async def health_check(self, request: Request):
# TODO: do health check in background
pass
async def _start_background_health_check(self):
asyncio.create_task(self._health_check_loop())

async def _check_worker_health(self, url):
"""Encapsulated health check logic for better maintainability."""
try:
response = await self.client.get(f"{url}/health", timeout=5.0)
if response.status_code == 200:
return url, True
logger.warning(f"[miles-router] Worker {url} is unhealthy (Status: {response.status_code})")
except Exception as e:
logger.warning(f"[miles-router] Worker {url} health check failed: {e}")
return url, False

async def _health_check_loop(self):
interval = self.args.rollout_health_check_interval

while True:
await asyncio.sleep(interval)

urls = list(self.worker_urls.keys())
if not urls:
continue

# Concurrent execution using a generator expression
results = await asyncio.gather(*(self._check_worker_health(url) for url in urls))

for url, is_healthy in results:
if not is_healthy:
logger.warning(f"[miles-router] Removing unhealthy worker from pool: {url}")
self.worker_urls.pop(url, None)

logger.info(f"[miles-router] Health check complete. {len(self.worker_urls)} workers active.")

async def proxy(self, request: Request, path: str):
"""Proxy all other requests to the SGLang router"""
Expand Down