fix(redis): log a timeout streak once per interval instead of one line per cache call

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/caching/redis_cache.py

@@ -15,6 +15,7 @@ import hashlib
 import inspect
 import json
 import logging
+import threading
 import time
 from collections.abc import Awaitable, Callable, Sequence
 from contextvars import ContextVar
@@ -32,6 +33,7 @@ from litellm.constants import (
     REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD,
     REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT,
     REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION,
+    REDIS_TIMEOUT_LOG_INTERVAL,
 )
 from litellm.litellm_core_utils.core_helpers import _get_parent_otel_span_from_kwargs
 from litellm.litellm_core_utils.coroutine_checker import coroutine_checker
@@ -404,13 +406,58 @@ class RedisCircuitBreakerOpenError(Exception):
     pass
 
 
+class _RedisTimeoutLogThrottle:
+    """Admits one Redis timeout log line per interval and counts the timeouts it suppressed in between."""
+
+    def __init__(self, interval: float, clock: Callable[[], float] = time.time) -> None:
+        self.interval = interval
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._last_logged_at: float | None = None
+        self._suppressed = 0
+
+    def admit(self) -> int | None:
+        """Return the number of timeouts suppressed since the last admitted line, or None to suppress this one."""
+        with self._lock:
+            now: Final = self._clock()
+            if self._last_logged_at is not None and now - self._last_logged_at < self.interval:
+                self._suppressed += 1
+                return None
+            suppressed: Final = self._suppressed
+            self._suppressed = 0
+            self._last_logged_at = now
+            return suppressed
+
+
+_redis_timeout_log_throttle: Final = _RedisTimeoutLogThrottle(REDIS_TIMEOUT_LOG_INTERVAL)
+
+
 def log_redis_failure(
     logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False
 ) -> None:
     if isinstance(exc, RedisCircuitBreakerOpenError):
-        logger.debug("%s: %s", message, exc)
+        logger.debug("%s: %s", message, exc, stacklevel=2)
         return
-    logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None)
+    exc_info: Final = exc if with_traceback else None
+    if not _is_redis_timeout_failure(exc):
+        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+        return
+    suppressed: Final = _redis_timeout_log_throttle.admit()
+    if suppressed is None:
+        logger.debug("%s: %s", message, exc, stacklevel=2)
+        return
+    if suppressed == 0:
+        logger.log(level, "%s: %s", message, exc, exc_info=exc_info, stacklevel=2)
+        return
+    logger.log(
+        level,
+        "%s: %s (%d more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        message,
+        exc,
+        suppressed,
+        exc_info=exc_info,
+        stacklevel=2,
+    )
 
 
 @dataclass(frozen=True, slots=True)
@@ -783,10 +830,8 @@ class RedisCache(BaseCache):
             ## LOGGING ##
             end_time = time.time()
             _duration = end_time - start_time
-            verbose_logger.error(
-                "LiteLLM Redis Caching: increment_cache() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: increment_cache() - Got exception from REDIS", e
             )
             raise e
 
@@ -992,11 +1037,8 @@ class RedisCache(BaseCache):
                     call_type=f"async_set_cache <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, key=%r, value=%r",
-                str(e),
-                key,
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             raise e
 
@@ -1044,10 +1086,8 @@ class RedisCache(BaseCache):
                     event_metadata={"key": key},
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1094,7 +1134,6 @@ class RedisCache(BaseCache):
         start_time: Final = time.time()
 
         print_verbose(f"Set Async Redis Cache: key list: {cache_list}\nttl={ttl}, redis_version={self.redis_version}")
-        cache_value: Final = None
         try:
             async with _redis_client.pipeline(transaction=False) as pipe:
                 results: Final = await self._pipeline_helper(pipe, cache_list, ttl)
@@ -1131,10 +1170,11 @@ class RedisCache(BaseCache):
                 )
             )
 
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                cache_value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS",
+                e,
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1177,10 +1217,8 @@ class RedisCache(BaseCache):
                 )
             )
             # NON blocking - notify users Redis is throwing an exception
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "LiteLLM Redis Caching: async set() - Got exception from REDIS", e
             )
             raise e
 
@@ -1216,10 +1254,11 @@ class RedisCache(BaseCache):
                 )
             )
             # NON blocking - notify users Redis is throwing an exception
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS",
+                e,
             )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
@@ -1288,10 +1327,11 @@ class RedisCache(BaseCache):
                     parent_otel_span=parent_otel_span,
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS %s, Writing value=%s",
-                str(e),
-                value,
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -1377,7 +1417,9 @@ class RedisCache(BaseCache):
             print_verbose(f"Got Redis Cache: key: {key}, cached_response {cached_response}")
             return self._get_cache_logic(cached_response=cached_response)
         except Exception as e:
