diff --git a/litellm/proxy/common_utils/http_parsing_utils.py b/litellm/proxy/common_utils/http_parsing_utils.py index ee12f8814e60..a182e9e1915d 100644 --- a/litellm/proxy/common_utils/http_parsing_utils.py +++ b/litellm/proxy/common_utils/http_parsing_utils.py @@ -44,7 +44,19 @@ async def _read_request_body(request: Optional[Request]) -> Dict: parsed_body = {} else: try: + # Parse and immediately clear body reference to prevent memory accumulation parsed_body = orjson.loads(body) + del body # Clear body reference immediately + + # Force garbage collection every 100 requests to prevent accumulation + import gc + if hasattr(_read_request_body, '_gc_counter'): + _read_request_body._gc_counter += 1 + else: + _read_request_body._gc_counter = 1 + + if _read_request_body._gc_counter % 100 == 0: + gc.collect() except orjson.JSONDecodeError as e: # First try the standard json module which is more forgiving # First decode bytes to string if needed @@ -90,18 +102,64 @@ async def _read_request_body(request: Optional[Request]) -> Dict: return {} + + def _safe_get_request_parsed_body(request: Optional[Request]) -> Optional[dict]: if request is None: return None + + # Check if we already have a parsed body stored directly on the request object + if hasattr(request, "_litellm_parsed_body"): + return getattr(request, "_litellm_parsed_body") + + # Fallback to checking request.scope for backward compatibility if ( hasattr(request, "scope") and "parsed_body" in request.scope and isinstance(request.scope["parsed_body"], tuple) ): accepted_keys, parsed_body = request.scope["parsed_body"] - return {key: parsed_body[key] for key in accepted_keys} + result = {key: parsed_body[key] for key in accepted_keys} + # Clean up the scope to free memory and store on request object + del request.scope["parsed_body"] + setattr(request, "_litellm_parsed_body", result) + return result return None +def _safe_get_request_query_params(request: Optional[Request]) -> Dict: + if request is None: + return {} + try: + if hasattr(request, "query_params"): + return dict(request.query_params) + return {} + except Exception as e: + verbose_proxy_logger.debug( + "Unexpected error reading request query params - {}".format(e) + ) + return {} + +def cleanup_request_memory(request: Optional[Request]) -> None: + """ + Explicitly cleanup request memory to prevent leaks. + Call this after request processing is complete. + """ + if request is None: + return + + try: + # Remove parsed body from request object + if hasattr(request, "_litellm_parsed_body"): + delattr(request, "_litellm_parsed_body") + + # Clean up any remaining scope data + if hasattr(request, "scope") and "parsed_body" in request.scope: + del request.scope["parsed_body"] + + except Exception: + pass # Silent cleanup - don't break request processing + + def _safe_get_request_query_params(request: Optional[Request]) -> Dict: if request is None: return {} @@ -122,7 +180,12 @@ def _safe_set_request_parsed_body( try: if request is None: return - request.scope["parsed_body"] = (tuple(parsed_body.keys()), parsed_body) + + # Store the parsed body directly on the request object + # This prevents memory leaks and data cross-contamination since + # the data is tied to the specific request object lifecycle + setattr(request, "_litellm_parsed_body", parsed_body) + except Exception as e: verbose_proxy_logger.debug( "Unexpected error setting request parsed body - {}".format(e) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index fda5e9414840..cbe3e464ebff 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -57,6 +57,9 @@ pass_through_endpoint_logging = PassThroughEndpointLogging() +# Global registry to track registered pass-through routes and prevent memory leaks +_registered_pass_through_routes: Dict[str, Dict[str, str]] = {} + def get_response_body(response: httpx.Response) -> Optional[dict]: try: @@ -970,8 +973,19 @@ def add_exact_path_route( merge_query_params: Optional[bool], dependencies: Optional[List], cost_per_request: Optional[float], + endpoint_id: str, ): """Add exact path route for pass-through endpoint""" + route_key = f"{endpoint_id}:exact:{path}" + + # Check if this exact route is already registered + if route_key in _registered_pass_through_routes: + verbose_proxy_logger.debug( + "Skipping duplicate exact pass through endpoint: %s (already registered)", + path, + ) + return + verbose_proxy_logger.debug( "adding exact pass through endpoint: %s, dependencies: %s", path, @@ -992,6 +1006,13 @@ def add_exact_path_route( methods=["GET", "POST", "PUT", "DELETE", "PATCH"], dependencies=dependencies, ) + + # Register the route to prevent duplicates + _registered_pass_through_routes[route_key] = { + "endpoint_id": endpoint_id, + "path": path, + "type": "exact" + } @staticmethod def add_subpath_route( @@ -1003,9 +1024,20 @@ def add_subpath_route( merge_query_params: Optional[bool], dependencies: Optional[List], cost_per_request: Optional[float], + endpoint_id: str, ): """Add wildcard route for sub-paths""" wildcard_path = f"{path}/{{subpath:path}}" + route_key = f"{endpoint_id}:subpath:{path}" + + # Check if this subpath route is already registered + if route_key in _registered_pass_through_routes: + verbose_proxy_logger.debug( + "Skipping duplicate wildcard pass through endpoint: %s (already registered)", + wildcard_path, + ) + return + verbose_proxy_logger.debug( "adding wildcard pass through endpoint: %s, dependencies: %s", wildcard_path, @@ -1027,6 +1059,26 @@ def add_subpath_route( methods=["GET", "POST", "PUT", "DELETE", "PATCH"], dependencies=dependencies, ) + + # Register the route to prevent duplicates + _registered_pass_through_routes[route_key] = { + "endpoint_id": endpoint_id, + "path": path, + "type": "subpath" + } + + @staticmethod + def remove_endpoint_routes(endpoint_id: str): + """Remove all routes for a specific endpoint ID from the registry""" + keys_to_remove = [ + key for key, value in _registered_pass_through_routes.items() + if value["endpoint_id"] == endpoint_id + ] + for key in keys_to_remove: + del _registered_pass_through_routes[key] + verbose_proxy_logger.debug( + "Removed pass-through route from registry: %s", key + ) async def initialize_pass_through_endpoints( @@ -1093,6 +1145,7 @@ async def initialize_pass_through_endpoints( merge_query_params=_merge_query_params, dependencies=_dependencies, cost_per_request=endpoint.get("cost_per_request", None), + endpoint_id=endpoint.get("id"), ) # Add wildcard route for sub-paths @@ -1106,6 +1159,7 @@ async def initialize_pass_through_endpoints( merge_query_params=_merge_query_params, dependencies=_dependencies, cost_per_request=endpoint.get("cost_per_request", None), + endpoint_id=endpoint.get("id"), ) verbose_proxy_logger.debug( @@ -1250,6 +1304,9 @@ async def update_pass_through_endpoints( # Update the list pass_through_endpoint_data[endpoint_index] = endpoint_dict + # Remove old routes from registry before they get re-registered + InitPassThroughEndpointHelpers.remove_endpoint_routes(endpoint_id) + ## Update db updated_data = ConfigFieldUpdate( field_name="pass_through_endpoints", @@ -1394,6 +1451,9 @@ async def delete_pass_through_endpoints( pass_through_endpoint_data.pop(endpoint_index) response_obj = found_endpoint + # Remove routes from registry + InitPassThroughEndpointHelpers.remove_endpoint_routes(endpoint_id) + ## Update db updated_data = ConfigFieldUpdate( field_name="pass_through_endpoints", diff --git a/litellm/router.py b/litellm/router.py index 3be88596b115..343bcb7b06e1 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -435,6 +435,9 @@ def __init__( # noqa: PLR0915 self.failed_calls = ( InMemoryCache() ) # cache to track failed call per deployment, if num failed calls within 1 minute > allowed fails, then add it to cooldown + self.failed_deployments = ( + InMemoryCache() + ) # cache to track deployments that failed due to configuration errors (missing credentials, etc.) to prevent repeated retry attempts if num_retries is not None: self.num_retries = num_retries @@ -5255,6 +5258,14 @@ def upsert_deployment(self, deployment: Deployment) -> Optional[Deployment]: # check if deployment already exists _deployment_model_id = deployment.model_info.id or "" + # Check if this deployment has previously failed due to configuration errors + _deployment_cache_key = f"failed_deployment_{_deployment_model_id}" + if self.failed_deployments.get_cache(_deployment_cache_key) is not None: + verbose_router_logger.debug( + f"Skipping deployment {_deployment_model_id} - previously failed due to configuration errors" + ) + return None + _deployment_on_router: Optional[Deployment] = self.get_deployment( model_id=_deployment_model_id ) @@ -5279,9 +5290,29 @@ def upsert_deployment(self, deployment: Deployment) -> Optional[Deployment]: return deployment except Exception as e: if self.ignore_invalid_deployments: - verbose_router_logger.debug( - f"Error upserting deployment: {e}, ignoring and continuing with other deployments." - ) + # Check if this is a configuration error (missing credentials, etc.) + _is_config_error = ( + "api_key is required" in str(e) or + "vertex_project, and vertex_location must be set" in str(e) or + "credentials" in str(e).lower() or + "authentication" in str(e).lower() + ) + + if _is_config_error: + # Cache this deployment as failed to prevent repeated retries + _deployment_cache_key = f"failed_deployment_{deployment.model_info.id or ''}" + self.failed_deployments.set_cache( + key=_deployment_cache_key, + value=str(e), + ttl=3600 # Cache for 1 hour + ) + verbose_router_logger.warning( + f"Error upserting deployment: {e}, caching as failed deployment to prevent retries." + ) + else: + verbose_router_logger.warning( + f"Error upserting deployment: {e}, ignoring and continuing with other deployments." + ) return None else: raise e @@ -5337,6 +5368,26 @@ def get_deployment_credentials(self, model_id: str) -> Optional[dict]: **deployment.litellm_params.model_dump(exclude_none=True) ).model_dump(exclude_none=True) + def clear_failed_deployments_cache(self, model_id: Optional[str] = None) -> None: + """ + Clear the cache of failed deployments to allow retrying them. + + Parameters: + - model_id (Optional[str]): If provided, only clear the cache for this specific model ID. + If None, clear all failed deployment entries. + """ + if model_id is not None: + # Clear cache for specific deployment + _deployment_cache_key = f"failed_deployment_{model_id}" + self.failed_deployments.delete_cache(_deployment_cache_key) + verbose_router_logger.info( + f"Cleared failed deployment cache for model_id: {model_id}" + ) + else: + # Clear all failed deployment entries + self.failed_deployments.flush_cache() + verbose_router_logger.info("Cleared all failed deployment cache entries") + def get_deployment_by_model_group_name( self, model_group_name: str ) -> Optional[Deployment]: