Litellm dev memory fix

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)

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",

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]: