scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 124

fix(redis): quiet every per-request Redis fallback while the breaker is open

Browse source
This commit is contained in:
mateo-berri 2026-09-10 14:18:53 -07:00
parent ad607516a2
commit 05f459d898
11 changed files with 228 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
litellm/caching/dual_cache.py
View file

@ -178,7 +178,9 @@ class DualCache(BaseCache):
print_verbose(f"get cache: cache result: {result}")
return result
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in get_cache", e)
log_redis_failure(
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in get_cache", e, with_traceback=True
)
def batch_get_cache(
self,
@ -218,7 +220,9 @@ class DualCache(BaseCache):
redis_result.get(key) if value is None else value for key, value in zip(keys, result)
)
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in batch_get_cache", e)
log_redis_failure(
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in batch_get_cache", e, with_traceback=True
)
async def async_get_cache(
self,
@ -251,7 +255,9 @@ class DualCache(BaseCache):
print_verbose(f"get cache: cache result: {result}")
return result
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async_get_cache", e)
log_redis_failure(
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async_get_cache", e, with_traceback=True
)
def _reserve_redis_batch_keys(
self,
@ -340,7 +346,13 @@ class DualCache(BaseCache):
return result
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async_batch_get_cache", e)
log_redis_failure(
verbose_logger,
logging.ERROR,
"LiteLLM Cache: exception in async_batch_get_cache",
e,
with_traceback=True,
)
async def async_set_cache(self, key, value, local_only: bool = False, **kwargs):
print_verbose(f"async set cache: cache key: {key}; local_only: {local_only}; value: {value}")
@ -353,7 +365,9 @@ class DualCache(BaseCache):
if self.redis_cache is not None and local_only is False:
await self.redis_cache.async_set_cache(key, value, **kwargs)
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async add_cache", e)
log_redis_failure(
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async add_cache", e, with_traceback=True
)
# async_batch_set_cache
async def async_set_cache_pipeline(self, cache_list: list, local_only: bool = False, **kwargs):
@ -372,7 +386,9 @@ class DualCache(BaseCache):
cache_list=cache_list, ttl=kwargs.pop("ttl", None), **kwargs
)
except Exception as e:
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async add_cache", e)
log_redis_failure(
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async add_cache", e, with_traceback=True
)
async def async_increment_cache(
self,

6
litellm/caching/redis_cache.py
View file

@ -396,7 +396,9 @@ class RedisCircuitBreakerOpenError(Exception):
"""Raised in place of a Redis call while the circuit breaker is open."""
def log_redis_failure(logger: logging.Logger, level: int, message: str, exc: BaseException) -> None:
def log_redis_failure(
logger: logging.Logger, level: int, message: str, exc: BaseException, with_traceback: bool = False
) -> None:
"""Log a Redis failure the caller is about to swallow.
An open breaker refuses every call until Redis recovers and announced itself once when it
@ -405,7 +407,7 @@ def log_redis_failure(logger: logging.Logger, level: int, message: str, exc: Bas
if isinstance(exc, RedisCircuitBreakerOpenError):
logger.debug("%s: %s", message, exc)
return
logger.log(level, "%s: %s", message, exc, exc_info=exc if level >= logging.ERROR else None)
logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None)
def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> int:

23
litellm/proxy/hooks/batch_enqueued_tokens.py
View file

@ -9,6 +9,7 @@ the reservation is refunded when the batch reaches a terminal state
"""
import asyncio
import logging
import math
import time
import uuid
@ -19,6 +20,7 @@ from typing import TYPE_CHECKING, Annotated, Final, Literal, Protocol, TypeAlias
from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, ValidationError
from litellm._logging import verbose_proxy_logger
from litellm.caching.redis_cache import log_redis_failure
from litellm.constants import BATCH_ENQUEUED_TOKEN_LIMIT_METADATA_KEY, BATCH_ENQUEUED_TOKEN_TTL_SECONDS
from litellm.proxy._types import UserAPIKeyAuth
@ -233,8 +235,11 @@ class BatchEnqueuedTokenStore:
try:
return await self._reserve_via_redis(reserve_script, refund_script, tokens=tokens, scopes=scopes)
except Exception as e: # noqa: BLE001 # any Redis failure must fall back to the in-memory counters
verbose_proxy_logger.warning(
"Redis enqueued-token reserve failed, falling back to in-memory: %s", str(e)
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reserve failed, falling back to in-memory",
e,
)
return await self._reserve_in_memory(tokens=tokens, scopes=scopes, span=litellm_parent_otel_span)
@ -374,8 +379,11 @@ class BatchEnqueuedTokenStore:
(serialized, ttl),
)
except Exception as e: # noqa: BLE001 # any Redis failure must fall back to the in-memory record
verbose_proxy_logger.warning(
"Redis enqueued-token reservation save failed, falling back to in-memory: %s", str(e)
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reservation save failed, falling back to in-memory",
e,
)
else:
return
@ -421,8 +429,11 @@ class BatchEnqueuedTokenStore:
await pop_script((self._record_key(batch_id),), (BATCH_ENQUEUED_TOKEN_TTL_SECONDS,))
)
except Exception as e: # noqa: BLE001 # any Redis failure must fall back to the in-memory record
verbose_proxy_logger.warning(
"Redis enqueued-token reservation pop failed, falling back to in-memory: %s", str(e)
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"Redis enqueued-token reservation pop failed, falling back to in-memory",
e,
)
return None

