diff --git a/.config/e2e-distributed-selection.txt b/.config/e2e-distributed-selection.txt index 15cf1285b..32e35a5cd 100644 --- a/.config/e2e-distributed-selection.txt +++ b/.config/e2e-distributed-selection.txt @@ -1,2 +1,2 @@ -sha256-linux=3dc5c2dc931c76b9a1ac28275fab1279c5d58507a88fe7ff59faad60deb1c7f7 -sha256-darwin=3dc5c2dc931c76b9a1ac28275fab1279c5d58507a88fe7ff59faad60deb1c7f7 +sha256-linux=cf31ed27e4e6bf42da5bf81f248d9d2619ac8e8615186cdacb30bd63996d7607 +sha256-darwin=cf31ed27e4e6bf42da5bf81f248d9d2619ac8e8615186cdacb30bd63996d7607 diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index bdba4346c..6acc5347e 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -1272,6 +1272,7 @@ pub struct RustFSTestClusterEnvironment { /// Optional socket proxies used for the corresponding node's volume /// endpoints. Proxies must be installed before [`Self::start`]. volume_proxy_addresses: Vec>, + tls_ca_pem: Option>, } impl RustFSTestClusterEnvironment { @@ -1377,9 +1378,35 @@ impl RustFSTestClusterEnvironment { node_capture_log_paths: vec![None; topology.node_count], topology, volume_proxy_addresses: vec![None; node_count], + tls_ca_pem: None, }) } + /// Configure verified HTTPS for both volume endpoints and S3 clients. + /// The bundle must be generated before any node starts. + pub async fn enable_tls(&mut self, directory: &Path) -> std::io::Result<()> { + if self.nodes.iter().any(|node| node.process.is_some()) { + return Err(std::io::Error::new( + ErrorKind::InvalidInput, + "configure cluster TLS before starting nodes", + )); + } + let directory = fs::canonicalize(directory).await?; + let ca_pem = fs::read(directory.join("ca.crt")).await?; + reqwest::Certificate::from_pem(&ca_pem).map_err(|error| std::io::Error::new(ErrorKind::InvalidData, error))?; + self.set_env( + "RUSTFS_TLS_PATH", + directory + .to_str() + .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "TLS directory is not UTF-8"))?, + ); + for node in &mut self.nodes { + node.url = format!("https://{}", node.address); + } + self.tls_ca_pem = Some(ca_pem); + Ok(()) + } + /// Add an extra environment variable applied to every cluster node process. pub fn set_env(&mut self, key: K, value: V) where @@ -1501,6 +1528,7 @@ impl RustFSTestClusterEnvironment { fn build_volumes_arg(&self) -> String { let pools = self.topology.normalized_pools(); + let scheme = if self.tls_ca_pem.is_some() { "https" } else { "http" }; if pools.len() <= 1 { // Single pool: explicit enumeration of every drive on every node. @@ -1510,7 +1538,7 @@ impl RustFSTestClusterEnvironment { .enumerate() .flat_map(|(node_idx, n)| { let address = self.volume_address(node_idx); - n.data_dirs.iter().map(move |dir| format!("http://{}{}", address, dir)) + n.data_dirs.iter().map(move |dir| format!("{scheme}://{address}{dir}")) }) .collect::>() .join(" "); @@ -1530,7 +1558,7 @@ impl RustFSTestClusterEnvironment { .and_then(|d| d.rsplit_once('/').map(|(parent, _)| parent)) .unwrap_or(&node.data_dir); format!( - "http://{}{}/drive{{0...{}}}", + "{scheme}://{}{}/drive{{0...{}}}", self.volume_address(node_idx), base, self.topology.drives_per_node - 1 @@ -1719,13 +1747,20 @@ impl RustFSTestClusterEnvironment { if node_idx >= self.nodes.len() { return Err("node_idx is invalid".into()); } - Ok(Client::from_conf(build_test_s3_config( - &self.nodes[node_idx].url, - &self.access_key, - &self.secret_key, - None, - "cluster-test", - ))) + let mut config = + build_test_s3_config(&self.nodes[node_idx].url, &self.access_key, &self.secret_key, None, "cluster-test"); + if let Some(ca_pem) = &self.tls_ca_pem { + use aws_smithy_http_client::tls::{Provider, TlsContext, TrustStore, rustls_provider::CryptoMode}; + let tls = TlsContext::builder() + .with_trust_store(TrustStore::empty().with_pem_certificate(ca_pem.clone())) + .build()?; + let client = SmithyHttpClientBuilder::new() + .tls_provider(Provider::Rustls(CryptoMode::AwsLc)) + .tls_context(tls) + .build_https(); + config = config.to_builder().http_client(client).build(); + } + Ok(Client::from_conf(config)) } /// Create S3 clients for all nodes in the RustFS cluster and collect them into a vector. @@ -1850,7 +1885,7 @@ impl RustFSTestClusterEnvironment { } self.nodes.push(ClusterNode { - url: format!("http://{address}"), + url: format!("{}://{address}", if self.tls_ca_pem.is_some() { "https" } else { "http" }), address, data_dir: data_dirs[0].clone(), data_dirs, @@ -2464,6 +2499,7 @@ mod tests { node_capture_log_paths: vec![None; topology.node_count], topology, volume_proxy_addresses: vec![None; node_count], + tls_ca_pem: None, } } @@ -2481,8 +2517,8 @@ mod tests { ); } - #[test] - fn volumes_single_pool_multidrive_enumerates_every_drive() { + #[tokio::test] + async fn volumes_single_pool_multidrive_enumerates_every_drive() { // 4 nodes x 2 drives -> 8 explicit endpoints, one legacy pool. No // ellipses, so the server parser keeps this as a single DistErasure pool. let env = fake_cluster(ClusterTopology::single_pool_multidrive(4, 2)); @@ -2494,6 +2530,25 @@ mod tests { .join(" "); assert_eq!(env.build_volumes_arg(), expected); assert_eq!(env.build_volumes_arg().split(' ').count(), 8); + + let directory = tempfile::tempdir().expect("TLS fixture directory"); + crate::tls_gen::run(crate::tls_gen::Args { + out_dir: directory.path().to_owned(), + days: 1, + force: false, + }) + .expect("generate TLS bundle"); + for topology in [ + ClusterTopology::single_pool_multidrive(4, 4), + ClusterTopology::per_node_pools(4, vec![vec![0], vec![1]]), + ] { + let mut cluster = fake_cluster(topology); + let plaintext_volumes = cluster.build_volumes_arg(); + cluster.enable_tls(directory.path()).await.expect("configure TLS"); + assert_eq!(cluster.build_volumes_arg(), plaintext_volumes.replace("http://", "https://")); + assert!(cluster.nodes.iter().all(|node| node.url.starts_with("https://"))); + assert!(cluster.create_s3_client(0).is_ok(), "trusted SDK client should build"); + } } #[test] diff --git a/crates/e2e_test/src/distributed/harness.rs b/crates/e2e_test/src/distributed/harness.rs index e710d2eae..64d120f8b 100644 --- a/crates/e2e_test/src/distributed/harness.rs +++ b/crates/e2e_test/src/distributed/harness.rs @@ -34,7 +34,7 @@ use crate::common::{ }; use crate::replication_extension_test::LOOPBACK_REPLICATION_TARGET_ENV; use aws_sdk_s3::Client; -use aws_sdk_s3::config::retry::RetryConfig; +use aws_sdk_s3::config::{Credentials, retry::RetryConfig}; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::operation::{get_object::GetObjectError, put_object::PutObjectError}; use aws_sdk_s3::primitives::ByteStream; @@ -185,13 +185,12 @@ impl DistCluster { if node_idx >= self.cluster.nodes.len() { return Err("node_idx is invalid".into()); } - Ok(Client::from_conf(build_test_s3_config( - &self.cluster.nodes[node_idx].url, - access_key, - secret_key, - None, - "cluster-iam-test", - ))) + let config = self.cluster.create_s3_client(node_idx)?.config().to_builder(); + Ok(Client::from_conf( + config + .credentials_provider(Credentials::new(access_key, secret_key, None, None, "cluster-iam-test")) + .build(), + )) } /// Stop the fleet, drop retired pools from the command line, and restart diff --git a/crates/e2e_test/src/distributed/internode_tls_test.rs b/crates/e2e_test/src/distributed/internode_tls_test.rs new file mode 100644 index 000000000..54842b817 --- /dev/null +++ b/crates/e2e_test/src/distributed/internode_tls_test.rs @@ -0,0 +1,250 @@ +// Copyright 2026 RustFS Team +// Licensed under the Apache License, Version 2.0. + +use super::harness::{DistCluster, DistLayout, TestResult, unique_bucket}; +use crate::tls_gen; +use aws_sdk_s3::{Client, config::retry::RetryConfig, primitives::ByteStream}; +use futures::future::try_join_all; +use rand::{Rng, SeedableRng, rngs::StdRng}; +use sha2::{Digest, Sha256}; +use std::io::{Read, Seek, SeekFrom, Write}; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant}; + +const MIB: u64 = 1024 * 1024; +const READ_TIMEOUT: Duration = Duration::from_secs(180); + +struct ExpectedRead { + range: Option<(u64, u64)>, + length: u64, + sha256: String, +} + +struct ObjectFixture { + key: String, + path: PathBuf, + length: u64, + reads: Vec, +} + +fn fixture(directory: &Path, length: u64) -> TestResult { + let key = format!("object-{length}"); + let path = directory.join(&key); + let mut file = std::fs::File::create(&path)?; + let mut rng = StdRng::seed_from_u64(length); + let mut block = vec![0; usize::try_from(MIB)?]; + let mut remaining = length; + let mut digest = Sha256::new(); + while remaining != 0 { + let count = usize::try_from(remaining.min(MIB))?; + rng.fill_bytes(&mut block[..count]); + file.write_all(&block[..count])?; + digest.update(&block[..count]); + remaining -= u64::try_from(count)?; + } + drop(file); + let mut reads = vec![ExpectedRead { + range: None, + length, + sha256: hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower), + }]; + // Unaligned ranges cross TLS frames and EC blocks; the tail detects + // truncation and incorrect range offsets independently of a full GET. + for (start, end) in [(17, (MIB + 31).min(length - 1)), (length.saturating_sub(65537), length - 1)] { + let mut input = std::fs::File::open(&path)?; + input.seek(SeekFrom::Start(start))?; + let range_length = end - start + 1; + let mut hasher = Sha256::new(); + let mut input = input.take(range_length); + loop { + let count = input.read(&mut block)?; + if count == 0 { + break; + } + hasher.update(&block[..count]); + } + reads.push(ExpectedRead { + range: Some((start, end)), + length: range_length, + sha256: hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower), + }); + } + Ok(ObjectFixture { + key, + path, + length, + reads, + }) +} + +async fn verified_get( + client: &Client, + node: usize, + bucket: &str, + object: &ObjectFixture, + expected: &ExpectedRead, + concurrent: bool, +) -> TestResult { + tokio::time::timeout(READ_TIMEOUT, async { + let started = Instant::now(); + let mut request = client.get_object().bucket(bucket).key(&object.key); + if let Some((start, end)) = expected.range { + request = request.range(format!("bytes={start}-{end}")); + } + let mut output = request.send().await?; + assert_eq!(output.content_length(), Some(i64::try_from(expected.length)?)); + if let Some((start, end)) = expected.range { + assert_eq!(output.content_range(), Some(format!("bytes {start}-{end}/{}", object.length).as_str())); + } else { + assert_eq!(output.content_range(), None); + } + let mut count = 0u64; + let mut digest = Sha256::new(); + while let Some(chunk) = output.body.next().await { + let chunk = chunk?; + count += u64::try_from(chunk.len())?; + assert!(count <= expected.length, "node {node}: GET exceeded the requested length"); + digest.update(&chunk); + } + let actual_sha256 = hex_simd::encode_to_string(digest.finalize(), hex_simd::AsciiCase::Lower); + assert_eq!(count, expected.length, "node {node}: truncated GET {}", object.key); + assert_eq!(actual_sha256, expected.sha256, "node {node}: corrupt GET {}", object.key); + Ok(serde_json::json!({ + "operation": "get", "node": node, "key": object.key, + "range": expected.range, "bytes": count, "sha256": actual_sha256, + "concurrent": concurrent, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0, + })) + }) + .await? +} + +/// This case always exercises the full acceptance sizes; do not substitute +/// tiny objects and count that as a large distributed-read pass. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn four_node_tls_large_and_range_gets_preserve_bytes() -> TestResult { + let mode = std::env::var("RUSTFS_E2E_INTERNODE_HTTP_MODE").unwrap_or_else(|_| "http1".into()); + let protocol = match mode.as_str() { + "http1" => "http1", + "auto" | "unpooled" => "auto", + _ => return Err("RUSTFS_E2E_INTERNODE_HTTP_MODE must be http1, auto or unpooled".into()), + }; + let mut dist = DistCluster::new_stopped_with_env( + DistLayout::FourByFour, + &[ + ("RUSTFS_INTERNODE_HTTP_PROTOCOL", protocol), + ("RUSTFS_SCANNER_ENABLED", "true"), + ("RUSTFS_SCANNER_START_DELAY_SECS", "0"), + ("RUSTFS_SCANNER_CYCLE", "1"), + ("RUSTFS_HEAL_ENABLED", "true"), + ("RUST_LOG", "warn"), + ], + ) + .await?; + if mode == "unpooled" { + dist.cluster.set_env("RUSTFS_INTERNODE_HTTP_POOL_MAX_IDLE_PER_HOST", "0"); + } + let directory = PathBuf::from(&dist.cluster.temp_dir); + let tls_directory = directory.join("tls"); + tls_gen::run(tls_gen::Args { + out_dir: tls_directory.clone(), + days: 1, + force: false, + })?; + dist.cluster.enable_tls(&tls_directory).await?; + let report_path = std::env::var_os("RUSTFS_INTERNODE_ACCEPTANCE_REPORT").map(PathBuf::from); + if let Some(path) = &report_path + && (path.exists() || path.with_extension("pids.json").exists()) + { + return Err("acceptance report paths must be unused".into()); + } + let log_directory = report_path + .as_ref() + .map_or_else(|| directory.join("logs"), |path| path.with_extension("logs")); + std::fs::create_dir(&log_directory)?; + dist.capture_logs_under(&log_directory)?; + dist.cluster.start().await?; + let clients = dist + .clients()? + .into_iter() + .map(|client| Client::from_conf(client.config().to_builder().retry_config(RetryConfig::disabled()).build())) + .collect::>(); + let bucket = unique_bucket("internode-tls"); + clients[0].create_bucket().bucket(&bucket).send().await?; + let pids = dist + .cluster + .nodes + .iter() + .map(|node| node.process.as_ref().map(std::process::Child::id)) + .collect::>(); + if let Some(path) = &report_path { + std::fs::write(path.with_extension("pids.json"), serde_json::to_vec(&pids)?)?; + } + + let mut results = Vec::new(); + for length in [32 * 1024, 100 * MIB, 1024 * MIB] { + let fixture_directory = directory.clone(); + let object = tokio::task::spawn_blocking(move || fixture(&fixture_directory, length)).await??; + let started = Instant::now(); + tokio::time::timeout( + READ_TIMEOUT, + clients[0] + .put_object() + .bucket(&bucket) + .key(&object.key) + .body(ByteStream::from_path(&object.path).await?) + .send(), + ) + .await??; + results.push( + serde_json::json!({"operation": "put", "bytes": length, "elapsed_ms": started.elapsed().as_secs_f64() * 1000.0}), + ); + if length == 32 * 1024 { + let alternate = dist.client_with_credentials(0, &dist.cluster.access_key, &dist.cluster.secret_key)?; + let alternate = Client::from_conf(alternate.config().to_builder().retry_config(RetryConfig::disabled()).build()); + results.push(verified_get(&alternate, 0, &bucket, &object, &object.reads[0], false).await?); + let invalid = dist.client_with_credentials(0, "invalid-tls-user", "invalid-tls-secret-password")?; + let invalid = Client::from_conf(invalid.config().to_builder().retry_config(RetryConfig::disabled()).build()); + let denied = invalid + .list_buckets() + .send() + .await + .expect_err("alternate credentials must not inherit the root identity"); + assert_eq!(denied.raw_response().map(|response| response.status().as_u16()), Some(403)); + } + for expected in &object.reads { + for (node, client) in clients.iter().enumerate() { + results.push(verified_get(client, node, &bucket, &object, expected, false).await?); + } + } + results.extend( + try_join_all( + clients + .iter() + .enumerate() + .map(|(node, client)| verified_get(client, node, &bucket, &object, &object.reads[0], true)), + ) + .await?, + ); + tokio::fs::remove_file(&object.path).await?; + } + for node in 0..clients.len() { + let log = tokio::fs::read_to_string(log_directory.join(format!("node-{node}.log"))).await?; + assert!( + !log.contains("internode body stalled"), + "node {node}: internode body stall during verified reads" + ); + } + let report = serde_json::json!({ + "schema": 1, "mode": mode, "tls_verified": true, "nodes": 4, "drives_per_node": 4, + "source_revision": env!("RUSTFS_E2E_BUILD_COMMIT"), + "source_dirty": env!("RUSTFS_E2E_BUILD_DIRTY"), + "scanner_enabled": true, "heal_enabled": true, + "scanner_progress_verified": false, "heal_progress_verified": false, + "results": results, + }); + if let Some(path) = report_path { + std::fs::write(path, serde_json::to_vec_pretty(&report)?)?; + } + eprintln!("internode_tls_acceptance={report}"); + Ok(()) +} diff --git a/crates/e2e_test/src/distributed/mod.rs b/crates/e2e_test/src/distributed/mod.rs index eb01e118b..f627b339b 100644 --- a/crates/e2e_test/src/distributed/mod.rs +++ b/crates/e2e_test/src/distributed/mod.rs @@ -26,6 +26,7 @@ mod expand_decommission_rebalance_test; mod extra_test; mod harness; mod heal_test; +mod internode_tls_test; mod multipart_fence_acceptance_test; mod object_lock_test; mod observability_test; diff --git a/docs/operations/internode-http-transport.md b/docs/operations/internode-http-transport.md index 19a2d7942..7ac9741f1 100644 --- a/docs/operations/internode-http-transport.md +++ b/docs/operations/internode-http-transport.md @@ -14,12 +14,32 @@ This setting covers internode shard reads, writes, and directory walks. It does Set the protocol on every node and restart the processes. The node initiating a read, scan, or heal chooses the outbound protocol. With `http1`, retain the normal connection-pool settings; there is no need to set the maximum idle count to zero to isolate active bodies. +During a rolling upgrade, keep `RUSTFS_INTERNODE_HTTP_POOL_MAX_IDLE_PER_HOST=0` on older nodes that lack the protocol selector. Their coordinators, scanner and heal workers still choose the older outbound policy. After a node runs a release with the selector, use `http1` and restore its normal idle-pool setting. New response framing alone does not isolate streams initiated by an older HTTP/2 client. + HTTP/2 window and adaptive-window settings apply only with `auto` and a TLS connection that negotiates HTTP/2. The `legacy`, `balanced`, and `throughput` tuning profiles continue to control their existing pool and window settings; none overrides the protocol policy. HTTP/1.1 uses more concurrent connections than multiplexed HTTP/2. Size file-descriptor and connection budgets for the workload. For bounded reads, consuming the full advertised byte count completes HTTP framing and permits reuse without an additional EOF poll. Unknown-length reads still require EOF. Cancelled or partially consumed responses can close their connections. A bounded response that ends early is a transport error; the server retains its existing read cap and never pads or shrinks a short response into success. Disabling pooling also adds TCP/TLS setup and connection churn to subsequent requests. No universal throughput improvement is implied by the protocol default. For validation, compare complete 100 MiB and 1 GiB GETs through every coordinator, verify lengths and hashes, and include concurrent GET, scanner, and heal activity. Record latency, stall errors, CPU, connection counts, and file descriptors. Include small objects and uploads when assessing a deployment's performance. Local transport tests exercise TLS ALPN, shared HTTP/2 backpressure, full-body delivery, and HTTP/1.1 idle reuse; they do not replace a distributed cluster run. +The distributed E2E case below starts four processes with four drives each and a generated CA. Both S3 and volume endpoints use verified TLS. It streams 32 KiB, 100 MiB and 1 GiB uploads from files, checks complete and unaligned Range responses through every coordinator, and repeats full reads with all four coordinators active concurrently. Each GET checks Content-Length, Content-Range where applicable, the received byte count and SHA-256. SDK GET retries are disabled, and captured node logs must contain no internode body stalls. + +```bash +export RUSTUP_TOOLCHAIN=stable +export NO_PROXY=127.0.0.1,localhost HTTP_PROXY= HTTPS_PROXY= +python3 scripts/e2e_binary.py build --features e2e-test-hooks +mkdir -p target/internode-acceptance +RUSTFS_E2E_INTERNODE_HTTP_MODE=http1 \ +RUSTFS_INTERNODE_ACCEPTANCE_REPORT="$PWD/target/internode-acceptance/http1.json" \ +python3 scripts/e2e_binary.py run --features e2e-test-hooks -- \ + cargo nextest run -p e2e_test --profile e2e-distributed \ + -E 'test(four_node_tls_large_and_range_gets_preserve_bytes)' --no-capture +``` + +Use an unused report filename for each invocation. The report has per-request timings and integrity results; adjacent files retain node logs and process IDs for external resource sampling. `RUSTFS_E2E_INTERNODE_HTTP_MODE=auto` selects pooled protocol negotiation, and `unpooled` selects the same negotiation with the idle pool disabled. Scanner and heal are enabled, but this case does not inject damage or certify their progress. A passing result does not complete mixed-version, mTLS or certificate-rotation acceptance. + +These correctness timings are not a capacity benchmark: the client hashes every byte and shares a host with the four servers. For performance attribution, use the same release binary, topology and workload in isolated ABBA runs, gather CPU, FD, connections and TIME_WAIT externally, and reject candidate conclusions when the repeated baseline drifts. A few correctness requests cannot establish p99 or universal protocol superiority. + The mechanism follows [HTTP/2 flow control](https://www.rfc-editor.org/rfc/rfc9113.html#name-flow-control): stream and connection DATA credit both apply. In the client dependency, credit is released when the application polls body frames. Multiple paused bodies can therefore consume shared connection credit required by a stream the caller is still waiting for. Increasing a finite window or retrying the GET does not remove this coupling. Control frames and response headers are not subject to DATA credit. S3 [GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html) defines the returned object body, content length, and byte-range semantics; it does not require HTTP/2 for internode transport. This policy changes the transport below the existing integrity, quorum, timeout, and identity-fenced resume behavior. diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 76b4fa3f5..72b8a5669 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -714,7 +714,7 @@ fn read_file_body_stream( reader: R, length: usize, operation: &'static str, -) -> Pin> + Send + Sync>> +) -> impl futures::Stream> + Send + Sync where R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static, { @@ -725,11 +725,10 @@ where } else { u64::try_from(length).unwrap_or(u64::MAX) }; - let stream = ReaderStream::with_capacity(reader.take(read_limit), read_buffer_size).map_ok(move |bytes| { + ReaderStream::with_capacity(reader.take(read_limit), read_buffer_size).map_ok(move |bytes| { metrics.record_sent_bytes_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, bytes.len()); bytes - }); - Box::pin(stream) + }) } fn read_file_stream_buffer_size(length: usize) -> usize {