From 107b4ec4db64985de0b3651f401b290ea09e81ed Mon Sep 17 00:00:00 2001 From: yassin Date: Sat, 12 Sep 2026 01:12:39 +0000 Subject: [PATCH 1/7] 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 | 140 ++++++++++++------ litellm/constants.py | 3 + tests/test_litellm/caching/test_dual_cache.py | 57 +++++++ .../test_litellm/caching/test_redis_cache.py | 57 +++++++ 4 files changed, 211 insertions(+), 46 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 2c36995c4f8..eaac7ef7b0b 100644 --- a/litellm/caching/redis_cache.py +++ b/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 diff --git a/litellm/constants.py b/litellm/constants.py index 6b984c2673c..a32551b4480 100644 --- a/litellm/constants.py +++ b/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 diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 4c9068722b8..850fa14106b 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/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)", + ) + ] diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index bcae33b976e..d0974b2420c 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/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 From 9c84e98fb22bd0f6e2c359f335bbc329181bb8bd Mon Sep 17 00:00:00 2001 From: yassin Date: Sat, 12 Sep 2026 01:25:32 +0000 Subject: [PATCH 2/7] fix(proxy): treat a Redis timeout in spend counter increments as an already-logged cache failure The cost tracking callback logged its own ERROR with a traceback for every request whose spend counter increment timed out, on top of the cache layer's throttled line. Timeouts now take the same path as breaker-open refusals: invalidate the counters and return. Also exposes is_redis_timeout_failure publicly for that caller and drops the comment on the new constant Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/redis_cache.py | 10 +++---- litellm/constants.py | 2 -- litellm/proxy/proxy_server.py | 4 +-- .../test_litellm/caching/test_redis_cache.py | 26 +++++++++---------- .../proxy/proxy_server/test_spend_counters.py | 22 ++++++++++++++++ 5 files changed, 42 insertions(+), 22 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index eaac7ef7b0b..7a3e689a667 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -330,7 +330,7 @@ def _redis_timeout_error_types() -> tuple[type, ...]: return (RedisTimeoutError, TimeoutError) -def _is_redis_timeout_failure(exc: BaseException) -> bool: +def is_redis_timeout_failure(exc: BaseException) -> bool: return isinstance(exc, _redis_timeout_error_types()) @@ -398,7 +398,7 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep """ if not _is_redis_health_failure(exc): return - breaker.record_failure(is_timeout=_is_redis_timeout_failure(exc)) + breaker.record_failure(is_timeout=is_redis_timeout_failure(exc)) _swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1) @@ -439,7 +439,7 @@ def log_redis_failure( logger.debug("%s: %s", message, exc, stacklevel=2) return exc_info: Final = exc if with_traceback else None - if not _is_redis_timeout_failure(exc): + 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() @@ -504,7 +504,7 @@ async def _run_under_circuit_breaker( result: Final = await call() except Exception as e: if _is_redis_health_failure(e): - breaker.record_failure(is_timeout=_is_redis_timeout_failure(e)) + breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) raise _exit_circuit_breaker(breaker, admission) return result @@ -521,7 +521,7 @@ def _run_under_circuit_breaker_sync( result: Final = call() except Exception as e: if _is_redis_health_failure(e): - breaker.record_failure(is_timeout=_is_redis_timeout_failure(e)) + breaker.record_failure(is_timeout=is_redis_timeout_failure(e)) raise _exit_circuit_breaker(breaker, admission) return result diff --git a/litellm/constants.py b/litellm/constants.py index a32551b4480..60e1c682238 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -459,8 +459,6 @@ 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 diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index c8fd7edfed6..318ea96dbcb 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -251,7 +251,7 @@ import litellm._redis from litellm import Router from litellm._logging import _redact_string, verbose_proxy_logger, verbose_router_logger from litellm.caching.caching import DualCache, RedisCache -from litellm.caching.redis_cache import RedisCircuitBreakerOpenError +from litellm.caching.redis_cache import RedisCircuitBreakerOpenError, is_redis_timeout_failure from litellm.caching.redis_cluster_cache import RedisClusterCache from litellm.constants import ( _REALTIME_BODY_CACHE_SIZE, @@ -3411,7 +3411,7 @@ async def _apply_spend_counter_increments(pending: Sequence[_PendingSpendIncreme results: Final = await redis_cache.async_increment_pipeline(increment_list=increment_list) except Exception as e: await asyncio.gather(*(_invalidate_spend_counter(counter_key=item.counter_key) for item in pending)) - if isinstance(e, RedisCircuitBreakerOpenError): + if isinstance(e, RedisCircuitBreakerOpenError) or is_redis_timeout_failure(e): return raise for item, current_value in zip(pending, results or ()): diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index d0974b2420c..5840f450ac6 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -977,17 +977,17 @@ async def test_stale_timeout_does_not_let_sub_threshold_hard_failures_open_the_b from redis.exceptions import ConnectionError as RedisConnectionError from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure + from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=0.05) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out"))) await asyncio.sleep(0.06) for _ in range(breaker.failure_threshold - 1): - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused"))) assert breaker.is_open() is False, "2 hard failures and 1 stale timeout are below both thresholds" - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused"))) assert breaker.is_open() is True, "the threshold-th hard failure must still open it" @@ -999,19 +999,19 @@ async def test_hard_failure_resets_timeout_streak_so_a_later_burst_must_earn_its from redis.exceptions import ConnectionError as RedisConnectionError from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure + from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=0.05) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out"))) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused"))) await asyncio.sleep(0.06) for _ in range(breaker.failure_threshold): - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out"))) assert breaker.is_open() is False, "the burst is instantaneous, so the duration gate must hold it closed" await asyncio.sleep(0.06) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("read timed out"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("read timed out"))) assert breaker.is_open() is True, "the same run of timeouts persisting past the duration must open it" @@ -1022,7 +1022,7 @@ async def test_breaker_metrics_track_state_and_failure_class(): from redis.exceptions import ConnectionError as RedisConnectionError from redis.exceptions import TimeoutError as RedisTimeoutError - from litellm.caching.redis_cache import RedisCircuitBreaker, _is_redis_timeout_failure + from litellm.caching.redis_cache import RedisCircuitBreaker, is_redis_timeout_failure def sample(name, labels=None): return REGISTRY.get_sample_value(name, labels) or 0.0 @@ -1034,9 +1034,9 @@ async def test_breaker_metrics_track_state_and_failure_class(): closed_gauge_before = sample("litellm_redis_circuit_breaker_state", {"state": "closed"}) breaker = RedisCircuitBreaker(failure_threshold=2, recovery_timeout=60, timeout_min_duration=5.0) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisTimeoutError("t"))) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) - breaker.record_failure(is_timeout=_is_redis_timeout_failure(RedisConnectionError("refused"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisTimeoutError("t"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused"))) + breaker.record_failure(is_timeout=is_redis_timeout_failure(RedisConnectionError("refused"))) assert sample("litellm_redis_circuit_breaker_failures_total", {"failure_class": "timeout"}) == timeout_before + 1 assert sample("litellm_redis_circuit_breaker_failures_total", {"failure_class": "connectivity"}) == hard_before + 2 diff --git a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py index 2f47736a398..19b5a11af33 100644 --- a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py +++ b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py @@ -1180,6 +1180,28 @@ async def test_apply_spend_counter_increments_open_breaker_invalidates_and_retur fake_cache.in_memory_cache.set_cache.assert_not_called() +@pytest.mark.asyncio +async def test_apply_spend_counter_increments_redis_timeout_invalidates_and_returns(monkeypatch): + """A Redis timeout is the streak the breaker is already counting and the cache layer already logged. + + Re-raising it sent every request in the pre-open window through the cost callback's error + path, which logged a traceback and fired the failed-tracking alert once per request. + """ + from redis.exceptions import TimeoutError as RedisTimeoutError + + fake_cache = _make_spend_counter_cache() + fake_cache.redis_cache.async_increment_pipeline = AsyncMock( + side_effect=RedisTimeoutError("Timeout reading from 127.0.0.1:6379") + ) + monkeypatch.setattr(ps, "spend_counter_cache", fake_cache) + + await ps._apply_spend_counter_increments(_two_pending_increments()) + + deleted_keys = sorted(call.kwargs["key"] for call in fake_cache.in_memory_cache.delete_cache.call_args_list) + assert deleted_keys == ["spend:key:k", "spend:team:t"] + fake_cache.in_memory_cache.set_cache.assert_not_called() + + @pytest.mark.asyncio async def test_apply_spend_counter_increments_other_redis_error_invalidates_and_raises(monkeypatch): fake_cache = _make_spend_counter_cache() From f681a978f06baa13da0f0c24f7b1ac3a20d9a02a Mon Sep 17 00:00:00 2001 From: yassin Date: Sat, 12 Sep 2026 01:37:36 +0000 Subject: [PATCH 3/7] fix(redis): use a monotonic clock for the timeout log throttle and trim test docstrings Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/redis_cache.py | 2 +- tests/test_litellm/caching/test_dual_cache.py | 7 +------ tests/test_litellm/caching/test_redis_cache.py | 9 +-------- .../proxy/proxy_server/test_spend_counters.py | 6 +----- 4 files changed, 4 insertions(+), 20 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 7a3e689a667..e5e27d1e02b 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -409,7 +409,7 @@ class RedisCircuitBreakerOpenError(Exception): 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: + def __init__(self, interval: float, clock: Callable[[], float] = time.monotonic) -> None: self.interval = interval self._clock = clock self._lock = threading.Lock() diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 850fa14106b..6f29be00b30 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -708,12 +708,7 @@ async def test_open_breaker_keeps_async_batch_read_memory_hits_and_releases_rese @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. - """ + """The first fallback WARNING of a timeout streak logs, the rest stay at DEBUG until the summary.""" from redis.exceptions import TimeoutError as RedisTimeoutError from litellm.caching import redis_cache as redis_cache_module diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 5840f450ac6..4ca33894aed 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1205,14 +1205,7 @@ async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new 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. - """ + """A timeout streak logs its first failure plus one summary per interval; other failures log per call.""" import logging from redis.exceptions import TimeoutError as RedisTimeoutError diff --git a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py index 19b5a11af33..4a1fc389d3e 100644 --- a/tests/test_litellm/proxy/proxy_server/test_spend_counters.py +++ b/tests/test_litellm/proxy/proxy_server/test_spend_counters.py @@ -1182,11 +1182,7 @@ async def test_apply_spend_counter_increments_open_breaker_invalidates_and_retur @pytest.mark.asyncio async def test_apply_spend_counter_increments_redis_timeout_invalidates_and_returns(monkeypatch): - """A Redis timeout is the streak the breaker is already counting and the cache layer already logged. - - Re-raising it sent every request in the pre-open window through the cost callback's error - path, which logged a traceback and fired the failed-tracking alert once per request. - """ + """A Redis timeout invalidates the counters and returns without reaching the cost callback's error path.""" from redis.exceptions import TimeoutError as RedisTimeoutError fake_cache = _make_spend_counter_cache() From 28f2d1f0168aa31639a23447d391516129267069 Mon Sep 17 00:00:00 2001 From: yassin Date: Sat, 12 Sep 2026 01:58:36 +0000 Subject: [PATCH 4/7] test(redis): cover the write and list timeout paths going through the shared log throttle Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../test_litellm/caching/test_redis_cache.py | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 4ca33894aed..5ea21dae539 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1252,3 +1252,60 @@ def test_timeouts_during_a_blip_log_once_per_interval_not_once_per_call(sync_bat 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 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "call_method", + [ + pytest.param(lambda c: c.async_set_cache_pipeline([("lit7520", "v")]), id="async_set_cache_pipeline"), + pytest.param(lambda c: c.async_set_cache_sadd("lit7520", ["v"], ttl=None), id="async_set_cache_sadd"), + pytest.param(lambda c: c.async_increment("lit7520", 1.0), id="async_increment"), + pytest.param( + lambda c: c.async_increment_pipeline([{"key": "lit7520", "increment_value": 1.0, "ttl": 60}]), + id="async_increment_pipeline", + ), + pytest.param(lambda c: c.async_rpush("lit7520", ["v"]), id="async_rpush"), + pytest.param( + lambda c: c.async_rpush_pipeline([{"key": "lit7520", "values": ["v"]}]), id="async_rpush_pipeline" + ), + pytest.param(lambda c: c.async_lpop("lit7520"), id="async_lpop"), + pytest.param(lambda c: c.async_lpop_pipeline([{"key": "lit7520", "count": 1}]), id="async_lpop_pipeline"), + ], +) +async def test_write_path_timeouts_inside_the_interval_stay_at_debug(call_method, caplog, monkeypatch, redis_no_ping): + """A write or list operation timing out mid-streak is counted by the throttle instead of logging its own ERROR.""" + import contextlib + 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) + throttle = _RedisTimeoutLogThrottle(interval=5.0, clock=clock) + assert throttle.admit() == 0 + monkeypatch.setattr(redis_cache_module, "_redis_timeout_log_throttle", throttle) + + timeout = RedisTimeoutError("Timeout reading from 127.0.0.1:6379") + client = MagicMock() + client.pipeline.return_value.__aenter__.side_effect = timeout + client.sadd = AsyncMock(side_effect=timeout) + client.incrbyfloat = AsyncMock(side_effect=timeout) + client.rpush = AsyncMock(side_effect=timeout) + client.lpop = AsyncMock(side_effect=timeout) + monkeypatch.setenv("REDIS_HOST", "https://my-test-host") + cache = RedisCache() + + with ( + patch.object(cache, "init_async_client", return_value=client), + caplog.at_level(logging.DEBUG, logger="LiteLLM"), + ): + with contextlib.suppress(RedisTimeoutError): + await call_method(cache) + + timeout_records = [r for r in caplog.records if "Timeout reading from" in r.getMessage()] + assert [(r.levelno, r.filename) for r in timeout_records] == [(logging.DEBUG, "redis_cache.py")] + clock.return_value += 5.0 + assert throttle.admit() == 1 From c4354c248a78efec9842c59c1dee13c7e556310f Mon Sep 17 00:00:00 2001 From: yassin Date: Mon, 14 Sep 2026 19:38:14 +0000 Subject: [PATCH 5/7] fix(redis): route the per-TTL pipeline write timeout log through the shared throttle Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/redis_cache.py | 7 +++++-- tests/test_litellm/caching/test_redis_cache.py | 4 ++++ 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 3c3539eb8a9..b4b2b1a334c 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -1231,8 +1231,11 @@ class RedisCache(BaseCache): end_time=time.time(), ) ) - verbose_logger.error( - "LiteLLM Redis Caching: async_set_cache_pipeline_with_ttls() - Got exception from REDIS %s", str(e) + log_redis_failure( + verbose_logger, + logging.ERROR, + "LiteLLM Redis Caching: async_set_cache_pipeline_with_ttls() - Got exception from REDIS", + e, ) _record_swallowed_redis_failure(self._circuit_breaker, e) diff --git a/tests/test_litellm/caching/test_redis_cache.py b/tests/test_litellm/caching/test_redis_cache.py index 9fdddd3c085..19638c60b4b 100644 --- a/tests/test_litellm/caching/test_redis_cache.py +++ b/tests/test_litellm/caching/test_redis_cache.py @@ -1260,6 +1260,10 @@ def test_timeouts_during_a_blip_log_once_per_interval_not_once_per_call(sync_bat "call_method", [ pytest.param(lambda c: c.async_set_cache_pipeline([("lit7520", "v")]), id="async_set_cache_pipeline"), + pytest.param( + lambda c: c.async_set_cache_pipeline_with_ttls([("lit7520", "v", 60.0)]), + id="async_set_cache_pipeline_with_ttls", + ), pytest.param(lambda c: c.async_set_cache_sadd("lit7520", ["v"], ttl=None), id="async_set_cache_sadd"), pytest.param(lambda c: c.async_increment("lit7520", 1.0), id="async_increment"), pytest.param( From bea22df6c27b30d060e6f7e239e94396c9f38ad7 Mon Sep 17 00:00:00 2001 From: yassin Date: Mon, 14 Sep 2026 19:59:10 +0000 Subject: [PATCH 6/7] fix(caching): keep generic add_cache failures at ERROR unless the backend is Redis Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/caching/caching.py | 13 ++++++++--- tests/test_litellm/caching/test_caching.py | 26 ++++++++++++++++++++++ 2 files changed, 36 insertions(+), 3 deletions(-) diff --git a/litellm/caching/caching.py b/litellm/caching/caching.py index d6dd2a073af..be82f5def1f 100644 --- a/litellm/caching/caching.py +++ b/litellm/caching/caching.py @@ -679,7 +679,14 @@ class Cache: cache_key, cached_data, kwargs = self._add_cache_logic(result=result, **kwargs) self.cache.set_cache(cache_key, cached_data, **kwargs) except Exception as e: - log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e) + self._log_add_cache_failure(e) + + def _log_add_cache_failure(self, exc: Exception) -> None: + message: Final = "LiteLLM Cache: exception in add_cache" + if isinstance(self.cache, RedisCache): + log_redis_failure(verbose_logger, logging.ERROR, message, exc) + return + verbose_logger.error("%s: %s", message, exc) async def async_add_cache(self, result, dynamic_cache_object: BaseCache | None = None, **kwargs): """ @@ -698,7 +705,7 @@ class Cache: else: await self.cache.async_set_cache(cache_key, cached_data, **kwargs) except Exception as e: - log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e) + self._log_add_cache_failure(e) def _convert_to_cached_embedding( self, @@ -877,7 +884,7 @@ class Cache: else: await self.cache.async_set_cache_pipeline(cache_list=cache_list, **kwargs) except Exception as e: - log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e) + self._log_add_cache_failure(e) def should_use_cache(self, **kwargs): """ diff --git a/tests/test_litellm/caching/test_caching.py b/tests/test_litellm/caching/test_caching.py index 4d0ec0fb677..c7ec8abc31e 100644 --- a/tests/test_litellm/caching/test_caching.py +++ b/tests/test_litellm/caching/test_caching.py @@ -1,9 +1,12 @@ import logging import re +from unittest.mock import MagicMock import pytest +import litellm.caching.redis_cache as redis_cache_module from litellm.caching.caching import Cache +from litellm.caching.redis_cache import RedisCache, _RedisTimeoutLogThrottle from litellm.types.caching import LiteLLMCacheType, SemanticCacheScope from litellm.types.utils import Embedding, EmbeddingResponse, Usage @@ -53,6 +56,29 @@ def test_cache_key_debug_log_does_not_include_prompt_material(caplog): assert any(cache_key in message for message in created_cache_key_logs) +@pytest.mark.parametrize( + ("backend", "expected_level"), + [ + pytest.param(MagicMock(spec=RedisCache), logging.DEBUG, id="redis_backend_is_throttled"), + pytest.param(MagicMock(), logging.ERROR, id="other_backend_logs_every_timeout"), + ], +) +def test_add_cache_timeout_only_joins_redis_throttle_for_redis_backends(backend, expected_level, caplog, monkeypatch): + throttle = _RedisTimeoutLogThrottle(interval=5.0, clock=MagicMock(return_value=1_000.0)) + assert throttle.admit() == 0 + monkeypatch.setattr(redis_cache_module, "_redis_timeout_log_throttle", throttle) + + cache = Cache(type=LiteLLMCacheType.LOCAL) + backend.set_cache.side_effect = TimeoutError("lit7520 backend timed out") + cache.cache = backend + + with caplog.at_level(logging.DEBUG, logger="LiteLLM"): + cache.add_cache("result", model="gpt-4.1-mini", messages=[{"role": "user", "content": "hi"}]) + + records = [r for r in caplog.records if "lit7520 backend timed out" in r.getMessage()] + assert [r.levelno for r in records] == [expected_level] + + def _embedding_response(prompt_tokens, num_items): return EmbeddingResponse( model="amazon.titan-embed-image-v1", From a4e34d6e1b9905736acd5d13ea03d7e32f3e3f0e Mon Sep 17 00:00:00 2001 From: yassin Date: Mon, 14 Sep 2026 20:05:14 +0000 Subject: [PATCH 7/7] test(caching): wrap the DualCache fixture line to the 120 character limit Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- tests/test_litellm/caching/test_dual_cache.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 6f29be00b30..95395878c25 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -726,7 +726,10 @@ async def test_redis_timeouts_falling_back_to_memory_log_once_per_interval(caplo 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 + 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"):