[Test] L5 Long-Term Stability Test and GPU Memory Monitoring Main L5 last (#1799)

This commit is contained in:
zhumingjue138 authored and GitHub committed 2026-03-23 19:04:05 +08:00
1 parent 65d76a4127
commit c14ab819e3
15 files changed
+1613 -101

No files matched your search

+1 -1
View File
@@ -184,7 +184,7 @@ steps:
commands:
- export VLLM_WORKER_MULTIPROC_METHOD=spawn
- export BENCHMARK_DIR=tests
- pytest -s -v tests/perf/scripts/run_benchmark.py
- pytest -s -v tests/dfx/perf/scripts/run_benchmark.py
- buildkite-agent artifact upload "tests/*.json"
- python tools/nightly/buildkite_testcase_statistics.py -o tests/buildkite_testcase_statistics.html
- buildkite-agent artifact upload "tests/*.html"
+77 -10
View File
@@ -86,7 +86,7 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
/tests/e2e/online_serving/test_{model_name}_expansion.py<br>
/tests/e2e/offline_inference/test_{model_name}_expansion.py<br>
<strong>Performance:</strong><br>
/tests/e2e/perf/nightly.json<br>
/tests/dfx/perf/tests/test.json<br>
<strong>Doc Test:</strong><br>
tests/example/online_serving/test_{model_name}.py<br>
tests/example/offline_inference/test_{model_name}.py
@@ -104,7 +104,7 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari
<td> Depends on reality </td>
<td>
<strong>Stability:</strong><br>
tests/e2e/stability/weekly.json<br>
/tests/dfx/stability/tests/test.json<br>
<strong>Reliability:</strong><br>
tests/e2e/reliability/test_{model_name}.py
</td>
@@ -529,13 +529,13 @@ L4 level testing is a comprehensive quality audit before a version release. It e
### 3.2 Testing Content and Scope
- ***Full Functionality Testing***: Executes all test cases defined in `test_{model_name}_expansion.py`, covering all implemented features, positive flows, boundary conditions, and exception handling.
- ***Performance Testing***: Uses the `/tests/e2e/perf/nightly.json` configuration file to drive performance testing tools for stress, load, and endurance tests, collecting metrics like throughput, response time, and resource utilization.
- ***Performance Testing***: Uses the `tests/dfx/perf/tests/test.json` configuration file to drive performance testing tools for stress, load, and endurance tests, collecting metrics like throughput, response time, and resource utilization.
- ***Documentation Testing***: Verifies whether the example code provided to users is runnable and its results match the description.
### 3.3 Test Directory and Execution Files
- ***Functional Testing***: Same directories as L3.
- ***Performance Test Configuration***: `/tests/e2e/perf/nightly.json`
- ***Performance Test Configuration***: `tests/dfx/perf/tests/test.json`
- ***Documentation Example Tests***:
- - `tests/example/online_serving/test_{model_name}.py`
- `tests/example/offline_inference/test_{model_name}.py`
@@ -552,7 +552,7 @@ L4 level testing is a comprehensive quality audit before a version release. It e
--8<-- "docs/contributing/ci/test_examples/doc_example_tests.inc.md"
When you want to add L4-level ***performance test*** cases, you can refer to the following format for case addition in tests/perf/tests/test.json:
When you want to add L4-level ***performance test*** cases, you can refer to the following format for case addition in tests/dfx/perf/tests/test.json:
```JSON
{
@@ -654,12 +654,12 @@ L5 level testing focuses on the performance of model services under ***long-runn
### 4.2 Testing Content and Scope
- ***Long-term Stability (Stability) Testing***: Uses the `tests/e2e/stability/weekly.json` configuration to run the service under moderate load for an extended period (e.g., over 12 hours), monitoring whether metrics like memory/VRAM usage, response time, and throughput degrade over time, and whether the service process remains stable.
- ***Long-term Stability (Stability) Testing***: Uses the `tests/dfx/stability/tests/test.json` configuration to run the service under moderate load for an extended period (e.g., over 12 hours), monitoring whether metrics like memory/VRAM usage, response time, and throughput degrade over time, and whether the service process remains stable.
- ***Reliability Testing***: Uses `tests/e2e/reliability/test_{model_name}.py` to actively simulate various fault and abnormal scenarios, such as: dependent service interruption, abnormal input data, network flicker, hardware resource preemption, etc., to verify the system's fault tolerance, self-healing, and graceful degradation capabilities.
### 4.3 Test Directory and Execution Files
- ***Stability Test Configuration***: `tests/e2e/stability/weekly.json`
- ***Stability Test Configuration***: `tests/dfx/stability/tests/test.json`
- ***Reliability Test Suite***: `tests/e2e/reliability/test_{model_name}.py`
### 4.4 Execution Method and Example
@@ -669,12 +669,79 @@ L5 level testing focuses on the performance of model services under ***long-runn
- ***Script Example***:
<details>
<summary> Test Examples</summary>
```python
# WIP
When you want to add L5-level stability test cases, you can refer to the following format for case addition in `tests/dfx/stability/tests/test.json`:
```json
{
"test_name": "test_qwen3_omni_stability",
"server_params": {
"model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
"stage_config_name": "qwen3_omni.yaml"
},
"benchmark_params": [
{
"dataset_name": "random",
"backend": "openai-chat-omni",
"endpoint": "/v1/chat/completions",
"duration_sec": 43200,
"request_rate": 0.5,
"num_prompts_per_batch": 20,
"random_input_len": 2500,
"random_output_len": 900,
"ignore_eos": true,
"percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
}
]
}
```
#### Parameter Explanation
***Overview***
| Field | Required | Description |
| ---------------- | -------- | --------------------------------------------------------------------------- |
| test_name | Yes | Unique identifier for the stability test case |
| server_params | Yes | Server-side configuration parameters (model, stage configuration, etc.) |
| benchmark_params | Yes | Stability benchmark running parameters (supports multiple configurations) |
#### server_params Configuration
##### Basic Parameters
| Parameter | Required | Example | Description |
| ----------------- | -------- | ---------------------------------- | ----------------------------------- |
| model | Yes | "Qwen/Qwen3-Omni-30B-A3B-Instruct" | Model name or path |
| stage_config_name | Yes | "qwen3_omni.yaml" | Stage configuration file name |
##### Dynamic Configuration (update/delete)
Supports incremental modifications based on the basic configuration:
| Operation | Description |
| --------- | ------------------------------------ |
| update | Update or add configuration items |
| delete | Delete specified configuration items |
***Example***:
You can refer to Test Examples in Chapter 3.4
#### benchmark_params Configuration
For stability testing, the key parameters are:
- **duration_sec**: Total duration (in seconds) during which benchmark traffic is sent. The stability benchmark will keep sending batches until this duration is reached.
- **request_rate** / **max_concurrency**: Exactly one of them must be specified. They control how the traffic is generated for each batch:
- `request_rate`: Number of requests per second. The benchmark will send `num_prompts_per_batch` requests at the given rate.
- `max_concurrency`: Maximum number of concurrent requests. When this is used, `request_rate` is set to `inf` internally.
- **num_prompts_per_batch**: Number of prompts sent in each batch. Multiple batches will be executed sequentially within `duration_sec`.
All other optional parameters follow the same rules as the in Chapter 3.4.
</details>
- - ***Stability***: (Execution would be driven by the configuration in `weekly.json`).
- - ***Stability***: `pytest -s -v tests/dfx/stability/scripts/test_{model_name}.py`
- ***Reliability***: `pytest -s -v tests/e2e/reliability/test_{model_name}.py`
## Summary
+97
View File
@@ -0,0 +1,97 @@
import json
from pathlib import Path
from typing import Any
from tests.conftest import modify_stage_config
def load_configs(config_path: str) -> list[dict[str, Any]]:
try:
abs_path = Path(config_path).resolve()
with open(abs_path, encoding="utf-8") as f:
configs = json.load(f)
return configs
except json.JSONDecodeError as e:
raise ValueError(f"JSON parsing error: {str(e)}")
except FileNotFoundError:
raise ValueError(f"Configuration file not found: {config_path}")
except Exception as e:
raise RuntimeError(f"Failed to load configuration file: {str(e)}")
def modify_stage(default_path, updates, deletes):
kwargs = {}
if updates is not None:
kwargs["updates"] = updates
if deletes is not None:
kwargs["deletes"] = deletes
if kwargs:
path = modify_stage_config(default_path, **kwargs)
else:
path = default_path
return path
def create_unique_server_params(
configs: list[dict[str, Any]],
stage_configs_dir: Path,
) -> list[tuple[str, str, str]]:
unique_params = []
seen = set()
for config in configs:
test_name = config["test_name"]
model = config["server_params"]["model"]
stage_config_name = config["server_params"].get("stage_config_name")
if stage_config_name:
stage_config_path = str(stage_configs_dir / stage_config_name)
delete = config["server_params"].get("delete", None)
update = config["server_params"].get("update", None)
stage_config_path = modify_stage(stage_config_path, update, delete)
else:
stage_config_path = None
server_param = (test_name, model, stage_config_path)
if server_param not in seen:
seen.add(server_param)
unique_params.append(server_param)
return unique_params
def create_test_parameter_mapping(configs: list[dict[str, Any]]) -> dict[str, dict]:
mapping = {}
for config in configs:
test_name = config["test_name"]
if test_name not in mapping:
mapping[test_name] = {
"test_name": test_name,
"benchmark_params": [],
}
mapping[test_name]["benchmark_params"].extend(config["benchmark_params"])
return mapping
def get_benchmark_params_for_server(test_name: str, server_to_benchmark_mapping: dict[str, dict]) -> list:
if test_name not in server_to_benchmark_mapping:
return []
return server_to_benchmark_mapping[test_name]["benchmark_params"]
def create_benchmark_indices(
benchmark_configs: list[dict[str, Any]],
server_to_benchmark_mapping: dict[str, dict],
) -> list[tuple[str, int]]:
indices = []
seen = set()
for config in benchmark_configs:
test_name = config["test_name"]
if test_name not in seen:
seen.add(test_name)
params_list = get_benchmark_params_for_server(test_name, server_to_benchmark_mapping)
for idx in range(len(params_list)):
indices.append((test_name, idx))
return indices
@@ -8,83 +8,25 @@ from typing import Any
import pytest
from tests.conftest import OmniServer, modify_stage_config
from tests.conftest import OmniServer
from tests.dfx.conftest import (
create_benchmark_indices,
create_test_parameter_mapping,
create_unique_server_params,
get_benchmark_params_for_server,
load_configs,
)
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
os.environ["VLLM_TEST_CLEAN_GPU_MEMORY"] = "0"
def load_configs(config_path: str) -> list[dict[str, Any]]:
try:
abs_path = Path(config_path).resolve()
with open(abs_path, encoding="utf-8") as f:
configs = json.load(f)
return configs
except json.JSONDecodeError as e:
raise ValueError(f"JSON parsing error: {str(e)}")
except FileNotFoundError:
raise ValueError(f"Configuration file not found: {config_path}")
except Exception as e:
raise RuntimeError(f"Failed to load configuration file: {str(e)}")
def modify_stage(default_path, updates, deletes):
kwargs = {}
if updates is not None:
kwargs["updates"] = updates
if deletes is not None:
kwargs["deletes"] = deletes
if kwargs:
path = modify_stage_config(default_path, **kwargs)
else:
path = default_path
return path
def create_unique_server_params(configs: list[dict[str, Any]]) -> list[tuple[str, str, str]]:
unique_params = []
seen = set()
for config in configs:
test_name = config["test_name"]
model = config["server_params"]["model"]
stage_config_name = config["server_params"].get("stage_config_name")
if stage_config_name:
stage_config_path = str(Path(__file__).parent.parent / "stage_configs" / stage_config_name)
delete = config["server_params"].get("delete", None)
update = config["server_params"].get("update", None)
stage_config_path = modify_stage(stage_config_path, update, delete)
else:
stage_config_path = None
server_param = (test_name, model, stage_config_path)
if server_param not in seen:
seen.add(server_param)
unique_params.append(server_param)
return unique_params
def create_test_parameter_mapping(configs: list[dict[str, Any]]) -> dict[str, dict]:
mapping = {}
for config in configs:
test_name = config["test_name"]
if test_name not in mapping:
mapping[test_name] = {
"test_name": test_name,
"benchmark_params": [],
}
mapping[test_name]["benchmark_params"].extend(config["benchmark_params"])
return mapping
CONFIG_FILE_PATH = str(Path(__file__).parent.parent / "tests" / "test.json")
BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
test_params = create_unique_server_params(BENCHMARK_CONFIGS)
STAGE_CONFIGS_DIR = Path(__file__).parent.parent / "stage_configs"
test_params = create_unique_server_params(BENCHMARK_CONFIGS, STAGE_CONFIGS_DIR)
server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS)
_omni_server_lock = threading.Lock()
@@ -157,27 +99,7 @@ def run_benchmark(
return result
def get_benchmark_params_for_server(test_name: str) -> list:
if test_name not in server_to_benchmark_mapping:
return []
return server_to_benchmark_mapping[test_name]["benchmark_params"]
def create_benchmark_indices():
indices = []
seen = set()
for config in BENCHMARK_CONFIGS:
test_name = config["test_name"]
if test_name not in seen:
seen.add(test_name)
params_list = get_benchmark_params_for_server(test_name)
for idx in range(len(params_list)):
indices.append((test_name, idx))
return indices
benchmark_indices = create_benchmark_indices()
benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
@pytest.fixture(params=benchmark_indices)
@@ -188,7 +110,7 @@ def benchmark_params(request, omni_server):
if test_name != omni_server.test_name:
pytest.skip(f"Skipping parameter for {test_name} - current server is {omni_server.test_name}")
all_params = get_benchmark_params_for_server(test_name)
all_params = get_benchmark_params_for_server(test_name, server_to_benchmark_mapping)
if not all_params:
raise ValueError(f"No benchmark parameters found for test: {test_name}")
File renamed without changes.
+90
View File
@@ -0,0 +1,90 @@
# Resource Monitor Script Guide (GPU / Reserved CPU·NPU)
The **`scripts/resource_monitor.sh`** script in this directory is the unified entry point for resource monitoring: it collects GPU memory and related metrics while running any command (such as long-running stability tests or unit tests), then bundles the results and generates a single-file HTML report when the command finishes. The script generates **`report.html` and CSV** files; open `report.html` in the output directory after the run completes.
Currently only the **GPU** backend is implemented. The backend is selected with the subcommand option `--backend gpu|cpu|npu` (`gpu` by default), with CPU/NPU reserved for future extension.
---
## Subcommands
All functionality is provided by a single script and should be run from the repository root:
| Subcommand | Description |
|--------|------|
| `scripts/resource_monitor.sh start [--backend gpu\|cpu\|npu] [gpu_ids] [interval]` | Collect data in the background (currently only `gpu`: `nvidia-smi` writes CSV) |
| `scripts/resource_monitor.sh finalize [--backend gpu\|cpu\|npu] [run_id]` | Bundle the current run, generate `report.html`, and print `GPU_MONITOR_BUNDLE_DIR=` / `RESOURCE_MONITOR_BUNDLE_DIR=` |
| `scripts/resource_monitor.sh run [--backend gpu\|cpu\|npu] -- <command>` | Complete everything in one step: `start` -> run your command -> `finalize` |
`--backend` is optional. If omitted, the **gpu** backend is used by default.
---
## Environment Variables (monitoring script only)
| Environment Variable | Description | Default |
|----------|------|--------|
| `RESOURCE_MONITOR_DATA_ROOT` | Root directory for monitoring data | `tests/dfx/stability/gpu_monitor_data` |
| `RESOURCE_MONITOR_INTERVAL` | Sampling interval (seconds) | 5 |
| `RESOURCE_MONITOR_LOG_INTERVAL` | Log print interval (seconds) | 15 |
| `GPU_MONITOR_DEVICES` | [GPU backend] GPU device IDs to monitor, such as `0,1` or `all` | all |
---
## Recommended Usage: one-step `run`
Run this in **finally** or **after script** blocks so cleanup still happens even if the tested command fails:
```bash
# Monitor + run any command, then automatically bundle results and generate report.html
# If --backend is omitted, gpu is used by default; use --backend cpu or --backend npu for other backends
bash tests/dfx/stability/scripts/resource_monitor.sh run [--backend gpu|cpu|npu] -- <your-command>
```
Examples (from the repository root):
```bash
# Example: run a pytest case (gpu backend by default, so --backend can be omitted)
bash tests/dfx/stability/scripts/resource_monitor.sh run -- pytest -s -v tests/e2e/online_serving/test_foo.py -k test_xxx
# Explicitly select the gpu backend, customize the sampling interval and GPUs 0,1, and print one log line every 30s
export RESOURCE_MONITOR_INTERVAL=10
export GPU_MONITOR_DEVICES=0,1
export RESOURCE_MONITOR_LOG_INTERVAL=30
bash tests/dfx/stability/scripts/resource_monitor.sh run --backend gpu -- pytest -s -v tests/e2e/online_serving/test_foo.py
```
During execution, the log will show `[GPU] ...` every few seconds. When the run ends, the log prints the bundle path, for example: `Line chart: open in browser: .../report.html`.
---
## Step-by-step Usage: `start` -> your command -> `finalize`
If you need to start monitoring first and then manually run a long task, you can call the steps separately:
```bash
# 1. Start monitoring (run from the `scripts` directory or set `DATA_ROOT`; gpu is the default if `--backend` is omitted)
cd tests/dfx/stability/scripts
./resource_monitor.sh start [--backend gpu] all 5 &
MONITOR_PID=$!
# 2. Run your long-running/stability test command (any command)
# ...
# 3. Finalize (must run before environment cleanup; putting it in `finally` / `after script` is recommended; the backend must match `start`)
BUNDLE_LINE=$(./resource_monitor.sh finalize [--backend gpu] 2>/dev/null | grep '^GPU_MONITOR_BUNDLE_DIR=')
eval "$BUNDLE_LINE"
echo "Report directory: $GPU_MONITOR_BUNDLE_DIR"
```
---
## Directories and Outputs (monitoring only)
- **Scripts**: `tests/dfx/stability/scripts/resource_monitor.sh` (entry point) and `scripts/generate_report.py` (called by `finalize` to generate HTML).
- **Data directory**: `tests/dfx/stability/gpu_monitor_data/` by default (can be overridden with `RESOURCE_MONITOR_DATA_ROOT`).
- Each run generates `run_<run_id>/gpu_metrics.csv`.
- After `finalize`, you get `gpu_monitor_bundle_<run_id>/` containing `gpu_metrics.csv`, `report.html`, and `README.txt`.
- **View the report**: open `report.html` in the bundle directory with a browser to inspect memory usage curves and statistics.
The script only generates `report.html` and CSV files. If you need to keep the report, archive or download it from the working directory yourself.
+169
View File
@@ -0,0 +1,169 @@
"""
Stability-specific conftest: when pytest is executed under this directory,
resource monitoring is started before each test and finalized after each test,
so each stability test case gets its own HTML report (one report per case).
No need to wrap pytest with `bash resource_monitor.sh run -- pytest ...`.
"""
import os
import subprocess
import sys
import threading
import time
from pathlib import Path
import pytest
STABILITY_DIR = Path(__file__).resolve().parent
RESOURCE_MONITOR_SCRIPT = STABILITY_DIR / "scripts" / "resource_monitor.sh"
REPO_ROOT = STABILITY_DIR.parent.parent.parent
def _start_resource_monitor():
"""Start `resource_monitor.sh start` in the background and return `Popen` or `None`."""
if not RESOURCE_MONITOR_SCRIPT.is_file():
return None
try:
proc = subprocess.Popen(
["bash", str(RESOURCE_MONITOR_SCRIPT), "start", "--backend", "gpu"],
cwd=str(REPO_ROOT),
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
start_new_session=True,
)
try:
proc.wait(timeout=2)
if proc.returncode != 0:
stderr = proc.stderr.read().decode("utf-8", errors="ignore") if proc.stderr else ""
if stderr.strip():
sys.stderr.write(f"[Stability] Resource monitor failed to start: {stderr.strip()}\n")
return None
except subprocess.TimeoutExpired:
pass
return proc
except (FileNotFoundError, OSError):
return None
def _get_monitor_data_root() -> Path:
data_root = os.environ.get("RESOURCE_MONITOR_DATA_ROOT") or os.environ.get("GPU_MONITOR_DATA_ROOT")
if data_root:
return Path(data_root)
return STABILITY_DIR / "gpu_monitor_data"
def _wait_for_run_dir(timeout_sec: int = 10) -> Path | None:
data_root = _get_monitor_data_root()
run_id_file = data_root / "current_run_id"
deadline = time.time() + timeout_sec
while time.time() < deadline:
if run_id_file.is_file():
run_id = run_id_file.read_text(encoding="utf-8").strip()
if run_id:
run_dir = data_root / run_id
if run_dir.is_dir():
return run_dir
time.sleep(0.5)
return None
def _report_latest_gpu_samples(stop_event: threading.Event) -> None:
"""Periodically print the latest sampled GPU line."""
log_interval = int(
os.environ.get("RESOURCE_MONITOR_LOG_INTERVAL") or os.environ.get("GPU_MONITOR_LOG_INTERVAL") or "15"
)
log_interval = max(log_interval, 1)
last_line = ""
time.sleep(min(log_interval, 5))
while not stop_event.wait(log_interval):
run_dir = _wait_for_run_dir(timeout_sec=1)
if run_dir is None:
continue
csv_file = run_dir / "gpu_metrics.csv"
if not csv_file.is_file():
continue
try:
lines = csv_file.read_text(encoding="utf-8").splitlines()
except OSError:
continue
if len(lines) <= 1:
continue
latest = lines[-1].strip()
if latest and latest != last_line:
last_line = latest
sys.stderr.write(f"[GPU] {latest}\n")
def _finalize_resource_monitor() -> str | None:
"""
Run `resource_monitor.sh finalize` for the current run and generate the report.
Returns the bundle dir path (for this test case's report) if successful, else None.
"""
if not RESOURCE_MONITOR_SCRIPT.is_file():
return None
try:
result = subprocess.run(
["bash", str(RESOURCE_MONITOR_SCRIPT), "finalize", "--backend", "gpu"],
cwd=str(REPO_ROOT),
capture_output=True,
text=True,
timeout=60,
check=False,
)
if result.returncode != 0:
return None
for line in (result.stdout or "").splitlines():
if line.startswith("GPU_MONITOR_BUNDLE_DIR=") or line.startswith("RESOURCE_MONITOR_BUNDLE_DIR="):
_, _, value = line.partition("=")
return value.strip() if value else None
return None
except (FileNotFoundError, OSError, subprocess.TimeoutExpired):
return None
@pytest.fixture(autouse=True)
def stability_resource_monitor_per_test(request: pytest.FixtureRequest):
"""
For each test under this directory: start GPU monitor before the test,
then finalize after the test so this case gets its own report.html.
"""
proc = _start_resource_monitor()
stop_event = threading.Event()
reporter: threading.Thread | None = None
if proc is not None:
reporter = threading.Thread(
target=_report_latest_gpu_samples,
args=(stop_event,),
name="stability-resource-monitor-reporter",
daemon=True,
)
reporter.start()
run_dir = _wait_for_run_dir(timeout_sec=5)
node_name = request.node.name
if run_dir is not None:
sys.stderr.write(f"[Stability] Resource monitor started for test: {node_name} | run dir: {run_dir}\n")
else:
sys.stderr.write(f"[Stability] Resource monitor started for test: {node_name} (run dir not ready yet)\n")
yield
# Teardown: stop reporter, stop monitor, finalize → one HTML per test
if proc is not None:
stop_event.set()
if reporter is not None and reporter.is_alive():
reporter.join(timeout=2)
if proc.poll() is None:
proc.terminate()
try:
proc.wait(timeout=10)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
bundle_dir = _finalize_resource_monitor()
node_name = request.node.name
if bundle_dir:
sys.stderr.write(f"[Stability] Report for test «{node_name}»: {bundle_dir}/report.html\n")
else:
sys.stderr.write(f"[Stability] Finalize skipped or failed for test «{node_name}»\n")
+1
View File
@@ -0,0 +1 @@
# scripts package for stability benchmark
@@ -0,0 +1,364 @@
#!/usr/bin/env python3
"""
Generate a GPU memory monitoring report (HTML with charts and simple anomaly markers)
from the CSV produced by `resource_monitor.sh`.
This is used to generate an archivable report in CI after a long-running stability
test so that the report remains available even after environment cleanup.
"""
from __future__ import annotations
import csv
import os
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from vllm.logger import init_logger
logger = init_logger(__name__)
def load_csv(csv_path: str) -> list[dict]:
"""Load and parse rows from a GPU monitoring CSV file.
Args:
csv_path: Path to the CSV file.
Returns:
A list of parsed row dicts. Each row contains keys such as
`timestamp_epoch`, `gpu_index`, and `memory_*`.
Invalid rows are skipped and logged.
"""
rows = []
with open(csv_path, newline="", encoding="utf-8") as f:
for r in csv.DictReader(f):
try:
r["timestamp_epoch"] = int(float(r["timestamp_epoch"]))
r["gpu_index"] = int(r["gpu_index"])
r["memory_used_mb"] = int(r["memory_used_mb"])
r["memory_total_mb"] = int(r["memory_total_mb"])
r["memory_util_pct"] = int(r["memory_util_pct"])
rows.append(r)
except (KeyError, ValueError) as e:
logger.debug("Skip invalid CSV row: %s", e)
continue
return rows
def compute_stats(rows: list[dict]) -> dict:
"""Compute per-GPU min/max/avg/P50/P95 memory utilization and sample counts.
Args:
rows: Rows returned by `load_csv`.
Returns:
A stats dict keyed by `gpu_index`, with values containing
`min`/`max`/`avg`/`p50`/`p95`/`samples`.
"""
by_gpu = defaultdict(list)
for r in rows:
by_gpu[r["gpu_index"]].append(r["memory_util_pct"])
stats = {}
for gpu, pcts in by_gpu.items():
if not pcts:
continue
pcts_sorted = sorted(pcts)
n = len(pcts_sorted)
stats[gpu] = {
"min": min(pcts),
"max": max(pcts),
"avg": round(sum(pcts) / n, 1),
"p50": pcts_sorted[n // 2] if n else 0,
"p95": pcts_sorted[int(n * 0.95)] if n > 1 else pcts_sorted[0],
"samples": n,
}
return stats
def find_anomalies(rows: list[dict], high_pct: int = 95, low_pct: int = 5) -> list[dict]:
"""Find simple anomalies where memory utilization exceeds `high_pct` or drops below `low_pct`."""
anomalies = []
for r in rows:
pct = r["memory_util_pct"]
ts_iso = r.get("timestamp_iso") or datetime.fromtimestamp(r["timestamp_epoch"]).strftime("%Y-%m-%d %H:%M:%S")
extra = {"timestamp_iso": ts_iso}
if pct >= high_pct:
anomalies.append({**r, **extra, "type": "high", "threshold": high_pct})
elif pct <= low_pct:
anomalies.append({**r, **extra, "type": "low", "threshold": low_pct})
return anomalies
def _anomaly_sort_key(a: dict) -> tuple[int, str, int]:
return (a["gpu_index"], a["type"], a["timestamp_epoch"])
def merge_anomalies_into_periods(anomalies: list[dict], max_gap_seconds: int = 120) -> list[dict]:
"""Merge consecutive anomalies (same GPU, same type) into time periods for display.
This avoids truncation: instead of showing only the first N raw points, we show
each continuous period once, so all below-threshold (or above-threshold) periods
are visible in the report.
"""
if not anomalies:
return []
# Sort by GPU, type, then time so we can merge consecutive same-GPU same-type.
sorted_anomalies = sorted(anomalies, key=_anomaly_sort_key)
periods = []
for a in sorted_anomalies:
ts = a["timestamp_epoch"]
ts_iso = a.get("timestamp_iso") or datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
if not periods:
periods.append(
{
"gpu_index": a["gpu_index"],
"type": a["type"],
"threshold": a["threshold"],
"start_epoch": ts,
"end_epoch": ts,
"start_iso": ts_iso,
"end_iso": ts_iso,
"min_pct": a["memory_util_pct"],
"max_pct": a["memory_util_pct"],
"samples": 1,
}
)
continue
last = periods[-1]
if (
last["gpu_index"] == a["gpu_index"]
and last["type"] == a["type"]
and (ts - last["end_epoch"]) <= max_gap_seconds
):
last["end_epoch"] = ts
last["end_iso"] = ts_iso
last["min_pct"] = min(last["min_pct"], a["memory_util_pct"])
last["max_pct"] = max(last["max_pct"], a["memory_util_pct"])
last["samples"] += 1
else:
periods.append(
{
"gpu_index": a["gpu_index"],
"type": a["type"],
"threshold": a["threshold"],
"start_epoch": ts,
"end_epoch": ts,
"start_iso": ts_iso,
"end_iso": ts_iso,
"min_pct": a["memory_util_pct"],
"max_pct": a["memory_util_pct"],
"samples": 1,
}
)
return periods
def build_series_by_gpu(rows: list[dict]) -> tuple[list[float], dict[int, list[float]]]:
"""Deduplicate timestamps in time order and build a memory usage series (GB) for each GPU."""
times = []
by_ts_gpu = defaultdict(dict)
for r in rows:
t = r["timestamp_epoch"]
g = r["gpu_index"]
# Convert MB to GB for chart display.
by_ts_gpu[t][g] = r["memory_used_mb"] / 1024.0
for t in sorted(by_ts_gpu.keys()):
times.append(t)
gpu_series = defaultdict(list)
for t in times:
gpus = by_ts_gpu[t]
for g in sorted(gpus.keys()):
gpu_series[g].append(gpus[g])
return times, dict(gpu_series)
def render_html(
run_id: str,
csv_path: str,
rows: list[dict],
stats: dict,
anomalies: list[dict],
out_path: str,
) -> None:
"""Generate a single-file HTML report and write it to `out_path`.
Args:
run_id: Run identifier used in the title.
csv_path: Source CSV path, used only for displaying the file name.
rows: Raw data rows.
stats: Statistics returned by `compute_stats`.
anomalies: Anomaly list returned by `find_anomalies`.
out_path: Output HTML path; parent directories are created if needed.
"""
times, gpu_series = build_series_by_gpu(rows)
# X-axis time, e.g. 02-27 11:38:07 (local timezone), which is easier to read for long stability runs.
labels_js = [f'"{datetime.fromtimestamp(t).strftime("%m-%d %H:%M:%S")}"' for t in times]
datasets_js = []
colors = ["#e94560", "#0f3460", "#533483", "#16c79a"]
for i, (gpu, series) in enumerate(sorted(gpu_series.items())):
color = colors[i % len(colors)]
data_str = ",".join(str(v) for v in series)
datasets_js.append(
f'{{ label: "GPU {gpu}", data: [{data_str}], borderColor: "{color}", '
f'backgroundColor: "{color}20", fill: true, tension: 0.2 }}'
)
stats_rows = []
for gpu in sorted(stats.keys()):
s = stats[gpu]
stats_rows.append(
f"<tr><td>GPU {gpu}</td><td>{s['min']}%</td><td>{s['max']}%</td>"
f"<td>{s['avg']}%</td><td>{s['p50']}</td><td>{s['p95']}</td><td>{s['samples']}</td></tr>"
)
stats_table = "\n".join(stats_rows)
# Use merged periods so every below/above-threshold time period is shown (no truncation).
periods = merge_anomalies_into_periods(anomalies)
anomaly_cells = []
for p in periods:
time_range = f"{p['start_iso']} — {p['end_iso']}" if p["start_iso"] != p["end_iso"] else p["start_iso"]
pct_str = f"{p['min_pct']}–{p['max_pct']}%" if p["min_pct"] != p["max_pct"] else f"{p['min_pct']}%"
anomaly_cells.append(
f"<tr><td>{time_range}</td><td>GPU {p['gpu_index']}</td>"
f"<td>{pct_str}</td><td>{p['type']} (threshold {p['threshold']}%)</td><td>{p['samples']}</td></tr>"
)
anomaly_table = "\n".join(anomaly_cells) if anomaly_cells else "<tr><td colspan='5'>None</td></tr>"
html = f"""<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>GPU memory monitor report - {run_id}</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4.4.1/dist/chart.umd.min.js"></script>
<style>
body {{ font-family: system-ui,sans-serif; margin: 1rem; background: #1a1a2e; color: #eee; }}
h1 {{ font-size: 1.25rem; }} .meta {{ color: #888; margin-bottom: 1rem; }}
table {{ border-collapse: collapse; margin: 1rem 0; }} th, td {{ border: 1px solid #333; padding: 0.4rem 0.6rem; text-align: left; }}
th {{ background: #16213e; }} canvas {{ max-height: 400px; }}
</style>
</head>
<body>
<h1>Stability GPU memory monitor report</h1>
<p class="meta">Run: {run_id} | Data file: {os.path.basename(csv_path)} | Samples: {len(rows)}</p>
<h2>Statistics</h2>
<table>
<tr><th>GPU</th><th>Min %</th><th>Max %</th><th>Avg %</th><th>P50</th><th>P95</th><th>Samples</th></tr>
{stats_table}
</table>
<h2>Memory utilization over time</h2>
<button id="toggleMode">Split per GPU</button>
<div id="chartsContainer">
<canvas id="chart"></canvas>
</div>
<h2>Anomalies (high/low threshold)</h2>
<p class="meta">Consecutive samples merged into periods; all periods listed.</p>
<table>
<tr><th>Period</th><th>GPU</th><th>Util %</th><th>Type</th><th>Samples</th></tr>
{anomaly_table}
</table>
<script>
const labels = [{",".join(labels_js)}];
const gpuDatasets = [{",".join(datasets_js)}];
let mode = "combined";
let charts = [];
const commonOptions = {{
responsive: true,
scales: {{
y: {{
title: {{
display: true,
text: "Memory used (GB)"
}}
}}
}}
}};
function destroyCharts() {{
charts.forEach(c => c.destroy());
charts = [];
}}
function renderCombined() {{
const container = document.getElementById("chartsContainer");
container.innerHTML = '<canvas id="chart"></canvas>';
const ctx = document.getElementById("chart");
charts = [new Chart(ctx, {{
type: "line",
data: {{ labels, datasets: gpuDatasets }},
options: commonOptions
}})];
}}
function renderSplit() {{
const container = document.getElementById("chartsContainer");
container.innerHTML = "";
charts = [];
gpuDatasets.forEach((ds, idx) => {{
const canvas = document.createElement("canvas");
canvas.id = "chart-gpu-" + idx;
container.appendChild(canvas);
charts.push(new Chart(canvas, {{
type: "line",
data: {{ labels, datasets: [ds] }},
options: commonOptions
}}));
}});
}}
document.getElementById("toggleMode").addEventListener("click", () => {{
destroyCharts();
if (mode === "combined") {{
mode = "split";
document.getElementById("toggleMode").textContent = "Combine all GPUs";
renderSplit();
}} else {{
mode = "combined";
document.getElementById("toggleMode").textContent = "Split per GPU";
renderCombined();
}}
}});
// Default: combined chart
renderCombined();
</script>
</body>
</html>
"""
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
with open(out_path, "w", encoding="utf-8") as f:
f.write(html)
def main() -> int:
"""Entry point: read the CSV path and optional output path from the command line and generate `report.html`.
Returns:
0 on success, 1 on invalid arguments or invalid data.
"""
if len(sys.argv) < 2:
logger.error("Usage: generate_report.py <gpu_metrics.csv> [output.html]")
return 1
csv_path = sys.argv[1]
out_path = sys.argv[2] if len(sys.argv) > 2 else csv_path.replace(".csv", "_report.html")
if not os.path.isfile(csv_path):
logger.error("File not found: %s", csv_path)
return 1
run_id = Path(csv_path).parent.name
rows = load_csv(csv_path)
if not rows:
logger.error("CSV has no valid data")
return 1
stats = compute_stats(rows)
anomalies = find_anomalies(rows)
render_html(run_id, csv_path, rows, stats, anomalies, out_path)
print(out_path)
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,329 @@
#!/bin/bash
#
# Stability resource monitor script (single entry point, extendable to CPU/NPU later)
# Only GPU monitoring is implemented for now; --backend reserves cpu/npu for future expansion.
#
# Subcommands: start | finalize | run -- <command>
#
# start - collect data in the background (currently only gpu: `nvidia-smi` writes CSV)
# finalize - bundle the current run (CSV + report.html) and print BUNDLE_DIR
# run - start monitoring -> execute command -> finalize (generate report.html only, no upload)
#
# Argument entry point (reserved for multiple backends):
# --backend, -b gpu | cpu | npu default: gpu; only gpu is implemented right now
#
# Environment variables:
# RESOURCE_MONITOR_DATA_ROOT data root directory (compatible with GPU_MONITOR_DATA_ROOT)
# RESOURCE_MONITOR_INTERVAL sampling interval in seconds (compatible with GPU_MONITOR_INTERVAL)
# RESOURCE_MONITOR_LOG_INTERVAL log print interval in seconds (compatible with GPU_MONITOR_LOG_INTERVAL)
# GPU_MONITOR_DEVICES [GPU backend] GPU device IDs to monitor, e.g. 0,1 or all
#
set -e
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# Data root directory: placed under the stability directory (the parent of SCRIPT_DIR)
# so it stays alongside stage_configs/tests.
STABILITY_DIR="$(cd "$SCRIPT_DIR/.." && pwd)"
DATA_ROOT="${RESOURCE_MONITOR_DATA_ROOT:-${GPU_MONITOR_DATA_ROOT:-$STABILITY_DIR/gpu_monitor_data}}"
SUBCMD="${1:-}"
# Parse optional --backend|-b arguments, store the result in BACKEND,
# and keep the remaining positional arguments in the REST_ARGS array.
parse_backend_and_rest() {
BACKEND="gpu"
REST_ARGS=()
while [[ $# -gt 0 ]]; do
case "$1" in
--backend|-b)
if [[ -z "${2:-}" ]]; then
echo "Error: --backend requires a value (gpu|cpu|npu)" >&2
exit 1
fi
BACKEND="$2"
shift 2
;;
*)
REST_ARGS+=("$1")
shift
;;
esac
done
}
# ---------- subcommand: start ----------
cmd_start() {
parse_backend_and_rest "$@"
local POS1="${REST_ARGS[0]:-${GPU_MONITOR_DEVICES:-all}}"
local POS2="${REST_ARGS[1]:-${RESOURCE_MONITOR_INTERVAL:-${GPU_MONITOR_INTERVAL:-5}}}"
case "$BACKEND" in
gpu)
start_backend_gpu "$POS1" "$POS2"
;;
cpu|npu)
echo "Error: backend '$BACKEND' not implemented yet. Currently only 'gpu' is supported." >&2
exit 1
;;
*)
echo "Error: invalid backend '$BACKEND'. Use gpu|cpu|npu." >&2
exit 1
;;
esac
}
start_backend_gpu() {
local GPU_IDS_RAW="${1:-${GPU_MONITOR_DEVICES:-all}}"
local INTERVAL="${2:-${RESOURCE_MONITOR_INTERVAL:-${GPU_MONITOR_INTERVAL:-5}}}"
[[ "$INTERVAL" =~ ^[0-9]+$ ]] && [[ "$INTERVAL" -ge 1 ]] || {
echo "Error: interval must be a positive integer (seconds)"
echo "Usage: $0 start [--backend gpu|cpu|npu] [gpu_ids] [interval_seconds]"
exit 1
}
local RUN_ID="run_$(date +%Y%m%d_%H%M%S)"
local RUN_DIR="$DATA_ROOT/$RUN_ID"
mkdir -p "$RUN_DIR"
echo "$RUN_ID" > "$DATA_ROOT/current_run_id"
local CSV_FILE="$RUN_DIR/gpu_metrics.csv"
echo "timestamp_iso,timestamp_epoch,gpu_index,memory_used_mb,memory_total_mb,memory_util_pct" > "$CSV_FILE"
local NVSMI_QUERY="index,memory.used,memory.total"
local NVSMI_IDS=""
[[ "$GPU_IDS_RAW" != "all" ]] && NVSMI_IDS="-i $GPU_IDS_RAW"
trap 'echo "[$(date +%H:%M:%S)] Stopping; data saved to $RUN_DIR"; exit 0' SIGTERM SIGINT
echo "========================================"
echo "Stability resource monitor (backend=gpu) started"
echo "RUN_ID: $RUN_ID"
echo "Data dir: $RUN_DIR"
echo "Interval: ${INTERVAL}s | GPU: $GPU_IDS_RAW"
echo "========================================"
while true; do
local TS_ISO TS_EPOCH RAW
TS_ISO=$(date -Iseconds 2>/dev/null || date '+%Y-%m-%dT%H:%M:%S%z')
TS_EPOCH=$(date +%s)
RAW=$(nvidia-smi --query-gpu="$NVSMI_QUERY" --format=csv,noheader,nounits $NVSMI_IDS 2>/dev/null) || true
if [[ -z "$RAW" ]]; then
sleep "$INTERVAL"
continue
fi
while IFS= read -r line; do
line=$(echo "$line" | tr -d ' ')
[[ -z "$line" ]] && continue
local idx used total pct
idx=$(echo "$line" | cut -d',' -f1)
used=$(echo "$line" | cut -d',' -f2)
total=$(echo "$line" | cut -d',' -f3)
used=${used:-0}
total=${total:-1}
[[ "$total" -le 0 ]] && total=1
pct=$((used * 100 / total))
echo "${TS_ISO},${TS_EPOCH},${idx},${used},${total},${pct}" >> "$CSV_FILE"
done <<< "$RAW"
sleep "$INTERVAL"
done
}
# ---------- subcommand: finalize ----------
cmd_finalize() {
parse_backend_and_rest "$@"
local RUN_ID="${REST_ARGS[0]:-}"
case "$BACKEND" in
gpu)
finalize_backend_gpu "$RUN_ID"
;;
cpu|npu)
echo "Error: backend '$BACKEND' not implemented yet. Currently only 'gpu' is supported." >&2
exit 1
;;
*)
echo "Error: invalid backend '$BACKEND'. Use gpu|cpu|npu." >&2
exit 1
;;
esac
}
finalize_backend_gpu() {
local RUN_ID="${1:-}"
if [[ -z "$RUN_ID" ]]; then
if [[ -f "$DATA_ROOT/current_run_id" ]]; then
RUN_ID=$(cat "$DATA_ROOT/current_run_id")
else
echo "Error: run_id not specified and $DATA_ROOT/current_run_id does not exist" >&2
exit 1
fi
fi
local RUN_DIR="$DATA_ROOT/$RUN_ID"
if [[ ! -d "$RUN_DIR" ]]; then
echo "Error: run dir does not exist: $RUN_DIR" >&2
exit 1
fi
local CSV_FILE="$RUN_DIR/gpu_metrics.csv"
if [[ ! -f "$CSV_FILE" ]]; then
echo "Error: CSV not found: $CSV_FILE" >&2
exit 1
fi
local BUNDLE_DIR="$DATA_ROOT/gpu_monitor_bundle_${RUN_ID}"
rm -rf "$BUNDLE_DIR"
mkdir -p "$BUNDLE_DIR"
cp "$CSV_FILE" "$BUNDLE_DIR/gpu_metrics.csv"
local REPORT_HTML="$BUNDLE_DIR/report.html"
if command -v python3 &>/dev/null; then
if python3 "$SCRIPT_DIR/generate_report.py" "$CSV_FILE" "$REPORT_HTML"; then
echo "Report generated: $REPORT_HTML"
else
echo "Warning: report generation failed; only CSV archived" >&2
fi
else
echo "Warning: python3 not found; skipping report" >&2
fi
cat > "$BUNDLE_DIR/README.txt" << EOF
Stability resource monitor (gpu) bundle - ${RUN_ID}
- gpu_metrics.csv: raw samples
- report.html: report with charts (open in browser to view)
EOF
local BUNDLE_ABS
BUNDLE_ABS=$(cd "$BUNDLE_DIR" && pwd)
echo "GPU_MONITOR_BUNDLE_DIR=$BUNDLE_ABS"
echo "RESOURCE_MONITOR_BUNDLE_DIR=$BUNDLE_ABS"
echo "Archive path: $BUNDLE_ABS"
}
# ---------- subcommand: run ----------
cmd_run() {
shift
local BACKEND="gpu"
local CMD=()
while [[ $# -gt 0 ]]; do
case "$1" in
--backend|-b)
if [[ -n "${2:-}" ]]; then
BACKEND="$2"
shift 2
else
shift
fi
;;
--)
shift
CMD=("$@")
break
;;
*)
shift
;;
esac
done
if [[ ${#CMD[@]} -eq 0 ]]; then
echo "Usage: $0 run [--backend gpu|cpu|npu] -- <command to run>" >&2
exit 1
fi
case "$BACKEND" in
gpu)
run_backend_gpu "${CMD[@]}"
;;
cpu|npu)
echo "Error: backend '$BACKEND' not implemented yet. Currently only 'gpu' is supported." >&2
exit 1
;;
*)
echo "Error: invalid backend '$BACKEND'. Use gpu|cpu|npu." >&2
exit 1
;;
esac
}
run_backend_gpu() {
local CMD=("$@")
export RESOURCE_MONITOR_DATA_ROOT="${RESOURCE_MONITOR_DATA_ROOT:-${GPU_MONITOR_DATA_ROOT:-$STABILITY_DIR/gpu_monitor_data}}"
export GPU_MONITOR_DATA_ROOT="${GPU_MONITOR_DATA_ROOT:-$RESOURCE_MONITOR_DATA_ROOT}"
export RESOURCE_MONITOR_INTERVAL="${RESOURCE_MONITOR_INTERVAL:-${GPU_MONITOR_INTERVAL:-5}}"
export GPU_MONITOR_INTERVAL="${GPU_MONITOR_INTERVAL:-${GPU_MONITOR_INTERVAL:-5}}"
export GPU_MONITOR_DEVICES="${GPU_MONITOR_DEVICES:-all}"
export RESOURCE_MONITOR_LOG_INTERVAL="${RESOURCE_MONITOR_LOG_INTERVAL:-${GPU_MONITOR_LOG_INTERVAL:-15}}"
export GPU_MONITOR_LOG_INTERVAL="${RESOURCE_MONITOR_LOG_INTERVAL:-${GPU_MONITOR_LOG_INTERVAL:-15}}"
local REPO_ROOT
REPO_ROOT="$(cd "$SCRIPT_DIR/../../../.." && pwd)"
local MONITOR_PID="" LOG_REPORTER_PID=""
local TEST_EXIT_CODE=0
cleanup() {
[[ -n "$LOG_REPORTER_PID" ]] && kill -0 "$LOG_REPORTER_PID" 2>/dev/null && kill "$LOG_REPORTER_PID" 2>/dev/null || true
[[ -n "$MONITOR_PID" ]] && kill -0 "$MONITOR_PID" 2>/dev/null && kill "$MONITOR_PID" 2>/dev/null || true
if [[ -f "$GPU_MONITOR_DATA_ROOT/current_run_id" ]]; then
echo "--- Finalizing: bundling resource (gpu) monitor data ---"
local TMPF BUNDLE_LINE
TMPF=$(mktemp)
"$SCRIPT_DIR/resource_monitor.sh" finalize --backend gpu 2>&1 | tee "$TMPF"
BUNDLE_LINE=$(grep '^GPU_MONITOR_BUNDLE_DIR=' "$TMPF" || true)
rm -f "$TMPF"
if [[ -n "$BUNDLE_LINE" ]]; then
eval "$BUNDLE_LINE"
if [[ -d "${GPU_MONITOR_BUNDLE_DIR:-}" ]]; then
echo "--- Resource monitor bundle dir: $GPU_MONITOR_BUNDLE_DIR ---"
echo "--- Line chart: open in browser: $GPU_MONITOR_BUNDLE_DIR/report.html ---"
fi
fi
fi
exit "${TEST_EXIT_CODE:-0}"
}
trap cleanup EXIT
if command -v nvidia-smi &>/dev/null; then
"$SCRIPT_DIR/resource_monitor.sh" start --backend gpu "$GPU_MONITOR_DEVICES" "$RESOURCE_MONITOR_INTERVAL" &
MONITOR_PID=$!
echo "[Resource Monitor (gpu)] Started (PID $MONITOR_PID), interval=${RESOURCE_MONITOR_INTERVAL}s, devices=$GPU_MONITOR_DEVICES; log every ${RESOURCE_MONITOR_LOG_INTERVAL}s."
else
echo "[Resource Monitor (gpu)] nvidia-smi not found; skipping."
fi
(
sleep 10
while true; do
sleep "$RESOURCE_MONITOR_LOG_INTERVAL"
local RID_FILE RUN_ID CSV LINE
RID_FILE="$GPU_MONITOR_DATA_ROOT/current_run_id"
[[ -f "$RID_FILE" ]] || continue
RUN_ID=$(cat "$RID_FILE" 2>/dev/null)
CSV="$GPU_MONITOR_DATA_ROOT/$RUN_ID/gpu_metrics.csv"
[[ -f "$CSV" ]] || continue
LINE=$(tail -1 "$CSV" 2>/dev/null)
[[ -n "$LINE" ]] && echo "[GPU] $LINE"
done
) &
LOG_REPORTER_PID=$!
(cd "$REPO_ROOT" && "${CMD[@]}") || TEST_EXIT_CODE=$?
exit $TEST_EXIT_CODE
}
# ---------- dispatch ----------
case "$SUBCMD" in
start) cmd_start "${@:2}" ;;
finalize) cmd_finalize "${@:2}" ;;
run) cmd_run "$@" ;;
*)
echo "Usage: $0 { start [--backend gpu|cpu|npu] [gpu_ids] [interval] | finalize [--backend gpu|cpu|npu] [run_id] | run [--backend gpu|cpu|npu] -- <command> }" >&2
echo " start - background monitor (currently only backend=gpu)" >&2
echo " finalize - bundle current run, print GPU_MONITOR_BUNDLE_DIR= / RESOURCE_MONITOR_BUNDLE_DIR=" >&2
echo " run - start + command + finalize (generate report.html only)" >&2
echo " --backend gpu|cpu|npu (default: gpu; cpu/npu reserved for future use)" >&2
exit 1
;;
esac
@@ -0,0 +1,286 @@
"""
Stability test cases: start OmniServer first, then run benchmark traffic with either
`request-rate` or `max-concurrency` for a fixed duration. No new requests are sent
after the duration is reached, and the test asserts that there are no failed requests.
The overall flow matches the perf logic: `load_configs`, `modify_stage`,
`create_unique_server_params`, `create_test_parameter_mapping`,
`get_benchmark_params_for_server`, `create_benchmark_indices`, and the
`omni_server` fixture are aligned with perf. Only the benchmark execution
(`run_stability_benchmark`, which is duration-based here) and the test cases differ.
All test-specific parameters, such as `duration_sec`, `request_rate` /
`max_concurrency`, and `num_prompts_per_batch`, are configured in
`tests/dfx/stability/tests/test.json` and are no longer overridden
through environment variables.
"""
import json
import os
import threading
import time
from pathlib import Path
from typing import Any
import pytest
from tests.conftest import OmniServer
from tests.dfx.conftest import (
create_benchmark_indices,
create_test_parameter_mapping,
create_unique_server_params,
get_benchmark_params_for_server,
load_configs,
)
from tests.dfx.perf.scripts.run_benchmark import run_benchmark
STABILITY_DIR = Path(__file__).resolve().parent.parent
STAGE_CONFIGS_DIR = STABILITY_DIR / "stage_configs"
CONFIG_FILE_PATH = str(STABILITY_DIR / "tests" / "test.json")
DEFAULT_NUM_PROMPTS_PER_BATCH = 20
try:
BENCHMARK_CONFIGS = load_configs(CONFIG_FILE_PATH)
except FileNotFoundError:
BENCHMARK_CONFIGS = []
test_params = create_unique_server_params(BENCHMARK_CONFIGS, STAGE_CONFIGS_DIR) if BENCHMARK_CONFIGS else []
server_to_benchmark_mapping = create_test_parameter_mapping(BENCHMARK_CONFIGS) if BENCHMARK_CONFIGS else {}
_omni_server_lock = threading.Lock()
benchmark_indices = create_benchmark_indices(BENCHMARK_CONFIGS, server_to_benchmark_mapping)
def _build_base_args(params: dict[str, Any], host: str, port: int) -> list[str]:
exclude = {
"request_rate",
"max_concurrency",
"num_prompts",
"baseline",
"duration_sec",
"num_prompts_per_batch",
}
args = ["--host", host, "--port", str(port)]
for key, value in params.items():
if key in exclude or value is None:
continue
arg_name = f"--{key.replace('_', '-')}"
if isinstance(value, bool) and value:
args.append(arg_name)
elif isinstance(value, dict):
args.extend([arg_name, json.dumps(value, ensure_ascii=False, separators=(",", ":"))])
elif not isinstance(value, bool):
args.extend([arg_name, str(value)])
return args
def _run_one_benchmark_batch(
host: str,
port: int,
params: dict[str, Any],
num_prompts: int,
request_rate: float | None,
max_concurrency: int | None,
result_dir: str,
batch_index: int,
) -> dict[str, Any]:
base = _build_base_args(params, host, port)
if request_rate is not None:
args = base + ["--request-rate", str(request_rate), "--num-prompts", str(num_prompts)]
flow = request_rate
else:
args = base + [
"--max-concurrency",
str(max_concurrency),
"--num-prompts",
str(num_prompts),
"--request-rate",
"inf",
]
flow = max_concurrency
dataset_name = params.get("dataset_name", "random")
old_benchmark_dir = os.environ.get("BENCHMARK_DIR")
try:
os.environ["BENCHMARK_DIR"] = result_dir
result = run_benchmark(
args=args,
test_name="stability",
flow=flow,
dataset_name=dataset_name,
num_prompt=num_prompts,
)
return result
except (FileNotFoundError, OSError) as e:
# Surface batch failure so the stability test does not false-pass when
# run_benchmark fails before writing JSON (e.g. command not found).
return {
"completed": 0,
"failed": 1,
"duration": 0.0,
"errors": [f"Benchmark batch failed: {type(e).__name__}: {e}"],
}
finally:
if old_benchmark_dir is not None:
os.environ["BENCHMARK_DIR"] = old_benchmark_dir
elif "BENCHMARK_DIR" in os.environ:
os.environ.pop("BENCHMARK_DIR")
def _merge_batch_results(batch_results: list[dict[str, Any]], total_duration_sec: float) -> dict[str, Any]:
if not batch_results:
return {"completed": 0, "failed": 0, "duration": total_duration_sec, "errors": []}
completed = sum(r.get("completed", 0) for r in batch_results)
failed = sum(r.get("failed", 0) for r in batch_results)
merged: dict[str, Any] = {
"completed": completed,
"failed": failed,
"duration": total_duration_sec,
"errors": [],
}
for r in batch_results:
merged["errors"].extend(r.get("errors") or [])
return merged
def _print_merged_report(result: dict[str, Any]) -> None:
"""Print the final summary: successful requests, failed requests, and total duration only."""
fmt = "{:<40} {:<10}"
fmt_float = "{:<40} {:<10.2f}"
completed = result.get("completed", 0)
failed = result.get("failed", 0)
duration = float(result.get("duration", 0.0) or 0.0)
print("\n============ Stability Benchmark Summary ============")
print(fmt.format("Successful requests:", completed))
print(fmt.format("Failed requests:", failed))
print(fmt_float.format("Total duration (s):", duration))
print("==================================================\n")
def run_stability_benchmark(
host: str,
port: int,
duration_sec: int | float,
params: dict[str, Any],
*,
request_rate: float | None = None,
max_concurrency: int | None = None,
result_filename: str | None = None,
result_dir: str = "./",
num_prompts_per_batch: int = DEFAULT_NUM_PROMPTS_PER_BATCH,
) -> dict[str, Any]:
if (request_rate is None) == (max_concurrency is None):
raise ValueError("Exactly one of request_rate or max_concurrency must be specified")
start_time = time.perf_counter()
batch_results: list[dict[str, Any]] = []
batch_index = 0
while True:
if (time.perf_counter() - start_time) >= duration_sec:
break
result = _run_one_benchmark_batch(
host=host,
port=port,
params=params,
num_prompts=num_prompts_per_batch,
request_rate=request_rate,
max_concurrency=max_concurrency,
result_dir=result_dir,
batch_index=batch_index,
)
batch_results.append(result)
batch_index += 1
if (time.perf_counter() - start_time) >= duration_sec:
break
total_duration = time.perf_counter() - start_time
merged = _merge_batch_results(batch_results, total_duration)
_print_merged_report(merged)
if result_filename and result_dir:
result_path = Path(result_dir) / result_filename
with open(result_path, "w", encoding="utf-8") as f:
json.dump(merged, f, indent=2, ensure_ascii=False)
return merged
@pytest.fixture(scope="module")
def omni_server(request):
"""Start vLLM-Omni server as a subprocess with actual model weights.
Uses session scope so the server starts only once for the entire test session.
Multi-stage initialization can take 10-20+ minutes.
"""
with _omni_server_lock:
test_name, model, stage_config_path = request.param
print(f"Starting OmniServer with test: {test_name}, model: {model}")
with OmniServer(model, ["--stage-configs-path", stage_config_path, "--stage-init-timeout", "120"]) as server:
server.test_name = test_name
print("OmniServer started successfully")
yield server
print("OmniServer stopping...")
print("OmniServer stopped")
@pytest.fixture(params=benchmark_indices)
def stability_benchmark_params(request, omni_server):
"""Benchmark parameters fixture with proper parametrization (same as perf)."""
test_name, param_index = request.param
if test_name != omni_server.test_name:
pytest.skip(f"Skipping parameter for {test_name} - current server is {omni_server.test_name}")
all_params = get_benchmark_params_for_server(test_name, server_to_benchmark_mapping)
if not all_params:
raise ValueError(f"No benchmark parameters found for test: {test_name}")
if param_index >= len(all_params):
raise ValueError(f"No benchmark parameters found for index {param_index} in test: {test_name}")
current = param_index + 1
total = len(all_params)
print(f"\n Running benchmark {current}/{total} for {test_name}")
return {"test_name": test_name, "params": all_params[param_index]}
@pytest.mark.slow
@pytest.mark.parametrize("omni_server", test_params, indirect=True)
@pytest.mark.parametrize("stability_benchmark_params", benchmark_indices, indirect=True)
def test_benchmark_stability(omni_server, stability_benchmark_params):
"""Run the benchmark for a fixed duration using request-rate or max-concurrency and assert zero failed requests."""
test_name = stability_benchmark_params["test_name"]
params = stability_benchmark_params["params"]
duration_sec = params.get("duration_sec", 300)
num_prompts_per_batch = params.get("num_prompts_per_batch", DEFAULT_NUM_PROMPTS_PER_BATCH)
request_rate = params.get("request_rate")
max_concurrency = params.get("max_concurrency")
bench_params = {
k: v
for k, v in params.items()
if k not in ("duration_sec", "request_rate", "max_concurrency", "num_prompts_per_batch")
}
result = run_stability_benchmark(
host=omni_server.host,
port=omni_server.port,
duration_sec=duration_sec,
params=bench_params,
request_rate=request_rate,
max_concurrency=max_concurrency,
result_dir=str(STABILITY_DIR),
num_prompts_per_batch=num_prompts_per_batch,
)
assert result.get("failed", 0) == 0, f"[{test_name}] Failed requests detected: {result.get('errors', [])}"
assert result.get("completed", 0) > 0, f"[{test_name}] No requests completed"
@@ -0,0 +1,101 @@
# Stage config for running Qwen3-Omni-MoE with 3-stage architecture
# Stage 0: Thinker (multimodal understanding + text generation)
# Stage 1: Talker (text embeddings → 8-layer RVQ codec codes)
# Stage 2: Code2Wav (8-layer RVQ codes → audio waveform)
# The following config has been verified on 2x H100-80G GPUs.
async_chunk: false
stage_args:
- stage_id: 0
stage_type: llm # Use llm stage type to launch OmniLLM
runtime:
devices: "0"
max_batch_size: 64
engine_args:
model_stage: thinker
model_arch: Qwen3OmniMoeForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.9
enforce_eager: false
trust_remote_code: true
engine_output_type: latent # Output hidden states for talker
distributed_executor_backend: "mp"
enable_prefix_caching: false
max_num_batched_tokens: 32768
hf_config_name: thinker_config
tensor_parallel_size: 1
final_output: true
final_output_type: text
is_comprehension: true
default_sampling_params:
temperature: 0.4
top_p: 0.9
top_k: 1
max_tokens: 2048
seed: 42
detokenize: True
repetition_penalty: 1.05
- stage_id: 1
stage_type: llm # Use llm stage type to launch OmniLLM
runtime:
devices: "1"
max_batch_size: 64
engine_args:
model_stage: talker
model_arch: Qwen3OmniMoeForConditionalGeneration
worker_type: ar
scheduler_cls: vllm_omni.core.sched.omni_ar_scheduler.OmniARScheduler
gpu_memory_utilization: 0.6
enforce_eager: false
trust_remote_code: true
engine_output_type: latent # Output codec codes for code2wav
enable_prefix_caching: false
max_num_batched_tokens: 32768
distributed_executor_backend: "mp"
hf_config_name: talker_config
engine_input_source: [0]
custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker
# final_output: true
# final_output_type: text
default_sampling_params:
temperature: 0.9
top_k: 50
max_tokens: 4096
seed: 42
detokenize: False
repetition_penalty: 1.05
stop_token_ids: [2150]
- stage_id: 2
stage_type: llm # Use llm stage type to launch OmniLLM
runtime:
devices: "1"
max_batch_size: 64
engine_args:
model_stage: code2wav
model_arch: Qwen3OmniMoeForConditionalGeneration
worker_type: generation
scheduler_cls: vllm_omni.core.sched.omni_generation_scheduler.OmniGenerationScheduler
enforce_eager: true
trust_remote_code: true
async_scheduling: false
enable_prefix_caching: false
engine_output_type: audio # Final output: audio waveform
gpu_memory_utilization: 0.1
distributed_executor_backend: "mp"
max_num_batched_tokens: 1000000
hf_config_name: thinker_config
engine_input_source: [1]
custom_process_input_func: vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav
final_output: true
final_output_type: audio
default_sampling_params:
temperature: 0.0
top_p: 1.0
top_k: -1
max_tokens: 65536
seed: 42
detokenize: True
repetition_penalty: 1.1
+86
View File
@@ -0,0 +1,86 @@
[
{
"test_name": "test_qwen3_omni_stability",
"server_params": {
"model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
"stage_config_name": "qwen3_omni.yaml"
},
"benchmark_params": [
{
"dataset_name": "random",
"backend": "openai-chat-omni",
"endpoint": "/v1/chat/completions",
"duration_sec": 300,
"request_rate": 0.2,
"num_prompts_per_batch": 10,
"random_input_len": 2500,
"random_output_len": 900,
"ignore_eos": true,
"percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
},
{
"dataset_name": "random",
"backend": "openai-chat-omni",
"endpoint": "/v1/chat/completions",
"duration_sec": 300,
"max_concurrency": 2,
"num_prompts_per_batch": 10,
"random_input_len": 2500,
"random_output_len": 900,
"ignore_eos": true,
"percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
}
]
},
{
"test_name": "test_qwen3_omni_stability_async_chunk",
"server_params": {
"model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
"stage_config_name": "qwen3_omni.yaml",
"update": {
"async_chunk": true,
"stage_args": {
"0": {
"engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.thinker2talker_async_chunk"
},
"1": {
"engine_args.custom_process_next_stage_input_func": "vllm_omni.model_executor.stage_input_processors.qwen3_omni.talker2code2wav_async_chunk"
}
}
},
"delete": {
"stage_args": {
"2": [
"custom_process_input_func"
]
}
}
},
"benchmark_params": [
{
"dataset_name": "random",
"backend": "openai-chat-omni",
"endpoint": "/v1/chat/completions",
"duration_sec": 300,
"request_rate": 0.2,
"num_prompts_per_batch": 10,
"random_input_len": 2500,
"random_output_len": 900,
"ignore_eos": true,
"percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
},
{
"dataset_name": "random",
"backend": "openai-chat-omni",
"endpoint": "/v1/chat/completions",
"duration_sec": 300,
"max_concurrency": 2,
"num_prompts_per_batch": 10,
"random_input_len": 2500,
"random_output_len": 900,
"ignore_eos": true,
"percentile-metrics": "ttft,tpot,itl,e2el,audio_rtf,audio_ttfp,audio_duration"
}
]
}
]