diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 79a6f6864..cb942830a 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -34,6 +34,7 @@ use serde_json; use std::ffi::OsStr; use std::fs as stdfs; use std::io::ErrorKind; +use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; use std::sync::Once; @@ -1214,6 +1215,9 @@ pub struct RustFSTestClusterEnvironment { pub node_extra_env: Vec>, pub node_capture_log_paths: Vec>, pub topology: ClusterTopology, + /// Optional socket proxies used for the corresponding node's volume + /// endpoints. Proxies must be installed before [`Self::start`]. + volume_proxy_addresses: Vec>, } impl RustFSTestClusterEnvironment { @@ -1305,6 +1309,7 @@ impl RustFSTestClusterEnvironment { extra_env.push(("RUSTFS_UNSAFE_BYPASS_DISK_CHECK".to_string(), "true".to_string())); } + let node_count = topology.node_count; Ok(Self { nodes, temp_dir, @@ -1314,6 +1319,7 @@ impl RustFSTestClusterEnvironment { node_extra_env: vec![Vec::new(); topology.node_count], node_capture_log_paths: vec![None; topology.node_count], topology, + volume_proxy_addresses: vec![None; node_count], }) } @@ -1381,6 +1387,34 @@ impl RustFSTestClusterEnvironment { self.build_volumes_arg() } + /// Start a socket proxy for one node's volume endpoints and route all + /// subsequent `RUSTFS_VOLUMES` references for that node through it. + /// + /// Call this before [`Self::start`], then use the returned proxy's + /// [`crate::fault_proxy::FaultProxy::set_mode`] to inject latency, + /// blackhole, or one-way partition faults. The node's own listen address + /// remains direct, so S3 clients can still reach it while peer disk/RPC + /// traffic is steered through the proxy. + pub async fn start_volume_proxy_for_node( + &mut self, + node_idx: usize, + ) -> Result> { + self.ensure_node_index(node_idx)?; + if self.volume_proxy_addresses[node_idx].is_some() { + return Err(format!("a volume proxy is already configured for node {node_idx}").into()); + } + let target = self.nodes[node_idx].address.parse::()?; + let proxy = crate::fault_proxy::FaultProxy::start(target).await?; + self.volume_proxy_addresses[node_idx] = Some(proxy.local_addr()); + Ok(proxy) + } + + fn volume_address(&self, node_idx: usize) -> String { + self.volume_proxy_addresses[node_idx] + .map(|address| address.to_string()) + .unwrap_or_else(|| self.nodes[node_idx].address.clone()) + } + fn build_volumes_arg(&self) -> String { let pools = self.topology.normalized_pools(); @@ -1389,7 +1423,11 @@ impl RustFSTestClusterEnvironment { return self .nodes .iter() - .flat_map(|n| n.data_dirs.iter().map(move |dir| format!("http://{}{}", n.address, dir))) + .enumerate() + .flat_map(|(node_idx, n)| { + let address = self.volume_address(node_idx); + n.data_dirs.iter().map(move |dir| format!("http://{}{}", address, dir)) + }) .collect::>() .join(" "); } @@ -1400,13 +1438,19 @@ impl RustFSTestClusterEnvironment { pools .iter() .map(|nodes| { - let node = &self.nodes[nodes[0]]; + let node_idx = nodes[0]; + let node = &self.nodes[node_idx]; let base = node .data_dirs .first() .and_then(|d| d.rsplit_once('/').map(|(parent, _)| parent)) .unwrap_or(&node.data_dir); - format!("http://{}{}/drive{{0...{}}}", node.address, base, self.topology.drives_per_node - 1) + format!( + "http://{}{}/drive{{0...{}}}", + self.volume_address(node_idx), + base, + self.topology.drives_per_node - 1 + ) }) .collect::>() .join(" ") @@ -2000,7 +2044,7 @@ mod tests { } let multidrive = topology.drives_per_node > 1; - let nodes = (0..topology.node_count) + let nodes: Vec = (0..topology.node_count) .map(|i| { let address = format!("127.0.0.1:{}", 9000 + i); let data_dirs: Vec = if multidrive { @@ -2021,6 +2065,7 @@ mod tests { }) .collect(); + let node_count = nodes.len(); RustFSTestClusterEnvironment { nodes, temp_dir, @@ -2030,6 +2075,7 @@ mod tests { node_extra_env: vec![Vec::new(); topology.node_count], node_capture_log_paths: vec![None; topology.node_count], topology, + volume_proxy_addresses: vec![None; node_count], } } @@ -2114,6 +2160,25 @@ mod tests { assert!(ClusterTopology::single_pool_multidrive(1, 1).validate().is_ok()); } + #[tokio::test] + async fn volume_proxy_rewrites_cluster_volume_endpoint() { + let mut env = RustFSTestClusterEnvironment::new(1) + .await + .expect("cluster environment should allocate a node"); + let direct = env.nodes[0].address.clone(); + let proxy = env + .start_volume_proxy_for_node(0) + .await + .expect("volume proxy should bind before the target server starts"); + let proxied = proxy.local_addr().to_string(); + let volumes = env.rustfs_volumes_arg(); + + assert!(volumes.contains(&proxied), "volumes must use the proxy address: {volumes}"); + assert!(!volumes.contains(&direct), "volumes must not retain the direct address: {volumes}"); + + proxy.shutdown().await; + } + #[test] fn cluster_node_env_supports_per_node_overrides() { let mut env = fake_cluster(ClusterTopology::single_pool(4));