scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 280

feat(redis-client): stop rejecting redis-py settings and support health checks, Unix sockets and CA bundles

Browse source 
A Redis setting the Rust client cannot honor is now warned and dropped with
what that means, so a config that works on Python never fails to build on
Rust. health_check_interval, unix_socket_path, ssl_ca_data and ssl_ca_path
are read and applied like redis-py.

health_check_interval resolves to a three-state HealthCheck, so 0 stays off
on a cluster instead of turning on Python's 25 second cluster default, which
only an unset interval takes. unix_socket_path resolves to Target::Unix. A
url's ?db=, ?username= and ?password= are read like parse_url reads them
instead of being dropped, and ssl_min_version no longer warns at TLS 1.2 or
lower, which rustls already meets.
This commit is contained in:
Yujong Lee 2026-11-04 01:48:28 +09:00 • committed by yujonglee
parent c07ef4568c
commit 5b75f64f71
23 changed files with 1010 additions and 224 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
litellm-rust/Cargo.lock generated
View file

@ -4381,6 +4381,7 @@ dependencies = [
"socket2 0.6.5",
"thiserror 2.0.19",
"tokio",
"tracing",
]
[[package]]

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

@ -1,22 +1,26 @@
## Redis settings
- The boundary is surface in, `litellm_redis_types::Resolution` 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 the `Resolution`: the target, the auth and the timeouts Rust models, every other plain setting verbatim in `passthrough`, the names of object-valued settings in `host_objects`, and every accepted-but-ignored setting by source, which the host logs. A host passes its `credential_provider` object as a `ProviderHandle` through `ConfiguredRedis::with_credential_provider`, and a `credential_provider` key without one is `RedisError::ProviderWithoutHandle`. 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, splits the unmodeled ones into passthrough and host objects, collects ignored ones and fills Python's `get_secret` fallbacks. `endpoint.rs` and `auth.rs` pick the target and the credentials from `Options` alone
- The boundary is surface in, `litellm_redis_types::Resolution` 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 the `Resolution`: the target, the auth and the timeouts Rust models, every other plain setting verbatim in `passthrough`, the names of object-valued settings in `host_objects`, and every setting the Rust client ignores as an `Ignored` with its source and consequence, which the host logs. A host passes its `credential_provider` object as a `ProviderHandle` through `ConfiguredRedis::with_credential_provider`, and a `credential_provider` key without one is `RedisError::ProviderWithoutHandle`. 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`, splits the unmodeled settings into passthrough and host objects, 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
- 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
- `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`
- Every Redis setting Python reads is one row in `SETTINGS` in `src/redis/keys.rs`: how Rust handles it (read, or warn and drop), 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 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
- 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_ca_data`, `ssl_ca_path`, `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
- 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`
- 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`
- `ssl_ca_path` is rejected on both sides. redis-py 5.3.1's Redis constructor raises TypeError and Rust returns Unsupported naming the setting
- `retry_on_timeout`, `retry_on_error`, `protocol`, `cache`, `cache_config`, `health_check_interval`, `socket_keepalive` and `max_connections` are accepted with a warning and ignored. Cluster clients still turn TCP keepalive on, like Python's async cluster default, but redis-rs has no periodic health check, so Python's 25s cluster default has no equivalent
- A boolean `REDIS_RETRY_ON_ERROR` reaches redis-py's connection constructor as a non-iterable and raises TypeError. Rust ignores it with a warning and otherwise resolves the default host connection, verified against the host-default fixture
- The resolver never fails on a setting it cannot honor. A `Warn` row is dropped and returned as an `Ignored` naming its source (`ssl_ciphers` or `REDIS_SSL_CIPHERS`) and the consequence from `keys::consequence`, because a Python deployment that worked must keep working when its stores move to Rust. Only a value that is malformed (`RedisError::Invalid`) or a target that is missing fails. Promoting a `Warn` row to `Read` is then a pure improvement with no behavior cliff
- `ssl_ciphers`, `ssl_min_version` above TLS 1.2, `ssl_password`, the OCSP options, `connection_pool`, `event_dispatcher`, `redis_connect_func` and `retry` are ignored. Ignoring OCSP connects without revocation checks, and the consequence says so
- `ssl_ca_data` and `ssl_ca_path` are read into `TlsOptions` and trusted next to `ssl_ca_certs`. `unix_socket_path` resolves to `Target::Unix`, which like redis-py's `UnixDomainSocketConnection` replaces host, port and TLS. Python still requires `host` next to it, so Rust does too
- A url's `?db=`, `?username=` and `?password=` are read like `parse_url` reads them, so `?db=` wins over the path and the userinfo wins over `?username=` and `?password=`. Any other query parameter except the TLS ones on a `rediss://` url is ignored naming `url ?<name>`. redis-py turns each one into a client kwarg
- `retry_on_timeout`, `retry_on_error`, `protocol`, `cache`, `cache_config`, `socket_keepalive` and `max_connections` are ignored. Cluster clients still turn TCP keepalive on, like Python's async cluster default
- `health_check_interval` is read as an `int` of seconds like `_coerce_redis_kwargs_types` into `HealthCheck`. Zero is `Disabled` and stays off on every target, a negative interval checks before every command like redis-py, and only `Unset` takes a default. `redis-connect` gives an unset cluster Python's 25s default and `redis-client` pings every node before the next command once the client sat idle that long, outside the breaker and the call events
- Cases where Rust and Python disagree on purpose are cases in both contract tests with a different expected outcome on each side
- `ssl_ca_path`: redis-py 5.3.1's async `Redis` has no such parameter and raises `TypeError` building the client, while Rust trusts the directory like `load_verify_locations(capath=...)`
- `on`, `t` and `y` for a flag are `RedisError::Invalid` on every surface. pydantic reads them as true for `coordination_redis` and `_str_to_bool` reads them as false everywhere else, so Rust refuses to pick one. Other unrecognized text is false in Python and an error here
- Empty `sentinel_nodes` with no `REDIS_SENTINEL_NODES` is no target at all, so it fails as "either host or url must be specified". redis-py keeps the empty list and fails building the Sentinel client

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

