From 37069f45cbc240ff7bb097f45e6b96fc464e7a2d Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 23 Apr 2024 18:31:15 -0700 Subject: [PATCH 1/5] fix - use one slack class --- litellm/integrations/slack_alerting.py | 419 +++++++++++++++++++++++++ 1 file changed, 419 insertions(+) create mode 100644 litellm/integrations/slack_alerting.py diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py new file mode 100644 index 00000000000..e7e92bcb543 --- /dev/null +++ b/litellm/integrations/slack_alerting.py @@ -0,0 +1,419 @@ +#### What this does #### +# Class for sending Slack Alerts # +import dotenv, os + +dotenv.load_dotenv() # Loading env variables using dotenv +import copy +import traceback +from litellm._logging import verbose_logger, verbose_proxy_logger +import litellm +from typing import List, Literal, Any, Union, Optional +from litellm.caching import DualCache +import asyncio +import aiohttp + + +class SlackAlerting: + # Class variables or attributes + def __init__( + self, + alerting_threshold: float = 300, + alerting: Optional[List] = [], + alert_types: Optional[ + List[ + Literal[ + "llm_exceptions", + "llm_too_slow", + "llm_requests_hanging", + "budget_alerts", + "db_exceptions", + ] + ] + ] = [ + "llm_exceptions", + "llm_too_slow", + "llm_requests_hanging", + "budget_alerts", + "db_exceptions", + ], + ): + self.alerting_threshold = alerting_threshold + self.alerting = alerting + self.alert_types = alert_types + self.internal_usage_cache = DualCache() + + pass + + def update_values( + self, + alerting: Optional[List] = None, + alerting_threshold: Optional[float] = None, + alert_types: Optional[List] = None, + ): + if alerting is not None: + self.alerting = alerting + if alerting_threshold is not None: + self.alerting_threshold = alerting_threshold + if alert_types is not None: + self.alert_types = alert_types + + async def deployment_in_cooldown(self): + pass + + async def deployment_removed_from_cooldown(self): + pass + + def _all_possible_alert_types(self): + # used by the UI to show all supported alert types + # Note: This is not the alerts the user has configured, instead it's all possible alert types a user can select + return [ + "llm_exceptions", + "llm_too_slow", + "llm_requests_hanging", + "budget_alerts", + "db_exceptions", + ] + + def _add_langfuse_trace_id_to_alert( + self, + request_info: str, + request_data: Optional[dict] = None, + kwargs: Optional[dict] = None, + ): + import uuid + + if request_data is not None: + trace_id = request_data.get("metadata", {}).get( + "trace_id", None + ) # get langfuse trace id + if trace_id is None: + trace_id = "litellm-alert-trace-" + str(uuid.uuid4()) + request_data["metadata"]["trace_id"] = trace_id + elif kwargs is not None: + _litellm_params = kwargs.get("litellm_params", {}) + trace_id = _litellm_params.get("metadata", {}).get( + "trace_id", None + ) # get langfuse trace id + if trace_id is None: + trace_id = "litellm-alert-trace-" + str(uuid.uuid4()) + _litellm_params["metadata"]["trace_id"] = trace_id + + _langfuse_host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com") + _langfuse_project_id = os.environ.get("LANGFUSE_PROJECT_ID") + + # langfuse urls look like: https://us.cloud.langfuse.com/project/************/traces/litellm-alert-trace-ididi9dk-09292-************ + + _langfuse_url = ( + f"{_langfuse_host}/project/{_langfuse_project_id}/traces/{trace_id}" + ) + request_info += f"\n🪢 Langfuse Trace: {_langfuse_url}" + return request_info + + def _response_taking_too_long_callback( + self, + kwargs, # kwargs to completion + start_time, + end_time, # start/end time + ): + try: + time_difference = end_time - start_time + # Convert the timedelta to float (in seconds) + time_difference_float = time_difference.total_seconds() + litellm_params = kwargs.get("litellm_params", {}) + model = kwargs.get("model", "") + api_base = litellm.get_api_base(model=model, optional_params=litellm_params) + messages = kwargs.get("messages", None) + # if messages does not exist fallback to "input" + if messages is None: + messages = kwargs.get("input", None) + + # only use first 100 chars for alerting + _messages = str(messages)[:100] + + return time_difference_float, model, api_base, _messages + except Exception as e: + raise e + + async def response_taking_too_long_callback( + self, + kwargs, # kwargs to completion + completion_response, # response from completion + start_time, + end_time, # start/end time + ): + if self.alerting is None or self.alert_types is None: + return + + if "llm_too_slow" not in self.alert_types: + return + time_difference_float, model, api_base, messages = ( + self._response_taking_too_long_callback( + kwargs=kwargs, + start_time=start_time, + end_time=end_time, + ) + ) + request_info = f"\nRequest Model: `{model}`\nAPI Base: `{api_base}`\nMessages: `{messages}`" + slow_message = f"`Responses are slow - {round(time_difference_float,2)}s response time > Alerting threshold: {self.alerting_threshold}s`" + if time_difference_float > self.alerting_threshold: + if "langfuse" in litellm.success_callback: + request_info = self._add_langfuse_trace_id_to_alert( + request_info=request_info, kwargs=kwargs + ) + await self.send_alert( + message=slow_message + request_info, + level="Low", + ) + + async def log_failure_event(self, original_exception: Exception): + pass + + async def response_taking_too_long( + self, + start_time: Optional[float] = None, + end_time: Optional[float] = 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 + 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: + messages = "" + request_info = f"\nRequest Model: `{model}`\nMessages: `{messages}`" + if "langfuse" in litellm.success_callback: + request_info = self._add_langfuse_trace_id_to_alert( + request_info=request_info, request_data=request_data + ) + else: + request_info = "" + + if type == "hanging_request": + # Simulate a long-running operation that could take more than 5 minutes + if "llm_requests_hanging" not in self.alert_types: + return + 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 + if ( + request_data is not None + and request_data.get("litellm_status", "") != "success" + and request_data.get("litellm_status", "") != "fail" + ): + 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 = request_data["metadata"] + _api_base = _metadata.get("api_base", "") + if _api_base is None: + _api_base = "" + 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`" + ) + await self.send_alert( + message=alerting_message + request_info, + level="Medium", + ) + + async def budget_alerts( + self, + type: Literal[ + "token_budget", + "user_budget", + "user_and_proxy_budget", + "failed_budgets", + "failed_tracking", + "projected_limit_exceeded", + ], + user_max_budget: float, + user_current_spend: float, + user_info=None, + error_message="", + ): + if self.alerting is None or self.alert_types is None: + # do nothing if alerting is not switched on + return + if "budget_alerts" not in self.alert_types: + return + _id: str = "default_id" # used for caching + if type == "user_and_proxy_budget": + user_info = dict(user_info) + user_id = user_info["user_id"] + _id = user_id + max_budget = user_info["max_budget"] + spend = user_info["spend"] + user_email = user_info["user_email"] + user_info = f"""\nUser ID: {user_id}\nMax Budget: ${max_budget}\nSpend: ${spend}\nUser Email: {user_email}""" + elif type == "token_budget": + token_info = dict(user_info) + token = token_info["token"] + _id = token + spend = token_info["spend"] + max_budget = token_info["max_budget"] + user_id = token_info["user_id"] + user_info = f"""\nToken: {token}\nSpend: ${spend}\nMax Budget: ${max_budget}\nUser ID: {user_id}""" + elif type == "failed_tracking": + user_id = str(user_info) + _id = user_id + user_info = f"\nUser ID: {user_id}\n Error {error_message}" + message = "Failed Tracking Cost for" + user_info + await self.send_alert( + message=message, + level="High", + ) + return + elif type == "projected_limit_exceeded" and user_info is not None: + """ + Input variables: + user_info = { + "key_alias": key_alias, + "projected_spend": projected_spend, + "projected_exceeded_date": projected_exceeded_date, + } + user_max_budget=soft_limit, + user_current_spend=new_spend + """ + message = f"""\n🚨 `ProjectedLimitExceededError` 💸\n\n`Key Alias:` {user_info["key_alias"]} \n`Expected Day of Error`: {user_info["projected_exceeded_date"]} \n`Current Spend`: {user_current_spend} \n`Projected Spend at end of month`: {user_info["projected_spend"]} \n`Soft Limit`: {user_max_budget}""" + await self.send_alert( + message=message, + level="High", + ) + return + else: + user_info = str(user_info) + + # percent of max_budget left to spend + if user_max_budget > 0: + percent_left = (user_max_budget - user_current_spend) / user_max_budget + else: + percent_left = 0 + verbose_proxy_logger.debug( + f"Budget Alerts: Percent left: {percent_left} for {user_info}" + ) + + ## PREVENTITIVE ALERTING ## - https://github.com/BerriAI/litellm/issues/2727 + # - Alert once within 28d period + # - Cache this information + # - Don't re-alert, if alert already sent + _cache: DualCache = self.internal_usage_cache + + # check if crossed budget + if user_current_spend >= user_max_budget: + verbose_proxy_logger.debug("Budget Crossed for %s", user_info) + message = "Budget Crossed for" + user_info + result = await _cache.async_get_cache(key=message) + if result is None: + await self.send_alert( + message=message, + level="High", + ) + await _cache.async_set_cache(key=message, value="SENT", ttl=2419200) + return + + # check if 5% of max budget is left + if percent_left <= 0.05: + message = "5% budget left for" + user_info + cache_key = "alerting:{}".format(_id) + result = await _cache.async_get_cache(key=cache_key) + if result is None: + await self.send_alert( + message=message, + level="Medium", + ) + + await _cache.async_set_cache(key=cache_key, value="SENT", ttl=2419200) + + return + + # check if 15% of max budget is left + if percent_left <= 0.15: + message = "15% budget left for" + user_info + result = await _cache.async_get_cache(key=message) + if result is None: + await self.send_alert( + message=message, + level="Low", + ) + await _cache.async_set_cache(key=message, value="SENT", ttl=2419200) + return + + return + + async def send_alert(self, message: str, level: Literal["Low", "Medium", "High"]): + """ + Alerting based on thresholds: - https://github.com/BerriAI/litellm/issues/1298 + + - Responses taking too long + - Requests are hanging + - Calls are failing + - DB Read/Writes are failing + - Proxy Close to max budget + - Key Close to max budget + + Parameters: + level: str - Low|Medium|High - if calls might fail (Medium) or are failing (High); Currently, no alerts would be 'Low'. + message: str - what is the alert about + """ + print( + "inside send alert for slack, message: ", + message, + "self.alerting: ", + self.alerting, + ) + if self.alerting is None: + return + + from datetime import datetime + + # Get the current timestamp + current_time = datetime.now().strftime("%H:%M:%S") + _proxy_base_url = os.getenv("PROXY_BASE_URL", None) + formatted_message = ( + f"Level: `{level}`\nTimestamp: `{current_time}`\n\nMessage: {message}" + ) + if _proxy_base_url is not None: + formatted_message += f"\n\nProxy URL: `{_proxy_base_url}`" + + slack_webhook_url = os.getenv("SLACK_WEBHOOK_URL", None) + if slack_webhook_url is None: + raise Exception("Missing SLACK_WEBHOOK_URL from environment") + payload = {"text": formatted_message} + headers = {"Content-type": "application/json"} + async with aiohttp.ClientSession( + connector=aiohttp.TCPConnector(ssl=False) + ) as session: + async with session.post( + slack_webhook_url, json=payload, headers=headers + ) as response: + if response.status == 200: + pass + else: + print("Error sending slack alert. Error=", response.text) From 31979f83d26d568fd5f80a719d12d4146b1a3799 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 23 Apr 2024 18:34:11 -0700 Subject: [PATCH 2/5] fix - refactor slack alerting --- litellm/proxy/utils.py | 325 ++++------------------------------------- 1 file changed, 29 insertions(+), 296 deletions(-) diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index c2bd0a4f00a..5d9940b2da4 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -31,6 +31,7 @@ import smtplib, re from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime, timedelta +from litellm.integrations.slack_alerting import SlackAlerting def print_verbose(print_statement): @@ -80,17 +81,11 @@ class ProxyLogging: "budget_alerts", "db_exceptions", ] - - def _all_possible_alert_types(self): - # used by the UI to show all supported alert types - # Note: This is not the alerts the user has configured, instead it's all possible alert types a user can select - return [ - "llm_exceptions", - "llm_too_slow", - "llm_requests_hanging", - "budget_alerts", - "db_exceptions", - ] + self.slack_alerting_instance = SlackAlerting( + alerting_threshold=self.alerting_threshold, + alerting=self.alerting, + alert_types=self.alert_types, + ) def update_values( self, @@ -112,13 +107,18 @@ class ProxyLogging: self.alerting = alerting if alerting_threshold is not None: self.alerting_threshold = alerting_threshold + if alert_types is not None: + self.alert_types = alert_types + + self.slack_alerting_instance.update_values( + alerting=self.alerting, + alerting_threshold=self.alerting_threshold, + alert_types=self.alert_types, + ) if redis_cache is not None: self.internal_usage_cache.redis_cache = redis_cache - if alert_types is not None: - self.alert_types = alert_types - def _init_litellm_callbacks(self): print_verbose(f"INITIALIZING LITELLM CALLBACKS!") self.service_logging_obj = ServiceLogging() @@ -127,7 +127,9 @@ class ProxyLogging: litellm.callbacks.append(self.max_budget_limiter) litellm.callbacks.append(self.cache_control_check) litellm.callbacks.append(self.service_logging_obj) - litellm.success_callback.append(self.response_taking_too_long_callback) + litellm.success_callback.append( + self.slack_alerting_instance.response_taking_too_long_callback + ) for callback in litellm.callbacks: if callback not in litellm.input_callback: litellm.input_callback.append(callback) @@ -176,7 +178,9 @@ class ProxyLogging: """ print_verbose(f"Inside Proxy Logging Pre-call hook!") ### ALERTING ### - asyncio.create_task(self.response_taking_too_long(request_data=data)) + asyncio.create_task( + self.slack_alerting_instance.response_taking_too_long(request_data=data) + ) try: for callback in litellm.callbacks: @@ -225,169 +229,6 @@ class ProxyLogging: raise e return data - def _response_taking_too_long_callback( - self, - kwargs, # kwargs to completion - start_time, - end_time, # start/end time - ): - try: - time_difference = end_time - start_time - # Convert the timedelta to float (in seconds) - time_difference_float = time_difference.total_seconds() - litellm_params = kwargs.get("litellm_params", {}) - model = kwargs.get("model", "") - api_base = litellm.get_api_base(model=model, optional_params=litellm_params) - messages = kwargs.get("messages", None) - # if messages does not exist fallback to "input" - if messages is None: - messages = kwargs.get("input", None) - - # only use first 100 chars for alerting - _messages = str(messages)[:100] - - return time_difference_float, model, api_base, _messages - except Exception as e: - raise e - - async def response_taking_too_long_callback( - self, - kwargs, # kwargs to completion - completion_response, # response from completion - start_time, - end_time, # start/end time - ): - if self.alerting is None: - return - if "llm_too_slow" not in self.alert_types: - return - time_difference_float, model, api_base, messages = ( - self._response_taking_too_long_callback( - kwargs=kwargs, - start_time=start_time, - end_time=end_time, - ) - ) - request_info = f"\nRequest Model: `{model}`\nAPI Base: `{api_base}`\nMessages: `{messages}`" - slow_message = f"`Responses are slow - {round(time_difference_float,2)}s response time > Alerting threshold: {self.alerting_threshold}s`" - if time_difference_float > self.alerting_threshold: - if "langfuse" in litellm.success_callback: - request_info = self._add_langfuse_trace_id_to_alert( - request_info=request_info, kwargs=kwargs - ) - await self.alerting_handler( - message=slow_message + request_info, - level="Low", - ) - - def _add_langfuse_trace_id_to_alert( - self, - request_info: str, - request_data: Optional[dict] = None, - kwargs: Optional[dict] = None, - ): - import uuid - - if request_data is not None: - trace_id = request_data.get("metadata", {}).get( - "trace_id", None - ) # get langfuse trace id - if trace_id is None: - trace_id = "litellm-alert-trace-" + str(uuid.uuid4()) - request_data["metadata"]["trace_id"] = trace_id - elif kwargs is not None: - _litellm_params = kwargs.get("litellm_params", {}) - trace_id = _litellm_params.get("metadata", {}).get( - "trace_id", None - ) # get langfuse trace id - if trace_id is None: - trace_id = "litellm-alert-trace-" + str(uuid.uuid4()) - _litellm_params["metadata"]["trace_id"] = trace_id - - _langfuse_host = os.environ.get("LANGFUSE_HOST", "https://cloud.langfuse.com") - _langfuse_project_id = os.environ.get("LANGFUSE_PROJECT_ID") - - # langfuse urls look like: https://us.cloud.langfuse.com/project/************/traces/litellm-alert-trace-ididi9dk-09292-************ - - _langfuse_url = ( - f"{_langfuse_host}/project/{_langfuse_project_id}/traces/{trace_id}" - ) - request_info += f"\n🪢 Langfuse Trace: {_langfuse_url}" - return request_info - - async def response_taking_too_long( - self, - start_time: Optional[float] = None, - end_time: Optional[float] = None, - type: Literal["hanging_request", "slow_response"] = "hanging_request", - request_data: Optional[dict] = None, - ): - 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: - messages = "" - request_info = f"\nRequest Model: `{model}`\nMessages: `{messages}`" - if "langfuse" in litellm.success_callback: - request_info = self._add_langfuse_trace_id_to_alert( - request_info=request_info, request_data=request_data - ) - else: - request_info = "" - - if type == "hanging_request": - # Simulate a long-running operation that could take more than 5 minutes - if "llm_requests_hanging" not in self.alert_types: - return - 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 - if ( - request_data is not None - and request_data.get("litellm_status", "") != "success" - and request_data.get("litellm_status", "") != "fail" - ): - 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 = request_data["metadata"] - _api_base = _metadata.get("api_base", "") - if _api_base is None: - _api_base = "" - 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`" - ) - await self.alerting_handler( - message=alerting_message + request_info, - level="Medium", - ) - async def budget_alerts( self, type: Literal[ @@ -406,112 +247,14 @@ class ProxyLogging: if self.alerting is None: # do nothing if alerting is not switched on return - if "budget_alerts" not in self.alert_types: - return - _id: str = "default_id" # used for caching - if type == "user_and_proxy_budget": - user_info = dict(user_info) - user_id = user_info["user_id"] - _id = user_id - max_budget = user_info["max_budget"] - spend = user_info["spend"] - user_email = user_info["user_email"] - user_info = f"""\nUser ID: {user_id}\nMax Budget: ${max_budget}\nSpend: ${spend}\nUser Email: {user_email}""" - elif type == "token_budget": - token_info = dict(user_info) - token = token_info["token"] - _id = token - spend = token_info["spend"] - max_budget = token_info["max_budget"] - user_id = token_info["user_id"] - user_info = f"""\nToken: {token}\nSpend: ${spend}\nMax Budget: ${max_budget}\nUser ID: {user_id}""" - elif type == "failed_tracking": - user_id = str(user_info) - _id = user_id - user_info = f"\nUser ID: {user_id}\n Error {error_message}" - message = "Failed Tracking Cost for" + user_info - await self.alerting_handler( - message=message, - level="High", - ) - return - elif type == "projected_limit_exceeded" and user_info is not None: - """ - Input variables: - user_info = { - "key_alias": key_alias, - "projected_spend": projected_spend, - "projected_exceeded_date": projected_exceeded_date, - } - user_max_budget=soft_limit, - user_current_spend=new_spend - """ - message = f"""\n🚨 `ProjectedLimitExceededError` 💸\n\n`Key Alias:` {user_info["key_alias"]} \n`Expected Day of Error`: {user_info["projected_exceeded_date"]} \n`Current Spend`: {user_current_spend} \n`Projected Spend at end of month`: {user_info["projected_spend"]} \n`Soft Limit`: {user_max_budget}""" - await self.alerting_handler( - message=message, - level="High", - ) - return - else: - user_info = str(user_info) - - # percent of max_budget left to spend - if user_max_budget > 0: - percent_left = (user_max_budget - user_current_spend) / user_max_budget - else: - percent_left = 0 - verbose_proxy_logger.debug( - f"Budget Alerts: Percent left: {percent_left} for {user_info}" + await self.slack_alerting_instance.budget_alerts( + type=type, + user_max_budget=user_max_budget, + user_current_spend=user_current_spend, + user_info=user_info, + error_message=error_message, ) - ## PREVENTITIVE ALERTING ## - https://github.com/BerriAI/litellm/issues/2727 - # - Alert once within 28d period - # - Cache this information - # - Don't re-alert, if alert already sent - _cache: DualCache = self.internal_usage_cache - - # check if crossed budget - if user_current_spend >= user_max_budget: - verbose_proxy_logger.debug("Budget Crossed for %s", user_info) - message = "Budget Crossed for" + user_info - result = await _cache.async_get_cache(key=message) - if result is None: - await self.alerting_handler( - message=message, - level="High", - ) - await _cache.async_set_cache(key=message, value="SENT", ttl=2419200) - return - - # check if 5% of max budget is left - if percent_left <= 0.05: - message = "5% budget left for" + user_info - cache_key = "alerting:{}".format(_id) - result = await _cache.async_get_cache(key=cache_key) - if result is None: - await self.alerting_handler( - message=message, - level="Medium", - ) - - await _cache.async_set_cache(key=cache_key, value="SENT", ttl=2419200) - - return - - # check if 15% of max budget is left - if percent_left <= 0.15: - message = "15% budget left for" + user_info - result = await _cache.async_get_cache(key=message) - if result is None: - await self.alerting_handler( - message=message, - level="Low", - ) - await _cache.async_set_cache(key=message, value="SENT", ttl=2419200) - return - - return - async def alerting_handler( self, message: str, level: Literal["Low", "Medium", "High"] ): @@ -545,19 +288,9 @@ class ProxyLogging: for client in self.alerting: if client == "slack": - slack_webhook_url = os.getenv("SLACK_WEBHOOK_URL", None) - if slack_webhook_url is None: - raise Exception("Missing SLACK_WEBHOOK_URL from environment") - payload = {"text": formatted_message} - headers = {"Content-type": "application/json"} - async with aiohttp.ClientSession( - connector=aiohttp.TCPConnector(ssl=False) - ) as session: - async with session.post( - slack_webhook_url, json=payload, headers=headers - ) as response: - if response.status == 200: - pass + await self.slack_alerting_instance.send_alert( + message=message, level=level + ) elif client == "sentry": if litellm.utils.sentry_sdk_instance is not None: litellm.utils.sentry_sdk_instance.capture_message(formatted_message) From bfc84a7e81e4c61a99bb5af79eaf1511a09d8688 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 23 Apr 2024 18:47:39 -0700 Subject: [PATCH 3/5] fix - slack alerting use one httpx client for alerts --- litellm/integrations/slack_alerting.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/litellm/integrations/slack_alerting.py b/litellm/integrations/slack_alerting.py index e7e92bcb543..a54b825e8a8 100644 --- a/litellm/integrations/slack_alerting.py +++ b/litellm/integrations/slack_alerting.py @@ -11,6 +11,7 @@ from typing import List, Literal, Any, Union, Optional from litellm.caching import DualCache import asyncio import aiohttp +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler class SlackAlerting: @@ -41,6 +42,7 @@ class SlackAlerting: self.alerting = alerting self.alert_types = alert_types self.internal_usage_cache = DualCache() + self.async_http_handler = AsyncHTTPHandler() pass @@ -392,6 +394,7 @@ class SlackAlerting: return from datetime import datetime + import json # Get the current timestamp current_time = datetime.now().strftime("%H:%M:%S") @@ -407,13 +410,13 @@ class SlackAlerting: raise Exception("Missing SLACK_WEBHOOK_URL from environment") payload = {"text": formatted_message} headers = {"Content-type": "application/json"} - async with aiohttp.ClientSession( - connector=aiohttp.TCPConnector(ssl=False) - ) as session: - async with session.post( - slack_webhook_url, json=payload, headers=headers - ) as response: - if response.status == 200: - pass - else: - print("Error sending slack alert. Error=", response.text) + + response = await self.async_http_handler.post( + url=slack_webhook_url, + headers=headers, + data=json.dumps(payload), + ) + if response.status_code == 200: + pass + else: + print("Error sending slack alert. Error=", response.text) # noqa From 9d18e4770d64347f8b7fe24dc8b006ac8d4c3ace Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 23 Apr 2024 19:05:50 -0700 Subject: [PATCH 4/5] fix using slack alerting through admin ui --- litellm/proxy/proxy_server.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 8bb2c2d475a..568121dbad1 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -2624,9 +2624,16 @@ class ProxyConfig: if "alerting" in _general_settings: general_settings["alerting"] = _general_settings["alerting"] proxy_logging_obj.alerting = general_settings["alerting"] + proxy_logging_obj.slack_alerting_instance.alerting = general_settings[ + "alerting" + ] + if "alert_types" in _general_settings: general_settings["alert_types"] = _general_settings["alert_types"] proxy_logging_obj.alert_types = general_settings["alert_types"] + proxy_logging_obj.slack_alerting_instance.alert_types = general_settings[ + "alert_types" + ] # router settings if llm_router is not None: @@ -8470,8 +8477,10 @@ async def get_config(): ) _slack_env_vars[_var] = _decrypted_value - _alerting_types = proxy_logging_obj.alert_types - _all_alert_types = proxy_logging_obj._all_possible_alert_types() + _alerting_types = proxy_logging_obj.slack_alerting_instance.alert_types + _all_alert_types = ( + proxy_logging_obj.slack_alerting_instance._all_possible_alert_types() + ) _data_to_return.append( { "name": "slack", From c1ea3597bcb7cb3bca6182893674fb1c2250c286 Mon Sep 17 00:00:00 2001 From: Ishaan Jaff Date: Tue, 23 Apr 2024 19:10:06 -0700 Subject: [PATCH 5/5] fix test alerting --- litellm/tests/test_alerting.py | 32 ++++++++++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/litellm/tests/test_alerting.py b/litellm/tests/test_alerting.py index 6cb0e29edc4..311b803110c 100644 --- a/litellm/tests/test_alerting.py +++ b/litellm/tests/test_alerting.py @@ -4,7 +4,7 @@ import sys import os import io, asyncio -from datetime import datetime +from datetime import datetime, timedelta # import logging # logging.basicConfig(level=logging.DEBUG) @@ -13,6 +13,10 @@ from litellm.proxy.utils import ProxyLogging from litellm.caching import DualCache import litellm import pytest +import asyncio +from unittest.mock import patch, MagicMock +from litellm.caching import DualCache +from litellm.integrations.slack_alerting import SlackAlerting @pytest.mark.asyncio @@ -43,7 +47,7 @@ async def test_get_api_base(): end_time = datetime.now() time_difference_float, model, api_base, messages = ( - _pl._response_taking_too_long_callback( + _pl.slack_alerting_instance._response_taking_too_long_callback( kwargs={ "model": model, "messages": messages, @@ -65,3 +69,27 @@ async def test_get_api_base(): message=slow_message + request_info, level="Low", ) + print("passed test_get_api_base") + + +# Create a mock environment for testing +@pytest.fixture +def mock_env(monkeypatch): + monkeypatch.setenv("SLACK_WEBHOOK_URL", "https://example.com/webhook") + monkeypatch.setenv("LANGFUSE_HOST", "https://cloud.langfuse.com") + monkeypatch.setenv("LANGFUSE_PROJECT_ID", "test-project-id") + + +# Test the __init__ method +def test_init(): + slack_alerting = SlackAlerting( + alerting_threshold=32, alerting=["slack"], alert_types=["llm_exceptions"] + ) + assert slack_alerting.alerting_threshold == 32 + assert slack_alerting.alerting == ["slack"] + assert slack_alerting.alert_types == ["llm_exceptions"] + + slack_no_alerting = SlackAlerting() + assert slack_no_alerting.alerting == [] + + print("passed testing slack alerting init")