-            verbose_logger.error("litellm.caching.caching: get() - Got exception from REDIS: %s", e)
+            log_redis_failure(
+                verbose_logger, logging.ERROR, "litellm.caching.caching: get() - Got exception from REDIS", e
+            )
             _record_swallowed_redis_failure(self._circuit_breaker, e)
 
     def _run_redis_mget_operation(self, keys: list[str]) -> Sequence[bytes | str | None]:
@@ -1455,7 +1497,7 @@ class RedisCache(BaseCache):
                 end_time=failed_at,
                 parent_otel_span=parent_otel_span,
             )
-            verbose_logger.error("Error occurred in batch get cache - %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "Error occurred in batch get cache", e)
             _record_swallowed_redis_failure(self._circuit_breaker, e)
             return key_value_dict
 
@@ -1574,7 +1616,7 @@ class RedisCache(BaseCache):
                     parent_otel_span=parent_otel_span,
                 )
             )
-            verbose_logger.error("Error occurred in async batch get cache - %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "Error occurred in async batch get cache", e)
             _record_swallowed_redis_failure(self._circuit_breaker, e)
             return key_value_dict
 
@@ -1799,9 +1841,11 @@ class RedisCache(BaseCache):
                     parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs),
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -1878,7 +1922,7 @@ class RedisCache(BaseCache):
                     call_type=f"async_rpush <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error("LiteLLM Redis Cache RPUSH: - Got exception from REDIS : %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Redis Cache RPUSH: - Got exception from REDIS", e)
             raise e
 
     async def _pipeline_rpush_helper(
@@ -1946,9 +1990,11 @@ class RedisCache(BaseCache):
                     call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e
 
@@ -2024,7 +2070,7 @@ class RedisCache(BaseCache):
                     call_type=f"async_lpop <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error("LiteLLM Redis Cache LPOP: - Got exception from REDIS : %s", e)
+            log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Redis Cache LPOP: - Got exception from REDIS", e)
             raise e
 
     async def _pipeline_lpop_helper(
@@ -2135,8 +2181,10 @@ class RedisCache(BaseCache):
                     call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}",
                 )
             )
