[Metrics] Add admission rejection metrics (#55812)
Signed-off-by: Thuongvu Ho <thuongvuho@gmail.com> Co-authored-by: Nicolò Lucchesi <nicolo.lucchesi@mistral.ai>
This commit is contained in:
co-authored by
Nicolò Lucchesi
parent
5bed289707
commit
99b44c7a42
@@ -30,6 +30,7 @@ In v1, an extensive set of metrics are exposed via a Prometheus-compatible `/met
|
||||
- `vllm:prompt_tokens_total` (Counter) - Total number of prompt tokens processed.
|
||||
- `vllm:generation_tokens_total` (Counter) - Total number of generated tokens.
|
||||
- `vllm:request_success_total` (Counter) - Number of finished requests (by finish reason).
|
||||
- `vllm:admission_rejections_total` (Counter) - Admission attempts rejected by a queue limit, by reason.
|
||||
- `vllm:request_prompt_tokens` (Histogram) - Histogram of input prompt token counts.
|
||||
- `vllm:request_generation_tokens` (Histogram) - Histogram of generation token counts.
|
||||
- `vllm:time_to_first_token_seconds` (Histogram) - Time to first token (TTFT).
|
||||
|
||||
@@ -188,6 +188,7 @@ EXPECTED_METRICS_V1 = [
|
||||
"vllm:prefix_cache_queries",
|
||||
"vllm:prefix_cache_hits",
|
||||
"vllm:num_preemptions_total",
|
||||
"vllm:admission_rejections_total",
|
||||
"vllm:prompt_tokens_total",
|
||||
"vllm:generation_tokens_total",
|
||||
"vllm:iteration_tokens_total",
|
||||
|
||||
@@ -6,6 +6,7 @@
|
||||
These tests cover:
|
||||
- OutputProcessor.get_num_queued_tokens() token counting
|
||||
- AsyncLLM.check_admission() admission control logic
|
||||
- Admission rejection Prometheus samples
|
||||
- Exception classes (GracefulHTTPError, QueueOverflowError, MaxQueuedTokensError)
|
||||
- create_error_response() mapping GracefulHTTPError to HTTP 503
|
||||
- SchedulerConfig field defaults and validation
|
||||
@@ -14,6 +15,7 @@ These tests cover:
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import multiprocessing
|
||||
from collections.abc import Awaitable, Callable
|
||||
from http import HTTPStatus
|
||||
@@ -21,6 +23,7 @@ from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from prometheus_client import REGISTRY
|
||||
from pydantic import ValidationError
|
||||
|
||||
from vllm.config.scheduler import SchedulerConfig
|
||||
@@ -40,6 +43,10 @@ from vllm.v1.engine import EngineCoreRequest
|
||||
from vllm.v1.engine.admission_control import SharedAdmissionStats
|
||||
from vllm.v1.engine.async_llm import AsyncLLM
|
||||
from vllm.v1.engine.output_processor import OutputProcessor
|
||||
from vllm.v1.metrics.loggers import StatLoggerManager
|
||||
|
||||
_ADMISSION_METRIC_MODEL = "admission-metric-test"
|
||||
_ADMISSION_LOGGER = "vllm.v1.engine.async_llm"
|
||||
|
||||
pytestmark = pytest.mark.cpu_test
|
||||
|
||||
@@ -70,10 +77,43 @@ def _make_async_llm(
|
||||
llm.output_processor = MagicMock()
|
||||
llm.output_processor.get_num_unfinished_requests.return_value = num_unfinished
|
||||
llm.output_processor.get_num_queued_tokens.return_value = num_queued_tokens
|
||||
llm.logger_manager = None
|
||||
llm.admission_stats = None
|
||||
return llm
|
||||
|
||||
|
||||
def _attach_prometheus_logger(
|
||||
llm: AsyncLLM, engine_indexes: list[int] | None = None
|
||||
) -> None:
|
||||
"""Attach a real Prometheus logger manager to a lightweight AsyncLLM."""
|
||||
vllm_config = MagicMock()
|
||||
vllm_config.observability_config.show_hidden_metrics = False
|
||||
vllm_config.observability_config.kv_cache_metrics = False
|
||||
vllm_config.observability_config.custom_histogram_buckets = None
|
||||
vllm_config.model_config.served_model_name = _ADMISSION_METRIC_MODEL
|
||||
vllm_config.model_config.max_model_len = 2048
|
||||
vllm_config.model_config.is_diffusion = False
|
||||
vllm_config.speculative_config = None
|
||||
vllm_config.kv_transfer_config = None
|
||||
vllm_config.ec_transfer_config = None
|
||||
vllm_config.lora_config = None
|
||||
llm.logger_manager = StatLoggerManager(
|
||||
vllm_config=vllm_config,
|
||||
engine_idxs=engine_indexes,
|
||||
enable_default_loggers=False,
|
||||
)
|
||||
|
||||
|
||||
def _admission_rejection_count(reason: str) -> float:
|
||||
"""Read the Prometheus sample for one admission rejection reason."""
|
||||
value = REGISTRY.get_sample_value(
|
||||
"vllm:admission_rejections_total",
|
||||
{"model_name": _ADMISSION_METRIC_MODEL, "reason": reason},
|
||||
)
|
||||
assert value is not None
|
||||
return value
|
||||
|
||||
|
||||
def _make_output_processor(**request_states) -> OutputProcessor:
|
||||
op = OutputProcessor.__new__(OutputProcessor)
|
||||
op.request_states = request_states
|
||||
@@ -157,11 +197,13 @@ def test_shared_request_snapshot_enforces_global_limit():
|
||||
server_0.set_num_requests(2)
|
||||
llm = _make_async_llm(max_num_queued_reqs=2)
|
||||
llm.admission_stats = server_1
|
||||
_attach_prometheus_logger(llm)
|
||||
|
||||
with pytest.raises(QueueOverflowError):
|
||||
llm.check_admission()
|
||||
|
||||
assert server_0.get_num_requests() == 2
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -247,19 +289,35 @@ def test_admission_no_limits_allows_everything():
|
||||
|
||||
def test_admission_reqs_allows_when_under_limit():
|
||||
llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=5)
|
||||
_attach_prometheus_logger(llm)
|
||||
llm.check_admission()
|
||||
llm.check_admission()
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 0
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 0
|
||||
|
||||
|
||||
def test_admission_reqs_rejects_at_limit():
|
||||
def test_admission_reqs_rejects_at_limit(caplog):
|
||||
llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=10)
|
||||
with pytest.raises(QueueOverflowError):
|
||||
_attach_prometheus_logger(llm, engine_indexes=[0, 1, 2])
|
||||
with (
|
||||
caplog.at_level(logging.INFO, logger=_ADMISSION_LOGGER),
|
||||
pytest.raises(QueueOverflowError),
|
||||
):
|
||||
llm.check_admission()
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 1
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 0
|
||||
assert not any(
|
||||
"Request queue full" in record.getMessage() for record in caplog.records
|
||||
)
|
||||
|
||||
|
||||
def test_admission_reqs_rejects_with_n():
|
||||
llm = _make_async_llm(max_num_queued_reqs=10, num_unfinished=8)
|
||||
_attach_prometheus_logger(llm)
|
||||
with pytest.raises(QueueOverflowError):
|
||||
llm.check_admission(3)
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 1
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 0
|
||||
|
||||
|
||||
def test_admission_reqs_allows_n_at_boundary():
|
||||
@@ -283,8 +341,11 @@ def test_admission_tokens_allows_when_under_limit():
|
||||
|
||||
def test_admission_tokens_rejects_at_limit():
|
||||
llm = _make_async_llm(max_num_queued_tokens=1000, num_queued_tokens=1000)
|
||||
_attach_prometheus_logger(llm)
|
||||
with pytest.raises(MaxQueuedTokensError):
|
||||
llm.check_admission()
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 0
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 1
|
||||
|
||||
|
||||
def test_admission_tokens_rejects_over_limit():
|
||||
@@ -320,8 +381,11 @@ def test_admission_req_limit_checked_before_token_limit():
|
||||
num_unfinished=10,
|
||||
num_queued_tokens=1000,
|
||||
)
|
||||
_attach_prometheus_logger(llm)
|
||||
with pytest.raises(QueueOverflowError):
|
||||
llm.check_admission()
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 1
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 0
|
||||
|
||||
|
||||
# -- n derived from params at the add_request call site ---------------------
|
||||
@@ -381,6 +445,17 @@ async def test_concurrent_single_request_admission_respects_limit():
|
||||
assert len(request_states) == 1
|
||||
|
||||
|
||||
def test_admission_preflight_pass_then_reject_records_once():
|
||||
llm = _make_async_llm(max_num_queued_reqs=1, num_unfinished=0)
|
||||
_attach_prometheus_logger(llm)
|
||||
llm.check_admission()
|
||||
llm.output_processor.get_num_unfinished_requests.return_value = 1
|
||||
with pytest.raises(QueueOverflowError):
|
||||
llm.check_admission()
|
||||
assert _admission_rejection_count("max_num_queued_reqs") == 1
|
||||
assert _admission_rejection_count("max_num_queued_tokens") == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("first_n", [1, 3])
|
||||
async def test_parallel_admission_is_all_or_nothing(first_n: int):
|
||||
|
||||
@@ -348,7 +348,7 @@ class AsyncLLM(EngineClient):
|
||||
else self.get_num_unfinished_requests()
|
||||
)
|
||||
if current_requests + n > max_num_reqs:
|
||||
logger.info(
|
||||
logger.debug(
|
||||
"Request queue full - rejecting request %s "
|
||||
"(current=%d, n=%d, max=%d).",
|
||||
request_id,
|
||||
@@ -356,21 +356,28 @@ class AsyncLLM(EngineClient):
|
||||
n,
|
||||
max_num_reqs,
|
||||
)
|
||||
self._record_admission_rejection("max_num_queued_reqs")
|
||||
raise QueueOverflowError()
|
||||
|
||||
max_queued_tokens = self.scheduler_config.max_num_queued_tokens
|
||||
if max_queued_tokens is not None:
|
||||
current_tokens = self.get_num_queued_tokens()
|
||||
if current_tokens >= max_queued_tokens:
|
||||
logger.info(
|
||||
logger.debug(
|
||||
"Max queued tokens reached - rejecting request %s "
|
||||
"(current_tokens=%d, max=%d).",
|
||||
request_id,
|
||||
current_tokens,
|
||||
max_queued_tokens,
|
||||
)
|
||||
self._record_admission_rejection("max_num_queued_tokens")
|
||||
raise MaxQueuedTokensError()
|
||||
|
||||
def _record_admission_rejection(self, reason: str) -> None:
|
||||
"""Increment the frontend admission rejection counter if logging is on."""
|
||||
if self.logger_manager is not None:
|
||||
self.logger_manager.record_admission_rejection(reason)
|
||||
|
||||
async def get_supported_tasks(self) -> tuple[SupportedTask, ...]:
|
||||
if not hasattr(self, "_supported_tasks"):
|
||||
# Cache the result
|
||||
|
||||
@@ -43,6 +43,9 @@ logger = init_logger(__name__)
|
||||
WAITING_REASON_CAPACITY = "capacity"
|
||||
WAITING_REASON_DEFERRED = "deferred"
|
||||
|
||||
# User-facing reason labels for frontend admission rejections.
|
||||
ADMISSION_REASONS = ("max_num_queued_reqs", "max_num_queued_tokens")
|
||||
|
||||
PerEngineStatLoggerFactory = Callable[[VllmConfig, int], "StatLoggerBase"]
|
||||
AggregateStatLoggerFactory = type["AggregateStatLoggerBase"]
|
||||
StatLoggerFactory = AggregateStatLoggerFactory | PerEngineStatLoggerFactory
|
||||
@@ -102,6 +105,9 @@ class AggregateStatLoggerBase(StatLoggerBase):
|
||||
@abstractmethod
|
||||
def __init__(self, vllm_config: VllmConfig, engine_indexes: list[int]): ...
|
||||
|
||||
def record_admission_rejection(self, reason: str) -> None: # noqa
|
||||
"""Record a frontend admission rejection. Optional for custom loggers."""
|
||||
|
||||
|
||||
class LoggingStatLogger(StatLoggerBase):
|
||||
def __init__(self, vllm_config: VllmConfig, engine_index: int = 0):
|
||||
@@ -816,6 +822,23 @@ class PrometheusStatLogger(AggregateStatLoggerBase):
|
||||
"request_num_preemptions", overrides=custom_buckets
|
||||
)
|
||||
|
||||
#
|
||||
# Frontend admission
|
||||
#
|
||||
counter_admission_rejections = self._counter_cls(
|
||||
name="vllm:admission_rejections",
|
||||
documentation=(
|
||||
"Number of admission attempts rejected by a queue limit, by reason."
|
||||
),
|
||||
labelnames=["model_name", "reason"],
|
||||
)
|
||||
self.admission_rejections = {
|
||||
reason: counter_admission_rejections.labels(
|
||||
model_name=model_name, reason=reason
|
||||
)
|
||||
for reason in ADMISSION_REASONS
|
||||
}
|
||||
|
||||
#
|
||||
# Histograms of counts
|
||||
#
|
||||
@@ -1276,6 +1299,9 @@ class PrometheusStatLogger(AggregateStatLoggerBase):
|
||||
finished_request.max_tokens_param
|
||||
)
|
||||
|
||||
def record_admission_rejection(self, reason: str) -> None:
|
||||
self.admission_rejections[reason].inc()
|
||||
|
||||
def record_sleep_state(self, sleep: int = 0, level: int = 0):
|
||||
awake = 1
|
||||
discard_all = 0
|
||||
@@ -1383,6 +1409,11 @@ class StatLoggerManager:
|
||||
for logger in self.stat_loggers:
|
||||
logger.record_sleep_state(sleep, level)
|
||||
|
||||
def record_admission_rejection(self, reason: str) -> None:
|
||||
"""Record one frontend admission rejection on aggregate loggers."""
|
||||
for stat_logger in self.stat_loggers:
|
||||
stat_logger.record_admission_rejection(reason)
|
||||
|
||||
def log(self):
|
||||
for logger in self.stat_loggers:
|
||||
logger.log()
|
||||
|
||||
Reference in New Issue
Block a user