diff --git a/litellm/integrations/SlackAlerting/slack_alerting.py b/litellm/integrations/SlackAlerting/slack_alerting.py index 65f4774a693..7b46616c360 100644 --- a/litellm/integrations/SlackAlerting/slack_alerting.py +++ b/litellm/integrations/SlackAlerting/slack_alerting.py @@ -106,6 +106,7 @@ class SlackAlerting(CustomBatchLogger): self.default_webhook_url = default_webhook_url self.flush_lock = asyncio.Lock() self.periodic_started = False + self._periodic_flush_task: asyncio.Task | None = None self.hanging_request_check = AlertingHangingRequestCheck( slack_alerting_object=self, ) @@ -117,6 +118,20 @@ class SlackAlerting(CustomBatchLogger): self.digest_lock = asyncio.Lock() super().__init__(**kwargs, flush_lock=self.flush_lock) + def _ensure_periodic_flush_started(self) -> None: + """Start the periodic flush loop, at most once. + + ``update_values`` is re-invoked on a timer by the proxy's deployment + refresh, so creating the task unconditionally leaks one ``while True`` + task per call for the lifetime of the process. + """ + if self.periodic_started: + return + # Keep a reference: a bare create_task may be garbage collected + # mid-execution, and this is now the only flush task. + self._periodic_flush_task = asyncio.create_task(self.periodic_flush()) + self.periodic_started = True + def update_values( self, alerting: list | None = None, @@ -129,17 +144,14 @@ class SlackAlerting(CustomBatchLogger): ): if alerting is not None: self.alerting = alerting - asyncio.create_task(self.periodic_flush()) - self.periodic_started = True + self._ensure_periodic_flush_started() if alerting_threshold is not None: self.alerting_threshold = alerting_threshold if alert_types is not None: self.alert_types = alert_types if alerting_args is not None: self.alerting_args = SlackAlertingArgs(**alerting_args) - if not self.periodic_started: - asyncio.create_task(self.periodic_flush()) - self.periodic_started = True + self._ensure_periodic_flush_started() if alert_type_config is not None: for key, val in alert_type_config.items(): self.alert_type_config[key] = AlertTypeConfig(**val) if isinstance(val, dict) else val @@ -1420,9 +1432,8 @@ Model Info: return # Start periodic flush if not already started - if not self.periodic_started and self.alerting is not None and len(self.alerting) > 0: - asyncio.create_task(self.periodic_flush()) - self.periodic_started = True + if len(self.alerting) > 0: + self._ensure_periodic_flush_started() if "webhook" in self.alerting and alert_type == "budget_alerts" and user_info is not None: await self.send_webhook_alert(webhook_event=user_info) diff --git a/tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py b/tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py index cfbd3e76a88..167b6c110fb 100644 --- a/tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py +++ b/tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py @@ -164,6 +164,20 @@ class TestSlackAlerting(unittest.TestCase): self.slack_alerting.update_values(alerting_args={"slack_alerting": "True"}) assert self.slack_alerting.periodic_started == True + # The proxy re-invokes update_values on a timer; it must not spawn a new + # periodic_flush task each time (each one is a `while True` loop that + # never exits, so they accumulate for the lifetime of the process). + @patch("asyncio.create_task") + def test_update_values_starts_periodic_task_only_once(self, mock_create_task): + mock_create_task.return_value = AsyncMock() + + for _ in range(5): + self.slack_alerting.update_values(alerting=["slack"]) + self.slack_alerting.update_values(alerting_args={"slack_alerting": "True"}) + + assert self.slack_alerting.periodic_started is True + assert mock_create_task.call_count == 1 + @patch("litellm.integrations.SlackAlerting.slack_alerting.datetime") def test_alert_type_in_formatted_message(self, mock_datetime): # Setup mocks