mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-13 08:52:02 +00:00
feat(redis-client): stop rejecting redis-py settings and support health checks, Unix sockets and CA bundles
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 (with Python's 25 second cluster default), unix_socket_path, ssl_ca_data and ssl_ca_path are read and applied like redis-py.
This commit is contained in:
parent
541bc2037c
commit
5154104d7b
20 changed files with 651 additions and 179 deletions
1
litellm-rust/Cargo.lock
generated
1
litellm-rust/Cargo.lock
generated
|
|
@ -4433,6 +4433,7 @@ dependencies = [
|
|||
"socket2 0.6.5",
|
||||
"thiserror 2.0.19",
|
||||
"tokio",
|
||||
"tracing",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
## Redis settings
|
||||
|
||||
- `RedisSettings::resolve` follows `litellm/_redis.py` `_get_redis_client_logic` plus `_credential_provider_auth_kwargs` and the per-client allowlists of `get_redis_async_client` (redis-py 5.3.1). Python and Rust pods share one Redis during the switch, so a change there is a change to a wire contract
|
||||
- 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 `RedisSettings::secret_names` lists, read the way Python's `get_secret` reads it (secret manager, then environment). The host fills it once before resolving, python-bridge through `get_secret` and rust-proxy from its environment, so the resolver never calls back into the host. The environment lookup is only for what Python reads straight from `os.environ` (`REDIS_USERNAME` for Azure). Reading a name the snapshot does not list fails a debug assertion
|
||||
- Every Redis setting Python reads is one row in `SETTINGS` in `src/redis/keys.rs`: how Rust handles it (read, ignore, warn, reject), whether `_get_redis_env_kwarg_mapping` reads it as `REDIS_<KWARG>`, and the names `_get_redis_client_logic` falls back to. The env layer, the rejections, the warnings and `secret_names` all come from that table, so a new setting is one row. It is the shape the `#[derive(Layered)]` attributes in `litellm-layered` would generate, and moves onto them once that crate lands
|
||||
- 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 warnings and `secret_names` all come from that table, so a new setting is one row. It is the shape the `#[derive(Layered)]` attributes in `litellm-layered` would generate, and moves onto them once that crate lands
|
||||
- Lookups go through `litellm_core_utils::settings::Lookup`, never a closure type of our own
|
||||
- A url carries its own TLS. redis-py's url allowlist is the plain `Connection` signature, so every `ssl_*` setting outside the url is dropped, and a credential provider strips the url's userinfo. A `rediss://` url's `ssl_cert_reqs`, `ssl_ca_certs`, `ssl_certfile`, `ssl_keyfile` and `ssl_check_hostname` query parameters are read like `parse_url` reads them
|
||||
- TLS options and `db` are only read for the endpoint that uses them, so a bad `ssl_cert_reqs` never fails a url target. A cluster with a non-zero `db`, and Sentinel with a `host` or `port`, are errors because redis-py raises building those clients
|
||||
|
|
@ -11,9 +11,10 @@
|
|||
- Precedence matches Python. The target is cluster, then URL, then Sentinel (`sentinel_nodes` plus `service_name`), then host. Auth is an explicit `credential_provider`, then GCP, then Azure, then AWS, and the first one wins with a warning
|
||||
- Numbers are read with `litellm_python_compat::number`, so whitespace and digit underscores parse like Python's `int()` and `float()`. A value they cannot parse (`port: abc`, `socket_timeout: slow`) is dropped with a warning and the default is used, like `_coerce_redis_kwargs_types`. A number that parses but is out of range (negative seconds, a port above 65535) fails with `RedisError::Invalid`
|
||||
- `validate_coordination` mirrors the proxy's `CoordinationRedisParams` pydantic model, which rejects a malformed `port` or `ssl` in `general_settings.coordination_redis` before redis-py coerces anything. Hosts call it for that block only
|
||||
- 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 with a warning that names 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`, `ssl_password`, the OCSP options, `connection_pool`, `event_dispatcher`, `redis_connect_func` and `retry` are warned and dropped. Ignoring OCSP connects without revocation checks, and the warning says so
|
||||
- `ssl_ca_data` and `ssl_ca_path` are read into `RedisTls` and trusted next to `ssl_ca_certs`. `unix_socket_path` is read into `RedisHost` and, like redis-py's `UnixDomainSocketConnection`, replaces host, port and TLS. Python still requires `host` next to it, so Rust does too
|
||||
- A url query parameter other than the TLS ones, or any query parameter on a plain `redis://` url, is warned and dropped 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 warned and dropped. 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`, zero meaning off. `redis-connect` gives cluster clients 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` is one, 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=...)`
|
||||
|
|
|
|||
|
|
@ -73,6 +73,8 @@ pub struct ResolvedRedis {
|
|||
pub namespace: Option<String>,
|
||||
pub timeout: Option<Duration>,
|
||||
pub connect_timeout: Option<Duration>,
|
||||
/// redis-py's `health_check_interval`, which `_coerce_redis_kwargs_types` reads as an `int`
|
||||
pub health_check_interval: Option<Duration>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
|
|
@ -132,7 +134,6 @@ impl RedisSettings {
|
|||
.filter(|(key, _)| !(key == "url" && drop_environment_url))
|
||||
.chain(block)
|
||||
.collect();
|
||||
reject_unsupported(&merged)?;
|
||||
warn_ignored(&merged);
|
||||
|
||||
let startup_nodes = match configured(&merged, "startup_nodes") {
|
||||
|
|
@ -197,6 +198,7 @@ impl RedisSettings {
|
|||
port: integer(&merged, "port")?.unwrap_or(6379),
|
||||
tls: tls()?,
|
||||
db: db()?,
|
||||
unix_socket_path: text_value(&merged, "unix_socket_path").map(Into::into),
|
||||
}),
|
||||
},
|
||||
};
|
||||
|
|
@ -213,6 +215,9 @@ impl RedisSettings {
|
|||
namespace: text_value(&merged, "namespace"),
|
||||
timeout: seconds(&merged, "socket_timeout")?,
|
||||
connect_timeout: seconds(&merged, "socket_connect_timeout")?,
|
||||
health_check_interval: integer::<i64>(&merged, "health_check_interval")?
|
||||
.filter(|seconds| *seconds > 0)
|
||||
.map(|seconds| Duration::from_secs(seconds.unsigned_abs())),
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -279,15 +284,6 @@ pub fn environment_has_target(lookup: &dyn Lookup) -> bool {
|
|||
.any(|name| lookup.get(name).is_some())
|
||||
}
|
||||
|
||||
fn reject_unsupported(merged: &Kwargs) -> Result<(), RedisError> {
|
||||
match merged.iter().find(|(key, _)| is_unsupported(key)) {
|
||||
Some((_, entry)) => Err(RedisError::Unsupported {
|
||||
setting: entry.source.clone(),
|
||||
}),
|
||||
None => Ok(()),
|
||||
}
|
||||
}
|
||||
|
||||
fn invalid(setting: &str, expected: Expected) -> RedisError {
|
||||
RedisError::Invalid {
|
||||
setting: setting.to_owned(),
|
||||
|
|
@ -339,17 +335,17 @@ fn coerced<N, T>(
|
|||
}
|
||||
|
||||
fn warn_ignored(merged: &Kwargs) {
|
||||
for key in merged.keys() {
|
||||
for (key, entry) in merged {
|
||||
if keys::setting(key).is_some_and(|setting| setting.handling == Handling::Warn) {
|
||||
tracing::warn!(setting = %key, "Redis setting is accepted but ignored");
|
||||
tracing::warn!(
|
||||
"Redis setting {} is not supported by the Rust client and is ignored: {}",
|
||||
entry.source,
|
||||
keys::consequence(key)
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn is_unsupported(key: &str) -> bool {
|
||||
keys::setting(key).is_some_and(|setting| setting.handling == Handling::Reject)
|
||||
}
|
||||
|
||||
fn environment_kwargs(lookup: &dyn Lookup) -> Kwargs {
|
||||
keys::SETTINGS
|
||||
.iter()
|
||||
|
|
|
|||
|
|
@ -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")]
|
||||
|
|
|
|||
|
|
@ -1,14 +1,13 @@
|
|||
use litellm_core_utils::settings::Lookup;
|
||||
|
||||
// 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 {
|
||||
|
|
@ -76,16 +75,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,
|
||||
|
|
@ -98,17 +97,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(
|
||||
|
|
@ -129,37 +128,61 @@ 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, &[]),
|
||||
];
|
||||
|
||||
/// What ignoring a warned setting means for the connection, so the operator reads the consequence
|
||||
/// rather than only the key name.
|
||||
pub(super) 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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,10 +4,12 @@ use super::{Entry, Expected, Kwargs, RedisError, bool_value, parsed, text_value}
|
|||
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
|
||||
// equivalent, so any other parameter, and any parameter on a plain redis:// url, is warned and dropped
|
||||
const URL_TLS_SETTINGS: &[&str] = &[
|
||||
"ssl_cert_reqs",
|
||||
"ssl_ca_certs",
|
||||
"ssl_ca_data",
|
||||
"ssl_ca_path",
|
||||
"ssl_certfile",
|
||||
"ssl_keyfile",
|
||||
"ssl_check_hostname",
|
||||
|
|
@ -35,6 +37,8 @@ pub(super) fn tls_options(
|
|||
.map(Into::into),
|
||||
certfile: text_value(merged, "ssl_certfile").map(Into::into),
|
||||
keyfile: text_value(merged, "ssl_keyfile").map(Into::into),
|
||||
ca_data: text_value(merged, "ssl_ca_data"),
|
||||
ca_path: text_value(merged, "ssl_ca_path").map(Into::into),
|
||||
verify_certificate,
|
||||
check_hostname: bool_value(merged, "ssl_check_hostname"),
|
||||
})
|
||||
|
|
@ -57,14 +61,16 @@ pub(super) fn url_tls(url: &str) -> Result<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(),
|
||||
});
|
||||
let (honored, ignored): (Kwargs, Kwargs) = query
|
||||
.into_iter()
|
||||
.partition(|(name, _)| tls_scheme && URL_TLS_SETTINGS.contains(&name.as_str()));
|
||||
for entry in ignored.values() {
|
||||
tracing::warn!(
|
||||
"Redis setting {} is not supported by the Rust client and is ignored",
|
||||
entry.source
|
||||
);
|
||||
}
|
||||
let query = honored;
|
||||
Ok(TlsOptions {
|
||||
check_hostname: text_value(&query, "ssl_check_hostname")
|
||||
.is_some_and(|text| url_bool(&text)),
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ use std::{
|
|||
collections::HashMap,
|
||||
fmt::Write,
|
||||
sync::{Arc, Mutex},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use litellm_config::{
|
||||
|
|
@ -136,14 +137,31 @@ fn resolves_redis_through_the_parsed_config(
|
|||
&[],
|
||||
"connection_pool"
|
||||
)]
|
||||
fn rejected_settings_name_their_source(
|
||||
fn unsupported_settings_resolve_with_a_warning_naming_their_source(
|
||||
#[case] yaml: &str,
|
||||
#[case] environment: &[(&str, &str)],
|
||||
#[case] expected: &str,
|
||||
) {
|
||||
let error = resolve(yaml, environment).unwrap_err();
|
||||
let events = Arc::new(Mutex::new(Vec::new()));
|
||||
let subscriber = tracing_subscriber::registry().with(WarningCapture(events.clone()));
|
||||
let resolved = tracing::subscriber::with_default(subscriber, || resolve(yaml, environment));
|
||||
let warnings = events.lock().unwrap().join(" ");
|
||||
|
||||
assert!(matches!(error, RedisError::Unsupported { setting } if setting == expected));
|
||||
assert_eq!(
|
||||
resolved.unwrap().endpoint,
|
||||
Endpoint::Host(litellm_redis_types::HostEndpoint {
|
||||
host: "redis".into(),
|
||||
port: 6379,
|
||||
db: 0,
|
||||
tls: None,
|
||||
unix_socket_path: None,
|
||||
})
|
||||
);
|
||||
assert!(
|
||||
warnings.contains(&format!("Redis setting {expected} is not supported")),
|
||||
"{warnings}"
|
||||
);
|
||||
assert!(warnings.contains("is ignored: "), "{warnings}");
|
||||
}
|
||||
|
||||
fn static_auth(username: Option<&str>, password: Option<&str>) -> RedisAuth {
|
||||
|
|
@ -258,22 +276,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(
|
||||
|
|
@ -522,19 +540,23 @@ fn a_missing_canonical_startup_nodes_reference_shadows_the_cache_alias() {
|
|||
}
|
||||
|
||||
#[rstest]
|
||||
fn rejected_environment_settings_never_print_the_value() {
|
||||
let error = resolve(
|
||||
"model_list: []",
|
||||
&[
|
||||
("REDIS_HOST", "redis"),
|
||||
("REDIS_SSL_CIPHERS", "private-option-value"),
|
||||
],
|
||||
)
|
||||
.unwrap_err();
|
||||
let message = error.to_string();
|
||||
fn unsupported_environment_settings_never_print_the_value() {
|
||||
let events = Arc::new(Mutex::new(Vec::new()));
|
||||
let subscriber = tracing_subscriber::registry().with(WarningCapture(events.clone()));
|
||||
tracing::subscriber::with_default(subscriber, || {
|
||||
resolve(
|
||||
"model_list: []",
|
||||
&[
|
||||
("REDIS_HOST", "redis"),
|
||||
("REDIS_SSL_CIPHERS", "private-option-value"),
|
||||
],
|
||||
)
|
||||
.unwrap();
|
||||
});
|
||||
let warnings = events.lock().unwrap().join(" ");
|
||||
|
||||
assert!(message.contains("REDIS_SSL_CIPHERS"));
|
||||
assert!(!message.contains("private-option-value"));
|
||||
assert!(warnings.contains("REDIS_SSL_CIPHERS"), "{warnings}");
|
||||
assert!(!warnings.contains("private-option-value"), "{warnings}");
|
||||
}
|
||||
|
||||
#[rstest]
|
||||
|
|
@ -649,6 +671,8 @@ 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,
|
||||
|
|
@ -668,24 +692,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: &ResolvedRedis) -> serde_json::Value {
|
||||
let target = match &resolved.endpoint {
|
||||
Endpoint::Host(host) if host.unix_socket_path.is_some() => serde_json::json!({
|
||||
"mode": "unix", "path": host.unix_socket_path, "tls": null,
|
||||
}),
|
||||
Endpoint::Host(host) => serde_json::json!({
|
||||
"mode": "host",
|
||||
"host": host.host,
|
||||
|
|
@ -740,6 +751,10 @@ fn view(resolved: &ResolvedRedis) -> 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": resolved.health_check_interval
|
||||
.or_else(|| matches!(resolved.endpoint, Endpoint::Cluster(_)).then_some(Duration::from_secs(25)))
|
||||
.filter(|interval| !interval.is_zero())
|
||||
.map(|interval| interval.as_secs_f64()),
|
||||
});
|
||||
let serde_json::Value::Object(rest) = rest else {
|
||||
unreachable!("the view is an object")
|
||||
|
|
@ -747,51 +762,68 @@ fn view(resolved: &ResolvedRedis) -> 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::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"}),
|
||||
|
|
@ -806,7 +838,7 @@ fn view(resolved: &ResolvedRedis) -> 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!({}),
|
||||
|
|
@ -821,42 +853,42 @@ fn view(resolved: &ResolvedRedis) -> 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"}),
|
||||
|
|
@ -871,47 +903,47 @@ fn view(resolved: &ResolvedRedis) -> 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"}),
|
||||
|
|
@ -926,22 +958,22 @@ fn view(resolved: &ResolvedRedis) -> 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}),
|
||||
|
|
@ -951,22 +983,22 @@ fn view(resolved: &ResolvedRedis) -> 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}]}),
|
||||
|
|
@ -981,7 +1013,7 @@ fn view(resolved: &ResolvedRedis) -> 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}]}),
|
||||
|
|
@ -991,7 +1023,7 @@ fn view(resolved: &ResolvedRedis) -> 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}]}),
|
||||
|
|
@ -1001,52 +1033,87 @@ fn view(resolved: &ResolvedRedis) -> 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::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,
|
||||
|
|
|
|||
|
|
@ -254,6 +254,7 @@ mod tests {
|
|||
port,
|
||||
tls: None,
|
||||
db: 0,
|
||||
unix_socket_path: None,
|
||||
}),
|
||||
auth: RedisAuth::Static {
|
||||
username: None,
|
||||
|
|
@ -262,6 +263,7 @@ mod tests {
|
|||
namespace: None,
|
||||
timeout: None,
|
||||
connect_timeout: None,
|
||||
health_check_interval: None,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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-redis-types.workspace = true
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use std::{
|
||||
future::Future,
|
||||
sync::Arc,
|
||||
sync::{Arc, Mutex, PoisonError},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
|
|
@ -20,6 +20,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)]
|
||||
|
|
@ -36,6 +38,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())),
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -159,6 +163,7 @@ impl RedisClient {
|
|||
.collect();
|
||||
}
|
||||
};
|
||||
self.check_health().await;
|
||||
let results = pipeline.run(&self.connection).await;
|
||||
let transport = results.iter().find_map(|result| {
|
||||
result
|
||||
|
|
@ -216,6 +221,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()));
|
||||
|
|
@ -229,6 +235,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,
|
||||
|
|
@ -238,6 +248,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()
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -156,9 +187,16 @@ pub(crate) fn connection_info(
|
|||
let (info, tls) = match endpoint {
|
||||
Endpoint::Url { url, tls } => (url.expose().into_connection_info()?, Some(tls)),
|
||||
Endpoint::Host(host) => {
|
||||
let info = addr(&host.host, host.port, host.tls.as_ref()).into_connection_info()?;
|
||||
let (addr, tls) = match &host.unix_socket_path {
|
||||
Some(path) => (ConnectionAddr::Unix(path.clone()), None),
|
||||
None => (
|
||||
addr(&host.host, host.port, host.tls.as_ref()),
|
||||
host.tls.as_ref(),
|
||||
),
|
||||
};
|
||||
let info = addr.into_connection_info()?;
|
||||
let settings = info.redis_settings().clone().set_db(host.db);
|
||||
(info.set_redis_settings(settings), host.tls.as_ref())
|
||||
(info.set_redis_settings(settings), tls)
|
||||
}
|
||||
Endpoint::Cluster(_) | Endpoint::Sentinel(_) => {
|
||||
unreachable!("cluster and sentinel endpoints build their own connection info")
|
||||
|
|
@ -185,3 +223,26 @@ fn with_static_credentials(info: ConnectionInfo, credentials: &Credentials) -> C
|
|||
};
|
||||
info.set_redis_settings(settings)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use litellm_redis_types::HostEndpoint;
|
||||
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_unix_socket_path_replaces_host_port_and_tls_like_redis_py() {
|
||||
let endpoint = Endpoint::Host(HostEndpoint {
|
||||
host: "ignored".to_owned(),
|
||||
port: 1,
|
||||
tls: Some(TlsOptions::verified()),
|
||||
db: 3,
|
||||
unix_socket_path: Some("/tmp/redis.sock".into()),
|
||||
});
|
||||
|
||||
let info = connection_info(&endpoint, &Credentials::NONE).unwrap();
|
||||
|
||||
assert_eq!(*info.addr(), ConnectionAddr::Unix("/tmp/redis.sock".into()));
|
||||
assert_eq!(info.redis_settings().db(), 3);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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},
|
||||
};
|
||||
|
|
@ -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 {
|
||||
|
|
@ -253,6 +278,7 @@ impl SentinelConnection {
|
|||
port,
|
||||
tls: self.0.endpoint.tls.clone(),
|
||||
db: self.0.endpoint.db,
|
||||
unix_socket_path: None,
|
||||
});
|
||||
let info =
|
||||
config::connection_info(&endpoint, &self.0.credentials).map_err(
|
||||
|
|
|
|||
|
|
@ -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};
|
||||
|
|
@ -254,6 +254,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();
|
||||
|
|
@ -308,6 +309,7 @@ async fn every_call_reports_its_operation_and_outcome(
|
|||
timeouts: Timeouts::default(),
|
||||
breaker: None,
|
||||
events: Some(recorded.clone()),
|
||||
health_check: None,
|
||||
})
|
||||
.unwrap();
|
||||
client.rpush("list", "item", KEEP).await.unwrap();
|
||||
|
|
@ -737,6 +739,7 @@ async fn a_host_endpoint_connects_to_its_database() {
|
|||
timeouts: Timeouts::default(),
|
||||
breaker: None,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
};
|
||||
let url = |url: String| Endpoint::Url {
|
||||
|
|
@ -748,6 +751,7 @@ async fn a_host_endpoint_connects_to_its_database() {
|
|||
port,
|
||||
tls: None,
|
||||
db: 3,
|
||||
unix_socket_path: None,
|
||||
}))
|
||||
.unwrap();
|
||||
let in_db_0 = connect(url(server.url().to_owned())).unwrap();
|
||||
|
|
@ -764,3 +768,94 @@ async fn a_host_endpoint_connects_to_its_database() {
|
|||
Some("3")
|
||||
);
|
||||
}
|
||||
|
||||
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 {
|
||||
endpoint: 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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -52,12 +63,14 @@ async fn ping_over_tls(
|
|||
port: server.port(),
|
||||
tls: Some(tls),
|
||||
db: 0,
|
||||
unix_socket_path: None,
|
||||
}),
|
||||
credentials: Credentials::NONE,
|
||||
namespace: litellm_redis_client::Namespace::default(),
|
||||
timeouts: litellm_redis_client::Timeouts::default(),
|
||||
breaker: None,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
.unwrap();
|
||||
client.ping().await
|
||||
|
|
@ -71,6 +84,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)]
|
||||
|
|
@ -128,12 +149,14 @@ fn a_missing_tls_file_names_the_path() {
|
|||
port: 6379,
|
||||
tls: Some(tls),
|
||||
db: 0,
|
||||
unix_socket_path: None,
|
||||
}),
|
||||
credentials: Credentials::NONE,
|
||||
namespace: litellm_redis_client::Namespace::default(),
|
||||
timeouts: litellm_redis_client::Timeouts::default(),
|
||||
breaker: None,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
.err()
|
||||
.unwrap();
|
||||
|
|
|
|||
|
|
@ -25,12 +25,14 @@ async fn an_out_of_range_database_is_a_server_rejection() {
|
|||
port,
|
||||
tls: None,
|
||||
db: 999,
|
||||
unix_socket_path: None,
|
||||
}),
|
||||
credentials: Credentials::NONE,
|
||||
namespace: Namespace::default(),
|
||||
timeouts: Timeouts::new(Duration::from_secs(2), None),
|
||||
breaker: None,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
.unwrap();
|
||||
let result = client.ping().await;
|
||||
|
|
@ -79,6 +81,7 @@ fn sentinel_client(sentinel: &RedisSentinel) -> RedisClient {
|
|||
timeouts: Timeouts::new(Duration::from_secs(1), None),
|
||||
breaker: None,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -48,6 +48,7 @@ pub fn connect_guarded(
|
|||
timeouts: Timeouts::new(timeout, None),
|
||||
breaker,
|
||||
events: None,
|
||||
health_check: None,
|
||||
})
|
||||
.unwrap()
|
||||
}
|
||||
|
|
@ -60,6 +61,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()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,13 +1,25 @@
|
|||
use std::time::Duration;
|
||||
|
||||
use litellm_circuit_breaker::Breaker;
|
||||
use litellm_config::{RedisAuth, ResolvedRedis};
|
||||
use litellm_redis_client::{
|
||||
Credentials, DEFAULT_TIMEOUT, Namespace, RedisClient, RedisConfig, Timeouts,
|
||||
Credentials, DEFAULT_TIMEOUT, Endpoint, Namespace, RedisClient, RedisConfig, Timeouts,
|
||||
};
|
||||
|
||||
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: ResolvedRedis, breaker: Option<Breaker>) -> Result<RedisClient, Error> {
|
||||
tracing::info!(endpoint = ?resolved.endpoint, namespace = ?resolved.namespace, "redis client configured");
|
||||
let health_check = match &resolved.endpoint {
|
||||
Endpoint::Cluster(_) => resolved
|
||||
.health_check_interval
|
||||
.or(Some(CLUSTER_HEALTH_CHECK_INTERVAL)),
|
||||
_ => resolved.health_check_interval,
|
||||
}
|
||||
.filter(|interval| !interval.is_zero());
|
||||
Ok(RedisClient::new(RedisConfig {
|
||||
endpoint: resolved.endpoint,
|
||||
credentials: credentials(resolved.auth)?,
|
||||
|
|
@ -18,6 +30,7 @@ pub fn connect(resolved: ResolvedRedis, breaker: Option<Breaker>) -> Result<Redi
|
|||
),
|
||||
breaker,
|
||||
events: None,
|
||||
health_check,
|
||||
})?)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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?;
|
||||
|
|
|
|||
|
|
@ -16,6 +16,9 @@ pub struct HostEndpoint {
|
|||
pub port: u16,
|
||||
pub tls: Option<TlsOptions>,
|
||||
pub db: i64,
|
||||
/// Connect over this unix socket instead of host, port and TLS, like redis-py's
|
||||
/// `UnixDomainSocketConnection`
|
||||
pub unix_socket_path: Option<PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
|
|
@ -42,6 +45,10 @@ pub struct SentinelEndpoint {
|
|||
#[derive(Clone, Debug, Default, 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,
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
@ -2158,6 +2141,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,
|
||||
|
|
@ -2184,6 +2169,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,
|
||||
|
|
@ -2208,6 +2195,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"},
|
||||
|
|
@ -2250,6 +2258,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,
|
||||
|
|
@ -2273,6 +2283,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,
|
||||
|
|
@ -2293,6 +2305,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"],
|
||||
}
|
||||
|
|
@ -2302,7 +2315,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(
|
||||
|
|
@ -2315,10 +2328,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,
|
||||
|
|
@ -2430,6 +2446,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,
|
||||
|
|
@ -2512,6 +2530,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,
|
||||
|
|
@ -2545,6 +2565,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,
|
||||
|
|
@ -2576,6 +2598,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,
|
||||
|
|
@ -2632,6 +2656,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,
|
||||
|
|
@ -2652,6 +2678,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,
|
||||
|
|
@ -2707,6 +2735,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",
|
||||
|
|
@ -2724,6 +2754,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,
|
||||
|
|
@ -2755,6 +2787,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,
|
||||
|
|
@ -2790,7 +2824,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(
|
||||
|
|
@ -2807,7 +2841,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(
|
||||
|
|
@ -2824,7 +2858,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(
|
||||
|
|
@ -2838,6 +2872,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,
|
||||
|
|
@ -2895,6 +2931,78 @@ 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(
|
||||
{"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):
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue