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

fix(redis): resolve Rust checks

Browse source
This commit is contained in:
Devin AI 2026-11-01 05:14:39 +00:00
parent 3df56d9366
commit 5c5b7796f4
6 changed files with 51 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
litellm-rust/crates/gateway/src/redis.rs
View file

@ -1,8 +1,8 @@
use std::{sync::Arc, time::Duration};
use litellm_config::{
CacheSettings, Config, RedisSettings, ResolvedRedis, SecretSnapshot,
Value, environment_has_target,
CacheSettings, Config, RedisSettings, ResolvedRedis, SecretSnapshot, Value,
environment_has_target,
};
use litellm_core_utils::settings::Lookup;
use litellm_redis_client::{RedisClient, RedisEvents};

2
litellm-rust/crates/python-bridge/src/lib.rs
View file

@ -118,6 +118,8 @@ mod tests {
"aresponses",
"ResponsesWebSocketConnection",
"NativeDiagnosticProcessor",
"NativeRouterCooldowns",
"NativeRouterUsage",
"NativeClickHouseSpendConfig",
"NativeClickHouseSpendStorage",
"NativeDatabase",

45
litellm-rust/crates/python-bridge/src/redis/events.rs
View file

@ -51,9 +51,10 @@ impl ServiceLoggingEvents {
};
runtime.spawn(async move {
let reported = attach_blocking(context, move |py| match args {
ReportArgs::Call(call_type, duration, failure, caller) => {
record.bind(py).call1((call_type, duration.as_secs_f64(), failure, caller)).map(drop)
}
ReportArgs::Call(call_type, duration, failure, caller) => record
.bind(py)
.call1((call_type, duration.as_secs_f64(), failure, caller))
.map(drop),
ReportArgs::Breaker(state) => record.bind(py).call1((state,)).map(drop),
})
.await
@ -102,12 +103,21 @@ impl RedisEvents for ServiceLoggingEvents {
let caller = CALL.try_with(Clone::clone).ok();
let (context, call_type, caller) = match caller {
Some(call) => (call.context, call.call_type, call.caller),
None => (self.context.clone(), operation_call_type(event.operation), None),
None => (
self.context.clone(),
operation_call_type(event.operation),
None,
),
};
self.report(
context,
self.record_call.clone(),
ReportArgs::Call(call_type, event.duration, failure_label(event.outcome), caller),
ReportArgs::Call(
call_type,
event.duration,
failure_label(event.outcome),
caller,
),
);
}
@ -145,8 +155,13 @@ def record_redis_breaker(state):
Python::initialize();
Python::attach(|py| {
let name = std::ffi::CString::new(format!("redis_events_recorder_{name}")).unwrap();
let module = PyModule::from_code(py, &std::ffi::CString::new(RECORDER).unwrap(), &name, &name).unwrap();
(ServiceLoggingEvents::from_module(py, &module).unwrap(), module.unbind())
let module =
PyModule::from_code(py, &std::ffi::CString::new(RECORDER).unwrap(), &name, &name)
.unwrap();
(
ServiceLoggingEvents::from_module(py, &module).unwrap(),
module.unbind(),
)
})
}
@ -205,9 +220,16 @@ def record_redis_breaker(state):
) {
let (events, namespace) = recorder(expected);
events.breaker(&litellm_circuit_breaker::Transition { from: litellm_circuit_breaker::State::HalfOpen, to, generation: 1 });
events.breaker(&litellm_circuit_breaker::Transition {
from: litellm_circuit_breaker::State::HalfOpen,
to,
generation: 1,
});
assert_eq!(seen(&namespace, 1).await, format!("[('breaker', '{expected}')]"));
assert_eq!(
seen(&namespace, 1).await,
format!("[('breaker', '{expected}')]")
);
}
#[rstest]
@ -221,6 +243,9 @@ def record_redis_breaker(state):
outcome: Outcome::Succeeded,
});
assert_eq!(seen(&namespace, 1).await, "[('pipeline', 0.005, None, None, None)]");
assert_eq!(
seen(&namespace, 1).await,
"[('pipeline', 0.005, None, None, None)]"
);
}
}

4
litellm-rust/crates/redis-connect/src/connect.rs
View file

@ -1,7 +1,9 @@
use std::sync::Arc;
use litellm_circuit_breaker::{Breaker, Config as BreakerConfig};
use litellm_config::{Expected, RedisAuth, RedisEndpoint, RedisError, RedisTls, ResolvedRedis, parse_seconds};
use litellm_config::{
Expected, RedisAuth, RedisEndpoint, RedisError, RedisTls, ResolvedRedis, parse_seconds,
};
use litellm_core_utils::settings::Lookup;
use litellm_redis_client::{
ClusterEndpoint, ClusterNode, Credentials, CredentialsProvider, DEFAULT_TIMEOUT, Endpoint,

10
litellm/rust_bridge/_native.pyi
View file

@ -26,7 +26,7 @@ class NativeClickHouseSpendStorage:
def ensure_schema(self) -> Future[None]: ...
def insert_rows(self, rows: Sequence[Mapping[str, object]]) -> Future[None]: ...
class NativeCooldown(TypedDict):
class _NativeCooldown(TypedDict):
exception_received: str
status_code: str
timestamp: float
@ -39,12 +39,12 @@ class NativeRouterCooldowns:
) -> NativeRouterCooldowns: ...
def put(self, model_id: str, cooldown: Mapping[str, object], ttl_seconds: float) -> None: ...
def increment_allowed_fails(self, model_id: str, suffix: str | None, ttl_seconds: float) -> int: ...
def active(self, model_ids: Sequence[str]) -> list[tuple[str, NativeCooldown]]: ...
def aactive(self, model_ids: Sequence[str]) -> Future[list[tuple[str, NativeCooldown]]]: ...
def active(self, model_ids: Sequence[str]) -> list[tuple[str, _NativeCooldown]]: ...
def aactive(self, model_ids: Sequence[str]) -> Future[list[tuple[str, _NativeCooldown]]]: ...
def minimum(self, model_ids: Sequence[str]) -> float | None: ...
def evict_local(self, model_id: str) -> None: ...
def flush_local_counters(self) -> None: ...
def snapshot(self) -> dict[str, NativeCooldown]: ...
def snapshot(self) -> dict[str, _NativeCooldown]: ...
@final
class NativeRouterUsage:
@ -253,6 +253,8 @@ __all__ = [
"NativeClickHouseSpendStorage",
"NativeDatabase",
"NativeDiagnosticProcessor",
"NativeRouterCooldowns",
"NativeRouterUsage",
"ProcessReservedForForking",
"ResponsesWebSocketConnection",
"RustBridgeDeclined",

6
litellm/rust_bridge/host/redis_events.py
View file

@ -3,13 +3,13 @@
from __future__ import annotations
from functools import lru_cache
from typing import Final, Literal
from typing import Final, Literal, TypeAlias
from litellm._service_logger import ServiceLogging
from litellm.types.services import ServiceTypes
Failure = Literal["timeout", "transport", "server", "decode", "breaker_open"]
BreakerState = Literal["closed", "open", "half_open"]
Failure: TypeAlias = Literal["timeout", "transport", "server", "decode", "breaker_open"]
BreakerState: TypeAlias = Literal["closed", "open", "half_open"]
_HEALTH_FAILURE_CLASSES: Final = {"timeout": "timeout", "transport": "connectivity"}