18
litellm/proxy/hooks/max_budget_per_session_limiter.py
View file

@ -14,11 +14,13 @@ Works across multiple proxy instances via DualCache (in-memory + Redis).
Follows the same pattern as max_iterations_limiter.py.
"""
import logging
import os
from typing import TYPE_CHECKING, Any, Final
from litellm import DualCache
from litellm._logging import verbose_proxy_logger
from litellm.caching.redis_cache import log_redis_failure
from litellm.exceptions import RateLimitType
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy._types import UserAPIKeyAuth
@ -215,9 +217,11 @@ class _PROXY_MaxBudgetPerSessionHandler(CustomLogger):
return float(result)
return 0.0
except Exception as e:
verbose_proxy_logger.warning(
"MaxBudgetPerSessionHandler: Redis GET failed, falling back to in-memory: %s",
str(e),
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"MaxBudgetPerSessionHandler: Redis GET failed, falling back to in-memory",
e,
)
result = await self.internal_usage_cache.async_get_cache(
@ -239,9 +243,11 @@ class _PROXY_MaxBudgetPerSessionHandler(CustomLogger):
)
return float(result)
except Exception as e:
verbose_proxy_logger.warning(
"MaxBudgetPerSessionHandler: Redis INCRBYFLOAT failed, falling back to in-memory: %s",
str(e),
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"MaxBudgetPerSessionHandler: Redis INCRBYFLOAT failed, falling back to in-memory",
e,
)
return await self._in_memory_increment_spend(cache_key, amount)

47
litellm/proxy/hooks/parallel_request_limiter_v3.py
View file

@ -1230,7 +1230,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
all_cache_values.extend(group_cache_values)
except Exception as e:
verbose_proxy_logger.warning("Redis Lua script failed for hash tag %s: %s", hash_tag, e)
log_redis_failure(
verbose_proxy_logger, logging.WARNING, f"Redis Lua script failed for hash tag {hash_tag}", e
)
# Fallback to in-memory cache for this group
group_cache_values = await self.in_memory_cache_sliding_window(
keys=group_keys,
@ -1477,7 +1479,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
counts = [max(0, int(value)) for value in raw_counts]
except Exception as e: # noqa: BLE001 - any Redis/Lua failure degrades to the local mirror, never a 500
verbose_proxy_logger.warning("parallel_count_script failed, using local mirror: %s", e)
log_redis_failure(
verbose_proxy_logger, logging.WARNING, "parallel_count_script failed, using local mirror", e
)
counts = await self._read_local_gauge_counts(gauge_keys, parent_otel_span)
else:
counts = await self._read_local_gauge_counts(gauge_keys, parent_otel_span)
@ -1507,7 +1511,12 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
],
)
except Exception as e: # noqa: BLE001 - any Redis/Lua failure degrades to in-memory enforcement, never a 500
verbose_proxy_logger.warning("parallel_acquire_script failed, falling back to in-memory gauge: %s", e)
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"parallel_acquire_script failed, falling back to in-memory gauge",
e,
)
async with self._check_and_increment_lock:
return await self._acquire_parallel_slots_in_memory(gauges, slot_id, parent_otel_span)
if int(raw[0]) == 1:
@ -1633,7 +1642,12 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
return
except Exception as e: # noqa: BLE001 - any Redis/Lua failure degrades to the in-memory release, never a 500
verbose_proxy_logger.warning("parallel_release_script failed, falling back to in-memory release: %s", e)
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"parallel_release_script failed, falling back to in-memory release",
e,
)
async with self._check_and_increment_lock:
for counter_key in counter_keys:
@ -1816,12 +1830,13 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
# state ambiguous. Refund any prior groups so Redis returns
# to its pre-call state, then fall back to in-memory for the
# whole call (counters there are independent of Redis).
verbose_proxy_logger.error(
"atomic_check_and_increment_by_n: Redis Lua execution failed (%s: %s). Refunding %s prior descriptors and falling back to in-memory enforcement — counters will diverge from Redis until window expires (window_size=%ss).",
type(e).__name__,
log_redis_failure(
verbose_proxy_logger,
logging.ERROR,
f"atomic_check_and_increment_by_n: Redis Lua execution failed ({type(e).__name__}). Refunding "
f"{len(applied)} prior descriptors and falling back to in-memory enforcement, counters will "
f"diverge from Redis until window expires (window_size={self.window_size}s)",
e,
len(applied),
self.window_size,
)
await self._refund_applied_descriptor_groups(applied)
flat_meta: list[AtomicCounterMeta] = [m for _k, _a, group_meta in descriptor_groups for m in group_meta]
@ -1868,8 +1883,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
value=-entry["increment"],
)
except Exception as e:
verbose_proxy_logger.warning(
"Failed to refund %s on cross-descriptor rollback: %s", entry["counter_key"], e
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
f"Failed to refund {entry['counter_key']} on cross-descriptor rollback",
e,
)
def _build_atomic_response(
@ -3926,9 +3944,10 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
continue
except Exception as e: # noqa: BLE001 # Redis failures use the plain increment fallback
verbose_proxy_logger.warning(
"Window-guarded token adjustment failed for %s: %s",
operation["key"],
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
f"Window-guarded token adjustment failed for {operation['key']}",
e,
)
if operation["increment_value"] > 0:

18
litellm/proxy/hooks/sensitive_data_routing.py
View file

@ -10,11 +10,13 @@ this hook manages:
Works across multiple proxy instances via DualCache (in-memory + Redis).
"""
import logging
import os
from typing import TYPE_CHECKING, Any, Final
from litellm._logging import verbose_proxy_logger
from litellm.caching.caching import DualCache
from litellm.caching.redis_cache import log_redis_failure
from litellm.integrations.custom_guardrail import get_session_id_from_request_data
from litellm.integrations.custom_logger import CustomLogger
from litellm.proxy._types import UserAPIKeyAuth
@ -96,9 +98,11 @@ class _PROXY_SensitiveDataRoutingHandler(CustomLogger):
)
return routed_model
except Exception as e:
verbose_proxy_logger.warning(
"SensitiveDataRoutingHandler: Redis GET failed, falling back to in-memory: %s",
str(e),
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"SensitiveDataRoutingHandler: Redis GET failed, falling back to in-memory",
e,
)
result = await self.internal_usage_cache.async_get_cache(
@ -142,9 +146,11 @@ class _PROXY_SensitiveDataRoutingHandler(CustomLogger):
ttl=self.ttl,
)
except Exception as e:
verbose_proxy_logger.warning(
"SensitiveDataRoutingHandler: Redis SET failed, falling back to in-memory: %s",
str(e),
log_redis_failure(
verbose_proxy_logger,
logging.WARNING,
"SensitiveDataRoutingHandler: Redis SET failed, falling back to in-memory",
e,
)
await self.internal_usage_cache.async_set_cache(

7
litellm/router.py
View file

@ -53,6 +53,7 @@ from litellm.caching.caching import (
RedisCache,
RedisClusterCache,
)
from litellm.caching.redis_cache import log_redis_failure
from litellm.constants import (
CLIENT_OUTPUT_CEILING_METADATA_KEY,
CONSUMED_REQUEST_TAGS_METADATA_KEY,
@ -13338,8 +13339,10 @@ class Router:
return await session_cache.async_get_cache(key=cache_key)
return await session_cache.redis_cache.async_get_cache(key=cache_key)
except Exception as e: # noqa: BLE001 # an optional binding must not make routing depend on Redis
verbose_router_logger.warning(
"Failed to read Claude Code session router binding; using the requested model: %s",
log_redis_failure(
verbose_router_logger,
logging.WARNING,
"Failed to read Claude Code session router binding; using the requested model",
e,
)
return None

36
tests/test_litellm/proxy/hooks/test_max_budget_per_session_limiter.py
View file

@ -10,10 +10,12 @@ Tests that session-scoped budget tracking works correctly:
from unittest.mock import patch
import logging
import pytest
from fastapi import HTTPException
from litellm.caching.caching import DualCache
from litellm.caching.redis_cache import _redis_circuit_breaker_guard
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.hooks.max_budget_per_session_limiter import (
_PROXY_MaxBudgetPerSessionHandler,
@ -163,3 +165,37 @@ async def test_no_agent_id_passes():
call_type="",
)
assert result is None
class _OpenBreakerRedis:
def __init__(self) -> None:
from litellm.caching.redis_cache import RedisCircuitBreaker
self._circuit_breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
for _ in range(3):
self._circuit_breaker.record_failure()
@_redis_circuit_breaker_guard
async def async_get_cache(self, key, **kwargs):
raise AssertionError("never reached")
def async_register_script(self, script):
@_redis_circuit_breaker_guard
async def refused(_self, keys, args):
raise AssertionError("never reached")
return lambda keys, args: refused(self, keys, args)
@pytest.mark.asyncio
async def test_an_open_circuit_breaker_reads_session_spend_locally_without_a_warning(caplog):
cache = DualCache(redis_cache=_OpenBreakerRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
handler = _PROXY_MaxBudgetPerSessionHandler(internal_usage_cache=InternalUsageCache(cache))
caplog.clear()
with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
spend = await handler._get_current_spend("{session_budget:quiet}:spend")
assert spend == 0.0
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)

21
tests/test_litellm/proxy/hooks/test_parallel_request_limiter_v3.py
View file

@ -6257,3 +6257,24 @@ async def test_an_open_circuit_breaker_falls_back_to_the_pipeline_without_a_warn
assert await handler.internal_usage_cache.dual_cache.async_get_cache("quiet_key") == 10.0
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
@pytest.mark.asyncio
async def test_an_open_circuit_breaker_reads_the_sliding_window_locally_without_a_warning(caplog):
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(DualCache()))
async def refused_script(keys, args):
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
handler.batch_rate_limiter_script = refused_script
with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
values = await handler._execute_redis_batch_rate_limiter_script(
["{quiet}:window", "{quiet}:counter"], now_int=int(time.time())
)
assert isinstance(values, list)
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)

35
tests/test_litellm/proxy/hooks/test_sensitive_data_routing.py
View file

@ -6,6 +6,7 @@ This feature allows guardrails to route requests to a different model
All subsequent requests in the same session are routed to the same model.
"""
import logging
import asyncio
from typing import Any, Dict, Optional
from unittest.mock import AsyncMock, MagicMock, patch
@ -13,12 +14,14 @@ from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from litellm.caching.caching import DualCache
from litellm.caching.redis_cache import _redis_circuit_breaker_guard
from litellm.exceptions import SensitiveDataRouteException
from litellm.integrations.custom_guardrail import (
CustomGuardrail,
get_session_id_from_request_data,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.utils import InternalUsageCache
from litellm.proxy.hooks.sensitive_data_routing import (
_PROXY_SensitiveDataRoutingHandler,
SENSITIVE_ROUTING_CACHE_PREFIX,
@ -1034,3 +1037,35 @@ class TestPreCallHookDeferredRouting:
metrics_kwargs = prom._record_guardrail_metrics.call_args.kwargs
assert metrics_kwargs["status"] == "intervened"
assert metrics_kwargs["error_type"] is None
class _OpenBreakerRedis:
def __init__(self) -> None:
from litellm.caching.redis_cache import RedisCircuitBreaker
self._circuit_breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
for _ in range(3):
self._circuit_breaker.record_failure()
@_redis_circuit_breaker_guard
async def async_get_cache(self, key, **kwargs):
raise AssertionError("never reached")
@_redis_circuit_breaker_guard
async def async_set_cache(self, key, value, **kwargs):
raise AssertionError("never reached")
@pytest.mark.asyncio
async def test_an_open_circuit_breaker_keeps_session_routing_in_memory_without_a_warning(caplog):
cache = DualCache(redis_cache=_OpenBreakerRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
handler = _PROXY_SensitiveDataRoutingHandler(internal_usage_cache=InternalUsageCache(cache))
caplog.clear()
with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
await handler.set_session_routing("quiet-session", "safe-model")
routed = await handler._get_routed_model("quiet-session", None)
assert routed == "safe-model"
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)

31
tests/test_litellm/test_router.py
View file

@ -19,6 +19,8 @@ import respx
import litellm
from litellm.caching.caching import DualCache
from litellm.caching.redis_cache import _redis_circuit_breaker_guard
from litellm import Router
from litellm.exceptions import MidStreamFallbackError
from litellm.integrations.custom_guardrail import CustomGuardrail
@ -15161,3 +15163,32 @@ def test_cached_model_info_lookups_match_uncached_and_reset_on_model_list_change
assert router.cached_deployment_model_info("dep-a", "openai/gpt-4o")["max_output_tokens"] == 200
assert router.cached_model_group_info("grp").max_output_tokens == 200
class _OpenBreakerRedis:
def __init__(self) -> None:
from litellm.caching.redis_cache import RedisCircuitBreaker
self._circuit_breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60)
for _ in range(3):
self._circuit_breaker.record_failure()
@_redis_circuit_breaker_guard
async def async_get_cache(self, key, **kwargs):
raise AssertionError("never reached")
@pytest.mark.asyncio
async def test_an_open_circuit_breaker_skips_the_session_binding_without_a_warning(caplog):
router = litellm.Router(
model_list=[{"model_name": "haiku", "litellm_params": {"model": "anthropic/claude-haiku-4-5", "api_key": "k"}}]
)
router._claude_code_session_router_cache = DualCache(redis_cache=_OpenBreakerRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
caplog.clear()
with caplog.at_level(logging.DEBUG, logger="LiteLLM Router"):
binding = await router._get_claude_code_session_router_binding("quiet-session")
assert binding is None
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)