diff --git a/litellm/router_strategy/base_routing_strategy.py b/litellm/router_strategy/base_routing_strategy.py index 8235761ca98..686d57e2b77 100644 --- a/litellm/router_strategy/base_routing_strategy.py +++ b/litellm/router_strategy/base_routing_strategy.py @@ -3,12 +3,13 @@ Base class across routing strategies to abstract commmon functions like batch in """ import asyncio +import logging from abc import ABC from typing import Final from litellm._logging import verbose_router_logger from litellm.caching.caching import DualCache -from litellm.caching.redis_cache import RedisPipelineIncrementOperation +from litellm.caching.redis_cache import RedisPipelineIncrementOperation, log_redis_failure from litellm.constants import DEFAULT_REDIS_SYNC_INTERVAL @@ -147,7 +148,7 @@ class BaseRoutingStrategy(ABC): return return_result except Exception as e: - verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e) + log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e) self.redis_increment_operation_queue = [] def add_to_in_memory_keys_to_update(self, key: str): diff --git a/litellm/router_strategy/budget_limiter.py b/litellm/router_strategy/budget_limiter.py index a8d51f95e45..acdc57d706d 100644 --- a/litellm/router_strategy/budget_limiter.py +++ b/litellm/router_strategy/budget_limiter.py @@ -20,6 +20,7 @@ anthropic: import asyncio import builtins +import logging from collections.abc import Mapping from datetime import datetime, timedelta, timezone from typing import Any, Final @@ -27,7 +28,7 @@ from typing import Any, Final import litellm from litellm._logging import verbose_router_logger from litellm.caching.caching import DualCache -from litellm.caching.redis_cache import RedisPipelineIncrementOperation +from litellm.caching.redis_cache import RedisPipelineIncrementOperation, log_redis_failure from litellm.integrations.custom_logger import CustomLogger, Span from litellm.litellm_core_utils.core_helpers import ( get_metadata_variable_name_from_kwargs, @@ -536,17 +537,13 @@ class RouterBudgetLimiting(CustomLogger): "Pushing Redis Increment Pipeline for queue: %s", self.redis_increment_operation_queue, ) - if len(self.redis_increment_operation_queue) > 0: - asyncio.create_task( - self.dual_cache.redis_cache.async_increment_pipeline( - increment_list=self.redis_increment_operation_queue, - ) - ) - + queued: Final = self.redis_increment_operation_queue self.redis_increment_operation_queue = [] + if queued: + await self.dual_cache.redis_cache.async_increment_pipeline(increment_list=queued) except Exception as e: - verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e) + log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e) async def _sync_in_memory_spend_with_redis(self): """ @@ -601,7 +598,7 @@ class RouterBudgetLimiting(CustomLogger): verbose_router_logger.debug("Updated in-memory cache for %s: %s", key, value) except Exception as e: - verbose_router_logger.error("Error syncing in-memory cache with Redis: %s", e) + log_redis_failure(verbose_router_logger, logging.ERROR, "Error syncing in-memory cache with Redis", e) def _get_budget_config_for_deployment( self, diff --git a/tests/test_litellm/router_strategy/test_base_routing_strategy.py b/tests/test_litellm/router_strategy/test_base_routing_strategy.py index 154042692d0..dc75c3d2916 100644 --- a/tests/test_litellm/router_strategy/test_base_routing_strategy.py +++ b/tests/test_litellm/router_strategy/test_base_routing_strategy.py @@ -1,4 +1,5 @@ import json +import logging from typing import Any, Dict, List, Optional, Set, Union import pytest @@ -9,7 +10,7 @@ from unittest.mock import MagicMock, patch from litellm.caching.caching import DualCache -from litellm.caching.redis_cache import RedisPipelineIncrementOperation +from litellm.caching.redis_cache import RedisCircuitBreakerOpenError, RedisPipelineIncrementOperation from litellm.router_strategy.base_routing_strategy import BaseRoutingStrategy @@ -146,3 +147,18 @@ async def test_cache_keys_management(base_strategy): # Test resetting cache keys base_strategy.reset_in_memory_keys_to_update() assert len(base_strategy.get_in_memory_keys_to_update()) == 0 + + +@pytest.mark.asyncio +async def test_push_refused_by_the_open_circuit_breaker_is_not_logged_as_an_error(base_strategy, mock_dual_cache, caplog): + """The sync loop pushes every 100 ms under usage-based routing, so an open breaker must not add an error line per cycle.""" + mock_dual_cache.redis_cache.async_increment_pipeline.side_effect = RedisCircuitBreakerOpenError( + "Redis circuit breaker is open - skipping async_increment_pipeline" + ) + base_strategy.redis_increment_operation_queue = [{"key": "k", "increment_value": 1.0, "ttl": 60}] + + with caplog.at_level(logging.ERROR): + await base_strategy._push_in_memory_increments_to_redis() + + assert caplog.records == [] + assert base_strategy.redis_increment_operation_queue == [] diff --git a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py b/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py index 36fa38bacb5..128b2dbd842 100644 --- a/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py +++ b/tests/test_litellm/router_strategy/test_budget_limiter_hotpath.py @@ -1,7 +1,13 @@ +import asyncio +import gc +import logging +from unittest.mock import AsyncMock, MagicMock + import pytest import litellm from litellm.caching.caching import DualCache +from litellm.caching.redis_cache import RedisCache, RedisCircuitBreakerOpenError from litellm.router_strategy.budget_limiter import RouterBudgetLimiting from litellm.types.router import LiteLLM_Params from litellm.types.utils import BudgetConfig @@ -303,3 +309,34 @@ def test_router_add_deployment_registers_deployment_budget( ) assert config is not None assert config.max_budget == 0.000000000001 + + +@pytest.mark.asyncio +async def test_sync_refused_by_the_open_circuit_breaker_is_quiet_and_leaks_no_task(disable_budget_sync, caplog): + """The budget sync runs every second, so an open breaker must not add an error line or an unretrieved task exception per cycle.""" + refused = RedisCircuitBreakerOpenError("Redis circuit breaker is open - skipping async_increment_pipeline") + redis_cache = MagicMock(spec=RedisCache) + redis_cache.async_increment_pipeline = AsyncMock(side_effect=refused) + redis_cache.async_batch_get_cache = AsyncMock(side_effect=refused) + limiter = RouterBudgetLimiting( + dual_cache=DualCache(redis_cache=redis_cache), + provider_budget_config={"openai": BudgetConfig(max_budget=1.0, budget_duration="1d")}, + ) + await asyncio.gather(*(task for task in asyncio.all_tasks() if task is not asyncio.current_task())) + limiter.redis_increment_operation_queue = [{"key": "provider_spend:openai:1d", "increment_value": 0.5, "ttl": 60}] + loop = asyncio.get_running_loop() + unretrieved = MagicMock() + loop.set_exception_handler(unretrieved) + + try: + with caplog.at_level(logging.ERROR): + await limiter._sync_in_memory_spend_with_redis() + await asyncio.sleep(0) + gc.collect() + finally: + loop.set_exception_handler(None) + + assert caplog.records == [] + unretrieved.assert_not_called() + assert limiter.redis_increment_operation_queue == [] + assert redis_cache.async_increment_pipeline.await_count == 1