diff --git a/litellm/__init__.py b/litellm/__init__.py index e61ef25057..617c102fb8 100644 --- a/litellm/__init__.py +++ b/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 ) diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index 147420cd53..65b8a9fc4b 100644 --- a/litellm/integrations/prometheus.py +++ b/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( diff --git a/litellm/integrations/prometheus_helpers/bounded_prometheus_series_tracker.py b/litellm/integrations/prometheus_helpers/bounded_prometheus_series_tracker.py new file mode 100644 index 0000000000..d834ae2014 --- /dev/null +++ b/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 diff --git a/tests/test_litellm/integrations/test_prometheus_end_user_cardinality.py b/tests/test_litellm/integrations/test_prometheus_end_user_cardinality.py new file mode 100644 index 0000000000..868d86a6c2 --- /dev/null +++ b/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