feat(prometheus): skip metrics for invalid API key requests

Prevent invalid API key requests (401 status code) from being recorded
in Prometheus metrics to avoid polluting metrics with authentication
failures.

- Add helper methods to detect invalid API key requests:
  - _extract_status_code: Centralized status code extraction
  - _is_invalid_api_key_request: Check for 401 or auth error patterns
  - _should_skip_metrics_for_invalid_key: Main validation method

- Integrate validation checks into all metric recording entry points:
  - async_post_call_failure_hook
  - async_post_call_success_hook
  - async_log_failure_event
  - async_log_success_event
  - set_llm_deployment_failure_metrics

- Handles both ProxyException (uses 'code') and standard exceptions
- Also checks exception messages for auth error patterns to catch
  AssertionError cases before conversion to ProxyException

litellm/integrations/prometheus.py

@@ -14,6 +14,7 @@ from typing import (
     Literal,
     Optional,
     Tuple,
+    Union,
     cast,
 )
 
@@ -791,6 +792,11 @@ class PrometheusLogger(CustomLogger):
                 f"standard_logging_object is required, got={standard_logging_payload}"
             )
 
+        if self._should_skip_metrics_for_invalid_key(
+            kwargs=kwargs, standard_logging_payload=standard_logging_payload
+        ):
+            return
+
         model = kwargs.get("model", "")
         litellm_params = kwargs.get("litellm_params", {}) or {}
         _metadata = litellm_params.get("metadata", {})
@@ -1189,11 +1195,17 @@ class PrometheusLogger(CustomLogger):
             f"prometheus Logging - Enters failure logging function for kwargs {kwargs}"
         )
 
-        # unpack kwargs
-        model = kwargs.get("model", "")
         standard_logging_payload: StandardLoggingPayload = kwargs.get(
             "standard_logging_object", {}
         )
+        
+        if self._should_skip_metrics_for_invalid_key(
+            kwargs=kwargs, standard_logging_payload=standard_logging_payload
+        ):
+            return
+        
+        model = kwargs.get("model", "")
+        
         litellm_params = kwargs.get("litellm_params", {}) or {}
         get_end_user_id_for_cost_tracking = _get_cached_end_user_id_for_cost_tracking()
         
@@ -1207,7 +1219,6 @@ class PrometheusLogger(CustomLogger):
         user_api_team_alias = standard_logging_payload["metadata"][
             "user_api_key_team_alias"
         ]
