github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 22:53:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix redis sentinel tunnel route

Browse Source
This commit is contained in:
clj1566317 authored and GitHub committed 2026-07-02 14:21:52 +08:00
1 parent fd99fc8448
commit 486b9ee8ed
3 files changed
+255 -36
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/connection.rs
+158 -3
View File
@@ -809,7 +809,7 @@ impl AppState {
)
} else if db_config.uses_redis_sentinel() {
db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new(
db::redis_driver::connect_sentinel(&db_config).await?,
self.connect_redis_sentinel(connection_id, &db_config).await?,
))
} else {
db::redis_driver::RedisConnection::Direct(tokio::sync::Mutex::new(
@@ -1093,6 +1093,129 @@ impl AppState {
Ok(("127.0.0.1".to_string(), local_port))
}
pub async fn connect_redis_sentinel(
&self,
connection_id: &str,
config: &ConnectionConfig,
) -> Result<redis::aio::MultiplexedConnection, String> {
let transport_layers = config.effective_transport_layers();
if transport_layers.is_empty() {
return db::redis_driver::connect_sentinel(config).await;
}
let result = async {
let sentinel_nodes = db::redis_driver::redis_sentinel_node_endpoints(config)?;
let connect_timeout = std::time::Duration::from_secs(config.effective_connect_timeout_secs());
let layer_count = transport_layers.len();
let mut last_error = None;
for sentinel in sentinel_nodes {
let sentinel_tunnel_id = redis_sentinel_transport_id(connection_id, "sentinel", &sentinel);
let sentinel_local_port = match db::transport_layer_tunnel::start_transport_layers(
&sentinel_tunnel_id,
&transport_layers,
&sentinel.host,
sentinel.port,
&self.tunnels,
&self.proxy_tunnels,
)
.await
{
Ok(port) => port,
Err(err) => {
last_error =
Some(format!("Redis Sentinel {}:{} transport failed: {err}", sentinel.host, sentinel.port));
continue;
}
};
let master =
match db::redis_driver::discover_sentinel_master(config, "127.0.0.1", sentinel_local_port).await {
Ok(master) => master,
Err(err) => {
last_error = Some(format!(
"Redis Sentinel {}:{} master lookup failed: {err}",
sentinel.host, sentinel.port
));
db::transport_layer_tunnel::stop_transport_layers(
&sentinel_tunnel_id,
layer_count,
&self.tunnels,
&self.proxy_tunnels,
)
.await;
continue;
}
};
let master_tunnel_id = redis_sentinel_transport_id(connection_id, "master", &master);
let master_local_port = match db::transport_layer_tunnel::start_transport_layers(
&master_tunnel_id,
&transport_layers,
&master.host,
master.port,
&self.tunnels,
&self.proxy_tunnels,
)
.await
{
Ok(port) => port,
Err(err) => {
last_error = Some(format!(
"Redis Sentinel master {}:{} transport failed: {err}",
master.host, master.port
));
db::transport_layer_tunnel::stop_transport_layers(
&sentinel_tunnel_id,
layer_count,
&self.tunnels,
&self.proxy_tunnels,
)
.await;
continue;
}
};
match db::redis_driver::connect_standalone(config, "127.0.0.1", master_local_port, connect_timeout)
.await
{
Ok(con) => return Ok(con),
Err(err) => {
last_error = Some(format!(
"Redis Sentinel master {}:{} connection failed: {err}",
master.host, master.port
));
db::transport_layer_tunnel::stop_transport_layers(
&master_tunnel_id,
layer_count,
&self.tunnels,
&self.proxy_tunnels,
)
.await;
db::transport_layer_tunnel::stop_transport_layers(
&sentinel_tunnel_id,
layer_count,
&self.tunnels,
&self.proxy_tunnels,
)
.await;
}
}
}
Err(last_error.unwrap_or_else(|| "Redis Sentinel master discovery failed".to_string()))
}
.await;
if result.is_err() {
let redis_sentinel_prefix = redis_sentinel_transport_prefix(connection_id);
self.tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await;
self.proxy_tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await;
}
result
}
pub async fn connect_redis_cluster(
&self,
connection_id: &str,
@@ -1584,6 +1707,9 @@ impl AppState {
let redis_cluster_prefix = redis_cluster_transport_prefix(connection_id);
self.tunnels.stop_tunnels_with_prefix(&redis_cluster_prefix).await;
self.proxy_tunnels.stop_tunnels_with_prefix(&redis_cluster_prefix).await;
let redis_sentinel_prefix = redis_sentinel_transport_prefix(connection_id);
self.tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await;
self.proxy_tunnels.stop_tunnels_with_prefix(&redis_sentinel_prefix).await;
db::transport_layer_tunnel::stop_transport_layers(
connection_id,
layer_count,
@@ -2051,6 +2177,23 @@ fn redis_cluster_transport_id(connection_id: &str, endpoint: &db::redis_driver::
)
}
fn redis_sentinel_transport_prefix(connection_id: &str) -> String {
format!("{connection_id}:redis-sentinel:")
}
fn redis_sentinel_transport_id(
connection_id: &str,
role: &str,
endpoint: &db::redis_driver::RedisNodeEndpoint,
) -> String {
format!(
"{}{role}:{host}:{port}",
redis_sentinel_transport_prefix(connection_id),
host = endpoint.host,
port = endpoint.port
)
}
fn session_scoped_pool_key(base_pool_key: String, client_session_id: Option<&str>) -> String {
normalize_client_session_id(client_session_id)
.map(|session| format!("{base_pool_key}:session:{session}"))
@@ -2430,8 +2573,9 @@ mod tests {
use super::{
agent_connect_timeout, connection_remote_endpoint, connection_url_for_endpoint, database_connection_config,
metadata_connection_config, mysql_metadata_fallback_url, oceanbase_mysql_setup_queries,
prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, uses_bare_mysql_pool, uses_tcp_probe,
validate_h2_database_path, AppState, PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS,
prestosql_jdbc_config_for_endpoint, redacted_connection_url_for_endpoint, redis_sentinel_transport_id,
redis_sentinel_transport_prefix, uses_bare_mysql_pool, uses_tcp_probe, validate_h2_database_path, AppState,
PoolKind, PRESTOSQL_JDBC_DRIVER_CLASS,
};
use crate::agent_connection::{
agent_connect_params, mongo_legacy_error_with_auth_hint, mongo_uses_legacy_driver,
@@ -3226,6 +3370,17 @@ mod tests {
);
}
#[test]
fn redis_sentinel_transport_ids_are_connection_scoped_by_role_and_endpoint() {
let endpoint = db::redis_driver::RedisNodeEndpoint { host: "10.0.0.8".to_string(), port: 6379 };
assert_eq!(redis_sentinel_transport_prefix("redis-prod"), "redis-prod:redis-sentinel:");
assert_eq!(
redis_sentinel_transport_id("redis-prod", "master", &endpoint),
"redis-prod:redis-sentinel:master:10.0.0.8:6379"
);
}
#[test]
fn mysql_pool_size_keeps_session_pools_single_connection() {
assert_eq!(super::mysql_pool_max_connections_for_session(None), 10);
crates/dbx-core/src/db/redis_driver.rs
+94 -31
View File
@@ -263,6 +263,38 @@ pub async fn connect_sentinel(config: &ConnectionConfig) -> Result<redis::aio::M
connect_client(client).await
}
pub async fn discover_sentinel_master(
config: &ConnectionConfig,
host: &str,
port: u16,
) -> Result<RedisNodeEndpoint, 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 client = redis::Client::open(connection_info(
host,
port,
config.redis_sentinel_tls,
config.redis_tls_insecure(),
&config.redis_sentinel_username,
&config.redis_sentinel_password,
0,
))
.map_err(|e| format!("Redis Sentinel connect failed: {e}"))?;
let mut con = connect_client_with_timeout(client, super::connection_timeout(), "Redis Sentinel").await?;
let raw = tokio::time::timeout(
super::connection_timeout(),
redis::cmd("SENTINEL").arg("get-master-addr-by-name").arg(service_name).query_async::<RedisRawValue>(&mut con),
)
.await
.map_err(|_| format!("Redis Sentinel master lookup timed out ({}s)", super::CONNECTION_TIMEOUT_SECS))?
.map_err(|e| format!("Redis Sentinel master lookup failed: {e}"))?;
redis_sentinel_master_endpoint(raw)
}
pub async fn connect_cluster(config: &ConnectionConfig) -> Result<RedisClusterPool, String> {
let seed_nodes = redis_cluster_seed_nodes(config)?;
let mut last_error = None;
@@ -429,23 +461,30 @@ where
}
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([',', ';', '\n', '\r'])
.map(str::trim)
.filter(|node| !node.is_empty())
.map(ToOwned::to_owned)
.collect()
};
redis_sentinel_node_endpoints(config)?
.iter()
.map(|endpoint| {
Ok(connection_info(
&endpoint.host,
endpoint.port,
config.redis_sentinel_tls,
config.redis_tls_insecure(),
&config.redis_sentinel_username,
&config.redis_sentinel_password,
0,
))
})
.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()
pub fn redis_sentinel_node_endpoints(config: &ConnectionConfig) -> Result<Vec<RedisNodeEndpoint>, String> {
redis_node_endpoints(
config.redis_sentinel_nodes.trim(),
config.host.trim(),
config.port,
"Redis Sentinel node",
26379,
)
}
pub fn redis_cluster_seed_nodes(config: &ConnectionConfig) -> Result<Vec<RedisNodeEndpoint>, String> {
@@ -458,19 +497,6 @@ pub fn redis_cluster_seed_nodes(config: &ConnectionConfig) -> Result<Vec<RedisNo
)
}
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_tls_insecure(),
&config.redis_sentinel_username,
&config.redis_sentinel_password,
0,
))
}
fn redis_node_endpoints(
raw_nodes: &str,
fallback_host: &str,
@@ -502,6 +528,23 @@ fn redis_node_endpoints(
.collect()
}
fn redis_sentinel_master_endpoint(value: RedisRawValue) -> Result<RedisNodeEndpoint, String> {
let RedisRawValue::Array(parts) = value else {
return Err("Redis Sentinel master lookup returned an invalid response".to_string());
};
if parts.len() != 2 {
return Err("Redis Sentinel master lookup returned an invalid endpoint".to_string());
}
let Some(host) = redis_value_to_string(parts[0].clone()) else {
return Err("Redis Sentinel master lookup returned an invalid host".to_string());
};
let Some(port_text) = redis_value_to_string(parts[1].clone()) else {
return Err("Redis Sentinel master lookup returned an invalid port".to_string());
};
let port = parse_redis_port(&port_text)?;
Ok(RedisNodeEndpoint { host, port })
}
fn connection_info(
host: &str,
port: u16,
@@ -2329,8 +2372,8 @@ mod tests {
parse_stream_entries, redis_auth_candidates, redis_cluster_slot, redis_command_raw_to_json,
redis_database_index, redis_json_raw_to_json, redis_json_value_preview, redis_key_bytes_to_display,
redis_key_bytes_to_raw, redis_key_matches_query, redis_key_raw_to_bytes, redis_key_value_preview,
redis_raw_to_json, redis_value_contains_binary, redis_value_matches_query, RedisAuthCandidate,
RedisClusterSlotRange, RedisCommandSafety, RedisNodeEndpoint, RedisRawValue,
redis_raw_to_json, redis_sentinel_master_endpoint, redis_value_contains_binary, redis_value_matches_query,
RedisAuthCandidate, RedisClusterSlotRange, RedisCommandSafety, RedisNodeEndpoint, RedisRawValue,
};
use crate::models::connection::ConnectionConfig;
use redis::ConnectionAddr;
@@ -2579,6 +2622,26 @@ mod tests {
assert_eq!(parse_redis_endpoint("::1", 26379).unwrap(), ("::1".to_string(), 26379));
}
#[test]
fn parses_redis_sentinel_master_lookup_response() {
let raw = RedisRawValue::Array(vec![bulk("10.0.0.8"), bulk("6379")]);
assert_eq!(
redis_sentinel_master_endpoint(raw).unwrap(),
RedisNodeEndpoint { host: "10.0.0.8".to_string(), port: 6379 }
);
}
#[test]
fn rejects_invalid_redis_sentinel_master_lookup_response() {
let raw = RedisRawValue::Array(vec![bulk("10.0.0.8")]);
assert_eq!(
redis_sentinel_master_endpoint(raw).unwrap_err(),
"Redis Sentinel master lookup returned an invalid endpoint"
);
}
#[test]
fn redis_connection_info_marks_tls_as_insecure_when_requested() {
let info = connection_info("cache.example.com", 6379, true, true, "default", "secret", 0);