diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index 6b673a662e3..63137949f38 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -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", diff --git a/litellm-rust/crates/config/Cargo.toml b/litellm-rust/crates/config/Cargo.toml index 23d53923362..9746a002bb9 100644 --- a/litellm-rust/crates/config/Cargo.toml +++ b/litellm-rust/crates/config/Cargo.toml @@ -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"] } diff --git a/litellm-rust/crates/config/src/error.rs b/litellm-rust/crates/config/src/error.rs index 61d19491abc..95c33967cf0 100644 --- a/litellm-rust/crates/config/src/error.rs +++ b/litellm-rust/crates/config/src/error.rs @@ -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, + }, } diff --git a/litellm-rust/crates/config/src/lib.rs b/litellm-rust/crates/config/src/lib.rs index 8603ef3279c..b8af6f11d2e 100644 --- a/litellm-rust/crates/config/src/lib.rs +++ b/litellm-rust/crates/config/src/lib.rs @@ -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, diff --git a/litellm-rust/crates/config/src/redis.rs b/litellm-rust/crates/config/src/redis.rs new file mode 100644 index 00000000000..5674f993025 --- /dev/null +++ b/litellm-rust/crates/config/src/redis.rs @@ -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; + +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, + pub host: Option>, + pub port: Option>, + pub username: Option, + pub password: Option, + pub ssl: Option>, + pub db: Option>, + pub socket_timeout: Option>, + pub startup_nodes: Option, + pub namespace: Option>, + pub additional_fields: AdditionalFields, + unparsed: DeferredInputs, +} + +impl<'de> Deserialize<'de> for RedisSettings { + fn deserialize>(deserializer: D) -> Result { + let fields = BTreeMap::::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(|_| ""), + ) + .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), + Other(Object), +} + +impl<'de> Deserialize<'de> for CacheSettings { + fn deserialize>(deserializer: D) -> Result { + let fields = BTreeMap::::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, + pub timeout: Option, +} + +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(&"").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, + pub password: Option, + pub tls: bool, + pub db: i64, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct RedisCluster { + pub nodes: Vec, + pub username: Option, + pub password: Option, + 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; + +impl RedisSettings { + pub fn resolve(&self, lookup: Lookup<'_>) -> Result { + self.resolve_with(lookup, false) + } + + pub fn resolve_cache(&self, lookup: Lookup<'_>) -> Result { + self.resolve_with(lookup, true) + } + + fn resolve_with(&self, lookup: Lookup<'_>, cache: bool) -> Result { + 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::() + .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 { + 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 { + 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 { + kwargs.get(key).and_then(|entry| as_text(&entry.value)) +} + +fn secret_value(kwargs: &Kwargs, key: &str) -> Option { + 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(kwargs: &Kwargs, key: &'static str) -> Result, 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, Error> { + match value { + Value::String(json) => { + let nodes: Vec> = + 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, port: Option) -> Result { + 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 { + match value { + serde_json::Value::String(text) => Some(text.clone()), + serde_json::Value::Number(number) => Some(number.to_string()), + _ => None, + } +} + +fn spelled_text(value: Option<&Spelled>) -> Option { + value.map(|value| match value { + Spelled::Value(value) => value.to_string(), + Spelled::Text(value) => value.clone(), + }) +} + +fn spelled(text: Option) -> Option> { + text.map(|text| match text.parse() { + Ok(value) => Spelled::Value(value), + Err(_) => Spelled::Text(text), + }) +} + +fn cloned( + fields: &BTreeMap, + 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; + +fn deferred_inputs(fields: &BTreeMap) -> 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(), + } +} diff --git a/litellm-rust/crates/config/src/settings.rs b/litellm-rust/crates/config/src/settings.rs index 1094de1638e..9571c4fa5de 100644 --- a/litellm-rust/crates/config/src/settings.rs +++ b/litellm-rust/crates/config/src/settings.rs @@ -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, pub plugins: Option>, - pub coordination_redis: Option, + pub coordination_redis: Option, pub mcp_allowed_hosts: Option>, pub mcp_allowed_origins: Box<[String]>, #[serde(flatten)] @@ -220,7 +222,7 @@ pub struct LiteLlmSettings { pub request_timeout: Option>, pub num_retries: Option, pub cache: Option, - pub cache_params: Option, + pub cache_params: Option, #[serde(deserialize_with = "one_or_many")] pub callbacks: Option>, #[serde(deserialize_with = "one_or_many")] diff --git a/litellm-rust/crates/config/src/value.rs b/litellm-rust/crates/config/src/value.rs index 5becdc87b70..667a5afa130 100644 --- a/litellm-rust/crates/config/src/value.rs +++ b/litellm-rust/crates/config/src/value.rs @@ -10,6 +10,10 @@ pub type AdditionalFields = BTreeMap; pub struct Object(BTreeMap); impl Object { + pub(crate) fn from_fields(fields: BTreeMap) -> Self { + Self(fields) + } + pub fn get(&self, key: &str) -> Option<&Value> { self.0.get(key) } diff --git a/litellm-rust/crates/config/tests/redis.rs b/litellm-rust/crates/config/tests/redis.rs new file mode 100644 index 00000000000..ab409541c8e --- /dev/null +++ b/litellm-rust/crates/config/tests/redis.rs @@ -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 { + let environment: HashMap = 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 { + 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 { + 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::>() + .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>>); + +impl Layer 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")); +} diff --git a/litellm-rust/crates/gateway/AGENTS.md b/litellm-rust/crates/gateway/AGENTS.md index d54fa062f2b..a33bd19cc80 100644 --- a/litellm-rust/crates/gateway/AGENTS.md +++ b/litellm-rust/crates/gateway/AGENTS.md @@ -3,3 +3,10 @@ - Endpoint paths, request handling, model resolution, and response encoding belong to the mounted crates. Provider execution belongs to the `inference-` 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 diff --git a/litellm-rust/crates/gateway/Cargo.toml b/litellm-rust/crates/gateway/Cargo.toml index 387bcd43231..b2a2f09bfc7 100644 --- a/litellm-rust/crates/gateway/Cargo.toml +++ b/litellm-rust/crates/gateway/Cargo.toml @@ -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 diff --git a/litellm-rust/crates/gateway/src/error.rs b/litellm-rust/crates/gateway/src/error.rs index 75dc04d179d..170a8284c18 100644 --- a/litellm-rust/crates/gateway/src/error.rs +++ b/litellm-rust/crates/gateway/src/error.rs @@ -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), } diff --git a/litellm-rust/crates/gateway/src/lib.rs b/litellm-rust/crates/gateway/src/lib.rs index 0b3aba81181..2ed2fc80a51 100644 --- a/litellm-rust/crates/gateway/src/lib.rs +++ b/litellm-rust/crates/gateway/src/lib.rs @@ -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, Error> { )?)) } +pub async fn build_redis(config: &Config) -> Result { + 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, @@ -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 { + 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); diff --git a/litellm-rust/crates/gateway/src/main.rs b/litellm-rust/crates/gateway/src/main.rs index f8150e6e860..47c1fe83f50 100644 --- a/litellm-rust/crates/gateway/src/main.rs +++ b/litellm-rust/crates/gateway/src/main.rs @@ -46,6 +46,7 @@ async fn main() -> Result<(), Box> { 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())?; diff --git a/litellm-rust/crates/gateway/src/redis.rs b/litellm-rust/crates/gateway/src/redis.rs new file mode 100644 index 00000000000..197f40e7988 --- /dev/null +++ b/litellm-rust/crates/gateway/src/redis.rs @@ -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, + pub transaction_buffer: Option, + pub response_cache: Option, +} + +#[derive(Debug, PartialEq)] +pub(crate) struct Plan { + coordination: Option, + transaction_buffer_fallback: Option, + response_cache: Option, +} + +#[derive(Debug, PartialEq)] +enum Target { + Configured(Params), + Environment(Params), +} + +pub(crate) fn plan(config: &Config, lookup: Lookup<'_>) -> Result { + 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 { + 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, 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 { + 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 { + 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) -> Result { + 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) { + 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, 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 { + lookup(name) + .map(|value| { + value + .parse::() + .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 { + let environment: HashMap = 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) -> 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::("rate-limit").await.unwrap().as_deref(), + Some("1") + ); + assert_eq!( + cache.get::("spend").await.unwrap().as_deref(), + Some("2") + ); + assert_eq!( + unnamespaced + .get::("responses:rate-limit") + .await + .unwrap() + .as_deref(), + Some("1") + ); + assert_eq!( + unnamespaced + .get::("responses:spend") + .await + .unwrap() + .as_deref(), + Some("2") + ); + assert!( + unnamespaced + .get::("rate-limit") + .await + .unwrap() + .is_none() + ); + assert!(unnamespaced.get::("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()); + } +} diff --git a/litellm-rust/crates/gateway/tests/server.rs b/litellm-rust/crates/gateway/tests/server.rs index 71edd3c3650..53a97e7098c 100644 --- a/litellm-rust/crates/gateway/tests/server.rs +++ b/litellm-rust/crates/gateway/tests/server.rs @@ -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, + #[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::(&body).unwrap(), + json!({"status": "healthy", "db": "Not connected"}) + ); +} diff --git a/litellm-rust/crates/redis-client/src/client.rs b/litellm-rust/crates/redis-client/src/client.rs index 28fd16ecd23..a02b0847663 100644 --- a/litellm-rust/crates/redis-client/src/client.rs +++ b/litellm-rust/crates/redis-client/src/client.rs @@ -26,15 +26,15 @@ pub enum Ttl { } impl RedisClient { - pub async fn connect(config: RedisConfig) -> Result { + pub fn new(config: RedisConfig) -> Result { 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(&self, key: &str) -> Result, 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)) diff --git a/litellm-rust/crates/redis-client/src/commands.rs b/litellm-rust/crates/redis-client/src/commands.rs index 34ec7432c98..fb8a4a28c94 100644 --- a/litellm-rust/crates/redis-client/src/commands.rs +++ b/litellm-rust/crates/redis-client/src/commands.rs @@ -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() } diff --git a/litellm-rust/crates/redis-client/src/config.rs b/litellm-rust/crates/redis-client/src/config.rs index 81dd532d0cf..64934c68891 100644 --- a/litellm-rust/crates/redis-client/src/config.rs +++ b/litellm-rust/crates/redis-client/src/config.rs @@ -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, } +#[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, pub username: Option, @@ -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(|_| "")) + .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(|_| "")) + .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}://@{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://@cache:6379/0")] + #[case::url_with_at_in_password(Endpoint::Url("rediss://:p@ss@cache:6380".to_owned()), "rediss://@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, + }), + "" + )] + 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}"); + } +} diff --git a/litellm-rust/crates/redis-client/src/connection.rs b/litellm-rust/crates/redis-client/src/connection.rs index f70449bd43e..aaf26f105da 100644 --- a/litellm-rust/crates/redis-client/src/connection.rs +++ b/litellm-rust/crates/redis-client/src/connection.rs @@ -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 { + pub(crate) fn standalone(info: ConnectionInfo, timeout: Duration) -> RedisResult { 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 { + pub(crate) fn cluster(endpoint: &ClusterEndpoint, timeout: Duration) -> RedisResult { 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), )) } } diff --git a/litellm-rust/crates/redis-client/src/namespace.rs b/litellm-rust/crates/redis-client/src/namespace.rs index b64293db421..4db02b5b7bb 100644 --- a/litellm-rust/crates/redis-client/src/namespace.rs +++ b/litellm-rust/crates/redis-client/src/namespace.rs @@ -1,4 +1,4 @@ -#[derive(Clone, Default)] +#[derive(Clone, Debug, Default, PartialEq)] pub struct Namespace(Option); 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) => { diff --git a/litellm-rust/crates/redis-client/src/test_support.rs b/litellm-rust/crates/redis-client/src/test_support.rs index af1d1c46b9e..64a174b1543 100644 --- a/litellm-rust/crates/redis-client/src/test_support.rs +++ b/litellm-rust/crates/redis-client/src/test_support.rs @@ -56,13 +56,12 @@ impl Server { timeout: Duration, breaker: Option, ) -> 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(), diff --git a/litellm-rust/crates/redis-testing/src/cluster.rs b/litellm-rust/crates/redis-testing/src/cluster.rs index 324f3cf1d6b..e61f7832b1c 100644 --- a/litellm-rust/crates/redis-testing/src/cluster.rs +++ b/litellm-rust/crates/redis-testing/src/cluster.rs @@ -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> { diff --git a/litellm-rust/crates/redis-testing/src/standalone.rs b/litellm-rust/crates/redis-testing/src/standalone.rs index 97d2414d9d9..8060ccbf190 100644 --- a/litellm-rust/crates/redis-testing/src/standalone.rs +++ b/litellm-rust/crates/redis-testing/src/standalone.rs @@ -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> {