scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

feat(gateway): connect to Redis from config.yaml (#48284)

Browse source 
* feat(gateway): connect to Redis from config.yaml

* feat(config): own Redis settings resolution

* fix(config): preserve Redis input precedence and presence

* fix(config): box Redis cache settings

* fix(gateway): preserve borrowed redis namespace

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yujonglee 2026-10-26 18:41:47 -07:00 • committed by GitHub
parent 0c0908c36e
commit bdfe7ae195
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
23 changed files with 1915 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm-rust/Cargo.lock generated
View file

@ -3676,6 +3676,8 @@ dependencies = [
"strum",
"tempfile",
"thiserror 2.0.19",
"tracing",
"tracing-subscriber",
]
[[package]]
@ -3764,6 +3766,7 @@ dependencies = [
"futures-util",
"http-body-util",
"litellm-auth-types",
"litellm-circuit-breaker",
"litellm-config",
"litellm-gateway-auth",
"litellm-gateway-inference",
@ -3772,6 +3775,8 @@ dependencies = [
"litellm-http",
"litellm-inference",
"litellm-llms",
"litellm-redis-client",
"litellm-redis-testing",
"litellm-secrets",
"litellm-tracing",
"rstest",

4
litellm-rust/crates/config/Cargo.toml
View file

@ -9,11 +9,13 @@ repository.workspace = true
litellm-auth-types.workspace = true
litellm-router-types.workspace = true
serde.workspace = true
serde_json.workspace = true
serde_yaml_ng = "0.10.0"
strum.workspace = true
thiserror.workspace = true
tracing.workspace = true
[dev-dependencies]
rstest.workspace = true
serde_json.workspace = true
tempfile.workspace = true
tracing-subscriber = { version = "0.3", default-features = false, features = ["registry", "std"] }

7
litellm-rust/crates/config/src/error.rs
View file

@ -4,4 +4,11 @@ pub enum Error {
Read(#[from] std::io::Error),
#[error("invalid YAML config")]
Parse(#[from] serde_yaml_ng::Error),
#[error("Redis host does not yet support configured setting {0}")]
RedisSetting(String),
#[error("invalid Redis setting {setting}: {reason}")]
RedisInvalid {
setting: &'static str,
reason: &'static str,
},
}

5
litellm-rust/crates/config/src/lib.rs
View file

@ -2,6 +2,7 @@ mod error;
mod includes;
mod mcp;
mod model;
mod redis;
mod settings;
mod value;
@ -13,6 +14,10 @@ pub use error::Error;
pub use litellm_router_types::{LitellmParams, Spelled};
pub use mcp::{McpAuth, McpServer, McpTransport};
pub use model::Model;
pub use redis::{
CacheSettings, Lookup, RedisCluster, RedisEndpoint, RedisHost, RedisNode, RedisSettings,
ResolvedRedis, environment_has_target,
};
pub use settings::{
ClickHouseStoreSettings, GeneralSettings, LiteLlmSettings, RouterSettings, TracingSettings,
TracingStoreSettings,

667
litellm-rust/crates/config/src/redis.rs Normal file
View file

@ -0,0 +1,667 @@
use std::{collections::BTreeMap, fmt, time::Duration};
use litellm_auth_types::SecretValue;
use litellm_router_types::Spelled;
use serde::{Deserialize, Deserializer, de::Error as _};
use crate::{AdditionalFields, Error, Object, Value};
pub type Lookup<'a> = &'a dyn Fn(&str) -> Option<String>;
const SUPPORTED: &[&str] = &[
"url",
"host",
"port",
"username",
"password",
"ssl",
"db",
"socket_timeout",
"startup_nodes",
"namespace",
];
const IGNORED: &[&str] = &[
"max_connections",
"single_connection_client",
"health_check_interval",
"socket_keepalive",
"socket_keepalive_options",
"decode_responses",
"encoding",
"encoding_errors",
"charset",
"errors",
"client_name",
"lib_name",
"lib_version",
"sentinel_password",
"gcp_ssl_ca_certs",
"azure_client_id",
"azure_tenant_id",
"azure_client_secret",
"aws_iam_user_name",
"aws_iam_cache_name",
"aws_iam_region",
"aws_iam_serverless",
];
const WARN_IGNORED: &[&str] = &[
"max_connections",
"health_check_interval",
"socket_keepalive",
"socket_keepalive_options",
];
const REDIS_PY_KWARGS: &[&str] = &[
"aws_iam_auth",
"aws_iam_cache_name",
"aws_iam_region",
"aws_iam_serverless",
"aws_iam_user_name",
"azure_client_id",
"azure_client_secret",
"azure_redis_ad_token",
"azure_tenant_id",
"cache",
"cache_config",
"charset",
"client_name",
"db",
"decode_responses",
"encoding",
"encoding_errors",
"errors",
"event_dispatcher",
"gcp_service_account",
"gcp_ssl_ca_certs",
"health_check_interval",
"host",
"lib_name",
"lib_version",
"max_connections",
"password",
"port",
"protocol",
"redis_connect_func",
"retry_on_error",
"retry_on_timeout",
"single_connection_client",
"socket_connect_timeout",
"socket_keepalive",
"socket_keepalive_options",
"socket_timeout",
"ssl",
"ssl_ca_certs",
"ssl_ca_data",
"ssl_ca_path",
"ssl_cert_reqs",
"ssl_certfile",
"ssl_check_hostname",
"ssl_ciphers",
"ssl_keyfile",
"ssl_min_version",
"ssl_ocsp_context",
"ssl_ocsp_expected_cert",
"ssl_password",
"ssl_validate_ocsp",
"ssl_validate_ocsp_stapled",
"unix_socket_path",
"url",
"username",
];
const UNSUPPORTED_EXTRA: &[&str] = &[
"credential_provider",
"sentinel_nodes",
"retry",
"connection_pool",
];
const CACHE_ONLY: &[&str] = &[
"type",
"mode",
"ttl",
"default_in_memory_ttl",
"default_in_redis_ttl",
"supported_call_types",
"similarity_threshold",
"max_messages",
"_backend",
];
#[derive(Clone, Default)]
pub struct RedisSettings {
pub url: Option<SecretValue>,
pub host: Option<Spelled<String>>,
pub port: Option<Spelled<u16>>,
pub username: Option<SecretValue>,
pub password: Option<SecretValue>,
pub ssl: Option<Spelled<bool>>,
pub db: Option<Spelled<i64>>,
pub socket_timeout: Option<Spelled<f64>>,
pub startup_nodes: Option<Value>,
pub namespace: Option<Spelled<String>>,
pub additional_fields: AdditionalFields,
unparsed: DeferredInputs,
}
impl<'de> Deserialize<'de> for RedisSettings {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let fields = BTreeMap::<String, Value>::deserialize(deserializer)?;
let scalar = |key: &str| fields.get(key).and_then(as_text);
Ok(Self {
url: scalar("url").map(SecretValue::new),
host: scalar("host").map(Spelled::Value),
port: spelled(scalar("port")),
username: scalar("username").map(SecretValue::new),
password: scalar("password").map(SecretValue::new),
ssl: spelled(scalar("ssl")),
db: spelled(scalar("db")),
socket_timeout: spelled(scalar("socket_timeout")),
startup_nodes: fields
.get("startup_nodes")
.filter(|value| !value.is_null())
.cloned(),
namespace: scalar("namespace").map(Spelled::Value),
additional_fields: cloned(&fields, |key, _| !SUPPORTED.contains(&key)),
unparsed: deferred_inputs(&fields),
})
}
}
impl fmt::Debug for RedisSettings {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("RedisSettings")
.field("url", &self.url)
.field("host", &self.host)
.field("port", &self.port)
.field("username", &self.username)
.field("password", &self.password)
.field("ssl", &self.ssl)
.field("db", &self.db)
.field("socket_timeout", &self.socket_timeout)
.field(
"startup_nodes",
&self.startup_nodes.as_ref().map(|_| "<configured>"),
)
.field("namespace", &self.namespace)
.field("additional_fields", &self.additional_fields.keys())
.field("unparsed", &self.unparsed.keys())
.finish()
}
}
#[derive(Clone, Debug)]
pub enum CacheSettings {
Redis(Box<RedisSettings>),
Other(Object),
}
impl<'de> Deserialize<'de> for CacheSettings {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let fields = BTreeMap::<String, Value>::deserialize(deserializer)?;
let is_redis = matches!(fields.get("type"), Some(Value::String(kind)) if kind == "redis");
if !is_redis {
return Ok(Self::Other(Object::from_fields(fields)));
}
let mapping = fields
.into_iter()
.map(|(key, value)| (Value::String(key), value))
.collect();
serde_yaml_ng::from_value(Value::Mapping(mapping))
.map(|settings| Self::Redis(Box::new(settings)))
.map_err(D::Error::custom)
}
}
#[derive(Clone, PartialEq)]
pub struct ResolvedRedis {
pub endpoint: RedisEndpoint,
pub namespace: Option<String>,
pub timeout: Option<Duration>,
}
impl fmt::Debug for ResolvedRedis {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ResolvedRedis")
.field("endpoint", &self.endpoint)
.field("namespace", &self.namespace)
.field("timeout", &self.timeout)
.finish()
}
}
#[derive(Clone, PartialEq)]
pub enum RedisEndpoint {
Url(SecretValue),
Host(RedisHost),
Cluster(RedisCluster),
}
impl fmt::Debug for RedisEndpoint {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Url(_) => formatter.debug_tuple("Url").field(&"<redacted>").finish(),
Self::Host(host) => host.fmt(formatter),
Self::Cluster(cluster) => cluster.fmt(formatter),
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct RedisHost {
pub host: String,
pub port: u16,
pub username: Option<SecretValue>,
pub password: Option<SecretValue>,
pub tls: bool,
pub db: i64,
}
#[derive(Clone, Debug, PartialEq)]
pub struct RedisCluster {
pub nodes: Vec<RedisNode>,
pub username: Option<SecretValue>,
pub password: Option<SecretValue>,
pub tls: bool,
}
#[derive(Clone, Debug, PartialEq)]
pub struct RedisNode {
pub host: String,
pub port: u16,
}
#[derive(Clone)]
struct Entry {
value: Value,
source: String,
}
type Kwargs = BTreeMap<String, Entry>;
impl RedisSettings {
pub fn resolve(&self, lookup: Lookup<'_>) -> Result<ResolvedRedis, Error> {
self.resolve_with(lookup, false)
}
pub fn resolve_cache(&self, lookup: Lookup<'_>) -> Result<ResolvedRedis, Error> {
self.resolve_with(lookup, true)
}
fn resolve_with(&self, lookup: Lookup<'_>, cache: bool) -> Result<ResolvedRedis, Error> {
let block = self.kwargs(cache);
let block: Kwargs = block
.into_iter()
.filter_map(|(key, entry)| {
let value = environment_reference(&entry.value, lookup)?;
Some((key, Entry { value, ..entry }))
})
.collect();
let environment = environment_kwargs(lookup);
let names_a_target = ["host", "startup_nodes", "sentinel_nodes"]
.iter()
.any(|key| block.contains_key(*key));
let drop_environment_url = names_a_target && !block.contains_key("url");
let merged: Kwargs = environment
.into_iter()
.filter(|(key, _)| !(key == "url" && drop_environment_url))
.chain(block)
.collect();
reject_unsupported(&merged, lookup)?;
warn_ignored(&merged);
let startup_nodes = match merged.get("startup_nodes") {
Some(entry) => Some(cluster_nodes(&entry.value)?),
None => lookup("REDIS_CLUSTER_NODES")
.map(|nodes| cluster_nodes(&Value::String(nodes)))
.transpose()?,
};
let username = secret_value(&merged, "username");
let password = secret_value(&merged, "password");
let tls = bool_value(&merged, "ssl");
let endpoint = match (startup_nodes, text_value(&merged, "url")) {
(Some(nodes), _) if !nodes.is_empty() => RedisEndpoint::Cluster(RedisCluster {
nodes,
username,
password,
tls,
}),
(_, Some(url)) => RedisEndpoint::Url(SecretValue::new(url)),
_ => RedisEndpoint::Host(RedisHost {
host: text_value(&merged, "host").ok_or(Error::RedisInvalid {
setting: "host",
reason: "either host or url must be specified for redis",
})?,
port: integer(&merged, "port")?.unwrap_or(6379),
username,
password,
tls,
db: integer(&merged, "db")?.unwrap_or(0),
}),
};
let timeout = merged
.get("socket_timeout")
.and_then(|entry| as_text(&entry.value))
.map(|seconds| {
seconds
.parse::<f64>()
.ok()
.and_then(|seconds| Duration::try_from_secs_f64(seconds).ok())
.ok_or(Error::RedisInvalid {
setting: "socket_timeout",
reason: "must be a non-negative number of seconds",
})
})
.transpose()?;
Ok(ResolvedRedis {
endpoint,
namespace: text_value(&merged, "namespace"),
timeout,
})
}
fn kwargs(&self, cache: bool) -> Kwargs {
let known = [
(
"url",
self.url.as_ref().map(|value| value.expose().to_owned()),
),
("host", spelled_text(self.host.as_ref())),
("port", spelled_text(self.port.as_ref())),
(
"username",
self.username
.as_ref()
.map(|value| value.expose().to_owned()),
),
(
"password",
self.password
.as_ref()
.map(|value| value.expose().to_owned()),
),
("ssl", spelled_text(self.ssl.as_ref())),
("db", spelled_text(self.db.as_ref())),
("socket_timeout", spelled_text(self.socket_timeout.as_ref())),
("namespace", spelled_text(self.namespace.as_ref())),
];
let known = known.into_iter().filter_map(|(key, value)| {
value.map(|value| {
(
key.to_owned(),
Entry {
value: Value::String(value),
source: key.to_owned(),
},
)
})
});
let startup_nodes = self.startup_nodes.clone().map(|value| {
(
"startup_nodes".to_owned(),
Entry {
value,
source: "startup_nodes".to_owned(),
},
)
});
let extras = self.additional_fields.iter().filter_map(|(key, value)| {
if cache && CACHE_ONLY.contains(&key.as_str()) {
return None;
}
let key = match (cache, key.as_str()) {
(true, "redis_startup_nodes") => "startup_nodes".to_owned(),
_ => key.clone(),
};
Some((
key.clone(),
Entry {
value: value.clone(),
source: key,
},
))
});
let unparsed = self.unparsed.iter().map(|(key, value)| {
(
key.clone(),
Entry {
value: value.0.clone(),
source: key.clone(),
},
)
});
extras
.chain(unparsed)
.chain(known)
.chain(startup_nodes)
.collect()
}
}
pub fn environment_has_target(lookup: Lookup<'_>) -> bool {
[
"REDIS_HOST",
"REDIS_URL",
"REDIS_CLUSTER_NODES",
"REDIS_SENTINEL_NODES",
]
.iter()
.any(|name| lookup(name).is_some())
}
fn reject_unsupported(merged: &Kwargs, lookup: Lookup<'_>) -> Result<(), Error> {
let active_auth = [
merged
.get("gcp_service_account")
.map(|entry| entry.source.as_str())
.or_else(|| lookup("REDIS_GCP_SERVICE_ACCOUNT").map(|_| "REDIS_GCP_SERVICE_ACCOUNT")),
text_value(merged, "azure_redis_ad_token")
.or_else(|| lookup("REDIS_AZURE_AD_TOKEN"))
.filter(|value| value.eq_ignore_ascii_case("true"))
.map(|_| {
merged
.get("azure_redis_ad_token")
.map_or("REDIS_AZURE_AD_TOKEN", |entry| entry.source.as_str())
}),
bool_value(merged, "aws_iam_auth").then(|| merged["aws_iam_auth"].source.as_str()),
merged
.get("sentinel_nodes")
.map(|entry| entry.source.as_str())
.or_else(|| lookup("REDIS_SENTINEL_NODES").map(|_| "REDIS_SENTINEL_NODES")),
];
if let Some(setting) = active_auth.into_iter().flatten().next() {
return Err(Error::RedisSetting(setting.to_owned()));
}
match merged
.iter()
.find(|(key, _)| is_unsupported(key))
.map(|(_, entry)| entry.source.clone())
{
Some(setting) => Err(Error::RedisSetting(setting)),
None => Ok(()),
}
}
fn warn_ignored(merged: &Kwargs) {
for setting in WARN_IGNORED {
if merged.contains_key(*setting) {
tracing::warn!(setting, "Redis setting is accepted but ignored");
}
}
}
fn is_unsupported(key: &str) -> bool {
let known = REDIS_PY_KWARGS.contains(&key) || UNSUPPORTED_EXTRA.contains(&key);
let handled = SUPPORTED.contains(&key)
|| IGNORED.contains(&key)
|| matches!(
key,
"gcp_service_account" | "azure_redis_ad_token" | "aws_iam_auth"
);
known && !handled
}
fn environment_kwargs(lookup: Lookup<'_>) -> Kwargs {
REDIS_PY_KWARGS
.iter()
.filter_map(|key| {
let source = format!("REDIS_{}", key.to_ascii_uppercase());
let value = lookup(&source)?;
Some((
(*key).to_owned(),
Entry {
value: Value::String(value),
source,
},
))
})
.collect()
}
fn environment_reference(value: &Value, lookup: Lookup<'_>) -> Option<Value> {
match value {
Value::String(text) => match text.strip_prefix("os.environ/") {
Some(name) => lookup(name).map(Value::String),
None => Some(value.clone()),
},
Value::Null => None,
_ => Some(value.clone()),
}
}
fn as_text(value: &Value) -> Option<String> {
match value {
Value::String(text) => Some(text.clone()),
Value::Number(number) => Some(number.to_string()),
Value::Bool(flag) => Some(flag.to_string()),
_ => None,
}
}
fn text_value(kwargs: &Kwargs, key: &str) -> Option<String> {
kwargs.get(key).and_then(|entry| as_text(&entry.value))
}
fn secret_value(kwargs: &Kwargs, key: &str) -> Option<SecretValue> {
text_value(kwargs, key).map(SecretValue::new)
}
fn bool_value(kwargs: &Kwargs, key: &str) -> bool {
match kwargs.get(key).map(|entry| &entry.value) {
Some(Value::Bool(flag)) => *flag,
Some(value) => as_text(value)
.is_some_and(|text| matches!(text.to_ascii_lowercase().as_str(), "true" | "1" | "yes")),
None => false,
}
}
fn integer<T: std::str::FromStr>(kwargs: &Kwargs, key: &'static str) -> Result<Option<T>, Error> {
kwargs
.get(key)
.and_then(|entry| as_text(&entry.value))
.map(|text| {
text.parse().map_err(|_| Error::RedisInvalid {
setting: key,
reason: "must be an integer",
})
})
.transpose()
}
const INVALID_NODES: Error = Error::RedisInvalid {
setting: "startup_nodes",
reason: "must be a list of {host, port} objects",
};
fn cluster_nodes(value: &Value) -> Result<Vec<RedisNode>, Error> {
match value {
Value::String(json) => {
let nodes: Vec<serde_json::Map<String, serde_json::Value>> =
serde_json::from_str(json).map_err(|_| INVALID_NODES)?;
nodes
.iter()
.map(|node| {
cluster_node(
node.get("host").and_then(json_text),
node.get("port").and_then(json_text),
)
})
.collect()
}
Value::Sequence(nodes) => nodes
.iter()
.map(|node| {
cluster_node(
node.get("host").and_then(as_text),
node.get("port").and_then(as_text),
)
})
.collect(),
_ => Err(INVALID_NODES),
}
}
fn cluster_node(host: Option<String>, port: Option<String>) -> Result<RedisNode, Error> {
match (host, port.and_then(|port| port.parse().ok())) {
(Some(host), Some(port)) => Ok(RedisNode { host, port }),
_ => Err(INVALID_NODES),
}
}
fn json_text(value: &serde_json::Value) -> Option<String> {
match value {
serde_json::Value::String(text) => Some(text.clone()),
serde_json::Value::Number(number) => Some(number.to_string()),
_ => None,
}
}
fn spelled_text<T: ToString>(value: Option<&Spelled<T>>) -> Option<String> {
value.map(|value| match value {
Spelled::Value(value) => value.to_string(),
Spelled::Text(value) => value.clone(),
})
}
fn spelled<T: std::str::FromStr>(text: Option<String>) -> Option<Spelled<T>> {
text.map(|text| match text.parse() {
Ok(value) => Spelled::Value(value),
Err(_) => Spelled::Text(text),
})
}
fn cloned(
fields: &BTreeMap<String, Value>,
keep: impl Fn(&str, &Value) -> bool,
) -> AdditionalFields {
fields
.iter()
.filter(|(key, value)| keep(key, value))
.map(|(key, value)| (key.clone(), value.clone()))
.collect()
}
#[derive(Clone)]
struct DeferredInput(Value);
type DeferredInputs = BTreeMap<String, DeferredInput>;
fn deferred_inputs(fields: &BTreeMap<String, Value>) -> DeferredInputs {
fields
.iter()
.filter(|(key, value)| is_unparsed(key, value))
.map(|(key, value)| (key.clone(), DeferredInput(value.clone())))
.collect()
}
fn is_unparsed(key: &str, value: &Value) -> bool {
SUPPORTED.contains(&key)
&& match key {
"startup_nodes" => value.is_null(),
_ => as_text(value).is_none(),
}
}

8
litellm-rust/crates/config/src/settings.rs
View file

@ -3,7 +3,9 @@ use std::fmt;
use litellm_auth_types::SecretValue;
use serde::Deserialize;
use crate::{AdditionalFields, Object, Spelled, Value, value::one_or_many};
use crate::{
AdditionalFields, CacheSettings, Object, RedisSettings, Spelled, Value, value::one_or_many,
};
#[derive(Clone, Debug, Deserialize)]
#[serde(rename_all = "lowercase")]
@ -76,7 +78,7 @@ pub struct GeneralSettings {
pub enable_public_model_hub: bool,
pub dangerously_permit_weak_or_unset_master_key: Option<bool>,
pub plugins: Option<Box<[Object]>>,
pub coordination_redis: Option<Object>,
pub coordination_redis: Option<RedisSettings>,
pub mcp_allowed_hosts: Option<Box<[String]>>,
pub mcp_allowed_origins: Box<[String]>,
#[serde(flatten)]
@ -220,7 +222,7 @@ pub struct LiteLlmSettings {
pub request_timeout: Option<Spelled<f64>>,
pub num_retries: Option<u64>,
pub cache: Option<bool>,
pub cache_params: Option<Object>,
pub cache_params: Option<CacheSettings>,
#[serde(deserialize_with = "one_or_many")]
pub callbacks: Option<Vec<Value>>,
#[serde(deserialize_with = "one_or_many")]

4
litellm-rust/crates/config/src/value.rs
View file

@ -10,6 +10,10 @@ pub type AdditionalFields = BTreeMap<String, Value>;
pub struct Object(BTreeMap<String, Value>);
impl Object {
pub(crate) fn from_fields(fields: BTreeMap<String, Value>) -> Self {
Self(fields)
}
pub fn get(&self, key: &str) -> Option<&Value> {
self.0.get(key)
}

405
litellm-rust/crates/config/tests/redis.rs Normal file
View file

@ -0,0 +1,405 @@
use std::{
collections::HashMap,
fmt::Write,
sync::{Arc, Mutex},
};
use litellm_config::{CacheSettings, Config, Error, RedisEndpoint, RedisSettings, ResolvedRedis};
use rstest::rstest;
use tracing::{Event, Subscriber, field::Visit};
use tracing_subscriber::{Layer, layer::Context, prelude::*};
fn resolve(
yaml: &str,
environment: &[(&str, &str)],
) -> Result<litellm_config::ResolvedRedis, Error> {
let environment: HashMap<String, String> = environment
.iter()
.map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
.collect();
let config = Config::from_yaml(yaml).unwrap();
config
.general_settings
.coordination_redis
.as_ref()
.unwrap_or(&RedisSettings::default())
.resolve(&|name| environment.get(name).cloned())
}
fn resolve_cache(yaml: &str) -> Result<ResolvedRedis, Error> {
let config = Config::from_yaml(yaml).unwrap();
match config.litellm_settings.cache_params.unwrap() {
CacheSettings::Redis(settings) => settings.resolve_cache(&|_| None),
CacheSettings::Other(_) => panic!("expected a Redis cache"),
}
}
fn cluster_ports(resolved: &ResolvedRedis) -> Vec<u16> {
match &resolved.endpoint {
RedisEndpoint::Cluster(cluster) => cluster.nodes.iter().map(|node| node.port).collect(),
endpoint => panic!("expected a cluster, got {endpoint:?}"),
}
}
#[rstest]
#[case::host(
"general_settings:\n coordination_redis:\n host: redis\n port: 6380\n db: 2\n ssl: true\n namespace: app\n socket_timeout: 1.5",
&[],
"host:redis:6380:2:true:app:1500"
)]
#[case::environment_url(
"model_list: []",
&[("REDIS_URL", "redis://environment:6379")],
"url:redis://environment:6379"
)]
#[case::explicit_host_drops_environment_url(
"general_settings:\n coordination_redis:\n host: configured",
&[("REDIS_URL", "redis://environment:6379")],
"host:configured:6379:0:false::none"
)]
#[case::yaml_cluster(
"general_settings:\n coordination_redis:\n startup_nodes:\n - host: node-a\n port: 7000\n - host: node-b\n port: \"7001\"",
&[],
"cluster:node-a:7000,node-b:7001"
)]
#[case::json_environment_cluster(
"model_list: []",
&[("REDIS_CLUSTER_NODES", r#"[{"host":"node-a","port":"7000"},{"host":"node-b","port":7001}]"#)],
"cluster:node-a:7000,node-b:7001"
)]
#[case::missing_reference_keeps_environment_fallback(
"general_settings:\n coordination_redis:\n host: os.environ/MISSING_HOST",
&[("REDIS_HOST", "environment")],
"host:environment:6379:0:false::none"
)]
#[case::configured_values_beat_the_same_environment_values(
"general_settings:\n coordination_redis:\n host: configured\n port: 6380\n db: 3",
&[("REDIS_HOST", "environment"), ("REDIS_PORT", "6390"), ("REDIS_DB", "9")],
"host:configured:6380:3:false::none"
)]
#[case::environment_url_beats_the_environment_host(
"model_list: []",
&[("REDIS_URL", "redis://environment:6379"), ("REDIS_HOST", "ignored")],
"url:redis://environment:6379"
)]
#[case::startup_nodes_beat_a_url(
"general_settings:\n coordination_redis:\n url: redis://single:6379\n startup_nodes:\n - host: node-a\n port: 7000",
&[],
"cluster:node-a:7000"
)]
#[case::environment_references_resolve_the_host_and_port(
"general_settings:\n coordination_redis:\n host: os.environ/COORD_HOST\n port: os.environ/COORD_PORT",
&[("COORD_HOST", "referenced"), ("COORD_PORT", "6400")],
"host:referenced:6400:0:false::none"
)]
#[case::a_null_host_falls_back_to_the_environment(
"general_settings:\n coordination_redis:\n host: null",
&[("REDIS_HOST", "environment")],
"host:environment:6379:0:false::none"
)]
fn resolves_redis_through_the_parsed_config(
#[case] yaml: &str,
#[case] environment: &[(&str, &str)],
#[case] expected: &str,
) {
let resolved = resolve(yaml, environment).unwrap();
let description = match resolved.endpoint {
RedisEndpoint::Url(url) => format!("url:{}", url.expose()),
RedisEndpoint::Host(host) => format!(
"host:{}:{}:{}:{}:{}:{}",
host.host,
host.port,
host.db,
host.tls,
resolved.namespace.unwrap_or_default(),
resolved
.timeout
.map(|timeout| timeout.as_millis().to_string())
.unwrap_or_else(|| "none".to_owned())
),
RedisEndpoint::Cluster(cluster) => format!(
"cluster:{}",
cluster
.nodes
.iter()
.map(|node| format!("{}:{}", node.host, node.port))
.collect::<Vec<_>>()
.join(",")
),
};
assert_eq!(description, expected);
}
#[rstest]
#[case::custom_tls_file(
"general_settings:\n coordination_redis:\n host: redis\n ssl_ca_certs: /ca.pem",
&[],
"ssl_ca_certs"
)]
#[case::gcp_auth_from_config(
"general_settings:\n coordination_redis:\n host: redis\n gcp_service_account: account",
&[],
"gcp_service_account"
)]
#[case::gcp_auth_from_environment(
"model_list: []",
&[("REDIS_HOST", "redis"), ("REDIS_GCP_SERVICE_ACCOUNT", "account")],
"REDIS_GCP_SERVICE_ACCOUNT"
)]
#[case::azure_auth_from_config(
"general_settings:\n coordination_redis:\n host: redis\n azure_redis_ad_token: true",
&[],
"azure_redis_ad_token"
)]
#[case::azure_auth_from_environment(
"model_list: []",
&[("REDIS_HOST", "redis"), ("REDIS_AZURE_AD_TOKEN", "true")],
"REDIS_AZURE_AD_TOKEN"
)]
#[case::aws_auth_from_config(
"general_settings:\n coordination_redis:\n host: redis\n aws_iam_auth: true",
&[],
"aws_iam_auth"
)]
#[case::aws_auth_from_environment(
"model_list: []",
&[("REDIS_HOST", "redis"), ("REDIS_AWS_IAM_AUTH", "true")],
"REDIS_AWS_IAM_AUTH"
)]
#[case::sentinel_from_config(
"general_settings:\n coordination_redis:\n sentinel_nodes: [[redis, 26379]]",
&[],
"sentinel_nodes"
)]
#[case::sentinel_from_environment(
"model_list: []",
&[("REDIS_SENTINEL_NODES", "[[\"redis\",26379]]")],
"REDIS_SENTINEL_NODES"
)]
fn rejected_settings_name_their_source(
#[case] yaml: &str,
#[case] environment: &[(&str, &str)],
#[case] expected: &str,
) {
let error = resolve(yaml, environment).unwrap_err();
assert!(matches!(error, Error::RedisSetting(setting) if setting == expected));
}
#[rstest]
fn an_explicit_host_keeps_environment_credentials() {
let resolved = resolve(
"general_settings:\n coordination_redis:\n host: configured",
&[
("REDIS_PASSWORD", "environment-secret"),
("REDIS_SSL", "True"),
],
)
.unwrap();
let RedisEndpoint::Host(host) = resolved.endpoint else {
panic!("expected a host endpoint");
};
assert_eq!(
host.password.as_ref().map(|password| password.expose()),
Some("environment-secret")
);
assert!(host.tls);
}
#[rstest]
#[case::only_a_port("general_settings:\n coordination_redis:\n port: 6379", &[])]
#[case::a_structured_host_with_an_environment_target(
"general_settings:\n coordination_redis:\n host: []",
&[("REDIS_HOST", "environment"), ("REDIS_URL", "redis://environment:6379")]
)]
fn a_block_without_a_usable_target_fails(#[case] yaml: &str, #[case] environment: &[(&str, &str)]) {
let error = resolve(yaml, environment).unwrap_err();
assert!(matches!(
error,
Error::RedisInvalid {
setting: "host",
..
}
));
}
#[rstest]
fn the_cache_alias_names_cluster_nodes() {
let resolved = resolve_cache(
"litellm_settings:\n cache_params:\n type: redis\n redis_startup_nodes:\n - host: node-a\n port: 7000",
)
.unwrap();
assert_eq!(cluster_ports(&resolved), vec![7000]);
}
#[rstest]
fn canonical_startup_nodes_beat_the_cache_alias() {
let resolved = resolve_cache(
"litellm_settings:\n cache_params:\n type: redis\n startup_nodes:\n - host: canonical\n port: 7000\n redis_startup_nodes:\n - host: alias\n port: 7100\n - host: alias\n port: 7101",
)
.unwrap();
assert_eq!(cluster_ports(&resolved), vec![7000]);
}
#[rstest]
fn a_null_canonical_startup_nodes_setting_shadows_the_cache_alias() {
let resolved = resolve_cache(
"litellm_settings:\n cache_params:\n type: redis\n host: cache\n startup_nodes: null\n redis_startup_nodes:\n - host: alias\n port: 7100",
)
.unwrap();
assert!(matches!(resolved.endpoint, RedisEndpoint::Host(_)));
}
#[rstest]
fn a_missing_canonical_startup_nodes_reference_shadows_the_cache_alias() {
let resolved = resolve_cache(
"litellm_settings:\n cache_params:\n type: redis\n host: cache\n startup_nodes: os.environ/MISSING_NODES\n redis_startup_nodes:\n - host: alias\n port: 7100",
)
.unwrap();
assert!(matches!(resolved.endpoint, RedisEndpoint::Host(_)));
}
#[rstest]
fn rejected_environment_settings_never_print_the_value() {
let error = resolve(
"model_list: []",
&[
("REDIS_HOST", "redis"),
("REDIS_SSL_CERT_REQS", "private-option-value"),
],
)
.unwrap_err();
let message = error.to_string();
assert!(message.contains("REDIS_SSL_CERT_REQS"));
assert!(!message.contains("private-option-value"));
}
#[rstest]
#[case::negative("-1")]
#[case::not_a_number("slow")]
fn rejects_bad_socket_timeout(#[case] timeout: &str) {
let yaml = format!(
"general_settings:\n coordination_redis:\n host: redis\n socket_timeout: {timeout}"
);
let error = resolve(&yaml, &[]).unwrap_err();
assert!(matches!(
error,
Error::RedisInvalid {
setting: "socket_timeout",
..
}
));
}
#[rstest]
fn defaults_socket_timeout_for_the_gateway() {
let resolved = resolve(
"general_settings:\n coordination_redis:\n host: redis",
&[],
)
.unwrap();
assert_eq!(resolved.timeout, None);
}
#[rstest]
fn debug_redacts_redis_secrets_and_url() {
let config = Config::from_yaml(
"general_settings:\n coordination_redis:\n url: redis://user:password@redis:6379\n username: credential-user\n password: credential-password",
)
.unwrap();
let settings = config.general_settings.coordination_redis.unwrap();
let resolved = settings.resolve(&|_| None).unwrap();
let debug = format!("{settings:?} {resolved:?}");
assert!(!debug.contains("redis://user:password"));
assert!(!debug.contains("credential-user"));
assert!(!debug.contains("credential-password"));
assert!(debug.contains("redacted") || debug.contains("REDACTED"));
}
#[rstest]
fn inactive_redis_cache_keeps_malformed_known_inputs_deferred() {
let config = Config::from_yaml(
"litellm_settings:\n cache: false\n cache_params:\n type: redis\n host: []\n port: invalid\n socket_timeout: invalid",
)
.unwrap();
assert!(matches!(
config.litellm_settings.cache_params,
Some(litellm_config::CacheSettings::Redis(_))
));
}
#[rstest]
fn non_redis_cache_settings_remain_opaque() {
let config = Config::from_yaml(
"litellm_settings:\n cache_params:\n type: local\n custom_option:\n nested: true",
)
.unwrap();
assert!(matches!(
config.litellm_settings.cache_params,
Some(litellm_config::CacheSettings::Other(_))
));
}
#[rstest]
fn legacy_unknown_settings_do_not_change_resolution() {
let resolved = resolve(
"general_settings:\n coordination_redis:\n host: redis\n redis_flush_size: 42\n service_name: primary",
&[],
)
.unwrap();
assert!(matches!(resolved.endpoint, RedisEndpoint::Host(_)));
}
struct WarningCapture(Arc<Mutex<Vec<String>>>);
impl<S: Subscriber> Layer<S> for WarningCapture {
fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
let mut fields = Fields::default();
event.record(&mut fields);
self.0.lock().unwrap().push(fields.0);
}
}
#[derive(Default)]
struct Fields(String);
impl Visit for Fields {
fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
write!(self.0, "{}={value:?} ", field.name()).unwrap();
}
fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
write!(self.0, "{}={value} ", field.name()).unwrap();
}
}
#[rstest]
fn ignored_setting_warning_names_the_setting_without_its_value() {
let events = Arc::new(Mutex::new(Vec::new()));
let subscriber = tracing_subscriber::registry().with(WarningCapture(events.clone()));
tracing::subscriber::with_default(subscriber, || {
resolve(
"general_settings:\n coordination_redis:\n host: redis\n max_connections: 47",
&[],
)
.unwrap();
});
let warnings = events.lock().unwrap().join(" ");
assert!(warnings.contains("max_connections"));
assert!(!warnings.contains("47"));
}

7
litellm-rust/crates/gateway/AGENTS.md
View file

@ -3,3 +3,10 @@
- Endpoint paths, request handling, model resolution, and response encoding belong to the mounted crates. Provider execution belongs to the `inference-<fmt>` crates and `llms`
- Inject shared state and infrastructure. Avoid global runtimes, duplicate client pools, and abstractions for hypothetical endpoint groups
- Test mounting and server lifecycle through public HTTP behavior. Test endpoint semantics in the owning crate
## Redis
- `build_redis` builds the Redis clients once at boot (`src/redis.rs`). Components get a clone as a field, never a per-request extractor or a pool. redis-rs multiplexes one connection per node, so there is no connection pool and pool settings like `max_connections` are accepted and ignored
- Which Redis backs which role mirrors the proxy: coordination is `general_settings.coordination_redis`, else the `cache_params` Redis when `cache: true` and `type: redis`, else `REDIS_*` env. The transaction buffer reuses coordination, else env when `use_redis_transaction_buffer` is on. The response cache gets its own client so large values never queue in front of coordination commands
- `litellm-config` follows `litellm/_redis.py`'s merge of config blocks and `REDIS_*` env. A redis-py setting we can't honor (sentinel, IAM, custom TLS files, ...) fails boot with its name, never silently dropped
- Boot never waits for a configured Redis (clients connect lazily). Only the env-only coordination fallback is pinged, for 2s, and treated as absent when unreachable, as in Python
- `/health/readiness` never touches Redis, matching the proxy

3
litellm-rust/crates/gateway/Cargo.toml
View file

@ -19,6 +19,8 @@ litellm-gateway-auth.workspace = true
litellm-gateway-ui.workspace = true
litellm-auth-types.workspace = true
litellm-config.workspace = true
litellm-circuit-breaker.workspace = true
litellm-redis-client.workspace = true
litellm-http = { workspace = true, features = ["mcp"] }
litellm-llms.workspace = true
litellm-secrets.workspace = true
@ -31,6 +33,7 @@ uuid.workspace = true
tower-sessions-moka-store = "0.15.0"
[dev-dependencies]
litellm-redis-testing.workspace = true
base64.workspace = true
rstest.workspace = true
tempfile.workspace = true

9
litellm-rust/crates/gateway/src/error.rs
View file

@ -8,6 +8,15 @@ pub enum Error {
McpSetting(String),
#[error(transparent)]
Mcp(#[from] litellm_gateway_mcp::ConnectError),
#[error(transparent)]
Config(#[from] litellm_config::Error),
#[error("invalid Redis setting {setting}: {reason}")]
RedisInvalid {
setting: &'static str,
reason: &'static str,
},
#[error(transparent)]
RedisConnect(#[from] litellm_redis_client::ConnectError),
#[error("MCP requires a configured master key")]
Auth(#[from] litellm_gateway_auth::Error),
}

22
litellm-rust/crates/gateway/src/lib.rs
View file

@ -1,7 +1,9 @@
mod auth;
mod error;
mod redis;
mod secrets;
pub use error::Error;
pub use redis::RedisClients;
use std::{sync::Arc, time::Instant};
@ -70,6 +72,17 @@ pub fn build_inference(config: &Config) -> Result<Arc<Gateway>, Error> {
)?))
}
pub async fn build_redis(config: &Config) -> Result<RedisClients, Error> {
let environment = secrets::environment_values(&config.environment_variables)?;
let lookup = |name: &str| {
environment
.get(name)
.map(|value| value.expose().to_owned())
.or_else(|| std::env::var(name).ok())
};
redis::connect(config, &lookup).await
}
pub async fn build_mcp(
config: &Config,
secrets: Arc<dyn litellm_secrets::source::SecretSource>,
@ -128,12 +141,17 @@ pub fn router(
litellm_gateway_auth::authenticate,
))
.layer(axum::middleware::from_fn(log_request));
let health = Router::new().route("/health/readiness", axum::routing::get(readiness));
match ui {
Some(ui) => inference.merge(ui),
None => inference,
Some(ui) => inference.merge(ui).merge(health),
None => inference.merge(health),
}
}
async fn readiness() -> axum::Json<serde_json::Value> {
axum::Json(serde_json::json!({"status": "healthy", "db": "Not connected"}))
}
async fn log_request(request: Request, next: Next) -> Response {
let request_id = Uuid::new_v4().to_string();
let log_body_chunks = tracing::enabled!(tracing::Level::DEBUG);

1
litellm-rust/crates/gateway/src/main.rs
View file

@ -46,6 +46,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
Logger::new(StderrSink { level }).install_global()?;
let config = Config::load(settings.litellm_config)?;
let inference = litellm_gateway::build_inference(&config)?;
let _redis = litellm_gateway::build_redis(&config).await?;
let ui = match std::env::var_os("LITELLM_UI_PATH") {
Some(directory) => {
let ui_settings: UiSettings = settings::from_iter(std::env::vars_os())?;

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

@ -0,0 +1,577 @@
use std::time::Duration;
use litellm_circuit_breaker::{Breaker, Config as BreakerConfig, State};
use litellm_config::{
CacheSettings, Config, Lookup, RedisEndpoint, RedisSettings, ResolvedRedis, Value,
environment_has_target,
};
use litellm_redis_client::{
ClusterEndpoint, ClusterNode, DEFAULT_TIMEOUT, Endpoint, HostEndpoint, Namespace, RedisClient,
RedisConfig,
};
use crate::Error;
const ENVIRONMENT_PING_TIMEOUT: Duration = Duration::from_secs(2);
type Params = ResolvedRedis;
pub struct RedisClients {
pub coordination: Option<RedisClient>,
pub transaction_buffer: Option<RedisClient>,
pub response_cache: Option<RedisClient>,
}
#[derive(Debug, PartialEq)]
pub(crate) struct Plan {
coordination: Option<Target>,
transaction_buffer_fallback: Option<Params>,
response_cache: Option<Params>,
}
#[derive(Debug, PartialEq)]
enum Target {
Configured(Params),
Environment(Params),
}
pub(crate) fn plan(config: &Config, lookup: Lookup<'_>) -> Result<Plan, Error> {
let response_cache = redis_cache_params(config)
.map(|settings| settings.resolve_cache(lookup))
.transpose()?;
let coordination = match &config.general_settings.coordination_redis {
Some(settings) => Some(Target::Configured(settings.resolve(lookup)?)),
None => match (&response_cache, config.litellm_settings.cache == Some(true)) {
(Some(cache), _) => Some(Target::Configured(cache.clone())),
(None, true) => environment(lookup)?.map(Target::Configured),
(None, false) => optional_environment(lookup).map(Target::Environment),
},
};
let transaction_buffer_fallback = match (&coordination, use_transaction_buffer(config)) {
(Some(Target::Configured(_)), _) | (_, false) => None,
(Some(Target::Environment(_)) | None, true) => environment(lookup)?,
};
Ok(Plan {
coordination,
transaction_buffer_fallback,
response_cache,
})
}
pub async fn connect(config: &Config, lookup: Lookup<'_>) -> Result<RedisClients, Error> {
let plan = plan(config, lookup)?;
let breaker = breaker_config(lookup)?;
let coordination = match plan.coordination {
Some(Target::Configured(params)) => Some(client(params, breaker)?),
Some(Target::Environment(params)) => match client(params, breaker) {
Ok(client) => reachable(client).await,
Err(_) => {
tracing::warn!(
"REDIS_* environment variables did not form a valid Redis client, cross-pod coordination stays in memory"
);
None
}
},
None => None,
};
let transaction_buffer = match (&coordination, plan.transaction_buffer_fallback) {
(Some(coordination), _) => Some(coordination.clone()),
(None, Some(params)) => Some(client(params, breaker)?),
(None, None) => None,
};
let response_cache = plan
.response_cache
.map(|params| client(params, breaker))
.transpose()?;
Ok(RedisClients {
coordination,
transaction_buffer,
response_cache,
})
}
fn redis_cache_params(config: &Config) -> Option<&RedisSettings> {
let settings = &config.litellm_settings;
let params = settings.cache_params.as_ref()?;
match (settings.cache, params) {
(Some(true), CacheSettings::Redis(redis)) => Some(redis),
_ => None,
}
}
fn environment(lookup: Lookup<'_>) -> Result<Option<Params>, Error> {
match environment_has_target(lookup) {
true => RedisSettings::default()
.resolve(lookup)
.map(Some)
.map_err(Into::into),
false => Ok(None),
}
}
fn optional_environment(lookup: Lookup<'_>) -> Option<Params> {
match environment(lookup) {
Ok(params) => params,
Err(error) => {
tracing::warn!(error = %error, "REDIS_* environment variables are invalid, cross-pod coordination stays in memory");
None
}
}
}
fn use_transaction_buffer(config: &Config) -> bool {
match config
.general_settings
.additional_fields
.get("use_redis_transaction_buffer")
{
Some(Value::Bool(enabled)) => *enabled,
Some(Value::String(text)) => matches!(
text.to_ascii_lowercase().as_str(),
"true" | "1" | "yes" | "on"
),
_ => false,
}
}
async fn reachable(client: RedisClient) -> Option<RedisClient> {
match tokio::time::timeout(ENVIRONMENT_PING_TIMEOUT, client.ping()).await {
Ok(Ok(())) => Some(client),
Ok(Err(error)) => {
tracing::warn!(%error, "REDIS_* environment variables named a Redis that is not reachable, cross-pod coordination stays in memory");
None
}
Err(_) => {
tracing::warn!(
"REDIS_* environment variables named a Redis that did not answer PING in 2s, cross-pod coordination stays in memory"
);
None
}
}
}
fn client(params: Params, breaker: Option<BreakerConfig>) -> Result<RedisClient, Error> {
tracing::info!(endpoint = ?params.endpoint, namespace = ?params.namespace, "redis client configured");
let endpoint = match params.endpoint {
RedisEndpoint::Url(url) => Endpoint::Url(url.expose().to_owned()),
RedisEndpoint::Host(host) => Endpoint::Host(HostEndpoint {
host: host.host,
port: host.port,
username: host.username.map(|value| value.expose().to_owned()),
password: host.password.map(|value| value.expose().to_owned()),
tls: host.tls,
db: host.db,
}),
RedisEndpoint::Cluster(cluster) => Endpoint::Cluster(ClusterEndpoint {
nodes: cluster
.nodes
.into_iter()
.map(|node| ClusterNode {
host: node.host,
port: node.port,
})
.collect(),
username: cluster.username.map(|value| value.expose().to_owned()),
password: cluster.password.map(|value| value.expose().to_owned()),
tls: cluster.tls,
}),
};
Ok(RedisClient::new(RedisConfig {
endpoint,
namespace: params.namespace.map(Namespace::new).unwrap_or_default(),
timeout: params.timeout.unwrap_or(DEFAULT_TIMEOUT),
breaker: breaker.map(|config| Breaker::new(config, Box::new(log_transition))),
})?)
}
fn log_transition(transition: &litellm_circuit_breaker::Transition<State>) {
match transition.to {
State::Open { .. } => tracing::warn!(
generation = transition.generation,
"redis circuit breaker opened, fast-failing redis calls"
),
State::HalfOpen => tracing::info!(
generation = transition.generation,
"redis circuit breaker half open, probing redis"
),
State::Closed { .. } => tracing::info!(
generation = transition.generation,
"redis circuit breaker closed, redis recovered"
),
}
}
fn breaker_config(lookup: Lookup<'_>) -> Result<Option<BreakerConfig>, Error> {
let enabled = lookup("REDIS_CIRCUIT_BREAKER_ENABLED")
.is_none_or(|value| value.eq_ignore_ascii_case("true"));
if !enabled {
return Ok(None);
}
let defaults = BreakerConfig::default();
let seconds = |name: &'static str, default: Duration| -> Result<Duration, Error> {
lookup(name)
.map(|value| {
value
.parse::<f64>()
.ok()
.and_then(|seconds| Duration::try_from_secs_f64(seconds).ok())
.ok_or(Error::RedisInvalid {
setting: name,
reason: "must be a non-negative number of seconds",
})
})
.unwrap_or(Ok(default))
};
let failure_threshold = lookup("REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD")
.map(|value| {
value.parse().map_err(|_| Error::RedisInvalid {
setting: "REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD",
reason: "must be an integer",
})
})
.unwrap_or(Ok(defaults.failure_threshold))?;
Ok(Some(BreakerConfig {
failure_threshold,
recovery_timeout: seconds(
"REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT",
defaults.recovery_timeout,
)?,
timeout_min_duration: seconds(
"REDIS_CIRCUIT_BREAKER_TIMEOUT_MIN_DURATION",
defaults.timeout_min_duration,
)?,
}))
}
#[cfg(test)]
mod tests {
use std::{collections::HashMap, time::Duration};
use litellm_config::{Config, RedisEndpoint, RedisHost};
use litellm_redis_testing::RedisServer;
use rstest::rstest;
use super::{Params, Plan, Target, connect, plan};
use crate::Error;
fn host(name: &str, port: u16) -> Params {
Params {
endpoint: RedisEndpoint::Host(RedisHost {
host: name.to_owned(),
port,
username: None,
password: None,
tls: false,
db: 0,
}),
namespace: None,
timeout: None,
}
}
fn resolved(yaml: &str, environment: &[(&str, &str)]) -> Result<Plan, Error> {
let environment: HashMap<String, String> = environment
.iter()
.map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
.collect();
plan(&Config::from_yaml(yaml).unwrap(), &|name| {
environment.get(name).cloned()
})
}
fn configured(coordination: Params, response_cache: Option<Params>) -> Plan {
Plan {
coordination: Some(Target::Configured(coordination)),
transaction_buffer_fallback: None,
response_cache,
}
}
const NO_REDIS: Plan = Plan {
coordination: None,
transaction_buffer_fallback: None,
response_cache: None,
};
#[rstest]
#[case::nothing_configured("model_list: []", &[], NO_REDIS)]
#[case::coordination_block_wins_over_the_cache_redis(
"general_settings:\n coordination_redis:\n host: coord\nlitellm_settings:\n cache: true\n cache_params:\n type: redis\n host: cache",
&[],
configured(host("coord", 6379), Some(host("cache", 6379)))
)]
#[case::cache_params_without_cache_true_are_unused(
"litellm_settings:\n cache_params:\n type: redis\n host: cache",
&[],
NO_REDIS
)]
#[case::non_redis_cache_takes_coordination_from_the_environment_without_a_ping(
"litellm_settings:\n cache: true\n cache_params:\n type: local",
&[("REDIS_HOST", "env")],
configured(host("env", 6379), None)
)]
#[case::environment_only_coordination_is_probed(
"model_list: []",
&[("REDIS_HOST", "env")],
Plan { coordination: Some(Target::Environment(host("env", 6379))), transaction_buffer_fallback: None, response_cache: None }
)]
#[case::the_transaction_buffer_falls_back_to_the_environment_only_when_enabled(
"general_settings:\n use_redis_transaction_buffer: \"true\"",
&[("REDIS_HOST", "env")],
Plan { coordination: Some(Target::Environment(host("env", 6379))), transaction_buffer_fallback: Some(host("env", 6379)), response_cache: None }
)]
fn resolves_each_redis_role_like_the_proxy(
#[case] yaml: &str,
#[case] environment: &[(&str, &str)],
#[case] expected: Plan,
) {
assert_eq!(resolved(yaml, environment).unwrap(), expected);
}
#[rstest]
fn redis_migration_regression_preserves_borrowed_cache_namespace() {
let yaml = "litellm_settings:\n cache: true\n cache_params:\n type: redis\n host: cache\n namespace: responses\n ttl: 600";
let cache = Params {
namespace: Some("responses".to_owned()),
..host("cache", 6379)
};
assert_eq!(
resolved(yaml, &[]).unwrap(),
configured(cache.clone(), Some(cache))
);
}
#[rstest]
#[tokio::test]
async fn borrowed_coordination_and_transaction_buffer_share_the_cache_namespace() {
let server = RedisServer::start().await.unwrap();
let config = Config::from_yaml(&format!(
"litellm_settings:\n cache: true\n cache_params:\n type: redis\n host: 127.0.0.1\n port: {}\n namespace: responses",
port_of(&server)
))
.unwrap();
let clients = connect(&config, &|_| None).await.unwrap();
let coordination = clients.coordination.unwrap();
let buffer = clients.transaction_buffer.unwrap();
let cache = clients.response_cache.unwrap();
let unnamespaced = connect(&coordination_config(&server), &|_| None)
.await
.unwrap()
.coordination
.unwrap();
coordination
.set("rate-limit", "1", Duration::from_secs(60))
.await
.unwrap();
buffer
.set("spend", "2", Duration::from_secs(60))
.await
.unwrap();
assert_eq!(
cache.get::<String>("rate-limit").await.unwrap().as_deref(),
Some("1")
);
assert_eq!(
cache.get::<String>("spend").await.unwrap().as_deref(),
Some("2")
);
assert_eq!(
unnamespaced
.get::<String>("responses:rate-limit")
.await
.unwrap()
.as_deref(),
Some("1")
);
assert_eq!(
unnamespaced
.get::<String>("responses:spend")
.await
.unwrap()
.as_deref(),
Some("2")
);
assert!(
unnamespaced
.get::<String>("rate-limit")
.await
.unwrap()
.is_none()
);
assert!(unnamespaced.get::<String>("spend").await.unwrap().is_none());
}
#[rstest]
#[case::malformed_port(&[("REDIS_HOST", "env"), ("REDIS_PORT", "not-a-port")])]
#[case::malformed_cluster_nodes(&[("REDIS_CLUSTER_NODES", "[{\"host\": ")])]
fn redis_migration_regression_malformed_optional_environment_is_absent(
#[case] environment: &[(&str, &str)],
) {
assert_eq!(resolved("model_list: []", environment).unwrap(), NO_REDIS);
}
#[rstest]
fn redis_migration_regression_explicit_malformed_setting_still_fails() {
let error = resolved(
"general_settings:\n coordination_redis:\n host: coord\n port: not-a-port",
&[],
)
.unwrap_err();
assert!(matches!(
error,
Error::Config(litellm_config::Error::RedisInvalid {
setting: "port",
..
})
));
}
#[rstest]
fn redis_migration_regression_enabled_transaction_buffer_rejects_invalid_environment() {
let error = resolved(
"general_settings:\n use_redis_transaction_buffer: true",
&[("REDIS_HOST", "env"), ("REDIS_PORT", "not-a-port")],
)
.unwrap_err();
assert!(matches!(
error,
Error::Config(litellm_config::Error::RedisInvalid {
setting: "port",
..
})
));
}
#[rstest]
fn redis_migration_regression_invalid_response_cache_still_fails() {
let error = resolved(
"litellm_settings:\n cache: true\n cache_params:\n type: redis\n host: cache\n port: not-a-port",
&[],
)
.unwrap_err();
assert!(matches!(
error,
Error::Config(litellm_config::Error::RedisInvalid {
setting: "port",
..
})
));
}
#[rstest]
#[tokio::test]
async fn redis_migration_regression_invalid_optional_environment_url_is_absent() {
let config = Config::from_yaml("model_list: []").unwrap();
let clients = connect(&config, &|name| {
(name == "REDIS_URL").then(|| "not a redis URL".to_owned())
})
.await
.unwrap();
assert!(clients.coordination.is_none());
}
fn port_of(server: &RedisServer) -> String {
server.url().rsplit(':').next().unwrap().to_owned()
}
fn coordination_config(server: &RedisServer) -> Config {
Config::from_yaml(&format!(
"general_settings:\n coordination_redis:\n host: 127.0.0.1\n port: {}\n socket_timeout: 0.5",
port_of(server)
))
.unwrap()
}
async fn eventually_pings(client: &litellm_redis_client::RedisClient) -> bool {
let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
while tokio::time::Instant::now() < deadline {
if client.ping().await.is_ok() {
return true;
}
tokio::time::sleep(Duration::from_millis(200)).await;
}
false
}
#[rstest]
#[tokio::test]
async fn boot_does_not_wait_for_a_configured_redis_that_is_down() {
let server = RedisServer::start().await.unwrap();
server.stop().await.unwrap();
let clients = connect(&coordination_config(&server), &|_| None)
.await
.unwrap();
let coordination = clients.coordination.unwrap();
let while_down = coordination.ping().await;
server.resume().await.unwrap();
assert!(while_down.is_err());
assert!(eventually_pings(&coordination).await);
}
#[rstest]
#[tokio::test]
async fn the_coordination_client_survives_a_redis_restart_and_backs_the_transaction_buffer() {
let server = RedisServer::start().await.unwrap();
let clients = connect(&coordination_config(&server), &|_| None)
.await
.unwrap();
let coordination = clients.coordination.unwrap();
coordination
.set("written", "1", Duration::from_secs(60))
.await
.unwrap();
server.restart().await.unwrap();
assert!(eventually_pings(&coordination).await);
let buffer = clients.transaction_buffer.unwrap();
assert!(buffer.ping().await.is_ok());
}
#[rstest]
#[tokio::test]
async fn an_unreachable_environment_only_redis_leaves_coordination_in_memory() {
let server = RedisServer::start().await.unwrap();
let port = port_of(&server);
server.stop().await.unwrap();
let config = Config::from_yaml("model_list: []").unwrap();
let clients = connect(&config, &|name| match name {
"REDIS_HOST" => Some("127.0.0.1".to_owned()),
"REDIS_PORT" => Some(port.clone()),
_ => None,
})
.await
.unwrap();
assert!(clients.coordination.is_none());
assert!(clients.transaction_buffer.is_none());
}
#[rstest]
#[tokio::test]
async fn a_reachable_environment_only_redis_becomes_coordination() {
let server = RedisServer::start().await.unwrap();
let port = port_of(&server);
let config = Config::from_yaml("model_list: []").unwrap();
let clients = connect(&config, &|name| match name {
"REDIS_HOST" => Some("127.0.0.1".to_owned()),
"REDIS_PORT" => Some(port.clone()),
_ => None,
})
.await
.unwrap();
assert!(clients.coordination.unwrap().ping().await.is_ok());
}
}

