diff --git a/sgl-model-gateway/e2e_test/backends.py b/sgl-model-gateway/e2e_test/backends.py deleted file mode 100644 index 588c5be5e..000000000 --- a/sgl-model-gateway/e2e_test/backends.py +++ /dev/null @@ -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 diff --git a/sgl-model-gateway/e2e_test/fixtures/router_manager.py b/sgl-model-gateway/e2e_test/fixtures/router_manager.py deleted file mode 100644 index ed7e7ace4..000000000 --- a/sgl-model-gateway/e2e_test/fixtures/router_manager.py +++ /dev/null @@ -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() diff --git a/sgl-model-gateway/e2e_test/fixtures/setup_backend.py b/sgl-model-gateway/e2e_test/fixtures/setup_backend.py index 2f61f1bf0..bd481baf7 100644 --- a/sgl-model-gateway/e2e_test/fixtures/setup_backend.py +++ b/sgl-model-gateway/e2e_test/fixtures/setup_backend.py @@ -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): diff --git a/sgl-model-gateway/e2e_test/infra/__init__.py b/sgl-model-gateway/e2e_test/infra/__init__.py index a9a2cf801..a138b3ce2 100644 --- a/sgl-model-gateway/e2e_test/infra/__init__.py +++ b/sgl-model-gateway/e2e_test/infra/__init__.py @@ -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", ] diff --git a/sgl-model-gateway/e2e_test/infra/gateway.py b/sgl-model-gateway/e2e_test/infra/gateway.py index b46fec184..132ff5967 100644 --- a/sgl-model-gateway/e2e_test/infra/gateway.py +++ b/sgl-model-gateway/e2e_test/infra/gateway.py @@ -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 diff --git a/sgl-model-gateway/e2e_test/infra/model_specs.py b/sgl-model-gateway/e2e_test/infra/model_specs.py index 81c31bba5..cb4417c6d 100644 --- a/sgl-model-gateway/e2e_test/infra/model_specs.py +++ b/sgl-model-gateway/e2e_test/infra/model_specs.py @@ -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", + }, +}