Migrate metrics tests to test/registered/metrics/ (#16466)
This commit is contained in:
Alison Shao
@@ -0,0 +1,178 @@
|
||||
import unittest
|
||||
from typing import Dict, List
|
||||
|
||||
import requests
|
||||
|
||||
from sglang.test.ci.ci_register import register_amd_ci, register_cuda_ci
|
||||
|
||||
register_cuda_ci(est_time=32, suite="stage-b-test-small-1-gpu")
|
||||
register_amd_ci(est_time=32, suite="stage-b-test-small-1-gpu")
|
||||
from prometheus_client.parser import text_string_to_metric_families
|
||||
from prometheus_client.samples import Sample
|
||||
|
||||
from sglang.srt.environ import envs
|
||||
from sglang.srt.utils import kill_process_tree
|
||||
from sglang.test.test_utils import (
|
||||
DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
DEFAULT_URL_FOR_TEST,
|
||||
CustomTestCase,
|
||||
is_in_ci,
|
||||
popen_launch_server,
|
||||
)
|
||||
|
||||
_MODEL_NAME = "Qwen/Qwen3-0.6B"
|
||||
|
||||
|
||||
class TestEnableMetrics(CustomTestCase):
|
||||
def test_metrics_1gpu(self):
|
||||
"""Test that metrics endpoint returns data when enabled"""
|
||||
self._execute_core(
|
||||
other_args=[],
|
||||
verify_metrics_extra=None,
|
||||
)
|
||||
|
||||
def test_metrics_2gpu(self):
|
||||
# TODO enable when we have 2-gpu runner in nightly CI
|
||||
if is_in_ci():
|
||||
print("Skip test_metrics_2gpu since in 1-gpu CI")
|
||||
return
|
||||
|
||||
def _verify_metrics_extra(metrics):
|
||||
metrics_to_check = [
|
||||
(
|
||||
"sglang:dp_cooperation_realtime_tokens_total",
|
||||
{"mode": "prefill_compute"},
|
||||
),
|
||||
("sglang:dp_cooperation_realtime_tokens_total", {"mode": "decode"}),
|
||||
(
|
||||
"sglang:dp_cooperation_gpu_execution_seconds_total",
|
||||
{"category": "forward_prefill"},
|
||||
),
|
||||
(
|
||||
"sglang:dp_cooperation_gpu_execution_seconds_total",
|
||||
{"category": "forward_decode"},
|
||||
),
|
||||
]
|
||||
_check_metrics_positive(self, metrics, metrics_to_check)
|
||||
|
||||
num_prefill_ranks_values = {
|
||||
s.labels["num_prefill_ranks"]
|
||||
for s in metrics["sglang:dp_cooperation_realtime_tokens_total"]
|
||||
}
|
||||
self.assertIn("0", num_prefill_ranks_values)
|
||||
self.assertIn("1", num_prefill_ranks_values)
|
||||
|
||||
self._execute_core(
|
||||
other_args=["--tp", "2", "--dp", "2", "--enable-dp-attention"],
|
||||
verify_metrics_extra=_verify_metrics_extra,
|
||||
)
|
||||
|
||||
def _execute_core(self, other_args, verify_metrics_extra):
|
||||
with (
|
||||
envs.SGLANG_ENABLE_METRICS_DP_ATTENTION.override(True),
|
||||
envs.SGLANG_ENABLE_METRICS_DEVICE_TIMER.override(True),
|
||||
envs.SGLANG_TEST_RETRACT.override(True),
|
||||
):
|
||||
process = popen_launch_server(
|
||||
_MODEL_NAME,
|
||||
DEFAULT_URL_FOR_TEST,
|
||||
timeout=DEFAULT_TIMEOUT_FOR_SERVER_LAUNCH,
|
||||
other_args=["--enable-metrics", "--cuda-graph-max-bs", 2, *other_args],
|
||||
)
|
||||
|
||||
try:
|
||||
# Make some requests to generate some metrics
|
||||
response = requests.get(f"{DEFAULT_URL_FOR_TEST}/health_generate")
|
||||
self.assertEqual(response.status_code, 200)
|
||||
|
||||
response = requests.post(
|
||||
f"{DEFAULT_URL_FOR_TEST}/generate",
|
||||
json={
|
||||
"text": ["The capital of France is"] * 20,
|
||||
"sampling_params": {
|
||||
"temperature": 0,
|
||||
"max_new_tokens": 50,
|
||||
},
|
||||
"stream": True,
|
||||
"ignore_eos": True,
|
||||
},
|
||||
stream=True,
|
||||
)
|
||||
for _ in response.iter_lines(decode_unicode=False):
|
||||
pass
|
||||
|
||||
# Get metrics
|
||||
metrics_response = requests.get(f"{DEFAULT_URL_FOR_TEST}/metrics")
|
||||
self.assertEqual(metrics_response.status_code, 200)
|
||||
metrics_text = metrics_response.text
|
||||
|
||||
print(f"metrics_text=\n{metrics_text}")
|
||||
|
||||
metrics = _parse_prometheus_metrics(metrics_text)
|
||||
self._verify_metrics_common(metrics_text, metrics)
|
||||
if verify_metrics_extra is not None:
|
||||
verify_metrics_extra(metrics)
|
||||
finally:
|
||||
kill_process_tree(process.pid)
|
||||
|
||||
def _verify_metrics_common(self, metrics_text, metrics):
|
||||
essential_metrics = [
|
||||
"sglang:num_running_reqs",
|
||||
"sglang:num_used_tokens",
|
||||
"sglang:token_usage",
|
||||
"sglang:gen_throughput",
|
||||
"sglang:num_queue_reqs",
|
||||
"sglang:num_grammar_queue_reqs",
|
||||
"sglang:cache_hit_rate",
|
||||
"sglang:spec_accept_length",
|
||||
"sglang:prompt_tokens_total",
|
||||
"sglang:generation_tokens_total",
|
||||
"sglang:cached_tokens_total",
|
||||
"sglang:num_requests_total",
|
||||
"sglang:time_to_first_token_seconds",
|
||||
"sglang:inter_token_latency_seconds",
|
||||
"sglang:e2e_request_latency_seconds",
|
||||
"sglang:http_requests_active",
|
||||
]
|
||||
for metric in essential_metrics:
|
||||
self.assertIn(metric, metrics_text, f"Missing metric: {metric}")
|
||||
|
||||
self.assertIn(f'model_name="{_MODEL_NAME}"', metrics_text)
|
||||
self.assertIn("_sum{", metrics_text)
|
||||
self.assertIn("_count{", metrics_text)
|
||||
self.assertIn("_bucket{", metrics_text)
|
||||
|
||||
metrics_to_check = [
|
||||
("sglang:realtime_tokens_total", {"mode": "prefill_compute"}),
|
||||
("sglang:realtime_tokens_total", {"mode": "decode"}),
|
||||
("sglang:gpu_execution_seconds_total", {"category": "forward_extend"}),
|
||||
("sglang:gpu_execution_seconds_total", {"category": "forward_decode"}),
|
||||
]
|
||||
_check_metrics_positive(self, metrics, metrics_to_check)
|
||||
|
||||
|
||||
def _parse_prometheus_metrics(metrics_text: str) -> Dict[str, List[Sample]]:
|
||||
result = {}
|
||||
for family in text_string_to_metric_families(metrics_text):
|
||||
for sample in family.samples:
|
||||
if sample.name not in result:
|
||||
result[sample.name] = []
|
||||
result[sample.name].append(sample)
|
||||
return result
|
||||
|
||||
|
||||
def _get_sample_value_by_labels(samples: List[Sample], labels: Dict[str, str]) -> float:
|
||||
for sample in samples:
|
||||
if all(sample.labels.get(k) == v for k, v in labels.items()):
|
||||
return sample.value
|
||||
raise KeyError(f"No sample found with labels {labels}")
|
||||
|
||||
|
||||
def _check_metrics_positive(test_case, metrics, metrics_to_check):
|
||||
for metric_name, labels in metrics_to_check:
|
||||
value = _get_sample_value_by_labels(metrics[metric_name], labels)
|
||||
test_case.assertGreater(value, 0, f"{metric_name} {labels}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,140 @@
|
||||
import unittest
|
||||
|
||||
from sglang.srt.metrics.utils import generate_buckets, two_sides_exponential_buckets
|
||||
from sglang.test.ci.ci_register import register_cpu_ci
|
||||
|
||||
register_cpu_ci(est_time=1, suite="stage-a-cpu-only")
|
||||
|
||||
|
||||
class TestMetricsUtils(unittest.TestCase):
|
||||
"""Test cases for metrics utility functions."""
|
||||
|
||||
def test_two_sides_exponential_buckets_basic(self):
|
||||
"""Test basic functionality of two_sides_exponential_buckets."""
|
||||
# Test with simple parameters
|
||||
count = 5
|
||||
buckets = two_sides_exponential_buckets(middle=10.0, base=2.0, count=count)
|
||||
|
||||
# Should contain the middle value
|
||||
self.assertIn(10.0, buckets)
|
||||
|
||||
# Should be sorted
|
||||
self.assertEqual(buckets, sorted(buckets))
|
||||
|
||||
# Should have unique values (no duplicates)
|
||||
self.assertEqual(len(buckets), len(set(buckets)))
|
||||
|
||||
# Should have reasonable number of buckets (not exactly count due to ceiling and deduplication)
|
||||
self.assertGreaterEqual(len(buckets), 3)
|
||||
self.assertLessEqual(len(buckets), count + 2)
|
||||
|
||||
def test_two_sides_exponential_buckets_specific_values(self):
|
||||
"""Test specific values for two_sides_exponential_buckets."""
|
||||
buckets = two_sides_exponential_buckets(middle=100.0, base=2.0, count=4)
|
||||
expected_values = [96.0, 98.0, 100.0, 102.0, 104.0]
|
||||
self.assertEqual(buckets, expected_values)
|
||||
|
||||
def test_two_sides_exponential_buckets_negative_values(self):
|
||||
"""Test two_sides_exponential_buckets with values that could go negative."""
|
||||
buckets = two_sides_exponential_buckets(middle=5.0, base=3.0, count=4)
|
||||
|
||||
# Should not contain negative values (max(0, middle - distance))
|
||||
for bucket in buckets:
|
||||
self.assertGreaterEqual(bucket, 0.0)
|
||||
|
||||
# Should contain the middle value
|
||||
self.assertIn(5.0, buckets)
|
||||
|
||||
def test_two_sides_exponential_buckets_edge_cases(self):
|
||||
"""Test edge cases for two_sides_exponential_buckets."""
|
||||
# Count = 1
|
||||
buckets = two_sides_exponential_buckets(middle=10.0, base=2.0, count=1)
|
||||
self.assertIn(10.0, buckets)
|
||||
|
||||
# Very small middle value
|
||||
buckets = two_sides_exponential_buckets(middle=0.1, base=2.0, count=2)
|
||||
self.assertIn(0.1, buckets)
|
||||
for bucket in buckets:
|
||||
self.assertGreaterEqual(bucket, 0.0)
|
||||
|
||||
def test_generate_buckets_default(self):
|
||||
"""Test generate_buckets with default rule."""
|
||||
default_buckets = [1.0, 5.0, 10.0, 50.0, 100.0]
|
||||
|
||||
# Test with "default" rule
|
||||
result = generate_buckets(["default"], default_buckets)
|
||||
self.assertEqual(result, default_buckets)
|
||||
|
||||
# Test with None (should default to "default")
|
||||
result = generate_buckets(None, default_buckets)
|
||||
self.assertEqual(result, default_buckets)
|
||||
|
||||
# Test with empty (should default to "default")
|
||||
result = generate_buckets(None, default_buckets)
|
||||
self.assertEqual(result, default_buckets)
|
||||
|
||||
def test_generate_buckets_tse(self):
|
||||
"""Test generate_buckets with tse (two sides exponential) rule."""
|
||||
default_buckets = [1.0, 5.0, 10.0]
|
||||
|
||||
# Test with "tse" rule
|
||||
result = generate_buckets(["tse", "10", "2.0", "4"], default_buckets)
|
||||
|
||||
# Should return the same as calling two_sides_exponential_buckets directly
|
||||
expected = two_sides_exponential_buckets(10.0, 2.0, 4)
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
def test_generate_buckets_custom(self):
|
||||
"""Test generate_buckets with custom rule."""
|
||||
default_buckets = [1.0, 5.0, 10.0]
|
||||
|
||||
# Test with "custom" rule
|
||||
result = generate_buckets(
|
||||
["custom", "1.5", "3.2", "7.8", "15.6"], default_buckets
|
||||
)
|
||||
expected = [1.5, 3.2, 7.8, 15.6]
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
def test_generate_buckets_custom_with_integers(self):
|
||||
"""Test generate_buckets with custom rule using integer strings."""
|
||||
default_buckets = [1.0, 5.0, 10.0]
|
||||
|
||||
# Test with integer strings
|
||||
result = generate_buckets(["custom", "1", "5", "10", "50"], default_buckets)
|
||||
expected = [1.0, 5.0, 10.0, 50.0]
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
def test_generate_buckets_preserves_order_and_type(self):
|
||||
"""Test that generate_buckets preserves order and returns floats."""
|
||||
default_buckets = [1, 5, 10, 50, 100] # integers
|
||||
|
||||
# Test default rule
|
||||
result = generate_buckets(["default"], default_buckets)
|
||||
self.assertEqual(result, default_buckets)
|
||||
self.assertIsInstance(result, list)
|
||||
|
||||
# Test custom rule with proper float conversion
|
||||
result = generate_buckets(
|
||||
["custom", "100", "50", "10", "5", "1"], default_buckets
|
||||
)
|
||||
expected = [1.0, 5.0, 10.0, 50.0, 100.0]
|
||||
self.assertEqual(result, expected)
|
||||
|
||||
# All values should be floats
|
||||
for value in result:
|
||||
self.assertIsInstance(value, float)
|
||||
|
||||
def test_integration_tse_through_generate_buckets(self):
|
||||
"""Test integration of TSE buckets through generate_buckets function."""
|
||||
default_buckets = [1.0, 10.0, 100.0]
|
||||
|
||||
# Generate buckets using both methods
|
||||
direct_result = two_sides_exponential_buckets(50.0, 1.5, 6)
|
||||
indirect_result = generate_buckets(["tse", "50.0", "1.5", "6"], default_buckets)
|
||||
|
||||
# Results should be identical
|
||||
self.assertEqual(direct_result, indirect_result)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user