mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
fix(proxy): seed a cold spend counter by adding when Redis refuses scripts
The cold seed raises the counter with a Lua script. A Redis whose ACL denies scripting, or that renamed EVAL away, refused it, the reservation skipped and customer and tag budgets stopped being enforced. The seed now adds the cached spend there, as main does async_set_max also reports to the Redis service metrics like async_increment, so a failed seed still counts in litellm_redis_failed_requests
This commit is contained in:
parent
b8e12c9144
commit
3227779577
5 changed files with 155 additions and 5 deletions
|
|
@ -453,6 +453,18 @@ def is_redis_timeout_failure(exc: BaseException) -> bool:
|
|||
return any(isinstance(link, timeout_types) for link in _explicit_causes(exc))
|
||||
|
||||
|
||||
def is_redis_scripting_unavailable(exc: BaseException) -> bool:
|
||||
"""True when Redis refused to run a Lua script for this client: an ACL without
|
||||
``@scripting``, or ``EVAL`` renamed away."""
|
||||
try:
|
||||
from redis.exceptions import NoPermissionError, ResponseError
|
||||
except ImportError:
|
||||
return False
|
||||
if isinstance(exc, NoPermissionError):
|
||||
return "permissions to run" in str(exc)
|
||||
return isinstance(exc, ResponseError) and "unknown command" in str(exc).lower()
|
||||
|
||||
|
||||
def _record_swallowed_redis_failure(breaker: RedisCircuitBreaker, exc: BaseException) -> None:
|
||||
"""Record a Redis failure that the calling method is about to swallow.
|
||||
|
||||
|
|
@ -1555,9 +1567,36 @@ class RedisCache(BaseCache):
|
|||
"return ARGV[1] end "
|
||||
"return cur"
|
||||
)
|
||||
result = cast(
|
||||
"str | bytes | int | float | None",
|
||||
await _redis_client.eval(lua, 1, key, str(value), str(int(_used_ttl or 0))),
|
||||
start_time: Final = time.time()
|
||||
try:
|
||||
result = cast(
|
||||
"str | bytes | int | float | None",
|
||||
await _redis_client.eval(lua, 1, key, str(value), str(int(_used_ttl or 0))),
|
||||
)
|
||||
except Exception as e:
|
||||
failed_at: Final = time.time()
|
||||
asyncio.create_task(
|
||||
self.service_logger_obj.async_service_failure_hook( # pyright: ignore[reportUnknownMemberType] # bare dict in sig
|
||||
service=ServiceTypes.REDIS,
|
||||
duration=failed_at - start_time,
|
||||
error=e,
|
||||
call_type="async_set_max",
|
||||
caller=_get_call_stack_info(),
|
||||
start_time=start_time,
|
||||
end_time=failed_at,
|
||||
)
|
||||
)
|
||||
raise
|
||||
end_time: Final = time.time()
|
||||
asyncio.create_task(
|
||||
self.service_logger_obj.async_service_success_hook( # pyright: ignore[reportUnknownMemberType] # bare dict in sig
|
||||
service=ServiceTypes.REDIS,
|
||||
duration=end_time - start_time,
|
||||
call_type="async_set_max",
|
||||
caller=_get_call_stack_info(),
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
)
|
||||
)
|
||||
if result is None:
|
||||
return None
|
||||
|
|
|
|||
|
|
@ -215,6 +215,7 @@ _SERVICE_VERB_BY_CALL_TYPE: Final[dict[str, str]] = {
|
|||
"async_increment": "incr",
|
||||
"async_increment_pipeline": "incr",
|
||||
"async_increment_if_present": "incr",
|
||||
"async_set_max": "set",
|
||||
"delete_cache": "delete",
|
||||
"async_delete_cache": "delete",
|
||||
"async_rpush": "rpush",
|
||||
|
|
|
|||
|
|
@ -313,7 +313,11 @@ from litellm.caching.redis_batch import (
|
|||
active_post_call_redis_batch,
|
||||
drain_post_call_redis_batches,
|
||||
)
|
||||
from litellm.caching.redis_cache import RedisCircuitBreakerOpenError, is_redis_timeout_failure
|
||||
from litellm.caching.redis_cache import (
|
||||
RedisCircuitBreakerOpenError,
|
||||
is_redis_scripting_unavailable,
|
||||
is_redis_timeout_failure,
|
||||
)
|
||||
from litellm.caching.redis_cluster_cache import RedisClusterCache
|
||||
from litellm.constants import (
|
||||
_REALTIME_BODY_CACHE_SIZE,
|
||||
|
|
@ -4169,7 +4173,10 @@ async def _lift_spend_counter_cache(counter_key: str, floor: float) -> None:
|
|||
if spend_counter_cache.redis_cache is not None:
|
||||
try:
|
||||
lifted: Final = await spend_counter_cache.redis_cache.async_set_max(key=counter_key, value=floor)
|
||||
except Exception:
|
||||
except Exception as e:
|
||||
if is_redis_scripting_unavailable(e):
|
||||
await increment_spend_counter_cache(counter_key=counter_key, increment=floor)
|
||||
return
|
||||
await invalidate_spend_counter(counter_key=counter_key)
|
||||
raise
|
||||
if lifted is not None:
|
||||
|
|
|
|||
|
|
@ -1697,3 +1697,43 @@ async def test_async_set_cache_pipeline_does_not_format_cached_values_for_loggin
|
|||
|
||||
assert _FormatCountingStr.format_calls == 0
|
||||
assert pipe.sets == [("k1", '"embedding"', timedelta(seconds=60))]
|
||||
|
||||
|
||||
def _cache_reporting_to(service_logger: ServiceLogging) -> RedisCache:
|
||||
with patch( # test-quality-ok: RedisCache.__init__ builds its client eagerly, with no injection point
|
||||
"litellm._redis.get_redis_client", return_value=MagicMock()
|
||||
):
|
||||
return RedisCache(host="127.0.0.1", port=6379, service_logger_obj=service_logger)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_set_max_reports_a_failed_script_to_the_redis_failure_metric(redis_no_ping: None):
|
||||
"""The cold spend-counter seed runs through this script, so its Redis errors have to keep reaching
|
||||
litellm_redis_failed_requests_total the way they did when the seed was an INCRBYFLOAT."""
|
||||
service_logger: Final = ServiceLogging(mock_testing=True)
|
||||
cache: Final = _cache_reporting_to(service_logger)
|
||||
client: Final = AsyncMock()
|
||||
client.eval.side_effect = OSError("redis unavailable")
|
||||
tasks_before: Final = asyncio.all_tasks()
|
||||
|
||||
with patch.object(cache, "init_async_client", return_value=client):
|
||||
with pytest.raises(OSError, match="redis unavailable"):
|
||||
await cache.async_set_max(key="spend:end_user:customer-1", value=6.0)
|
||||
await asyncio.wait_for(asyncio.gather(*(asyncio.all_tasks() - tasks_before)), timeout=5)
|
||||
|
||||
assert (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook) == (0, 1)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_set_max_reports_a_completed_script_to_the_redis_latency_metric(redis_no_ping: None):
|
||||
service_logger: Final = ServiceLogging(mock_testing=True)
|
||||
cache: Final = _cache_reporting_to(service_logger)
|
||||
client: Final = AsyncMock()
|
||||
client.eval.return_value = b"6.0"
|
||||
tasks_before: Final = asyncio.all_tasks()
|
||||
|
||||
with patch.object(cache, "init_async_client", return_value=client):
|
||||
assert await cache.async_set_max(key="spend:end_user:customer-1", value=6.0) == 6.0
|
||||
await asyncio.wait_for(asyncio.gather(*(asyncio.all_tasks() - tasks_before)), timeout=5)
|
||||
|
||||
assert (service_logger.mock_testing_async_success_hook, service_logger.mock_testing_async_failure_hook) == (1, 0)
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ from opentelemetry.sdk.trace import TracerProvider
|
|||
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
|
||||
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
|
||||
from opentelemetry.trace import StatusCode
|
||||
from redis.exceptions import NoPermissionError, ResponseError
|
||||
|
||||
import litellm
|
||||
import litellm.proxy.proxy_server as ps
|
||||
|
|
@ -1415,6 +1416,68 @@ async def test_a_redis_failure_while_lifting_a_cold_counter_drops_the_local_copy
|
|||
assert counters.in_memory_cache.get_cache(key=counter_key) is None
|
||||
|
||||
|
||||
class _FakeRedisRefusingScripts(FakeAsyncRedis):
|
||||
def __init__(self, server: FakeServer, refusal: ResponseError) -> None:
|
||||
super().__init__(server=server)
|
||||
self._refusal: Final = refusal
|
||||
|
||||
async def eval(self, script: str, numkeys: int, *keys_and_args: str | bytes | float) -> bytes | None:
|
||||
raise self._refusal
|
||||
|
||||
|
||||
def _cold_start_on(monkeypatch: pytest.MonkeyPatch, redis: FakeAsyncRedis, cached_spend: float) -> None:
|
||||
objects: Final = MagicMock()
|
||||
objects.async_get_cache = AsyncMock(return_value={"spend": cached_spend})
|
||||
monkeypatch.setattr(
|
||||
ps, "spend_counter_cache", DualCache(in_memory_cache=InMemoryCache(), redis_cache=_RedisCacheOnClient(redis))
|
||||
)
|
||||
monkeypatch.setattr(ps, "user_api_key_cache", objects)
|
||||
monkeypatch.setattr(ps, "prisma_client", None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"refusal",
|
||||
(
|
||||
NoPermissionError("User noscript has no permissions to run the 'eval' command"),
|
||||
ResponseError("unknown command 'eval', with args beginning with: "),
|
||||
),
|
||||
ids=("acl-without-scripting", "eval-renamed-away"),
|
||||
)
|
||||
async def test_a_cold_counter_is_still_seeded_on_a_redis_that_refuses_scripts(
|
||||
monkeypatch: pytest.MonkeyPatch, refusal: ResponseError
|
||||
) -> None:
|
||||
redis: Final = _FakeRedisRefusingScripts(FakeServer(), refusal)
|
||||
_cold_start_on(monkeypatch, redis, cached_spend=6.0)
|
||||
|
||||
await ps.ensure_spend_counter_initialized(counter_key="spend:end_user:customer-1", source_cache_key="customer-1")
|
||||
|
||||
assert await redis.get("spend:end_user:customer-1") == b"6"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"refusal",
|
||||
(
|
||||
NoPermissionError("No permissions to access a key"),
|
||||
ResponseError("WRONGTYPE Operation against a key holding the wrong kind of value"),
|
||||
),
|
||||
ids=("key-outside-the-acl", "wrong-type"),
|
||||
)
|
||||
async def test_a_lift_refused_for_another_reason_is_raised_instead_of_seeded_by_adding(
|
||||
monkeypatch: pytest.MonkeyPatch, refusal: ResponseError
|
||||
) -> None:
|
||||
redis: Final = _FakeRedisRefusingScripts(FakeServer(), refusal)
|
||||
_cold_start_on(monkeypatch, redis, cached_spend=6.0)
|
||||
|
||||
with pytest.raises(type(refusal)):
|
||||
await ps.ensure_spend_counter_initialized(
|
||||
counter_key="spend:end_user:customer-1", source_cache_key="customer-1"
|
||||
)
|
||||
|
||||
assert await redis.get("spend:end_user:customer-1") is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _get_source_cache_base_spend
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue