diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index 8061835f205..7e592b3bb0f 100644 --- a/litellm/integrations/prometheus.py +++ b/litellm/integrations/prometheus.py @@ -717,6 +717,29 @@ class PrometheusLogger(CustomLogger): labelnames=[], ) + ######################################## + # Per-pod request pressure + ######################################## + self.litellm_requests_shed_total = self._counter_factory( + name="litellm_requests_shed_total", + documentation=( + "Responses where the proxy declined to serve rather than failed to, by status. " + "429 is a rate or concurrency limit, 503 is the database being unavailable" + ), + labelnames=("status",), + ) + + self.litellm_global_max_parallel_requests_limit = self._gauge_factory( + "litellm_global_max_parallel_requests_limit", + ( + "Concurrency ceiling actually applied on this worker. +Inf means nothing bounds " + "concurrency, either because global_max_parallel_requests is unset or because the " + "active limiter does not enforce it" + ), + labelnames=(), + multiprocess_mode="livemax", + ) + ######################################## # Scheduled background jobs ######################################## @@ -3204,6 +3227,14 @@ class PrometheusLogger(CustomLogger): except Exception as e: verbose_logger.warning("Error recording check batch cost metrics: %s", e) + def record_request_shed(self, status: int) -> None: + """Count one response where the proxy declined to serve.""" + self.litellm_requests_shed_total.labels(status=str(status)).inc() + + def set_global_max_parallel_requests_limit(self, limit: float) -> None: + """Publish the per-pod concurrency ceiling actually in force.""" + self.litellm_global_max_parallel_requests_limit.set(limit) + def record_scheduled_job_run(self, run: JobRun) -> None: """Publish one completed run of a scheduled background job.""" self.litellm_scheduled_job_runs_total.labels(job_name=run.job_name, result=run.result.value).inc() diff --git a/litellm/proxy/common_utils/request_pressure_metrics.py b/litellm/proxy/common_utils/request_pressure_metrics.py new file mode 100644 index 00000000000..d7623d62c34 --- /dev/null +++ b/litellm/proxy/common_utils/request_pressure_metrics.py @@ -0,0 +1,58 @@ +"""Publishes the per-pod concurrency ceiling that is actually in force. + +``global_max_parallel_requests`` is only read by the v1 parallel-request +limiter, which is off unless ``LEGACY_MULTI_INSTANCE_RATE_LIMITING`` is set. The +default v3 limiter never looks at it, so publishing the configured number +regardless would hand operators a ceiling that nothing applies. + +A registered Prometheus gauge always exposes a value, so simply declining to set +it renders as ``0``, which reads as "no requests allowed". The gauge therefore +reports the effective ceiling in every state: the configured number when it is +enforced, and ``+Inf`` when nothing bounds concurrency. + +See LIT-5460 for the enforcement gap itself. +""" + +from __future__ import annotations + +from typing import Final + +from litellm._logging import verbose_proxy_logger + +UNBOUNDED: Final = float("inf") + + +def is_global_limit_enforced() -> bool: + """Whether the registered limiter reads ``global_max_parallel_requests``.""" + from litellm.proxy.hooks import PROXY_HOOKS + from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, + ) + + return PROXY_HOOKS.get("parallel_request_limiter") is _PROXY_MaxParallelRequestsHandler + + +def effective_global_limit(limit: int | None) -> float: + """The ceiling actually applied to this worker's concurrency.""" + if limit is None: + return UNBOUNDED + if not is_global_limit_enforced(): + verbose_proxy_logger.warning( + "global_max_parallel_requests=%s is set but the active rate limiter does not enforce it, so the " + "limit metric reports unbounded. Set LEGACY_MULTI_INSTANCE_RATE_LIMITING=true to enforce it", + limit, + ) + return UNBOUNDED + return float(limit) + + +def publish_global_max_parallel_requests(limit: int | None) -> None: + """Publish the effective ceiling, whatever it turns out to be.""" + try: + from litellm.integrations.prometheus import PrometheusLogger + + logger: Final = PrometheusLogger.get_instance() + if logger is not None: + logger.set_global_max_parallel_requests_limit(effective_global_limit(limit)) + except Exception as e: # noqa: BLE001 # telemetry must not block startup + verbose_proxy_logger.debug("global max parallel requests metric failed: %s", e) diff --git a/litellm/proxy/middleware/in_flight_requests_middleware.py b/litellm/proxy/middleware/in_flight_requests_middleware.py index 2430f2fb081..e414a85c946 100644 --- a/litellm/proxy/middleware/in_flight_requests_middleware.py +++ b/litellm/proxy/middleware/in_flight_requests_middleware.py @@ -1,14 +1,18 @@ """ -Tracks the number of HTTP requests currently in-flight on this uvicorn worker. +Tracks per-worker request pressure: how many HTTP requests are in flight, and +how many the proxy shed rather than served. -Used by /health/backlog to expose per-pod queue depth, and emitted as the -Prometheus gauge `litellm_in_flight_requests`. +Counting shed responses here rather than at each limiter means no rejection path +can be missed, and the count is per worker for the same reason the in-flight +gauge is. """ import os from typing import Any, Final -from starlette.types import ASGIApp, Receive, Scope, Send +from starlette.types import ASGIApp, Message, Receive, Scope, Send + +from litellm._logging import verbose_proxy_logger class InFlightRequestsMiddleware: @@ -43,7 +47,7 @@ class InFlightRequestsMiddleware: if gauge is not None: gauge.inc() try: - await self.app(scope, receive, send) + await self.app(scope, receive, send=_counting_send(send)) finally: InFlightRequestsMiddleware._in_flight -= 1 if gauge is not None: @@ -79,6 +83,33 @@ class InFlightRequestsMiddleware: return InFlightRequestsMiddleware._gauge +# Statuses that mean the proxy declined to serve rather than failed to. Kept to +# a fixed set so the metric label cannot grow. +_SHED_STATUSES: Final = frozenset({429, 503}) + + +def _record_shed_response(status: int) -> None: + if status not in _SHED_STATUSES: + return + try: + from litellm.integrations.prometheus import PrometheusLogger + + logger = PrometheusLogger.get_instance() + if logger is not None: + logger.record_request_shed(status) + except Exception as e: # noqa: BLE001 # counting a shed response must not break the response + verbose_proxy_logger.debug("request shed metric failed: %s", e) + + +def _counting_send(send: Send) -> Send: + async def wrapped(message: Message) -> None: + if message["type"] == "http.response.start": + _record_shed_response(int(message["status"])) + await send(message) + + return wrapped + + def get_in_flight_requests() -> int: """Module-level convenience wrapper used by the /health/backlog endpoint.""" return InFlightRequestsMiddleware.get_count() diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 233456032b0..5c9b9ef8fd2 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -347,6 +347,7 @@ from litellm.proxy.common_utils.periodic_reload_schedule import ( write_reload_interval, ) from litellm.proxy.common_utils.proxy_state import ProxyState +from litellm.proxy.common_utils.request_pressure_metrics import publish_global_max_parallel_requests from litellm.proxy.common_utils.reset_budget_job import ResetBudgetJob from litellm.proxy.common_utils.scheduled_job_metrics import ScheduledJobMetricsListener from litellm.proxy.common_utils.scheduled_job_stagger import ( @@ -9137,6 +9138,7 @@ class ProxyStartupEvent: ) # Registered before start so the first run of every job is observed. ScheduledJobMetricsListener().register(scheduler) + publish_global_max_parallel_requests(general_settings.get("global_max_parallel_requests")) # Start the scheduler immediately without processing backlogs scheduler.start(paused=False) diff --git a/litellm/types/integrations/prometheus.py b/litellm/types/integrations/prometheus.py index cfefb05822c..b4d1e4f751d 100644 --- a/litellm/types/integrations/prometheus.py +++ b/litellm/types/integrations/prometheus.py @@ -285,6 +285,9 @@ DEFINED_PROMETHEUS_METRICS = Literal[ # MCP tool call metrics "litellm_mcp_tool_calls_total", "litellm_mcp_tool_call_spend_metric", + # Per-pod request pressure + "litellm_requests_shed_total", + "litellm_global_max_parallel_requests_limit", # Scheduled background jobs "litellm_scheduled_job_runs_total", "litellm_scheduled_job_duration_seconds", @@ -794,8 +797,16 @@ class PrometheusMetricLabels: litellm_check_batch_cost_last_run_timestamp: list[str] = [] - # No pod label: pod identity is unbounded, and the lock result already - # identifies the owner. + # Per-pod request pressure. status is a fixed set of shed codes; the limit + # gauge is unlabelled because it describes this worker. + litellm_requests_shed_total: tuple[str, ...] = () + litellm_global_max_parallel_requests_limit: tuple[str, ...] = () + + # Scheduled background jobs. Labels are closed sets fixed at startup: job ids + # come from the scheduler registration, cronjob ids from the lock call sites, + # and results from an enum. No pod label: pod identity is unbounded, and the + # lock result already distinguishes the pod that owns a job from the ones + # that skipped it. litellm_scheduled_job_runs_total: tuple[str, ...] = () litellm_scheduled_job_duration_seconds: tuple[str, ...] = () litellm_scheduled_job_last_run_timestamp: tuple[str, ...] = () diff --git a/tests/test_litellm/proxy/common_utils/test_request_pressure_metrics.py b/tests/test_litellm/proxy/common_utils/test_request_pressure_metrics.py new file mode 100644 index 00000000000..7d2c3d663ac --- /dev/null +++ b/tests/test_litellm/proxy/common_utils/test_request_pressure_metrics.py @@ -0,0 +1,106 @@ +"""The concurrency-ceiling gauge must never claim a limit nothing enforces.""" + +import os +import sys +from unittest.mock import MagicMock, patch + +import pytest + +sys.path.insert(0, os.path.abspath("../../../..")) + +from litellm.proxy.common_utils.request_pressure_metrics import ( + UNBOUNDED, + effective_global_limit, + is_global_limit_enforced, + publish_global_max_parallel_requests, +) + + +def _with_limiter(handler): + from litellm.proxy import hooks + + return patch.dict(hooks.PROXY_HOOKS, {"parallel_request_limiter": handler}) + + +def test_the_v1_limiter_is_the_one_that_enforces_the_global_limit(): + from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, + ) + + with _with_limiter(_PROXY_MaxParallelRequestsHandler): + assert is_global_limit_enforced() is True + + +def test_the_default_v3_limiter_does_not_enforce_the_global_limit(): + """Verified live: with global_max_parallel_requests=3 and 20 concurrent + requests, the v3 limiter returned 20x 200 and zero rejections. Tracked as + LIT-5460.""" + from litellm.proxy.hooks.parallel_request_limiter_v3 import ( + _PROXY_MaxParallelRequestsHandler_v3, + ) + + with _with_limiter(_PROXY_MaxParallelRequestsHandler_v3): + assert is_global_limit_enforced() is False + + +def test_an_enforced_limit_is_reported_as_the_configured_number(): + from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, + ) + + logger = MagicMock() + with ( + _with_limiter(_PROXY_MaxParallelRequestsHandler), + patch("litellm.integrations.prometheus.PrometheusLogger.get_instance", return_value=logger), + ): + publish_global_max_parallel_requests(3) + + logger.set_global_max_parallel_requests_limit.assert_called_once_with(3.0) + + +@pytest.mark.parametrize("configured", [3, None]) +def test_an_unenforced_limit_is_reported_as_unbounded(configured): + """A registered gauge always exposes a value, so declining to set it renders + as 0, which reads as "no requests allowed". Reporting +Inf says what is + actually true: nothing bounds concurrency.""" + from litellm.proxy.hooks.parallel_request_limiter_v3 import ( + _PROXY_MaxParallelRequestsHandler_v3, + ) + + with _with_limiter(_PROXY_MaxParallelRequestsHandler_v3): + assert effective_global_limit(configured) == UNBOUNDED + + +def test_no_configured_limit_is_unbounded_even_when_enforcement_is_on(): + from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, + ) + + with _with_limiter(_PROXY_MaxParallelRequestsHandler): + assert effective_global_limit(None) == UNBOUNDED + + +def test_the_gauge_is_never_left_at_a_bare_zero(): + """0 would be indistinguishable from a real ceiling of zero.""" + from litellm.proxy.hooks.parallel_request_limiter_v3 import ( + _PROXY_MaxParallelRequestsHandler_v3, + ) + + with _with_limiter(_PROXY_MaxParallelRequestsHandler_v3): + assert effective_global_limit(3) != 0.0 + assert effective_global_limit(None) != 0.0 + + +def test_publishing_never_blocks_startup(): + from litellm.proxy.hooks.parallel_request_limiter import ( + _PROXY_MaxParallelRequestsHandler, + ) + + with ( + _with_limiter(_PROXY_MaxParallelRequestsHandler), + patch( + "litellm.integrations.prometheus.PrometheusLogger.get_instance", + side_effect=RuntimeError("metrics down"), + ), + ): + publish_global_max_parallel_requests(3) diff --git a/tests/test_litellm/proxy/middleware/test_in_flight_requests_middleware.py b/tests/test_litellm/proxy/middleware/test_in_flight_requests_middleware.py index d9fa62ed768..544ebb08251 100644 --- a/tests/test_litellm/proxy/middleware/test_in_flight_requests_middleware.py +++ b/tests/test_litellm/proxy/middleware/test_in_flight_requests_middleware.py @@ -95,3 +95,66 @@ def test_non_http_scopes_not_counted(): asyncio.run(mw({"type": "lifespan"}, None, None)) # type: ignore[arg-type] assert get_in_flight_requests() == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status, expected_calls", [(429, 1), (503, 1), (200, 0), (400, 0), (500, 0)]) +async def test_only_shed_responses_are_counted(status, expected_calls): + """A 500 is the proxy failing, not declining. Counting it here would blur the + signal an operator uses to decide between throttling and scaling out.""" + from unittest.mock import MagicMock, patch + + from litellm.proxy.middleware.in_flight_requests_middleware import ( + InFlightRequestsMiddleware, + ) + + sent = [] + + async def send(message): + sent.append(message) + + async def receive(): + return {"type": "http.request"} + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": status, "headers": []}) + await send({"type": "http.response.body", "body": b""}) + + logger = MagicMock() + with patch("litellm.integrations.prometheus.PrometheusLogger.get_instance", return_value=logger): + await InFlightRequestsMiddleware(app)({"type": "http"}, receive, send) + + assert logger.record_request_shed.call_count == expected_calls + if expected_calls: + logger.record_request_shed.assert_called_once_with(status) + assert [m["type"] for m in sent] == ["http.response.start", "http.response.body"], ( + "the wrapped send must still forward every message downstream" + ) + + +@pytest.mark.asyncio +async def test_a_broken_metric_never_breaks_the_response(): + from unittest.mock import patch + + from litellm.proxy.middleware.in_flight_requests_middleware import ( + InFlightRequestsMiddleware, + ) + + sent = [] + + async def send(message): + sent.append(message) + + async def receive(): + return {"type": "http.request"} + + async def app(scope, receive, send): + await send({"type": "http.response.start", "status": 429, "headers": []}) + + with patch( + "litellm.integrations.prometheus.PrometheusLogger.get_instance", + side_effect=RuntimeError("metrics down"), + ): + await InFlightRequestsMiddleware(app)({"type": "http"}, receive, send) + + assert len(sent) == 1