mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
Merge pull request #40620 from BerriAI/litellm_redis_breaker_quiet_open
fix(redis): log an open circuit breaker once instead of a traceback per request and count sync timeouts as timeouts
This commit is contained in:
Mateo Wang
GitHub
commit
ef1a37795c
17 changed files with 474 additions and 63 deletions
|
|
@ -10,6 +10,7 @@
|
|||
import ast
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
import traceback
|
||||
from collections.abc import Mapping
|
||||
|
|
@ -32,7 +33,7 @@ from .dual_cache import DualCache # noqa: F401
|
|||
from .gcs_cache import GCSCache
|
||||
from .in_memory_cache import InMemoryCache
|
||||
from .qdrant_semantic_cache import QdrantSemanticCache
|
||||
from .redis_cache import RedisCache
|
||||
from .redis_cache import RedisCache, log_redis_failure
|
||||
from .redis_cluster_cache import RedisClusterCache
|
||||
from .redis_semantic_cache import RedisSemanticCache
|
||||
from .s3_cache import S3Cache
|
||||
|
|
@ -678,7 +679,7 @@ 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:
|
||||
verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e)
|
||||
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
|
||||
|
||||
async def async_add_cache(self, result, dynamic_cache_object: BaseCache | None = None, **kwargs):
|
||||
"""
|
||||
|
|
@ -697,7 +698,7 @@ class Cache:
|
|||
else:
|
||||
await self.cache.async_set_cache(cache_key, cached_data, **kwargs)
|
||||
except Exception as e:
|
||||
verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e)
|
||||
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
|
||||
|
||||
def _convert_to_cached_embedding(
|
||||
self,
|
||||
|
|
@ -876,7 +877,7 @@ class Cache:
|
|||
else:
|
||||
await self.cache.async_set_cache_pipeline(cache_list=cache_list, **kwargs)
|
||||
except Exception as e:
|
||||
verbose_logger.exception("LiteLLM Cache: Excepton add_cache: %s", e)
|
||||
log_redis_failure(verbose_logger, logging.ERROR, "LiteLLM Cache: exception in add_cache", e)
|
||||
|
||||
def should_use_cache(self, **kwargs):
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -8,8 +8,8 @@ Has 4 primary methods:
|
|||
- async_get_cache
|
||||
"""
|
||||
|
||||
import logging
|
||||
import time
|
||||
import traceback
|
||||
from collections.abc import Sequence
|
||||
from threading import Lock
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
|
@ -23,7 +23,7 @@ from litellm.constants import DEFAULT_MAX_REDIS_BATCH_CACHE_SIZE
|
|||
|
||||
from .base_cache import BaseCache
|
||||
from .in_memory_cache import InMemoryCache
|
||||
from .redis_cache import RedisCache
|
||||
from .redis_cache import RedisCache, log_redis_failure
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from opentelemetry.trace import Span as _Span
|
||||
|
|
@ -177,8 +177,10 @@ class DualCache(BaseCache):
|
|||
|
||||
print_verbose(f"get cache: cache result: {result}")
|
||||
return result
|
||||
except Exception:
|
||||
verbose_logger.error(traceback.format_exc())
|
||||
except Exception as e:
|
||||
log_redis_failure(
|
||||
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in get_cache", e, with_traceback=True
|
||||
)
|
||||
|
||||
def batch_get_cache(
|
||||
self,
|
||||
|
|
@ -217,8 +219,10 @@ class DualCache(BaseCache):
|
|||
return list( # mutable-ok: public list contract
|
||||
redis_result.get(key) if value is None else value for key, value in zip(keys, result)
|
||||
)
|
||||
except Exception:
|
||||
verbose_logger.error(traceback.format_exc())
|
||||
except Exception as 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,
|
||||
|
|
@ -250,8 +254,10 @@ class DualCache(BaseCache):
|
|||
|
||||
print_verbose(f"get cache: cache result: {result}")
|
||||
return result
|
||||
except Exception:
|
||||
verbose_logger.error(traceback.format_exc())
|
||||
except Exception as 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,
|
||||
|
|
@ -339,8 +345,14 @@ class DualCache(BaseCache):
|
|||
await self.in_memory_cache.async_set_cache(key, value, **self._backfill_kwargs(kwargs))
|
||||
|
||||
return result
|
||||
except Exception:
|
||||
verbose_logger.error(traceback.format_exc())
|
||||
except Exception as 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:
|
||||
verbose_logger.exception("LiteLLM Cache: Excepton async add_cache: %s", 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:
|
||||
verbose_logger.exception("LiteLLM Cache: Excepton async add_cache: %s", 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,
|
||||
|
|
@ -410,8 +426,10 @@ class DualCache(BaseCache):
|
|||
|
||||
return result
|
||||
except Exception as e:
|
||||
verbose_logger.warning(
|
||||
"Redis async_increment_cache failed, falling back to in-memory result: %s",
|
||||
log_redis_failure(
|
||||
verbose_logger,
|
||||
logging.WARNING,
|
||||
"Redis async_increment_cache failed, falling back to in-memory result",
|
||||
e,
|
||||
)
|
||||
return result
|
||||
|
|
@ -439,8 +457,10 @@ class DualCache(BaseCache):
|
|||
|
||||
return result
|
||||
except Exception as e:
|
||||
verbose_logger.warning(
|
||||
"Redis async_increment_cache_pipeline failed, falling back to in-memory result: %s",
|
||||
log_redis_failure(
|
||||
verbose_logger,
|
||||
logging.WARNING,
|
||||
"Redis async_increment_cache_pipeline failed, falling back to in-memory result",
|
||||
e,
|
||||
)
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import functools
|
|||
import hashlib
|
||||
import inspect
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from collections.abc import Awaitable, Callable, Sequence
|
||||
from contextvars import ContextVar
|
||||
|
|
@ -391,10 +392,23 @@ def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseExcep
|
|||
_swallowed_redis_failures.set(_swallowed_redis_failures.get() + 1)
|
||||
|
||||
|
||||
class RedisCircuitBreakerOpenError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
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)
|
||||
return
|
||||
logger.log(level, "%s: %s", message, exc, exc_info=exc if with_traceback else None)
|
||||
|
||||
|
||||
def _enter_circuit_breaker(breaker: RedisCircuitBreaker, name: str) -> int:
|
||||
"""Reject the call if the breaker is open, else return the swallowed-failure count to compare against."""
|
||||
if breaker.is_open():
|
||||
raise Exception(f"Redis circuit breaker is open — skipping {name}")
|
||||
raise RedisCircuitBreakerOpenError(f"Redis circuit breaker is open — skipping {name}")
|
||||
return _swallowed_redis_failures.get()
|
||||
|
||||
|
||||
|
|
@ -440,7 +454,7 @@ def _run_under_circuit_breaker_sync(
|
|||
result: Final = call()
|
||||
except Exception as e:
|
||||
if _is_redis_health_failure(e):
|
||||
breaker.record_failure()
|
||||
breaker.record_failure(is_timeout=_is_redis_timeout_failure(e))
|
||||
raise
|
||||
_exit_circuit_breaker(breaker, swallowed_before)
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ This is currently in development and not yet ready for production.
|
|||
|
||||
import asyncio
|
||||
import binascii
|
||||
import logging
|
||||
import os
|
||||
import uuid
|
||||
from collections.abc import Awaitable, Callable, Mapping, Sequence, Set
|
||||
|
|
@ -26,6 +27,7 @@ from typing_extensions import NotRequired, ReadOnly
|
|||
|
||||
from litellm import DualCache
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.caching.redis_cache import log_redis_failure
|
||||
from litellm.constants import DYNAMIC_RATE_LIMIT_ERROR_THRESHOLD_PER_MINUTE, INTERNAL_CALL_ORIGIN_METADATA_KEY
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
|
|
@ -1228,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,
|
||||
|
|
@ -1475,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)
|
||||
|
|
@ -1505,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:
|
||||
|
|
@ -1631,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:
|
||||
|
|
@ -1814,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]
|
||||
|
|
@ -1866,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(
|
||||
|
|
@ -3856,7 +3876,9 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
|
|||
)
|
||||
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.warning("TTL preservation failed, falling back to regular pipeline: %s", e)
|
||||
log_redis_failure(
|
||||
verbose_proxy_logger, logging.WARNING, "TTL preservation failed, falling back to regular pipeline", e
|
||||
)
|
||||
# Fallback to regular pipeline on error
|
||||
await self.internal_usage_cache.dual_cache.async_increment_cache_pipeline(
|
||||
increment_list=pipeline_operations,
|
||||
|
|
@ -3922,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:
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -13395,8 +13396,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
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import logging
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
|
|
@ -6,6 +7,7 @@ from typing_extensions import ReadOnly, TypedDict
|
|||
|
||||
from litellm._logging import verbose_router_logger
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.caching.redis_cache import log_redis_failure
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
|
||||
IN_FLIGHT_COUNT_TTL_SECONDS: Final = 60 * 60
|
||||
|
|
@ -87,16 +89,22 @@ def _least_busy(
|
|||
|
||||
|
||||
def _warn_unreadable(model_group: str, error: Exception) -> None:
|
||||
verbose_router_logger.warning(
|
||||
"least-busy routing could not read the shared in-flight counts for %s, "
|
||||
"falling back to this worker's own counts: %s",
|
||||
model_group,
|
||||
log_redis_failure(
|
||||
verbose_router_logger,
|
||||
logging.WARNING,
|
||||
f"least-busy routing could not read the shared in-flight counts for {model_group}, "
|
||||
"falling back to this worker's own counts",
|
||||
error,
|
||||
)
|
||||
|
||||
|
||||
def _warn_unwritable(key: str, error: Exception) -> None:
|
||||
verbose_router_logger.warning("least-busy routing could not update the in-flight count under %s: %s", key, error)
|
||||
log_redis_failure(
|
||||
verbose_router_logger,
|
||||
logging.WARNING,
|
||||
f"least-busy routing could not update the in-flight count under {key}",
|
||||
error,
|
||||
)
|
||||
|
||||
|
||||
class LeastBusyLoggingHandler(CustomLogger):
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import asyncio
|
||||
import logging
|
||||
import time
|
||||
import uuid
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
|
@ -7,7 +8,8 @@ import pytest
|
|||
|
||||
from litellm.caching.dual_cache import DualCache
|
||||
from litellm.caching.in_memory_cache import InMemoryCache
|
||||
from litellm.caching.redis_cache import RedisCache
|
||||
from litellm.caching.redis_cache import RedisCache, _redis_circuit_breaker_guard, _redis_circuit_breaker_guard_sync
|
||||
from litellm.types.caching import RedisPipelineIncrementOperation
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -576,3 +578,102 @@ async def test_dual_cache_late_attach_redis_wires_writes_and_ttl_async():
|
|||
assert mock_redis.async_set_cache.call_args[0][:2] == (key_after, val_after)
|
||||
|
||||
assert in_memory.get_cache(key_after) == val_after
|
||||
|
||||
|
||||
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_batch_get_cache(self, key_list, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard
|
||||
async def async_set_cache(self, key, value, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard
|
||||
async def async_set_cache_pipeline(self, cache_list, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard
|
||||
async def async_increment_pipeline(self, increment_list, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard
|
||||
async def async_increment(self, key, value, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard_sync
|
||||
def get_cache(self, key, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
@_redis_circuit_breaker_guard_sync
|
||||
def batch_get_cache(self, key_list, **kwargs):
|
||||
raise AssertionError("never reached")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"call",
|
||||
[
|
||||
lambda cache: cache.async_get_cache("k"),
|
||||
lambda cache: cache.async_batch_get_cache(["k1", "k2"]),
|
||||
lambda cache: cache.async_set_cache("k", "v"),
|
||||
lambda cache: cache.async_set_cache_pipeline([("k", "v")]),
|
||||
lambda cache: cache.async_increment_cache_pipeline(
|
||||
increment_list=[RedisPipelineIncrementOperation(key="k", increment_value=1.0, ttl=60)]
|
||||
),
|
||||
lambda cache: cache.async_increment_cache("k", 1.0),
|
||||
],
|
||||
ids=["get", "batch_get", "set", "set_pipeline", "increment_pipeline", "increment"],
|
||||
)
|
||||
async def test_an_open_circuit_breaker_is_not_an_error_per_request(caplog, call):
|
||||
cache = DualCache(in_memory_cache=InMemoryCache(), redis_cache=_OpenBreakerRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
caplog.clear()
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
|
||||
await call(cache)
|
||||
|
||||
assert [record.levelno for record in caplog.records if record.levelno >= logging.WARNING] == []
|
||||
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"call",
|
||||
[lambda cache: cache.get_cache("k"), lambda cache: cache.batch_get_cache(["k1", "k2"])],
|
||||
ids=["get", "batch_get"],
|
||||
)
|
||||
def test_an_open_circuit_breaker_is_not_an_error_per_sync_request(caplog, call):
|
||||
cache = DualCache(in_memory_cache=InMemoryCache(), redis_cache=_OpenBreakerRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
caplog.clear()
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
|
||||
call(cache)
|
||||
|
||||
assert [record.levelno for record in caplog.records if record.levelno >= logging.WARNING] == []
|
||||
assert any("circuit breaker is open" in record.getMessage() for record in caplog.records)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_real_redis_failure_still_logs_an_error(caplog):
|
||||
class _BrokenRedis:
|
||||
async def async_get_cache(self, key, **kwargs):
|
||||
raise ConnectionError("redis is down")
|
||||
|
||||
cache = DualCache(in_memory_cache=InMemoryCache(), redis_cache=_BrokenRedis()) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM"):
|
||||
assert await cache.async_get_cache("k") is None
|
||||
|
||||
errors = [record for record in caplog.records if record.levelno == logging.ERROR]
|
||||
assert [record.getMessage() for record in errors] == ["LiteLLM Cache: exception in async_get_cache: redis is down"]
|
||||
assert errors[0].exc_info is not None
|
||||
|
|
|
|||
|
|
@ -1013,3 +1013,20 @@ async def test_breaker_metrics_track_state_and_failure_class():
|
|||
breaker.record_success()
|
||||
assert sample("litellm_redis_circuit_breaker_state", {"state": "open"}) == open_gauge_before
|
||||
assert sample("litellm_redis_circuit_breaker_state", {"state": "closed"}) == closed_gauge_before + 1
|
||||
|
||||
|
||||
def test_sync_guard_counts_a_timeout_as_a_timeout():
|
||||
from redis.exceptions import TimeoutError as RedisTimeoutError
|
||||
|
||||
from litellm.caching.redis_cache import RedisCircuitBreaker, _run_under_circuit_breaker_sync
|
||||
|
||||
breaker = RedisCircuitBreaker(failure_threshold=3, recovery_timeout=60, timeout_min_duration=5.0)
|
||||
|
||||
def timing_out_call() -> str:
|
||||
raise RedisTimeoutError("read timed out")
|
||||
|
||||
for _ in range(6):
|
||||
with pytest.raises(RedisTimeoutError):
|
||||
_run_under_circuit_breaker_sync(breaker, "op", timing_out_call)
|
||||
|
||||
assert breaker.is_open() is False
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ response-shape helpers the v3 limiter's post-call hooks rely on.
|
|||
"""
|
||||
|
||||
import base64
|
||||
import logging
|
||||
import socket
|
||||
import uuid
|
||||
from collections.abc import Mapping, Sequence
|
||||
|
|
@ -16,6 +17,7 @@ from typing import Final
|
|||
import pytest
|
||||
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
|
||||
from litellm.constants import BATCH_ENQUEUED_TOKEN_TTL_SECONDS
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.proxy.hooks.batch_enqueued_tokens import (
|
||||
|
|
@ -439,3 +441,29 @@ async def test_redis_lua_path_full_lifecycle():
|
|||
refill = await store.reserve(tokens=50, scopes=(key_scope, team_scope))
|
||||
assert isinstance(refill, BatchEnqueuedTokenReservation)
|
||||
await store.refund(refill)
|
||||
|
||||
|
||||
class _OpenBreakerRedis:
|
||||
def async_register_script(self, script: str):
|
||||
async def refused(keys: Sequence[str], args: Sequence[str | bytes | int | float]) -> object:
|
||||
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
|
||||
|
||||
return refused
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_open_circuit_breaker_keeps_reservations_in_memory_without_a_warning(caplog):
|
||||
scope = _scope(limit=100)
|
||||
store = BatchEnqueuedTokenStore(
|
||||
internal_usage_cache=InternalUsageCache(DualCache(redis_cache=_OpenBreakerRedis(), default_in_memory_ttl=60)) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
|
||||
reservation = await store.reserve(tokens=60, scopes=(scope,))
|
||||
assert isinstance(reservation, BatchEnqueuedTokenReservation)
|
||||
assert reservation.backend == "memory"
|
||||
await store.save_reservation("batch_quiet", reservation)
|
||||
assert await store.pop_reservation("batch_quiet") == reservation
|
||||
|
||||
assert not [record for record in caplog.records if record.levelno >= logging.WARNING]
|
||||
assert sum("circuit breaker is open" in record.getMessage() for record in caplog.records) == 3
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -6236,3 +6236,51 @@ async def test_post_call_success_hook_leaves_raw_provider_dict_untouched():
|
|||
)
|
||||
|
||||
assert response == {"id": "msg_123", "type": "message", "role": "assistant", "content": []}
|
||||
|
||||
|
||||
class _OpenBreakerRedis:
|
||||
def async_register_script(self, script: str):
|
||||
async def refused(keys, args):
|
||||
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
|
||||
|
||||
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
|
||||
|
||||
return refused
|
||||
|
||||
async def async_increment_pipeline(self, increment_list, **kwargs):
|
||||
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
|
||||
|
||||
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_open_circuit_breaker_falls_back_to_the_pipeline_without_a_warning(caplog):
|
||||
from litellm.types.caching import RedisPipelineIncrementOperation
|
||||
|
||||
handler = _PROXY_MaxParallelRequestsHandler(
|
||||
internal_usage_cache=InternalUsageCache(DualCache(redis_cache=_OpenBreakerRedis())) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
)
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM Proxy"):
|
||||
await handler.async_increment_tokens_with_ttl_preservation(
|
||||
pipeline_operations=[RedisPipelineIncrementOperation(key="quiet_key", increment_value=10.0, ttl=60)]
|
||||
)
|
||||
|
||||
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):
|
||||
handler = _PROXY_MaxParallelRequestsHandler(
|
||||
internal_usage_cache=InternalUsageCache(DualCache(redis_cache=_OpenBreakerRedis())) # pyright: ignore[reportArgumentType] # duck-typed Redis double
|
||||
)
|
||||
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -1,9 +1,11 @@
|
|||
import logging
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.caching.in_memory_cache import InMemoryCache
|
||||
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError
|
||||
from litellm.router_strategy.least_busy import IN_FLIGHT_COUNT_TTL_SECONDS, LeastBusyLoggingHandler
|
||||
|
||||
GROUP: Final = "least-busy-group"
|
||||
|
|
@ -185,3 +187,24 @@ def test_calls_without_a_deployment_are_ignored() -> None:
|
|||
worker.log_pre_api_call(model="m", messages=[], kwargs={})
|
||||
|
||||
assert shared.counts == {}
|
||||
|
||||
|
||||
class OpenBreakerRedis(SharedRedisCounters):
|
||||
def increment_with_floor(self, key: str, value: int, ttl: int) -> int:
|
||||
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
|
||||
|
||||
def batch_get_counts(self, key_list: list[str]) -> tuple[int | None, ...]:
|
||||
raise RedisCircuitBreakerOpenError("Redis circuit breaker is open")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_open_circuit_breaker_falls_back_without_a_warning_per_request(caplog: pytest.LogCaptureFixture) -> None:
|
||||
worker: Final = _worker(OpenBreakerRedis())
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger="LiteLLM Router"):
|
||||
worker.log_pre_api_call(model="m", messages=[], kwargs=_call_kwargs("dep-a"))
|
||||
picked: Final = worker.get_available_deployments(GROUP, HEALTHY)
|
||||
|
||||
assert picked is DEPLOYMENT_B
|
||||
assert [record.getMessage() for record in caplog.records if record.levelno >= logging.WARNING] == []
|
||||
assert sum("circuit breaker is open" in record.getMessage() for record in caplog.records) == 2
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -15359,3 +15361,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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue