[NPU] Improve the execution efficiency and maintainability of pr‑test‑npu (#33724)
Co-authored-by: Even Zhou <even.y.zhou@outlook.com> Co-authored-by: sglang-npu-bot <sglangnpu@163.com>
This commit is contained in:
Cherry_ming
co-authored by
Even Zhou
sglang-npu-bot
Even Zhou
sglang-npu-bot
parent
e732c0a9dc
commit
dd5d82bead
@@ -1,10 +1,15 @@
|
||||
import glob
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
@@ -289,9 +294,117 @@ class TestNpuAccuracyTestCaseBase(CustomTestCase):
|
||||
max_attempts = 2
|
||||
n_runs = 3
|
||||
accuracy = 0.1
|
||||
test_type = "accuracy"
|
||||
|
||||
@classmethod
|
||||
def _get_tc_name(cls):
|
||||
"""Derive the test case name from the test file (filename without
|
||||
extension). Mirrors the workflow's ``tc_name=${test_case##*/}`` logic
|
||||
so each case in a suite writes to its own output path."""
|
||||
try:
|
||||
tc_file = inspect.getfile(cls)
|
||||
except (TypeError, OSError):
|
||||
tc_file = getattr(sys.modules.get(cls.__module__), "__file__", "")
|
||||
return os.path.splitext(os.path.basename(tc_file))[0]
|
||||
|
||||
@classmethod
|
||||
def _setup_per_case_output(cls):
|
||||
"""Set up per-case output directories and env vars.
|
||||
|
||||
Extracted from ``nightly-test-npu-e2e-single-node.yml`` so that when a
|
||||
suite is executed, each case writes its metrics/plog to a path derived
|
||||
from the case file rather than the suite name.
|
||||
"""
|
||||
cls.tc_name = cls._get_tc_name()
|
||||
current_date = datetime.now().strftime("%Y%m%d")
|
||||
test_type = getattr(cls, "test_type", "accuracy")
|
||||
base_output = f"/root/.cache/tests/output/{test_type}/{current_date}"
|
||||
os.makedirs(base_output, exist_ok=True)
|
||||
cls.metrics_data_file = os.path.join(base_output, cls.tc_name)
|
||||
os.makedirs(cls.metrics_data_file, exist_ok=True)
|
||||
# Override env vars so evalscope/dump_metric write to per-case paths.
|
||||
os.environ["METRICS_DATA_FILE"] = cls.metrics_data_file
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = os.path.join(
|
||||
cls.metrics_data_file, "metrics"
|
||||
)
|
||||
logger.info(
|
||||
"Per-case output: tc_name=%s metrics_data_file=%s",
|
||||
cls.tc_name,
|
||||
cls.metrics_data_file,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _save_metrics_json(cls):
|
||||
"""Write per-case ``metrics.json`` from ``dump_metric`` JSONL files.
|
||||
|
||||
Replaces the workflow's stdout-parsing + ``dump_metrics.py`` logic so
|
||||
each case in a suite persists its own metrics snapshot.
|
||||
"""
|
||||
if not getattr(cls, "metrics_data_file", None):
|
||||
return
|
||||
metrics = {}
|
||||
baselines = {}
|
||||
pattern = os.path.join(cls.metrics_data_file, "metrics.*.jsonl")
|
||||
for jsonl_path in glob.glob(pattern):
|
||||
try:
|
||||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
record = json.loads(line)
|
||||
name = record.get("metric_name")
|
||||
value = record.get("value")
|
||||
if name is None:
|
||||
continue
|
||||
if name.endswith("_baseline"):
|
||||
baselines[name[: -len("_baseline")]] = value
|
||||
else:
|
||||
metrics[name] = value
|
||||
except Exception as e:
|
||||
logger.warning("Failed to read %s: %s", jsonl_path, e)
|
||||
out_path = os.path.join(cls.metrics_data_file, "metrics.json")
|
||||
payload = {
|
||||
"test_case": cls.tc_name,
|
||||
"test_type": getattr(cls, "test_type", "accuracy"),
|
||||
"metrics": metrics,
|
||||
"baselines": baselines,
|
||||
}
|
||||
try:
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
json.dump(payload, f, ensure_ascii=False)
|
||||
logger.info("Saved per-case metrics to %s", out_path)
|
||||
except Exception as e:
|
||||
logger.warning("Failed to write metrics.json: %s", e)
|
||||
|
||||
@classmethod
|
||||
def _backup_plog(cls):
|
||||
"""Backup Ascend plog files to a per-case path.
|
||||
|
||||
Replaces the workflow's ``Backup plog`` step so each case in a suite
|
||||
gets its own plog snapshot instead of all cases sharing the suite name.
|
||||
"""
|
||||
plog_path = "/root/ascend/log/debug/plog"
|
||||
if not os.path.isdir(plog_path):
|
||||
return
|
||||
tc_name = getattr(cls, "tc_name", None)
|
||||
if not tc_name:
|
||||
return
|
||||
hostname = os.getenv("HOSTNAME", "unknown")
|
||||
target = os.path.join("/root/.cache/tests/logs/plog", tc_name, hostname)
|
||||
os.makedirs(target, exist_ok=True)
|
||||
for name in os.listdir(plog_path):
|
||||
src = os.path.join(plog_path, name)
|
||||
if os.path.isfile(src):
|
||||
try:
|
||||
shutil.copy2(src, os.path.join(target, name))
|
||||
except Exception as e:
|
||||
logger.warning("Failed to copy plog %s: %s", name, e)
|
||||
logger.info("Backed up plog to %s", target)
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls._setup_per_case_output()
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
env = os.environ.copy()
|
||||
for key, value in env.items():
|
||||
@@ -318,6 +431,8 @@ class TestNpuAccuracyTestCaseBase(CustomTestCase):
|
||||
kill_process_tree(cls.process.pid)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during tearDown: {e}")
|
||||
cls._save_metrics_json()
|
||||
cls._backup_plog()
|
||||
|
||||
def run_accuracy(self):
|
||||
parsed_url = urlparse(self.base_url)
|
||||
|
||||
@@ -1,9 +1,15 @@
|
||||
import glob
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from functools import wraps
|
||||
from urllib.parse import urlparse
|
||||
|
||||
@@ -909,8 +915,115 @@ class TestNpuPerformanceTestCaseBase(CustomTestCase):
|
||||
generation_kwargs = None
|
||||
pop_sglang_is_in_ci_for_gsp = False
|
||||
|
||||
@classmethod
|
||||
def _get_tc_name(cls):
|
||||
"""Derive the test case name from the test file (filename without
|
||||
extension). Mirrors the workflow's ``tc_name=${test_case##*/}`` logic
|
||||
so each case in a suite writes to its own output path."""
|
||||
try:
|
||||
tc_file = inspect.getfile(cls)
|
||||
except (TypeError, OSError):
|
||||
tc_file = getattr(sys.modules.get(cls.__module__), "__file__", "")
|
||||
return os.path.splitext(os.path.basename(tc_file))[0]
|
||||
|
||||
@classmethod
|
||||
def _setup_per_case_output(cls):
|
||||
"""Set up per-case output directories and env vars.
|
||||
|
||||
Extracted from ``nightly-test-npu-e2e-single-node.yml`` so that when a
|
||||
suite is executed, each case writes its metrics/plog to a path derived
|
||||
from the case file rather than the suite name.
|
||||
"""
|
||||
cls.tc_name = cls._get_tc_name()
|
||||
current_date = datetime.now().strftime("%Y%m%d")
|
||||
test_type = getattr(cls, "test_type", "perf")
|
||||
base_output = f"/root/.cache/tests/output/{test_type}/{current_date}"
|
||||
os.makedirs(base_output, exist_ok=True)
|
||||
cls.metrics_data_file = os.path.join(base_output, cls.tc_name)
|
||||
os.makedirs(cls.metrics_data_file, exist_ok=True)
|
||||
# Override env vars so evalscope/dump_metric write to per-case paths.
|
||||
os.environ["METRICS_DATA_FILE"] = cls.metrics_data_file
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = os.path.join(
|
||||
cls.metrics_data_file, "metrics"
|
||||
)
|
||||
logger.info(
|
||||
"Per-case output: tc_name=%s metrics_data_file=%s",
|
||||
cls.tc_name,
|
||||
cls.metrics_data_file,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _save_metrics_json(cls):
|
||||
"""Write per-case ``metrics.json`` from ``dump_metric`` JSONL files.
|
||||
|
||||
Replaces the workflow's stdout-parsing + ``dump_metrics.py`` logic so
|
||||
each case in a suite persists its own metrics snapshot.
|
||||
"""
|
||||
if not getattr(cls, "metrics_data_file", None):
|
||||
return
|
||||
metrics = {}
|
||||
baselines = {}
|
||||
pattern = os.path.join(cls.metrics_data_file, "metrics.*.jsonl")
|
||||
for jsonl_path in glob.glob(pattern):
|
||||
try:
|
||||
with open(jsonl_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
record = json.loads(line)
|
||||
name = record.get("metric_name")
|
||||
value = record.get("value")
|
||||
if name is None:
|
||||
continue
|
||||
if name.endswith("_baseline"):
|
||||
baselines[name[: -len("_baseline")]] = value
|
||||
else:
|
||||
metrics[name] = value
|
||||
except Exception as e:
|
||||
logger.warning("Failed to read %s: %s", jsonl_path, e)
|
||||
out_path = os.path.join(cls.metrics_data_file, "metrics.json")
|
||||
payload = {
|
||||
"test_case": cls.tc_name,
|
||||
"test_type": getattr(cls, "test_type", "accuracy"),
|
||||
"metrics": metrics,
|
||||
"baselines": baselines,
|
||||
}
|
||||
try:
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
json.dump(payload, f, ensure_ascii=False)
|
||||
logger.info("Saved per-case metrics to %s", out_path)
|
||||
except Exception as e:
|
||||
logger.warning("Failed to write metrics.json: %s", e)
|
||||
|
||||
@classmethod
|
||||
def _backup_plog(cls):
|
||||
"""Backup Ascend plog files to a per-case path.
|
||||
|
||||
Replaces the workflow's ``Backup plog`` step so each case in a suite
|
||||
gets its own plog snapshot instead of all cases sharing the suite name.
|
||||
"""
|
||||
plog_path = "/root/ascend/log/debug/plog"
|
||||
if not os.path.isdir(plog_path):
|
||||
return
|
||||
tc_name = getattr(cls, "tc_name", None)
|
||||
if not tc_name:
|
||||
return
|
||||
hostname = os.getenv("HOSTNAME", "unknown")
|
||||
target = os.path.join("/root/.cache/tests/logs/plog", tc_name, hostname)
|
||||
os.makedirs(target, exist_ok=True)
|
||||
for name in os.listdir(plog_path):
|
||||
src = os.path.join(plog_path, name)
|
||||
if os.path.isfile(src):
|
||||
try:
|
||||
shutil.copy2(src, os.path.join(target, name))
|
||||
except Exception as e:
|
||||
logger.warning("Failed to copy plog %s: %s", name, e)
|
||||
logger.info("Backed up plog to %s", target)
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls._setup_per_case_output()
|
||||
cls.base_url = DEFAULT_URL_FOR_TEST
|
||||
env = os.environ.copy()
|
||||
for key, value in env.items():
|
||||
@@ -937,6 +1050,8 @@ class TestNpuPerformanceTestCaseBase(CustomTestCase):
|
||||
kill_process_tree(cls.process.pid)
|
||||
except Exception as e:
|
||||
logger.error(f"Error during tearDown: {e}")
|
||||
cls._save_metrics_json()
|
||||
cls._backup_plog()
|
||||
|
||||
@retry()
|
||||
def run_throughput(self):
|
||||
|
||||
Reference in New Issue
Block a user