scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 213

fix(proxy): treat a Redis timeout in spend counter increments as an already-logged cache failure

Browse source 
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>
This commit is contained in:
yassin 2026-09-12 01:25:32 +00:00
parent 107b4ec4db
commit 9c84e98fb2
5 changed files with 42 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/caching/redis_cache.py
View file

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

2
litellm/constants.py
View file

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

4
litellm/proxy/proxy_server.py
View file

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

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

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

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

@ -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()