chenchenghao/sglang
1
0
Fork 0
Code Issues Pull Requests Actions 12 Packages Projects Releases Wiki Activity

[smg][ci] rename 3rd models from cloud backend and delete dead code (#16692)

Browse Source
This commit is contained in:
Simo Lin
2026-01-07 18:19:44 -08:00
committed by GitHub
parent eec7dbd31e
commit a08dc5aa10
6 changed files with 65 additions and 333 deletions
Download Patch File Download Diff File Expand all files Collapse all files

87
sgl-model-gateway/e2e_test/backends.py
View File

@@ -1,87 +0,0 @@
"""Cloud runtime configurations for E2E tests.
This module handles cloud API runtimes (OpenAI, xAI) that don't need local GPU workers.
For local runtimes (gRPC, HTTP), use ModelPool from infra/ to launch workers,
then launch the gateway separately pointing to those workers.
Cloud runtimes vs History backends:
- Cloud runtimes: Where models run (openai, xai)
- History backends: Gateway plugin for conversation storage (memory, oracle)
These are orthogonal - any cloud runtime can use any history backend.
"""
from __future__ import annotations
import logging
from typing import Any
from infra import Gateway
logger = logging.getLogger(__name__)
# Cloud runtime configurations (where models run)
CLOUD_RUNTIMES: dict[str, dict[str, Any]] = {
"openai": {
"description": "OpenAI API",
"model": "gpt-5-nano",
"api_key_env": "OPENAI_API_KEY",
},
"xai": {
"description": "xAI API",
"model": "grok-2-latest",
"api_key_env": "XAI_API_KEY",
},
}
# Backward compatibility alias
CLOUD_BACKENDS = CLOUD_RUNTIMES
def get_cloud_runtime_config(runtime: str) -> dict[str, Any]:
"""Get configuration for a cloud runtime."""
if runtime not in CLOUD_RUNTIMES:
raise KeyError(
f"Unknown cloud runtime: {runtime}. Available: {list(CLOUD_RUNTIMES.keys())}"
)
return CLOUD_RUNTIMES[runtime]
def launch_cloud_gateway(
runtime: str, # "openai" or "xai"
*,
history_backend: str = "memory",
extra_args: list[str] | None = None,
timeout: float = 60,
show_output: bool | None = None,
) -> Gateway:
"""Launch gateway with cloud API runtime.
Args:
runtime: Cloud runtime ("openai" or "xai")
history_backend: History storage backend ("memory" or "oracle")
extra_args: Additional router arguments
timeout: Startup timeout in seconds
show_output: Show subprocess output
Returns:
Gateway instance with running router
"""
if runtime not in CLOUD_RUNTIMES:
raise ValueError(f"Unknown cloud runtime: {runtime}")
gateway = Gateway()
gateway.start(
cloud_backend=runtime,
history_backend=history_backend,
timeout=timeout,
show_output=show_output,
extra_args=extra_args,
)
return gateway
# Backward compatibility aliases
get_cloud_backend_config = get_cloud_runtime_config
launch_cloud_backend = launch_cloud_gateway
launch_cloud_router = launch_cloud_gateway

242
sgl-model-gateway/e2e_test/fixtures/router_manager.py
View File

@@ -1,242 +0,0 @@
"""Legacy router manager for integration tests.
DEPRECATED: This module will be removed during e2e_response_api migration.
Use infra.Gateway instead for all router management.
"""
import subprocess
import time
from dataclasses import dataclass
from typing import Dict, List, Optional
import requests
from infra import wait_for_health
from .ports import find_free_port
@dataclass
class ProcHandle:
process: subprocess.Popen
url: str
class RouterManager:
"""Helper to spawn a router process and interact with admin endpoints."""
def __init__(self):
self._children: List[subprocess.Popen] = []
def start_router(
self,
worker_urls: Optional[List[str]] = None,
policy: str = "round_robin",
port: Optional[int] = None,
extra: Optional[Dict] = None,
# PD options
pd_disaggregation: bool = False,
prefill_urls: Optional[List[tuple]] = None,
decode_urls: Optional[List[str]] = None,
prefill_policy: Optional[str] = None,
decode_policy: Optional[str] = None,
) -> ProcHandle:
worker_urls = worker_urls or []
port = port or find_free_port()
cmd = [
"python3",
"-m",
"sglang_router.launch_router",
"--host",
"127.0.0.1",
"--port",
str(port),
"--policy",
policy,
]
# Avoid Prometheus port collisions by assigning a free port per router
prom_port = find_free_port()
cmd.extend(
["--prometheus-port", str(prom_port), "--prometheus-host", "127.0.0.1"]
)
if worker_urls:
cmd.extend(["--worker-urls", *worker_urls])
# PD routing configuration
if pd_disaggregation:
cmd.append("--pd-disaggregation")
if prefill_urls:
for url, bport in prefill_urls:
if bport is None:
cmd.extend(["--prefill", url, "none"])
else:
cmd.extend(["--prefill", url, str(bport)])
if decode_urls:
for url in decode_urls:
cmd.extend(["--decode", url])
if prefill_policy:
cmd.extend(["--prefill-policy", prefill_policy])
if decode_policy:
cmd.extend(["--decode-policy", decode_policy])
# Map supported extras to CLI flags (subset for integration)
if extra:
flag_map = {
"max_payload_size": "--max-payload-size",
"dp_aware": "--dp-aware",
"api_key": "--api-key",
# Health/monitoring
"worker_startup_check_interval": "--worker-startup-check-interval",
# Cache-aware tuning
"cache_threshold": "--cache-threshold",
"balance_abs_threshold": "--balance-abs-threshold",
"balance_rel_threshold": "--balance-rel-threshold",
# Retry
"retry_max_retries": "--retry-max-retries",
"retry_initial_backoff_ms": "--retry-initial-backoff-ms",
"retry_max_backoff_ms": "--retry-max-backoff-ms",
"retry_backoff_multiplier": "--retry-backoff-multiplier",
"retry_jitter_factor": "--retry-jitter-factor",
"disable_retries": "--disable-retries",
# Circuit breaker
"cb_failure_threshold": "--cb-failure-threshold",
"cb_success_threshold": "--cb-success-threshold",
"cb_timeout_duration_secs": "--cb-timeout-duration-secs",
"cb_window_duration_secs": "--cb-window-duration-secs",
"disable_circuit_breaker": "--disable-circuit-breaker",
# Rate limiting
"max_concurrent_requests": "--max-concurrent-requests",
"queue_size": "--queue-size",
"queue_timeout_secs": "--queue-timeout-secs",
"rate_limit_tokens_per_second": "--rate-limit-tokens-per-second",
# mTLS configuration
"client_cert_path": "--client-cert-path",
"client_key_path": "--client-key-path",
"ca_cert_paths": "--ca-cert-paths",
}
for k, v in extra.items():
if v is None:
continue
flag = flag_map.get(k)
if not flag:
continue
if isinstance(v, bool):
if v:
cmd.append(flag)
elif isinstance(v, list):
# Handle list arguments (e.g., ca_cert_paths)
if v: # Only add if list is not empty
cmd.append(flag)
cmd.extend([str(item) for item in v])
else:
cmd.extend([flag, str(v)])
proc = subprocess.Popen(cmd)
self._children.append(proc)
url = f"http://127.0.0.1:{port}"
wait_for_health(url, timeout=30.0, check_interval=0.2)
return ProcHandle(process=proc, url=url)
def add_worker(self, base_url: str, worker_url: str, timeout: float = 30.0) -> None:
r = requests.post(f"{base_url}/workers", json={"url": worker_url})
assert (
r.status_code == 202
), f"add_worker failed: {r.status_code} {r.text}" # ACCEPTED status
payload = r.json()
worker_id = payload.get("worker_id")
assert worker_id, f"add_worker did not return worker_id: {payload}"
# Poll until worker is actually added and healthy
start = time.time()
with requests.Session() as s:
while time.time() - start < timeout:
try:
r = s.get(f"{base_url}/workers/{worker_id}", timeout=2)
if r.status_code == 200:
data = r.json()
# Check if registration job failed
job_status = data.get("job_status")
if job_status and job_status.get("state") == "failed":
raise RuntimeError(
f"Worker registration failed: {job_status.get('message', 'Unknown error')}"
)
# Check if worker is healthy and registered (not just in job queue)
if data.get("is_healthy", False):
return
# Worker not ready yet, continue polling
except requests.RequestException:
pass
time.sleep(0.1)
raise TimeoutError(
f"Worker {worker_url} was not added and healthy after {timeout}s"
)
def remove_worker(
self, base_url: str, worker_url: str, timeout: float = 30.0
) -> None:
# Resolve worker_id from the current registry snapshot
r_list = requests.get(f"{base_url}/workers")
assert (
r_list.status_code == 200
), f"list_workers failed: {r_list.status_code} {r_list.text}"
workers = r_list.json().get("workers", [])
worker_id = next(
(w.get("id") for w in workers if w.get("url") == worker_url), None
)
assert (
worker_id
), f"could not find worker_id for url={worker_url}. workers={workers}"
r = requests.delete(f"{base_url}/workers/{worker_id}")
assert (
r.status_code == 202
), f"remove_worker failed: {r.status_code} {r.text}" # ACCEPTED status
# Poll until worker is actually removed (GET returns 404) or timeout
start = time.time()
last_status = None
with requests.Session() as s:
while time.time() - start < timeout:
try:
r = s.get(f"{base_url}/workers/{worker_id}", timeout=2)
if r.status_code == 404:
# Worker successfully removed
return
elif r.status_code == 200:
# Check if removal job failed
data = r.json()
job_status = data.get("job_status")
if job_status:
last_status = job_status
if job_status.get("state") == "failed":
raise RuntimeError(
f"Worker removal failed: {job_status.get('message', 'Unknown error')}"
)
# Worker still being processed, continue polling
except requests.RequestException:
pass
time.sleep(0.1)
# Provide detailed timeout error with last known status
error_msg = f"Worker {worker_url} was not removed after {timeout}s"
if last_status:
error_msg += f". Last job status: {last_status}"
raise TimeoutError(error_msg)
def list_workers(self, base_url: str) -> list[str]:
r = requests.get(f"{base_url}/workers")
assert r.status_code == 200, f"list_workers failed: {r.status_code} {r.text}"
data = r.json()
# Extract URLs from WorkerInfo objects
workers = data.get("workers", [])
return [w["url"] for w in workers]
def stop_all(self):
for p in self._children:
if p.poll() is None:
p.terminate()
try:
p.wait(timeout=5)
except subprocess.TimeoutExpired:
p.kill()
self._children.clear()

6
sgl-model-gateway/e2e_test/fixtures/setup_backend.py
View File

@@ -373,12 +373,12 @@ def _setup_cloud_backend(
gateway_config: Gateway configuration from marker.
"""
import openai
from backends import CLOUD_RUNTIMES, launch_cloud_gateway
from infra import THIRD_PARTY_MODELS, launch_cloud_gateway
if backend_name not in CLOUD_RUNTIMES:
if backend_name not in THIRD_PARTY_MODELS:
pytest.fail(f"Unknown cloud runtime: {backend_name}")
cfg = CLOUD_RUNTIMES[backend_name]
cfg = THIRD_PARTY_MODELS[backend_name]
api_key_env = cfg.get("api_key_env")
if api_key_env and not os.environ.get(api_key_env):

6
sgl-model-gateway/e2e_test/infra/__init__.py
View File

@@ -26,7 +26,7 @@ from .constants import ( # Enums; Convenience sets; Fixture parameters; Default
Runtime,
WorkerType,
)
from .gateway import Gateway, WorkerInfo
from .gateway import Gateway, WorkerInfo, launch_cloud_gateway
from .gpu_allocator import (
GPUAllocator,
GPUInfo,
@@ -54,6 +54,7 @@ from .model_specs import ( # Default model paths; Model groups
FUNCTION_CALLING_MODELS,
MODEL_SPECS,
REASONING_MODELS,
THIRD_PARTY_MODELS,
)
from .process_utils import (
detect_ib_device,
@@ -121,6 +122,7 @@ __all__ = [
# Gateway
"Gateway",
"WorkerInfo",
"launch_cloud_gateway",
# Default model paths
"DEFAULT_MODEL_PATH",
"DEFAULT_SMALL_MODEL_PATH",
@@ -135,6 +137,8 @@ __all__ = [
"EMBEDDING_MODELS",
"REASONING_MODELS",
"FUNCTION_CALLING_MODELS",
# Third-party models
"THIRD_PARTY_MODELS",
# Evaluation
"run_eval",
]

39
sgl-model-gateway/e2e_test/infra/gateway.py
View File

@@ -554,3 +554,42 @@ class Gateway:
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
self.shutdown()
def launch_cloud_gateway(
runtime: str, # "openai" or "xai"
*,
history_backend: str = "memory",
extra_args: list[str] | None = None,
timeout: float = 60,
show_output: bool | None = None,
) -> Gateway:
"""Launch gateway with cloud API runtime.
Args:
runtime: Cloud runtime ("openai" or "xai")
history_backend: History storage backend ("memory" or "oracle")
extra_args: Additional router arguments
timeout: Startup timeout in seconds
show_output: Show subprocess output
Returns:
Gateway instance with running router
"""
from .model_specs import THIRD_PARTY_MODELS
if runtime not in THIRD_PARTY_MODELS:
raise ValueError(
f"Unknown cloud runtime: {runtime}. "
f"Available: {list(THIRD_PARTY_MODELS.keys())}"
)
gateway = Gateway()
gateway.start(
cloud_backend=runtime,
history_backend=history_backend,
timeout=timeout,
show_output=show_output,
extra_args=extra_args,
)
return gateway

18
sgl-model-gateway/e2e_test/infra/model_specs.py
View File

@@ -131,3 +131,21 @@ DEFAULT_QWEN_FUNCTION_CALLING_MODEL_PATH = MODEL_SPECS["qwen-7b"]["model"]
DEFAULT_MISTRAL_FUNCTION_CALLING_MODEL_PATH = MODEL_SPECS["mistral-7b"]["model"]
DEFAULT_GPT_OSS_MODEL_PATH = MODEL_SPECS["gpt-oss"]["model"]
DEFAULT_EMBEDDING_MODEL_PATH = MODEL_SPECS["embedding"]["model"]
# =============================================================================
# Third-party model configurations (cloud APIs)
# =============================================================================
THIRD_PARTY_MODELS: dict[str, dict] = {
"openai": {
"description": "OpenAI API",
"model": "gpt-5-nano",
"api_key_env": "OPENAI_API_KEY",
},
"xai": {
"description": "xAI API",
"model": "grok-4-fast",
"api_key_env": "XAI_API_KEY",
},
}