30
litellm-rust/crates/gateway/tests/server.rs
View file

@ -287,3 +287,33 @@ async fn ui_routes_are_absent_when_not_mounted(
.unwrap();
assert_eq!(response.status().as_u16(), 404);
}
#[rstest]
#[case::with_master_key("model_list: []\ngeneral_settings:\n master_key: gateway-key\n")]
#[case::without_master_key("model_list: []\n")]
#[tokio::test]
async fn readiness_answers_without_credentials_like_the_proxy_with_no_database(
inference: Arc<Gateway>,
#[case] yaml: &str,
) {
let config = Config::from_yaml(yaml).unwrap();
let app = litellm_gateway::router(inference, &config, None, None);
let response = app
.oneshot(
Request::get("/health/readiness")
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status().as_u16(), 200);
let body = axum::body::to_bytes(response.into_body(), 1024)
.await
.unwrap();
assert_eq!(
serde_json::from_slice::<Value>(&body).unwrap(),
json!({"status": "healthy", "db": "Not connected"})
);
}

52
litellm-rust/crates/redis-client/src/client.rs
View file

@ -26,15 +26,15 @@ pub enum Ttl {
}
impl RedisClient {
pub async fn connect(config: RedisConfig) -> Result<Self, ConnectError> {
pub fn new(config: RedisConfig) -> Result<Self, ConnectError> {
let connection = match &config.endpoint {
Endpoint::Url(url) => {
Connection::standalone(url.as_str().into_connection_info()?, config.timeout).await?
Connection::standalone(url.as_str().into_connection_info()?, config.timeout)?
}
Endpoint::Host(host) => {
Connection::standalone(host.connection_info()?, config.timeout).await?
Connection::standalone(host.connection_info()?, config.timeout)?
}
Endpoint::Cluster(cluster) => Connection::cluster(cluster, config.timeout).await?,
Endpoint::Cluster(cluster) => Connection::cluster(cluster, config.timeout)?,
};
Ok(Self {
connection,
@ -43,6 +43,10 @@ impl RedisClient {
})
}
pub async fn ping(&self) -> Result<(), Error> {
self.query(&commands::ping()).await
}
pub async fn get<T: FromRedisValue>(&self, key: &str) -> Result<Option<T>, Error> {
self.query(&commands::get(self.key(key))).await
}
@ -319,6 +323,35 @@ mod tests {
assert_eq!(recovered, Some(1));
}
#[rstest]
#[tokio::test]
async fn a_client_created_while_redis_is_down_works_once_redis_is_up(
#[values(Topology::Standalone, Topology::Cluster)] topology: Topology,
) {
let server = Server::start(topology).await;
server.stop().await;
let client = server.client("", Duration::from_millis(500)).await;
let while_down = client.ping().await;
server.resume().await;
assert!(
matches!(while_down, Err(Error::Transport(_))),
"{while_down:?}"
);
let deadline = Instant::now() + Duration::from_secs(30);
let recovered = loop {
match client.ping().await {
Ok(()) => break true,
Err(_) if Instant::now() < deadline => {
tokio::time::sleep(Duration::from_millis(200)).await
}
Err(_) => break false,
}
};
assert!(recovered);
}
#[rstest]
#[tokio::test]
async fn empty_batches_succeed_without_asking_redis(
@ -581,7 +614,7 @@ mod tests {
let server = RedisServer::start().await.unwrap();
let port = server.url().rsplit(':').next().unwrap().parse().unwrap();
let connect = |endpoint| {
RedisClient::connect(RedisConfig {
RedisClient::new(RedisConfig {
endpoint,
namespace: Namespace::default(),
timeout: DEFAULT_TIMEOUT,
@ -596,14 +629,9 @@ mod tests {
tls: false,
db: 3,
}))
.await
.unwrap();
let in_db_0 = connect(Endpoint::Url(server.url().to_owned()))
.await
.unwrap();
let via_url = connect(Endpoint::Url(format!("{}/3", server.url())))
.await
.unwrap();
let in_db_0 = connect(Endpoint::Url(server.url().to_owned())).unwrap();
let via_url = connect(Endpoint::Url(format!("{}/3", server.url()))).unwrap();
in_db_3
.set("k", "3", Duration::from_secs(60))

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

@ -2,6 +2,10 @@ use std::time::Duration;
use redis::{Cmd, ToRedisArgs, ToSingleRedisArg};
pub(crate) fn ping() -> Cmd {
redis::cmd("PING")
}
pub(crate) fn get(key: String) -> Cmd {
redis::cmd("GET").arg(key).to_owned()
}

74
litellm-rust/crates/redis-client/src/config.rs
View file

@ -1,4 +1,4 @@
use std::time::Duration;
use std::{fmt, time::Duration};
use redis::{ConnectionAddr, ConnectionInfo, IntoConnectionInfo, RedisConnectionInfo, RedisResult};
@ -15,12 +15,14 @@ pub struct RedisConfig {
pub breaker: Option<Breaker>,
}
#[derive(Clone, PartialEq)]
pub enum Endpoint {
Url(String),
Host(HostEndpoint),
Cluster(ClusterEndpoint),
}
#[derive(Clone, PartialEq)]
pub struct HostEndpoint {
pub host: String,
pub port: u16,
@ -30,6 +32,7 @@ pub struct HostEndpoint {
pub db: i64,
}
#[derive(Clone, PartialEq)]
pub struct ClusterEndpoint {
pub nodes: Vec<ClusterNode>,
pub username: Option<String>,
@ -37,6 +40,7 @@ pub struct ClusterEndpoint {
pub tls: bool,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ClusterNode {
pub host: String,
pub port: u16,
@ -78,3 +82,71 @@ fn addr(host: &str, port: u16, tls: bool) -> ConnectionAddr {
false => ConnectionAddr::Tcp(host.to_owned(), port),
}
}
impl fmt::Debug for Endpoint {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Url(url) => formatter
.debug_tuple("Url")
.field(&redacted_url(url))
.finish(),
Self::Host(host) => formatter
.debug_struct("Host")
.field("host", &host.host)
.field("port", &host.port)
.field("username", &host.username)
.field("password", &host.password.as_ref().map(|_| "<redacted>"))
.field("tls", &host.tls)
.field("db", &host.db)
.finish(),
Self::Cluster(cluster) => formatter
.debug_struct("Cluster")
.field("nodes", &cluster.nodes)
.field("username", &cluster.username)
.field("password", &cluster.password.as_ref().map(|_| "<redacted>"))
.field("tls", &cluster.tls)
.finish(),
}
}
}
fn redacted_url(url: &str) -> String {
match url
.split_once("://")
.and_then(|(scheme, rest)| Some((scheme, rest.rsplit_once('@')?.1)))
{
Some((scheme, host)) => format!("{scheme}://<redacted>@{host}"),
None => url.to_owned(),
}
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use super::{Endpoint, HostEndpoint};
#[rstest]
#[case::url_with_password(Endpoint::Url("redis://user:secret@cache:6379/0".to_owned()), "redis://<redacted>@cache:6379/0")]
#[case::url_with_at_in_password(Endpoint::Url("rediss://:p@ss@cache:6380".to_owned()), "rediss://<redacted>@cache:6380")]
#[case::host_password(
Endpoint::Host(HostEndpoint {
host: "cache".to_owned(),
port: 6379,
username: None,
password: Some("secret".to_owned()),
tls: false,
db: 0,
}),
"<redacted>"
)]
fn debug_output_never_shows_the_password(#[case] endpoint: Endpoint, #[case] shown: &str) {
let debug = format!("{endpoint:?}");
assert!(
!debug.contains("secret") && !debug.contains("p@ss"),
"{debug}"
);
assert!(debug.contains(shown), "{debug}");
}
}

20
litellm-rust/crates/redis-client/src/connection.rs
View file

@ -3,7 +3,7 @@ use std::time::Duration;
use redis::{
Cmd, ConnectionInfo, Pipeline, RedisFuture, RedisResult, Value,
aio::{ConnectionLike, ConnectionManager, ConnectionManagerConfig},
cluster::ClusterClientBuilder,
cluster::{ClusterClientBuilder, ClusterConfig},
cluster_async::ClusterConnection,
};
@ -18,21 +18,16 @@ pub(crate) enum Connection {
}
impl Connection {
pub(crate) async fn standalone(info: ConnectionInfo, timeout: Duration) -> RedisResult<Self> {
pub(crate) fn standalone(info: ConnectionInfo, timeout: Duration) -> RedisResult<Self> {
let config = ConnectionManagerConfig::new()
.set_connection_timeout(Some(timeout))
.set_response_timeout(Some(timeout))
.set_number_of_retries(RETRIES as usize);
let manager = redis::Client::open(info)?
.get_connection_manager_with_config(config)
.await?;
let manager = ConnectionManager::new_lazy_with_config(redis::Client::open(info)?, config)?;
Ok(Self::Standalone(manager))
}
pub(crate) async fn cluster(
endpoint: &ClusterEndpoint,
timeout: Duration,
) -> RedisResult<Self> {
pub(crate) fn cluster(endpoint: &ClusterEndpoint, timeout: Duration) -> RedisResult<Self> {
let builder = ClusterClientBuilder::new(endpoint.addrs())
.connection_timeout(timeout)
.response_timeout(timeout)
@ -45,8 +40,13 @@ impl Connection {
Some(password) => builder.password(password),
None => builder,
};
let config = ClusterConfig::new()
.set_connection_timeout(timeout)
.set_response_timeout(timeout);
Ok(Self::Cluster(
builder.build()?.get_async_connection().await?,
builder
.build()?
.get_pending_async_connection_with_config(config),
))
}
}

6
litellm-rust/crates/redis-client/src/namespace.rs
View file

@ -1,4 +1,4 @@
#[derive(Clone, Default)]
#[derive(Clone, Debug, Default, PartialEq)]
pub struct Namespace(Option<String>);
impl Namespace {
@ -7,6 +7,10 @@ impl Namespace {
Self((!namespace.is_empty()).then_some(namespace))
}
pub fn as_str(&self) -> Option<&str> {
self.0.as_deref()
}
pub(crate) fn apply(&self, key: &str) -> String {
match &self.0 {
Some(namespace) if !is_prefixed(key, namespace) => {

17
litellm-rust/crates/redis-client/src/test_support.rs
View file

@ -56,13 +56,12 @@ impl Server {
timeout: Duration,
breaker: Option<Breaker>,
) -> RedisClient {
RedisClient::connect(RedisConfig {
RedisClient::new(RedisConfig {
endpoint: self.endpoint(),
namespace: Namespace::new(namespace),
timeout,
breaker,
})
.await
.unwrap()
}
@ -73,6 +72,20 @@ impl Server {
}
}
pub(crate) async fn stop(&self) {
match self {
Self::Standalone(server) => server.stop().await.unwrap(),
Self::Cluster(cluster) => cluster.stop().await.unwrap(),
}
}
pub(crate) async fn resume(&self) {
match self {
Self::Standalone(server) => server.resume().await.unwrap(),
Self::Cluster(cluster) => cluster.resume().await.unwrap(),
}
}
pub(crate) async fn freeze(&self) {
match self {
Self::Standalone(server) => server.freeze().await.unwrap(),

13
litellm-rust/crates/redis-testing/src/cluster.rs
View file

@ -71,9 +71,16 @@ impl RedisCluster {
}
pub async fn restart(&self) -> Result<(), Error> {
self.container.stop().await?;
self.container.start().await?;
Ok(())
self.stop().await?;
self.resume().await
}
pub async fn stop(&self) -> Result<(), Error> {
Ok(self.container.stop().await?)
}
pub async fn resume(&self) -> Result<(), Error> {
Ok(self.container.start().await?)
}
pub async fn freeze(&self) -> Result<(), Error> {

13
litellm-rust/crates/redis-testing/src/standalone.rs
View file

@ -40,9 +40,16 @@ impl RedisServer {
}
pub async fn restart(&self) -> Result<(), Error> {
self.container.stop().await?;
self.container.start().await?;
Ok(())
self.stop().await?;
self.resume().await
}
pub async fn stop(&self) -> Result<(), Error> {
Ok(self.container.stop().await?)
}
pub async fn resume(&self) -> Result<(), Error> {
Ok(self.container.start().await?)
}
pub async fn freeze(&self) -> Result<(), Error> {