scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 49

Merge pull request #40817 from BerriAI/litellm_redis_timeout_log_throttle

Browse source 
fix(redis): log a timeout streak once per interval instead of one line per cache call
This commit is contained in:
Yassin Kortam 2026-09-14 13:34:45 -07:00 • committed by GitHub
commit 856aedd14a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 346 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/caching/caching.py
View file

@ -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):
"""

155
litellm/caching/redis_cache.py
View file

@ -15,6 +15,7 @@ import hashlib
import inspect
import json
import logging
import threading
import time
from collections.abc import Awaitable, Callable, Iterator, 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
@ -340,7 +342,7 @@ def _explicit_causes(exc: BaseException) -> Iterator[BaseException]:
current = current.__cause__
def _is_redis_timeout_failure(exc: BaseException) -> bool:
def is_redis_timeout_failure(exc: BaseException) -> bool:
"""True when ``exc`` or any exception it was explicitly raised ``from`` is a timeout.
redis-py's blocking pool reports a pool wait timeout as ``ConnectionError`` chained from
@ -414,7 +416,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)
@ -422,13 +424,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.monotonic) -> 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)
@ -475,7 +522,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
@ -492,7 +539,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
@ -801,10 +848,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
@ -1010,11 +1055,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
@ -1062,10 +1104,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)
@ -1112,7 +1152,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)
@ -1149,10 +1188,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)
@ -1191,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)
@ -1235,10 +1278,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
@ -1274,10 +1315,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)
@ -1359,10 +1401,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
@ -1448,7 +1491,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]:
@ -1526,7 +1571,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
@ -1645,7 +1690,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
@ -1870,9 +1915,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
@ -1949,7 +1996,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(
@ -2017,9 +2064,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
@ -2095,7 +2144,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(
@ -2206,8 +2255,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

1
litellm/constants.py
View file

@ -461,6 +461,7 @@ 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))
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

8
litellm/proxy/proxy_server.py
View file

@ -250,7 +250,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,
@ -3450,8 +3450,10 @@ async def _invalidate_spend_counter(counter_key: str):
async def _apply_spend_counter_increments(pending: Sequence[PendingSpendIncrement]) -> None:
try:
await increment_spend_counters_pipeline(pending=pending)
except RedisCircuitBreakerOpenError:
return
except Exception as e:
if isinstance(e, RedisCircuitBreakerOpenError) or is_redis_timeout_failure(e):
return
raise
async def increment_spend_counters_pipeline(pending: Sequence[PendingSpendIncrement]) -> None:

26
tests/test_litellm/caching/test_caching.py
View file

@ -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",

55
tests/test_litellm/caching/test_dual_cache.py
View file

@ -704,3 +704,58 @@ 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 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
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)",
)
]

145
tests/test_litellm/caching/test_redis_cache.py
View file

@ -978,17 +978,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"
@ -1000,19 +1000,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"
@ -1023,7 +1023,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
@ -1035,9 +1035,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
@ -1205,6 +1205,117 @@ async def test_a_probe_overtaken_by_a_later_outage_leaves_the_breaker_to_the_new
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 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
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
@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_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(
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
@pytest.mark.asyncio
async def test_pool_wait_timeout_is_a_timeout_failure_not_hard_connectivity():
"""A saturated blocking pool must not open the breaker before the timeout minimum duration.
@ -1241,7 +1352,7 @@ async def test_pool_wait_timeout_is_a_timeout_failure_not_hard_connectivity():
def test_timeout_classification_follows_the_explicit_cause_chain_only():
from redis.exceptions import ConnectionError as RedisConnectionError
from litellm.caching.redis_cache import _is_redis_timeout_failure
from litellm.caching.redis_cache import is_redis_timeout_failure
def raise_chained_from_timeout() -> None:
try:
@ -1260,9 +1371,9 @@ def test_timeout_classification_follows_the_explicit_cause_chain_only():
with pytest.raises(RedisConnectionError) as contextual:
raise_while_handling_timeout()
assert _is_redis_timeout_failure(chained.value) is True
assert _is_redis_timeout_failure(contextual.value) is False
assert _is_redis_timeout_failure(RedisConnectionError("refused")) is False
assert is_redis_timeout_failure(chained.value) is True
assert is_redis_timeout_failure(contextual.value) is False
assert is_redis_timeout_failure(RedisConnectionError("refused")) is False
class _RoundTripCountingRedis:

18
tests/test_litellm/proxy/proxy_server/test_spend_counters.py
View file

@ -1180,6 +1180,24 @@ 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 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()
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()