From 3c2c089a9e10df6f186ee8a5467d363725682e4b Mon Sep 17 00:00:00 2001 From: jeishod Date: Thu, 27 Aug 2026 14:37:21 +0100 Subject: [PATCH] fix(alerting): start the Slack periodic flush task once, not per config reload update_values is re-invoked by the proxy's periodic deployment reconcile (add_deployment_job -> _update_llm_router -> _add_general_settings_from_db_config), which passes `alerting` whenever the DB config carries that key. That branch created a periodic_flush task unconditionally, and each one is a `while True` loop that never exits, so a long-running proxy accumulates one task per reload interval for the lifetime of the process. The sibling `alerting_args` branch was already guarded; this routes every start site through one helper so the guard cannot be missed again. Also keeps a reference to the task: with the leak gone this is the only flush task, and a bare create_task may be garbage collected mid-execution. --- .../SlackAlerting/slack_alerting.py | 27 +++++++++++++------ .../SlackAlerting/test_slack_alerting.py | 14 ++++++++++ 2 files changed, 33 insertions(+), 8 deletions(-) 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