From c14ab819e38ce011962ef0cd3966355808c87311 Mon Sep 17 00:00:00 2001 From: zhumingjue138 Date: Mon, 23 Mar 2026 19:04:05 +0800 Subject: [PATCH] [Test] L5 Long-Term Stability Test and GPU Memory Monitoring Main L5 last (#1799) --- .buildkite/test-nightly.yml | 2 +- docs/contributing/ci/CI_5levels.md | 87 ++++- tests/dfx/conftest.py | 97 +++++ tests/{ => dfx}/perf/scripts/run_benchmark.py | 102 +---- .../perf/stage_configs/qwen3_omni.yaml | 0 .../perf/stage_configs/qwen3_tts.yaml | 0 tests/{ => dfx}/perf/tests/test.json | 0 tests/dfx/stability/README.md | 90 +++++ tests/dfx/stability/conftest.py | 169 ++++++++ tests/dfx/stability/scripts/__init__.py | 1 + .../dfx/stability/scripts/generate_report.py | 364 ++++++++++++++++++ .../dfx/stability/scripts/resource_monitor.sh | 329 ++++++++++++++++ .../scripts/test_benchmark_stability.py | 286 ++++++++++++++ .../stability/stage_configs/qwen3_omni.yaml | 101 +++++ tests/dfx/stability/tests/test.json | 86 +++++ 15 files changed, 1613 insertions(+), 101 deletions(-) create mode 100644 tests/dfx/conftest.py rename tests/{ => dfx}/perf/scripts/run_benchmark.py (70%) rename tests/{ => dfx}/perf/stage_configs/qwen3_omni.yaml (100%) rename tests/{ => dfx}/perf/stage_configs/qwen3_tts.yaml (100%) rename tests/{ => dfx}/perf/tests/test.json (100%) create mode 100644 tests/dfx/stability/README.md create mode 100644 tests/dfx/stability/conftest.py create mode 100644 tests/dfx/stability/scripts/__init__.py create mode 100644 tests/dfx/stability/scripts/generate_report.py create mode 100644 tests/dfx/stability/scripts/resource_monitor.sh create mode 100644 tests/dfx/stability/scripts/test_benchmark_stability.py create mode 100644 tests/dfx/stability/stage_configs/qwen3_omni.yaml create mode 100644 tests/dfx/stability/tests/test.json diff --git a/.buildkite/test-nightly.yml b/.buildkite/test-nightly.yml index 9d6b5f608..6a6123f24 100644 --- a/.buildkite/test-nightly.yml +++ b/.buildkite/test-nightly.yml @@ -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" diff --git a/docs/contributing/ci/CI_5levels.md b/docs/contributing/ci/CI_5levels.md index 1f9e6a188..e30456861 100644 --- a/docs/contributing/ci/CI_5levels.md +++ b/docs/contributing/ci/CI_5levels.md @@ -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
/tests/e2e/offline_inference/test_{model_name}_expansion.py
Performance:
- /tests/e2e/perf/nightly.json
+ /tests/dfx/perf/tests/test.json
Doc Test:
tests/example/online_serving/test_{model_name}.py
tests/example/offline_inference/test_{model_name}.py @@ -104,7 +104,7 @@ Through five levels (L1-L5) and common (Common) specifications, the system clari Depends on reality Stability:
- tests/e2e/stability/weekly.json
+ /tests/dfx/stability/tests/test.json
Reliability:
tests/e2e/reliability/test_{model_name}.py @@ -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***:
Test Examples -```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. +
-- - ***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 diff --git a/tests/dfx/conftest.py b/tests/dfx/conftest.py new file mode 100644 index 000000000..e54141b34 --- /dev/null +++ b/tests/dfx/conftest.py @@ -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 diff --git a/tests/perf/scripts/run_benchmark.py b/tests/dfx/perf/scripts/run_benchmark.py similarity index 70% rename from tests/perf/scripts/run_benchmark.py rename to tests/dfx/perf/scripts/run_benchmark.py index 704d47f9b..bc6abbb02 100644 --- a/tests/perf/scripts/run_benchmark.py +++ b/tests/dfx/perf/scripts/run_benchmark.py @@ -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}") diff --git a/tests/perf/stage_configs/qwen3_omni.yaml b/tests/dfx/perf/stage_configs/qwen3_omni.yaml similarity index 100% rename from tests/perf/stage_configs/qwen3_omni.yaml rename to tests/dfx/perf/stage_configs/qwen3_omni.yaml diff --git a/tests/perf/stage_configs/qwen3_tts.yaml b/tests/dfx/perf/stage_configs/qwen3_tts.yaml similarity index 100% rename from tests/perf/stage_configs/qwen3_tts.yaml rename to tests/dfx/perf/stage_configs/qwen3_tts.yaml diff --git a/tests/perf/tests/test.json b/tests/dfx/perf/tests/test.json similarity index 100% rename from tests/perf/tests/test.json rename to tests/dfx/perf/tests/test.json diff --git a/tests/dfx/stability/README.md b/tests/dfx/stability/README.md new file mode 100644 index 000000000..121bb610d --- /dev/null +++ b/tests/dfx/stability/README.md @@ -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] -- ` | 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] -- +``` + +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_/gpu_metrics.csv`. + - After `finalize`, you get `gpu_monitor_bundle_/` 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. diff --git a/tests/dfx/stability/conftest.py b/tests/dfx/stability/conftest.py new file mode 100644 index 000000000..3a0aee760 --- /dev/null +++ b/tests/dfx/stability/conftest.py @@ -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") diff --git a/tests/dfx/stability/scripts/__init__.py b/tests/dfx/stability/scripts/__init__.py new file mode 100644 index 000000000..6d73ea952 --- /dev/null +++ b/tests/dfx/stability/scripts/__init__.py @@ -0,0 +1 @@ +# scripts package for stability benchmark diff --git a/tests/dfx/stability/scripts/generate_report.py b/tests/dfx/stability/scripts/generate_report.py new file mode 100644 index 000000000..986a80831 --- /dev/null +++ b/tests/dfx/stability/scripts/generate_report.py @@ -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"GPU {gpu}{s['min']}%{s['max']}%" + f"{s['avg']}%{s['p50']}{s['p95']}{s['samples']}" + ) + 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"{time_range}GPU {p['gpu_index']}" + f"{pct_str}{p['type']} (threshold {p['threshold']}%){p['samples']}" + ) + anomaly_table = "\n".join(anomaly_cells) if anomaly_cells else "None" + + html = f""" + + + + GPU memory monitor report - {run_id} + + + + +

