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

refactor(config): resolve Redis through surface adapters and typed options
Some checks are pending
LiteLLM Rust DB / db-queries (push) Waiting to run
Details
LiteLLM Rust / rust-lint (push) Waiting to run
Details
LiteLLM Rust / rust-test (push) Waiting to run
Details
LiteLLM Rust / rust-wheel (push) Waiting to run
Details

Browse source 
Each config surface now turns its Redis section into a ConfiguredRedis
(coordination, cache_params, or a RedisCache.redis_kwargs dict), and one
resolve returns the ResolvedRedis plus the ignored settings by source, so
hosts log gaps instead of the resolver. The resolve/resolve_cache flag and
the coordination ssl normalization are gone

Resolution reads the layered kwargs into a typed Options struct once, and
endpoint and auth selection work from that struct alone. An empty value is
unset everywhere, and a flag Python surfaces read differently is an error

ResolvedRedis and RedisAuth move to redis-types, so redis-connect no longer
depends on litellm-config

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Yujong Lee 2026-11-04 11:56:55 +09:00
parent de1acf639e
commit 94fa31467d
23 changed files with 875 additions and 731 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm-rust/Cargo.lock generated
View file

@ -3717,7 +3717,6 @@ dependencies = [
"tempfile",
"thiserror 2.0.19",
"tracing",
"tracing-subscriber",
"url",
"veil",
]
@ -3834,6 +3833,7 @@ dependencies = [
"litellm-redis-client",
"litellm-redis-connect",
"litellm-redis-testing",
"litellm-redis-types",
"litellm-secret-value",
"litellm-secrets",
"litellm-tracing",
@ -4443,6 +4443,7 @@ dependencies = [
"litellm-config",
"litellm-redis-client",
"litellm-redis-testing",
"litellm-redis-types",
"rstest",
"rustls 0.23.42",
"serde_json",

13
litellm-rust/crates/config/AGENTS.md
View file

@ -1,16 +1,19 @@
## Redis settings
- `RedisSettings::resolve` follows `litellm/_redis.py` `_get_redis_client_logic` plus `_credential_provider_auth_kwargs` and the per-client allowlists of `get_redis_async_client` (redis-py 5.3.1). Python and Rust pods share one Redis during the switch, so a change there is a change to a wire contract
- The boundary is surface in, `ResolvedRedis` out. A surface turns what it was given into a `ConfiguredRedis`, the redis-py kwargs `_get_redis_client_logic` would receive: `RedisSettings::coordination` for `general_settings.coordination_redis`, `RedisSettings::cache_params` for `litellm_settings.cache_params`, and `ConfiguredRedis::redis_cache_kwargs` for a `RedisCache.redis_kwargs` dict. `ConfiguredRedis::resolve` is then the same for every host and returns a `Resolution`: the `litellm_redis_types::ResolvedRedis` plus every accepted-but-ignored setting by source, which the host logs. A new surface is a new constructor, never a flag on `resolve`
- Resolution runs in three steps, one file each. `layers.rs` puts the configured kwargs over their `REDIS_*` values. `options.rs` reads the result into the typed `Options`, rejects unsupported settings, collects ignored ones and fills Python's `get_secret` fallbacks. `endpoint.rs` and `auth.rs` pick the target and the credentials from `Options` alone
- `resolve` follows `litellm/_redis.py` `_get_redis_client_logic` plus `_credential_provider_auth_kwargs` and the per-client allowlists of `get_redis_async_client` (redis-py 5.3.1). Python and Rust pods share one Redis during the switch, so a change there is a change to a wire contract
- That contract is `resolves_like_python_redis` in `tests/redis.rs` and its mirror `test_redis_settings_resolve_like_the_rust_resolver_expects` in `tests/unit/test_redis.py`. Both hold the same named cases with the expected view written next to the input, the Python side reading the effective redis-py 5.3.1 connection objects without connecting. A new resolution rule is a new case in both files, never a generated fixture. A case where the two sides expect different outcomes is a divergence and is listed below
- `resolve` takes a `SecretSnapshot` and an environment `Lookup`. The snapshot holds every name `RedisSettings::secret_names` lists, read the way Python's `get_secret` reads it (secret manager, then environment). The host fills it once before resolving, python-bridge through `get_secret` and rust-proxy from its environment, so the resolver never calls back into the host. The environment lookup is only for what Python reads straight from `os.environ` (`REDIS_USERNAME` for Azure). Reading a name the snapshot does not list fails a debug assertion
- Every Redis setting Python reads is one row in `SETTINGS` in `src/redis/keys.rs`: how Rust handles it (read, ignore, warn, reject), whether `_get_redis_env_kwarg_mapping` reads it as `REDIS_<KWARG>`, and the names `_get_redis_client_logic` falls back to. The env layer, the rejections, the warnings and `secret_names` all come from that table, so a new setting is one row. It is the shape the `#[derive(Layered)]` attributes in `litellm-layered` would generate, and moves onto them once that crate lands
- `resolve` takes a `SecretSnapshot` and an environment `Lookup`. The snapshot holds every name `ConfiguredRedis::secret_names` lists, read the way Python's `get_secret` reads it (secret manager, then environment). The host fills it once before resolving, python-bridge through `get_secret` and rust-proxy from its environment, so the resolver never calls back into the host. The environment lookup is only for what Python reads straight from `os.environ` (`REDIS_USERNAME` for Azure). Reading a name the snapshot does not list fails a debug assertion
- Every Redis setting Python reads is one row in `SETTINGS` in `src/redis/keys.rs`: how Rust handles it (read, ignore, warn, reject), whether `_get_redis_env_kwarg_mapping` reads it as `REDIS_<KWARG>`, and the names `_get_redis_client_logic` falls back to. The env layer, the rejections, the ignored list and `secret_names` all come from that table. A read row is also a field of `Options`
- Lookups go through `litellm_core_utils::settings::Lookup`, never a closure type of our own
- A url carries its own TLS. redis-py's url allowlist is the plain `Connection` signature, so every `ssl_*` setting outside the url is dropped, and a credential provider strips the url's userinfo. A `rediss://` url's `ssl_cert_reqs`, `ssl_ca_certs`, `ssl_certfile`, `ssl_keyfile` and `ssl_check_hostname` query parameters are read like `parse_url` reads them
- TLS options and `db` are only read for the endpoint that uses them, so a bad `ssl_cert_reqs` never fails a url target. A cluster with a non-zero `db`, and Sentinel with a `host` or `port`, are errors because redis-py raises building those clients
- A falsy configured value (`""`, `[]`, `0`, `false`) falls through to the environment fallback like Python's `or` chain. `coordination_redis` drops every null like `model_dump(exclude_none=True)`, while a null in `cache_params` overrides the `REDIS_*` environment value unless `Cache` or `RedisCache` names that parameter. A missing `os.environ/` reference is a null, as `load_config` resolves it
- An empty value (`""`, `[]`, `{}`) is the same as an unset one, so it falls through to its environment fallback the way Python's `x or get_secret(...)` does. Nulls depend on the surface. `coordination_redis` drops every null like `model_dump(exclude_none=True)`, `cache_params` keeps a null as an explicit `None` that hides the `REDIS_*` value unless `Cache` or `RedisCache` names that parameter, and `RedisCache.redis_kwargs` keeps every null. A missing `os.environ/` reference is a null, as `load_config` resolves it
- Precedence matches Python. The target is cluster, then URL, then Sentinel (`sentinel_nodes` plus `service_name`), then host. Auth is an explicit `credential_provider`, then GCP, then Azure, then AWS, and the first one wins with a warning
- Numbers are read with `litellm_python_compat::number`, so whitespace and digit underscores parse like Python's `int()` and `float()`. A value they cannot parse (`port: abc`, `socket_timeout: slow`) is dropped with a warning and the default is used, like `_coerce_redis_kwargs_types`. A number that parses but is out of range (negative seconds, a port above 65535) fails with `RedisError::Invalid`
- `validate_coordination` mirrors the proxy's `CoordinationRedisParams` pydantic model, which rejects a malformed `port` or `ssl` in `general_settings.coordination_redis` before redis-py coerces anything. Hosts call it for that block only
- Flags (`ssl`, `ssl_check_hostname`, `aws_iam_auth`, `aws_iam_serverless`) are read one way on every surface. `true`, `1` and `yes` are true and `false`, `0`, `no`, `off`, `f` and `n` are false, the text redis-py's `_str_to_bool` and the proxy's pydantic models agree on. Any other text is `RedisError::Invalid`, so a TLS setting is never guessed
- `RedisSettings::coordination` mirrors the proxy's `CoordinationRedisParams` pydantic model, which rejects a malformed `port` before redis-py coerces anything
- Known divergences, kept on purpose. Each is a case in both contract tests with a different expected outcome on each side
- A url query parameter other than the TLS ones, or any query parameter on a plain `redis://` url, is `RedisError::Unsupported` naming `url ?<name>`. redis-py turns each one into a client kwarg
- Settings that would change the server, identity or TLS trust if ignored are `RedisError::Unsupported` naming the setting, including `ssl_ciphers`, `ssl_min_version`, `ssl_password`, `ssl_ca_data`, OCSP options, `unix_socket_path`, `connection_pool`, `event_dispatcher`, `redis_connect_func`, `retry`

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

@ -24,4 +24,3 @@ veil.workspace = true
litellm-error.workspace = true
rstest.workspace = true
tempfile.workspace = true
tracing-subscriber = { version = "0.3", default-features = false, features = ["registry", "std"] }

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

@ -15,8 +15,8 @@ pub use litellm_router_types::{LitellmParams, Spelled};
pub use mcp::{McpAuth, McpServer, McpTransport};
pub use model::Model;
pub use redis::{
AwsIamAuth, AzureAdAuth, CacheSettings, Expected, RedisAuth, RedisError, RedisSettings,
ResolvedRedis, SecretSnapshot, environment_has_target,
CacheSettings, ConfiguredRedis, Expected, RedisError, RedisSettings, Resolution,
SecretSnapshot, environment_has_target,
};
pub use settings::{
ClickHouseStoreSettings, GeneralSettings, LiteLlmSettings, RouterSettings, TracingSettings,

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

@ -1,29 +1,26 @@
use std::{collections::BTreeMap, time::Duration};
use std::collections::BTreeMap;
use litellm_core_utils::settings::Lookup;
use litellm_python_compat::{BigInt, number};
use litellm_redis_types::{
ClusterEndpoint, ClusterNode as RedisNode, Endpoint, HostEndpoint, SentinelEndpoint,
};
use litellm_secret_value::SecretValue;
use litellm_redis_types::{Endpoint, RedisAuth, ResolvedRedis};
use serde::{Deserialize, Deserializer, de::Error as _};
use veil::Redact;
use crate::{Object, Value};
mod auth;
mod endpoint;
mod error;
mod keys;
mod layers;
mod options;
mod secrets;
mod sentinel;
mod tls;
use keys::Handling;
pub use auth::{AwsIamAuth, AzureAdAuth, RedisAuth};
pub use error::{Expected, RedisError};
pub use secrets::SecretSnapshot;
use options::Options;
const CACHE_ONLY: &[&str] = &[
"type",
"mode",
@ -36,6 +33,8 @@ const CACHE_ONLY: &[&str] = &[
"_backend",
];
/// A Redis section as `config.yaml` spells it. Each surface reads it its own way, so it resolves
/// only through [`RedisSettings::coordination`] or [`RedisSettings::cache_params`].
#[derive(Clone, Default, Deserialize, Redact)]
#[serde(transparent)]
pub struct RedisSettings {
@ -66,140 +65,102 @@ impl<'de> Deserialize<'de> for CacheSettings {
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct ResolvedRedis {
pub endpoint: Endpoint,
pub auth: RedisAuth,
pub namespace: Option<String>,
pub timeout: Option<Duration>,
pub connect_timeout: Option<Duration>,
/// The redis-py kwargs one surface hands `_get_redis_client_logic`, before `REDIS_*` and secrets.
#[derive(Clone, Redact)]
pub struct ConfiguredRedis {
#[redact(with = "[REDACTED]")]
fields: BTreeMap<String, Value>,
nulls: Nulls,
}
#[derive(Clone)]
struct Entry {
value: Value,
source: String,
/// Whether a null setting reaches redis-py as an explicit `None`, which outranks its `REDIS_*` value.
#[derive(Clone, Copy, Debug)]
enum Nulls {
/// `coordination_redis` reaches redis-py through `model_dump(exclude_none=True)`.
Dropped,
/// `Cache` drops a null only for the parameters it names and passes any other key on.
KeptUnlessNamedByCache,
/// `RedisCache.redis_kwargs` already holds only the nulls redis-py receives.
Kept,
}
type Kwargs = BTreeMap<String, Entry>;
/// A resolved connection, plus the settings it was given that Rust accepts without honoring.
#[derive(Debug)]
pub struct Resolution {
pub redis: ResolvedRedis,
/// Where each ignored setting came from: its key, or the environment variable that set it.
pub ignored: Vec<String>,
}
impl RedisSettings {
/// `general_settings.coordination_redis`, which the proxy validates as a pydantic model first.
/// An `os.environ/` port is only known after resolution, where it is read like any other port.
pub fn coordination(&self) -> Result<ConfiguredRedis, RedisError> {
if let Some(port) = self.fields.get("port").and_then(options::as_text)
&& !port.starts_with("os.environ/")
&& port.parse::<u16>().is_err()
{
return Err(RedisError::Invalid {
setting: "port".to_owned(),
expected: Expected::Integer,
});
}
Ok(ConfiguredRedis {
fields: self.fields.clone(),
nulls: Nulls::Dropped,
})
}
/// `litellm_settings.cache_params` with `type: redis`, as `litellm.Cache` hands it to `RedisCache`.
pub fn cache_params(&self) -> ConfiguredRedis {
let alias = self
.fields
.get("redis_startup_nodes")
.filter(|_| !self.fields.contains_key("startup_nodes"))
.map(|nodes| ("startup_nodes".to_owned(), nodes.clone()));
let fields = self
.fields
.iter()
.filter(|(key, _)| {
key.as_str() != "redis_startup_nodes" && !CACHE_ONLY.contains(&key.as_str())
})
.map(|(key, value)| (key.clone(), value.clone()))
.chain(alias)
.collect();
ConfiguredRedis {
fields,
nulls: Nulls::KeptUnlessNamedByCache,
}
}
}
impl ConfiguredRedis {
/// `RedisCache.redis_kwargs`, the dict every Python surface ends in and the one python-bridge
/// hands over.
pub fn redis_cache_kwargs(fields: BTreeMap<String, Value>) -> Self {
Self {
fields,
nulls: Nulls::Kept,
}
}
/// No configured settings, so only `REDIS_*` and the other environment fallbacks apply.
pub fn environment_only() -> Self {
Self {
fields: BTreeMap::new(),
nulls: Nulls::Dropped,
}
}
pub fn resolve(
&self,
secrets: &SecretSnapshot,
environment: &dyn Lookup,
) -> Result<ResolvedRedis, RedisError> {
self.resolve_with(secrets, environment, false)
}
pub fn resolve_cache(
&self,
secrets: &SecretSnapshot,
environment: &dyn Lookup,
) -> Result<ResolvedRedis, RedisError> {
self.resolve_with(secrets, environment, true)
}
fn resolve_with(
&self,
secrets: &SecretSnapshot,
environment: &dyn Lookup,
cache: bool,
) -> Result<ResolvedRedis, RedisError> {
let lookup: &dyn Lookup = secrets;
let block = self.kwargs(cache);
let block: Kwargs = block
.into_iter()
.filter_map(|(key, entry)| {
let value = environment_reference(&entry.value, lookup);
let value = match (&value, cache, key.as_str()) {
(Value::String(text), false, "ssl") => {
coordination_bool(text).map(Value::Bool).unwrap_or(value)
}
_ => value,
};
(!value.is_null() || keeps_explicit_none(cache, &key))
.then(|| (key, Entry { value, ..entry }))
})
.collect();
let environment_kwargs = environment_kwargs(lookup);
let names_a_target = ["host", "startup_nodes", "sentinel_nodes"]
.iter()
.any(|key| is_set(&block, key));
let drop_environment_url = names_a_target && !is_set(&block, "url");
let merged: Kwargs = environment_kwargs
.into_iter()
.filter(|(key, _)| !(key == "url" && drop_environment_url))
.chain(block)
.collect();
reject_unsupported(&merged)?;
warn_ignored(&merged);
let startup_nodes = match configured(&merged, "startup_nodes") {
Some(entry) => Some(
cluster_nodes(&entry.value)
.ok_or_else(|| invalid(&entry.source, Expected::StartupNodes))?,
),
None => keys::fallback("startup_nodes", lookup)
.map(|(source, nodes)| {
cluster_nodes(&Value::String(nodes))
.ok_or_else(|| invalid(source, Expected::StartupNodes))
})
.transpose()?,
}
.filter(|nodes| !nodes.is_empty());
let url = text_value(&merged, "url");
let ssl = bool_value(&merged, "ssl");
let uses_tls = match (&startup_nodes, &url) {
(Some(_), _) | (None, None) => ssl,
(None, Some(url)) => url
.split_once("://")
.is_some_and(|(scheme, _)| scheme.eq_ignore_ascii_case("rediss")),
};
let auth = auth::redis_auth(&merged, lookup, environment, uses_tls)?;
let gcp_ca_certs = match &auth {
RedisAuth::GcpIam { .. } if ssl => {
text_or_fallback(&merged, "gcp_ssl_ca_certs", lookup)
.filter(|path| !path.is_empty())
}
_ => None,
};
let tls = || {
ssl.then(|| tls::tls_options(&merged, gcp_ca_certs.clone()))
.transpose()
};
let db = || integer(&merged, "db").map(|db| db.unwrap_or(0));
let endpoint = match (startup_nodes, url) {
(Some(nodes), _) => {
reject_cluster_db(&merged)?;
Endpoint::Cluster(ClusterEndpoint { nodes, tls: tls()? })
}
(None, Some(url)) => Endpoint::Url {
tls: tls::url_tls(&url)?,
url: SecretValue::new(match &auth {
RedisAuth::Static { .. } => url,
_ => without_userinfo(&url),
}),
},
(None, None) => match sentinel::sentinel_target(&merged, lookup)? {
Some(target) => {
reject_sentinel_host(&merged)?;
Endpoint::Sentinel(SentinelEndpoint {
sentinels: target.sentinels,
service_name: target.service_name,
sentinel_password: target.sentinel_password,
tls: tls()?,
db: db()?,
})
}
None => Endpoint::Host(HostEndpoint {
host: text_value(&merged, "host").ok_or(RedisError::NoTarget)?,
port: integer(&merged, "port")?.unwrap_or(6379),
tls: tls()?,
db: db()?,
}),
},
};
) -> Result<Resolution, RedisError> {
let kwargs = layers::merged(self, secrets);
let (options, ignored) = Options::read(kwargs, secrets)?;
let auth = auth::redis_auth(&options, environment)?;
let endpoint = endpoint::endpoint(&options, &auth)?;
let auth = match (&endpoint, auth) {
(Endpoint::Url { .. }, RedisAuth::Static { username, .. }) => RedisAuth::Static {
username,
@ -207,65 +168,17 @@ impl RedisSettings {
},
(_, auth) => auth,
};
Ok(ResolvedRedis {
endpoint,
auth,
namespace: text_value(&merged, "namespace"),
timeout: seconds(&merged, "socket_timeout")?,
connect_timeout: seconds(&merged, "socket_connect_timeout")?,
Ok(Resolution {
redis: ResolvedRedis {
endpoint,
auth,
namespace: options::text(options.namespace.as_ref()),
timeout: options::seconds(options.socket_timeout.as_ref())?,
connect_timeout: options::seconds(options.socket_connect_timeout.as_ref())?,
},
ignored,
})
}
pub fn validate_coordination(&self) -> Result<(), RedisError> {
let text = |key: &str| self.fields.get(key).and_then(as_text);
if text("port").is_some_and(|port| port.parse::<u16>().is_err()) {
return Err(invalid("port", Expected::Integer));
}
if text("ssl").is_some_and(|ssl| coordination_bool(&ssl).is_none()) {
return Err(invalid("ssl", Expected::Boolean));
}
Ok(())
}
fn kwargs(&self, cache: bool) -> Kwargs {
let entry = |key: &str, value: &Value| {
(
key.to_owned(),
Entry {
value: value.clone(),
source: key.to_owned(),
},
)
};
let aliased = self
.fields
.get("redis_startup_nodes")
.filter(|_| cache)
.map(|value| entry("startup_nodes", value));
let direct = self
.fields
.iter()
.filter(|(key, _)| {
!(cache
&& (key.as_str() == "redis_startup_nodes"
|| CACHE_ONLY.contains(&key.as_str())))
})
.map(|(key, value)| entry(key, value));
aliased.into_iter().chain(direct).collect()
}
}
// _get_redis_url_kwargs keeps only plain Connection kwargs, so redis-py drops every ssl_* setting for a url
// and a credential provider replaces the url's userinfo, which _url_without_userinfo strips
fn without_userinfo(url: &str) -> String {
let Some((scheme, rest)) = url.split_once("://") else {
return url.to_owned();
};
let (authority, tail) = rest.split_at(rest.find(['/', '?', '#']).unwrap_or(rest.len()));
let host = authority
.rsplit_once('@')
.map_or(authority, |(_, host)| host);
format!("{scheme}://{host}{tail}")
}
pub fn environment_has_target(lookup: &dyn Lookup) -> bool {
@ -279,293 +192,27 @@ pub fn environment_has_target(lookup: &dyn Lookup) -> bool {
.any(|name| lookup.get(name).is_some())
}
fn reject_unsupported(merged: &Kwargs) -> Result<(), RedisError> {
match merged.iter().find(|(key, _)| is_unsupported(key)) {
Some((_, entry)) => Err(RedisError::Unsupported {
setting: entry.source.clone(),
}),
None => Ok(()),
}
}
fn invalid(setting: &str, expected: Expected) -> RedisError {
RedisError::Invalid {
setting: setting.to_owned(),
expected,
}
}
fn parsed<T>(
kwargs: &Kwargs,
key: &str,
expected: Expected,
parse: impl Fn(&str) -> Option<T>,
) -> Result<Option<T>, RedisError> {
kwargs
.get(key)
.and_then(|entry| Some((as_text(&entry.value)?, &entry.source)))
.map(|(text, source)| parse(&text).ok_or_else(|| invalid(source, expected)))
.transpose()
}
fn seconds(merged: &Kwargs, key: &str) -> Result<Option<Duration>, RedisError> {
coerced(merged, key, Expected::Seconds, number::float, |seconds| {
Duration::try_from_secs_f64(seconds).ok()
})
}
fn coerced<N, T>(
kwargs: &Kwargs,
key: &str,
expected: Expected,
parse: impl Fn(&str) -> Option<N>,
convert: impl Fn(N) -> Option<T>,
) -> Result<Option<T>, RedisError> {
let Some((entry, text)) = kwargs
.get(key)
.and_then(|entry| Some((entry, as_text(&entry.value)?)))
else {
return Ok(None);
};
match parse(&text) {
Some(number) => convert(number)
.map(Some)
.ok_or_else(|| invalid(&entry.source, expected)),
None => {
tracing::warn!(setting = %entry.source, "Redis setting is not a number, so the default is used like redis-py");
Ok(None)
}
}
}
fn warn_ignored(merged: &Kwargs) {
for key in merged.keys() {
if keys::setting(key).is_some_and(|setting| setting.handling == Handling::Warn) {
tracing::warn!(setting = %key, "Redis setting is accepted but ignored");
}
}
}
fn is_unsupported(key: &str) -> bool {
keys::setting(key).is_some_and(|setting| setting.handling == Handling::Reject)
}
fn environment_kwargs(lookup: &dyn Lookup) -> Kwargs {
keys::SETTINGS
.iter()
.filter_map(|setting| {
let source = setting.environment_name()?;
let value = lookup.get(&source)?;
Some((
setting.kwarg.to_owned(),
Entry {
value: Value::String(value),
source,
},
))
})
.collect()
}
// load_config resolves os.environ/ to None when unset, so a missing reference is a null like any other
fn environment_reference(value: &Value, lookup: &dyn Lookup) -> Value {
match value {
Value::String(text) => match text.strip_prefix("os.environ/") {
Some(name) => lookup.get(name).map_or(Value::Null, Value::String),
None => value.clone(),
},
_ => value.clone(),
}
}
// coordination_redis reaches redis-py through model_dump(exclude_none=True), so every null is dropped. Cache
// and RedisCache drop a null only for the parameters they name, and pass any other key on as an explicit None
// that overrides the REDIS_* environment value
fn keeps_explicit_none(cache: bool, key: &str) -> bool {
const NAMED_BY_CACHE: &[&str] = &[
"host",
"port",
"password",
"socket_timeout",
"startup_nodes",
"namespace",
"gcp_service_account",
"gcp_ssl_ca_certs",
];
cache && !NAMED_BY_CACHE.contains(&key)
}
// redis-py's RedisCluster raises unless db is 0 or None
fn reject_cluster_db(merged: &Kwargs) -> Result<(), RedisError> {
let Some(entry) = merged.get("db").filter(|entry| !entry.value.is_null()) else {
return Ok(());
};
match &entry.value {
Value::String(_) if matches!(integer::<i64>(merged, "db"), Ok(None | Some(0))) => Ok(()),
value if !truthy(value) => Ok(()),
_ => Err(RedisError::DbInCluster {
setting: entry.source.clone(),
}),
}
}
// redis.asyncio.Sentinel passes the connection kwargs to each monitor's Redis(host=..., port=...), so a host or
// port setting raises TypeError when litellm builds the client
fn reject_sentinel_host(merged: &Kwargs) -> Result<(), RedisError> {
match ["host", "port"]
.iter()
.find_map(|key| merged.get(*key).filter(|entry| !entry.value.is_null()))
{
Some(entry) => Err(RedisError::SentinelWithHost {
setting: entry.source.clone(),
}),
None => Ok(()),
}
}
fn is_set(kwargs: &Kwargs, key: &str) -> bool {
kwargs.get(key).is_some_and(|entry| !entry.value.is_null())
}
// Python chains these settings with `or`, so a falsy configured value ("", [], 0, False) falls through too
fn configured<'a>(kwargs: &'a Kwargs, key: &str) -> Option<&'a Entry> {
kwargs.get(key).filter(|entry| truthy(&entry.value))
}
fn truthy(value: &Value) -> bool {
match value {
Value::Null => false,
Value::Bool(flag) => *flag,
Value::Number(number) => number.as_f64().is_some_and(|number| number != 0.0),
Value::String(text) => !text.is_empty(),
Value::Sequence(items) => !items.is_empty(),
Value::Mapping(entries) => !entries.is_empty(),
Value::Tagged(_) => true,
}
}
fn text_or_fallback(kwargs: &Kwargs, key: &str, lookup: &dyn Lookup) -> Option<String> {
configured(kwargs, key)
.and_then(|entry| as_text(&entry.value))
.or_else(|| keys::fallback_value(key, lookup))
}
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: TryFrom<BigInt>>(kwargs: &Kwargs, key: &str) -> Result<Option<T>, RedisError> {
coerced(kwargs, key, Expected::Integer, number::int, |number| {
T::try_from(number).ok()
})
}
fn cluster_nodes(value: &Value) -> Option<Vec<RedisNode>> {
nodes(value, |node| {
node_address(
node.get("host").and_then(as_text),
node.get("port").and_then(as_text),
)
})
}
fn nodes(value: &Value, parse: impl Fn(&Value) -> Option<RedisNode>) -> Option<Vec<RedisNode>> {
match value {
Value::String(json) => serde_json::from_str::<Vec<Value>>(json)
.ok()?
.iter()
.map(parse)
.collect(),
Value::Sequence(nodes) => nodes.iter().map(parse).collect(),
_ => None,
}
}
fn node_address(host: Option<String>, port: Option<String>) -> Option<RedisNode> {
Some(RedisNode {
host: host?,
port: port?.parse().ok()?,
})
}
fn coordination_bool(text: &str) -> Option<bool> {
match text.to_ascii_lowercase().as_str() {
"1" | "on" | "t" | "true" | "y" | "yes" => Some(true),
"0" | "off" | "f" | "false" | "n" | "no" => Some(false),
_ => None,
}
}
#[cfg(test)]
mod tests {
use super::{RedisSettings, SecretSnapshot};
use litellm_redis_types::Endpoint;
use super::RedisSettings;
use rstest::rstest;
#[rstest]
#[case::on("on", true)]
#[case::t("t", true)]
#[case::y("y", true)]
#[case::upper_on("ON", true)]
#[case::off("off", false)]
#[case::f("f", false)]
#[case::n("n", false)]
fn coordination_ssl_aliases_are_normalized_without_changing_raw_cache_kwargs(
#[case] value: &str,
#[case] tls: bool,
#[values(false, true)] cache: bool,
) {
let settings: RedisSettings = serde_json::from_value(serde_json::json!({
"host": "redis", "ssl": value,
}))
.unwrap();
settings.validate_coordination().unwrap();
let environment = |_: &str| None;
let secrets = SecretSnapshot::read(settings.secret_names(), &environment);
let resolved = if cache {
settings.resolve_cache(&secrets, &environment)
} else {
settings.resolve(&secrets, &environment)
}
.unwrap();
let Endpoint::Host(host) = resolved.endpoint else {
panic!("expected host endpoint");
};
assert_eq!(host.tls.is_some(), tls && !cache);
}
#[rstest]
#[case::scalar(serde_json::json!("credential-value"))]
#[case::object(serde_json::json!({"secret": "credential-value"}))]
#[case::array(serde_json::json!(["credential-value"]))]
fn credentials_are_redacted(#[case] password: serde_json::Value) {
fn credentials_are_redacted(
#[case] password: serde_json::Value,
#[values(false, true)] configured: bool,
) {
let settings: RedisSettings = serde_json::from_value(serde_json::json!({
"host": "redis", "password": password,
}))
.unwrap();
let debug = format!("{settings:?}");
let debug = match configured {
false => format!("{settings:?}"),
true => format!("{:?}", settings.cache_params()),
};
assert!(!debug.contains("credential-value"));
assert!(debug.contains("[REDACTED]"));
}

89
litellm-rust/crates/config/src/redis/auth.rs
View file

@ -1,52 +1,22 @@
use litellm_secret_value::SecretValue;
use litellm_core_utils::settings::Lookup;
use litellm_redis_types::{AwsIamAuth, AzureAdAuth, RedisAuth};
use super::{
Kwargs, Lookup, RedisError, bool_value, is_set, secret_value, text_or_fallback, text_value,
RedisError,
options::{self, AwsOptions, Options, flag, secret, text},
};
#[derive(Clone, Debug, PartialEq)]
pub enum RedisAuth {
Static {
username: Option<SecretValue>,
password: Option<SecretValue>,
},
Host,
GcpIam {
service_account: String,
},
AzureAd(AzureAdAuth),
AwsIam(AwsIamAuth),
}
#[derive(Clone, Debug, PartialEq)]
pub struct AzureAdAuth {
pub client_id: Option<String>,
pub tenant_id: Option<String>,
pub client_secret: Option<SecretValue>,
pub username: Option<String>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct AwsIamAuth {
pub user_name: String,
pub cache_name: String,
pub region: String,
pub serverless: bool,
}
pub(super) fn redis_auth(
merged: &Kwargs,
lookup: &dyn Lookup,
options: &Options,
environment: &dyn Lookup,
uses_tls: bool,
) -> Result<RedisAuth, RedisError> {
if is_set(merged, "credential_provider") {
if options.credential_provider.is_some() {
return Ok(RedisAuth::Host);
}
let gcp = text_or_fallback(merged, "gcp_service_account", lookup);
let azure = text_or_fallback(merged, "azure_redis_ad_token", lookup)
let gcp = text(options.gcp.service_account.as_ref());
let azure = text(options.azure.ad_token.as_ref())
.is_some_and(|enabled| enabled.eq_ignore_ascii_case("true"));
let aws = bool_value(merged, "aws_iam_auth");
let aws = flag(options.aws.auth.as_ref())?;
match (gcp, azure, aws) {
(Some(service_account), azure, aws) => {
if azure {
@ -68,37 +38,46 @@ pub(super) fn redis_auth(
);
}
Ok(RedisAuth::AzureAd(AzureAdAuth {
client_id: text_or_fallback(merged, "azure_client_id", lookup),
tenant_id: text_or_fallback(merged, "azure_tenant_id", lookup),
client_secret: text_or_fallback(merged, "azure_client_secret", lookup)
.map(SecretValue::new),
client_id: text(options.azure.client_id.as_ref()),
tenant_id: text(options.azure.tenant_id.as_ref()),
client_secret: secret(options.azure.client_secret.as_ref()),
username: environment
.get("REDIS_USERNAME")
.filter(|username| !username.is_empty()),
}))
}
(None, false, true) => aws_iam(merged, lookup, uses_tls).map(RedisAuth::AwsIam),
(None, false, true) => aws_iam(&options.aws, uses_tls(options)?).map(RedisAuth::AwsIam),
(None, false, false) => Ok(RedisAuth::Static {
username: secret_value(merged, "username"),
password: secret_value(merged, "password"),
username: secret(options.username.as_ref()),
password: secret(options.password.as_ref()),
}),
}
}
fn aws_iam(merged: &Kwargs, lookup: &dyn Lookup, uses_tls: bool) -> Result<AwsIamAuth, RedisError> {
// _uses_tls: a url's scheme decides unless cluster nodes are set, otherwise the ssl kwarg does
fn uses_tls(options: &Options) -> Result<bool, RedisError> {
match (&options.startup_nodes, text(options.url.as_ref())) {
(None, Some(url)) => Ok(url
.split_once("://")
.is_some_and(|(scheme, _)| scheme.eq_ignore_ascii_case("rediss"))),
_ => flag(options.ssl.as_ref()),
}
}
fn aws_iam(aws: &AwsOptions, uses_tls: bool) -> Result<AwsIamAuth, RedisError> {
if !uses_tls {
return Err(RedisError::AwsIamWithoutTls);
}
let user_name = present(merged, "aws_iam_user_name");
let cache_name = present(merged, "aws_iam_cache_name");
let region =
text_or_fallback(merged, "aws_iam_region", lookup).filter(|region| !region.is_empty());
match (user_name, cache_name, region) {
match (
text(aws.user_name.as_ref()),
text(aws.cache_name.as_ref()),
text(aws.region.as_ref()),
) {
(Some(user_name), Some(cache_name), Some(region)) => Ok(AwsIamAuth {
user_name,
cache_name,
region,
serverless: bool_value(merged, "aws_iam_serverless"),
serverless: options::flag(aws.serverless.as_ref())?,
}),
(user_name, cache_name, region) => Err(RedisError::AwsIamIncomplete {
missing: [
@ -112,7 +91,3 @@ fn aws_iam(merged: &Kwargs, lookup: &dyn Lookup, uses_tls: bool) -> Result<AwsIa
}),
}
}
fn present(merged: &Kwargs, key: &str) -> Option<String> {
text_value(merged, key).filter(|value| !value.is_empty())
}

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

@ -0,0 +1,156 @@
use litellm_redis_types::{
ClusterEndpoint, ClusterNode, Endpoint, HostEndpoint, RedisAuth, SentinelEndpoint, TlsOptions,
};
use litellm_secret_value::SecretValue;
use super::{
Expected, RedisError,
layers::Entry,
options::{Options, as_text, flag, integer, invalid, secret, text},
tls,
};
use crate::Value;
/// The target, first match wins like `get_redis_client`: cluster, url, Sentinel, then host.
pub(super) fn endpoint(options: &Options, auth: &RedisAuth) -> Result<Endpoint, RedisError> {
if let Some(nodes) = cluster_nodes(options.startup_nodes.as_ref())? {
reject_cluster_db(options.db.as_ref())?;
return Ok(Endpoint::Cluster(ClusterEndpoint {
nodes,
tls: tls_for(options, auth)?,
}));
}
if let Some(url) = text(options.url.as_ref()) {
return Ok(Endpoint::Url {
tls: tls::url_tls(&url)?,
url: SecretValue::new(match auth {
RedisAuth::Static { .. } => url,
_ => without_userinfo(&url),
}),
});
}
if let Some(sentinels) = options.sentinel.nodes.as_ref() {
return sentinel(options, auth, sentinels).map(Endpoint::Sentinel);
}
Ok(Endpoint::Host(HostEndpoint {
host: text(options.host.as_ref()).ok_or(RedisError::NoTarget)?,
port: integer(options.port.as_ref())?.unwrap_or(6379),
tls: tls_for(options, auth)?,
db: db(options)?,
}))
}
fn tls_for(options: &Options, auth: &RedisAuth) -> Result<Option<TlsOptions>, RedisError> {
if !flag(options.ssl.as_ref())? {
return Ok(None);
}
let gcp_ca_certs = match auth {
RedisAuth::GcpIam { .. } => options.gcp.ssl_ca_certs.as_ref(),
_ => None,
};
tls::tls_options(&options.tls, gcp_ca_certs).map(Some)
}
fn db(options: &Options) -> Result<i64, RedisError> {
integer(options.db.as_ref()).map(|db| db.unwrap_or(0))
}
fn sentinel(
options: &Options,
auth: &RedisAuth,
sentinels: &Entry,
) -> Result<SentinelEndpoint, RedisError> {
let nodes = sentinel_nodes(&sentinels.value)
.ok_or_else(|| invalid(&sentinels.source, Expected::SentinelNodes))?;
let service_name =
text(options.sentinel.service_name.as_ref()).ok_or(RedisError::NoSentinelService)?;
// redis.asyncio.Sentinel passes the connection kwargs to each monitor's Redis(host=..., port=...), so a
// host or port setting raises TypeError when litellm builds the client
if let Some(entry) = options.host.as_ref().or(options.port.as_ref()) {
return Err(RedisError::SentinelWithHost {
setting: entry.source.clone(),
});
}
Ok(SentinelEndpoint {
sentinels: nodes,
service_name,
sentinel_password: secret(options.sentinel.password.as_ref()),
tls: tls_for(options, auth)?,
db: db(options)?,
})
}
fn cluster_nodes(entry: Option<&Entry>) -> Result<Option<Vec<ClusterNode>>, RedisError> {
let Some(entry) = entry else {
return Ok(None);
};
let nodes = nodes(&entry.value, |node| {
node_address(
node.get("host").and_then(as_text),
node.get("port").and_then(as_text),
)
})
.ok_or_else(|| invalid(&entry.source, Expected::StartupNodes))?;
Ok((!nodes.is_empty()).then_some(nodes))
}
fn sentinel_nodes(value: &Value) -> Option<Vec<ClusterNode>> {
let nodes = nodes(value, |pair| match pair {
Value::Sequence(pair) if pair.len() == 2 => {
node_address(as_text(&pair[0]), as_text(&pair[1]))
}
_ => None,
})?;
(!nodes.is_empty()).then_some(nodes)
}
fn nodes(value: &Value, parse: impl Fn(&Value) -> Option<ClusterNode>) -> Option<Vec<ClusterNode>> {
match value {
Value::String(json) => serde_json::from_str::<Vec<Value>>(json)
.ok()?
.iter()
.map(parse)
.collect(),
Value::Sequence(nodes) => nodes.iter().map(parse).collect(),
_ => None,
}
}
fn node_address(host: Option<String>, port: Option<String>) -> Option<ClusterNode> {
Some(ClusterNode {
host: host?,
port: port?.parse().ok()?,
})
}
// redis-py's RedisCluster raises unless db is 0 or None
fn reject_cluster_db(db: Option<&Entry>) -> Result<(), RedisError> {
let Some(entry) = db else {
return Ok(());
};
let is_zero = match &entry.value {
Value::String(_) => matches!(integer::<i64>(Some(entry)), Ok(None | Some(0))),
Value::Bool(flag) => !flag,
Value::Number(number) => number.as_f64() == Some(0.0),
_ => false,
};
match is_zero {
true => Ok(()),
false => Err(RedisError::DbInCluster {
setting: entry.source.clone(),
}),
}
}
// _get_redis_url_kwargs keeps only plain Connection kwargs, so redis-py drops every ssl_* setting for a url
// and a credential provider replaces the url's userinfo, which _url_without_userinfo strips
fn without_userinfo(url: &str) -> String {
let Some((scheme, rest)) = url.split_once("://") else {
return url.to_owned();
};
let (authority, tail) = rest.split_at(rest.find(['/', '?', '#']).unwrap_or(rest.len()));
let host = authority
.rsplit_once('@')
.map_or(authority, |(_, host)| host);
format!("{scheme}://{host}{tail}")
}

26
litellm-rust/crates/config/src/redis/keys.rs
View file

@ -1,5 +1,8 @@
use litellm_core_utils::settings::Lookup;
use super::layers::Entry;
use crate::Value;
// Every Redis setting litellm/_redis.py reads, with how Rust handles it and where else Python looks for it.
// environment marks the kwargs _get_redis_env_kwarg_mapping reads as REDIS_<KWARG>, and fallback lists the names
// _get_redis_client_logic consults with get_secret when the merged kwargs lack the setting
@ -164,18 +167,15 @@ pub(super) fn setting(kwarg: &str) -> Option<&'static Setting> {
SETTINGS.iter().find(|setting| setting.kwarg == kwarg)
}
pub(super) fn fallback(kwarg: &str, lookup: &dyn Lookup) -> Option<(&'static str, String)> {
pub(super) fn fallback(kwarg: &str, lookup: &dyn Lookup) -> Option<Entry> {
let names = setting(kwarg).map_or(&[][..], |setting| setting.fallback);
let values: Vec<(&'static str, Option<String>)> =
names.iter().map(|name| (*name, lookup.get(name))).collect();
// Python chains these with `or`, so the first non-empty value wins and an empty one only when it is last
values
.iter()
.find(|(_, value)| value.as_deref().is_some_and(|value| !value.is_empty()))
.or(values.last())
.and_then(|(name, value)| Some((*name, value.clone()?)))
}
pub(super) fn fallback_value(kwarg: &str, lookup: &dyn Lookup) -> Option<String> {
fallback(kwarg, lookup).map(|(_, value)| value)
names.iter().find_map(|name| {
lookup
.get(name)
.filter(|value| !value.is_empty())
.map(|value| Entry {
value: Value::String(value),
source: (*name).to_owned(),
})
})
}

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

@ -0,0 +1,87 @@
use litellm_core_utils::settings::Lookup;
use super::{ConfiguredRedis, Nulls, keys, options::Kwargs};
use crate::Value;
#[derive(Clone)]
pub(super) struct Entry {
pub(super) value: Value,
pub(super) source: String,
}
const NAMED_BY_CACHE: &[&str] = &[
"host",
"port",
"password",
"socket_timeout",
"startup_nodes",
"namespace",
"gcp_service_account",
"gcp_ssl_ca_certs",
];
/// The configured settings over their `REDIS_*` values, `{**environment_kwargs, **env_overrides}`
/// in `_get_redis_client_logic`. A null left in the result is an explicit `None` that hid its
/// environment value.
pub(super) fn merged(configured: &ConfiguredRedis, lookup: &dyn Lookup) -> Kwargs {
let block: Kwargs = configured
.fields
.iter()
.map(|(key, value)| (key, environment_reference(value, lookup)))
.filter(|(key, value)| !value.is_null() || keeps_null(configured.nulls, key))
.map(|(key, value)| {
(
key.clone(),
Entry {
value,
source: key.clone(),
},
)
})
.collect();
let names_a_target = ["host", "startup_nodes", "sentinel_nodes"]
.iter()
.any(|key| is_set(&block, key));
let drop_environment_url = names_a_target && !is_set(&block, "url");
environment(lookup)
.filter(|(key, _)| !(key == "url" && drop_environment_url))
.chain(block)
.collect()
}
fn keeps_null(nulls: Nulls, key: &str) -> bool {
match nulls {
Nulls::Dropped => false,
Nulls::KeptUnlessNamedByCache => !NAMED_BY_CACHE.contains(&key),
Nulls::Kept => true,
}
}
fn environment(lookup: &dyn Lookup) -> impl Iterator<Item = (String, Entry)> + '_ {
keys::SETTINGS.iter().filter_map(|setting| {
let source = setting.environment_name()?;
let value = lookup.get(&source)?;
Some((
setting.kwarg.to_owned(),
Entry {
value: Value::String(value),
source,
},
))
})
}
// load_config resolves os.environ/ to None when unset, so a missing reference is a null like any other
fn environment_reference(value: &Value, lookup: &dyn Lookup) -> Value {
match value {
Value::String(text) => match text.strip_prefix("os.environ/") {
Some(name) => lookup.get(name).map_or(Value::Null, Value::String),
None => value.clone(),
},
_ => value.clone(),
}
}
fn is_set(kwargs: &Kwargs, key: &str) -> bool {
kwargs.get(key).is_some_and(|entry| !entry.value.is_null())
}

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

@ -0,0 +1,245 @@
use std::{collections::BTreeMap, time::Duration};
use litellm_core_utils::settings::Lookup;
use litellm_python_compat::{BigInt, number};
use litellm_secret_value::SecretValue;
use super::{
Expected, RedisError,
keys::{self, Handling},
layers::Entry,
};
use crate::Value;
pub(super) type Kwargs = BTreeMap<String, Entry>;
/// Every setting the resolver reads, after layering. A setting is `None` when it is unset, null or
/// empty, and then its environment fallback, if Python has one, stands in for it.
pub(super) struct Options {
pub(super) url: Option<Entry>,
pub(super) host: Option<Entry>,
pub(super) port: Option<Entry>,
pub(super) db: Option<Entry>,
pub(super) startup_nodes: Option<Entry>,
pub(super) sentinel: SentinelOptions,
pub(super) username: Option<Entry>,
pub(super) password: Option<Entry>,
pub(super) credential_provider: Option<Entry>,
pub(super) gcp: GcpOptions,
pub(super) azure: AzureOptions,
pub(super) aws: AwsOptions,
pub(super) ssl: Option<Entry>,
pub(super) tls: TlsInput,
pub(super) namespace: Option<Entry>,
pub(super) socket_timeout: Option<Entry>,
pub(super) socket_connect_timeout: Option<Entry>,
}
pub(super) struct SentinelOptions {
pub(super) nodes: Option<Entry>,
pub(super) service_name: Option<Entry>,
pub(super) password: Option<Entry>,
}
pub(super) struct GcpOptions {
pub(super) service_account: Option<Entry>,
pub(super) ssl_ca_certs: Option<Entry>,
}
pub(super) struct AzureOptions {
pub(super) ad_token: Option<Entry>,
pub(super) client_id: Option<Entry>,
pub(super) tenant_id: Option<Entry>,
pub(super) client_secret: Option<Entry>,
}
pub(super) struct AwsOptions {
pub(super) auth: Option<Entry>,
pub(super) user_name: Option<Entry>,
pub(super) cache_name: Option<Entry>,
pub(super) region: Option<Entry>,
pub(super) serverless: Option<Entry>,
}
/// The TLS settings, read from the kwargs or from a `rediss://` url's query.
pub(super) struct TlsInput {
pub(super) cert_reqs: Option<Entry>,
pub(super) ca_certs: Option<Entry>,
pub(super) certfile: Option<Entry>,
pub(super) keyfile: Option<Entry>,
pub(super) check_hostname: Option<Entry>,
}
impl Options {
/// The typed settings and where each accepted-but-ignored one came from.
pub(super) fn read(
kwargs: Kwargs,
lookup: &dyn Lookup,
) -> Result<(Self, Vec<String>), RedisError> {
if let Some((_, entry)) = kwargs
.iter()
.find(|(key, _)| handling(key) == Some(Handling::Reject))
{
return Err(RedisError::Unsupported {
setting: entry.source.clone(),
});
}
let ignored = kwargs
.iter()
.filter(|(key, _)| handling(key) == Some(Handling::Warn))
.map(|(_, entry)| entry.source.clone())
.collect();
let get = |key: &str| {
kwargs
.get(key)
.filter(|entry| is_present(&entry.value))
.cloned()
.or_else(|| keys::fallback(key, lookup))
};
let options = Self {
url: get("url"),
host: get("host"),
port: get("port"),
db: get("db"),
startup_nodes: get("startup_nodes"),
sentinel: SentinelOptions {
nodes: get("sentinel_nodes"),
service_name: get("service_name"),
password: get("sentinel_password"),
},
username: get("username"),
password: get("password"),
credential_provider: get("credential_provider"),
gcp: GcpOptions {
service_account: get("gcp_service_account"),
ssl_ca_certs: get("gcp_ssl_ca_certs"),
},
azure: AzureOptions {
ad_token: get("azure_redis_ad_token"),
client_id: get("azure_client_id"),
tenant_id: get("azure_tenant_id"),
client_secret: get("azure_client_secret"),
},
aws: AwsOptions {
auth: get("aws_iam_auth"),
user_name: get("aws_iam_user_name"),
cache_name: get("aws_iam_cache_name"),
region: get("aws_iam_region"),
serverless: get("aws_iam_serverless"),
},
ssl: get("ssl"),
tls: TlsInput {
cert_reqs: get("ssl_cert_reqs"),
ca_certs: get("ssl_ca_certs"),
certfile: get("ssl_certfile"),
keyfile: get("ssl_keyfile"),
check_hostname: get("ssl_check_hostname"),
},
namespace: get("namespace"),
socket_timeout: get("socket_timeout"),
socket_connect_timeout: get("socket_connect_timeout"),
};
Ok((options, ignored))
}
}
fn handling(key: &str) -> Option<Handling> {
keys::setting(key).map(|setting| setting.handling)
}
// Python reads most of these with `x or fallback`, so an empty value counts as unset everywhere
fn is_present(value: &Value) -> bool {
match value {
Value::Null => false,
Value::String(text) => !text.is_empty(),
Value::Sequence(items) => !items.is_empty(),
Value::Mapping(entries) => !entries.is_empty(),
Value::Bool(_) | Value::Number(_) | Value::Tagged(_) => true,
}
}
pub(super) fn invalid(setting: &str, expected: Expected) -> RedisError {
RedisError::Invalid {
setting: setting.to_owned(),
expected,
}
}
pub(super) 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,
}
}
pub(super) fn text(entry: Option<&Entry>) -> Option<String> {
entry.and_then(|entry| as_text(&entry.value))
}
pub(super) fn secret(entry: Option<&Entry>) -> Option<SecretValue> {
text(entry).map(SecretValue::new)
}
/// A flag every Python surface reads the same way. `_str_to_bool` takes only `true`, `1` and `yes`
/// as true while the proxy's pydantic models also take `on`, `t` and `y`, so text they disagree on,
/// or text neither accepts, is an error instead of a guess.
pub(super) fn flag(entry: Option<&Entry>) -> Result<bool, RedisError> {
let Some(entry) = entry else {
return Ok(false);
};
match &entry.value {
Value::Bool(flag) => Ok(*flag),
Value::Number(number) => Ok(number.as_f64().is_some_and(|number| number != 0.0)),
Value::String(text) => match text.to_ascii_lowercase().as_str() {
"true" | "1" | "yes" => Ok(true),
"false" | "0" | "no" | "off" | "f" | "n" => Ok(false),
_ => Err(invalid(&entry.source, Expected::Boolean)),
},
_ => Err(invalid(&entry.source, Expected::Boolean)),
}
}
pub(super) fn parsed<T>(
entry: Option<&Entry>,
expected: Expected,
parse: impl Fn(&str) -> Option<T>,
) -> Result<Option<T>, RedisError> {
entry
.and_then(|entry| Some((as_text(&entry.value)?, &entry.source)))
.map(|(text, source)| parse(&text).ok_or_else(|| invalid(source, expected)))
.transpose()
}
pub(super) fn integer<T: TryFrom<BigInt>>(entry: Option<&Entry>) -> Result<Option<T>, RedisError> {
coerced(entry, Expected::Integer, number::int, |number| {
T::try_from(number).ok()
})
}
pub(super) fn seconds(entry: Option<&Entry>) -> Result<Option<Duration>, RedisError> {
coerced(entry, Expected::Seconds, number::float, |seconds| {
Duration::try_from_secs_f64(seconds).ok()
})
}
fn coerced<N, T>(
entry: Option<&Entry>,
expected: Expected,
parse: impl Fn(&str) -> Option<N>,
convert: impl Fn(N) -> Option<T>,
) -> Result<Option<T>, RedisError> {
let Some((entry, text)) = entry.and_then(|entry| Some((entry, as_text(&entry.value)?))) else {
return Ok(None);
};
match parse(&text) {
Some(number) => convert(number)
.map(Some)
.ok_or_else(|| invalid(&entry.source, expected)),
None => {
tracing::warn!(setting = %entry.source, "Redis setting is not a number, so the default is used like redis-py");
Ok(None)
}
}
}

20
litellm-rust/crates/config/src/redis/secrets.rs
View file

@ -3,7 +3,7 @@ use std::collections::{BTreeMap, BTreeSet};
use litellm_core_utils::settings::Lookup;
use litellm_secret_value::SecretValue;
use super::{RedisSettings, Value, keys::SETTINGS};
use super::{ConfiguredRedis, Value, keys::SETTINGS};
#[derive(Clone, Debug, Default)]
pub struct SecretSnapshot(BTreeMap<String, Option<SecretValue>>);
@ -31,7 +31,10 @@ impl Lookup for SecretSnapshot {
match self.0.get(name) {
Some(value) => value.as_ref().map(|value| value.expose().to_owned()),
None => {
debug_assert!(false, "{name} is missing from RedisSettings::secret_names");
debug_assert!(
false,
"{name} is missing from ConfiguredRedis::secret_names"
);
tracing::error!(
name,
"a Redis secret was read without being in the snapshot"
@ -42,15 +45,12 @@ impl Lookup for SecretSnapshot {
}
}
impl RedisSettings {
impl ConfiguredRedis {
pub fn secret_names(&self) -> BTreeSet<String> {
let references = self
.kwargs(false)
.into_values()
.filter_map(|entry| match entry.value {
Value::String(text) => text.strip_prefix("os.environ/").map(str::to_owned),
_ => None,
});
let references = self.fields.values().filter_map(|value| match value {
Value::String(text) => text.strip_prefix("os.environ/").map(str::to_owned),
_ => None,
});
let declared = SETTINGS.iter().flat_map(|setting| {
setting
.environment_name()

51
litellm-rust/crates/config/src/redis/sentinel.rs
View file

@ -1,51 +0,0 @@
use litellm_redis_types::ClusterNode as RedisNode;
use litellm_secret_value::SecretValue;
use super::{
Expected, Kwargs, Lookup, RedisError, as_text, configured, invalid, keys, node_address, nodes,
text_or_fallback,
};
use crate::Value;
pub(super) struct SentinelTarget {
pub(super) sentinels: Vec<RedisNode>,
pub(super) service_name: String,
pub(super) sentinel_password: Option<SecretValue>,
}
pub(super) fn sentinel_target(
merged: &Kwargs,
lookup: &dyn Lookup,
) -> Result<Option<SentinelTarget>, RedisError> {
let (nodes, source) = match (
configured(merged, "sentinel_nodes"),
keys::fallback("sentinel_nodes", lookup),
merged
.get("sentinel_nodes")
.filter(|entry| !entry.value.is_null()),
) {
(Some(entry), _, _) => (sentinel_nodes(&entry.value), entry.source.as_str()),
(None, Some((source, json)), _) => (sentinel_nodes(&Value::String(json)), source),
(None, None, Some(empty)) => (None, empty.source.as_str()),
(None, None, None) => return Ok(None),
};
let nodes = nodes.ok_or_else(|| invalid(source, Expected::SentinelNodes))?;
let service_name =
text_or_fallback(merged, "service_name", lookup).ok_or(RedisError::NoSentinelService)?;
Ok(Some(SentinelTarget {
sentinels: nodes,
service_name,
sentinel_password: text_or_fallback(merged, "sentinel_password", lookup)
.map(SecretValue::new),
}))
}
fn sentinel_nodes(value: &Value) -> Option<Vec<RedisNode>> {
let nodes = nodes(value, |pair| match pair {
Value::Sequence(pair) if pair.len() == 2 => {
node_address(as_text(&pair[0]), as_text(&pair[1]))
}
_ => None,
})?;
(!nodes.is_empty()).then_some(nodes)
}

60
litellm-rust/crates/config/src/redis/tls.rs
View file

@ -1,6 +1,10 @@
use litellm_redis_types::TlsOptions;
use super::{Entry, Expected, Kwargs, RedisError, bool_value, parsed, text_value};
use super::{
Expected, RedisError,
layers::Entry,
options::{TlsInput, flag, parsed, text},
};
use crate::Value;
// redis-py's parse_url turns every query parameter into a client kwarg. Only the TLS ones have a redis-rs
@ -13,39 +17,35 @@ const URL_TLS_SETTINGS: &[&str] = &[
"ssl_check_hostname",
];
/// `gcp_ca_certs` replaces `ssl_ca_certs` when GCP IAM is on, as `_get_redis_client_logic` does.
pub(super) fn tls_options(
merged: &Kwargs,
gcp_ca_certs: Option<String>,
input: &TlsInput,
gcp_ca_certs: Option<&Entry>,
) -> Result<TlsOptions, RedisError> {
let verify_certificate =
parsed(
merged,
"ssl_cert_reqs",
Expected::CertRequirement,
|text| match text {
"required" | "optional" => Some(true),
"none" => Some(false),
_ => None,
},
)?
.unwrap_or(true);
Ok(TlsOptions {
ca_certs: gcp_ca_certs
.or_else(|| text_value(merged, "ssl_ca_certs"))
.map(Into::into),
certfile: text_value(merged, "ssl_certfile").map(Into::into),
keyfile: text_value(merged, "ssl_keyfile").map(Into::into),
verify_certificate,
check_hostname: bool_value(merged, "ssl_check_hostname"),
ca_certs: text(gcp_ca_certs.or(input.ca_certs.as_ref())).map(Into::into),
certfile: text(input.certfile.as_ref()).map(Into::into),
keyfile: text(input.keyfile.as_ref()).map(Into::into),
verify_certificate: verify_certificate(input.cert_reqs.as_ref())?,
check_hostname: flag(input.check_hostname.as_ref())?,
})
}
fn verify_certificate(cert_reqs: Option<&Entry>) -> Result<bool, RedisError> {
parsed(cert_reqs, Expected::CertRequirement, |text| match text {
"required" | "optional" => Some(true),
"none" => Some(false),
_ => None,
})
.map(|verify| verify.unwrap_or(true))
}
pub(super) fn url_tls(url: &str) -> Result<TlsOptions, RedisError> {
let Ok(parsed) = url::Url::parse(url) else {
return Ok(TlsOptions::verified());
};
let tls_scheme = parsed.scheme().eq_ignore_ascii_case("rediss");
let query: Kwargs = parsed
let query: Vec<(String, Entry)> = parsed
.query_pairs()
.map(|(name, value)| {
(
@ -65,10 +65,20 @@ pub(super) fn url_tls(url: &str) -> Result<TlsOptions, RedisError> {
setting: entry.source.clone(),
});
}
let get = |key: &str| {
query
.iter()
.rev()
.find(|(name, _)| name == key)
.map(|(_, entry)| entry.clone())
};
Ok(TlsOptions {
check_hostname: text_value(&query, "ssl_check_hostname")
ca_certs: text(get("ssl_ca_certs").as_ref()).map(Into::into),
certfile: text(get("ssl_certfile").as_ref()).map(Into::into),
keyfile: text(get("ssl_keyfile").as_ref()).map(Into::into),
verify_certificate: verify_certificate(get("ssl_cert_reqs").as_ref())?,
check_hostname: text(get("ssl_check_hostname").as_ref())
.is_some_and(|text| url_bool(&text)),
..tls_options(&query, None)?
})
}

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

@ -1,44 +1,57 @@
use std::{
collections::HashMap,
fmt::Write,
sync::{Arc, Mutex},
};
use std::collections::{BTreeMap, HashMap};
use litellm_config::{
AwsIamAuth, AzureAdAuth, CacheSettings, Config, Expected, RedisAuth, RedisError, RedisSettings,
ResolvedRedis, SecretSnapshot,
CacheSettings, Config, ConfiguredRedis, Expected, RedisError, Resolution, SecretSnapshot, Value,
};
use litellm_redis_types::{
AwsIamAuth, AzureAdAuth, ClusterNode, Endpoint, RedisAuth, ResolvedRedis, TlsOptions,
};
use litellm_redis_types::{ClusterNode, Endpoint, TlsOptions};
use rstest::rstest;
use serde_json::json;
use tracing::{Event, Subscriber, field::Visit};
use tracing_subscriber::{Layer, layer::Context, prelude::*};
fn resolve_settings(
settings: &RedisSettings,
fn resolution(
configured: &ConfiguredRedis,
environment: &HashMap<String, String>,
) -> Result<ResolvedRedis, RedisError> {
) -> Result<Resolution, RedisError> {
let lookup = |name: &str| environment.get(name).cloned();
settings.resolve(
&SecretSnapshot::read(settings.secret_names(), &lookup),
configured.resolve(
&SecretSnapshot::read(configured.secret_names(), &lookup),
&lookup,
)
}
fn resolve(
fn redis_cache_kwargs(kwargs: serde_json::Value) -> ConfiguredRedis {
ConfiguredRedis::redis_cache_kwargs(
serde_json::from_value::<BTreeMap<String, Value>>(kwargs).unwrap(),
)
}
fn resolve_kwargs(
kwargs: serde_json::Value,
environment: &HashMap<String, String>,
) -> Result<ResolvedRedis, RedisError> {
resolution(&redis_cache_kwargs(kwargs), environment).map(|resolution| resolution.redis)
}
fn coordination_resolution(
yaml: &str,
environment: &[(&str, &str)],
) -> Result<litellm_config::ResolvedRedis, RedisError> {
) -> Result<Resolution, RedisError> {
let environment: HashMap<String, String> = environment
.iter()
.map(|(key, value)| ((*key).to_owned(), (*value).to_owned()))
.collect();
let config = Config::from_yaml(yaml).unwrap();
let settings = config
let configured = config
.general_settings
.coordination_redis
.unwrap_or_default();
resolve_settings(&settings, &environment)
.unwrap_or_default()
.coordination()?;
resolution(&configured, &environment)
}
fn resolve(yaml: &str, environment: &[(&str, &str)]) -> Result<ResolvedRedis, RedisError> {
coordination_resolution(yaml, environment).map(|resolution| resolution.redis)
}
fn resolve_cache(yaml: &str) -> Result<ResolvedRedis, RedisError> {
@ -53,13 +66,16 @@ fn resolve_cache_in(yaml: &str, environment: &[(&str, &str)]) -> Result<Resolved
.map(|(_, value)| (*value).to_owned())
};
let config = Config::from_yaml(yaml).unwrap();
match config.litellm_settings.cache_params.unwrap() {
CacheSettings::Redis(settings) => settings.resolve_cache(
&SecretSnapshot::read(settings.secret_names(), &lookup),
&lookup,
),
let configured = match config.litellm_settings.cache_params.unwrap() {
CacheSettings::Redis(settings) => settings.cache_params(),
CacheSettings::Other(_) => panic!("expected a Redis cache"),
}
};
configured
.resolve(
&SecretSnapshot::read(configured.secret_names(), &lookup),
&lookup,
)
.map(|resolution| resolution.redis)
}
fn cluster_ports(resolved: &ResolvedRedis) -> Vec<u16> {
@ -400,7 +416,6 @@ fn a_host_or_port_cannot_reach_sentinel(
}
#[rstest]
#[case::empty("general_settings:\n coordination_redis:\n sentinel_nodes: []\n service_name: primary", &[], "sentinel_nodes")]
#[case::not_pairs(
"general_settings:\n coordination_redis:\n sentinel_nodes: [{host: s, port: 26379}]\n service_name: primary",
&[],
@ -559,8 +574,9 @@ fn debug_redacts_redis_secrets_and_url() {
)
.unwrap();
let settings = config.general_settings.coordination_redis.unwrap();
let resolved = resolve_settings(&settings, &HashMap::new()).unwrap();
let debug = format!("{settings:?} {resolved:?}");
let configured = settings.coordination().unwrap();
let resolved = resolution(&configured, &HashMap::new()).unwrap().redis;
let debug = format!("{settings:?} {configured:?} {resolved:?}");
assert!(!debug.contains("redis://user:password"));
assert!(!debug.contains("credential-user"));
@ -605,44 +621,25 @@ fn legacy_unknown_settings_do_not_change_resolution() {
assert!(matches!(resolved.endpoint, Endpoint::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(" ");
#[case::configured(
"general_settings:\n coordination_redis:\n host: redis\n max_connections: 47",
&[],
"max_connections"
)]
#[case::from_the_environment(
"general_settings:\n coordination_redis:\n host: redis",
&[("REDIS_MAX_CONNECTIONS", "47")],
"REDIS_MAX_CONNECTIONS"
)]
fn ignored_settings_are_named_by_their_source(
#[case] yaml: &str,
#[case] environment: &[(&str, &str)],
#[case] source: &str,
) {
let resolution = coordination_resolution(yaml, environment).unwrap();
assert!(warnings.contains("max_connections"));
assert!(!warnings.contains("47"));
assert_eq!(resolution.ignored, vec![source.to_owned()]);
}
fn tls_view(tls: Option<&TlsOptions>) -> serde_json::Value {
@ -773,6 +770,16 @@ fn view(resolved: &ResolvedRedis) -> serde_json::Value {
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
)]
#[case::an_ambiguous_flag_is_refused(
json!({"host": "redis", "ssl": "on"}),
&[],
Err("invalid Redis setting ssl: expected a boolean"),
)]
#[case::empty_sentinel_nodes_without_the_environment_name_no_target(
json!({"sentinel_nodes": [], "service_name": "primary"}),
&[],
Err("either host or url must be specified for redis"),
)]
#[case::ssl_ca_path_is_unsupported(
json!({"host": "redis", "ssl": true, "ssl_ca_path": "/ca"}),
&[],
@ -1053,13 +1060,12 @@ fn resolves_like_python_redis(
#[case] environment: &[(&str, &str)],
#[case] expected: Result<serde_json::Value, &str>,
) {
let settings: RedisSettings = serde_json::from_value(kwargs).unwrap();
let environment: HashMap<String, String> = environment
.iter()
.map(|(name, value)| ((*name).to_owned(), (*value).to_owned()))
.collect();
let outcome = resolve_settings(&settings, &environment)
let outcome = resolve_kwargs(kwargs, &environment)
.map(|resolved| view(&resolved))
.map_err(|error| error.to_string());
@ -1068,22 +1074,22 @@ fn resolves_like_python_redis(
fn snapshot(values: &[(&str, &str)]) -> SecretSnapshot {
let values: HashMap<&str, &str> = values.iter().copied().collect();
SecretSnapshot::read(RedisSettings::default().secret_names(), &|name: &str| {
values.get(name).map(|value| (*value).to_owned())
})
SecretSnapshot::read(
ConfiguredRedis::environment_only().secret_names(),
&|name: &str| values.get(name).map(|value| (*value).to_owned()),
)
}
#[rstest]
fn secret_manager_values_reach_the_resolver_through_the_snapshot() {
let settings: RedisSettings =
serde_json::from_str(r#"{"host": "os.environ/COORD_HOST"}"#).unwrap();
let settings = redis_cache_kwargs(json!({"host": "os.environ/COORD_HOST"}));
let secrets = SecretSnapshot::read(settings.secret_names(), &|name: &str| match name {
"COORD_HOST" => Some("from-the-manager".to_owned()),
"REDIS_PASSWORD" => Some("manager-secret".to_owned()),
_ => None,
});
let resolved = settings.resolve(&secrets, &|_: &str| None).unwrap();
let resolved = settings.resolve(&secrets, &|_: &str| None).unwrap().redis;
assert!(matches!(&resolved.endpoint, Endpoint::Host(host) if host.host == "from-the-manager"));
assert_eq!(resolved.auth, static_auth(None, Some("manager-secret")));
@ -1097,15 +1103,15 @@ fn the_azure_username_comes_from_the_process_environment_like_python(
#[case] environment: &[(&str, &str)],
#[case] expected: Option<&str>,
) {
let settings: RedisSettings =
serde_json::from_str(r#"{"host": "redis", "azure_redis_ad_token": "true"}"#).unwrap();
let settings = redis_cache_kwargs(json!({"host": "redis", "azure_redis_ad_token": "true"}));
let environment: HashMap<&str, &str> = environment.iter().copied().collect();
let resolved = settings
.resolve(&snapshot(secrets), &|name: &str| {
environment.get(name).map(|value| (*value).to_owned())
})
.unwrap();
.unwrap()
.redis;
assert!(
matches!(&resolved.auth, RedisAuth::AzureAd(azure) if azure.username.as_deref() == expected),

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

@ -24,6 +24,7 @@ litellm-error.workspace = true
litellm-circuit-breaker.workspace = true
litellm-redis-client.workspace = true
litellm-redis-connect.workspace = true
litellm-redis-types.workspace = true
rustls.workspace = true
litellm-http.workspace = true
litellm-llms.workspace = true

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

@ -2,11 +2,12 @@ use std::time::Duration;
use litellm_circuit_breaker::{Breaker, Config as BreakerConfig, State};
use litellm_config::{
CacheSettings, Config, Expected, RedisError, RedisSettings, ResolvedRedis, SecretSnapshot,
Value, environment_has_target,
CacheSettings, Config, ConfiguredRedis, Expected, RedisError, RedisSettings, Resolution,
SecretSnapshot, Value, environment_has_target,
};
use litellm_core_utils::settings::Lookup;
use litellm_redis_client::RedisClient;
use litellm_redis_types::ResolvedRedis;
use crate::Error;
@ -34,15 +35,13 @@ enum Target {
pub(crate) fn plan(config: &Config, lookup: &dyn Lookup) -> Result<Plan, Error> {
let response_cache = redis_cache_params(config)
.map(|settings| settings.resolve_cache(&snapshot(settings, lookup), lookup))
.map(|settings| resolve(&settings.cache_params(), lookup))
.transpose()?;
let coordination = match &config.general_settings.coordination_redis {
Some(settings) => {
settings.validate_coordination()?;
Some(Target::Configured(
settings.resolve(&snapshot(settings, lookup), lookup)?,
))
}
Some(settings) => Some(Target::Configured(resolve(
&settings.coordination()?,
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),
@ -107,16 +106,18 @@ fn redis_cache_params(config: &Config) -> Option<&RedisSettings> {
}
}
fn snapshot(settings: &RedisSettings, lookup: &dyn Lookup) -> SecretSnapshot {
SecretSnapshot::read(settings.secret_names(), lookup)
fn resolve(configured: &ConfiguredRedis, lookup: &dyn Lookup) -> Result<Params, RedisError> {
let secrets = SecretSnapshot::read(configured.secret_names(), lookup);
let Resolution { redis, ignored } = configured.resolve(&secrets, lookup)?;
for setting in ignored {
tracing::warn!(%setting, "Redis setting is accepted but ignored");
}
Ok(redis)
}
fn environment(lookup: &dyn Lookup) -> Result<Option<Params>, Error> {
match environment_has_target(lookup) {
true => RedisSettings::default()
.resolve(&snapshot(&RedisSettings::default(), lookup), lookup)
.map(Some)
.map_err(Into::into),
true => Ok(Some(resolve(&ConfiguredRedis::environment_only(), lookup)?)),
false => Ok(None),
}
}
@ -239,9 +240,10 @@ fn environment_value<T>(
mod tests {
use std::{collections::HashMap, time::Duration};
use litellm_config::{Config, RedisAuth, RedisError};
use litellm_config::{Config, RedisError};
use litellm_redis_client::{Endpoint, HostEndpoint};
use litellm_redis_testing::RedisServer;
use litellm_redis_types::RedisAuth;
use rstest::rstest;
use super::{Params, Plan, Target, connect, plan};

2
litellm-rust/crates/redis-connect/AGENTS.md
View file

@ -1,3 +1,3 @@
- Turns a `litellm_config::ResolvedRedis` into a `RedisClient`, the same way for every host. The endpoint passes through unchanged because config already resolves to `litellm-redis-types`. `redis-client` stays cloud-agnostic, so credentials are mapped here
- Turns a `litellm_redis_types::ResolvedRedis` into a `RedisClient`, the same way for every host. It depends on `litellm-redis-types`, never on `litellm-config`, so a host that resolves Redis some other way connects the same way. `redis-client` stays cloud-agnostic, so credentials are mapped here
- `RedisAuth::Host` (a Python `credential_provider` object) fails at connect with an error naming the key, since no Rust host can build one. Cloud auth (GCP, Azure, AWS) fails the same way until its token sources land, so a config that needs it never silently runs without Redis
- Tests run against real Redis from `litellm-redis-testing`

3
litellm-rust/crates/redis-connect/Cargo.toml
View file

@ -7,12 +7,13 @@ repository.workspace = true
[dependencies]
litellm-circuit-breaker.workspace = true
litellm-config.workspace = true
litellm-redis-types.workspace = true
litellm-redis-client.workspace = true
thiserror.workspace = true
tracing.workspace = true
[dev-dependencies]
litellm-config.workspace = true
litellm-redis-testing.workspace = true
rustls.workspace = true
serde_json.workspace = true

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

@ -1,8 +1,8 @@
use litellm_circuit_breaker::Breaker;
use litellm_config::{RedisAuth, ResolvedRedis};
use litellm_redis_client::{
Credentials, DEFAULT_TIMEOUT, Namespace, RedisClient, RedisConfig, Timeouts,
};
use litellm_redis_types::{RedisAuth, ResolvedRedis};
use crate::Error;

9
litellm-rust/crates/redis-connect/tests/connect.rs
View file

@ -1,20 +1,23 @@
use std::{collections::HashMap, time::Duration};
use litellm_config::{RedisSettings, ResolvedRedis, SecretSnapshot};
use litellm_config::{RedisSettings, SecretSnapshot};
use litellm_redis_connect::{Error, connect};
use litellm_redis_testing::{RedisSentinel, RedisServer, RedisTlsServer, SERVICE_NAME, eventually};
use litellm_redis_types::ResolvedRedis;
use rstest::rstest;
fn resolved(kwargs: serde_json::Value, environment: &[(&str, &str)]) -> ResolvedRedis {
let settings: RedisSettings = serde_json::from_value(kwargs).unwrap();
let environment: HashMap<&str, &str> = environment.iter().copied().collect();
let lookup = |name: &str| environment.get(name).map(|value| (*value).to_owned());
settings
let configured = settings.coordination().unwrap();
configured
.resolve(
&SecretSnapshot::read(settings.secret_names(), &lookup),
&SecretSnapshot::read(configured.secret_names(), &lookup),
&lookup,
)
.unwrap()
.redis
}
fn port(url: &str) -> u16 {

2
litellm-rust/crates/redis-types/AGENTS.md
View file

@ -1,2 +1,2 @@
- The Redis endpoint, node and TLS types that `config` resolves to and `redis-client` connects with, so neither converts the other's copy
- `ResolvedRedis` and the endpoint, node, TLS and auth types inside it: what `config` resolves to, `redis-connect` turns into a client and `redis-client` connects with, so none of them converts another's copy
- Data only. It depends on `secret-value` and nothing else, so `config` never pulls in redis-rs or tokio. Anything that needs redis-rs, like building a `ConnectionInfo` or reading TLS files, lives in `redis-client`

44
litellm-rust/crates/redis-types/src/lib.rs
View file

@ -1,7 +1,18 @@
use std::path::PathBuf;
use std::{path::PathBuf, time::Duration};
use litellm_secret_value::SecretValue;
/// One Redis connection with every setting, environment value and secret already applied, the
/// value a host hands to the connector.
#[derive(Clone, Debug, PartialEq)]
pub struct ResolvedRedis {
pub endpoint: Endpoint,
pub auth: RedisAuth,
pub namespace: Option<String>,
pub timeout: Option<Duration>,
pub connect_timeout: Option<Duration>,
}
#[derive(Clone, Debug, PartialEq)]
pub enum Endpoint {
Url { url: SecretValue, tls: TlsOptions },
@ -56,3 +67,34 @@ impl TlsOptions {
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub enum RedisAuth {
Static {
username: Option<SecretValue>,
password: Option<SecretValue>,
},
/// The host supplies credentials, like a redis-py `credential_provider` object.
Host,
GcpIam {
service_account: String,
},
AzureAd(AzureAdAuth),
AwsIam(AwsIamAuth),
}
#[derive(Clone, Debug, PartialEq)]
pub struct AzureAdAuth {
pub client_id: Option<String>,
pub tenant_id: Option<String>,
pub client_secret: Option<SecretValue>,
pub username: Option<String>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct AwsIamAuth {
pub user_name: String,
pub cache_name: String,
pub region: String,
pub serverless: bool,
}

17
tests/unit/test_redis.py
View file

@ -2125,6 +2125,23 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
_expected_view({"db": 0, "host": "redis", "mode": "host", "port": 6379}),
id="plaintext_ignores_tls_requirements",
),
pytest.param(
{"host": "redis", "ssl": "on"},
{},
_expected_view({"db": 0, "host": "redis", "mode": "host", "port": 6379}),
id="an_ambiguous_flag_is_refused",
),
pytest.param(
{"sentinel_nodes": [], "service_name": "primary"},
{},
{
"rejected": {
"message": "Both 'sentinel_nodes' and 'service_name' are required for Redis Sentinel.",
"type": "ValueError",
}
},
id="empty_sentinel_nodes_without_the_environment_name_no_target",
),
pytest.param(
{"host": "redis", "ssl": True, "ssl_ca_path": "/ca"},
{},