mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
[Feat] Performance - Don't create 1 task for every hanging request alert (#11385)
* feat: add async_get_oldest_n_keys in memory cache * fix: add add_request_to_hanging_request_check * test: alerting * feat: v2 hanging request check * fix: HangingRequestData * fix: AlertingHangingRequestCheck * fix: check_for_hanging_requests * fix: use correct metadata location for hanging requests * fix: formatting alert * test hanging request check * fix: add guard flags for background tasks alerting
This commit is contained in:
Ishaan Jaff
parent
a817ebc5bb
commit
27cb52ae3a
7 changed files with 487 additions and 113 deletions
|
|
@ -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]]
|
||||
|
|
|
|||
175
litellm/integrations/SlackAlerting/hanging_request_check.py
Normal file
175
litellm/integrations/SlackAlerting/hanging_request_check.py
Normal file
|
|
@ -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 {},
|
||||
)
|
||||
|
|
@ -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):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
Loading…
Add table
Reference in a new issue