Merge f16a63ebf9 into c2c2a623c0

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -114,6 +114,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,
         )
@@ -125,6 +126,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,
@@ -137,17 +152,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
@@ -1442,9 +1454,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)

tests/test_litellm/integrations/SlackAlerting/test_slack_alerting.py

@@ -15,6 +15,16 @@ from litellm.proxy._types import CallInfo, Litellm_EntityType
 from litellm.types.integrations.slack_alerting import AlertType, SlackAlertingCacheKeys
 
 
+def _close_and_stub(coro):
+    """asyncio.create_task stand-in: close the coroutine nothing will await.
+
+    Leaving it open emits an unawaited-coroutine warning, which is an error
+    under -W error.
+    """
+    coro.close()
+    return MagicMock()
+
+
 class TestSlackAlerting(unittest.TestCase):
     def setUp(self):
         self.slack_alerting = SlackAlerting()
@@ -171,14 +181,27 @@ class TestSlackAlerting(unittest.TestCase):
     # Calling update_values with alerting args should try to start the periodic task
     @patch("asyncio.create_task")
     def test_update_values_starts_periodic_task(self, mock_create_task):
-        # Make it do nothing (or return a dummy future)
-        mock_create_task.return_value = AsyncMock()  # prevents awaiting errors
+        mock_create_task.side_effect = _close_and_stub
 
         assert self.slack_alerting.periodic_started == False
 
         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.side_effect = _close_and_stub
+
+        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