From 55b7936582eeace2953d12f972f512b403f7a5f1 Mon Sep 17 00:00:00 2001 From: Simo Lin Date: Wed, 7 Jan 2026 06:59:49 -0800 Subject: [PATCH] refactor(e2e_test): fix smg ci e2e test code quality (#16664) --- .../e2e_test/benchmarks/conftest.py | 41 +++++++++--- sgl-model-gateway/e2e_test/conftest.py | 12 ++-- .../e2e_test/fixtures/router_manager.py | 16 +---- sgl-model-gateway/e2e_test/infra/__init__.py | 4 ++ sgl-model-gateway/e2e_test/infra/constants.py | 8 +++ sgl-model-gateway/e2e_test/infra/gateway.py | 63 +++++++++---------- .../e2e_test/infra/gpu_allocator.py | 2 +- .../e2e_test/infra/model_pool.py | 28 ++------- sgl-model-gateway/e2e_test/infra/run_eval.py | 4 +- .../e2e_test/infra/simple_eval_common.py | 10 +-- .../e2e_test/infra/simple_eval_mmlu.py | 3 + 11 files changed, 100 insertions(+), 91 deletions(-) diff --git a/sgl-model-gateway/e2e_test/benchmarks/conftest.py b/sgl-model-gateway/e2e_test/benchmarks/conftest.py index 6d99e4a3f..9261ca1a1 100644 --- a/sgl-model-gateway/e2e_test/benchmarks/conftest.py +++ b/sgl-model-gateway/e2e_test/benchmarks/conftest.py @@ -152,13 +152,20 @@ def genai_bench_runner(): ) timeout = timeout_sec or int(os.environ.get("GENAI_BENCH_TEST_TIMEOUT", "120")) - proc = subprocess.Popen( - cmd, - env=os.environ.copy(), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) + try: + proc = subprocess.Popen( + cmd, + env=os.environ.copy(), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except FileNotFoundError: + pytest.fail(f"genai-bench executable not found at {cli}") + except PermissionError: + pytest.fail(f"Permission denied executing {cli}") + except OSError as e: + pytest.fail(f"Failed to start genai-bench: {e}") # Start GPU monitor if needed gpu_monitor: GPUMonitor | None = None @@ -172,6 +179,16 @@ def genai_bench_runner(): except subprocess.TimeoutExpired: proc.kill() stdout, stderr = proc.communicate() + logger.error("genai-bench timed out after %ds", timeout) + + # Log output if process failed or for debugging + if proc.returncode != 0: + logger.error( + "genai-bench exited with code %d\nstdout:\n%s\nstderr:\n%s", + proc.returncode, + stdout or "(empty)", + stderr or "(empty)", + ) try: # Parse and validate results @@ -187,6 +204,16 @@ def genai_bench_runner(): gpu_monitor.log_summary() gpu_monitor.assert_thresholds(thresholds) + except AssertionError: + # Log genai-bench output when results not found + logger.error( + "genai-bench output (returncode=%d):\nstdout:\n%s\nstderr:\n%s", + proc.returncode, + stdout or "(empty)", + stderr or "(empty)", + ) + raise + finally: _cleanup_procs(kill_procs, drain_delay_sec) if gpu_monitor: diff --git a/sgl-model-gateway/e2e_test/conftest.py b/sgl-model-gateway/e2e_test/conftest.py index 7d261a015..98785d7f6 100644 --- a/sgl-model-gateway/e2e_test/conftest.py +++ b/sgl-model-gateway/e2e_test/conftest.py @@ -140,9 +140,9 @@ def pytest_runtest_logstart(nodeid: str, location: tuple) -> None: """Print clear test header at start of each test.""" # Extract test name from nodeid (e.g., "test_mmlu.py::TestMMLU::test_mmlu_basic[grpc]") test_name = nodeid.split("::")[-1] if "::" in nodeid else nodeid - print(f"\n{'='*60}") + print(f"\n{'=' * LOG_SEPARATOR_WIDTH}") print(f"TEST: {test_name}") - print(f"{'='*60}") + print(f"{'=' * LOG_SEPARATOR_WIDTH}") # Path setup for imports @@ -172,6 +172,7 @@ from infra import ( ENV_SKIP_MODEL_POOL, ENV_STARTUP_TIMEOUT, LOCAL_MODES, + LOG_SEPARATOR_WIDTH, PARAM_MODEL, PARAM_SETUP_BACKEND, ConnectionMode, @@ -434,17 +435,18 @@ def pytest_collection_finish(session: pytest.Session) -> None: max_required, available_gpus = validate_gpu_requirements() if max_required > available_gpus: + sep = "=" * LOG_SEPARATOR_WIDTH raise pytest.UsageError( - f"\n{'='*60}\n" + f"\n{sep}\n" f"GPU REQUIREMENTS EXCEEDED\n" - f"{'='*60}\n" + f"{sep}\n" f"Test '{_max_test_name}' requires {max_required} GPUs\n" f"Available: {available_gpus} GPUs\n" f"\nOptions:\n" f" 1. Run tests that fit: pytest -k 'not {_max_test_name.split('::')[0]}'\n" f" 2. Reduce workers: @pytest.mark.workers(prefill=1, decode=1)\n" f" 3. Skip GPU tests: SKIP_MODEL_POOL=1 pytest\n" - f"{'='*60}" + f"{sep}" ) logger.info( diff --git a/sgl-model-gateway/e2e_test/fixtures/router_manager.py b/sgl-model-gateway/e2e_test/fixtures/router_manager.py index 10d0b631f..23e98f579 100644 --- a/sgl-model-gateway/e2e_test/fixtures/router_manager.py +++ b/sgl-model-gateway/e2e_test/fixtures/router_manager.py @@ -4,6 +4,7 @@ from dataclasses import dataclass from typing import Dict, List, Optional import requests +from infra import wait_for_health from .ports import find_free_port @@ -126,22 +127,9 @@ class RouterManager: proc = subprocess.Popen(cmd) self._children.append(proc) url = f"http://127.0.0.1:{port}" - self._wait_health(url) + wait_for_health(url, timeout=30.0, check_interval=0.2) return ProcHandle(process=proc, url=url) - def _wait_health(self, base_url: str, timeout: float = 30.0): - start = time.time() - with requests.Session() as s: - while time.time() - start < timeout: - try: - r = s.get(f"{base_url}/health", timeout=2) - if r.status_code == 200: - return - except requests.RequestException: - pass - time.sleep(0.2) - raise TimeoutError(f"Router at {base_url} did not become healthy") - 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 ( diff --git a/sgl-model-gateway/e2e_test/infra/__init__.py b/sgl-model-gateway/e2e_test/infra/__init__.py index 53ddf9eef..a9a2cf801 100644 --- a/sgl-model-gateway/e2e_test/infra/__init__.py +++ b/sgl-model-gateway/e2e_test/infra/__init__.py @@ -17,6 +17,8 @@ from .constants import ( # Enums; Convenience sets; Fixture parameters; Default HEALTH_CHECK_INTERVAL, LOCAL_MODES, LOCAL_RUNTIMES, + LOG_SEPARATOR_WIDTH, + MAX_RETRY_ATTEMPTS, PARAM_BACKEND_ROUTER, PARAM_MODEL, PARAM_SETUP_BACKEND, @@ -82,6 +84,8 @@ __all__ = [ "DEFAULT_STARTUP_TIMEOUT", "DEFAULT_ROUTER_TIMEOUT", "HEALTH_CHECK_INTERVAL", + "MAX_RETRY_ATTEMPTS", + "LOG_SEPARATOR_WIDTH", # Env vars "ENV_MODELS", "ENV_BACKENDS", diff --git a/sgl-model-gateway/e2e_test/infra/constants.py b/sgl-model-gateway/e2e_test/infra/constants.py index 2fcd85c38..e12f8bb34 100644 --- a/sgl-model-gateway/e2e_test/infra/constants.py +++ b/sgl-model-gateway/e2e_test/infra/constants.py @@ -58,3 +58,11 @@ DEFAULT_HOST = "127.0.0.1" DEFAULT_STARTUP_TIMEOUT = 300 DEFAULT_ROUTER_TIMEOUT = 60 HEALTH_CHECK_INTERVAL = 5 + +# Retry configuration +MAX_RETRY_ATTEMPTS = ( + 6 # Max retries with exponential backoff (total ~63s: 1+2+4+8+16+32) +) + +# Display formatting +LOG_SEPARATOR_WIDTH = 60 # Width for log separator lines (e.g., "="*60) diff --git a/sgl-model-gateway/e2e_test/infra/gateway.py b/sgl-model-gateway/e2e_test/infra/gateway.py index e66ae9f90..067e1e26e 100644 --- a/sgl-model-gateway/e2e_test/infra/gateway.py +++ b/sgl-model-gateway/e2e_test/infra/gateway.py @@ -328,6 +328,30 @@ class Gateway: # Worker Management APIs # ------------------------------------------------------------------------- + def _worker_from_api_response(self, w: dict) -> WorkerInfo: + """Convert API response dict to WorkerInfo. + + Args: + w: Worker dict from API response. + + Returns: + WorkerInfo object. + """ + status = "healthy" if w.get("is_healthy", False) else "unhealthy" + return WorkerInfo( + id=w.get("id", ""), + url=w.get("url", ""), + model=w.get("model_id"), + status=status, + pending_requests=w.get("load", 0), + metadata={ + "worker_type": w.get("worker_type"), + "connection_mode": w.get("connection_mode"), + "priority": w.get("priority"), + "cost": w.get("cost"), + }, + ) + def list_workers(self, timeout: float = 5.0) -> list[WorkerInfo]: """List all workers connected to the gateway. @@ -338,26 +362,9 @@ class Gateway: resp = httpx.get(f"{self.base_url}/workers", timeout=timeout) if resp.status_code == 200: data = resp.json() - workers = [] - for w in data.get("workers", []): - # Map API fields to WorkerInfo - status = "healthy" if w.get("is_healthy", False) else "unhealthy" - workers.append( - WorkerInfo( - id=w.get("id", ""), - url=w.get("url", ""), - model=w.get("model_id"), - status=status, - pending_requests=w.get("load", 0), - metadata={ - "worker_type": w.get("worker_type"), - "connection_mode": w.get("connection_mode"), - "priority": w.get("priority"), - "cost": w.get("cost"), - }, - ) - ) - return workers + return [ + self._worker_from_api_response(w) for w in data.get("workers", []) + ] return [] except (httpx.RequestError, httpx.TimeoutException): return [] @@ -374,21 +381,7 @@ class Gateway: try: resp = httpx.get(f"{self.base_url}/workers/{worker_id}", timeout=timeout) if resp.status_code == 200: - w = resp.json() - status = "healthy" if w.get("is_healthy", False) else "unhealthy" - return WorkerInfo( - id=w.get("id", ""), - url=w.get("url", ""), - model=w.get("model_id"), - status=status, - pending_requests=w.get("load", 0), - metadata={ - "worker_type": w.get("worker_type"), - "connection_mode": w.get("connection_mode"), - "priority": w.get("priority"), - "cost": w.get("cost"), - }, - ) + return self._worker_from_api_response(resp.json()) return None except (httpx.RequestError, httpx.TimeoutException): return None diff --git a/sgl-model-gateway/e2e_test/infra/gpu_allocator.py b/sgl-model-gateway/e2e_test/infra/gpu_allocator.py index c6a690dc4..8fb5f2343 100644 --- a/sgl-model-gateway/e2e_test/infra/gpu_allocator.py +++ b/sgl-model-gateway/e2e_test/infra/gpu_allocator.py @@ -227,7 +227,7 @@ class GPUAllocator: name = pynvml.nvmlDeviceGetName(handle) # Handle bytes vs string return type (varies by pynvml version) if isinstance(name, bytes): - name = name.decode("utf-8") + name = name.decode("utf-8", errors="replace") mem_info = pynvml.nvmlDeviceGetMemoryInfo(handle) # Convert bytes to MB memory_mb = mem_info.total // (1024 * 1024) diff --git a/sgl-model-gateway/e2e_test/infra/model_pool.py b/sgl-model-gateway/e2e_test/infra/model_pool.py index 30b1e1368..1ff1ed5d2 100644 --- a/sgl-model-gateway/e2e_test/infra/model_pool.py +++ b/sgl-model-gateway/e2e_test/infra/model_pool.py @@ -305,9 +305,6 @@ class ModelPool: if ib_device: logger.info("Detected InfiniBand device: %s", ib_device) - # Track bootstrap ports for PD groups (all PD workers of same model/mode share one) - pd_bootstrap_ports: dict[tuple[str, ConnectionMode], int] = {} - deferred: list[str] = [] # Process requirements in order - all workers treated uniformly @@ -340,13 +337,8 @@ class ModelPool: deferred.append(str(identity)) continue - # Get bootstrap port for PD workers (shared within model/mode group) - bootstrap_port = None - if identity.is_prefill or identity.is_decode: - pd_key = (identity.model_id, identity.mode) - if pd_key not in pd_bootstrap_ports: - pd_bootstrap_ports[pd_key] = get_open_port() - bootstrap_port = pd_bootstrap_ports[pd_key] + # Each prefill worker needs its own bootstrap port for PD communication + bootstrap_port = get_open_port() if identity.is_prefill else None # Launch the worker self._launch_model( @@ -354,7 +346,7 @@ class ModelPool: mode=identity.mode, gpu_slot=slots[0], worker_type=identity.worker_type, - bootstrap_port=bootstrap_port if identity.is_prefill else None, + bootstrap_port=bootstrap_port, ib_device=( ib_device if (identity.is_prefill or identity.is_decode) else None ), @@ -888,25 +880,17 @@ class ModelPool: has_pd = any(w.is_prefill or w.is_decode for w in valid_workers) ib_device = detect_ib_device() if has_pd else None - # Track bootstrap ports for PD groups (shared within model/mode) - pd_bootstrap_ports: dict[tuple[str, ConnectionMode], int] = {} - instances: list[ModelInstance] = [] for w in valid_workers: - # Get bootstrap port for PD workers - bootstrap_port = None - if w.is_prefill or w.is_decode: - pd_key = (w.model_id, w.mode) - if pd_key not in pd_bootstrap_ports: - pd_bootstrap_ports[pd_key] = get_open_port() - bootstrap_port = pd_bootstrap_ports[pd_key] + # Each prefill worker needs its own bootstrap port for PD communication + bootstrap_port = get_open_port() if w.is_prefill else None instance = self._launch_model( model_id=w.model_id, mode=w.mode, gpu_slot=slot_map.get(w.key), worker_type=w.worker_type, - bootstrap_port=bootstrap_port if w.is_prefill else None, + bootstrap_port=bootstrap_port, ib_device=ib_device if (w.is_prefill or w.is_decode) else None, instance_key=w.key, ) diff --git a/sgl-model-gateway/e2e_test/infra/run_eval.py b/sgl-model-gateway/e2e_test/infra/run_eval.py index 753e27df9..fc87852db 100644 --- a/sgl-model-gateway/e2e_test/infra/run_eval.py +++ b/sgl-model-gateway/e2e_test/infra/run_eval.py @@ -30,12 +30,10 @@ if TYPE_CHECKING: from .simple_eval_common import Eval from .simple_eval_common import ChatCompletionSampler, set_ulimit +from .simple_eval_mmlu import MMLU_DATASET_URL logger = logging.getLogger(__name__) -# MMLU dataset URL -MMLU_DATASET_URL = "https://openaipublic.blob.core.windows.net/simple-evals/mmlu.csv" - @dataclass class EvalConfig: diff --git a/sgl-model-gateway/e2e_test/infra/simple_eval_common.py b/sgl-model-gateway/e2e_test/infra/simple_eval_common.py index 09edc252d..92e72937d 100644 --- a/sgl-model-gateway/e2e_test/infra/simple_eval_common.py +++ b/sgl-model-gateway/e2e_test/infra/simple_eval_common.py @@ -20,6 +20,8 @@ import requests from openai import OpenAI from tqdm import tqdm +from .constants import MAX_RETRY_ATTEMPTS + logger = logging.getLogger(__name__) OPENAI_SYSTEM_MESSAGE_API = "You are a helpful assistant." @@ -119,7 +121,6 @@ class ChatCompletionSampler(SamplerBase): image: str, encoding: str = "base64", format: str = "png", - fovea: int = 768, ): new_image = { "type": "image_url", @@ -141,7 +142,7 @@ class ChatCompletionSampler(SamplerBase): self._pack_message("system", self.system_message) ] + message_list trial = 0 - while trial < 6: # Max 63 seconds backoff (1+2+4+8+16+32) + while trial < MAX_RETRY_ATTEMPTS: try: response = self.client.chat.completions.create( model=self.model, @@ -162,14 +163,15 @@ class ChatCompletionSampler(SamplerBase): log_fn( "Request failed (retry %d/%d, backoff %ds): %s", trial + 1, - 6, + MAX_RETRY_ATTEMPTS, exception_backoff, e, ) time.sleep(exception_backoff) trial += 1 logger.warning( - "All retry attempts exhausted after 6 retries, returning empty response" + "All retry attempts exhausted after %d retries, returning empty response", + MAX_RETRY_ATTEMPTS, ) return "" diff --git a/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py b/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py index 80ec65cee..1083e56ca 100644 --- a/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py +++ b/sgl-model-gateway/e2e_test/infra/simple_eval_mmlu.py @@ -25,6 +25,9 @@ from .simple_eval_common import ( if TYPE_CHECKING: from .simple_eval_common import SamplerBase +# MMLU dataset URL (hosted by OpenAI) +MMLU_DATASET_URL = "https://openaipublic.blob.core.windows.net/simple-evals/mmlu.csv" + SUBJECT_TO_CATEGORY = { "abstract_algebra": "stem", "anatomy": "other",