@@ -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"