Stability GPU memory monitor report

+

Run: {run_id} | Data file: {os.path.basename(csv_path)} | Samples: {len(rows)}

+

Statistics

+ + + {stats_table} +
GPUMin %Max %Avg %P50P95Samples
+

Memory utilization over time

+ +
+ +
+

Anomalies (high/low threshold)

+

Consecutive samples merged into periods; all periods listed.

+ + + {anomaly_table} +
PeriodGPUUtil %TypeSamples
+ + + +""" + 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 [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()) diff --git a/tests/dfx/stability/scripts/resource_monitor.sh b/tests/dfx/stability/scripts/resource_monitor.sh new file mode 100644 index 000000000..dccc5585a --- /dev/null +++ b/tests/dfx/stability/scripts/resource_monitor.sh @@ -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 -- +# +# 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] -- " >&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] -- }" >&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 diff --git a/tests/dfx/stability/scripts/test_benchmark_stability.py b/tests/dfx/stability/scripts/test_benchmark_stability.py new file mode 100644 index 000000000..e8568652d --- /dev/null +++ b/tests/dfx/stability/scripts/test_benchmark_stability.py @@ -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" diff --git a/tests/dfx/stability/stage_configs/qwen3_omni.yaml b/tests/dfx/stability/stage_configs/qwen3_omni.yaml new file mode 100644 index 000000000..802f8dd24 --- /dev/null +++ b/tests/dfx/stability/stage_configs/qwen3_omni.yaml @@ -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 diff --git a/tests/dfx/stability/tests/test.json b/tests/dfx/stability/tests/test.json new file mode 100644 index 000000000..95993c9c5 --- /dev/null +++ b/tests/dfx/stability/tests/test.json @@ -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" + } + ] + } +]