feat(observability): expose per-pod request pressure and the enforced concurrency ceiling

Operators could see how many requests were in flight on a pod but not how many
the proxy was shedding, nor what ceiling was actually being applied, so there
was no way to tell "throttle upstream" apart from "add pods".

Shed responses are counted at the ASGI layer rather than at each limiter, so no
rejection path can be missed, and the count is per worker for the same reason
the in-flight gauge is. 500s are excluded: that is the proxy failing, not
declining.

The ceiling gauge reports what is actually in force. global_max_parallel_requests
is only read by the v1 limiter, which is off by default, so the gauge reports
+Inf when nothing bounds concurrency rather than echoing a configured number
that no limiter applies. A registered gauge always exposes a value, so leaving
it unset would have rendered as 0 and read as "no requests allowed".

Refs LIT-5435

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()

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)

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()

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)

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, ...] = ()

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)

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