-        kwargs.get("exception", None)
 
         try:
             self.litellm_llm_api_failed_requests_metric.labels(
@@ -1227,6 +1238,139 @@ class PrometheusLogger(CustomLogger):
             pass
         pass
 
+    def _extract_status_code(
+        self,
+        kwargs: Optional[dict] = None,
+        enum_values: Optional[Any] = None,
+        exception: Optional[Exception] = None,
+    ) -> Optional[int]:
+        """
+        Extract HTTP status code from various input formats for validation.
+        
+        This is a centralized helper to extract status code from different
+        callback function signatures. Handles both ProxyException (uses 'code')
+        and standard exceptions (uses 'status_code').
+        
+        Args:
+            kwargs: Dictionary potentially containing 'exception' key
+            enum_values: Object with 'status_code' attribute
+            exception: Exception object to extract status code from directly
+            
+        Returns:
+            Status code as integer if found, None otherwise
+        """
+        status_code = None
+        
+        # Try from enum_values first (most common in our callbacks)
+        if enum_values and hasattr(enum_values, "status_code") and enum_values.status_code:
+            try:
+                status_code = int(enum_values.status_code)
+            except (ValueError, TypeError):
+                pass
+        
+        if not status_code and exception:
+            # ProxyException uses 'code' attribute, other exceptions may use 'status_code'
+            status_code = getattr(exception, "status_code", None) or getattr(exception, "code", None)
+            if status_code is not None:
+                try:
+                    status_code = int(status_code)
+                except (ValueError, TypeError):
+                    status_code = None
+        
+        if not status_code and kwargs:
+            exception_in_kwargs = kwargs.get("exception")
+            if exception_in_kwargs:
+                status_code = getattr(exception_in_kwargs, "status_code", None) or getattr(exception_in_kwargs, "code", None)
+                if status_code is not None:
+                    try:
+                        status_code = int(status_code)
+                    except (ValueError, TypeError):
+                        status_code = None
+        
+        return status_code
+    
+    def _is_invalid_api_key_request(
+        self,
+        status_code: Optional[int],
+        exception: Optional[Exception] = None,
+    ) -> bool:
+        """
+        Determine if a request has an invalid API key based on status code and exception.
+        
+        This method prevents invalid authentication attempts from being recorded in
+        Prometheus metrics. A 401 status code is the definitive indicator of authentication
+        failure. Additionally, we check exception messages for authentication error patterns
+        to catch cases where the exception hasn't been converted to a ProxyException yet.
+        
+        Args:
+            status_code: HTTP status code (401 indicates authentication error)
+            exception: Exception object to check for auth-related error messages
+            
+        Returns:
+            True if the request has an invalid API key and metrics should be skipped,
+            False otherwise
+        """
+        if status_code == 401:
+            return True
+        
+        # Handle cases where AssertionError is raised before conversion to ProxyException
+        if exception is not None:
+            exception_str = str(exception).lower()
+            auth_error_patterns = [
+                "virtual key expected",
+                "expected to start with 'sk-'",
+                "authentication error",
+                "invalid api key",
+                "api key not valid",
+            ]
+            if any(pattern in exception_str for pattern in auth_error_patterns):
+                return True
+        
+        return False
+    
+    def _should_skip_metrics_for_invalid_key(
+        self,
+        kwargs: Optional[dict] = None,
+        user_api_key_dict: Optional[Any] = None,
+        enum_values: Optional[Any] = None,
+        standard_logging_payload: Optional[Union[dict, StandardLoggingPayload]] = None,
+        exception: Optional[Exception] = None,
+    ) -> bool:
+        """
+        Determine if Prometheus metrics should be skipped for invalid API key requests.
+        
+        This is a centralized validation method that extracts status code and exception
+        information from various callback function signatures and determines if the request
+        represents an invalid API key attempt that should be filtered from metrics.
+        
+        Args:
+            kwargs: Dictionary potentially containing exception and other data
+            user_api_key_dict: User API key authentication object (currently unused)
+            enum_values: Object with status_code attribute
+            standard_logging_payload: Standard logging payload dictionary
+            exception: Exception object to check directly
+            
+        Returns:
+            True if metrics should be skipped (invalid key detected), False otherwise
+        """
+        status_code = self._extract_status_code(
+            kwargs=kwargs,
+            enum_values=enum_values,
+            exception=exception,
+        )
+        
+        if exception is None and kwargs:
+            exception = kwargs.get("exception")
+        
+        if self._is_invalid_api_key_request(status_code, exception=exception):
+            verbose_logger.debug(
+                "Skipping Prometheus metrics for invalid API key request: "
+                f"status_code={status_code}, exception={type(exception).__name__ if exception else None}"
+            )
+            return True
+        
+        return False
+
     async def async_post_call_failure_hook(
         self,
         request_data: dict,
@@ -1252,6 +1396,14 @@ class PrometheusLogger(CustomLogger):
             StandardLoggingPayloadSetup,
         )
 
+        if self._should_skip_metrics_for_invalid_key(
+            user_api_key_dict=user_api_key_dict,
+            exception=original_exception,
+        ):
+            return
+
+        status_code = self._extract_status_code(exception=original_exception)
+
         try:
             _tags = StandardLoggingPayloadSetup._get_request_tags(
                 litellm_params=request_data,
@@ -1266,8 +1418,8 @@ class PrometheusLogger(CustomLogger):
                 team=user_api_key_dict.team_id,
                 team_alias=user_api_key_dict.team_alias,
                 requested_model=request_data.get("model", ""),
-                status_code=str(getattr(original_exception, "status_code", None)),
-                exception_status=str(getattr(original_exception, "status_code", None)),
+                status_code=str(status_code),
+                exception_status=str(status_code),
                 exception_class=self._get_exception_class_name(original_exception),
                 tags=_tags,
                 route=user_api_key_dict.request_route,
@@ -1305,6 +1457,11 @@ class PrometheusLogger(CustomLogger):
                 StandardLoggingPayloadSetup,
             )
 
+            if self._should_skip_metrics_for_invalid_key(
+                user_api_key_dict=user_api_key_dict
+            ):
+                return
+
             enum_values = UserAPIKeyLabelValues(
                 end_user=user_api_key_dict.end_user_id,
                 hashed_api_key=user_api_key_dict.api_key,
@@ -1360,6 +1517,15 @@ class PrometheusLogger(CustomLogger):
             exception = request_kwargs.get("exception", None)
 
             llm_provider = _litellm_params.get("custom_llm_provider", None)
+            
+            if self._should_skip_metrics_for_invalid_key(
+                kwargs=request_kwargs,
+                standard_logging_payload=standard_logging_payload,
+            ):
+                return
+            hashed_api_key = standard_logging_payload.get("metadata", {}).get(
+                "user_api_key_hash"
+            )
 
             # Create enum_values for the label factory (always create for use in different metrics)
             enum_values = UserAPIKeyLabelValues(
@@ -1374,9 +1540,7 @@ class PrometheusLogger(CustomLogger):
                     self._get_exception_class_name(exception) if exception else None
                 ),
                 requested_model=model_group,
-                hashed_api_key=standard_logging_payload["metadata"][
-                    "user_api_key_hash"
-                ],
+                hashed_api_key=hashed_api_key,
                 api_key_alias=standard_logging_payload["metadata"][
                     "user_api_key_alias"
                 ],
@@ -1441,6 +1605,14 @@ class PrometheusLogger(CustomLogger):
             if standard_logging_payload is None:
                 return
 
+            # Skip recording metrics for invalid API key requests
+            if self._should_skip_metrics_for_invalid_key(
+                kwargs=request_kwargs,
+                enum_values=enum_values,
+                standard_logging_payload=standard_logging_payload,
+            ):
+                return
+
             api_base = standard_logging_payload["api_base"]
             _litellm_params = request_kwargs.get("litellm_params", {}) or {}
             _metadata = _litellm_params.get("metadata", {})