@ -1,7 +1,7 @@
use std::collections::BTreeMap;
use std::{collections::BTreeMap, time::Duration};
use litellm_core_utils::settings::Lookup;
use litellm_redis_types::{Auth, ProviderHandle, Resolution, StaticAuth, Target};
use litellm_redis_types::{Auth, HealthCheck, ProviderHandle, Resolution, StaticAuth, Target};
use serde::{Deserialize, Deserializer, de::Error as _};
use veil::Redact;
@ -165,7 +165,7 @@ impl ConfiguredRedis {
let kwargs = layers::merged(self, secrets);
let (options, unmodeled) = Options::read(kwargs, secrets)?;
let auth = auth::redis_auth(&options, self.credential_provider, environment)?;
let target = endpoint::target(&options, &auth)?;
let (target, url_ignored) = endpoint::target(&options, &auth)?;
let auth = match (&target, auth) {
(Target::Url(_), Auth::Static(StaticAuth { username, .. })) => {
Auth::Static(StaticAuth {
@ -181,13 +181,24 @@ impl ConfiguredRedis {
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())?,
health_check: health_check(options.health_check_interval.as_ref())?,
passthrough: unmodeled.passthrough,
host_objects: unmodeled.host_objects,
ignored: unmodeled.ignored,
ignored: unmodeled.ignored.into_iter().chain(url_ignored).collect(),
})
}
}
// redis-py checks health only when the interval is truthy, so zero turns it off and a negative interval checks
// before every command
fn health_check(entry: Option<&layers::Entry>) -> Result<HealthCheck, RedisError> {
Ok(match options::integer::<i64>(entry)? {
None => HealthCheck::Unset,
Some(0) => HealthCheck::Disabled,
Some(seconds) => HealthCheck::Every(Duration::from_secs(seconds.try_into().unwrap_or(0))),
})
}
pub fn environment_has_target(lookup: &dyn Lookup) -> bool {
[
"REDIS_HOST",

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

@ -1,5 +1,6 @@
use litellm_redis_types::{
Auth, Cluster, Node, RedisUrl, Sentinel, Standalone, StaticAuth, Target, TlsOptions,
Auth, Cluster, Ignored, Node, RedisUrl, Sentinel, Standalone, StaticAuth, Target, TlsOptions,
UnixSocket,
};
use litellm_secret_value::SecretValue;
@ -11,33 +12,49 @@ use super::{
};
use crate::Value;
/// The target, first match wins like `get_redis_client`: cluster, url, Sentinel, then host.
pub(super) fn target(options: &Options, auth: &Auth) -> Result<Target, RedisError> {
/// The target, first match wins like `get_redis_client`: cluster, url, Sentinel, then host. A
/// url also returns the query parameters it ignores.
pub(super) fn target(options: &Options, auth: &Auth) -> Result<(Target, Vec<Ignored>), RedisError> {
if let Some(nodes) = cluster_nodes(options.startup_nodes.as_ref())? {
reject_cluster_db(options.db.as_ref())?;
return Ok(Target::Cluster(Cluster {
let cluster = Cluster {
nodes,
tls: tls_for(options, auth)?,
}));
};
return Ok((Target::Cluster(cluster), Vec::new()));
}
if let Some(url) = text(options.url.as_ref()) {
return Ok(Target::Url(RedisUrl {
tls: tls::url_tls(&url)?,
let query = tls::url_query(&url)?;
let url = RedisUrl {
url: SecretValue::new(match auth {
Auth::Static(_) => url,
_ => without_userinfo(&url),
}),
}));
tls: query.tls,
db: query.db,
username: query.username,
password: query.password,
};
return Ok((Target::Url(url), query.ignored));
}
if let Some(sentinels) = options.sentinel.nodes.as_ref() {
return sentinel(options, auth, sentinels).map(Target::Sentinel);
let sentinel = sentinel(options, auth, sentinels)?;
return Ok((Target::Sentinel(sentinel), Vec::new()));
}
Ok(Target::Standalone(Standalone {
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)?,
}))
let host = text(options.host.as_ref()).ok_or(RedisError::NoTarget)?;
let target = match text(options.unix_socket_path.as_ref()) {
Some(path) => Target::Unix(UnixSocket {
path: path.into(),
db: db(options)?,
}),
None => Target::Standalone(Standalone {
host,
port: integer(options.port.as_ref())?.unwrap_or(6379),
tls: tls_for(options, auth)?,
db: db(options)?,
}),
};
Ok((target, Vec::new()))
}
fn tls_for(options: &Options, auth: &Auth) -> Result<Option<TlsOptions>, RedisError> {

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

@ -1,7 +1,5 @@
#[derive(Debug, thiserror::Error)]
pub enum RedisError {
#[error("Redis host does not yet support configured setting {setting}")]
Unsupported { setting: String },
#[error("invalid Redis setting {setting}: expected {expected}")]
Invalid { setting: String, expected: Expected },
#[error("either host or url must be specified for redis")]

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

@ -1,17 +1,17 @@
use litellm_core_utils::settings::Lookup;
use litellm_redis_types::Ignored;
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.
// Warn rows are read by Python and dropped here with a warning naming the consequence, never an error.
// 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
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(super) enum Handling {
Read,
Ignore,
Warn,
Reject,
}
pub(super) struct Setting {
@ -79,16 +79,16 @@ pub(super) const SETTINGS: &[Setting] = &[
),
Setting::new("cache", Handling::Warn, true, &[]),
Setting::new("cache_config", Handling::Warn, true, &[]),
Setting::new("charset", Handling::Ignore, true, &[]),
Setting::new("client_name", Handling::Ignore, true, &[]),
Setting::new("connection_pool", Handling::Reject, false, &[]),
Setting::new("charset", Handling::Warn, true, &[]),
Setting::new("client_name", Handling::Warn, true, &[]),
Setting::new("connection_pool", Handling::Warn, false, &[]),
Setting::new("credential_provider", Handling::Read, false, &[]),
Setting::new("db", Handling::Read, true, &[]),
Setting::new("decode_responses", Handling::Ignore, true, &[]),
Setting::new("encoding", Handling::Ignore, true, &[]),
Setting::new("encoding_errors", Handling::Ignore, true, &[]),
Setting::new("errors", Handling::Ignore, true, &[]),
Setting::new("event_dispatcher", Handling::Reject, true, &[]),
Setting::new("decode_responses", Handling::Warn, true, &[]),
Setting::new("encoding", Handling::Warn, true, &[]),
Setting::new("encoding_errors", Handling::Warn, true, &[]),
Setting::new("errors", Handling::Warn, true, &[]),
Setting::new("event_dispatcher", Handling::Warn, true, &[]),
Setting::new(
"gcp_service_account",
Handling::Read,
@ -101,17 +101,17 @@ pub(super) const SETTINGS: &[Setting] = &[
true,
&["REDIS_GCP_SSL_CA_CERTS"],
),
Setting::new("health_check_interval", Handling::Warn, true, &[]),
Setting::new("health_check_interval", Handling::Read, true, &[]),
Setting::new("host", Handling::Read, true, &[]),
Setting::new("lib_name", Handling::Ignore, true, &[]),
Setting::new("lib_version", Handling::Ignore, true, &[]),
Setting::new("lib_name", Handling::Warn, true, &[]),
Setting::new("lib_version", Handling::Warn, true, &[]),
Setting::new("max_connections", Handling::Warn, true, &[]),
Setting::new("namespace", Handling::Read, false, &[]),
Setting::new("password", Handling::Read, true, &[]),
Setting::new("port", Handling::Read, true, &[]),
Setting::new("protocol", Handling::Warn, true, &[]),
Setting::new("redis_connect_func", Handling::Reject, true, &[]),
Setting::new("retry", Handling::Reject, false, &[]),
Setting::new("redis_connect_func", Handling::Warn, true, &[]),
Setting::new("retry", Handling::Warn, false, &[]),
Setting::new("retry_on_error", Handling::Warn, true, &[]),
Setting::new("retry_on_timeout", Handling::Warn, true, &[]),
Setting::new(
@ -132,37 +132,63 @@ pub(super) const SETTINGS: &[Setting] = &[
false,
&["REDIS_SERVICE_NAME"],
),
Setting::new("single_connection_client", Handling::Ignore, true, &[]),
Setting::new("single_connection_client", Handling::Warn, true, &[]),
Setting::new("socket_connect_timeout", Handling::Read, true, &[]),
Setting::new("socket_keepalive", Handling::Warn, true, &[]),
Setting::new("socket_keepalive_options", Handling::Warn, true, &[]),
Setting::new("socket_timeout", Handling::Read, true, &[]),
Setting::new("ssl", Handling::Read, true, &[]),
Setting::new("ssl_ca_certs", Handling::Read, true, &[]),
Setting::new("ssl_ca_data", Handling::Reject, true, &[]),
Setting::new("ssl_ca_path", Handling::Reject, true, &[]),
Setting::new("ssl_ca_data", Handling::Read, true, &[]),
Setting::new("ssl_ca_path", Handling::Read, true, &[]),
Setting::new("ssl_cert_reqs", Handling::Read, true, &[]),
Setting::new("ssl_certfile", Handling::Read, true, &[]),
Setting::new("ssl_check_hostname", Handling::Read, true, &[]),
Setting::new("ssl_ciphers", Handling::Reject, true, &[]),
Setting::new("ssl_ciphers", Handling::Warn, true, &[]),
Setting::new("ssl_keyfile", Handling::Read, true, &[]),
Setting::new("ssl_min_version", Handling::Reject, true, &[]),
Setting::new("ssl_ocsp_context", Handling::Reject, true, &[]),
Setting::new("ssl_ocsp_expected_cert", Handling::Reject, true, &[]),
Setting::new("ssl_password", Handling::Reject, true, &[]),
Setting::new("ssl_validate_ocsp", Handling::Reject, true, &[]),
Setting::new("ssl_validate_ocsp_stapled", Handling::Reject, true, &[]),
Setting::new("ssl_min_version", Handling::Warn, true, &[]),
Setting::new("ssl_ocsp_context", Handling::Warn, true, &[]),
Setting::new("ssl_ocsp_expected_cert", Handling::Warn, true, &[]),
Setting::new("ssl_password", Handling::Warn, true, &[]),
Setting::new("ssl_validate_ocsp", Handling::Warn, true, &[]),
Setting::new("ssl_validate_ocsp_stapled", Handling::Warn, true, &[]),
Setting::new(
"startup_nodes",
Handling::Read,
false,
&["REDIS_CLUSTER_NODES"],
),
Setting::new("unix_socket_path", Handling::Reject, true, &[]),
Setting::new("unix_socket_path", Handling::Read, true, &[]),
Setting::new("url", Handling::Read, true, &[]),
Setting::new("username", Handling::Read, true, &[]),
];
pub(super) fn ignored(setting: String, kwarg: &str) -> Ignored {
Ignored {
setting,
consequence: consequence(kwarg),
}
}
// rustls only speaks TLS 1.2 and 1.3, so a minimum of 1.2 or lower is already met. redis-py takes an
// ssl.TLSVersion, which reaches Rust as its name or its value
pub(super) fn is_met_by_rustls(kwarg: &str, value: &Value) -> bool {
if kwarg != "ssl_min_version" {
return false;
}
match value {
Value::Number(number) => matches!(number.as_i64(), Some(-2 | 769 | 770 | 771)),
Value::String(text) => matches!(
text.strip_prefix("TLSVersion.")
.unwrap_or(text)
.to_ascii_uppercase()
.as_str(),
"MINIMUM_SUPPORTED" | "TLSV1" | "TLSV1_1" | "TLSV1_2"
),
_ => false,
}
}
// redis-py parameters whose value is a Python object rather than plain data. credential_provider is modeled as
// Auth::Provider instead
const HOST_OBJECTS: &[&str] = &[
@ -185,6 +211,30 @@ pub(super) fn is_host_object(kwarg: &str) -> bool {
HOST_OBJECTS.contains(&kwarg)
}
/// What ignoring a warned setting means for the connection, so the operator reads the consequence
/// rather than only the key name.
fn consequence(kwarg: &str) -> &'static str {
match kwarg {
"ssl_validate_ocsp"
| "ssl_validate_ocsp_stapled"
| "ssl_ocsp_context"
| "ssl_ocsp_expected_cert" => {
"rustls has no client-side OCSP, so the connection is made without revocation checks"
}
"ssl_ciphers" | "ssl_min_version" => {
"rustls negotiates TLS 1.2+ with its own ECDHE AEAD suites"
}
"ssl_password" => "an encrypted client key cannot be read, so the TLS handshake fails",
"credential_provider" | "redis_connect_func" => {
"the Python object is not callable from Rust, so static username and password are used"
}
"connection_pool" | "retry" | "event_dispatcher" => {
"the Rust client uses its own pool, retries and events"
}
_ => "the Rust client has no equivalent",
}
}
pub(super) fn setting(kwarg: &str) -> Option<&'static Setting> {
SETTINGS.iter().find(|setting| setting.kwarg == kwarg)
}

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

@ -5,7 +5,7 @@ use std::{
use litellm_core_utils::settings::Lookup;
use litellm_python_compat::{BigInt, number};
use litellm_redis_types::Passthrough;
use litellm_redis_types::{Ignored, Passthrough};
use litellm_secret_value::SecretValue;
use super::{
@ -24,6 +24,7 @@ pub(super) struct Options {
pub(super) host: Option<Entry>,
pub(super) port: Option<Entry>,
pub(super) db: Option<Entry>,
pub(super) unix_socket_path: Option<Entry>,
pub(super) startup_nodes: Option<Entry>,
pub(super) sentinel: SentinelOptions,
pub(super) username: Option<Entry>,
@ -37,6 +38,7 @@ pub(super) struct Options {
pub(super) namespace: Option<Entry>,
pub(super) socket_timeout: Option<Entry>,
pub(super) socket_connect_timeout: Option<Entry>,
pub(super) health_check_interval: Option<Entry>,
}
pub(super) struct SentinelOptions {
@ -69,6 +71,8 @@ pub(super) struct AwsOptions {
pub(super) struct TlsInput {
pub(super) cert_reqs: Option<Entry>,
pub(super) ca_certs: Option<Entry>,
pub(super) ca_data: Option<Entry>,
pub(super) ca_path: Option<Entry>,
pub(super) certfile: Option<Entry>,
pub(super) keyfile: Option<Entry>,
pub(super) check_hostname: Option<Entry>,
@ -78,7 +82,7 @@ pub(super) struct TlsInput {
pub(super) struct Unmodeled {
pub(super) passthrough: Passthrough,
pub(super) host_objects: BTreeSet<String>,
pub(super) ignored: Vec<String>,
pub(super) ignored: Vec<Ignored>,
}
impl Unmodeled {
@ -104,8 +108,11 @@ impl Unmodeled {
passthrough: Passthrough::new(passthrough),
host_objects: objects.into_iter().map(|(key, _)| key.clone()).collect(),
ignored: unmodeled()
.filter(|(key, _)| handling(key) == Some(Handling::Warn))
.map(|(_, entry)| entry.source.clone())
.filter(|(key, entry)| {
handling(key) == Some(Handling::Warn)
&& !keys::is_met_by_rustls(key, &entry.value)
})
.map(|(key, entry)| keys::ignored(entry.source.clone(), key))
.collect(),
})
}
@ -117,14 +124,6 @@ impl Options {
kwargs: Kwargs,
lookup: &dyn Lookup,
) -> Result<(Self, Unmodeled), RedisError> {
if let Some((_, entry)) = kwargs
.iter()
.find(|(key, _)| handling(key) == Some(Handling::Reject))
{
return Err(RedisError::Unsupported {
setting: entry.source.clone(),
});
}
let unmodeled = Unmodeled::read(&kwargs)?;
let get = |key: &str| {
kwargs
@ -138,6 +137,7 @@ impl Options {
host: get("host"),
port: get("port"),
db: get("db"),
unix_socket_path: get("unix_socket_path"),
startup_nodes: get("startup_nodes"),
sentinel: SentinelOptions {
nodes: get("sentinel_nodes"),
@ -168,6 +168,8 @@ impl Options {
tls: TlsInput {
cert_reqs: get("ssl_cert_reqs"),
ca_certs: get("ssl_ca_certs"),
ca_data: get("ssl_ca_data"),
ca_path: get("ssl_ca_path"),
certfile: get("ssl_certfile"),
keyfile: get("ssl_keyfile"),
check_hostname: get("ssl_check_hostname"),
@ -175,6 +177,7 @@ impl Options {
namespace: get("namespace"),
socket_timeout: get("socket_timeout"),
socket_connect_timeout: get("socket_connect_timeout"),
health_check_interval: get("health_check_interval"),
};
Ok((options, unmodeled))
}

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

@ -1,17 +1,23 @@
use litellm_redis_types::TlsOptions;
use litellm_python_compat::number;
use litellm_redis_types::{Ignored, TlsOptions};
use litellm_secret_value::SecretValue;
use super::{
Expected, RedisError,
Expected, RedisError, keys,
layers::Entry,
options::{TlsInput, flag, parsed, text},
options::{TlsInput, flag, parsed, secret, 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
// equivalent, so any other parameter, and any parameter on a plain redis:// url, is rejected
// redis-py's parse_url turns every query parameter into a client kwarg. Only these and, on a rediss:// url, the
// TLS ones have a redis-rs equivalent, so any other parameter is ignored
const URL_CONNECTION_SETTINGS: &[&str] = &["db", "username", "password"];
const URL_TLS_SETTINGS: &[&str] = &[
"ssl_cert_reqs",
"ssl_ca_certs",
"ssl_ca_data",
"ssl_ca_path",
"ssl_certfile",
"ssl_keyfile",
"ssl_check_hostname",
@ -24,6 +30,8 @@ pub(super) fn tls_options(
) -> Result<TlsOptions, RedisError> {
Ok(TlsOptions {
ca_certs: text(gcp_ca_certs.or(input.ca_certs.as_ref())).map(Into::into),
ca_data: text(input.ca_data.as_ref()),
ca_path: text(input.ca_path.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())?,
@ -40,13 +48,29 @@ fn verify_certificate(cert_reqs: Option<&Entry>) -> Result<bool, RedisError> {
.map(|verify| verify.unwrap_or(true))
}
pub(super) fn url_tls(url: &str) -> Result<Option<TlsOptions>, RedisError> {
let Ok(parsed) = url::Url::parse(url) else {
return Ok(None);
/// What redis-py's `parse_url` reads from a url's query on top of the url itself.
pub(super) struct UrlQuery {
pub(super) tls: Option<TlsOptions>,
pub(super) db: Option<i64>,
pub(super) username: Option<SecretValue>,
pub(super) password: Option<SecretValue>,
pub(super) ignored: Vec<Ignored>,
}
pub(super) fn url_query(url: &str) -> Result<UrlQuery, RedisError> {
let Ok(url) = url::Url::parse(url) else {
return Ok(UrlQuery {
tls: None,
db: None,
username: None,
password: None,
ignored: Vec::new(),
});
};
let tls_scheme = parsed.scheme().eq_ignore_ascii_case("rediss");
let query: Vec<(String, Entry)> = parsed
let tls_scheme = url.scheme().eq_ignore_ascii_case("rediss");
let query: Vec<(String, Entry)> = url
.query_pairs()
.filter(|(_, value)| !value.is_empty())
.map(|(name, value)| {
(
name.to_string(),
@ -57,17 +81,10 @@ pub(super) fn url_tls(url: &str) -> Result<Option<TlsOptions>, RedisError> {
)
})
.collect();
if let Some((_, entry)) = query
.iter()
.find(|(name, _)| !tls_scheme || !URL_TLS_SETTINGS.contains(&name.as_str()))
{
return Err(RedisError::Unsupported {
setting: entry.source.clone(),
});
}
if !tls_scheme {
return Ok(None);
}
let (query, ignored): (Vec<_>, Vec<_>) = query.into_iter().partition(|(name, _)| {
URL_CONNECTION_SETTINGS.contains(&name.as_str())
|| (tls_scheme && URL_TLS_SETTINGS.contains(&name.as_str()))
});
let get = |key: &str| {
query
.iter()
@ -75,14 +92,32 @@ pub(super) fn url_tls(url: &str) -> Result<Option<TlsOptions>, RedisError> {
.find(|(name, _)| name == key)
.map(|(_, entry)| entry.clone())
};
Ok(Some(TlsOptions {
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)),
}))
let tls = match tls_scheme {
true => Some(TlsOptions {
ca_certs: text(get("ssl_ca_certs").as_ref()).map(Into::into),
ca_data: text(get("ssl_ca_data").as_ref()),
ca_path: text(get("ssl_ca_path").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)),
}),
false => None,
};
Ok(UrlQuery {
tls,
// parse_url reads ?db= with int() and raises on anything else
db: parsed(get("db").as_ref(), Expected::Integer, |text| {
number::int(text).and_then(|db| i64::try_from(db).ok())
})?,
username: secret(get("username").as_ref()),
password: secret(get("password").as_ref()),
ignored: ignored
.into_iter()
.map(|(name, entry)| keys::ignored(entry.source, &name))
.collect(),
})
}
// redis-py's to_bool: only 0, F, FALSE, N and NO (any case) are false

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

@ -4,7 +4,8 @@ use litellm_config::{
CacheSettings, Config, ConfiguredRedis, Expected, RedisError, SecretSnapshot, Value,
};
use litellm_redis_types::{
Auth, AwsIamAuth, AzureAdAuth, Node, ProviderHandle, Resolution, StaticAuth, Target, TlsOptions,
Auth, AwsIamAuth, AzureAdAuth, HealthCheck, Node, ProviderHandle, Resolution, StaticAuth,
Target, TlsOptions,
};
use rstest::rstest;
use serde_json::json;
@ -117,6 +118,7 @@ fn resolves_redis_through_the_parsed_config(
) {
let resolved = resolve(yaml, environment).unwrap();
let description = match resolved.target {
Target::Unix(socket) => format!("unix:{}", socket.path.display()),
Target::Url(url) => format!("url:{}", url.url.expose()),
Target::Sentinel(sentinel) => format!("sentinel:{}", sentinel.service_name),
Target::Standalone(host) => format!(
@ -161,14 +163,27 @@ fn resolves_redis_through_the_parsed_config(
&[],
"connection_pool"
)]
fn rejected_settings_name_their_source(
fn unsupported_settings_resolve_and_are_named_by_their_source(
#[case] yaml: &str,
#[case] environment: &[(&str, &str)],
#[case] expected: &str,
) {
let error = resolve(yaml, environment).unwrap_err();
let resolution = coordination_resolution(yaml, environment).unwrap();
assert!(matches!(error, RedisError::Unsupported { setting } if setting == expected));
assert_eq!(
resolution.target,
Target::Standalone(litellm_redis_types::Standalone {
host: "redis".into(),
port: 6379,
db: 0,
tls: None,
})
);
let [ignored] = resolution.ignored.as_slice() else {
panic!("expected one ignored setting, got {:?}", resolution.ignored);
};
assert_eq!(ignored.setting, expected);
assert!(!ignored.consequence.is_empty());
}
fn static_auth(username: Option<&str>, password: Option<&str>) -> Auth {
@ -312,22 +327,22 @@ fn incomplete_aws_iam_auth_fails(#[case] yaml: &str, #[case] message: &str) {
#[case::defaults_verify_the_certificate_but_not_the_hostname(
" ssl: true",
&[],
Some(TlsOptions { ca_certs: None, certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
Some(TlsOptions { ca_certs: None, ca_data: None, ca_path: None, certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
)]
#[case::every_option(
" ssl: true\n ssl_ca_certs: /ca.pem\n ssl_certfile: /client.pem\n ssl_keyfile: /client.key\n ssl_cert_reqs: none\n ssl_check_hostname: true",
&[],
Some(TlsOptions { ca_certs: Some("/ca.pem".into()), certfile: Some("/client.pem".into()), keyfile: Some("/client.key".into()), verify_certificate: false, check_hostname: true })
Some(TlsOptions { ca_certs: Some("/ca.pem".into()), ca_data: None, ca_path: None, certfile: Some("/client.pem".into()), keyfile: Some("/client.key".into()), verify_certificate: false, check_hostname: true })
)]
#[case::options_from_the_environment(
" ssl: true",
&[("REDIS_SSL_CA_CERTS", "/env-ca.pem"), ("REDIS_SSL_CERT_REQS", "optional")],
Some(TlsOptions { ca_certs: Some("/env-ca.pem".into()), certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
Some(TlsOptions { ca_certs: Some("/env-ca.pem".into()), ca_data: None, ca_path: None, certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
)]
#[case::gcp_ca_certs_replace_the_configured_ones(
" ssl: true\n ssl_ca_certs: /ca.pem\n gcp_service_account: account\n gcp_ssl_ca_certs: /gcp-ca.pem",
&[],
Some(TlsOptions { ca_certs: Some("/gcp-ca.pem".into()), certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
Some(TlsOptions { ca_certs: Some("/gcp-ca.pem".into()), ca_data: None, ca_path: None, certfile: None, keyfile: None, verify_certificate: true, check_hostname: false })
)]
#[case::options_without_ssl_stay_plaintext(" ssl_ca_certs: /ca.pem", &[], None)]
fn tls_options_resolve_like_redis_py(
@ -575,19 +590,19 @@ fn a_missing_canonical_startup_nodes_reference_shadows_the_cache_alias() {
}
#[rstest]
fn rejected_environment_settings_never_print_the_value() {
let error = resolve(
fn unsupported_environment_settings_never_carry_the_value() {
let resolution = coordination_resolution(
"model_list: []",
&[
("REDIS_HOST", "redis"),
("REDIS_SSL_CIPHERS", "private-option-value"),
],
)
.unwrap_err();
let message = error.to_string();
.unwrap();
let ignored = format!("{:?}", resolution.ignored);
assert!(message.contains("REDIS_SSL_CIPHERS"));
assert!(!message.contains("private-option-value"));
assert!(ignored.contains("REDIS_SSL_CIPHERS"), "{ignored}");
assert!(!ignored.contains("private-option-value"), "{ignored}");
}
#[rstest]
@ -678,7 +693,12 @@ fn ignored_settings_are_named_by_their_source(
) {
let resolution = coordination_resolution(yaml, environment).unwrap();
assert_eq!(resolution.ignored, vec![source.to_owned()]);
let settings: Vec<&str> = resolution
.ignored
.iter()
.map(|ignored| ignored.setting.as_str())
.collect();
assert_eq!(settings, vec![source]);
}
#[rstest]
@ -723,10 +743,73 @@ fn settings_rust_does_not_model_reach_the_host_instead_of_being_dropped(
);
}
#[rstest]
#[case::read_on_a_plain_url(
"redis://cache:6379/1?db=3&username=query-user&password=query-secret",
Some(3),
Some("query-user"),
Some("query-secret")
)]
#[case::absent("rediss://cache:6380/1", None, None, None)]
fn a_url_query_carries_db_and_credentials_like_parse_url(
#[case] url: &str,
#[case] db: Option<i64>,
#[case] username: Option<&str>,
#[case] password: Option<&str>,
) {
let resolved = resolve_kwargs(json!({"url": url}), &HashMap::new()).unwrap();
let Target::Url(parsed) = resolved.target else {
panic!("expected a url target, got {:?}", resolved.target);
};
assert_eq!(parsed.db, db);
assert_eq!(expose(parsed.username.as_ref()), username);
assert_eq!(expose(parsed.password.as_ref()), password);
assert!(resolved.ignored.is_empty(), "{:?}", resolved.ignored);
}
#[rstest]
fn a_url_query_db_that_is_not_an_integer_fails_like_parse_url() {
let error = resolve_kwargs(
json!({"url": "redis://cache:6379?db=first"}),
&HashMap::new(),
)
.unwrap_err();
assert!(
matches!(error, RedisError::Invalid { setting, expected: Expected::Integer } if setting == "url ?db")
);
}
#[rstest]
#[case::tls_1_2_is_already_met(json!("TLSv1_2"), false)]
#[case::the_enum_name(json!("TLSVersion.TLSv1_2"), false)]
#[case::the_enum_value(json!(771), false)]
#[case::tls_1_3_is_not_enforced(json!("TLSv1_3"), true)]
fn ssl_min_version_warns_only_when_rustls_cannot_meet_it(
#[case] version: serde_json::Value,
#[case] warns: bool,
) {
let resolved = resolve_kwargs(
json!({"host": "redis", "ssl": true, "ssl_min_version": version}),
&HashMap::new(),
)
.unwrap();
assert_eq!(
!resolved.ignored.is_empty(),
warns,
"{:?}",
resolved.ignored
);
}
fn tls_view(tls: Option<&TlsOptions>) -> serde_json::Value {
tls.map_or(serde_json::Value::Null, |tls| {
serde_json::json!({
"ca_certs": tls.ca_certs,
"ca_data": tls.ca_data,
"ca_path": tls.ca_path,
"certfile": tls.certfile,
"keyfile": tls.keyfile,
"verify_certificate": tls.verify_certificate,
@ -746,24 +829,11 @@ fn expose(value: Option<&litellm_secret_value::SecretValue>) -> Option<&str> {
value.map(litellm_secret_value::SecretValue::expose)
}
fn expected_view(fields: serde_json::Value) -> serde_json::Value {
let serde_json::Value::Object(fields) = fields else {
unreachable!("expected views are objects")
};
let defaults = json!({
"auth": {"kind": "static", "username": null, "password": null},
"tls": null,
"socket_timeout": 5.0,
"socket_connect_timeout": 5.0,
});
let serde_json::Value::Object(defaults) = defaults else {
unreachable!("the defaults are an object")
};
serde_json::Value::Object(defaults.into_iter().chain(fields).collect())
}
fn view(resolved: &Resolution) -> serde_json::Value {
let target = match &resolved.target {
Target::Unix(socket) => serde_json::json!({
"mode": "unix", "path": socket.path, "tls": null,
}),
Target::Standalone(host) => serde_json::json!({
"mode": "host",
"host": host.host,
@ -818,6 +888,11 @@ fn view(resolved: &Resolution) -> serde_json::Value {
"auth": auth,
"socket_timeout": resolved.timeout.map_or(5.0, |timeout| timeout.as_secs_f64()),
"socket_connect_timeout": resolved.connect_timeout.or(resolved.timeout).map_or(5.0, |timeout| timeout.as_secs_f64()),
"health_check_interval": match (&resolved.target, resolved.health_check) {
(_, HealthCheck::Every(interval)) => Some(interval.as_secs_f64()),
(Target::Cluster(_), HealthCheck::Unset) => Some(25.0),
(_, HealthCheck::Unset | HealthCheck::Disabled) => None,
},
});
let serde_json::Value::Object(rest) = rest else {
unreachable!("the view is an object")
@ -825,31 +900,48 @@ fn view(resolved: &Resolution) -> serde_json::Value {
serde_json::Value::Object(target.into_iter().chain(rest).collect())
}
fn expected_view(fields: serde_json::Value) -> serde_json::Value {
let serde_json::Value::Object(fields) = fields else {
unreachable!("expected views are objects")
};
let defaults = json!({
"auth": {"kind": "static", "username": null, "password": null},
"tls": null,
"socket_timeout": 5.0,
"socket_connect_timeout": 5.0,
"health_check_interval": null,
});
let serde_json::Value::Object(defaults) = defaults else {
unreachable!("the defaults are an object")
};
serde_json::Value::Object(defaults.into_iter().chain(fields).collect())
}
#[rstest]
#[case::host_defaults(
json!({"host": "redis"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::empty_cluster_nodes_use_the_environment(
json!({"startup_nodes": []}),
&[("REDIS_CLUSTER_NODES", r#"[{"host": "node-a", "port": 7000}]"#)],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "health_check_interval": 25.0}))),
)]
#[case::empty_sentinel_nodes_use_the_environment(
json!({"sentinel_nodes": [], "service_name": "primary"}),
&[("REDIS_SENTINEL_NODES", r#"[["sentinel-a", 26379]]"#)],
Ok(expected_view(json!({"db": 0, "mode": "sentinel", "nodes": ["sentinel-a:26379"], "sentinel_password": null, "service_name": "primary"}))),
Ok(expected_view(json!({"mode": "sentinel", "nodes": ["sentinel-a:26379"], "service_name": "primary", "sentinel_password": null, "db": 0}))),
)]
#[case::empty_service_name_uses_the_environment(
json!({"sentinel_nodes": [["sentinel-a", 26379]], "service_name": ""}),
&[("REDIS_SERVICE_NAME", "primary")],
Ok(expected_view(json!({"db": 0, "mode": "sentinel", "nodes": ["sentinel-a:26379"], "sentinel_password": null, "service_name": "primary"}))),
Ok(expected_view(json!({"mode": "sentinel", "nodes": ["sentinel-a:26379"], "service_name": "primary", "sentinel_password": null, "db": 0}))),
)]
#[case::plaintext_ignores_tls_requirements(
json!({"host": "redis", "ssl_cert_reqs": "strict"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::an_ambiguous_flag_is_refused(
json!({"host": "redis", "ssl": "on"}),
@ -864,22 +956,22 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::ssl_ca_path_is_unsupported(
json!({"host": "redis", "ssl": true, "ssl_ca_path": "/ca"}),
&[],
Err("Redis host does not yet support configured setting ssl_ca_path"),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": "/ca", "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::host_with_every_basic_setting(
json!({"db": 2, "host": "redis", "password": "pass", "port": 6380, "socket_timeout": 1.5, "ssl": true, "username": "user"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "static", "password": "pass", "username": "user"}, "db": 2, "host": "redis", "mode": "host", "port": 6380, "socket_connect_timeout": 1.5, "socket_timeout": 1.5, "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6380, "db": 2, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "static", "username": "user", "password": "pass"}, "socket_timeout": 1.5, "socket_connect_timeout": 1.5}))),
)]
#[case::environment_host_port_and_password(
json!({}),
&[("REDIS_HOST", "environment"), ("REDIS_PASSWORD", "environment-secret"), ("REDIS_PORT", "6390"), ("REDIS_SSL", "True")],
Ok(expected_view(json!({"auth": {"kind": "static", "password": "environment-secret", "username": null}, "db": 0, "host": "environment", "mode": "host", "port": 6390, "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "environment", "port": 6390, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "static", "username": null, "password": "environment-secret"}}))),
)]
#[case::string_numbers_are_coerced(
json!({"db": "3", "host": "redis", "port": "6380", "socket_timeout": "2.5"}),
&[],
Ok(expected_view(json!({"db": 3, "host": "redis", "mode": "host", "port": 6380, "socket_connect_timeout": 2.5, "socket_timeout": 2.5}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6380, "db": 3, "socket_timeout": 2.5, "socket_connect_timeout": 2.5}))),
)]
#[case::padded_and_underscored_numbers_parse_like_python_int_and_float(
json!({"host": "redis", "port": "6380\n", "db": " 2 ", "socket_timeout": "1_0.5\n"}),
@ -894,7 +986,7 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::socket_timeout_from_the_environment(
json!({"host": "redis"}),
&[("REDIS_SOCKET_TIMEOUT", "0.75")],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379, "socket_connect_timeout": 0.75, "socket_timeout": 0.75}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "socket_timeout": 0.75, "socket_connect_timeout": 0.75}))),
)]
#[case::environment_url(
json!({}),
@ -909,42 +1001,42 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::an_explicit_host_drops_the_environment_url(
json!({"host": "configured"}),
&[("REDIS_URL", "redis://environment:6379")],
Ok(expected_view(json!({"db": 0, "host": "configured", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "configured", "port": 6379, "db": 0}))),
)]
#[case::a_url_drops_host_db_and_password_and_keeps_the_username(
json!({"db": 4, "host": "dropped", "password": "dropped", "url": "rediss://cache:6380", "username": "user"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "static", "password": null, "username": "user"}, "mode": "url", "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}, "url": "rediss://cache:6380"}))),
Ok(expected_view(json!({"mode": "url", "url": "rediss://cache:6380", "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "static", "username": "user", "password": null}}))),
)]
#[case::a_url_ignores_tls_settings_outside_the_url(
json!({"ssl_ca_certs": "/ca.pem", "ssl_cert_reqs": "none", "ssl_check_hostname": true, "url": "rediss://cache:6380"}),
&[],
Ok(expected_view(json!({"mode": "url", "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}, "url": "rediss://cache:6380"}))),
Ok(expected_view(json!({"mode": "url", "url": "rediss://cache:6380", "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::startup_nodes_beat_a_url(
json!({"password": "cluster-secret", "startup_nodes": [{"host": "node-a", "port": 7000}], "url": "redis://single:6379"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "static", "password": "cluster-secret", "username": null}, "mode": "cluster", "nodes": ["node-a:7000"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "auth": {"kind": "static", "username": null, "password": "cluster-secret"}, "health_check_interval": 25.0}))),
)]
#[case::environment_cluster_nodes(
json!({}),
&[("REDIS_CLUSTER_NODES", r#"[{"host": "node-a", "port": "7000"}, {"host": "node-b", "port": 7001}]"#)],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000", "node-b:7001"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000", "node-b:7001"], "health_check_interval": 25.0}))),
)]
#[case::cluster_tls_options(
json!({"ssl": "true", "ssl_ca_certs": "/ca.pem", "ssl_check_hostname": true, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "tls": {"ca_certs": "/ca.pem", "certfile": null, "check_hostname": true, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "tls": {"ca_certs": "/ca.pem", "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": true}, "health_check_interval": 25.0}))),
)]
#[case::sentinel_from_settings(
json!({"password": "data-secret", "sentinel_nodes": [["sentinel-a", 26379], ["sentinel-b", 26380]], "sentinel_password": "sentinel-secret", "service_name": "primary"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "static", "password": "data-secret", "username": null}, "db": 0, "mode": "sentinel", "nodes": ["sentinel-a:26379", "sentinel-b:26380"], "sentinel_password": "sentinel-secret", "service_name": "primary"}))),
Ok(expected_view(json!({"mode": "sentinel", "nodes": ["sentinel-a:26379", "sentinel-b:26380"], "service_name": "primary", "sentinel_password": "sentinel-secret", "db": 0, "auth": {"kind": "static", "username": null, "password": "data-secret"}}))),
)]
#[case::sentinel_from_the_environment(
json!({}),
&[("REDIS_SENTINEL_NODES", r#"[["sentinel-a", 26379]]"#), ("REDIS_SENTINEL_PASSWORD", "sentinel-secret"), ("REDIS_SERVICE_NAME", "primary")],
Ok(expected_view(json!({"db": 0, "mode": "sentinel", "nodes": ["sentinel-a:26379"], "sentinel_password": "sentinel-secret", "service_name": "primary"}))),
Ok(expected_view(json!({"mode": "sentinel", "nodes": ["sentinel-a:26379"], "service_name": "primary", "sentinel_password": "sentinel-secret", "db": 0}))),
)]
#[case::a_host_breaks_sentinel(
json!({"host": "ignored", "sentinel_nodes": [["sentinel-a", 26379]], "service_name": "primary"}),
@ -959,47 +1051,47 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::an_explicit_credential_provider_beats_cloud_auth(
json!({"aws_iam_auth": true, "credential_provider": true, "gcp_service_account": "account", "host": "redis", "password": "dropped"}),
&[("REDIS_AZURE_AD_TOKEN", "true")],
Ok(expected_view(json!({"auth": {"kind": "host"}, "db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "auth": {"kind": "host"}}))),
)]
#[case::gcp_beats_azure_and_aws(
json!({"aws_iam_auth": true, "azure_redis_ad_token": "True", "host": "redis", "password": "dropped"}),
&[("REDIS_GCP_SERVICE_ACCOUNT", "projects/-/serviceAccounts/sa@project.iam.gserviceaccount.com")],
Ok(expected_view(json!({"auth": {"kind": "gcp_iam", "service_account": "projects/-/serviceAccounts/sa@project.iam.gserviceaccount.com"}, "db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "auth": {"kind": "gcp_iam", "service_account": "projects/-/serviceAccounts/sa@project.iam.gserviceaccount.com"}}))),
)]
#[case::gcp_ca_certs_replace_the_configured_ones_under_ssl(
json!({"gcp_service_account": "account", "gcp_ssl_ca_certs": "/gcp-ca.pem", "host": "redis", "ssl": true, "ssl_ca_certs": "/configured-ca.pem"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "gcp_iam", "service_account": "account"}, "db": 0, "host": "redis", "mode": "host", "port": 6379, "tls": {"ca_certs": "/gcp-ca.pem", "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": "/gcp-ca.pem", "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "gcp_iam", "service_account": "account"}}))),
)]
#[case::azure_beats_aws_and_takes_the_environment_username(
json!({"aws_iam_auth": true, "azure_client_id": "client", "azure_client_secret": "secret", "azure_redis_ad_token": true, "azure_tenant_id": "tenant", "host": "redis", "username": "dropped"}),
&[("REDIS_USERNAME", "environment-user")],
Ok(expected_view(json!({"auth": {"kind": "azure_ad", "username": "environment-user"}, "db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "auth": {"kind": "azure_ad", "username": "environment-user"}}))),
)]
#[case::azure_without_a_username_sends_only_the_token(
json!({"azure_client_id": "client", "azure_client_secret": "secret", "azure_redis_ad_token": true, "azure_tenant_id": "tenant", "host": "redis"}),
&[],
Ok(expected_view(json!({"auth": {"kind": "azure_ad", "username": null}, "db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "auth": {"kind": "azure_ad", "username": null}}))),
)]
#[case::azure_needs_the_literal_true(
json!({"azure_client_id": "client", "azure_client_secret": "secret", "azure_redis_ad_token": "1", "azure_tenant_id": "tenant", "host": "redis"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::aws_iam_over_tls_with_the_default_region(
json!({"aws_iam_auth": "yes", "aws_iam_cache_name": "cache-1", "aws_iam_user_name": "iam-user", "host": "cache-1", "password": "dropped", "ssl": true}),
&[("AWS_DEFAULT_REGION", "us-east-1")],
Ok(expected_view(json!({"auth": {"cache_name": "cache-1", "kind": "aws_iam", "region": "us-east-1", "serverless": false, "user_name": "iam-user"}, "db": 0, "host": "cache-1", "mode": "host", "port": 6379, "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "cache-1", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "aws_iam", "user_name": "iam-user", "cache_name": "cache-1", "region": "us-east-1", "serverless": false}}))),
)]
#[case::aws_region_beats_the_default_region(
json!({"aws_iam_auth": true, "aws_iam_cache_name": "cache-1", "aws_iam_user_name": "iam-user", "host": "cache-1", "ssl": true}),
&[("AWS_DEFAULT_REGION", "us-east-1"), ("AWS_REGION", "eu-central-1")],
Ok(expected_view(json!({"auth": {"cache_name": "cache-1", "kind": "aws_iam", "region": "eu-central-1", "serverless": false, "user_name": "iam-user"}, "db": 0, "host": "cache-1", "mode": "host", "port": 6379, "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "cache-1", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "aws_iam", "user_name": "iam-user", "cache_name": "cache-1", "region": "eu-central-1", "serverless": false}}))),
)]
#[case::aws_iam_serverless_over_a_rediss_url_drops_its_userinfo(
json!({"aws_iam_auth": true, "aws_iam_cache_name": "cache-1", "aws_iam_region": "eu-west-1", "aws_iam_serverless": true, "aws_iam_user_name": "iam-user", "url": "rediss://user:pass@cache-1:6379"}),
&[],
Ok(expected_view(json!({"auth": {"cache_name": "cache-1", "kind": "aws_iam", "region": "eu-west-1", "serverless": true, "user_name": "iam-user"}, "mode": "url", "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}, "url": "rediss://cache-1:6379"}))),
Ok(expected_view(json!({"mode": "url", "url": "rediss://cache-1:6379", "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}, "auth": {"kind": "aws_iam", "user_name": "iam-user", "cache_name": "cache-1", "region": "eu-west-1", "serverless": true}}))),
)]
#[case::aws_iam_without_tls_fails(
json!({"aws_iam_auth": true, "aws_iam_cache_name": "cache-1", "aws_iam_region": "us-east-1", "aws_iam_user_name": "iam-user", "host": "cache-1"}),
@ -1014,22 +1106,22 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::tls_options_from_settings(
json!({"host": "redis", "ssl": true, "ssl_ca_certs": "/ca.pem", "ssl_cert_reqs": "none", "ssl_check_hostname": true}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379, "tls": {"ca_certs": "/ca.pem", "certfile": null, "check_hostname": true, "keyfile": null, "verify_certificate": false}}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": "/ca.pem", "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": false, "check_hostname": true}}))),
)]
#[case::tls_options_from_the_environment(
json!({"host": "redis"}),
&[("REDIS_SSL", "true"), ("REDIS_SSL_CA_CERTS", "/env-ca.pem"), ("REDIS_SSL_CERT_REQS", "optional")],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379, "tls": {"ca_certs": "/env-ca.pem", "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": "/env-ca.pem", "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::connect_timeout_is_separate_from_the_socket_timeout(
json!({"host": "redis", "socket_connect_timeout": 0.25, "socket_timeout": 2.5}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379, "socket_connect_timeout": 0.25, "socket_timeout": 2.5}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "socket_timeout": 2.5, "socket_connect_timeout": 0.25}))),
)]
#[case::environment_references_resolve(
json!({"host": "os.environ/COORD_HOST", "password": "os.environ/COORD_PASSWORD"}),
&[("COORD_HOST", "referenced"), ("COORD_PASSWORD", "referenced-secret")],
Ok(expected_view(json!({"auth": {"kind": "static", "password": "referenced-secret", "username": null}, "db": 0, "host": "referenced", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "referenced", "port": 6379, "db": 0, "auth": {"kind": "static", "username": null, "password": "referenced-secret"}}))),
)]
#[case::no_target_fails(
json!({"port": 6379}),
@ -1039,22 +1131,22 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::client_certificates(
json!({"host": "redis", "ssl": true, "ssl_certfile": "/client.pem", "ssl_keyfile": "/client.key"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379, "tls": {"ca_certs": null, "certfile": "/client.pem", "check_hostname": false, "keyfile": "/client.key", "verify_certificate": true}}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": "/client.pem", "keyfile": "/client.key", "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::a_rediss_url_takes_tls_options_from_its_query(
json!({"url": "rediss://cache:6380/0?ssl_cert_reqs=none&ssl_ca_certs=/ca.pem&ssl_check_hostname=yes"}),
&[],
Ok(expected_view(json!({"mode": "url", "tls": {"ca_certs": "/ca.pem", "certfile": null, "check_hostname": true, "keyfile": null, "verify_certificate": false}, "url": "rediss://cache:6380/0?ssl_cert_reqs=none&ssl_ca_certs=/ca.pem&ssl_check_hostname=yes"}))),
Ok(expected_view(json!({"mode": "url", "url": "rediss://cache:6380/0?ssl_cert_reqs=none&ssl_ca_certs=/ca.pem&ssl_check_hostname=yes", "tls": {"ca_certs": "/ca.pem", "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": false, "check_hostname": true}}))),
)]
#[case::a_url_query_setting_without_a_redis_rs_equivalent(
json!({"url": "redis://cache:6379/0?socket_timeout=2"}),
&[],
Err("Redis host does not yet support configured setting url ?socket_timeout"),
Ok(expected_view(json!({"mode": "url", "url": "redis://cache:6379/0?socket_timeout=2"}))),
)]
#[case::tls_settings_outside_a_url_never_fail_it(
json!({"db": "99999999999999999999", "ssl_cert_reqs": "strict", "url": "rediss://cache:6380"}),
&[],
Ok(expected_view(json!({"mode": "url", "tls": {"ca_certs": null, "certfile": null, "check_hostname": false, "keyfile": null, "verify_certificate": true}, "url": "rediss://cache:6380"}))),
Ok(expected_view(json!({"mode": "url", "url": "rediss://cache:6380", "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::db_is_not_read_for_a_cluster(
json!({"db": 99_999_999_999_i64, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
@ -1069,7 +1161,7 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::cluster_db_false(
json!({"db": false, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "health_check_interval": 25.0}))),
)]
#[case::cluster_db_fraction(
json!({"db": 0.5, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
@ -1079,7 +1171,7 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::cluster_db_float_zero(
json!({"db": 0.0, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "health_check_interval": 25.0}))),
)]
#[case::cluster_db_nonempty_list(
json!({"db": [1], "startup_nodes": [{"host": "node-a", "port": 7000}]}),
@ -1089,52 +1181,92 @@ fn view(resolved: &Resolution) -> serde_json::Value {
#[case::cluster_db_empty_list(
json!({"db": [], "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"]}))),
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"], "health_check_interval": 25.0}))),
)]
#[case::ssl_ciphers(
json!({"host": "redis", "ssl": true, "ssl_ciphers": "HIGH"}),
&[],
Err("Redis host does not yet support configured setting ssl_ciphers"),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::retry_on_timeout(
json!({"host": "redis", "retry_on_timeout": true}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::malformed_port(
json!({"host": "redis", "port": "abc"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::malformed_db(
json!({"db": "first", "host": "redis"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::socket_timeout_not_a_number(
json!({"host": "redis", "socket_timeout": "slow"}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::malformed_port_from_the_environment(
json!({}),
&[("REDIS_HOST", "redis"), ("REDIS_PORT", "abc")],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::protocol_3(
json!({"host": "redis", "protocol": 3}),
&[],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::retry_on_error_from_the_environment(
json!({"host": "redis"}),
&[("REDIS_RETRY_ON_ERROR", "true")],
Ok(expected_view(json!({"db": 0, "host": "redis", "mode": "host", "port": 6379}))),
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::unix_socket_path(
json!({"host": "redis", "unix_socket_path": "/tmp/redis.sock"}),
&[],
Err("Redis host does not yet support configured setting unix_socket_path"),
Ok(expected_view(json!({"mode": "unix", "path": "/tmp/redis.sock"}))),
)]
#[case::unix_socket_path_keeps_tls_settings_out_of_the_target(
json!({"host": "redis", "ssl": true, "ssl_ca_certs": "/ca.pem", "unix_socket_path": "/tmp/redis.sock"}),
&[],
Ok(expected_view(json!({"mode": "unix", "path": "/tmp/redis.sock"}))),
)]
#[case::ssl_ca_data(
json!({"host": "redis", "ssl": true, "ssl_ca_data": "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----"}),
&[],
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----", "ca_path": null, "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::ssl_ca_path_from_the_environment(
json!({"host": "redis"}),
&[("REDIS_SSL", "true"), ("REDIS_SSL_CA_PATH", "/etc/ssl/private-cas")],
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "tls": {"ca_certs": null, "ca_data": null, "ca_path": "/etc/ssl/private-cas", "certfile": null, "keyfile": null, "verify_certificate": true, "check_hostname": false}}))),
)]
#[case::health_check_interval_is_whole_seconds(
json!({"health_check_interval": "7", "host": "redis"}),
&[],
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0, "health_check_interval": 7.0}))),
)]
#[case::health_check_interval_zero_is_off(
json!({"health_check_interval": 0, "host": "redis"}),
&[],
Ok(expected_view(json!({"mode": "host", "host": "redis", "port": 6379, "db": 0}))),
)]
#[case::health_check_interval_zero_keeps_a_cluster_off(
json!({"health_check_interval": 0, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Ok(expected_view(json!({"mode": "cluster", "nodes": ["node-a:7000"]}))),
)]
#[case::a_cluster_rejects_a_nonzero_db(
json!({"db": 3, "startup_nodes": [{"host": "node-a", "port": 7000}]}),
&[],
Err("db must be 0 in cluster mode"),
)]
#[case::unix_socket_path_without_a_host_fails_like_python(
json!({"unix_socket_path": "/tmp/redis.sock"}),
&[],
Err("either host or url must be specified for redis"),
)]
fn resolves_like_python_redis(
#[case] kwargs: serde_json::Value,

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

@ -109,8 +109,12 @@ fn redis_cache_params(config: &Config) -> Option<&RedisSettings> {
fn resolve(configured: &ConfiguredRedis, lookup: &dyn Lookup) -> Result<Params, RedisError> {
let secrets = SecretSnapshot::read(configured.secret_names(), lookup);
let resolution = configured.resolve(&secrets, lookup)?;
for setting in &resolution.ignored {
tracing::warn!(%setting, "Redis setting is accepted but ignored");
for ignored in &resolution.ignored {
tracing::warn!(
"Redis setting {} is not supported by the Rust client and is ignored: {}",
ignored.setting,
ignored.consequence
);
}
Ok(resolution)
}
@ -243,7 +247,7 @@ mod tests {
use litellm_config::{Config, RedisError};
use litellm_redis_client::{Standalone, StaticAuth};
use litellm_redis_testing::RedisServer;
use litellm_redis_types::{Auth, Passthrough};
use litellm_redis_types::{Auth, HealthCheck, Passthrough};
use rstest::rstest;
use super::{Params, Plan, Target, connect, plan};
@ -261,6 +265,7 @@ mod tests {
namespace: None,
timeout: None,
connect_timeout: None,
health_check: HealthCheck::Unset,
passthrough: Passthrough::default(),
host_objects: Default::default(),
ignored: Vec::new(),

1
litellm-rust/crates/redis-client/Cargo.toml
View file

@ -6,6 +6,7 @@ license.workspace = true
repository.workspace = true
[dependencies]
tracing.workspace = true
litellm-circuit-breaker.workspace = true
litellm-secret-value.workspace = true
litellm-python-compat.workspace = true

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

@ -1,6 +1,6 @@
use std::{
future::Future,
sync::Arc,
sync::{Arc, Mutex, PoisonError},
time::{Duration, Instant},
};
@ -18,6 +18,8 @@ pub struct RedisClient {
namespace: Namespace,
breaker: Option<Breaker>,
events: Option<Arc<dyn RedisEvents>>,
health_check: Option<Duration>,
last_used: Arc<Mutex<Instant>>,
}
#[derive(Debug, PartialEq, Eq)]
@ -34,6 +36,8 @@ impl RedisClient {
namespace: config.namespace,
breaker: config.breaker,
events: config.events,
health_check: config.health_check,
last_used: Arc::new(Mutex::new(Instant::now())),
})
}
@ -194,6 +198,7 @@ impl RedisClient {
.collect();
}
};
self.check_health().await;
let results = pipeline.run(&self.connection).await;
let transport = results.iter().find_map(|result| {
result
@ -237,6 +242,7 @@ impl RedisClient {
let started = Instant::now();
let result = match self.admit() {
Ok(admission) => {
self.check_health().await;
let result = call
.await
.map_err(|error| Error::classify(error, self.connection.deployment()));
@ -250,6 +256,10 @@ impl RedisClient {
}
fn observe(&self, operation: Operation, started: Instant, outcome: Outcome) {
*self
.last_used
.lock()
.unwrap_or_else(PoisonError::into_inner) = Instant::now();
if let Some(events) = &self.events {
events.call(&CallEvent {
operation,
@ -259,6 +269,26 @@ impl RedisClient {
}
}
/// redis-py pings a connection that sat idle longer than `health_check_interval` before
/// sending the next command on it, and reconnects when the ping fails. The ping is not a
/// call of the caller's and never reaches the breaker or the events.
async fn check_health(&self) {
let Some(interval) = self.health_check else {
return;
};
let idle = self
.last_used
.lock()
.unwrap_or_else(PoisonError::into_inner)
.elapsed();
if idle < interval {
return;
}
if let Err(error) = self.connection.ping_every_node().await {
tracing::debug!(%error, "idle Redis connection failed its health check, the next command reconnects");
}
}
fn admit(&self) -> Result<Option<Admission>, Error> {
self.breaker
.as_ref()

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

@ -9,7 +9,7 @@ use litellm_circuit_breaker::Breaker;
use litellm_secret_value::SecretValue;
use crate::{ConnectError, Namespace, RedisEvents};
use litellm_redis_types::{Cluster, Target, TlsOptions};
use litellm_redis_types::{Cluster, RedisUrl, Target, TlsOptions};
// litellm/constants.py REDIS_SOCKET_TIMEOUT, which _coerce_redis_kwargs_types sets when socket_timeout is unset
pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(5);
@ -21,6 +21,10 @@ pub struct RedisConfig {
pub timeouts: Timeouts,
pub breaker: Option<Breaker>,
pub events: Option<Arc<dyn RedisEvents>>,
/// Ping every node before the next command once the client sat idle this long, like
/// redis-py's `health_check_interval`, so a connection the server dropped silently is
/// replaced before a real command pays for it.
pub health_check: Option<Duration>,
}
#[derive(Clone, Copy, Debug, PartialEq)]
@ -72,17 +76,44 @@ pub(crate) fn tls_certificates(tls: &TlsOptions) -> Result<TlsCertificates, Conn
}),
(None, _) => None,
};
let from_file: Option<Vec<u8>> = tls.ca_certs.as_ref().map(read).transpose()?;
let from_directory: Vec<Vec<u8>> = tls
.ca_path
.as_ref()
.map(read_directory)
.transpose()?
.unwrap_or_default();
let configured: Vec<Vec<u8>> = from_file
.into_iter()
.chain(tls.ca_data.iter().map(|pem| pem.clone().into_bytes()))
.chain(from_directory)
.collect();
Ok(TlsCertificates {
client_tls,
root_cert: tls
.ca_certs
.as_ref()
.map(read)
.transpose()?
.map(with_system_roots),
root_cert: (!configured.is_empty()).then(|| with_system_roots(configured.join(&b'\n'))),
})
}
/// Every regular file in `path`, in name order, as redis-py hands `ssl_ca_path` to
/// `load_verify_locations(capath=...)`
fn read_directory(path: &PathBuf) -> Result<Vec<Vec<u8>>, ConnectError> {
let tls_file = |source| ConnectError::TlsFile {
path: path.display().to_string(),
source,
};
let mut files: Vec<PathBuf> = std::fs::read_dir(path)
.map_err(tls_file)?
.map(|entry| entry.map(|entry| entry.path()))
.collect::<Result<_, _>>()
.map_err(tls_file)?;
files.sort();
files
.iter()
.filter(|file| file.is_file())
.map(read)
.collect()
}
pub(crate) fn apply_tls(
tls: &TlsOptions,
info: ConnectionInfo,
@ -152,12 +183,17 @@ pub(crate) fn connection_info(
credentials: &Credentials,
) -> Result<ConnectionInfo, ConnectError> {
let (info, tls) = match target {
Target::Url(url) => (url.url.expose().into_connection_info()?, url.tls.as_ref()),
Target::Url(url) => (url_info(url)?, url.tls.as_ref()),
Target::Standalone(host) => {
let info = addr(&host.host, host.port, host.tls.as_ref()).into_connection_info()?;
let settings = info.redis_settings().clone().set_db(host.db);
(info.set_redis_settings(settings), host.tls.as_ref())
}
Target::Unix(socket) => {
let info = ConnectionAddr::Unix(socket.path.clone()).into_connection_info()?;
let settings = info.redis_settings().clone().set_db(socket.db);
(info.set_redis_settings(settings), None)
}
Target::Cluster(_) | Target::Sentinel(_) => {
unreachable!("cluster and sentinel endpoints build their own connection info")
}
@ -170,6 +206,24 @@ pub(crate) fn connection_info(
}
}
// parse_url reads ?db= over the path, and the url's userinfo over ?username= and ?password=
fn url_info(url: &RedisUrl) -> Result<ConnectionInfo, ConnectError> {
let info = url.url.expose().into_connection_info()?;
let settings = info.redis_settings().clone();
let settings = match url.db {
Some(db) => settings.set_db(db),
None => settings,
};
let query = Credentials {
username: url.username.clone(),
password: url.password.clone(),
};
Ok(with_static_credentials(
info.set_redis_settings(settings),
&query,
))
}
fn with_static_credentials(info: ConnectionInfo, credentials: &Credentials) -> ConnectionInfo {
let Credentials { username, password } = credentials;
let settings = info.redis_settings().clone();
@ -183,3 +237,69 @@ fn with_static_credentials(info: ConnectionInfo, credentials: &Credentials) -> C
};
info.set_redis_settings(settings)
}
#[cfg(test)]
mod tests {
use litellm_redis_types::{RedisUrl, UnixSocket};
use rstest::rstest;
use super::*;
#[test]
fn a_unix_socket_replaces_host_port_and_tls_like_redis_py() {
let target = Target::Unix(UnixSocket {
path: "/tmp/redis.sock".into(),
db: 3,
});
let info = connection_info(&target, &Credentials::NONE).unwrap();
assert_eq!(*info.addr(), ConnectionAddr::Unix("/tmp/redis.sock".into()));
assert_eq!(info.redis_settings().db(), 3);
}
#[rstest]
#[case::the_query_db_wins_over_the_path("redis://cache:6379/1", Some(3), None, 3, None)]
#[case::the_path_db_without_a_query_db("redis://cache:6379/1", None, None, 1, None)]
#[case::the_query_password_fills_a_missing_userinfo(
"redis://cache:6379",
None,
Some("query"),
0,
Some("query")
)]
#[case::the_userinfo_wins_over_the_query_password(
"redis://:userinfo@cache:6379",
None,
Some("query"),
0,
Some("userinfo")
)]
fn a_url_reads_its_query_like_parse_url(
#[case] url: &str,
#[case] db: Option<i64>,
#[case] password: Option<&str>,
#[case] expected_db: i64,
#[case] expected_password: Option<&str>,
) {
let target = Target::Url(RedisUrl {
url: SecretValue::new(url),
tls: None,
db,
username: None,
password: password.map(SecretValue::new),
});
let static_password = Credentials {
username: None,
password: Some(SecretValue::new("static")),
};
let info = connection_info(&target, &static_password).unwrap();
assert_eq!(info.redis_settings().db(), expected_db);
assert_eq!(
info.redis_settings().password(),
expected_password.or(Some("static"))
);
}
}

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

@ -12,6 +12,7 @@ use redis::{
aio::{ConnectionLike, ConnectionManager, ConnectionManagerConfig, MultiplexedConnection},
cluster::{ClusterClientBuilder, ClusterConfig},
cluster_async::ClusterConnection,
cluster_routing::{MultipleNodeRoutingInfo, ResponsePolicy, RoutingInfo},
io::tcp::TcpSettings,
sentinel::{SentinelClient, SentinelClientBuilder, SentinelServerType},
};
@ -44,7 +45,7 @@ impl Connection {
timeouts: Timeouts,
) -> Result<Self, ConnectError> {
match target {
Target::Url(_) | Target::Standalone(_) => {
Target::Url(_) | Target::Standalone(_) | Target::Unix(_) => {
Self::standalone(config::connection_info(target, credentials)?, timeouts)
}
Target::Cluster(cluster) => Self::cluster(cluster, credentials, timeouts),
@ -117,6 +118,30 @@ impl Connection {
}
}
impl Connection {
/// `PING` on the one multiplexed connection, or on every cluster node, so a dead socket is
/// found and replaced by redis-rs before a real command uses it.
pub(crate) async fn ping_every_node(&self) -> RedisResult<()> {
let ping = redis::cmd("PING");
match self {
Self::Cluster(connection) => connection
.clone()
.route_command(
ping,
RoutingInfo::MultiNode((
MultipleNodeRoutingInfo::AllNodes,
Some(ResponsePolicy::AllSucceeded),
)),
)
.await
.map(drop),
Self::Standalone(_) | Self::Sentinel(_) => {
ping.query_async::<Value>(&mut self.clone()).await.map(drop)
}
}
}
}
impl ConnectionLike for Connection {
fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {
match self {

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

@ -14,7 +14,7 @@ pub use config::{Credentials, DEFAULT_TIMEOUT, RedisConfig, Timeouts};
pub use error::{ConnectError, Error};
pub use events::{CallEvent, FailureKind, Operation, Outcome, RedisEvents};
pub use litellm_redis_types::{
Cluster, Node, RedisUrl, Sentinel, Standalone, StaticAuth, Target, TlsOptions,
Cluster, Node, RedisUrl, Sentinel, Standalone, StaticAuth, Target, TlsOptions, UnixSocket,
};
pub use namespace::Namespace;
pub use pipeline::Pipeline;

99
litellm-rust/crates/redis-client/tests/client.rs
View file

@ -5,7 +5,7 @@ use std::{
time::{Duration, Instant},
};
use litellm_redis_testing::{RedisDeployment, RedisServer, Topology, eventually};
use litellm_redis_testing::{RedisCluster, RedisDeployment, RedisServer, Topology, eventually};
use rstest::rstest;
use litellm_circuit_breaker::{Breaker, Config, State};
@ -229,6 +229,7 @@ async fn outage_events_report_the_failure_kind(
breaker: None,
events: Some(recorded.clone()),
credentials: Credentials::NONE,
health_check: None,
})
.unwrap();
client.ping().await.unwrap();
@ -283,6 +284,7 @@ async fn every_call_reports_its_operation_and_outcome(
timeouts: Timeouts::default(),
breaker: None,
events: Some(recorded.clone()),
health_check: None,
})
.unwrap();
client
@ -743,12 +745,16 @@ async fn a_host_endpoint_connects_to_its_database() {
timeouts: Timeouts::default(),
breaker: None,
events: None,
health_check: None,
})
};
let url = |url: String| {
Target::Url(RedisUrl {
url: SecretValue::new(url),
tls: None,
db: None,
username: None,
password: None,
})
};
let in_db_3 = connect(Target::Standalone(Standalone {
@ -773,6 +779,97 @@ async fn a_host_endpoint_connects_to_its_database() {
);
}
async fn ping_calls(cluster: &RedisCluster) -> Vec<u64> {
cluster
.cli_outputs(&["INFO", "commandstats"])
.await
.unwrap()
.iter()
.map(|info| {
info.lines()
.find_map(|line| line.strip_prefix("cmdstat_ping:calls="))
.and_then(|rest| rest.split(',').next())
.and_then(|calls| calls.trim().parse().ok())
.unwrap_or(0)
})
.collect()
}
fn health_checked(
server: &RedisDeployment,
interval: Duration,
timeout: Duration,
breaker: Option<Breaker>,
events: Option<Arc<Recorded>>,
) -> RedisClient {
RedisClient::new(RedisConfig {
target: endpoint(server),
credentials: Credentials::NONE,
namespace: Namespace::default(),
timeouts: Timeouts::new(timeout, None),
breaker,
events: events.map(|events| events as Arc<dyn RedisEvents>),
health_check: Some(interval),
})
.unwrap()
}
// redis-py 5.3.1 redis/asyncio/connection.py check_health: a connection idle longer than
// health_check_interval gets a PING before the next command, a busy one does not
#[rstest]
#[tokio::test]
async fn an_idle_client_pings_every_node_before_the_next_command_like_redis_py() {
let server = RedisDeployment::start(Topology::Cluster).await.unwrap();
let RedisDeployment::Cluster(cluster) = &server else {
unreachable!("started a cluster")
};
let client = health_checked(&server, Duration::from_secs(3), DEFAULT_TIMEOUT, None, None);
client.set("k", "1", Duration::from_secs(60)).await.unwrap();
let before = ping_calls(cluster).await;
client.set("k", "2", Duration::from_secs(60)).await.unwrap();
let busy = ping_calls(cluster).await;
tokio::time::sleep(Duration::from_millis(3500)).await;
client.set("k", "3", Duration::from_secs(60)).await.unwrap();
let idle = ping_calls(cluster).await;
assert_eq!(busy, before);
assert_eq!(
idle,
before.iter().map(|calls| calls + 1).collect::<Vec<_>>(),
"before {before:?}"
);
}
#[rstest]
#[tokio::test]
async fn a_failed_health_check_is_neither_a_call_nor_a_breaker_failure() {
let server = RedisDeployment::start(Topology::Cluster).await.unwrap();
let recorded = Arc::new(Recorded::default());
let breaker = breaker(2);
let client = health_checked(
&server,
Duration::from_millis(200),
Duration::from_millis(300),
Some(breaker.clone()),
Some(recorded.clone()),
);
client.set("k", "1", Duration::from_secs(60)).await.unwrap();
server.freeze().await.unwrap();
tokio::time::sleep(Duration::from_millis(300)).await;
let failed = client.get::<String>("k").await;
server.thaw().await.unwrap();
assert!(matches!(failed, Err(Error::Transport(_))), "{failed:?}");
assert!(
matches!(breaker.state(), State::Closed { timeouts: 1, .. }),
"{:?}",
breaker.state()
);
assert_eq!(recorded.0.lock().unwrap().len(), 2);
}
#[rstest]
#[case::int(Amount::Int(1), "1")]
#[case::float_keeps_its_point(Amount::Float(1.0), "1.0")]

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

@ -4,7 +4,7 @@ use litellm_redis_client::{Credentials, RedisUrl, Sentinel, StaticAuth, Target,
use litellm_secret_value::SecretValue;
#[rstest]
#[case::url(format!("{:?}", Target::Url(RedisUrl { url: SecretValue::new("redis://user:credential-value@cache:6379/0"), tls: None })))]
#[case::url(format!("{:?}", Target::Url(RedisUrl { url: SecretValue::new("redis://user:credential-value@cache:6379/0"), tls: None, db: None, username: None, password: None })))]
#[case::static_password(format!("{:?}", Credentials { username: None, password: Some(SecretValue::new("credential-value")) }))]
#[case::sentinel_password(format!("{:?}", Target::Sentinel(Sentinel {
monitors: Vec::new(),
@ -21,9 +21,18 @@ fn debug_output_never_shows_the_secret(#[case] debug: String) {
#[derive(Clone, Copy, Debug)]
enum Ca {
Trusted,
TrustedData,
TrustedDirectory,
SystemRoots,
}
fn directory_holding(ca: &std::path::Path) -> std::path::PathBuf {
let directory = std::env::temp_dir().join(format!("litellm-ca-dir-{}", std::process::id()));
std::fs::create_dir_all(&directory).unwrap();
std::fs::copy(ca, directory.join("ca.pem")).unwrap();
directory
}
async fn ping_over_tls(
require_client_certificate: bool,
host: &str,
@ -41,6 +50,8 @@ async fn ping_over_tls(
let files = server.files();
let tls = TlsOptions {
ca_certs: matches!(ca, Ca::Trusted).then(|| files.ca.clone()),
ca_data: matches!(ca, Ca::TrustedData).then(|| std::fs::read_to_string(&files.ca).unwrap()),
ca_path: matches!(ca, Ca::TrustedDirectory).then(|| directory_holding(&files.ca)),
certfile: client_certificate.then(|| files.client_cert.clone()),
keyfile: client_certificate.then(|| files.client_key.clone()),
verify_certificate,
@ -58,6 +69,7 @@ async fn ping_over_tls(
timeouts: litellm_redis_client::Timeouts::default(),
breaker: None,
events: None,
health_check: None,
})
.unwrap();
client.ping().await
@ -71,6 +83,14 @@ async fn ping_over_tls(
false,
true
)]
#[case::trusted_ca_as_pem_text_like_ssl_ca_data("127.0.0.1", Ca::TrustedData, true, false, true)]
#[case::trusted_ca_in_a_directory_like_ssl_ca_path(
"127.0.0.1",
Ca::TrustedDirectory,
true,
false,
true
)]
#[case::hostname_check_with_a_matching_name("localhost", Ca::Trusted, true, true, true)]
#[case::hostname_check_against_an_unlisted_address("127.0.0.1", Ca::Trusted, true, true, false)]
#[case::an_untrusted_ca_is_refused("127.0.0.1", Ca::SystemRoots, true, false, false)]
@ -134,6 +154,7 @@ fn a_missing_tls_file_names_the_path() {
timeouts: litellm_redis_client::Timeouts::default(),
breaker: None,
events: None,
health_check: None,
})
.err()
.unwrap();

2
litellm-rust/crates/redis-client/tests/connection.rs
View file

@ -31,6 +31,7 @@ async fn an_out_of_range_database_is_a_server_rejection() {
timeouts: Timeouts::new(Duration::from_secs(2), None),
breaker: None,
events: None,
health_check: None,
})
.unwrap();
let result = client.ping().await;
@ -79,6 +80,7 @@ fn sentinel_client(sentinel: &RedisSentinel) -> RedisClient {
timeouts: Timeouts::new(Duration::from_secs(1), None),
breaker: None,
events: None,
health_check: None,
})
.unwrap()
}

5
litellm-rust/crates/redis-client/tests/support/mod.rs
View file

@ -17,6 +17,9 @@ pub fn endpoint(deployment: &RedisDeployment) -> Target {
RedisDeployment::Standalone(server) => Target::Url(RedisUrl {
url: SecretValue::new(server.url()),
tls: None,
db: None,
username: None,
password: None,
}),
RedisDeployment::Cluster(cluster) => Target::Cluster(Cluster {
nodes: cluster
@ -48,6 +51,7 @@ pub fn connect_guarded(
timeouts: Timeouts::new(timeout, None),
breaker,
events: None,
health_check: None,
})
.unwrap()
}
@ -60,6 +64,7 @@ pub fn connect_as(deployment: &RedisDeployment, credentials: Credentials) -> Red
timeouts: Timeouts::new(Duration::from_secs(2), None),
breaker: None,
events: None,
health_check: None,
})
.unwrap()
}

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

@ -1,13 +1,19 @@
use std::time::Duration;
use litellm_circuit_breaker::Breaker;
use litellm_redis_client::{
Credentials, DEFAULT_TIMEOUT, Namespace, RedisClient, RedisConfig, Timeouts,
Credentials, DEFAULT_TIMEOUT, Namespace, RedisClient, RedisConfig, Target, Timeouts,
};
use litellm_redis_types::{Auth, Resolution, StaticAuth};
use litellm_redis_types::{Auth, HealthCheck, Resolution, StaticAuth};
use crate::Error;
// litellm/constants.py REDIS_CLUSTER_HEALTH_CHECK_INTERVAL, the default _redis.py sets for cluster clients only
const CLUSTER_HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(25);
pub fn connect(resolved: Resolution, breaker: Option<Breaker>) -> Result<RedisClient, Error> {
tracing::info!(target = ?resolved.target, namespace = ?resolved.namespace, "redis client configured");
let health_check = health_check(&resolved.target, resolved.health_check);
Ok(RedisClient::new(RedisConfig {
target: resolved.target,
credentials: credentials(resolved.auth)?,
@ -18,9 +24,19 @@ pub fn connect(resolved: Resolution, breaker: Option<Breaker>) -> Result<RedisCl
),
breaker,
events: None,
health_check,
})?)
}
fn health_check(target: &Target, configured: HealthCheck) -> Option<Duration> {
match (target, configured) {
(_, HealthCheck::Every(interval)) => Some(interval),
(_, HealthCheck::Disabled) => None,
(Target::Cluster(_), HealthCheck::Unset) => Some(CLUSTER_HEALTH_CHECK_INTERVAL),
(_, HealthCheck::Unset) => None,
}
}
fn credentials(auth: Auth) -> Result<Credentials, Error> {
match auth {
Auth::Static(StaticAuth { username, password }) => Ok(Credentials { username, password }),
@ -36,3 +52,56 @@ fn credentials(auth: Auth) -> Result<Credentials, Error> {
}),
}
}
#[cfg(test)]
mod tests {
use litellm_redis_client::{Cluster, Node, Standalone};
use rstest::rstest;
use super::*;
fn cluster() -> Target {
Target::Cluster(Cluster {
nodes: vec![Node {
host: "node".to_owned(),
port: 7000,
}],
tls: None,
})
}
fn standalone() -> Target {
Target::Standalone(Standalone {
host: "redis".to_owned(),
port: 6379,
tls: None,
db: 0,
})
}
#[rstest]
#[case::an_unset_cluster_takes_the_python_default(
cluster(),
HealthCheck::Unset,
Some(CLUSTER_HEALTH_CHECK_INTERVAL)
)]
#[case::zero_keeps_a_cluster_off(cluster(), HealthCheck::Disabled, None)]
#[case::an_unset_standalone_stays_off(standalone(), HealthCheck::Unset, None)]
#[case::a_configured_interval_wins(
cluster(),
HealthCheck::Every(Duration::from_secs(5)),
Some(Duration::from_secs(5))
)]
#[case::a_negative_interval_checks_every_command(
standalone(),
HealthCheck::Every(Duration::ZERO),
Some(Duration::ZERO)
)]
fn health_checks_follow_redis_py(
#[case] target: Target,
#[case] configured: HealthCheck,
#[case] expected: Option<Duration>,
) {
assert_eq!(health_check(&target, configured), expected);
}
}

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

@ -71,6 +71,15 @@ impl RedisCluster {
self.ports.iter().map(|&port| (HOST, port))
}
/// One redis-cli reply per node, in `nodes()` order.
pub async fn cli_outputs(&self, args: &[&str]) -> Result<Vec<String>, Error> {
let mut replies = Vec::with_capacity(self.ports.len());
for port in self.ports {
replies.push(crate::exec::run(&self.container, crate::exec::cli(port, args)).await?);
}
Ok(replies)
}
pub async fn cli_on_every_node(&self, args: &[&str]) -> Result<(), Error> {
for port in self.ports {
crate::exec::run(&self.container, crate::exec::cli(port, args)).await?;

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

@ -16,19 +16,37 @@ pub struct Resolution {
pub namespace: Option<String>,
pub timeout: Option<Duration>,
pub connect_timeout: Option<Duration>,
pub health_check: HealthCheck,
/// Plain settings the resolver does not model, kept verbatim for the redis-py projection.
pub passthrough: Passthrough,
/// Settings whose value is a host object, like redis-py's `retry` or `connection_class`, which
/// only the host can reattach.
pub host_objects: BTreeSet<String>,
/// Where each setting the Rust client accepts without honoring came from: its key, or the
/// environment variable that set it.
pub ignored: Vec<String>,
pub ignored: Vec<Ignored>,
}
/// redis-py's `health_check_interval`, which `_coerce_redis_kwargs_types` reads as an `int` of
/// seconds. Unset and zero differ, because a client may give an unset interval its own default.
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum HealthCheck {
Unset,
Disabled,
Every(Duration),
}
/// A setting the Rust client accepts without honoring, so the host can log what ignoring it means.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Ignored {
/// The key, the environment variable or the url query parameter that set it.
pub setting: String,
pub consequence: &'static str,
}
#[derive(Clone, Debug, PartialEq)]
pub enum Target {
Standalone(Standalone),
/// Like redis-py's `UnixDomainSocketConnection`, which replaces host, port and TLS.
Unix(UnixSocket),
Url(RedisUrl),
Cluster(Cluster),
Sentinel(Sentinel),
@ -42,11 +60,24 @@ pub struct Standalone {
pub db: i64,
}
#[derive(Clone, Debug, PartialEq)]
pub struct UnixSocket {
pub path: PathBuf,
pub db: i64,
}
/// A url kept verbatim, plus what redis-py's `parse_url` reads from its query that the url's own
/// grammar does not carry.
#[derive(Clone, Debug, PartialEq)]
pub struct RedisUrl {
pub url: SecretValue,
/// Set only for a `rediss://` url, from its query parameters.
pub tls: Option<TlsOptions>,
/// `?db=`, which wins over the path.
pub db: Option<i64>,
/// `?username=` and `?password=`, which the url's userinfo wins over.
pub username: Option<SecretValue>,
pub password: Option<SecretValue>,
}
#[derive(Clone, Debug, PartialEq)]
@ -74,6 +105,10 @@ pub struct Sentinel {
#[derive(Clone, Debug, PartialEq)]
pub struct TlsOptions {
pub ca_certs: Option<PathBuf>,
/// PEM text trusted alongside `ca_certs`, redis-py's `ssl_ca_data`
pub ca_data: Option<String>,
/// A directory whose files are PEM certificates to trust, redis-py's `ssl_ca_path`
pub ca_path: Option<PathBuf>,
pub certfile: Option<PathBuf>,
pub keyfile: Option<PathBuf>,
pub verify_certificate: bool,
@ -84,6 +119,8 @@ impl TlsOptions {
pub fn verified() -> Self {
Self {
ca_certs: None,
ca_data: None,
ca_path: None,
certfile: None,
keyfile: None,
verify_certificate: true,

166
tests/unit/test_redis.py
View file

@ -1972,6 +1972,8 @@ def _tls_view(connection: AbstractConnection) -> dict[str, object] | None:
context: Final = connection.ssl_context
return {
"ca_certs": context.ca_certs,
"ca_data": context.ca_data,
"ca_path": None,
"certfile": context.certfile,
"keyfile": context.keyfile,
"verify_certificate": context.cert_reqs != ssl.CERT_NONE,
@ -2006,6 +2008,7 @@ def _connection_view(connection: AbstractConnection, kwargs: dict[str, object])
"auth": _auth_view(kwargs),
"socket_timeout": connection.socket_timeout,
"socket_connect_timeout": connection.socket_connect_timeout,
"health_check_interval": float(connection.health_check_interval) or None,
}
@ -2047,6 +2050,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"tls": None,
"socket_timeout": 5.0,
"socket_connect_timeout": 5.0,
"health_check_interval": None,
**fields,
}
@ -2064,31 +2068,10 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
_expected_view({"db": 0, "host": "redis", "mode": "host", "port": 6379}),
id="host_defaults",
),
pytest.param(
{"host": "redis", "port": "6380\n", "db": " 2 ", "socket_timeout": "1_0.5\n"},
{},
_expected_view(
{
"db": 2,
"host": "redis",
"mode": "host",
"port": 6380,
"socket_timeout": 10.5,
"socket_connect_timeout": 10.5,
}
),
id="padded_and_underscored_numbers_parse_like_python_int_and_float",
),
pytest.param(
{},
{"REDIS_HOST": "redis", "REDIS_PORT": "6_380", "REDIS_DB": "3\n"},
_expected_view({"db": 3, "host": "redis", "mode": "host", "port": 6380}),
id="padded_environment_numbers_parse_like_python_int",
),
pytest.param(
{"startup_nodes": []},
{"REDIS_CLUSTER_NODES": '[{"host": "node-a", "port": 7000}]'},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000"]}),
_expected_view({"health_check_interval": 25.0, "mode": "cluster", "nodes": ["node-a:7000"]}),
id="empty_cluster_nodes_use_the_environment",
),
pytest.param(
@ -2175,6 +2158,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"socket_timeout": 1.5,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2201,6 +2186,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6390,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2225,6 +2212,27 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
),
id="string_numbers_are_coerced",
),
pytest.param(
{"host": "redis", "port": "6380\n", "db": " 2 ", "socket_timeout": "1_0.5\n"},
{},
_expected_view(
{
"db": 2,
"host": "redis",
"mode": "host",
"port": 6380,
"socket_timeout": 10.5,
"socket_connect_timeout": 10.5,
}
),
id="padded_and_underscored_numbers_parse_like_python_int_and_float",
),
pytest.param(
{},
{"REDIS_HOST": "redis", "REDIS_PORT": "6_380", "REDIS_DB": "3\n"},
_expected_view({"db": 3, "host": "redis", "mode": "host", "port": 6380}),
id="padded_environment_numbers_parse_like_python_int",
),
pytest.param(
{"host": "redis"},
{"REDIS_SOCKET_TIMEOUT": "0.75"},
@ -2267,6 +2275,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"mode": "url",
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2290,6 +2300,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"mode": "url",
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2310,6 +2322,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
_expected_view(
{
"auth": {"kind": "static", "password": "cluster-secret", "username": None},
"health_check_interval": 25.0,
"mode": "cluster",
"nodes": ["node-a:7000"],
}
@ -2319,7 +2332,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
pytest.param(
{},
{"REDIS_CLUSTER_NODES": '[{"host": "node-a", "port": "7000"}, {"host": "node-b", "port": 7001}]'},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000", "node-b:7001"]}),
_expected_view({"health_check_interval": 25.0, "mode": "cluster", "nodes": ["node-a:7000", "node-b:7001"]}),
id="environment_cluster_nodes",
),
pytest.param(
@ -2332,10 +2345,13 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
{},
_expected_view(
{
"health_check_interval": 25.0,
"mode": "cluster",
"nodes": ["node-a:7000"],
"tls": {
"ca_certs": "/ca.pem",
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": True,
"keyfile": None,
@ -2447,6 +2463,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": "/gcp-ca.pem",
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2529,6 +2547,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2562,6 +2582,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2593,6 +2615,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"mode": "url",
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2649,6 +2673,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": "/ca.pem",
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": True,
"keyfile": None,
@ -2669,6 +2695,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": "/env-ca.pem",
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2724,6 +2752,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": "/client.pem",
"check_hostname": False,
"keyfile": "/client.key",
@ -2741,6 +2771,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"mode": "url",
"tls": {
"ca_certs": "/ca.pem",
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": True,
"keyfile": None,
@ -2772,6 +2804,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"mode": "url",
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2807,7 +2841,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
pytest.param(
{"db": False, "startup_nodes": [{"host": "node-a", "port": 7000}]},
{},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000"]}),
_expected_view({"health_check_interval": 25.0, "mode": "cluster", "nodes": ["node-a:7000"]}),
id="cluster_db_false",
),
pytest.param(
@ -2824,7 +2858,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
pytest.param(
{"db": 0.0, "startup_nodes": [{"host": "node-a", "port": 7000}]},
{},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000"]}),
_expected_view({"health_check_interval": 25.0, "mode": "cluster", "nodes": ["node-a:7000"]}),
id="cluster_db_float_zero",
),
pytest.param(
@ -2841,7 +2875,7 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
pytest.param(
{"db": [], "startup_nodes": [{"host": "node-a", "port": 7000}]},
{},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000"]}),
_expected_view({"health_check_interval": 25.0, "mode": "cluster", "nodes": ["node-a:7000"]}),
id="cluster_db_empty_list",
),
pytest.param(
@ -2855,6 +2889,8 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
"port": 6379,
"tls": {
"ca_certs": None,
"ca_data": None,
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
@ -2912,6 +2948,84 @@ def _expected_view(fields: dict[str, object]) -> dict[str, object]:
_expected_view({"mode": "unix", "path": "/tmp/redis.sock"}),
id="unix_socket_path",
),
pytest.param(
{"host": "redis", "ssl": True, "ssl_ca_certs": "/ca.pem", "unix_socket_path": "/tmp/redis.sock"},
{},
_expected_view({"mode": "unix", "path": "/tmp/redis.sock"}),
id="unix_socket_path_keeps_tls_settings_out_of_the_target",
),
pytest.param(
{
"host": "redis",
"ssl": True,
"ssl_ca_data": "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----",
},
{},
_expected_view(
{
"db": 0,
"host": "redis",
"mode": "host",
"port": 6379,
"tls": {
"ca_certs": None,
"ca_data": "-----BEGIN CERTIFICATE-----\nMIIB\n-----END CERTIFICATE-----",
"ca_path": None,
"certfile": None,
"check_hostname": False,
"keyfile": None,
"verify_certificate": True,
},
}
),
id="ssl_ca_data",
),
pytest.param(
{"host": "redis"},
{"REDIS_SSL": "true", "REDIS_SSL_CA_PATH": "/etc/ssl/private-cas"},
{
"rejected": {
"message": "Redis.__init__() got an unexpected keyword argument 'ssl_ca_path'",
"type": "TypeError",
}
},
id="ssl_ca_path_from_the_environment",
),
pytest.param(
{"health_check_interval": "7", "host": "redis"},
{},
_expected_view({"db": 0, "health_check_interval": 7.0, "host": "redis", "mode": "host", "port": 6379}),
id="health_check_interval_is_whole_seconds",
),
pytest.param(
{"health_check_interval": 0, "host": "redis"},
{},
_expected_view({"db": 0, "host": "redis", "mode": "host", "port": 6379}),
id="health_check_interval_zero_is_off",
),
pytest.param(
{"health_check_interval": 0, "startup_nodes": [{"host": "node-a", "port": 7000}]},
{},
_expected_view({"mode": "cluster", "nodes": ["node-a:7000"]}),
id="health_check_interval_zero_keeps_a_cluster_off",
),
pytest.param(
{"db": 3, "startup_nodes": [{"host": "node-a", "port": 7000}]},
{},
{
"rejected": {
"message": "Argument 'db' must be 0 or None in cluster mode",
"type": "RedisClusterException",
}
},
id="a_cluster_rejects_a_nonzero_db",
),
pytest.param(
{"unix_socket_path": "/tmp/redis.sock"},
{},
{"rejected": {"message": "Either 'host' or 'url' must be specified for redis.", "type": "ValueError"}},
id="unix_socket_path_without_a_host_fails_like_python",
),
],
)
def test_redis_settings_resolve_like_the_rust_resolver_expects(monkeypatch, kwargs, environment, expected):