perf: cap Prometheus end-user metric cardinality with TTL + LRU eviction (#27272)

Co-authored-by: Yassin Kortam <yassinkortam@g.ucla.edu>

litellm/__init__.py

@@ -414,6 +414,9 @@ custom_prometheus_metadata_labels: List[str] = []
 custom_prometheus_tags: List[str] = []
 prometheus_metrics_config: Optional[List] = None
 prometheus_emit_stream_label: bool = False
+prometheus_end_user_metrics_max_series_per_metric: Optional[int] = 10000
+prometheus_end_user_metrics_ttl_seconds: Optional[float] = 3600.0
+prometheus_end_user_metrics_cleanup_interval_seconds: Optional[float] = 60.0
 disable_add_prefix_to_prompt: bool = (
     False  # used by anthropic, to disable adding prefix to prompt
 )

litellm/integrations/prometheus.py

@@ -25,6 +25,9 @@ from typing import (
 import litellm
 from litellm._logging import print_verbose, verbose_logger
 from litellm.integrations.custom_logger import CustomLogger
+from litellm.integrations.prometheus_helpers.bounded_prometheus_series_tracker import (
+    BoundedPrometheusSeriesTracker,
+)
 from litellm.integrations.prometheus_helpers import (
     PrometheusLabelFactoryContext,
     _get_cached_end_user_id_for_cost_tracking,
@@ -81,6 +84,7 @@ class PrometheusLogger(CustomLogger):
                 if _custom_buckets is not None
                 else LATENCY_BUCKETS
             )
+            self._bounded_prometheus_series_tracker = BoundedPrometheusSeriesTracker()
 
             # Create metric factory functions
             self._counter_factory = self._create_metric_factory(Counter)
@@ -984,6 +988,40 @@ class PrometheusLogger(CustomLogger):
 
         return filtered_labels
 
+    def _track_end_user_metric_series(
+        self,
+        metric: Any,
+        metric_name: DEFINED_PROMETHEUS_METRICS,
+        labels: Dict[str, Optional[str]],
+    ) -> None:
+        """
+        Cap the cardinality of metrics that include the ``end_user`` label.
+
+        Called *after* ``metric.labels(...).inc()/observe()`` so the emission is
+        recorded in prometheus-client's child map before any eviction runs.
+        Series that get evicted before the next scrape lose updates accrued
+        since the last scrape — this is inherent to any cardinality cap.
+        """
+        labelnames = self.get_labels_for_metric(metric_name)
+        if UserAPIKeyLabelNames.END_USER.value not in labelnames:
+            return
+        if labels.get(UserAPIKeyLabelNames.END_USER.value) is None:
+            return
+
+        max_series = litellm.prometheus_end_user_metrics_max_series_per_metric
+        ttl_seconds = litellm.prometheus_end_user_metrics_ttl_seconds
+        if max_series is None and ttl_seconds is None:
+            return
+
+        self._bounded_prometheus_series_tracker.track_series(
+            metric=metric,
+            metric_name=metric_name,
+            label_values=tuple(labels.get(label) for label in labelnames),
+            max_series=max_series,
+            ttl_seconds=ttl_seconds,
+            cleanup_interval_seconds=litellm.prometheus_end_user_metrics_cleanup_interval_seconds,
+        )
+
     def _inc_labeled_counter(
         self,
         counter: Any,
@@ -998,6 +1036,7 @@ class PrometheusLogger(CustomLogger):
             label_context=label_context,
         )
         counter.labels(**_labels).inc(amount)
+        self._track_end_user_metric_series(counter, metric_name, _labels)
 
     async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
         # Define prometheus client
@@ -1479,6 +1518,11 @@ class PrometheusLogger(CustomLogger):
             self.litellm_llm_api_time_to_first_token_metric.labels(
                 **_ttft_labels
             ).observe(time_to_first_token_seconds)
+            self._track_end_user_metric_series(
+                self.litellm_llm_api_time_to_first_token_metric,
+                "litellm_llm_api_time_to_first_token_metric",
+                _ttft_labels,
+            )
         else:
             verbose_logger.debug(
                 "Time to first token metric not emitted, stream option in model_parameters is not True"
@@ -1499,6 +1543,11 @@ class PrometheusLogger(CustomLogger):
             self.litellm_llm_api_latency_metric.labels(**_labels).observe(
                 api_call_total_time_seconds
             )
+            self._track_end_user_metric_series(
+                self.litellm_llm_api_latency_metric,
+                "litellm_llm_api_latency_metric",
+                _labels,
+            )
 
         # total request latency
         total_time_seconds = self._safe_duration_seconds(
@@ -1516,6 +1565,11 @@ class PrometheusLogger(CustomLogger):
             self.litellm_request_total_latency_metric.labels(**_labels).observe(
                 total_time_seconds
             )
+            self._track_end_user_metric_series(
+                self.litellm_request_total_latency_metric,
+                "litellm_request_total_latency_metric",
+                _labels,
+            )
 
         # request queue time (time from arrival to processing start)
         _litellm_params = kwargs.get("litellm_params", {}) or {}
@@ -1533,6 +1587,11 @@ class PrometheusLogger(CustomLogger):
             self.litellm_request_queue_time_metric.labels(**_labels).observe(
                 queue_time_seconds
             )
+            self._track_end_user_metric_series(
+                self.litellm_request_queue_time_metric,
+                "litellm_request_queue_time_seconds",
+                _labels,
+            )
 
     async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
         verbose_logger.debug(

litellm/integrations/prometheus_helpers/bounded_prometheus_series_tracker.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import time
+from collections import OrderedDict
+from threading import RLock
+from typing import Any, Dict, Optional
+
+
+class BoundedPrometheusSeriesTracker:
+    """
+    Tracks Prometheus child series and removes stale/excess labelsets.
+
+    The tracker is label-agnostic: callers decide which series should be tracked
+    and pass the full label tuple used by the Prometheus metric.
+    """
+
+    def __init__(self) -> None:
+        self._series: Dict[str, OrderedDict[tuple[Optional[str], ...], float]] = {}
+        self._last_ttl_cleanup: Dict[str, float] = {}
+        self.lock = RLock()
+
+    def track_series(
+        self,
+        metric: Any,
+        metric_name: str,
+        label_values: tuple[Optional[str], ...],
+        max_series: Optional[int],
+        ttl_seconds: Optional[float],
+        cleanup_interval_seconds: Optional[float],
+    ) -> None:
+        if max_series is None and ttl_seconds is None:
+            return
+
+        now = time.monotonic()
+
+        with self.lock:
+            series = self._series.setdefault(metric_name, OrderedDict())
+            series[label_values] = now
+            series.move_to_end(label_values)
+
+            if ttl_seconds is not None and self._should_run_ttl_cleanup(
+                metric_name=metric_name,
+                now=now,
+                cleanup_interval_seconds=cleanup_interval_seconds,
+            ):
+                expired_label_values = [
+                    tracked_label_values
+                    for tracked_label_values, last_seen in series.items()
+                    if now - last_seen > ttl_seconds
+                ]
+                for tracked_label_values in expired_label_values:
+                    self._remove_metric_series(metric, series, tracked_label_values)
+
+            # max_series <= 0 is treated as "unlimited" so a misconfigured zero
+            # value cannot silently drop every emission for this metric.
+            if max_series is not None and max_series > 0:
+                while len(series) > max_series:
+                    tracked_label_values = next(iter(series))
+                    if not self._remove_metric_child(metric, tracked_label_values):
+                        break
+                    del series[tracked_label_values]
+
+    def _should_run_ttl_cleanup(
+        self,
+        metric_name: str,
+        now: float,
+        cleanup_interval_seconds: Optional[float],
+    ) -> bool:
+        if cleanup_interval_seconds is None or cleanup_interval_seconds <= 0:
+            self._last_ttl_cleanup[metric_name] = now
+            return True
+
+        last_cleanup = self._last_ttl_cleanup.get(metric_name)
+        if last_cleanup is None or now - last_cleanup >= cleanup_interval_seconds:
+            self._last_ttl_cleanup[metric_name] = now
+            return True
+        return False
+
+    def _remove_metric_series(
+        self,
+        metric: Any,
+        series: OrderedDict[tuple[Optional[str], ...], float],
+        label_values: tuple[Optional[str], ...],
+    ) -> None:
+        if self._remove_metric_child(metric, label_values):
+            series.pop(label_values, None)
+
+    @staticmethod
+    def _remove_metric_child(
+        metric: Any, label_values: tuple[Optional[str], ...]
+    ) -> bool:
+        """
+        Remove the Prometheus child for ``label_values`` and report whether the
+        tracker should commit the matching state change.
+
+        Returns ``True`` when the child is no longer present in Prometheus
+        (either it was just removed or it was already gone), and ``False`` when
+        ``metric.remove()`` raised an unexpected error and the child likely
+        still exists.
+        """
+        try:
+            metric.remove(*label_values)
+            return True
+        except KeyError:
+            return True
+        except (AttributeError, ValueError):
+            return False

tests/test_litellm/integrations/test_prometheus_end_user_cardinality.py

@@ -0,0 +1,181 @@
+from time import monotonic
+
+import pytest
+from prometheus_client import REGISTRY
+
+import litellm
+from litellm.integrations.prometheus import PrometheusLogger
+from litellm.integrations.prometheus_helpers import bounded_prometheus_series_tracker
+from litellm.integrations.prometheus_helpers.bounded_prometheus_series_tracker import (
+    BoundedPrometheusSeriesTracker,
+)
+from litellm.types.integrations.prometheus import UserAPIKeyLabelValues
+
+
+@pytest.fixture(autouse=True)
+def cleanup_prometheus_registry():
+    collectors = list(REGISTRY._collector_to_names.keys())
+    for collector in collectors:
+        try:
+            REGISTRY.unregister(collector)
+        except Exception:
+            pass
+
+    old_enable_end_user = litellm.enable_end_user_cost_tracking_prometheus_only
+    old_metrics_config = litellm.prometheus_metrics_config
+    old_max_series = litellm.prometheus_end_user_metrics_max_series_per_metric
+    old_ttl_seconds = litellm.prometheus_end_user_metrics_ttl_seconds
+    old_cleanup_interval_seconds = (
+        litellm.prometheus_end_user_metrics_cleanup_interval_seconds
+    )
+
+    yield
+
+    litellm.enable_end_user_cost_tracking_prometheus_only = old_enable_end_user
+    litellm.prometheus_metrics_config = old_metrics_config
+    litellm.prometheus_end_user_metrics_max_series_per_metric = old_max_series
+    litellm.prometheus_end_user_metrics_ttl_seconds = old_ttl_seconds
+    litellm.prometheus_end_user_metrics_cleanup_interval_seconds = (
+        old_cleanup_interval_seconds
+    )
+
+    collectors = list(REGISTRY._collector_to_names.keys())
+    for collector in collectors:
+        try:
+            REGISTRY.unregister(collector)
+        except Exception:
+            pass
+
+
+def test_prometheus_end_user_series_are_capped_per_metric():
+    litellm.enable_end_user_cost_tracking_prometheus_only = True
+    litellm.prometheus_metrics_config = [
+        {
+            "group": "end-user-spend",
+            "metrics": ["litellm_spend_metric"],
+            "include_labels": ["end_user"],
+        }
+    ]
+    litellm.prometheus_end_user_metrics_max_series_per_metric = 3
+    litellm.prometheus_end_user_metrics_ttl_seconds = None
+    logger = PrometheusLogger()
+
+    for index in range(6):
+        PrometheusLogger._inc_labeled_counter(
+            logger,
+            logger.litellm_spend_metric,
+            "litellm_spend_metric",
+            UserAPIKeyLabelValues(end_user=f"end-user-{index}"),
+            amount=0.01,
+        )
+
+    assert len(logger.litellm_spend_metric._metrics) == 3
+    assert set(logger.litellm_spend_metric._metrics) == {
+        ("end-user-3",),
+        ("end-user-4",),
+        ("end-user-5",),
+    }
+
+
+def test_bounded_prometheus_series_tracker_is_label_agnostic():
+    class FakeMetric:
+        def __init__(self):
+            self.removed_label_values = []
+
+        def remove(self, *label_values):
+            self.removed_label_values.append(label_values)
+
+    metric = FakeMetric()
+    tracker = BoundedPrometheusSeriesTracker()
+
+    for index in range(4):
+        tracker.track_series(
+            metric=metric,
+            metric_name="generic_metric",
+            label_values=(f"route-{index}", "200"),
+            max_series=2,
+            ttl_seconds=None,
+            cleanup_interval_seconds=60.0,
+        )
+
+    assert metric.removed_label_values == [
+        ("route-0", "200"),
+        ("route-1", "200"),
+    ]
+
+
+def test_bounded_prometheus_series_tracker_treats_zero_max_as_unlimited():
+    # A misconfigured ``max_series=0`` must not silently evict every emission.
+    class FakeMetric:
+        def __init__(self):
+            self.removed_label_values = []
+
+        def remove(self, *label_values):
+            self.removed_label_values.append(label_values)
+
+    metric = FakeMetric()
+    tracker = BoundedPrometheusSeriesTracker()
+
+    for index in range(3):
+        tracker.track_series(
+            metric=metric,
+            metric_name="generic_metric",
+            label_values=(f"end-user-{index}",),
+            max_series=0,
+            ttl_seconds=None,
+            cleanup_interval_seconds=60.0,
+        )
+
+    assert metric.removed_label_values == []
+
+
+def test_prometheus_end_user_series_expire_by_ttl(monkeypatch):
+    litellm.enable_end_user_cost_tracking_prometheus_only = True
+    litellm.prometheus_metrics_config = [
+        {
+            "group": "end-user-spend",
+            "metrics": ["litellm_spend_metric"],
+            "include_labels": ["end_user"],
+        }
+    ]
+    litellm.prometheus_end_user_metrics_max_series_per_metric = None
+    litellm.prometheus_end_user_metrics_ttl_seconds = 10.0
+    litellm.prometheus_end_user_metrics_cleanup_interval_seconds = 0.0
+    logger = PrometheusLogger()
+
+    current_time = [monotonic()]
+    monkeypatch.setattr(
+        bounded_prometheus_series_tracker.time,
+        "monotonic",
+        lambda: current_time[0],
+    )
+    PrometheusLogger._inc_labeled_counter(
+        logger,
+        logger.litellm_spend_metric,
+        "litellm_spend_metric",
+        UserAPIKeyLabelValues(end_user="stale-end-user"),
+        amount=0.01,
+    )
+
+    current_time[0] += 11.0
+    PrometheusLogger._inc_labeled_counter(
+        logger,
+        logger.litellm_spend_metric,
+        "litellm_spend_metric",
+        UserAPIKeyLabelValues(end_user="fresh-end-user"),
+        amount=0.01,
+    )
+
+    assert set(logger.litellm_spend_metric._metrics) == {("fresh-end-user",)}
+
+
+def test_prometheus_end_user_not_tracked_by_default():
+    litellm.enable_end_user_cost_tracking_prometheus_only = None
+    labels = PrometheusLogger().get_labels_for_metric("litellm_spend_metric")
+    assert "end_user" in labels
+
+    label_values = UserAPIKeyLabelValues(end_user="not-exported")
+    from litellm.integrations.prometheus import prometheus_label_factory
+
+    prometheus_labels = prometheus_label_factory(labels, label_values)
+    assert prometheus_labels["end_user"] is None