feat: support Redis Sentinel connections
This commit is contained in:
1 parent
aa0ebac233
commit
32273b4d6e
17 files changed
+470
-12
No files matched your search
@@ -27,7 +27,7 @@ webpki-roots = "0.26"
|
||||
rusqlite = { version = "0.32", features = ["bundled"] }
|
||||
mysql_async = { version = "0.36", default-features = false, features = ["default-rustls", "client_ed25519", "chrono", "rust_decimal"] }
|
||||
sqlparser = "0.62.0"
|
||||
redis = { version = "0.32.2", features = ["tokio-comp", "tls-rustls", "tokio-rustls-comp"] }
|
||||
redis = { version = "0.32.2", features = ["tokio-comp", "tls-rustls", "tokio-rustls-comp", "sentinel"] }
|
||||
rustls = { version = "0.23", features = ["aws-lc-rs"] }
|
||||
duckdb = "1.3.2"
|
||||
tiberius = { version = "0.12.3", default-features = false, features = ["tds73", "chrono", "rust_decimal", "rustls", "sql-browser-tokio"] }
|
||||
|
||||
@@ -187,7 +187,11 @@ impl AppState {
|
||||
}
|
||||
DatabaseType::Sqlite => PoolKind::Sqlite(db::sqlite::connect_path(&expand_tilde(&db_config.host)).await?),
|
||||
DatabaseType::Redis => {
|
||||
let con = db::redis_driver::connect(&url).await?;
|
||||
let con = if db_config.uses_redis_sentinel() {
|
||||
db::redis_driver::connect_sentinel(&db_config).await?
|
||||
} else {
|
||||
db::redis_driver::connect(&url).await?
|
||||
};
|
||||
PoolKind::Redis(tokio::sync::Mutex::new(con))
|
||||
}
|
||||
DatabaseType::DuckDb => {
|
||||
@@ -907,6 +911,12 @@ mod tests {
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: None,
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
|
||||
@@ -7,6 +7,7 @@ pub const MAIN_PASSWORD_KEY: &str = "password";
|
||||
pub const SSH_PASSWORD_KEY: &str = "ssh_password";
|
||||
pub const SSH_KEY_PASSPHRASE_KEY: &str = "ssh_key_passphrase";
|
||||
pub const PROXY_PASSWORD_KEY: &str = "proxy_password";
|
||||
pub const REDIS_SENTINEL_PASSWORD_KEY: &str = "redis_sentinel_password";
|
||||
pub const CONNECTION_STRING_KEY: &str = "connection_string";
|
||||
|
||||
pub trait ConnectionSecretStore {
|
||||
@@ -63,6 +64,7 @@ pub fn save_connections_to_file(
|
||||
persist_secret(store, &config.id, SSH_PASSWORD_KEY, &config.ssh_password)?;
|
||||
persist_secret(store, &config.id, SSH_KEY_PASSPHRASE_KEY, &config.ssh_key_passphrase)?;
|
||||
persist_secret(store, &config.id, PROXY_PASSWORD_KEY, &config.proxy_password)?;
|
||||
persist_secret(store, &config.id, REDIS_SENTINEL_PASSWORD_KEY, &config.redis_sentinel_password)?;
|
||||
persist_optional_secret(store, &config.id, CONNECTION_STRING_KEY, config.connection_string.as_deref())?;
|
||||
}
|
||||
|
||||
@@ -116,6 +118,15 @@ pub fn load_connections_from_file(
|
||||
needs_rewrite = true;
|
||||
}
|
||||
|
||||
if config.redis_sentinel_password.is_empty() {
|
||||
if let Some(secret) = store.get_secret(&config.id, REDIS_SENTINEL_PASSWORD_KEY)? {
|
||||
config.redis_sentinel_password = secret;
|
||||
}
|
||||
} else {
|
||||
store.set_secret(&config.id, REDIS_SENTINEL_PASSWORD_KEY, &config.redis_sentinel_password)?;
|
||||
needs_rewrite = true;
|
||||
}
|
||||
|
||||
match config.connection_string.as_deref().filter(|secret| !secret.is_empty()) {
|
||||
Some(secret) => {
|
||||
store.set_secret(&config.id, CONNECTION_STRING_KEY, secret)?;
|
||||
@@ -207,6 +218,7 @@ fn sanitize_connections(configs: &[ConnectionConfig]) -> Vec<ConnectionConfig> {
|
||||
config.ssh_password.clear();
|
||||
config.ssh_key_passphrase.clear();
|
||||
config.proxy_password.clear();
|
||||
config.redis_sentinel_password.clear();
|
||||
config.connection_string = None;
|
||||
config
|
||||
})
|
||||
@@ -221,7 +233,7 @@ pub fn secret_account(connection_id: &str, key: &str) -> String {
|
||||
mod tests {
|
||||
use super::{
|
||||
load_connections_from_file, save_connections_to_file, ConnectionSecretStore, CONNECTION_STRING_KEY,
|
||||
MAIN_PASSWORD_KEY, SSH_PASSWORD_KEY,
|
||||
MAIN_PASSWORD_KEY, REDIS_SENTINEL_PASSWORD_KEY, SSH_PASSWORD_KEY,
|
||||
};
|
||||
use crate::models::connection::{ConnectionConfig, DatabaseType, ProxyType};
|
||||
use std::cell::RefCell;
|
||||
@@ -311,6 +323,12 @@ mod tests {
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: None,
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
@@ -327,15 +345,19 @@ mod tests {
|
||||
fn save_connections_moves_passwords_to_secret_store_and_redacts_file() {
|
||||
let path = temp_connections_file("save-redacts");
|
||||
let store = MemorySecretStore::default();
|
||||
let configs = vec![connection("main", "db-secret", "ssh-secret")];
|
||||
let mut config = connection("main", "db-secret", "ssh-secret");
|
||||
config.redis_sentinel_password = "sentinel-secret".to_string();
|
||||
let configs = vec![config];
|
||||
|
||||
save_connections_to_file(&path, &configs, &store).unwrap();
|
||||
|
||||
assert_eq!(store.get_existing("main", MAIN_PASSWORD_KEY).as_deref(), Some("db-secret"));
|
||||
assert_eq!(store.get_existing("main", SSH_PASSWORD_KEY).as_deref(), Some("ssh-secret"));
|
||||
assert_eq!(store.get_existing("main", REDIS_SENTINEL_PASSWORD_KEY).as_deref(), Some("sentinel-secret"));
|
||||
let persisted = read_configs(&path);
|
||||
assert_eq!(persisted[0].password, "");
|
||||
assert_eq!(persisted[0].ssh_password, "");
|
||||
assert_eq!(persisted[0].redis_sentinel_password, "");
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -344,6 +366,7 @@ mod tests {
|
||||
let store = MemorySecretStore::default();
|
||||
store.set_existing("main", MAIN_PASSWORD_KEY, "db-secret");
|
||||
store.set_existing("main", SSH_PASSWORD_KEY, "ssh-secret");
|
||||
store.set_existing("main", REDIS_SENTINEL_PASSWORD_KEY, "sentinel-secret");
|
||||
let sanitized = vec![connection("main", "", "")];
|
||||
std::fs::write(&path, serde_json::to_string_pretty(&sanitized).unwrap()).unwrap();
|
||||
|
||||
@@ -351,6 +374,7 @@ mod tests {
|
||||
|
||||
assert_eq!(loaded[0].password, "db-secret");
|
||||
assert_eq!(loaded[0].ssh_password, "ssh-secret");
|
||||
assert_eq!(loaded[0].redis_sentinel_password, "sentinel-secret");
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -1,5 +1,10 @@
|
||||
use crate::models::connection::ConnectionConfig;
|
||||
use base64::Engine;
|
||||
use redis::{FromRedisValue, Value as RedisRawValue};
|
||||
use redis::{
|
||||
sentinel::{Sentinel, SentinelNodeConnectionInfo},
|
||||
ConnectionAddr, ConnectionInfo, FromRedisValue, ProtocolVersion, RedisConnectionInfo, TlsMode,
|
||||
Value as RedisRawValue,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
const STREAM_ENTRY_LIMIT: usize = 100;
|
||||
@@ -71,6 +76,128 @@ pub async fn connect(url: &str) -> Result<redis::aio::MultiplexedConnection, Str
|
||||
Ok(con)
|
||||
}
|
||||
|
||||
pub async fn connect_sentinel(config: &ConnectionConfig) -> Result<redis::aio::MultiplexedConnection, String> {
|
||||
let service_name = config.redis_sentinel_master.trim();
|
||||
if service_name.is_empty() {
|
||||
return Err("Redis Sentinel master name is required".to_string());
|
||||
}
|
||||
|
||||
let nodes = redis_sentinel_nodes(config)?;
|
||||
let mut sentinel = Sentinel::build(nodes).map_err(|e| format!("Redis Sentinel connection failed: {e}"))?;
|
||||
let node_connection_info = SentinelNodeConnectionInfo {
|
||||
tls_mode: if config.ssl { Some(TlsMode::Secure) } else { None },
|
||||
redis_connection_info: Some(redis_connection_info(&config.username, &config.password, 0)),
|
||||
};
|
||||
let client = tokio::time::timeout(
|
||||
super::connection_timeout(),
|
||||
sentinel.async_master_for(service_name, Some(&node_connection_info)),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| format!("Redis Sentinel lookup timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
|
||||
.map_err(|e| format!("Redis Sentinel master lookup failed: {e}"))?;
|
||||
|
||||
let mut con = tokio::time::timeout(super::connection_timeout(), client.get_multiplexed_async_connection())
|
||||
.await
|
||||
.map_err(|_| format!("Redis connection timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
|
||||
.map_err(|e| format!("Redis connection failed: {e}"))?;
|
||||
|
||||
tokio::time::timeout(super::connection_timeout(), redis::cmd("PING").query_async::<String>(&mut con))
|
||||
.await
|
||||
.map_err(|_| format!("Redis ping timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
|
||||
.map_err(|e| format!("Redis authentication failed or command rejected: {e}"))?;
|
||||
|
||||
Ok(con)
|
||||
}
|
||||
|
||||
fn redis_sentinel_nodes(config: &ConnectionConfig) -> Result<Vec<ConnectionInfo>, String> {
|
||||
let raw_nodes = config.redis_sentinel_nodes.trim();
|
||||
let endpoints: Vec<String> = if raw_nodes.is_empty() {
|
||||
vec![format!("{}:{}", config.host.trim(), config.port)]
|
||||
} else {
|
||||
raw_nodes
|
||||
.split(|ch: char| ch == ',' || ch == ';' || ch == '\n' || ch == '\r')
|
||||
.map(str::trim)
|
||||
.filter(|node| !node.is_empty())
|
||||
.map(ToOwned::to_owned)
|
||||
.collect()
|
||||
};
|
||||
|
||||
if endpoints.is_empty() {
|
||||
return Err("At least one Redis Sentinel node is required".to_string());
|
||||
}
|
||||
|
||||
endpoints.iter().map(|endpoint| redis_sentinel_node_connection_info(config, endpoint)).collect()
|
||||
}
|
||||
|
||||
fn redis_sentinel_node_connection_info(config: &ConnectionConfig, endpoint: &str) -> Result<ConnectionInfo, String> {
|
||||
let (host, port) = parse_redis_endpoint(endpoint, 26379)?;
|
||||
Ok(connection_info(
|
||||
&host,
|
||||
port,
|
||||
config.redis_sentinel_tls,
|
||||
&config.redis_sentinel_username,
|
||||
&config.redis_sentinel_password,
|
||||
0,
|
||||
))
|
||||
}
|
||||
|
||||
fn connection_info(host: &str, port: u16, tls: bool, username: &str, password: &str, db: i64) -> ConnectionInfo {
|
||||
let addr = if tls {
|
||||
ConnectionAddr::TcpTls { host: host.to_string(), port, insecure: false, tls_params: None }
|
||||
} else {
|
||||
ConnectionAddr::Tcp(host.to_string(), port)
|
||||
};
|
||||
ConnectionInfo { addr, redis: redis_connection_info(username, password, db) }
|
||||
}
|
||||
|
||||
fn redis_connection_info(username: &str, password: &str, db: i64) -> RedisConnectionInfo {
|
||||
RedisConnectionInfo {
|
||||
db,
|
||||
username: non_empty_string(username),
|
||||
password: non_empty_string(password),
|
||||
protocol: ProtocolVersion::RESP2,
|
||||
}
|
||||
}
|
||||
|
||||
fn non_empty_string(value: &str) -> Option<String> {
|
||||
let value = value.trim();
|
||||
if value.is_empty() {
|
||||
None
|
||||
} else {
|
||||
Some(value.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_redis_endpoint(endpoint: &str, default_port: u16) -> Result<(String, u16), String> {
|
||||
let endpoint = endpoint.trim();
|
||||
if endpoint.is_empty() {
|
||||
return Err("Redis Sentinel node cannot be empty".to_string());
|
||||
}
|
||||
let endpoint = endpoint.strip_prefix("redis://").or_else(|| endpoint.strip_prefix("rediss://")).unwrap_or(endpoint);
|
||||
let endpoint = endpoint.rsplit_once('@').map(|(_, tail)| tail).unwrap_or(endpoint);
|
||||
let endpoint = endpoint.split(['/', '?', '#']).next().unwrap_or(endpoint);
|
||||
|
||||
if let Some(rest) = endpoint.strip_prefix('[') {
|
||||
let Some((host, tail)) = rest.split_once(']') else {
|
||||
return Err(format!("Invalid Redis Sentinel node '{endpoint}'"));
|
||||
};
|
||||
let port = tail.strip_prefix(':').filter(|value| !value.is_empty()).map(parse_redis_port).transpose()?;
|
||||
return Ok((host.to_string(), port.unwrap_or(default_port)));
|
||||
}
|
||||
|
||||
if let Some((host, port)) = endpoint.rsplit_once(':') {
|
||||
if !host.contains(':') && port.chars().all(|ch| ch.is_ascii_digit()) {
|
||||
return Ok((host.to_string(), parse_redis_port(port)?));
|
||||
}
|
||||
}
|
||||
|
||||
Ok((endpoint.to_string(), default_port))
|
||||
}
|
||||
|
||||
fn parse_redis_port(port: &str) -> Result<u16, String> {
|
||||
port.parse::<u16>().map_err(|_| format!("Invalid Redis Sentinel port '{port}'"))
|
||||
}
|
||||
|
||||
pub async fn list_databases(con: &mut redis::aio::MultiplexedConnection) -> Result<Vec<RedisDatabaseInfo>, String> {
|
||||
let configured_count =
|
||||
redis::cmd("CONFIG").arg("GET").arg("databases").query_async(con).await.ok().and_then(parse_database_count);
|
||||
@@ -899,10 +1026,11 @@ fn parse_scan_members(raw: RedisRawValue) -> Result<(u64, Vec<serde_json::Value>
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
classify_command, is_redis_json_type, parse_command_argv, parse_database_count, parse_scan_keys,
|
||||
parse_stream_entries, redis_command_raw_to_json, redis_json_raw_to_json, redis_json_value_preview,
|
||||
redis_key_bytes_to_display, redis_key_bytes_to_raw, redis_key_raw_to_bytes, redis_key_value_preview,
|
||||
redis_raw_to_json, redis_value_contains_binary, redis_value_matches_query, RedisCommandSafety, RedisRawValue,
|
||||
classify_command, is_redis_json_type, parse_command_argv, parse_database_count, parse_redis_endpoint,
|
||||
parse_scan_keys, parse_stream_entries, redis_command_raw_to_json, redis_json_raw_to_json,
|
||||
redis_json_value_preview, redis_key_bytes_to_display, redis_key_bytes_to_raw, redis_key_raw_to_bytes,
|
||||
redis_key_value_preview, redis_raw_to_json, redis_value_contains_binary, redis_value_matches_query,
|
||||
RedisCommandSafety, RedisRawValue,
|
||||
};
|
||||
|
||||
fn bulk(value: &str) -> RedisRawValue {
|
||||
@@ -1069,6 +1197,18 @@ mod tests {
|
||||
assert!(!is_redis_json_type("string"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_redis_sentinel_endpoints_with_default_ports() {
|
||||
assert_eq!(parse_redis_endpoint("sentinel.local:26380", 26379).unwrap(), ("sentinel.local".to_string(), 26380));
|
||||
assert_eq!(
|
||||
parse_redis_endpoint("redis://user:pass@sentinel.local:26380/0", 26379).unwrap(),
|
||||
("sentinel.local".to_string(), 26380)
|
||||
);
|
||||
assert_eq!(parse_redis_endpoint("sentinel.local", 26379).unwrap(), ("sentinel.local".to_string(), 26379));
|
||||
assert_eq!(parse_redis_endpoint("[::1]:26380", 26379).unwrap(), ("::1".to_string(), 26380));
|
||||
assert_eq!(parse_redis_endpoint("::1", 26379).unwrap(), ("::1".to_string(), 26379));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_redis_json_get_bulk_string() {
|
||||
let raw = bulk(r#"{"id":1,"embedding":[0.1,0.2],"meta":{"source":"test"}}"#);
|
||||
|
||||
@@ -63,6 +63,18 @@ pub struct ConnectionConfig {
|
||||
pub oracle_connection_type: Option<String>,
|
||||
#[serde(default)]
|
||||
pub connection_string: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub redis_connection_mode: Option<String>,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub redis_sentinel_master: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub redis_sentinel_nodes: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub redis_sentinel_username: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub redis_sentinel_password: String,
|
||||
#[serde(default, skip_serializing_if = "is_false")]
|
||||
pub redis_sentinel_tls: bool,
|
||||
/// Typed configuration for external tabular sources.
|
||||
#[serde(default)]
|
||||
pub external_config: Option<serde_json::Value>,
|
||||
@@ -234,6 +246,11 @@ impl ConnectionConfig {
|
||||
config
|
||||
}
|
||||
|
||||
pub fn uses_redis_sentinel(&self) -> bool {
|
||||
self.db_type == DatabaseType::Redis
|
||||
&& self.redis_connection_mode.as_deref().is_some_and(|mode| mode.eq_ignore_ascii_case("sentinel"))
|
||||
}
|
||||
|
||||
pub fn connection_url(&self) -> String {
|
||||
self.connection_url_with_host(&self.host, self.port)
|
||||
}
|
||||
@@ -777,6 +794,12 @@ mod tests {
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: None,
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
|
||||
@@ -1417,6 +1417,12 @@ mod tests {
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: Some("jdbc:h2:mem:test".to_string()),
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
|
||||
@@ -446,6 +446,7 @@ impl Storage {
|
||||
sanitized.ssh_password = String::new();
|
||||
sanitized.ssh_key_passphrase = String::new();
|
||||
sanitized.proxy_password = String::new();
|
||||
sanitized.redis_sentinel_password = String::new();
|
||||
sanitized.connection_string = None;
|
||||
let json = serde_json::to_string(&sanitized).map_err(|e| e.to_string())?;
|
||||
|
||||
@@ -456,6 +457,7 @@ impl Storage {
|
||||
persist_secret_in_tx(&tx, &config.id, "ssh_password", &config.ssh_password)?;
|
||||
persist_secret_in_tx(&tx, &config.id, "ssh_key_passphrase", &config.ssh_key_passphrase)?;
|
||||
persist_secret_in_tx(&tx, &config.id, "proxy_password", &config.proxy_password)?;
|
||||
persist_secret_in_tx(&tx, &config.id, "redis_sentinel_password", &config.redis_sentinel_password)?;
|
||||
if let Some(cs) = &config.connection_string {
|
||||
persist_secret_in_tx(&tx, &config.id, "connection_string", cs)?;
|
||||
} else {
|
||||
@@ -499,6 +501,7 @@ impl Storage {
|
||||
config.ssh_password = self.get_secret(&id, "ssh_password").await?.unwrap_or_default();
|
||||
config.ssh_key_passphrase = self.get_secret(&id, "ssh_key_passphrase").await?.unwrap_or_default();
|
||||
config.proxy_password = self.get_secret(&id, "proxy_password").await?.unwrap_or_default();
|
||||
config.redis_sentinel_password = self.get_secret(&id, "redis_sentinel_password").await?.unwrap_or_default();
|
||||
config.connection_string = self.get_secret(&id, "connection_string").await?;
|
||||
configs.push(config.canonicalized());
|
||||
}
|
||||
|
||||
@@ -1262,6 +1262,12 @@ mod tests {
|
||||
sysdba: false,
|
||||
oracle_connection_type: None,
|
||||
connection_string: None,
|
||||
redis_connection_mode: None,
|
||||
redis_sentinel_master: String::new(),
|
||||
redis_sentinel_nodes: String::new(),
|
||||
redis_sentinel_username: String::new(),
|
||||
redis_sentinel_password: String::new(),
|
||||
redis_sentinel_tls: false,
|
||||
external_config: None,
|
||||
jdbc_driver_class: None,
|
||||
jdbc_driver_paths: Vec::new(),
|
||||
|
||||
Reference in new issue
Block a user