Add unified metrics collection framework (v1) (#16064)
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
This commit is contained in:
Hudson Xing
@@ -51,6 +51,9 @@ def run_eval_once(args, base_url: str, eval_obj: Eval) -> dict:
|
||||
|
||||
|
||||
def run_eval(args):
|
||||
# Lazy import to avoid circular dependency with test_utils
|
||||
from sglang.test.test_utils import dump_metric
|
||||
|
||||
set_ulimit()
|
||||
|
||||
if "OPENAI_API_KEY" not in os.environ:
|
||||
@@ -132,6 +135,18 @@ def run_eval(args):
|
||||
metrics = result.metrics | {"score": result.score}
|
||||
print(f"Total latency: {latency:.3f} s")
|
||||
print(f"Score: {metrics['score']:.3f}")
|
||||
|
||||
# Report metrics to unified collection framework
|
||||
dump_metric(
|
||||
f"{args.eval_name}_score",
|
||||
metrics["score"],
|
||||
labels={"model": sampler.model, "eval": args.eval_name},
|
||||
)
|
||||
dump_metric(
|
||||
f"{args.eval_name}_latency",
|
||||
latency,
|
||||
labels={"model": sampler.model, "eval": args.eval_name},
|
||||
)
|
||||
else:
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
@@ -157,6 +172,17 @@ def run_eval(args):
|
||||
metrics = result.metrics | {"scores": scores_repeat}
|
||||
metrics = metrics | {"mean_score": mean_score}
|
||||
|
||||
# Report metrics to unified collection framework
|
||||
dump_metric(
|
||||
f"{args.eval_name}_mean_score",
|
||||
mean_score,
|
||||
labels={
|
||||
"model": sampler.model,
|
||||
"eval": args.eval_name,
|
||||
"repeat": args.repeat,
|
||||
},
|
||||
)
|
||||
|
||||
executor.shutdown()
|
||||
|
||||
# Dump reports
|
||||
|
||||
132
python/sglang/test/test_dump_metric.py
Normal file
132
python/sglang/test/test_dump_metric.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""Unit tests for dump_metric() function."""
|
||||
|
||||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from sglang.test.test_utils import dump_metric
|
||||
|
||||
|
||||
class TestDumpMetric(unittest.TestCase):
|
||||
"""Test suite for dump_metric() function."""
|
||||
|
||||
_ENV_KEYS_TO_CLEAN = ["SGLANG_TEST_METRICS_OUTPUT", "PYTEST_CURRENT_TEST"]
|
||||
|
||||
def setUp(self):
|
||||
"""Clean up env vars before each test."""
|
||||
for key in self._ENV_KEYS_TO_CLEAN:
|
||||
os.environ.pop(key, None)
|
||||
|
||||
def tearDown(self):
|
||||
"""Clean up env vars after each test."""
|
||||
for key in self._ENV_KEYS_TO_CLEAN:
|
||||
os.environ.pop(key, None)
|
||||
|
||||
def test_writes_valid_jsonl(self):
|
||||
"""Test that dump_metric writes one valid JSON line when env is set."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
base_path = os.path.join(tmpdir, "metrics")
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = base_path
|
||||
|
||||
dump_metric("test_accuracy", 0.95, labels={"model": "llama"})
|
||||
|
||||
# Check file exists with PID suffix
|
||||
pid = os.getpid()
|
||||
jsonl_path = f"{base_path}.{pid}.jsonl"
|
||||
self.assertTrue(os.path.exists(jsonl_path))
|
||||
|
||||
# Read and validate
|
||||
with open(jsonl_path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
self.assertEqual(len(lines), 1)
|
||||
record = json.loads(lines[0])
|
||||
|
||||
# Validate required fields
|
||||
self.assertIn("filename", record)
|
||||
self.assertIn("test_case", record)
|
||||
self.assertEqual(record["metric_name"], "test_accuracy")
|
||||
self.assertEqual(record["value"], 0.95)
|
||||
|
||||
# Validate optional fields
|
||||
self.assertIn("ts", record)
|
||||
self.assertIsInstance(record["ts"], (int, float))
|
||||
self.assertEqual(record["labels"], {"model": "llama"})
|
||||
|
||||
def test_no_env_no_file(self):
|
||||
"""Test that dump_metric doesn't create file when env var not set."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
# Don't set env var
|
||||
dump_metric("test_metric", 42)
|
||||
|
||||
# Verify no files created
|
||||
files = list(Path(tmpdir).glob("*.jsonl"))
|
||||
self.assertEqual(len(files), 0)
|
||||
|
||||
def test_labels_not_serializable_stringified(self):
|
||||
"""Test that non-serializable labels are stringified."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
base_path = os.path.join(tmpdir, "metrics")
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = base_path
|
||||
|
||||
# Non-serializable label
|
||||
class NonSerializable:
|
||||
pass
|
||||
|
||||
dump_metric("test_metric", 100, labels={"obj": NonSerializable()})
|
||||
|
||||
pid = os.getpid()
|
||||
jsonl_path = f"{base_path}.{pid}.jsonl"
|
||||
with open(jsonl_path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
record = json.loads(lines[0])
|
||||
self.assertIn("labels", record)
|
||||
self.assertIsInstance(record["labels"], str)
|
||||
|
||||
def test_bool_to_int(self):
|
||||
"""Test that bool values are converted to int."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
base_path = os.path.join(tmpdir, "metrics")
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = base_path
|
||||
|
||||
dump_metric("bool_true", True)
|
||||
dump_metric("bool_false", False)
|
||||
|
||||
pid = os.getpid()
|
||||
jsonl_path = f"{base_path}.{pid}.jsonl"
|
||||
with open(jsonl_path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
self.assertEqual(len(lines), 2)
|
||||
record1 = json.loads(lines[0])
|
||||
record2 = json.loads(lines[1])
|
||||
|
||||
self.assertEqual(record1["value"], 1) # True -> 1
|
||||
self.assertEqual(record2["value"], 0) # False -> 0
|
||||
|
||||
def test_pytest_current_test_parsing(self):
|
||||
"""Test PYTEST_CURRENT_TEST parsing for test_case."""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
base_path = os.path.join(tmpdir, "metrics")
|
||||
os.environ["SGLANG_TEST_METRICS_OUTPUT"] = base_path
|
||||
os.environ["PYTEST_CURRENT_TEST"] = (
|
||||
"test/srt/test_example.py::TestClass::test_method (call)"
|
||||
)
|
||||
|
||||
dump_metric("pytest_metric", 123)
|
||||
|
||||
pid = os.getpid()
|
||||
jsonl_path = f"{base_path}.{pid}.jsonl"
|
||||
with open(jsonl_path, encoding="utf-8") as f:
|
||||
lines = f.readlines()
|
||||
|
||||
record = json.loads(lines[0])
|
||||
# Only assert test_case parsing, not filename
|
||||
self.assertEqual(record["test_case"], "TestClass.test_method")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1952,3 +1952,165 @@ def intel_amx_benchmark(extra_args=None, min_throughput=None):
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
def dump_metric(metric_name: str, value: Any, labels: Optional[dict] = None):
|
||||
"""
|
||||
Output test metric to JSONL and stdout for CI performance tracking.
|
||||
|
||||
Schema (v1):
|
||||
- Required: filename, test_case, metric_name, value
|
||||
- Optional fields supported: ts, labels
|
||||
- ts is emitted by default for convenience
|
||||
- labels preferred as dict; if not JSON-serializable, stored as string
|
||||
|
||||
Value types (v1 contract):
|
||||
- Supported: int, float, str
|
||||
- Input may be bool (will be coerced to int: True=1, False=0)
|
||||
- Others: best-effort conversion to float, fallback to str
|
||||
|
||||
Output channels:
|
||||
- JSONL: ${SGLANG_TEST_METRICS_OUTPUT}.${pid}.jsonl (if env var set)
|
||||
- stdout: [METRIC] metric_name=value [labels=...]
|
||||
|
||||
This function never fails tests - all errors are silently caught.
|
||||
|
||||
Args:
|
||||
metric_name: Metric name (e.g., "gsm8k_accuracy", "cache_hit_rate")
|
||||
value: Metric value
|
||||
labels: Optional label dict (e.g., {"backend": "fa3"})
|
||||
"""
|
||||
try:
|
||||
# 1. Capture test context
|
||||
filename, test_case = _get_test_context()
|
||||
|
||||
# 2. Convert value to int/float/str
|
||||
# First unwrap numpy/torch scalars
|
||||
if hasattr(value, "item"):
|
||||
value = value.item()
|
||||
|
||||
# Now convert, ensuring no bool in final result
|
||||
if isinstance(value, bool):
|
||||
converted_value = int(value) # True->1, False->0
|
||||
elif isinstance(value, (int, float, str)):
|
||||
converted_value = value
|
||||
else:
|
||||
try:
|
||||
converted_value = float(value)
|
||||
except (ValueError, TypeError):
|
||||
converted_value = str(value)
|
||||
|
||||
# 3. Build record
|
||||
record = {
|
||||
"filename": filename,
|
||||
"test_case": test_case,
|
||||
"metric_name": metric_name,
|
||||
"value": converted_value,
|
||||
"ts": time.time(),
|
||||
}
|
||||
|
||||
# 4. Handle labels (best-effort JSON serialization)
|
||||
labels_for_output = None
|
||||
if labels:
|
||||
try:
|
||||
json.dumps(labels, ensure_ascii=False) # Test serializability
|
||||
record["labels"] = labels
|
||||
labels_for_output = labels
|
||||
except (TypeError, ValueError):
|
||||
# If not serializable, stringify
|
||||
stringified = str(labels)
|
||||
record["labels"] = stringified
|
||||
labels_for_output = stringified
|
||||
|
||||
# 5. Write JSONL
|
||||
base_path = os.getenv("SGLANG_TEST_METRICS_OUTPUT")
|
||||
if base_path:
|
||||
try:
|
||||
jsonl_path = f"{base_path}.{os.getpid()}.jsonl"
|
||||
with open(jsonl_path, "a", encoding="utf-8") as f:
|
||||
f.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||||
except Exception as e:
|
||||
logging.warning(
|
||||
f"sglang.test.dump_metric: failed to write to {jsonl_path}: {e}"
|
||||
)
|
||||
|
||||
# 6. Output to stdout (use same labels as record)
|
||||
if labels_for_output:
|
||||
if isinstance(labels_for_output, str):
|
||||
labels_str = f" labels='{labels_for_output}'"
|
||||
else:
|
||||
labels_str = (
|
||||
f" labels={json.dumps(labels_for_output, ensure_ascii=False)}"
|
||||
)
|
||||
else:
|
||||
labels_str = ""
|
||||
print(f"[METRIC] {metric_name}={converted_value}{labels_str}")
|
||||
|
||||
except Exception as e:
|
||||
# Silent failure - never break tests
|
||||
logging.warning(
|
||||
f"sglang.test.dump_metric: failed to dump metric '{metric_name}': {e}",
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
|
||||
def _get_test_context() -> tuple[str, str]:
|
||||
"""
|
||||
Get current test's filename and test_case.
|
||||
|
||||
Tries PYTEST_CURRENT_TEST first, falls back to inspect.stack().
|
||||
"""
|
||||
# 1. Try parsing PYTEST_CURRENT_TEST
|
||||
pytest_current = os.getenv("PYTEST_CURRENT_TEST")
|
||||
if pytest_current:
|
||||
# Format: "path/to/test.py::TestClass::test_method (call)"
|
||||
parts = pytest_current.split(" ")[0].split("::", 1)
|
||||
if len(parts) == 2:
|
||||
filename = _repo_relative_path(parts[0])
|
||||
test_case = parts[1].replace("::", ".")
|
||||
return filename, test_case
|
||||
|
||||
# 2. Fallback to inspect
|
||||
import inspect
|
||||
|
||||
frame = inspect.currentframe()
|
||||
# Assumes direct callsite: frame -> _get_test_context -> dump_metric -> caller
|
||||
# If dump_metric gets wrapped, may need to scan upward
|
||||
if frame and frame.f_back and frame.f_back.f_back:
|
||||
caller = frame.f_back.f_back
|
||||
filename = _repo_relative_path(caller.f_code.co_filename)
|
||||
|
||||
# Try to get class name from self
|
||||
test_self = caller.f_locals.get("self")
|
||||
if test_self and hasattr(test_self, "__class__"):
|
||||
test_case = f"{test_self.__class__.__name__}.{caller.f_code.co_name}"
|
||||
else:
|
||||
test_case = caller.f_code.co_name
|
||||
|
||||
return filename, test_case
|
||||
|
||||
return "unknown.py", "unknown_test"
|
||||
|
||||
|
||||
def _repo_relative_path(filepath: str) -> str:
|
||||
"""Convert absolute path to repo-relative, preferring GITHUB_WORKSPACE"""
|
||||
# Path is imported at module top (line 20)
|
||||
try:
|
||||
abs_path = Path(filepath).resolve()
|
||||
|
||||
# Try GITHUB_WORKSPACE first
|
||||
workspace = os.getenv("GITHUB_WORKSPACE")
|
||||
if workspace:
|
||||
try:
|
||||
return str(abs_path.relative_to(Path(workspace).resolve()))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
# Fallback to cwd
|
||||
try:
|
||||
return str(abs_path.relative_to(Path.cwd()))
|
||||
except ValueError:
|
||||
return abs_path.name
|
||||
|
||||
except Exception:
|
||||
return Path(filepath).name
|
||||
|
||||
Reference in New Issue
Block a user