diff --git a/litellm/caching/in_memory_cache.py b/litellm/caching/in_memory_cache.py index a3ee8813449..f644fdfefd7 100644 --- a/litellm/caching/in_memory_cache.py +++ b/litellm/caching/in_memory_cache.py @@ -234,3 +234,11 @@ class InMemoryCache(BaseCache): Get the remaining TTL of a key in in-memory cache """ return self.ttl_dict.get(key, None) + + async def async_get_oldest_n_keys(self, n: int) -> List[str]: + """ + Get the oldest n keys in the cache + """ + # sorted ttl dict by ttl + sorted_ttl_dict = sorted(self.ttl_dict.items(), key=lambda x: x[1]) + return [key for key, _ in sorted_ttl_dict[:n]] diff --git a/litellm/integrations/SlackAlerting/hanging_request_check.py b/litellm/integrations/SlackAlerting/hanging_request_check.py new file mode 100644 index 00000000000..713e790ba90 --- /dev/null +++ b/litellm/integrations/SlackAlerting/hanging_request_check.py @@ -0,0 +1,175 @@ +""" +Class to check for LLM API hanging requests + + +Notes: +- Do not create tasks that sleep, that can saturate the event loop +- Do not store large objects (eg. messages in memory) that can increase RAM usage +""" + +import asyncio +from typing import TYPE_CHECKING, Any, Optional + +import litellm +from litellm._logging import verbose_proxy_logger +from litellm.caching.in_memory_cache import InMemoryCache +from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs +from litellm.types.integrations.slack_alerting import ( + HANGING_ALERT_BUFFER_TIME_SECONDS, + MAX_OLDEST_HANGING_REQUESTS_TO_CHECK, + HangingRequestData, +) + +if TYPE_CHECKING: + from litellm.integrations.SlackAlerting.slack_alerting import SlackAlerting +else: + SlackAlerting = Any + + +class AlertingHangingRequestCheck: + """ + Class to safely handle checking hanging requests alerts + """ + + def __init__( + self, + slack_alerting_object: SlackAlerting, + ): + self.slack_alerting_object = slack_alerting_object + self.hanging_request_cache = InMemoryCache( + default_ttl=int( + self.slack_alerting_object.alerting_threshold + + HANGING_ALERT_BUFFER_TIME_SECONDS + ), + ) + + async def add_request_to_hanging_request_check( + self, + request_data: Optional[dict] = None, + ): + """ + Add a request to the hanging request cache. This is the list of request_ids that gets periodicall checked for hanging requests + """ + if request_data is None: + return + + request_metadata = get_litellm_metadata_from_kwargs(kwargs=request_data) + model = request_data.get("model", "") + api_base: Optional[str] = None + + if request_data.get("deployment", None) is not None and isinstance( + request_data["deployment"], dict + ): + api_base = litellm.get_api_base( + model=model, + optional_params=request_data["deployment"].get("litellm_params", {}), + ) + + hanging_request_data = HangingRequestData( + request_id=request_data.get("litellm_call_id", ""), + model=model, + api_base=api_base, + key_alias=request_metadata.get("user_api_key_alias", ""), + team_alias=request_metadata.get("user_api_key_team_alias", ""), + ) + + await self.hanging_request_cache.async_set_cache( + key=hanging_request_data.request_id, + value=hanging_request_data, + ttl=int( + self.slack_alerting_object.alerting_threshold + + HANGING_ALERT_BUFFER_TIME_SECONDS + ), + ) + return + + async def send_alerts_for_hanging_requests(self): + """ + Send alerts for hanging requests + """ + from litellm.proxy.proxy_server import proxy_logging_obj + + ######################################################### + # Find all requests that have been hanging for more than the alerting threshold + # Get the last 50 oldest items in the cache and check if they have completed + ######################################################### + # check if request_id is in internal usage cache + if proxy_logging_obj.internal_usage_cache is None: + return + + hanging_requests = await self.hanging_request_cache.async_get_oldest_n_keys( + n=MAX_OLDEST_HANGING_REQUESTS_TO_CHECK, + ) + + for request_id in hanging_requests: + hanging_request_data: Optional[HangingRequestData] = ( + await self.hanging_request_cache.async_get_cache( + key=request_id, + ) + ) + + if hanging_request_data is None: + continue + + request_status = ( + await proxy_logging_obj.internal_usage_cache.async_get_cache( + key="request_status:{}".format(hanging_request_data.request_id), + litellm_parent_otel_span=None, + local_only=True, + ) + ) + # this means the request status was either success or fail + # and is not hanging + if request_status is not None: + # clear this request from hanging request cache since the request was either success or failed + self.hanging_request_cache._remove_key( + key=request_id, + ) + continue + + ################ + # Send the Alert on Slack + ################ + await self.send_hanging_request_alert( + hanging_request_data=hanging_request_data + ) + + return + + async def check_for_hanging_requests( + self, + ): + """ + Background task that checks all request ids in self.hanging_request_cache to check if they have completed + + Runs every alerting_threshold/2 seconds to check for hanging requests + """ + while True: + verbose_proxy_logger.debug("Checking for hanging requests....") + await self.send_alerts_for_hanging_requests() + await asyncio.sleep(self.slack_alerting_object.alerting_threshold / 2) + + async def send_hanging_request_alert( + self, + hanging_request_data: HangingRequestData, + ): + """ + Send a hanging request alert + """ + from litellm.integrations.SlackAlerting.slack_alerting import AlertType + + ################ + # Send the Alert on Slack + ################ + request_info = f"""Request Model: `{hanging_request_data.model}` +API Base: `{hanging_request_data.api_base}` +Key Alias: `{hanging_request_data.key_alias}` +Team Alias: `{hanging_request_data.team_alias}`""" + + alerting_message = f"`Requests are hanging - {self.slack_alerting_object.alerting_threshold}s+ request time`" + await self.slack_alerting_object.send_alert( + message=alerting_message + "\n" + request_info, + level="Medium", + alert_type=AlertType.llm_requests_hanging, + alerting_metadata=hanging_request_data.alerting_metadata or {}, + ) diff --git a/litellm/integrations/SlackAlerting/slack_alerting.py b/litellm/integrations/SlackAlerting/slack_alerting.py index 16305061ec8..41db4a551bd 100644 --- a/litellm/integrations/SlackAlerting/slack_alerting.py +++ b/litellm/integrations/SlackAlerting/slack_alerting.py @@ -19,6 +19,9 @@ from litellm.caching.caching import DualCache from litellm.constants import HOURS_IN_A_DAY from litellm.integrations.custom_batch_logger import CustomBatchLogger from litellm.integrations.SlackAlerting.budget_alert_types import get_budget_alert_type +from litellm.integrations.SlackAlerting.hanging_request_check import ( + AlertingHangingRequestCheck, +) from litellm.litellm_core_utils.duration_parser import duration_in_seconds from litellm.litellm_core_utils.exception_mapping_utils import ( _add_key_name_and_team_to_alert, @@ -38,7 +41,7 @@ from litellm.types.integrations.slack_alerting import * from ..email_templates.templates import * from .batching_handler import send_to_webhook, squash_payloads -from .utils import _add_langfuse_trace_id_to_alert, process_slack_alerting_variables +from .utils import process_slack_alerting_variables if TYPE_CHECKING: from litellm.router import Router as _Router @@ -86,6 +89,9 @@ class SlackAlerting(CustomBatchLogger): self.default_webhook_url = default_webhook_url self.flush_lock = asyncio.Lock() self.periodic_started = False + self.hanging_request_check = AlertingHangingRequestCheck( + slack_alerting_object=self, + ) super().__init__(**kwargs, flush_lock=self.flush_lock) def update_values( @@ -107,10 +113,10 @@ class SlackAlerting(CustomBatchLogger): self.alert_types = alert_types if alerting_args is not None: self.alerting_args = SlackAlertingArgs(**alerting_args) - if not self.periodic_started: + if not self.periodic_started: asyncio.create_task(self.periodic_flush()) self.periodic_started = True - + if alert_to_webhook_url is not None: # update the dict if self.alert_to_webhook_url is None: @@ -451,106 +457,17 @@ class SlackAlerting(CustomBatchLogger): async def response_taking_too_long( self, - start_time: Optional[datetime.datetime] = None, - end_time: Optional[datetime.datetime] = None, - type: Literal["hanging_request", "slow_response"] = "hanging_request", request_data: Optional[dict] = None, ): if self.alerting is None or self.alert_types is None: return - model: str = "" - if request_data is not None: - model = request_data.get("model", "") - messages = request_data.get("messages", None) - if messages is None: - # if messages does not exist fallback to "input" - messages = request_data.get("input", None) - # try casting messages to str and get the first 100 characters, else mark as None - try: - messages = str(messages) - messages = messages[:100] - except Exception: - messages = "" + if AlertType.llm_requests_hanging not in self.alert_types: + return - if ( - litellm.turn_off_message_logging - or litellm.redact_messages_in_exceptions - ): - messages = ( - "Message not logged. litellm.redact_messages_in_exceptions=True" - ) - request_info = f"\nRequest Model: `{model}`\nMessages: `{messages}`" - else: - request_info = "" - - if type == "hanging_request": - await asyncio.sleep( - self.alerting_threshold - ) # Set it to 5 minutes - i'd imagine this might be different for streaming, non-streaming, non-completion (embedding + img) requests - alerting_metadata: dict = {} - if await self._request_is_completed(request_data=request_data) is True: - return - - if request_data is not None: - if request_data.get("deployment", None) is not None and isinstance( - request_data["deployment"], dict - ): - _api_base = litellm.get_api_base( - model=model, - optional_params=request_data["deployment"].get( - "litellm_params", {} - ), - ) - - if _api_base is None: - _api_base = "" - - request_info += f"\nAPI Base: {_api_base}" - elif request_data.get("metadata", None) is not None and isinstance( - request_data["metadata"], dict - ): - # In hanging requests sometime it has not made it to the point where the deployment is passed to the `request_data`` - # in that case we fallback to the api base set in the request metadata - _metadata: dict = request_data["metadata"] - _api_base = _metadata.get("api_base", "") - - request_info = _add_key_name_and_team_to_alert( - request_info=request_info, metadata=_metadata - ) - - if _api_base is None: - _api_base = "" - - if "alerting_metadata" in _metadata: - alerting_metadata = _metadata["alerting_metadata"] - request_info += f"\nAPI Base: `{_api_base}`" - # only alert hanging responses if they have not been marked as success - alerting_message = ( - f"`Requests are hanging - {self.alerting_threshold}s+ request time`" - ) - - if "langfuse" in litellm.success_callback: - langfuse_url = await _add_langfuse_trace_id_to_alert( - request_data=request_data, - ) - - if langfuse_url is not None: - request_info += "\n🪢 Langfuse Trace: {}".format(langfuse_url) - - # add deployment latencies to alert - _deployment_latency_map = self._get_deployment_latencies_to_alert( - metadata=request_data.get("metadata", {}) - ) - if _deployment_latency_map is not None: - request_info += f"\nDeployment Latencies\n{_deployment_latency_map}" - - await self.send_alert( - message=alerting_message + request_info, - level="Medium", - alert_type=AlertType.llm_requests_hanging, - alerting_metadata=alerting_metadata, - ) + await self.hanging_request_check.add_request_to_hanging_request_check( + request_data=request_data + ) async def failed_tracking_alert(self, error_message: str, failing_model: str): """ diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 56f553852f8..5d6d5332858 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -280,6 +280,10 @@ class ProxyLogging: self.db_spend_update_writer = DBSpendUpdateWriter() self.proxy_hook_mapping: Dict[str, CustomLogger] = {} + # Guard flags to prevent duplicate background tasks + self.daily_report_started: bool = False + self.hanging_requests_check_started: bool = False + def startup_event( self, llm_router: Optional[Router], @@ -301,12 +305,25 @@ class ProxyLogging: if ( self.slack_alerting_instance is not None and "daily_reports" in self.slack_alerting_instance.alert_types + and not self.daily_report_started ): asyncio.create_task( self.slack_alerting_instance._run_scheduled_daily_report( llm_router=llm_router ) ) # RUN DAILY REPORT (if scheduled) + self.daily_report_started = True + + if ( + self.slack_alerting_instance is not None + and AlertType.llm_requests_hanging + in self.slack_alerting_instance.alert_types + and not self.hanging_requests_check_started + ): + asyncio.create_task( + self.slack_alerting_instance.hanging_request_check.check_for_hanging_requests() + ) # RUN HANGING REQUEST CHECK (if user wants to alert on hanging requests) + self.hanging_requests_check_started = True def update_values( self, diff --git a/litellm/types/integrations/slack_alerting.py b/litellm/types/integrations/slack_alerting.py index 052fd05ea83..804c41eaf87 100644 --- a/litellm/types/integrations/slack_alerting.py +++ b/litellm/types/integrations/slack_alerting.py @@ -9,6 +9,8 @@ from litellm.types.utils import LiteLLMPydanticObjectBase SLACK_ALERTING_THRESHOLD_5_PERCENT = 0.05 SLACK_ALERTING_THRESHOLD_15_PERCENT = 0.15 +MAX_OLDEST_HANGING_REQUESTS_TO_CHECK = 20 +HANGING_ALERT_BUFFER_TIME_SECONDS = 60 class BaseOutageModel(TypedDict): @@ -187,3 +189,12 @@ DEFAULT_ALERT_TYPES: List[AlertType] = [ # Fallback alerts AlertType.fallback_reports, ] + + +class HangingRequestData(BaseModel): + request_id: str + model: str + api_base: Optional[str] = None + key_alias: Optional[str] = None + team_alias: Optional[str] = None + alerting_metadata: Optional[dict] = None diff --git a/tests/logging_callback_tests/test_alerting.py b/tests/logging_callback_tests/test_alerting.py index ef6b47154bf..e431f99129f 100644 --- a/tests/logging_callback_tests/test_alerting.py +++ b/tests/logging_callback_tests/test_alerting.py @@ -143,22 +143,6 @@ def slack_alerting(): ) -# Test for hanging LLM responses -@pytest.mark.asyncio -async def test_response_taking_too_long_hanging(slack_alerting): - request_data = { - "model": "test_model", - "messages": "test_messages", - "litellm_status": "running", - } - with patch.object(slack_alerting, "send_alert", new=AsyncMock()) as mock_send_alert: - await slack_alerting.response_taking_too_long( - type="hanging_request", request_data=request_data - ) - - mock_send_alert.assert_awaited_once() - - # Test for slow LLM responses @pytest.mark.asyncio async def test_response_taking_too_long_callback(slack_alerting): diff --git a/tests/test_litellm/integrations/SlackAlerting/test_hanging_request_check.py b/tests/test_litellm/integrations/SlackAlerting/test_hanging_request_check.py new file mode 100644 index 00000000000..0bece97b6f0 --- /dev/null +++ b/tests/test_litellm/integrations/SlackAlerting/test_hanging_request_check.py @@ -0,0 +1,262 @@ +import json +import os +import sys +from typing import Optional +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# Adds the grandparent directory to sys.path to allow importing project modules +sys.path.insert(0, os.path.abspath("../..")) + +from litellm.integrations.SlackAlerting.hanging_request_check import ( + AlertingHangingRequestCheck, +) +from litellm.types.integrations.slack_alerting import HangingRequestData + + +class TestAlertingHangingRequestCheck: + """Test suite for AlertingHangingRequestCheck class""" + + @pytest.fixture + def mock_slack_alerting(self): + """Create a mock SlackAlerting object for testing""" + mock_slack = MagicMock() + mock_slack.alerting_threshold = 300 # 5 minutes + mock_slack.send_alert = AsyncMock() + return mock_slack + + @pytest.fixture + def hanging_request_checker(self, mock_slack_alerting): + """Create an AlertingHangingRequestCheck instance for testing""" + return AlertingHangingRequestCheck(slack_alerting_object=mock_slack_alerting) + + @pytest.mark.asyncio + async def test_init_creates_cache_with_correct_ttl(self, mock_slack_alerting): + """ + Test that initialization creates a hanging request cache with correct TTL. + The TTL should be alerting_threshold + buffer time. + """ + checker = AlertingHangingRequestCheck(slack_alerting_object=mock_slack_alerting) + + # The cache should be created with TTL = alerting_threshold + buffer time + expected_ttl = ( + mock_slack_alerting.alerting_threshold + 60 + ) # HANGING_ALERT_BUFFER_TIME_SECONDS + assert checker.hanging_request_cache.default_ttl == expected_ttl + + @pytest.mark.asyncio + async def test_add_request_to_hanging_request_check_success( + self, hanging_request_checker + ): + """ + Test successfully adding a request to the hanging request cache. + Should extract metadata and store HangingRequestData in cache. + """ + request_data = { + "litellm_call_id": "test_request_123", + "model": "gpt-4", + "deployment": {"litellm_params": {"api_base": "https://api.openai.com/v1"}}, + "metadata": { + "user_api_key_alias": "test_key", + "user_api_key_team_alias": "test_team", + }, + } + + with patch("litellm.get_api_base", return_value="https://api.openai.com/v1"): + await hanging_request_checker.add_request_to_hanging_request_check( + request_data + ) + + # Verify the request was added to cache + cached_data = ( + await hanging_request_checker.hanging_request_cache.async_get_cache( + key="test_request_123" + ) + ) + + assert cached_data is not None + assert isinstance(cached_data, HangingRequestData) + assert cached_data.request_id == "test_request_123" + assert cached_data.model == "gpt-4" + assert cached_data.api_base == "https://api.openai.com/v1" + + @pytest.mark.asyncio + async def test_add_request_to_hanging_request_check_none_request_data( + self, hanging_request_checker + ): + """ + Test that passing None request_data returns early without error. + Should handle gracefully when no request data is provided. + """ + result = await hanging_request_checker.add_request_to_hanging_request_check( + None + ) + assert result is None + + @pytest.mark.asyncio + async def test_add_request_to_hanging_request_check_minimal_data( + self, hanging_request_checker + ): + """ + Test adding request with minimal required data. + Should handle cases where optional fields are missing. + """ + request_data = { + "litellm_call_id": "minimal_request_456", + "model": "gpt-3.5-turbo", + } + + await hanging_request_checker.add_request_to_hanging_request_check(request_data) + + cached_data = ( + await hanging_request_checker.hanging_request_cache.async_get_cache( + key="minimal_request_456" + ) + ) + + assert cached_data is not None + assert cached_data.request_id == "minimal_request_456" + assert cached_data.model == "gpt-3.5-turbo" + assert cached_data.api_base is None + assert cached_data.key_alias == "" + assert cached_data.team_alias == "" + + @pytest.mark.asyncio + async def test_send_hanging_request_alert(self, hanging_request_checker): + """ + Test sending a hanging request alert. + Should format the alert message correctly and call slack alerting. + """ + hanging_request_data = HangingRequestData( + request_id="test_hanging_request", + model="gpt-4", + api_base="https://api.openai.com/v1", + key_alias="test_key", + team_alias="test_team", + ) + + await hanging_request_checker.send_hanging_request_alert(hanging_request_data) + + # Verify slack alert was called + hanging_request_checker.slack_alerting_object.send_alert.assert_called_once() + + # Check the alert message format + call_args = hanging_request_checker.slack_alerting_object.send_alert.call_args + message = call_args[1]["message"] + + assert "Requests are hanging - 300s+ request time" in message + assert "Request Model: `gpt-4`" in message + assert "API Base: `https://api.openai.com/v1`" in message + assert "Key Alias: `test_key`" in message + assert "Team Alias: `test_team`" in message + assert call_args[1]["level"] == "Medium" + + @pytest.mark.asyncio + async def test_send_alerts_for_hanging_requests_no_proxy_logging( + self, hanging_request_checker + ): + """ + Test send_alerts_for_hanging_requests when proxy_logging_obj.internal_usage_cache is None. + Should return early without processing when internal usage cache is unavailable. + """ + with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy: + mock_proxy.internal_usage_cache = None + + result = await hanging_request_checker.send_alerts_for_hanging_requests() + assert result is None + + @pytest.mark.asyncio + async def test_send_alerts_for_hanging_requests_with_completed_request( + self, hanging_request_checker + ): + """ + Test send_alerts_for_hanging_requests when request has completed (not hanging). + Should remove completed requests from cache and not send alerts. + """ + # Add a request to the hanging cache + hanging_data = HangingRequestData( + request_id="completed_request_789", + model="gpt-4", + api_base="https://api.openai.com/v1", + ) + await hanging_request_checker.hanging_request_cache.async_set_cache( + key="completed_request_789", value=hanging_data, ttl=300 + ) + + with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy: + # Mock internal usage cache to return a request status (meaning request completed) + mock_internal_cache = AsyncMock() + mock_internal_cache.async_get_cache.return_value = {"status": "success"} + mock_proxy.internal_usage_cache = mock_internal_cache + + # Mock the cache method to return our test request + hanging_request_checker.hanging_request_cache.async_get_oldest_n_keys = ( + AsyncMock(return_value=["completed_request_789"]) + ) + + await hanging_request_checker.send_alerts_for_hanging_requests() + + # Verify no alert was sent since request completed + hanging_request_checker.slack_alerting_object.send_alert.assert_not_called() + + @pytest.mark.asyncio + async def test_send_alerts_for_hanging_requests_with_actual_hanging_request( + self, hanging_request_checker + ): + """ + Test send_alerts_for_hanging_requests when request is actually hanging. + Should send alert for requests that haven't completed within threshold. + """ + # Add a hanging request to the cache + hanging_data = HangingRequestData( + request_id="hanging_request_999", + model="gpt-4", + api_base="https://api.openai.com/v1", + key_alias="test_key", + team_alias="test_team", + ) + await hanging_request_checker.hanging_request_cache.async_set_cache( + key="hanging_request_999", value=hanging_data, ttl=300 + ) + + with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy: + # Mock internal usage cache to return None (meaning request is still hanging) + mock_internal_cache = AsyncMock() + mock_internal_cache.async_get_cache.return_value = None + mock_proxy.internal_usage_cache = mock_internal_cache + + # Mock the cache method to return our test request + hanging_request_checker.hanging_request_cache.async_get_oldest_n_keys = ( + AsyncMock(return_value=["hanging_request_999"]) + ) + + await hanging_request_checker.send_alerts_for_hanging_requests() + + # Verify alert was sent for hanging request + hanging_request_checker.slack_alerting_object.send_alert.assert_called_once() + + @pytest.mark.asyncio + async def test_send_alerts_for_hanging_requests_with_missing_hanging_data( + self, hanging_request_checker + ): + """ + Test send_alerts_for_hanging_requests when hanging request data is missing from cache. + Should continue processing other requests when individual request data is missing. + """ + with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy: + mock_internal_cache = AsyncMock() + mock_proxy.internal_usage_cache = mock_internal_cache + + # Mock cache to return request ID but no data (simulating expired or missing data) + hanging_request_checker.hanging_request_cache.async_get_oldest_n_keys = ( + AsyncMock(return_value=["missing_request_111"]) + ) + hanging_request_checker.hanging_request_cache.async_get_cache = AsyncMock( + return_value=None + ) + + await hanging_request_checker.send_alerts_for_hanging_requests() + + # Should not crash and should not send any alerts + hanging_request_checker.slack_alerting_object.send_alert.assert_not_called()