-            verbose_logger.error(
-                "LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS %s",
-                str(e),
+            log_redis_failure(
+                verbose_logger,
+                logging.ERROR,
+                "LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS",
+                e,
             )
             raise e

litellm/constants.py

@@ -459,6 +459,9 @@ REDIS_CIRCUIT_BREAKER_ENABLED: Final = os.getenv("REDIS_CIRCUIT_BREAKER_ENABLED"
 # minimum seconds a timeout-only failure streak must span before it can open the breaker,
 # so one event-loop stall timing out many queued calls at once does not trip it
 REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION: Final = float(os.getenv("REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION", 5.0))
+# seconds between Redis timeout log lines: the first timeout of a streak logs at the caller's level,
+# later ones log at DEBUG until the interval passes and one line summarizes how many were suppressed
+REDIS_TIMEOUT_LOG_INTERVAL: Final = float(os.getenv("REDIS_TIMEOUT_LOG_INTERVAL", "5.0"))
 # Seconds of idle before a Redis cluster connection is validated with a PING and
 # reconnected if dead, so a connection silently dropped by a cluster restart
 # (e.g. ElastiCache Serverless maintenance) is not reused while broken

tests/test_litellm/caching/test_dual_cache.py

@@ -704,3 +704,60 @@ async def test_open_breaker_keeps_async_batch_read_memory_hits_and_releases_rese
 
     assert list(await cache.async_batch_get_cache(["k1", "k2"])) == ["v1", None]
     assert "k2" not in cache.last_redis_batch_access_time
+
+
+@pytest.mark.asyncio
+async def test_redis_timeouts_falling_back_to_memory_log_once_per_interval(caplog, monkeypatch):
+    """The in-memory fallback WARNING must not repeat for every timed-out increment during a blip.
+
+    The rate limiter's pipeline increments and the dual cache increments each logged a WARNING per
+    call while Redis timed out, hundreds of lines per second before the breaker opened. The first
+    timeout of a streak keeps its WARNING, the rest are DEBUG until the summary interval passes.
+    """
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    from litellm.caching import redis_cache as redis_cache_module
+    from litellm.caching.redis_cache import _RedisTimeoutLogThrottle
+
+    clock = MagicMock(return_value=1_000.0)
+    monkeypatch.setattr(
+        redis_cache_module, "_redis_timeout_log_throttle", _RedisTimeoutLogThrottle(interval=5.0, clock=clock)
+    )
+
+    class _TimingOutRedis:
+        async def async_increment_pipeline(self, increment_list, **kwargs):
+            raise RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+        async def async_increment(self, key, value, **kwargs):
+            raise RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+    cache = DualCache(in_memory_cache=InMemoryCache(), redis_cache=_TimingOutRedis())  # pyright: ignore[reportArgumentType]  # duck-typed Redis double
+    increments = [RedisPipelineIncrementOperation(key="k", increment_value=1.0, ttl=60)]
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(100):
+            await cache.async_increment_cache_pipeline(increment_list=increments)
+            await cache.async_increment_cache("k", 1.0)
+
+    visible = [r for r in caplog.records if r.levelno >= logging.WARNING]
+    assert [(r.levelno, r.getMessage()) for r in visible] == [
+        (
+            logging.WARNING,
+            "Redis async_increment_cache_pipeline failed, falling back to in-memory result:"
+            " Timeout reading from 127.0.0.1:6379",
+        )
+    ]
+    assert visible[0].filename == "dual_cache.py"
+    assert sum("Timeout reading from" in r.getMessage() for r in caplog.records) == 200
+
+    caplog.clear()
+    clock.return_value += 5.0
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        await cache.async_increment_cache("k", 1.0)
+    assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
+        (
+            logging.WARNING,
+            "Redis async_increment_cache failed, falling back to in-memory result: Timeout reading from 127.0.0.1:6379"
+            " (199 more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        )
+    ]

tests/test_litellm/caching/test_redis_cache.py

@@ -1202,3 +1202,60 @@ async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new
     new_probe_release.set()
     assert await new_probe == "new probe"
     assert breaker._state == breaker.CLOSED
+
+
+def test_timeouts_during_a_blip_log_once_per_interval_not_once_per_call(sync_batch_redis_cache, caplog, monkeypatch):
+    """A Redis latency blip must not write one ERROR line per timed-out cache call.
+
+    Before the breaker opens (up to REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION of timeouts) every
+    cache operation logged its own ERROR or WARNING line, so one single-worker proxy wrote
+    ~1100 lines in 5 s at LITELLM_LOG=WARNING. A timeout streak now logs its first failure, then
+    one summary line per REDIS_TIMEOUT_LOG_INTERVAL carrying the count of suppressed timeouts,
+    while every timeout stays visible at DEBUG. Hard connectivity failures keep their per-call line.
+    """
+    import logging
+
+    from redis.exceptions import TimeoutError as RedisTimeoutError
+
+    from litellm.caching import redis_cache as redis_cache_module
+    from litellm.caching.redis_cache import _RedisTimeoutLogThrottle
+
+    clock = MagicMock(return_value=1_000.0)
+    monkeypatch.setattr(
+        redis_cache_module, "_redis_timeout_log_throttle", _RedisTimeoutLogThrottle(interval=5.0, clock=clock)
+    )
+    sync_batch_redis_cache.redis_client.get.side_effect = RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+    sync_batch_redis_cache.redis_client.mget.side_effect = RedisTimeoutError("Timeout reading from 127.0.0.1:6379")
+
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(200):
+            assert sync_batch_redis_cache.get_cache("lit7520") is None
+        assert sync_batch_redis_cache.batch_get_cache(key_list=["lit7520"]) == {}
+
+    timeout_records = [r for r in caplog.records if "Timeout reading from" in r.getMessage()]
+    assert len(timeout_records) == 201, "every timeout must stay visible at DEBUG"
+    assert [r.getMessage() for r in timeout_records if r.levelno >= logging.WARNING] == [
+        "litellm.caching.caching: get() - Got exception from REDIS: Timeout reading from 127.0.0.1:6379"
+    ]
+    assert timeout_records[0].levelno == logging.ERROR
+    assert timeout_records[0].filename == "redis_cache.py"
+    assert timeout_records[0].lineno != timeout_records[-1].lineno, "the record must point at the cache operation"
+
+    caplog.clear()
+    clock.return_value += 5.0
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        assert sync_batch_redis_cache.batch_get_cache(key_list=["lit7520"]) == {}
+    assert [(r.levelno, r.getMessage()) for r in caplog.records] == [
+        (
+            logging.ERROR,
+            "Error occurred in batch get cache: Timeout reading from 127.0.0.1:6379"
+            " (200 more Redis timeouts since the previous Redis timeout line were logged at DEBUG)",
+        )
+    ]
+
+    caplog.clear()
+    sync_batch_redis_cache.redis_client.get.side_effect = OSError("redis unavailable")
+    with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
+        for _ in range(3):
+            assert sync_batch_redis_cache.get_cache("lit7520") is None
+    assert [r.levelno for r in caplog.records if "redis unavailable" in r.getMessage()] == [logging.ERROR] * 3