Cap Prometheus end-user metric cardinality

litellm/__init__.py

@@ -414,6 +414,9 @@
 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 @@
 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 @@ def __init__(  # noqa: PLR0915
                 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 @@ def get_labels_for_metric(
 
         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 @@ def _inc_labeled_counter(
             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 @@ def _set_latency_metrics(
             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 @@ def _set_latency_metrics(
             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 @@ def _set_latency_metrics(
             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 @@ def _set_latency_metrics(
             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