diff --git a/Cargo.lock b/Cargo.lock index b8204f597..62aa4e4ed 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7914,6 +7914,7 @@ name = "rustfs-rio" version = "0.0.5" dependencies = [ "aes-gcm", + "axum", "base64 0.22.1", "bytes", "crc-fast", @@ -7921,10 +7922,12 @@ dependencies = [ "futures", "hex-simd", "http 1.4.0", + "http-body-util", "md-5 0.11.0-rc.5", "pin-project-lite", "rand 0.10.0", "reqwest 0.13.2", + "rustfs-common", "rustfs-config", "rustfs-utils", "s3s", diff --git a/crates/audit/src/registry.rs b/crates/audit/src/registry.rs index e3620b668..c6eccae57 100644 --- a/crates/audit/src/registry.rs +++ b/crates/audit/src/registry.rs @@ -313,6 +313,12 @@ impl AuditRegistry { } } + if &new_config == config { + info!("Audit target configuration unchanged, skip persisting server config"); + info!(count = successful_targets.len(), "All target processing completed"); + return Ok(successful_targets); + } + let Some(store) = rustfs_ecstore::global::new_object_layer_fn() else { return Err(AuditError::StorageNotAvailable( "Failed to save target configuration: server storage not initialized".to_string(), diff --git a/crates/audit/tests/performance_test.rs b/crates/audit/tests/performance_test.rs index b96e92eb3..47780abc9 100644 --- a/crates/audit/tests/performance_test.rs +++ b/crates/audit/tests/performance_test.rs @@ -159,7 +159,8 @@ async fn test_audit_log_dispatch_performance() { let start = Instant::now(); - // Dispatch audit log (should be fast since no targets are configured) + // Dispatch audit log against an unstarted system state. Empty config keeps + // the audit system stopped, so dispatch should fail fast without targets. let result = system.dispatch(Arc::new(audit_entry)).await; let elapsed = start.elapsed(); @@ -168,8 +169,10 @@ async fn test_audit_log_dispatch_performance() { // Should be very fast (sub-millisecond for no targets) assert!(elapsed < Duration::from_millis(100), "Dispatch took too long: {elapsed:?}"); - // Should succeed even with no targets - assert!(result.is_ok(), "Dispatch should succeed with no targets"); + assert!( + matches!(result, Err(AuditError::NotInitialized(_))), + "Dispatch on a stopped system should return NotInitialized, got: {result:?}" + ); // Clean up let _ = system.close().await; @@ -186,11 +189,11 @@ async fn test_system_state_transitions() { let config = rustfs_ecstore::config::Config(std::collections::HashMap::new()); let start_result = system.start(config).await; - // Should be running (or failed due to server storage) + // Empty config keeps the audit system stopped even when start() succeeds. let state = system.get_state().await; match start_result { Ok(_) => { - assert_eq!(state, rustfs_audit::system::AuditSystemState::Running); + assert_eq!(state, rustfs_audit::system::AuditSystemState::Stopped); } Err(_) => { // Expected in test environment due to server storage not being initialized diff --git a/crates/audit/tests/system_integration_test.rs b/crates/audit/tests/system_integration_test.rs index d60c6f18b..d108227e0 100644 --- a/crates/audit/tests/system_integration_test.rs +++ b/crates/audit/tests/system_integration_test.rs @@ -29,27 +29,21 @@ async fn test_complete_audit_system_lifecycle() { assert_eq!(system.get_state().await, system::AuditSystemState::Stopped); assert!(!system.is_running().await); - // 2. Start with empty config (will fail due to no server storage in test) + // 2. Start with empty config. The current implementation returns Ok(()) + // but keeps the system stopped when no audit targets are enabled. let config = Config(HashMap::new()); let start_result = system.start(config).await; - // Should fail in test environment but state handling should work + // State handling should remain consistent for both empty-config success and + // storage-unavailable failure paths. match start_result { Err(AuditError::StorageNotAvailable(_)) => { // Expected in test environment assert_eq!(system.get_state().await, system::AuditSystemState::Stopped); } Ok(_) => { - // If it somehow succeeds, verify running state - assert_eq!(system.get_state().await, system::AuditSystemState::Running); - assert!(system.is_running().await); - - // Test pause/resume - system.pause().await.expect("Should pause successfully"); - assert_eq!(system.get_state().await, system::AuditSystemState::Paused); - - system.resume().await.expect("Should resume successfully"); - assert_eq!(system.get_state().await, system::AuditSystemState::Running); + assert_eq!(system.get_state().await, system::AuditSystemState::Stopped); + assert!(!system.is_running().await); } Err(e) => { panic!("Unexpected error: {e}"); diff --git a/crates/common/src/internode_metrics.rs b/crates/common/src/internode_metrics.rs new file mode 100644 index 000000000..2cae60253 --- /dev/null +++ b/crates/common/src/internode_metrics.rs @@ -0,0 +1,170 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use metrics::{counter, gauge}; +use std::sync::{ + Arc, LazyLock, + atomic::{AtomicU64, Ordering}, +}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct InternodeMetricsSnapshot { + pub sent_bytes_total: u64, + pub recv_bytes_total: u64, + pub outgoing_requests_total: u64, + pub incoming_requests_total: u64, + pub errors_total: u64, + pub dial_errors_total: u64, + pub dial_avg_time_nanos: u64, + pub last_dial_unix_millis: u64, +} + +#[derive(Debug, Default)] +pub struct InternodeMetrics { + sent_bytes_total: AtomicU64, + recv_bytes_total: AtomicU64, + outgoing_requests_total: AtomicU64, + incoming_requests_total: AtomicU64, + errors_total: AtomicU64, + dial_errors_total: AtomicU64, + dial_total_time_nanos: AtomicU64, + dial_samples_total: AtomicU64, + last_dial_unix_millis: AtomicU64, +} + +impl InternodeMetrics { + pub fn record_sent_bytes(&self, bytes: usize) { + let bytes = bytes as u64; + if bytes == 0 { + return; + } + self.sent_bytes_total.fetch_add(bytes, Ordering::Relaxed); + counter!("rustfs.internode.sent.bytes.total").increment(bytes); + } + + pub fn record_recv_bytes(&self, bytes: usize) { + let bytes = bytes as u64; + if bytes == 0 { + return; + } + self.recv_bytes_total.fetch_add(bytes, Ordering::Relaxed); + counter!("rustfs.internode.recv.bytes.total").increment(bytes); + } + + pub fn record_outgoing_request(&self) { + self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed); + counter!("rustfs.internode.requests.outgoing.total").increment(1); + } + + pub fn record_incoming_request(&self) { + self.incoming_requests_total.fetch_add(1, Ordering::Relaxed); + counter!("rustfs.internode.requests.incoming.total").increment(1); + } + + pub fn record_error(&self) { + self.errors_total.fetch_add(1, Ordering::Relaxed); + counter!("rustfs.internode.errors.total").increment(1); + } + + pub fn record_dial_result(&self, duration: Duration, success: bool) { + let elapsed_nanos = duration.as_nanos().min(u128::from(u64::MAX)) as u64; + self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed); + let samples = self.dial_samples_total.fetch_add(1, Ordering::Relaxed) + 1; + let total = self.dial_total_time_nanos.load(Ordering::Relaxed); + gauge!("rustfs.internode.dial.avg_time.nanos").set(total as f64 / samples as f64); + + if !success { + self.dial_errors_total.fetch_add(1, Ordering::Relaxed); + counter!("rustfs.internode.dial.errors.total").increment(1); + } + + let now_ms = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(u128::from(u64::MAX)) as u64; + self.last_dial_unix_millis.store(now_ms, Ordering::Relaxed); + } + + pub fn snapshot(&self) -> InternodeMetricsSnapshot { + let dial_samples_total = self.dial_samples_total.load(Ordering::Relaxed); + let dial_total_time_nanos = self.dial_total_time_nanos.load(Ordering::Relaxed); + let dial_avg_time_nanos = if dial_samples_total == 0 { + 0 + } else { + dial_total_time_nanos / dial_samples_total + }; + + InternodeMetricsSnapshot { + sent_bytes_total: self.sent_bytes_total.load(Ordering::Relaxed), + recv_bytes_total: self.recv_bytes_total.load(Ordering::Relaxed), + outgoing_requests_total: self.outgoing_requests_total.load(Ordering::Relaxed), + incoming_requests_total: self.incoming_requests_total.load(Ordering::Relaxed), + errors_total: self.errors_total.load(Ordering::Relaxed), + dial_errors_total: self.dial_errors_total.load(Ordering::Relaxed), + dial_avg_time_nanos, + last_dial_unix_millis: self.last_dial_unix_millis.load(Ordering::Relaxed), + } + } + + #[doc(hidden)] + pub fn reset_for_test(&self) { + self.sent_bytes_total.store(0, Ordering::Relaxed); + self.recv_bytes_total.store(0, Ordering::Relaxed); + self.outgoing_requests_total.store(0, Ordering::Relaxed); + self.incoming_requests_total.store(0, Ordering::Relaxed); + self.errors_total.store(0, Ordering::Relaxed); + self.dial_errors_total.store(0, Ordering::Relaxed); + self.dial_total_time_nanos.store(0, Ordering::Relaxed); + self.dial_samples_total.store(0, Ordering::Relaxed); + self.last_dial_unix_millis.store(0, Ordering::Relaxed); + } +} + +pub fn global_internode_metrics() -> &'static Arc { + static GLOBAL_INTERNODE_METRICS: LazyLock> = LazyLock::new(|| Arc::new(InternodeMetrics::default())); + &GLOBAL_INTERNODE_METRICS +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn snapshot_reports_recorded_values() { + let metrics = global_internode_metrics(); + metrics.reset_for_test(); + + metrics.record_sent_bytes(64); + metrics.record_recv_bytes(32); + metrics.record_outgoing_request(); + metrics.record_incoming_request(); + metrics.record_error(); + metrics.record_dial_result(Duration::from_millis(9), true); + metrics.record_dial_result(Duration::from_millis(3), false); + + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.sent_bytes_total, 64); + assert_eq!(snapshot.recv_bytes_total, 32); + assert_eq!(snapshot.outgoing_requests_total, 1); + assert_eq!(snapshot.incoming_requests_total, 1); + assert_eq!(snapshot.errors_total, 1); + assert_eq!(snapshot.dial_errors_total, 1); + assert_eq!(snapshot.dial_avg_time_nanos, 6_000_000); + assert!(snapshot.last_dial_unix_millis > 0); + + metrics.reset_for_test(); + } +} diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index c239d4b37..3c532fcf0 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -17,6 +17,7 @@ pub mod bucket_stats; pub mod data_usage; pub mod globals; pub mod heal_channel; +pub mod internode_metrics; pub mod last_minute; pub mod metrics; mod readiness; diff --git a/crates/e2e_test/src/reliant/grpc_lock_client.rs b/crates/e2e_test/src/reliant/grpc_lock_client.rs index dc89af754..fc772d520 100644 --- a/crates/e2e_test/src/reliant/grpc_lock_client.rs +++ b/crates/e2e_test/src/reliant/grpc_lock_client.rs @@ -61,6 +61,7 @@ impl GrpcLockClient { metadata: LockMetadata::default(), priority: LockPriority::Normal, deadlock_detection: false, + suppress_contention_logs: false, } } } diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index fb8b17b68..7a1a10657 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -15,6 +15,7 @@ use crate::config::{Config, GLOBAL_STORAGE_CLASS, storageclass}; use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; +use crate::global::is_first_cluster_node_local; use crate::store_api::{ObjectInfo, ObjectOptions, PutObjReader, StorageAPI}; use http::HeaderMap; use rustfs_config::{DEFAULT_DELIMITER, RUSTFS_REGION}; @@ -44,6 +45,19 @@ pub async fn read_config(api: Arc, file: &str) -> Result(api: Arc, file: &str) -> Result> { + let (data, _obj) = read_config_with_metadata( + api, + file, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await?; + Ok(data) +} + pub async fn read_config_with_metadata( api: Arc, file: &str, @@ -287,7 +301,14 @@ fn is_object_not_found(err: &Error) -> bool { pub async fn try_migrate_server_config(api: Arc) { let config_file = get_config_file(); match api - .get_object_info(RUSTFS_META_BUCKET, &config_file, &ObjectOptions::default()) + .get_object_info( + RUSTFS_META_BUCKET, + &config_file, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await { Ok(_) => { @@ -360,7 +381,13 @@ pub async fn try_migrate_server_config(api: Arc) { /// Handle the situation where the configuration file does not exist, create and save a new configuration async fn handle_missing_config(api: Arc, context: &str) -> Result { warn!("Configuration not found ({}): Start initializing new configuration", context); - let cfg = new_and_save_server_config(api).await?; + let cfg = if is_first_cluster_node_local().await { + new_and_save_server_config(api.clone()).await? + } else { + let mut cfg = new_server_config(); + lookup_configs(&mut cfg, api).await; + cfg + }; warn!("Configuration initialization complete ({})", context); Ok(cfg) } @@ -375,7 +402,7 @@ pub async fn read_config_without_migrate(api: Arc) -> Result read_server_config(api, &data).await, Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration").await, Err(err) => handle_config_read_error(err, &config_file), @@ -389,7 +416,7 @@ async fn read_server_config(api: Arc, data: &[u8]) -> Result { // TODO: decrypt let cfg = decode_server_config_blob(&cfg_data)?; diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs index be2c26cda..ca46e0297 100644 --- a/crates/ecstore/src/config/mod.rs +++ b/crates/ecstore/src/config/mod.rs @@ -79,7 +79,7 @@ pub async fn try_migrate_server_config(api: Arc) { com::try_migrate_server_config(api).await } -#[derive(Debug, Deserialize, Serialize, Clone)] +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)] pub struct KV { pub key: String, pub value: String, @@ -87,7 +87,7 @@ pub struct KV { pub hidden_if_empty: bool, } -#[derive(Debug, Deserialize, Serialize, Clone)] +#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Eq)] pub struct KVS(pub Vec); impl Default for KVS { @@ -163,7 +163,7 @@ impl KVS { } } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub struct Config(pub HashMap>); impl Default for Config { diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index f35e9c65b..7f5a8541e 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -18,6 +18,7 @@ use crate::disk::{ WalkDirOptions, local::{LocalDisk, ScanGuard}, }; +use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; use bytes::Bytes; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use std::{ @@ -410,7 +411,19 @@ impl LocalDiskWrapper { /// Set the disk ID pub async fn set_disk_id_internal(&self, id: Option) -> Result<()> { let mut disk_id = self.disk_id.write().await; + let previous = *disk_id; *disk_id = id; + drop(disk_id); + + if self.disk.is_local() { + let mut disk_id_map = GLOBAL_LOCAL_DISK_ID_MAP.write().await; + if let Some(previous_id) = previous { + disk_id_map.remove(&previous_id); + } + if let Some(current_id) = id { + disk_id_map.insert(current_id, self.disk.endpoint().to_string()); + } + } Ok(()) } diff --git a/crates/ecstore/src/erasure_coding/bitrot.rs b/crates/ecstore/src/erasure_coding/bitrot.rs index 4f4b7f1e7..01e2f4c3c 100644 --- a/crates/ecstore/src/erasure_coding/bitrot.rs +++ b/crates/ecstore/src/erasure_coding/bitrot.rs @@ -15,6 +15,7 @@ use bytes::Bytes; use pin_project_lite::pin_project; use rustfs_utils::HashAlgorithm; +use std::io::IoSlice; use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use tracing::error; use uuid::Uuid; @@ -155,23 +156,49 @@ where error!("bitrot writer write hash error: hash is empty"); return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, "hash is empty")); } - self.inner.write_all(hash.as_ref()).await?; + write_all_vectored(&mut self.inner, hash.as_ref(), buf).await?; + } else { + self.inner.write_all(buf).await?; } - self.inner.write_all(buf).await?; - - self.inner.flush().await?; - let n = buf.len(); Ok(n) } pub async fn shutdown(&mut self) -> std::io::Result<()> { + self.inner.flush().await?; self.inner.shutdown().await } } +async fn write_all_vectored(writer: &mut W, hash: &[u8], data: &[u8]) -> std::io::Result<()> +where + W: AsyncWrite + Unpin, +{ + let mut hash_offset = 0; + let mut data_offset = 0; + + while hash_offset < hash.len() || data_offset < data.len() { + let slices = [IoSlice::new(&hash[hash_offset..]), IoSlice::new(&data[data_offset..])]; + let written = writer.write_vectored(&slices).await?; + if written == 0 { + return Err(std::io::Error::new(std::io::ErrorKind::WriteZero, "failed to write hash and data")); + } + + let hash_remaining = hash.len() - hash_offset; + if written < hash_remaining { + hash_offset += written; + continue; + } + + hash_offset = hash.len(); + data_offset += written - hash_remaining; + } + + Ok(()) +} + pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: HashAlgorithm) -> usize { if algo != HashAlgorithm::HighwayHash256S && algo != HashAlgorithm::HighwayHash256SLegacy { return size; @@ -292,6 +319,33 @@ impl AsyncWrite for CustomWriter { } } } + + fn poll_write_vectored( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + bufs: &[IoSlice<'_>], + ) -> std::task::Poll> { + match self.get_mut() { + Self::InlineBuffer(data) => { + let total = bufs.iter().map(|buf| buf.len()).sum::(); + for buf in bufs { + data.extend_from_slice(buf); + } + std::task::Poll::Ready(Ok(total)) + } + Self::Other(writer) => { + let pinned_writer = std::pin::Pin::new(writer.as_mut()); + pinned_writer.poll_write_vectored(cx, bufs) + } + } + } + + fn is_write_vectored(&self) -> bool { + match self { + Self::InlineBuffer(_) => true, + Self::Other(writer) => writer.is_write_vectored(), + } + } } /// Wrapper around BitrotWriter that uses our custom writer @@ -361,7 +415,74 @@ mod tests { use super::BitrotReader; use super::BitrotWriter; use rustfs_utils::HashAlgorithm; - use std::io::Cursor; + use std::io::{Cursor, IoSlice}; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + use std::task::{Context, Poll}; + use tokio::io::AsyncWrite; + + #[derive(Default)] + struct VectoredCountingWriter { + vectored_writes: Arc, + writes: Vec, + } + + impl AsyncWrite for VectoredCountingWriter { + fn poll_write(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { + Poll::Ready(Err(std::io::Error::other("poll_write should not be used"))) + } + + fn poll_flush(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_write_vectored( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut Context<'_>, + bufs: &[IoSlice<'_>], + ) -> Poll> { + self.vectored_writes.fetch_add(1, Ordering::SeqCst); + let total = bufs.iter().map(|buf| buf.len()).sum::(); + for buf in bufs { + self.writes.extend_from_slice(buf); + } + Poll::Ready(Ok(total)) + } + + fn is_write_vectored(&self) -> bool { + true + } + } + + #[derive(Default)] + struct CountingWriter { + flushes: Arc, + shutdowns: Arc, + writes: Vec, + } + + impl AsyncWrite for CountingWriter { + fn poll_write(mut self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.writes.extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + self.flushes.fetch_add(1, Ordering::SeqCst); + Poll::Ready(Ok(())) + } + + fn poll_shutdown(self: std::pin::Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + self.shutdowns.fetch_add(1, Ordering::SeqCst); + Poll::Ready(Ok(())) + } + } #[tokio::test] async fn test_bitrot_read_write_ok() { @@ -471,4 +592,41 @@ mod tests { assert_eq!(n, data_size); assert_eq!(data, &out[..]); } + + #[tokio::test] + async fn test_bitrot_writer_flushes_once_on_shutdown() { + let flushes = Arc::new(AtomicUsize::new(0)); + let shutdowns = Arc::new(AtomicUsize::new(0)); + let writer = CountingWriter { + flushes: flushes.clone(), + shutdowns: shutdowns.clone(), + writes: Vec::new(), + }; + let mut bitrot_writer = BitrotWriter::new(writer, 8, HashAlgorithm::None); + + bitrot_writer.write(b"12345678").await.unwrap(); + bitrot_writer.write(b"abc").await.unwrap(); + + assert_eq!(flushes.load(Ordering::SeqCst), 0); + assert_eq!(shutdowns.load(Ordering::SeqCst), 0); + + bitrot_writer.shutdown().await.unwrap(); + + assert_eq!(flushes.load(Ordering::SeqCst), 1); + assert_eq!(shutdowns.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn test_bitrot_writer_uses_vectored_write_for_hash_and_data() { + let vectored_writes = Arc::new(AtomicUsize::new(0)); + let writer = VectoredCountingWriter { + vectored_writes: vectored_writes.clone(), + writes: Vec::new(), + }; + let mut bitrot_writer = BitrotWriter::new(writer, 8, HashAlgorithm::HighwayHash256); + + bitrot_writer.write(b"payload").await.unwrap(); + + assert!(vectored_writes.load(Ordering::SeqCst) > 0); + } } diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index cb7bd7ed4..e029f64a4 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -112,11 +112,66 @@ impl<'a> MultiWriter<'a> { ))) } - pub async fn _shutdown(&mut self) -> std::io::Result<()> { - for writer in self.writers.iter_mut().flatten() { - writer.shutdown().await?; + async fn shutdown_writer(writer_opt: &mut Option, err: &mut Option) { + match writer_opt { + Some(writer) => match writer.shutdown().await { + Ok(()) => { + *err = None; + } + Err(e) => { + *err = Some(Error::from(e)); + *writer_opt = None; + } + }, + None => { + *err = Some(Error::DiskNotFound); + } } - Ok(()) + } + + pub async fn shutdown(&mut self) -> std::io::Result<()> { + { + let mut futures = FuturesUnordered::new(); + for (writer_opt, err) in self.writers.iter_mut().zip(self.errs.iter_mut()) { + if err.is_some() { + continue; + } + futures.push(Self::shutdown_writer(writer_opt, err)); + } + while let Some(()) = futures.next().await {} + } + + let nil_count = self.errs.iter().filter(|&e| e.is_none()).count(); + if nil_count >= self.write_quorum { + return Ok(()); + } + + if let Some(write_err) = reduce_write_quorum_errs(&self.errs, OBJECT_OP_IGNORED_ERRS, self.write_quorum) { + error!( + "reduce_write_quorum_errs during shutdown: {:?}, offline-disks={}/{}, errs={:?}", + write_err, + count_errs(&self.errs, &Error::DiskNotFound), + self.writers.len(), + self.errs + ); + return Err(std::io::Error::other(format!( + "Failed to shutdown writers: {} (offline-disks={}/{})", + write_err, + count_errs(&self.errs, &Error::DiskNotFound), + self.writers.len() + ))); + } + + Err(std::io::Error::other(format!( + "Failed to shutdown writers: (offline-disks={}/{}): {}", + count_errs(&self.errs, &Error::DiskNotFound), + self.writers.len(), + self.errs + .iter() + .map(|e| e.as_ref().map_or("".to_string(), |e| e.to_string())) + .collect::>() + .join(", ") + ))) } } @@ -176,7 +231,69 @@ impl Erasure { } let (reader, total) = task.await??; - // writers.shutdown().await?; + writers.shutdown().await?; Ok((reader, total)) } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::erasure_coding::{BitrotWriterWrapper, CustomWriter}; + use rustfs_utils::HashAlgorithm; + use std::pin::Pin; + use std::sync::{Arc, Mutex}; + use std::task::{Context, Poll}; + use tokio::io::AsyncWrite; + + #[derive(Clone, Default)] + struct DeferredCommitWriter { + buffered: Vec, + committed: Arc>>, + } + + impl DeferredCommitWriter { + fn new(committed: Arc>>) -> Self { + Self { + buffered: Vec::new(), + committed, + } + } + } + + impl AsyncWrite for DeferredCommitWriter { + fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + self.buffered.extend_from_slice(buf); + Poll::Ready(Ok(buf.len())) + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + let buffered = std::mem::take(&mut self.buffered); + let mut committed = self.committed.lock().unwrap(); + committed.extend_from_slice(&buffered); + Poll::Ready(Ok(())) + } + } + + #[tokio::test] + async fn encode_shutdowns_writers_after_small_shards() { + let committed = Arc::new(Mutex::new(Vec::new())); + let writer = DeferredCommitWriter::new(committed.clone()); + let mut writers = vec![Some(BitrotWriterWrapper::new( + CustomWriter::new_tokio_writer(writer), + 16, + HashAlgorithm::HighwayHash256S, + ))]; + + let erasure = Arc::new(Erasure::new(1, 0, 16)); + let reader = tokio::io::BufReader::new(std::io::Cursor::new(b"small payload".to_vec())); + let (_reader, written) = erasure.encode(reader, &mut writers, 1).await.unwrap(); + + assert_eq!(written, b"small payload".len()); + assert!(!committed.lock().unwrap().is_empty()); + } +} diff --git a/crates/ecstore/src/global.rs b/crates/ecstore/src/global.rs index 559f109c2..c6ce63408 100644 --- a/crates/ecstore/src/global.rs +++ b/crates/ecstore/src/global.rs @@ -46,6 +46,7 @@ lazy_static! { pub static ref GLOBAL_IsDistErasure: RwLock = RwLock::new(false); pub static ref GLOBAL_IsErasureSD: RwLock = RwLock::new(false); pub static ref GLOBAL_LOCAL_DISK_MAP: Arc>>> = Arc::new(RwLock::new(HashMap::new())); + pub static ref GLOBAL_LOCAL_DISK_ID_MAP: Arc>> = Arc::new(RwLock::new(HashMap::new())); pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc> = Arc::new(RwLock::new(Vec::new())); pub static ref GLOBAL_Endpoints: OnceLock = OnceLock::new(); pub static ref GLOBAL_RootDiskThreshold: RwLock = RwLock::new(0); @@ -153,6 +154,10 @@ pub fn get_global_endpoints_opt() -> Option { GLOBAL_Endpoints.get().cloned() } +pub async fn is_first_cluster_node_local() -> bool { + get_global_endpoints().first_local() +} + pub fn get_global_tier_config_mgr() -> Arc> { GLOBAL_TierConfigMgr.clone() } diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index 8f1b159d1..0d1fa07a8 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -14,8 +14,11 @@ use crate::{admin_server_info::get_local_server_property, new_object_layer_fn, store_api::StorageAPI}; use chrono::Utc; -use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics}; -use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics}; +use rustfs_common::{ + GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, internode_metrics::global_internode_metrics, + metrics::global_metrics, +}; +use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, NetDevLine, NetMetrics, RPCMetrics, RealtimeMetrics}; use rustfs_utils::os::get_drive_stats; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; @@ -120,13 +123,50 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts) // if types.contains(&MetricType::SITE_RESYNC) {} - // if types.contains(&MetricType::NET) {} + if types.contains(&MetricType::NET) { + let snapshot = global_internode_metrics().snapshot(); + real_time_metrics.aggregated.net = Some(NetMetrics { + collected_at: Utc::now(), + interface_name: "internode".to_string(), + net_stats: NetDevLine { + name: "internode".to_string(), + rx_bytes: snapshot.recv_bytes_total, + tx_bytes: snapshot.sent_bytes_total, + ..Default::default() + }, + }); + } // if types.contains(&MetricType::MEM) {} // if types.contains(&MetricType::CPU) {} - // if types.contains(&MetricType::RPC) {} + if types.contains(&MetricType::RPC) { + let collected_at = Utc::now(); + let snapshot = global_internode_metrics().snapshot(); + let last_connect_time = + chrono::DateTime::::from_timestamp_millis(snapshot.last_dial_unix_millis as i64).unwrap_or(collected_at); + + real_time_metrics.aggregated.rpc = Some(RPCMetrics { + collected_at, + connected: i32::from(snapshot.last_dial_unix_millis > 0), + reconnect_count: snapshot.dial_errors_total.min(i32::MAX as u64) as i32, + disconnected: 0, + outgoing_streams: 0, + incoming_streams: 0, + outgoing_bytes: snapshot.sent_bytes_total.min(i64::MAX as u64) as i64, + incoming_bytes: snapshot.recv_bytes_total.min(i64::MAX as u64) as i64, + outgoing_messages: snapshot.outgoing_requests_total.min(i64::MAX as u64) as i64, + incoming_messages: snapshot.incoming_requests_total.min(i64::MAX as u64) as i64, + out_queue: 0, + last_pong_time: collected_at, + last_ping_ms: snapshot.dial_avg_time_nanos as f64 / 1_000_000.0, + max_ping_dur_ms: snapshot.dial_avg_time_nanos as f64 / 1_000_000.0, + last_connect_time, + by_destination: None, + by_caller: None, + }); + } real_time_metrics .by_host @@ -211,7 +251,9 @@ async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap 0.0); + + metrics.reset_for_test(); + } } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index eb17ea583..a931505e5 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -23,6 +23,7 @@ use crate::disk::{ }; use crate::disk::{disk_store::DiskHealthTracker, error::DiskError, local::ScanGuard}; use crate::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client}; +use crate::set_disk::DEFAULT_READ_BUFFER_SIZE; use crate::{ disk::error::{Error, Result}, rpc::build_auth_headers, @@ -40,7 +41,9 @@ use rustfs_protos::proto_gen::node_service::{ node_service_client::NodeServiceClient, }; use rustfs_rio::{HttpReader, HttpWriter}; +use serde::{Serialize, de::DeserializeOwned}; use std::{ + io::Cursor, path::PathBuf, sync::{ Arc, @@ -49,12 +52,36 @@ use std::{ time::Duration, }; use tokio::time; -use tokio::{io::AsyncWrite, net::TcpStream, time::timeout}; +use tokio::{ + io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::TcpStream, + time::timeout, +}; use tokio_util::sync::CancellationToken; use tonic::{Request, service::interceptor::InterceptedService, transport::Channel}; use tracing::{debug, info, warn}; use uuid::Uuid; +async fn copy_stream_with_buffer(reader: &mut R, writer: &mut W, buffer_size: usize) -> io::Result +where + R: AsyncRead + Unpin, + W: AsyncWrite + Unpin, +{ + let mut copied = 0_u64; + let mut buffer = vec![0_u8; buffer_size]; + + loop { + let bytes_read = reader.read(&mut buffer).await?; + if bytes_read == 0 { + writer.flush().await?; + return Ok(copied); + } + + writer.write_all(&buffer[..bytes_read]).await?; + copied += bytes_read as u64; + } +} + #[derive(Debug)] pub struct RemoteDisk { pub id: Mutex>, @@ -259,6 +286,27 @@ impl RemoteDisk { .await .map_err(|err| Error::other(format!("can not get client, err: {err}"))) } + + async fn disk_ref(&self) -> String { + (*self.id.lock().await) + .map(|id| id.to_string()) + .unwrap_or_else(|| self.endpoint.to_string()) + } +} + +fn encode_msgpack(value: &T) -> Result> { + let mut serializer = rmp_serde::Serializer::new(Vec::new()); + value.serialize(&mut serializer)?; + Ok(serializer.into_inner()) +} + +fn decode_msgpack_or_json(binary: &[u8], json: &str) -> Result { + if !binary.is_empty() { + let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); + return T::deserialize(&mut deserializer).map_err(Error::from); + } + + serde_json::from_str(json).map_err(Error::from) } // TODO: all api need to handle errors @@ -707,18 +755,21 @@ impl DiskAPI for RemoteDisk { async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { info!("write_metadata {}/{}", volume, path); let file_info = serde_json::to_string(&fi)?; + let file_info_bin = encode_msgpack(&fi)?; self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(WriteMetadataRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), file_info: file_info.clone(), + file_info_bin: file_info_bin.clone(), }); let response = client.write_metadata(request).await?.into_inner(); @@ -735,6 +786,7 @@ impl DiskAPI for RemoteDisk { } async fn read_metadata(&self, volume: &str, path: &str) -> Result { + let disk = self.disk_ref().await; let mut client = self .get_client() .await @@ -742,7 +794,7 @@ impl DiskAPI for RemoteDisk { let request = Request::new(ReadMetadataRequest { volume: volume.to_string(), path: path.to_string(), - disk: self.endpoint.to_string(), + disk, }); let response = client.read_metadata(request).await?.into_inner(); @@ -759,19 +811,24 @@ impl DiskAPI for RemoteDisk { info!("update_metadata"); let file_info = serde_json::to_string(&fi)?; let opts_str = serde_json::to_string(&opts)?; + let file_info_bin = encode_msgpack(&fi)?; + let opts_bin = encode_msgpack(opts)?; self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(UpdateMetadataRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), file_info: file_info.clone(), opts: opts_str.clone(), + file_info_bin: file_info_bin.clone(), + opts_bin: opts_bin.clone(), }); let response = client.update_metadata(request).await?.into_inner(); @@ -798,19 +855,22 @@ impl DiskAPI for RemoteDisk { ) -> Result { info!("read_version"); let opts_str = serde_json::to_string(opts)?; + let opts_bin = encode_msgpack(opts)?; self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ReadVersionRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), version_id: version_id.to_string(), opts: opts_str.clone(), + opts_bin: opts_bin.clone(), }); let response = client.read_version(request).await?.into_inner(); @@ -819,7 +879,7 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let file_info = serde_json::from_str::(&response.file_info)?; + let file_info = decode_msgpack_or_json::(&response.file_info_bin, &response.file_info)?; Ok(file_info) }, @@ -834,12 +894,13 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ReadXlRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), read_data, @@ -851,7 +912,7 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let raw_file_info = serde_json::from_str::(&response.raw_file_info)?; + let raw_file_info = decode_msgpack_or_json::(&response.raw_file_info_bin, &response.raw_file_info)?; Ok(raw_file_info) }, @@ -909,13 +970,14 @@ impl DiskAPI for RemoteDisk { if self.health.is_faulty() { return Err(DiskError::FaultyDisk); } + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ListDirRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), dir_path: dir_path.to_string(), count, @@ -937,12 +999,9 @@ impl DiskAPI for RemoteDisk { if self.health.is_faulty() { return Err(DiskError::FaultyDisk); } + let disk = self.disk_ref().await; - let url = format!( - "{}/rustfs/rpc/walk_dir?disk={}", - self.endpoint.grid_host(), - urlencoding::encode(self.endpoint.to_string().as_str()), - ); + let url = format!("{}/rustfs/rpc/walk_dir?disk={}", self.endpoint.grid_host(), urlencoding::encode(&disk),); let opts = serde_json::to_vec(&opts)?; @@ -952,33 +1011,14 @@ impl DiskAPI for RemoteDisk { let mut reader = HttpReader::new(url, Method::GET, headers, Some(opts)).await?; - tokio::io::copy(&mut reader, wr).await?; + copy_stream_with_buffer(&mut reader, wr, DEFAULT_READ_BUFFER_SIZE).await?; Ok(()) } #[tracing::instrument(level = "debug", skip(self))] async fn read_file(&self, volume: &str, path: &str) -> Result { - info!("read_file {}/{}", volume, path); - - if self.health.is_faulty() { - return Err(DiskError::FaultyDisk); - } - - let url = format!( - "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", - self.endpoint.grid_host(), - urlencoding::encode(self.endpoint.to_string().as_str()), - urlencoding::encode(volume), - urlencoding::encode(path), - 0, - 0 - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - build_auth_headers(&url, &Method::GET, &mut headers); - Ok(Box::new(HttpReader::new(url, Method::GET, headers, None).await?)) + self.read_file_stream(volume, path, 0, 0).await } #[tracing::instrument(level = "debug", skip(self))] @@ -995,11 +1035,12 @@ impl DiskAPI for RemoteDisk { if self.health.is_faulty() { return Err(DiskError::FaultyDisk); } + let disk = self.disk_ref().await; let url = format!( "{}/rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}", self.endpoint.grid_host(), - urlencoding::encode(self.endpoint.to_string().as_str()), + urlencoding::encode(&disk), urlencoding::encode(volume), urlencoding::encode(path), offset, @@ -1019,11 +1060,12 @@ impl DiskAPI for RemoteDisk { if self.health.is_faulty() { return Err(DiskError::FaultyDisk); } + let disk = self.disk_ref().await; let url = format!( "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", self.endpoint.grid_host(), - urlencoding::encode(self.endpoint.to_string().as_str()), + urlencoding::encode(&disk), urlencoding::encode(volume), urlencoding::encode(path), true, @@ -1049,11 +1091,12 @@ impl DiskAPI for RemoteDisk { if self.health.is_faulty() { return Err(DiskError::FaultyDisk); } + let disk = self.disk_ref().await; let url = format!( "{}/rustfs/rpc/put_file_stream?disk={}&volume={}&path={}&append={}&size={}", self.endpoint.grid_host(), - urlencoding::encode(self.endpoint.to_string().as_str()), + urlencoding::encode(&disk), urlencoding::encode(volume), urlencoding::encode(path), false, @@ -1261,13 +1304,16 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { let read_multiple_req = serde_json::to_string(&req)?; + let read_multiple_req_bin = encode_msgpack(&req)?; + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ReadMultipleRequest { - disk: self.endpoint.to_string(), + disk, read_multiple_req, + read_multiple_req_bin, }); let response = client.read_multiple(request).await?.into_inner(); @@ -1276,11 +1322,19 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let read_multiple_resps = response - .read_multiple_resps - .into_iter() - .filter_map(|json_str| serde_json::from_str::(&json_str).ok()) - .collect(); + let read_multiple_resps = if !response.read_multiple_resps_bin.is_empty() { + response + .read_multiple_resps_bin + .into_iter() + .filter_map(|buf| decode_msgpack_or_json::(&buf, "").ok()) + .collect() + } else { + response + .read_multiple_resps + .into_iter() + .filter_map(|json_str| serde_json::from_str::(&json_str).ok()) + .collect() + }; Ok(read_multiple_resps) }, @@ -1295,12 +1349,13 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(WriteAllRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), data, @@ -1325,12 +1380,13 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + let disk = self.disk_ref().await; let mut client = self .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ReadAllRequest { - disk: self.endpoint.to_string(), + disk, volume: volume.to_string(), path: path.to_string(), }); @@ -1386,6 +1442,7 @@ impl DiskAPI for RemoteDisk { mod tests { use super::*; use std::sync::Once; + use tokio::io::duplex; use tokio::net::TcpListener; use tracing::Level; use uuid::Uuid; @@ -1543,6 +1600,24 @@ mod tests { assert!(!remote_disk.is_online().await); } + #[tokio::test] + async fn test_copy_stream_with_buffer_copies_full_payload() { + let payload = b"walk-dir-stream".repeat(1024); + let expected = payload.clone(); + let (mut write_half, mut read_half) = duplex(128); + + let copy_task = tokio::spawn(async move { + let mut cursor = Cursor::new(payload); + copy_stream_with_buffer(&mut cursor, &mut write_half, 4 * 1024).await.unwrap(); + }); + + let mut copied = Vec::new(); + read_half.read_to_end(&mut copied).await.unwrap(); + copy_task.await.unwrap(); + + assert_eq!(copied, expected); + } + #[tokio::test] async fn test_remote_disk_disk_id() { let url = url::Url::parse("http://remote-server:9000").unwrap(); @@ -1579,6 +1654,30 @@ mod tests { assert!(cleared_id.is_none()); } + #[tokio::test] + async fn test_remote_disk_ref_prefers_disk_id() { + let url = url::Url::parse("http://remote-server:9000").unwrap(); + let endpoint = Endpoint { + url, + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }; + let disk_option = DiskOption { + cleanup: false, + health_check: false, + }; + + let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap(); + assert_eq!(remote_disk.disk_ref().await, endpoint.to_string()); + + let disk_id = Uuid::new_v4(); + remote_disk.set_disk_id(Some(disk_id)).await.unwrap(); + + assert_eq!(remote_disk.disk_ref().await, disk_id.to_string()); + } + #[tokio::test] async fn test_remote_disk_endpoints_with_different_schemes() { let test_cases = vec![ diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index 64798303d..e3f34ed4d 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -52,6 +52,7 @@ impl RemoteClient { metadata: LockMetadata::default(), priority: LockPriority::Normal, deadlock_detection: false, + suppress_contention_logs: false, } } diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index aafed2a8b..b77c09b4e 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -142,8 +142,8 @@ mod rebalance; use peer::init_local_peer; pub use peer::{ - all_local_disk, all_local_disk_path, find_local_disk, get_disk_infos, get_disk_via_endpoint, has_space_for, init_local_disks, - init_lock_clients, + all_local_disk, all_local_disk_path, find_local_disk, find_local_disk_by_ref, get_disk_infos, get_disk_via_endpoint, + has_space_for, init_local_disks, init_lock_clients, }; #[derive(Debug)] diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 2947cf082..8c508f041 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::global::is_first_cluster_node_local; impl ECStore { #[allow(clippy::new_ret_no_self)] @@ -203,6 +204,7 @@ impl ECStore { let mut meta = PoolMeta::default(); meta.load(self.pools[0].clone(), self.pools.clone()).await?; let update = meta.validate(self.pools.clone())?; + let should_persist_pool_meta = is_first_cluster_node_local().await; if !update { { @@ -211,7 +213,9 @@ impl ECStore { } } else { let new_meta = PoolMeta::new(&self.pools, &meta); - new_meta.save(self.pools.clone()).await?; + if should_persist_pool_meta { + new_meta.save(self.pools.clone()).await?; + } { let mut pool_meta = self.pool_meta.write().await; *pool_meta = new_meta; diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs index 9c33e683b..ecd0f706e 100644 --- a/crates/ecstore/src/store/peer.rs +++ b/crates/ecstore/src/store/peer.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::global::GLOBAL_LOCAL_DISK_ID_MAP; pub async fn find_local_disk(disk_path: &String) -> Option { let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; @@ -24,6 +25,19 @@ pub async fn find_local_disk(disk_path: &String) -> Option { } } +pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option { + if let Some(disk) = find_local_disk(&disk_ref.to_string()).await { + return Some(disk); + } + + let Ok(disk_id) = Uuid::parse_str(disk_ref) else { + return None; + }; + + let disk_path = GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(&disk_id).cloned()?; + find_local_disk(&disk_path).await +} + pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option { let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; if global_set_drives.is_empty() { diff --git a/crates/ecstore/src/tier/tier.rs b/crates/ecstore/src/tier/tier.rs index 93012baaf..ace4f9e4e 100644 --- a/crates/ecstore/src/tier/tier.rs +++ b/crates/ecstore/src/tier/tier.rs @@ -49,6 +49,7 @@ use crate::{ StorageAPI, config::com::{CONFIG_PREFIX, read_config}, disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}, + global::{get_global_endpoints, is_first_cluster_node_local}, store::ECStore, store_api::{ObjectIO as _, ObjectOptions, PutObjReader}, }; @@ -1118,7 +1119,15 @@ async fn load_tier_config(api: Arc) -> std::result::Result { warn!("config not found, start to init"); - new_and_save_tiering_config(api).await.map_err(io::Error::other) + if is_first_cluster_node_local().await { + new_and_save_tiering_config(api).await.map_err(io::Error::other) + } else { + Ok(TierConfigMgr { + driver_cache: HashMap::new(), + tiers: HashMap::new(), + last_refreshed_at: OffsetDateTime::now_utc(), + }) + } } Err(legacy_err) => Err(io::Error::other(legacy_err)), } @@ -1166,7 +1175,14 @@ async fn write_tier_config_to_rustfs(api: Arc, path: &str, dat pub async fn try_migrate_tiering_config(api: Arc) { let target_path = tier_config_path(TIER_CONFIG_FILE); if api - .get_object_info(RUSTFS_META_BUCKET, &target_path, &ObjectOptions::default()) + .get_object_info( + RUSTFS_META_BUCKET, + &target_path, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .is_ok() { diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 0e95639f2..16dcb53de 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -25,6 +25,7 @@ use crate::{ }; use futures::future::join_all; use rustfs_credentials::{Credentials, EMBEDDED_POLICY_TYPE, INHERITED_POLICY_TYPE, get_global_action_cred}; +use rustfs_ecstore::global::is_first_cluster_node_local; use rustfs_madmin::{AccountStatus, AddOrUpdateUserReq, GroupDesc}; use rustfs_policy::{ arn::ARN, @@ -303,6 +304,10 @@ where return Ok(()); } + if !is_first_cluster_node_local().await { + return Ok(()); + } + self.api.save_iam_config(IAMFormat::new_version_1(), path).await } pub async fn get_user(&self, access_key: &str) -> Option { diff --git a/crates/iam/src/store/object.rs b/crates/iam/src/store/object.rs index 4f8327ee1..c76c6bf69 100644 --- a/crates/iam/src/store/object.rs +++ b/crates/iam/src/store/object.rs @@ -25,7 +25,7 @@ use rustfs_ecstore::store_api::{ListOperations as _, ObjectInfoOrErr, WalkOption use rustfs_ecstore::{ config::{ RUSTFS_CONFIG_PREFIX, - com::{delete_config, read_config, read_config_with_metadata, save_config}, + com::{delete_config, read_config, read_config_no_lock, read_config_with_metadata, save_config}, }, store::ECStore, store_api::{ObjectInfo, ObjectOptions}, @@ -391,7 +391,7 @@ impl ObjectStore { // If it doesn't exist, the system bucket or metadata is not ready. let probe_path = format!("{}/format.json", *IAM_CONFIG_PREFIX); - match read_config(self.object_api.clone(), &probe_path).await { + match read_config_no_lock(self.object_api.clone(), &probe_path).await { Ok(_) => Ok(()), Err(rustfs_ecstore::error::StorageError::ConfigNotFound) => Err(Error::other(format!( "Storage metadata not ready: probe object '{}' not found (expected IAM config to be initialized)", diff --git a/crates/lock/src/distributed_lock.rs b/crates/lock/src/distributed_lock.rs index bd897b5bd..3f32218f6 100644 --- a/crates/lock/src/distributed_lock.rs +++ b/crates/lock/src/distributed_lock.rs @@ -230,7 +230,21 @@ impl DistributedLock { } else { // Check if it's a timeout or quorum failure if let Some(error_msg) = &resp.error { - warn!("acquire_lock_quorum error: {}", error_msg); + if request.suppress_contention_logs { + tracing::debug!( + resource = %request.resource, + owner = %request.owner, + "acquire_lock_quorum contention: {}", + error_msg + ); + } else { + warn!( + resource = %request.resource, + owner = %request.owner, + "acquire_lock_quorum error: {}", + error_msg + ); + } if error_msg.contains("quorum") { // This is a quorum failure - return appropriate error // Extract achieved count from error message or use individual_locks.len() @@ -266,6 +280,21 @@ impl DistributedLock { self.acquire_guard(&req).await } + /// Convenience: acquire exclusive lock with expected contention logs suppressed + pub async fn lock_guard_quiet( + &self, + resource: ObjectKey, + owner: &str, + timeout: Duration, + ttl: Duration, + ) -> Result> { + let req = LockRequest::new(resource, LockType::Exclusive, owner) + .with_acquire_timeout(timeout) + .with_ttl(ttl) + .with_suppress_contention_logs(true); + self.acquire_guard(&req).await + } + /// Convenience: acquire shared lock as a guard pub async fn rlock_guard( &self, @@ -307,11 +336,24 @@ impl DistributedLock { individual_locks.push((lock_info.id.clone(), self.clients[idx].clone())); } } else { - tracing::warn!( - "Failed to acquire lock on client from response: {}, error: {}", - idx, - resp.error.unwrap_or_else(|| "unknown error".to_string()) - ); + let error = resp.error.unwrap_or_else(|| "unknown error".to_string()); + if request.suppress_contention_logs { + tracing::debug!( + resource = %request.resource, + owner = %request.owner, + "Failed to acquire lock on client from response: {}, error: {}", + idx, + error + ); + } else { + tracing::warn!( + resource = %request.resource, + owner = %request.owner, + "Failed to acquire lock on client from response: {}, error: {}", + idx, + error + ); + } } } Err(e) => { diff --git a/crates/lock/src/namespace/mod.rs b/crates/lock/src/namespace/mod.rs index 9b364beff..bc1dfc3b1 100644 --- a/crates/lock/src/namespace/mod.rs +++ b/crates/lock/src/namespace/mod.rs @@ -58,6 +58,16 @@ impl NamespaceLockWrapper { self.lock.get_write_lock(self.resource.clone(), &self.owner, timeout).await } + /// Acquire write lock with expected contention logs suppressed + pub async fn get_write_lock_quiet( + &self, + timeout: Duration, + ) -> std::result::Result { + self.lock + .get_write_lock_quiet(self.resource.clone(), &self.owner, timeout) + .await + } + /// Acquire read lock (shared lock) with timeout /// Returns the guard if acquisition succeeds, or an error if it fails pub async fn get_read_lock(&self, timeout: Duration) -> std::result::Result { @@ -237,6 +247,29 @@ impl NamespaceLock { } } + /// Acquire write lock while suppressing expected contention warnings + pub async fn get_write_lock_quiet( + &self, + resource: ObjectKey, + owner: &str, + timeout: Duration, + ) -> std::result::Result { + let ttl = crate::fast_lock::DEFAULT_LOCK_TIMEOUT; + let resource_str = format!("{}", resource); + match self { + Self::Distributed(lock) => match lock.lock_guard_quiet(resource, owner, timeout, ttl).await { + Ok(Some(guard)) => Ok(NamespaceLockGuard::Standard(guard)), + Ok(None) => Err(crate::error::LockError::timeout(resource_str, timeout)), + Err(e) => Err(e), + }, + Self::Local(lock) => match lock.lock_guard(resource, owner, timeout, ttl).await { + Ok(Some(guard)) => Ok(NamespaceLockGuard::Fast(guard)), + Ok(None) => Err(crate::error::LockError::timeout(resource_str, timeout)), + Err(e) => Err(e), + }, + } + } + /// Acquire read lock (shared lock) with timeout /// Returns the guard if acquisition succeeds, or an error if it fails pub async fn get_read_lock( diff --git a/crates/lock/src/types.rs b/crates/lock/src/types.rs index ec5309541..b531082e5 100644 --- a/crates/lock/src/types.rs +++ b/crates/lock/src/types.rs @@ -225,6 +225,8 @@ pub struct LockRequest { pub priority: LockPriority, /// Deadlock detection pub deadlock_detection: bool, + /// Suppress warning logs for expected contention failures + pub suppress_contention_logs: bool, } impl LockRequest { @@ -240,6 +242,7 @@ impl LockRequest { metadata: LockMetadata::default(), priority: LockPriority::default(), deadlock_detection: false, + suppress_contention_logs: false, } } @@ -272,6 +275,12 @@ impl LockRequest { self.deadlock_detection = enabled; self } + + /// Suppress warning logs for expected contention failures + pub fn with_suppress_contention_logs(mut self, enabled: bool) -> Self { + self.suppress_contention_logs = enabled; + self + } } /// Lock response structure diff --git a/crates/notify/src/registry.rs b/crates/notify/src/registry.rs index 68462f52a..74ccbdfb9 100644 --- a/crates/notify/src/registry.rs +++ b/crates/notify/src/registry.rs @@ -293,6 +293,12 @@ impl TargetRegistry { } } + if &new_config == config { + info!("Notification target configuration unchanged, skip persisting server config"); + info!(count = successful_targets.len(), "All target processing completed"); + return Ok(successful_targets); + } + let store = match rustfs_ecstore::global::new_object_layer_fn() { Some(s) => s, None => { diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 37badbbf8..51cc40308 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -467,6 +467,10 @@ pub struct UpdateMetadataRequest { pub file_info: ::prost::alloc::string::String, #[prost(string, tag = "5")] pub opts: ::prost::alloc::string::String, + #[prost(bytes = "vec", tag = "6")] + pub file_info_bin: ::prost::alloc::vec::Vec, + #[prost(bytes = "vec", tag = "7")] + pub opts_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct UpdateMetadataResponse { @@ -486,6 +490,8 @@ pub struct WriteMetadataRequest { pub path: ::prost::alloc::string::String, #[prost(string, tag = "4")] pub file_info: ::prost::alloc::string::String, + #[prost(bytes = "vec", tag = "5")] + pub file_info_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct WriteMetadataResponse { @@ -506,6 +512,8 @@ pub struct ReadVersionRequest { pub version_id: ::prost::alloc::string::String, #[prost(string, tag = "5")] pub opts: ::prost::alloc::string::String, + #[prost(bytes = "vec", tag = "6")] + pub opts_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ReadVersionResponse { @@ -515,6 +523,8 @@ pub struct ReadVersionResponse { pub file_info: ::prost::alloc::string::String, #[prost(message, optional, tag = "3")] pub error: ::core::option::Option, + #[prost(bytes = "vec", tag = "4")] + pub file_info_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ReadXlRequest { @@ -535,6 +545,8 @@ pub struct ReadXlResponse { pub raw_file_info: ::prost::alloc::string::String, #[prost(message, optional, tag = "3")] pub error: ::core::option::Option, + #[prost(bytes = "vec", tag = "4")] + pub raw_file_info_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteVersionRequest { @@ -586,6 +598,8 @@ pub struct ReadMultipleRequest { pub disk: ::prost::alloc::string::String, #[prost(string, tag = "2")] pub read_multiple_req: ::prost::alloc::string::String, + #[prost(bytes = "vec", tag = "3")] + pub read_multiple_req_bin: ::prost::alloc::vec::Vec, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct ReadMultipleResponse { @@ -595,6 +609,8 @@ pub struct ReadMultipleResponse { pub read_multiple_resps: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, #[prost(message, optional, tag = "3")] pub error: ::core::option::Option, + #[prost(bytes = "vec", repeated, tag = "4")] + pub read_multiple_resps_bin: ::prost::alloc::vec::Vec<::prost::alloc::vec::Vec>, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteVolumeRequest { diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index 45ddb5bb5..60073bb18 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -16,8 +16,13 @@ mod generated; use proto_gen::node_service::node_service_client::NodeServiceClient; -use rustfs_common::{GLOBAL_CONN_MAP, GLOBAL_MTLS_IDENTITY, GLOBAL_ROOT_CERT, evict_connection}; -use std::{error::Error, time::Duration}; +use rustfs_common::{ + GLOBAL_CONN_MAP, GLOBAL_MTLS_IDENTITY, GLOBAL_ROOT_CERT, evict_connection, internode_metrics::global_internode_metrics, +}; +use std::{ + error::Error, + time::{Duration, Instant}, +}; use tonic::{ Request, Status, service::interceptor::InterceptedService, @@ -65,6 +70,7 @@ const RUSTFS_HTTPS_PREFIX: &str = "https://"; /// - Overall RPC timeout of 30s (reduced from 60s) pub async fn create_new_channel(addr: &str) -> Result> { debug!("Creating new gRPC channel to: {}", addr); + let dial_started_at = Instant::now(); let mut connector = Endpoint::from_shared(addr.to_string())? // Fast connection timeout for dead peer detection @@ -119,7 +125,16 @@ pub async fn create_new_channel(addr: &str) -> Result> { } } - let channel = connector.connect().await?; + let channel = match connector.connect().await { + Ok(channel) => { + global_internode_metrics().record_dial_result(dial_started_at.elapsed(), true); + channel + } + Err(err) => { + global_internode_metrics().record_dial_result(dial_started_at.elapsed(), false); + return Err(err.into()); + } + }; // Cache the new connection { diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index 13e061065..23024f09e 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -331,6 +331,8 @@ message UpdateMetadataRequest { string path = 3; string file_info = 4; string opts = 5; + bytes file_info_bin = 6; + bytes opts_bin = 7; } message UpdateMetadataResponse { @@ -343,6 +345,7 @@ message WriteMetadataRequest { string volume = 2; string path = 3; string file_info = 4; + bytes file_info_bin = 5; } message WriteMetadataResponse { @@ -356,12 +359,14 @@ message ReadVersionRequest { string path = 3; string version_id = 4; string opts = 5; + bytes opts_bin = 6; } message ReadVersionResponse { bool success = 1; string file_info = 2; optional Error error = 3; + bytes file_info_bin = 4; } message ReadXLRequest { @@ -375,6 +380,7 @@ message ReadXLResponse { bool success = 1; string raw_file_info = 2; optional Error error = 3; + bytes raw_file_info_bin = 4; } message DeleteVersionRequest { @@ -408,12 +414,14 @@ message DeleteVersionsResponse { message ReadMultipleRequest { string disk = 1; string read_multiple_req = 2; + bytes read_multiple_req_bin = 3; } message ReadMultipleResponse { bool success = 1; repeated string read_multiple_resps = 2; optional Error error = 3; + repeated bytes read_multiple_resps_bin = 4; } message DeleteVolumeRequest { diff --git a/crates/rio/Cargo.toml b/crates/rio/Cargo.toml index 3b50e1998..42f30d419 100644 --- a/crates/rio/Cargo.toml +++ b/crates/rio/Cargo.toml @@ -42,6 +42,7 @@ tokio-util.workspace = true faster-hex.workspace = true futures.workspace = true rustfs-config = { workspace = true, features = ["constants"] } +rustfs-common.workspace = true rustfs-utils = { workspace = true, features = ["io", "hash", "compress", "tls"] } serde_json.workspace = true md-5 = { workspace = true } @@ -55,3 +56,5 @@ hex-simd.workspace = true [dev-dependencies] tokio-test = { workspace = true } +axum = { workspace = true } +http-body-util = { workspace = true } diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 5377cba7f..86186db2d 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -13,13 +13,14 @@ // limitations under the License. use crate::{EtagResolvable, HashReaderDetector, HashReaderMut}; -use bytes::Bytes; +use bytes::{Bytes, BytesMut}; use futures::{Stream, TryStreamExt as _}; use http::HeaderMap; use pin_project_lite::pin_project; use reqwest::{Certificate, Client, Identity, Method, RequestBuilder}; +use rustfs_common::internode_metrics::global_internode_metrics; use rustfs_utils::get_env_opt_str; -use std::error::Error as _; +use std::io::IoSlice; use std::io::{self, Error}; use std::ops::Not as _; use std::pin::Pin; @@ -28,6 +29,7 @@ use std::task::{Context, Poll}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; use tokio::sync::mpsc; use tokio_util::io::StreamReader; +use tokio_util::sync::PollSender; use tracing::error; /// Get the TLS path from the RUSTFS_TLS_PATH environment variable. @@ -111,24 +113,12 @@ fn get_http_client() -> Client { CLIENT.clone() } -static HTTP_DEBUG_LOG: bool = false; -#[inline(always)] -fn http_debug_log(args: std::fmt::Arguments) { - if HTTP_DEBUG_LOG { - println!("{args}"); - } -} -macro_rules! http_log { - ($($arg:tt)*) => { - http_debug_log(format_args!($($arg)*)); - }; -} - pin_project! { pub struct HttpReader { url:String, method: Method, headers: HeaderMap, + track_internode_metrics: bool, #[pin] inner: StreamReader>+Send+Sync>>, Bytes>, } @@ -147,53 +137,47 @@ impl HttpReader { body: Option>, _read_buf_size: usize, ) -> io::Result { - // http_log!( - // "[HttpReader::with_capacity] url: {url}, method: {method:?}, headers: {headers:?}, buf_size: {}", - // _read_buf_size - // ); - // First, check if the connection is available (HEAD) - let client = get_http_client(); - let head_resp = client.head(&url).headers(headers.clone()).send().await; - match head_resp { - Ok(resp) => { - http_log!("[HttpReader::new] HEAD status: {}", resp.status()); - if !resp.status().is_success() { - return Err(Error::other(format!("HEAD failed: url: {}, status {}", url, resp.status()))); - } - } - Err(e) => { - http_log!("[HttpReader::new] HEAD error: {e}"); - return Err(Error::other(e.source().map(|s| s.to_string()).unwrap_or_else(|| e.to_string()))); - } - } - + let track_internode_metrics = is_internode_rpc_url(&url); let client = get_http_client(); let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone()); if let Some(body) = body { request = request.body(body); } - let resp = request - .send() - .await - .map_err(|e| Error::other(format!("HttpReader HTTP request error: {e}")))?; + let resp = request.send().await.map_err(|e| { + if track_internode_metrics { + global_internode_metrics().record_error(); + } + Error::other(format!("HttpReader HTTP request error: {e}")) + })?; if resp.status().is_success().not() { + if track_internode_metrics { + global_internode_metrics().record_error(); + } return Err(Error::other(format!( "HttpReader HTTP request failed with non-200 status {}", resp.status() ))); } - let stream = resp - .bytes_stream() - .map_err(|e| Error::other(format!("HttpReader stream error: {e}"))); + if track_internode_metrics { + global_internode_metrics().record_outgoing_request(); + } + + let stream = resp.bytes_stream().map_err(move |e| { + if track_internode_metrics { + global_internode_metrics().record_error(); + } + Error::other(format!("HttpReader stream error: {e}")) + }); Ok(Self { inner: StreamReader::new(Box::pin(stream)), url, method, headers, + track_internode_metrics, }) } pub fn url(&self) -> &str { @@ -209,14 +193,17 @@ impl HttpReader { impl AsyncRead for HttpReader { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { - // http_log!( - // "[HttpReader::poll_read] url: {}, method: {:?}, buf.remaining: {}", - // self.url, - // self.method, - // buf.remaining() - // ); - // Read from the inner stream - Pin::new(&mut self.inner).poll_read(cx, buf) + let filled_before = buf.filled().len(); + match Pin::new(&mut self.inner).poll_read(cx, buf) { + Poll::Ready(Ok(())) => { + let bytes_read = buf.filled().len().saturating_sub(filled_before); + if self.track_internode_metrics && bytes_read > 0 { + global_internode_metrics().record_recv_bytes(bytes_read); + } + Poll::Ready(Ok(())) + } + other => other, + } } } @@ -241,6 +228,7 @@ impl HashReaderDetector for HttpReader { struct ReceiverStream { receiver: mpsc::Receiver>, + track_internode_metrics: bool, } impl Stream for ReceiverStream { @@ -262,7 +250,12 @@ impl Stream for ReceiverStream { // } // } match poll { - Poll::Ready(Some(Some(bytes))) => Poll::Ready(Some(Ok(bytes))), + Poll::Ready(Some(Some(bytes))) => { + if self.track_internode_metrics { + global_internode_metrics().record_sent_bytes(bytes.len()); + } + Poll::Ready(Some(Ok(bytes))) + } Poll::Ready(Some(None)) => Poll::Ready(None), // Sender shutdown Poll::Ready(None) => Poll::Ready(None), Poll::Pending => Poll::Pending, @@ -276,13 +269,17 @@ pin_project! { method: Method, headers: HeaderMap, err_rx: tokio::sync::oneshot::Receiver, - sender: tokio::sync::mpsc::Sender>, + sender: PollSender>, handle: tokio::task::JoinHandle>, + pending_chunk: BytesMut, finish:bool, } } +const HTTP_WRITER_CHANNEL_CAPACITY: usize = 8; +const HTTP_WRITER_BUFFER_SIZE: usize = 1024 * 1024; + impl HttpWriter { /// Create a new HttpWriter for the given URL. The HTTP request is performed in the background. pub async fn new(url: String, method: Method, headers: HeaderMap) -> io::Result { @@ -290,28 +287,16 @@ impl HttpWriter { let url_clone = url.clone(); let method_clone = method.clone(); let headers_clone = headers.clone(); + let track_internode_metrics = is_internode_rpc_url(&url); - // First, try to write empty data to check if writable - let client = get_http_client(); - let resp = client.put(&url).headers(headers.clone()).body(Vec::new()).send().await; - match resp { - Ok(resp) => { - // http_log!("[HttpWriter::new] empty PUT status: {}", resp.status()); - if !resp.status().is_success() { - return Err(Error::other(format!("Empty PUT failed: status {}", resp.status()))); - } - } - Err(e) => { - // http_log!("[HttpWriter::new] empty PUT error: {e}"); - return Err(Error::other(format!("Empty PUT failed: {e}"))); - } - } - - let (sender, receiver) = tokio::sync::mpsc::channel::>(8); + let (sender, receiver) = tokio::sync::mpsc::channel::>(HTTP_WRITER_CHANNEL_CAPACITY); let (err_tx, err_rx) = tokio::sync::oneshot::channel::(); let handle = tokio::spawn(async move { - let stream = ReceiverStream { receiver }; + let stream = ReceiverStream { + receiver, + track_internode_metrics, + }; let body = reqwest::Body::wrap_stream(stream); // http_log!( // "[HttpWriter::spawn] sending HTTP request: url={url_clone}, method={method_clone:?}, headers={headers_clone:?}" @@ -330,6 +315,9 @@ impl HttpWriter { Ok(resp) => { // http_log!("[HttpWriter::spawn] got response: status={}", resp.status()); if !resp.status().is_success() { + if track_internode_metrics { + global_internode_metrics().record_error(); + } let _ = err_tx.send(Error::other(format!( "HttpWriter HTTP request failed with non-200 status {}", resp.status() @@ -338,6 +326,9 @@ impl HttpWriter { } } Err(e) => { + if track_internode_metrics { + global_internode_metrics().record_error(); + } // http_log!("[HttpWriter::spawn] HTTP request error: {e}"); let _ = err_tx.send(Error::other(format!("HTTP request failed: {e}"))); return Err(Error::other(format!("HTTP request failed: {e}"))); @@ -349,13 +340,17 @@ impl HttpWriter { }); // http_log!("[HttpWriter::new] connection established successfully"); + if track_internode_metrics { + global_internode_metrics().record_outgoing_request(); + } Ok(Self { url, method, headers, err_rx, - sender, + sender: PollSender::new(sender), handle, + pending_chunk: BytesMut::with_capacity(HTTP_WRITER_BUFFER_SIZE), finish: false, }) } @@ -373,8 +368,40 @@ impl HttpWriter { } } +fn is_internode_rpc_url(url: &str) -> bool { + url.contains("/rustfs/rpc/") +} + +fn poll_send_error_to_io(err: tokio_util::sync::PollSendError, context: &str) -> io::Error { + Error::other(format!("{context}: {err}")) +} + +fn send_error_to_io(err: tokio_util::sync::PollSendError, context: &str) -> io::Error { + Error::other(format!("{context}: {err}")) +} + +impl HttpWriter { + fn poll_send_pending_chunk(&mut self, cx: &mut Context<'_>) -> Poll> { + if self.pending_chunk.is_empty() { + return Poll::Ready(Ok(())); + } + + match self.sender.poll_reserve(cx) { + Poll::Ready(Ok(())) => { + let chunk = self.pending_chunk.split().freeze(); + self.sender + .send_item(Some(chunk)) + .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?; + Poll::Ready(Ok(())) + } + Poll::Ready(Err(e)) => Poll::Ready(Err(poll_send_error_to_io(e, "HttpWriter send error"))), + Poll::Pending => Poll::Pending, + } + } +} + impl AsyncWrite for HttpWriter { - fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { // http_log!( // "[HttpWriter::poll_write] url: {}, method: {:?}, buf.len: {}", // self.url, @@ -385,26 +412,104 @@ impl AsyncWrite for HttpWriter { return Poll::Ready(Err(e)); } - self.sender - .try_send(Some(Bytes::copy_from_slice(buf))) - .map_err(|e| Error::other(format!("HttpWriter send error: {e}")))?; + let this = self.as_mut().get_mut(); + + if this.pending_chunk.len() >= HTTP_WRITER_BUFFER_SIZE { + match this.poll_send_pending_chunk(cx) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => return Poll::Pending, + } + } + + if buf.len() >= HTTP_WRITER_BUFFER_SIZE && this.pending_chunk.is_empty() { + match this.sender.poll_reserve(cx) { + Poll::Ready(Ok(())) => { + this.sender + .send_item(Some(Bytes::copy_from_slice(buf))) + .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?; + return Poll::Ready(Ok(buf.len())); + } + Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))), + Poll::Pending => return Poll::Pending, + } + } + + this.pending_chunk.extend_from_slice(buf); Poll::Ready(Ok(buf.len())) } - fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { - Poll::Ready(Ok(())) + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.as_mut().get_mut().poll_send_pending_chunk(cx) } - fn poll_shutdown(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + fn poll_write_vectored(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>]) -> Poll> { + if let Ok(e) = Pin::new(&mut self.err_rx).try_recv() { + return Poll::Ready(Err(e)); + } + + let this = self.as_mut().get_mut(); + + if this.pending_chunk.len() >= HTTP_WRITER_BUFFER_SIZE { + match this.poll_send_pending_chunk(cx) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => return Poll::Pending, + } + } + + let total_len = bufs.iter().map(|buf| buf.len()).sum::(); + if total_len == 0 { + return Poll::Ready(Ok(0)); + } + + if bufs.len() == 1 && this.pending_chunk.is_empty() && total_len >= HTTP_WRITER_BUFFER_SIZE { + match this.sender.poll_reserve(cx) { + Poll::Ready(Ok(())) => { + this.sender + .send_item(Some(Bytes::copy_from_slice(bufs[0].as_ref()))) + .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?; + return Poll::Ready(Ok(total_len)); + } + Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))), + Poll::Pending => return Poll::Pending, + } + } + + for buf in bufs { + this.pending_chunk.extend_from_slice(buf); + } + + Poll::Ready(Ok(total_len)) + } + + fn is_write_vectored(&self) -> bool { + true + } + + fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // let url = self.url.clone(); // let method = self.method.clone(); + match self.as_mut().get_mut().poll_send_pending_chunk(cx) { + Poll::Ready(Ok(())) => {} + Poll::Ready(Err(err)) => return Poll::Ready(Err(err)), + Poll::Pending => return Poll::Pending, + } + if !self.finish { // http_log!("[HttpWriter::poll_shutdown] url: {}, method: {:?}", url, method); - self.sender - .try_send(None) - .map_err(|e| Error::other(format!("HttpWriter shutdown error: {e}")))?; + let this = self.as_mut().get_mut(); + match this.sender.poll_reserve(cx) { + Poll::Ready(Ok(())) => { + this.sender + .send_item(None) + .map_err(|e| send_error_to_io(e, "HttpWriter shutdown error"))?; + } + Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter shutdown error"))), + Poll::Pending => return Poll::Pending, + } // http_log!( // "[HttpWriter::poll_shutdown] sent shutdown signal to HTTP request, url: {}, method: {:?}", // url, @@ -415,7 +520,7 @@ impl AsyncWrite for HttpWriter { } // Wait for the HTTP request to complete use futures::FutureExt; - match Pin::new(&mut self.get_mut().handle).poll_unpin(_cx) { + match Pin::new(&mut self.get_mut().handle).poll_unpin(cx) { Poll::Ready(Ok(_)) => { // http_log!( // "[HttpWriter::poll_shutdown] HTTP request finished successfully, url: {}, method: {:?}", @@ -437,77 +542,126 @@ impl AsyncWrite for HttpWriter { } } -// #[cfg(test)] -// mod tests { -// use super::*; -// use reqwest::Method; -// use std::vec; -// use tokio::io::{AsyncReadExt, AsyncWriteExt}; +#[cfg(test)] +mod tests { + use super::*; + use axum::{Router, body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get}; + use http_body_util::BodyExt as _; + use std::io::IoSlice; + use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }; + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + sync::Mutex, + }; -// #[tokio::test] -// async fn test_http_writer_err() { -// // Use a real local server for integration, or mockito for unit test -// // Here, we use the Go test server at 127.0.0.1:8081 (scripts/testfile.go) -// let url = "http://127.0.0.1:8081/testfile".to_string(); -// let data = vec![42u8; 8]; + #[derive(Clone, Default)] + struct TestState { + head_count: Arc, + get_count: Arc, + put_count: Arc, + put_bodies: Arc>>>, + } -// // Write -// // Add header X-Deny-Write = 1 to simulate non-writable situation -// let mut headers = HeaderMap::new(); -// headers.insert("X-Deny-Write", "1".parse().unwrap()); -// // Here we use PUT method -// let writer_result = HttpWriter::new(url.clone(), Method::PUT, headers).await; -// match writer_result { -// Ok(mut writer) => { -// // If creation succeeds, write should fail -// let write_result = writer.write_all(&data).await; -// assert!(write_result.is_err(), "write_all should fail when server denies write"); -// if let Err(e) = write_result { -// println!("write_all error: {e}"); -// } -// let shutdown_result = writer.shutdown().await; -// if let Err(e) = shutdown_result { -// println!("shutdown error: {e}"); -// } -// } -// Err(e) => { -// // Direct construction failure is also acceptable -// println!("HttpWriter::new error: {e}"); -// assert!( -// e.to_string().contains("Empty PUT failed") || e.to_string().contains("Forbidden"), -// "unexpected error: {e}" -// ); -// return; -// } -// } -// // Should not reach here -// panic!("HttpWriter should not allow writing when server denies write"); -// } + async fn get_stream(State(state): State) -> impl IntoResponse { + state.get_count.fetch_add(1, Ordering::SeqCst); + (StatusCode::OK, Body::from("hello")) + } -// #[tokio::test] -// async fn test_http_writer_and_reader_ok() { -// // Use local Go test server -// let url = "http://127.0.0.1:8081/testfile".to_string(); -// let data = vec![99u8; 512 * 1024]; // 512KB of data + async fn reject_head(State(state): State) -> impl IntoResponse { + state.head_count.fetch_add(1, Ordering::SeqCst); + StatusCode::METHOD_NOT_ALLOWED + } -// // Write (without X-Deny-Write) -// let headers = HeaderMap::new(); -// let mut writer = HttpWriter::new(url.clone(), Method::PUT, headers).await.unwrap(); -// writer.write_all(&data).await.unwrap(); -// writer.shutdown().await.unwrap(); + async fn accept_put(State(state): State, body: Body) -> impl IntoResponse { + state.put_count.fetch_add(1, Ordering::SeqCst); + let bytes = body.collect().await.unwrap().to_bytes(); + state.put_bodies.lock().await.push(bytes.to_vec()); + StatusCode::OK + } -// http_log!("Wrote {} bytes to {} (ok case)", data.len(), url); + async fn start_test_server(state: TestState) -> (String, tokio::task::JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .route("/stream", get(get_stream).head(reject_head).put(accept_put)) + .with_state(state); -// // Read back -// let mut reader = HttpReader::with_capacity(url.clone(), Method::GET, HeaderMap::new(), 8192) -// .await -// .unwrap(); -// let mut buf = Vec::new(); -// reader.read_to_end(&mut buf).await.unwrap(); -// assert_eq!(buf, data); + let handle = tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); -// // println!("Read {} bytes from {} (ok case)", buf.len(), url); -// // tokio::time::sleep(std::time::Duration::from_secs(2)).await; // Wait for server to process -// // println!("[test_http_writer_and_reader_ok] completed successfully"); -// } -// } + (format!("http://{addr}/stream"), handle) + } + + #[tokio::test] + async fn http_reader_does_not_send_preflight_head() { + let state = TestState::default(); + let (url, handle) = start_test_server(state.clone()).await; + + let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None).await.unwrap(); + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await.unwrap(); + + assert_eq!(buf, b"hello"); + assert_eq!(state.head_count.load(Ordering::SeqCst), 0); + assert_eq!(state.get_count.load(Ordering::SeqCst), 1); + + handle.abort(); + } + + #[tokio::test] + async fn http_writer_does_not_send_empty_preflight_put() { + let state = TestState::default(); + let (url, handle) = start_test_server(state.clone()).await; + + let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap(); + writer.write_all(b"payload").await.unwrap(); + writer.shutdown().await.unwrap(); + + assert_eq!(state.put_count.load(Ordering::SeqCst), 1); + assert_eq!(state.put_bodies.lock().await.as_slice(), &[b"payload".to_vec()]); + + handle.abort(); + } + + #[tokio::test] + async fn http_writer_handles_many_small_writes() { + let state = TestState::default(); + let (url, handle) = start_test_server(state.clone()).await; + + let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap(); + let chunk = b"0123456789abcdef"; + let mut expected = Vec::new(); + for _ in 0..256 { + writer.write_all(chunk).await.unwrap(); + expected.extend_from_slice(chunk); + } + writer.shutdown().await.unwrap(); + + assert_eq!(state.put_count.load(Ordering::SeqCst), 1); + assert_eq!(state.put_bodies.lock().await.as_slice(), &[expected]); + + handle.abort(); + } + + #[tokio::test] + async fn http_writer_supports_vectored_writes() { + let state = TestState::default(); + let (url, handle) = start_test_server(state.clone()).await; + + let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new()).await.unwrap(); + let bufs = [IoSlice::new(b"hello "), IoSlice::new(b"world")]; + let written = writer.write_vectored(&bufs).await.unwrap(); + assert_eq!(written, 11); + writer.shutdown().await.unwrap(); + + assert_eq!(state.put_count.load(Ordering::SeqCst), 1); + assert_eq!(state.put_bodies.lock().await.as_slice(), &[b"hello world".to_vec()]); + + handle.abort(); + } +} diff --git a/crates/rio/src/writer.rs b/crates/rio/src/writer.rs index 570183dcc..98d19f2d8 100644 --- a/crates/rio/src/writer.rs +++ b/crates/rio/src/writer.rs @@ -20,7 +20,7 @@ use crate::HttpWriter; pub enum Writer { Cursor(Cursor>), - Http(HttpWriter), + Http(Box), Other(Box), } @@ -38,7 +38,7 @@ impl Writer { } pub fn from_http(w: HttpWriter) -> Self { - Writer::Http(w) + Writer::Http(Box::new(w)) } pub fn into_cursor_inner(self) -> Option> { @@ -56,14 +56,14 @@ impl Writer { } pub fn as_http(&mut self) -> Option<&mut HttpWriter> { match self { - Writer::Http(w) => Some(w), + Writer::Http(w) => Some(w.as_mut()), _ => None, } } pub fn into_http(self) -> Option { match self { - Writer::Http(w) => Some(w), + Writer::Http(w) => Some(*w), _ => None, } } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 0638812aa..d70d76509 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -71,6 +71,16 @@ fn randomized_cycle_delay_for(interval: Duration) -> Duration { delay.max(Duration::from_secs(1)) } +fn initial_scanner_delay() -> Duration { + initial_scanner_delay_for(scanner_start_delay_secs()) +} + +fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { + start_delay_secs + .map(|secs| randomized_cycle_delay_for(Duration::from_secs(secs))) + .unwrap_or_else(|| Duration::from_secs(rand::random::() % 5)) +} + pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { // Force init global sleeper so config is read once at startup. let _ = &*SCANNER_SLEEPER; @@ -78,7 +88,7 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { let ctx_clone = ctx.clone(); let storeapi_clone = storeapi.clone(); tokio::spawn(async move { - let sleep_time = Duration::from_secs(rand::random::() % 5); + let sleep_time = initial_scanner_delay(); tokio::time::sleep(sleep_time).await; loop { @@ -246,7 +256,7 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc) -> Result<(), ScannerError> { // Acquire leader lock (write lock) to ensure only one scanner runs let _guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await { - Ok(ns_lock) => match ns_lock.get_write_lock(get_lock_acquire_timeout()).await { + Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await { Ok(guard) => { debug!("run_data_scanner: acquired leader write lock"); guard @@ -358,6 +368,14 @@ mod tests { assert!(delay <= Duration::from_secs(132)); } + #[test] + #[serial] + fn test_initial_scanner_delay_uses_configured_start_delay() { + let delay = initial_scanner_delay_for(Some(120)); + assert!(delay >= Duration::from_secs(108)); + assert!(delay <= Duration::from_secs(132)); + } + #[test] #[serial] fn test_randomized_cycle_delay_handles_small_start_delay() { diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 1334312cc..d2bbaa5a5 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -16,7 +16,6 @@ mod auth; pub mod console; pub mod handlers; pub mod router; -mod rpc; pub mod utils; #[cfg(test)] @@ -28,7 +27,6 @@ use handlers::{ bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, sts, system, tier, user, }; use router::{AdminOperation, S3Router}; -use rpc::register_rpc_route; use s3s::route::S3Route; /// Create admin router @@ -44,7 +42,6 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result health::register_health_route(&mut r)?; sts::register_admin_auth_route(&mut r)?; - register_rpc_route(&mut r)?; user::register_user_route(&mut r)?; system::register_system_route(&mut r)?; pools::register_pool_route(&mut r)?; diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 07bbf9b5b..f2be9d6a3 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -15,7 +15,6 @@ use crate::admin::{ handlers::{bucket_meta, heal, health, kms, pools, profile_admin, quota, rebalance, replication, sts, system, tier, user}, router::{AdminOperation, S3Router}, - rpc, }; use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH}; use hyper::Method; @@ -54,8 +53,6 @@ fn test_register_routes_cover_representative_admin_paths() { replication::register_replication_route(&mut router).expect("register replication route"); profile_admin::register_profiling_route(&mut router).expect("register profile route"); kms::register_kms_route(&mut router).expect("register kms route"); - rpc::register_rpc_route(&mut router).expect("register rpc route"); - assert_route(&router, Method::GET, HEALTH_PREFIX); assert_route(&router, Method::HEAD, HEALTH_PREFIX); assert_route(&router, Method::GET, HEALTH_READY_PATH); @@ -117,8 +114,11 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::POST, &admin_path("/v3/kms/keys")); assert_route(&router, Method::GET, &admin_path("/v3/kms/keys")); assert_route(&router, Method::GET, &admin_path("/v3/kms/keys/test-key")); - assert_route(&router, Method::GET, "/rustfs/rpc/read_file_stream"); - assert_route(&router, Method::HEAD, "/rustfs/rpc/read_file_stream"); + + assert!( + !router.contains_route(Method::GET, "/rustfs/rpc/read_file_stream"), + "internode rpc routes should no longer be registered inside the admin router" + ); } #[test] @@ -160,9 +160,8 @@ fn test_admin_alias_paths_match_existing_admin_routes() { } #[test] -fn test_phase5_admin_info_and_rpc_read_file_contract() { +fn test_phase5_admin_info_contract() { let system_src = include_str!("handlers/system.rs"); - let rpc_src = include_str!("rpc.rs"); let server_info_impl_marker = "impl Operation for ServerInfoHandler"; let server_info_impl_start = system_src @@ -175,24 +174,4 @@ fn test_phase5_admin_info_and_rpc_read_file_contract() { && server_info_impl_block.contains("execute_query_server_info(QueryServerInfoRequest { include_pools: true })"), "admin server info path must be served through DefaultAdminUsecase::execute_query_server_info" ); - - let register_route_marker = "pub fn register_rpc_route"; - let register_route_start = rpc_src - .find(register_route_marker) - .expect("Expected register_rpc_route in rpc.rs"); - let register_route_block = &rpc_src[register_route_start..]; - - let read_file_marker = "pub struct ReadFile {}"; - let read_file_start = rpc_src.find(read_file_marker).expect("Expected ReadFile operation in rpc.rs"); - let read_file_block = &rpc_src[read_file_start..]; - - assert!( - register_route_block.contains("format!(\"{}{}\", RPC_PREFIX, \"/read_file_stream\")"), - "rpc read_file_stream route path must remain registered with RPC_PREFIX" - ); - - assert!( - read_file_block.contains(".read_file_stream(&query.volume, &query.path, query.offset, query.length)"), - "rpc read_file_stream route must remain wired to disk.read_file_stream" - ); } diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index abf61af7e..0197723be 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -14,9 +14,7 @@ use crate::admin::console::{is_console_path, make_console_server}; use crate::admin::handlers::oidc::is_oidc_path; -use crate::server::{ - ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, -}; +use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH}; use hyper::HeaderMap; use hyper::Method; use hyper::StatusCode; @@ -24,7 +22,6 @@ use hyper::Uri; use hyper::http::Extensions; use matchit::Params; use matchit::Router; -use rustfs_ecstore::rpc::verify_rpc_signature; use s3s::Body; use s3s::S3Request; use s3s::S3Response; @@ -33,7 +30,6 @@ use s3s::header; use s3s::route::S3Route; use s3s::s3_error; use tower::Service; -use tracing::error; pub struct S3Router { router: Router, @@ -140,7 +136,7 @@ where return true; } - is_admin_path(path) || path.starts_with(RPC_PREFIX) || is_console_path(path) + is_admin_path(path) || is_console_path(path) } // check_access before call @@ -168,18 +164,6 @@ where return Ok(()); } - // Check RPC signature verification - if req.uri.path().starts_with(RPC_PREFIX) { - // Skip signature verification for HEAD requests (health checks) - if req.method != Method::HEAD { - verify_rpc_signature(&req.uri.to_string(), &req.method, &req.headers).map_err(|e| { - error!("RPC signature verification failed: {}", e); - s3_error!(AccessDenied, "{}", e) - })?; - } - return Ok(()); - } - // Allow unauthenticated STS requests to POST / (AssumeRoleWithWebIdentity // doesn't use SigV4 — the JWT token in the request body is the authentication). // The handler dispatches on the Action parameter: AssumeRole will reject if diff --git a/rustfs/src/admin/rpc.rs b/rustfs/src/admin/rpc.rs deleted file mode 100644 index 0c6fbc66d..000000000 --- a/rustfs/src/admin/rpc.rs +++ /dev/null @@ -1,218 +0,0 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::admin::router::{AdminOperation, Operation, S3Router}; -use crate::server::RPC_PREFIX; -use futures::StreamExt; -use http::StatusCode; -use hyper::Method; -use matchit::Params; -use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; -use rustfs_ecstore::disk::DiskAPI; -use rustfs_ecstore::disk::WalkDirOptions; -use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; -use rustfs_ecstore::store::find_local_disk; -use rustfs_utils::net::bytes_stream; -use s3s::Body; -use s3s::S3Request; -use s3s::S3Response; -use s3s::S3Result; -use s3s::dto::StreamingBlob; -use s3s::s3_error; -use serde_urlencoded::from_bytes; -use tokio::io::AsyncWriteExt; -use tokio_util::io::ReaderStream; -use tracing::warn; - -pub fn register_rpc_route(r: &mut S3Router) -> std::io::Result<()> { - r.insert( - Method::GET, - format!("{}{}", RPC_PREFIX, "/read_file_stream").as_str(), - AdminOperation(&ReadFile {}), - )?; - - r.insert( - Method::HEAD, - format!("{}{}", RPC_PREFIX, "/read_file_stream").as_str(), - AdminOperation(&ReadFile {}), - )?; - - r.insert( - Method::PUT, - format!("{}{}", RPC_PREFIX, "/put_file_stream").as_str(), - AdminOperation(&PutFile {}), - )?; - - r.insert( - Method::GET, - format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(), - AdminOperation(&WalkDir {}), - )?; - - r.insert( - Method::HEAD, - format!("{}{}", RPC_PREFIX, "/walk_dir").as_str(), - AdminOperation(&WalkDir {}), - )?; - - Ok(()) -} - -// /rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}" -#[derive(Debug, Default, serde::Deserialize)] -pub struct ReadFileQuery { - disk: String, - volume: String, - path: String, - offset: usize, - length: usize, -} -pub struct ReadFile {} -#[async_trait::async_trait] -impl Operation for ReadFile { - async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - if req.method == Method::HEAD { - return Ok(S3Response::new((StatusCode::OK, Body::empty()))); - } - let query = { - if let Some(query) = req.uri.query() { - let input: ReadFileQuery = - from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?; - input - } else { - ReadFileQuery::default() - } - }; - - let Some(disk) = find_local_disk(&query.disk).await else { - return Err(s3_error!(InvalidArgument, "disk not found")); - }; - - let file = disk - .read_file_stream(&query.volume, &query.path, query.offset, query.length) - .await - .map_err(|e| s3_error!(InternalError, "read file err {}", e))?; - - Ok(S3Response::new(( - StatusCode::OK, - Body::from(StreamingBlob::wrap(bytes_stream( - ReaderStream::with_capacity(file, DEFAULT_READ_BUFFER_SIZE), - query.length, - ))), - ))) - } -} - -#[derive(Debug, Default, serde::Deserialize)] -pub struct WalkDirQuery { - disk: String, -} - -pub struct WalkDir {} - -#[async_trait::async_trait] -impl Operation for WalkDir { - async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - if req.method == Method::HEAD { - return Ok(S3Response::new((StatusCode::OK, Body::empty()))); - } - - let query = { - if let Some(query) = req.uri.query() { - let input: WalkDirQuery = - from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?; - input - } else { - WalkDirQuery::default() - } - }; - - let mut input = req.input; - let body = match input.store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE).await { - Ok(b) => b, - Err(e) => { - warn!("get body failed, e: {:?}", e); - return Err(s3_error!(InvalidRequest, "RPC request body too large or failed to read")); - } - }; - - // let body_bytes = decrypt_data(input_cred.secret_key.expose().as_bytes(), &body) - // .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidArgument, format!("decrypt_data err {}", e)))?; - - let args: WalkDirOptions = - serde_json::from_slice(&body).map_err(|e| s3_error!(InternalError, "unmarshal body err {}", e))?; - let Some(disk) = find_local_disk(&query.disk).await else { - return Err(s3_error!(InvalidArgument, "disk not found")); - }; - - let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); - - tokio::spawn(async move { - if let Err(e) = disk.walk_dir(args, &mut wd).await { - warn!("walk dir err {}", e); - } - }); - - let body = Body::from(StreamingBlob::wrap(ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE))); - Ok(S3Response::new((StatusCode::OK, body))) - } -} - -// /rustfs/rpc/read_file_stream?disk={}&volume={}&path={}&offset={}&length={}" -#[derive(Debug, Default, serde::Deserialize)] -pub struct PutFileQuery { - disk: String, - volume: String, - path: String, - append: bool, - size: i64, -} -pub struct PutFile {} -#[async_trait::async_trait] -impl Operation for PutFile { - async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { - let query = { - if let Some(query) = req.uri.query() { - let input: PutFileQuery = - from_bytes(query.as_bytes()).map_err(|e| s3_error!(InvalidArgument, "get query failed1 {:?}", e))?; - input - } else { - PutFileQuery::default() - } - }; - - let Some(disk) = find_local_disk(&query.disk).await else { - return Err(s3_error!(InvalidArgument, "disk not found")); - }; - - let mut file = if query.append { - disk.append_file(&query.volume, &query.path) - .await - .map_err(|e| s3_error!(InternalError, "append file err {}", e))? - } else { - disk.create_file("", &query.volume, &query.path, query.size) - .await - .map_err(|e| s3_error!(InternalError, "read file err {}", e))? - }; - - let mut body = req.input; - while let Some(item) = body.next().await { - let bytes = item.map_err(|e| s3_error!(InternalError, "body stream err {}", e))?; - let result = file.write_all(&bytes).await; - result.map_err(|e| s3_error!(InternalError, "write file err {}", e))?; - } - - Ok(S3Response::new((StatusCode::OK, Body::empty()))) - } -} diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 07705a1c1..31d59c048 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -530,14 +530,17 @@ async fn run(config: config::Config) -> Result<()> { "Background services configuration: scanner={}, heal={}", enable_scanner, enable_heal ); - // Initialize heal manager and scanner based on environment variables + // Scanner depends on the heal channel/manager, so scanner implies heal. if enable_heal || enable_scanner { let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone())); - init_heal_manager(heal_storage, None).await?; + } + if enable_scanner { init_data_scanner(ctx.clone(), store.clone()).await; - } else { + } + + if !enable_heal && !enable_scanner { info!(target: "rustfs::main::run","Both scanner and heal are disabled, skipping AHM service initialization"); } @@ -623,20 +626,24 @@ async fn handle_shutdown( let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true); let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true); - // Stop background services based on what was enabled - if enable_scanner || enable_heal { + // Stop background services based on what was enabled. + if enable_scanner { info!( target: "rustfs::main::handle_shutdown", - "Stopping background services (data scanner and auto heal)..." + "Stopping background services (data scanner)..." ); shutdown_background_services(); + } + if enable_heal || enable_scanner { info!( target: "rustfs::main::handle_shutdown", "Stopping AHM services..." ); shutdown_ahm_services(); - } else { + } + + if !enable_scanner && !enable_heal { info!( target: "rustfs::main::handle_shutdown", "Background services were disabled, skipping AHM shutdown" diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index edb0aec15..85dc8e932 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -24,6 +24,7 @@ use crate::server::{ layer::{AdminChunkedContentLengthCompatLayer, ConditionalCorsLayer, ObjectAttributesEtagFixLayer, RedirectLayer}, }; use crate::storage; +use crate::storage::rpc::InternodeRpcService; use crate::storage::tonic_service::make_server; use bytes::Bytes; use http::{HeaderMap, Method, Request as HttpRequest, Response}; @@ -592,6 +593,7 @@ fn process_connection( let http_service = SwiftService::new(true, None, s3_service); #[cfg(not(feature = "swift"))] let http_service = s3_service; + let http_service = InternodeRpcService::new(http_service); let service = hybrid(http_service, rpc_service); diff --git a/rustfs/src/storage/rpc/disk.rs b/rustfs/src/storage/rpc/disk.rs index d1f469f2d..63749477b 100644 --- a/rustfs/src/storage/rpc/disk.rs +++ b/rustfs/src/storage/rpc/disk.rs @@ -13,6 +13,26 @@ // limitations under the License. use super::*; +use serde::de::DeserializeOwned; +use std::io::Cursor; + +fn decode_msgpack_or_json(binary: &[u8], json: &str, value_name: &str) -> std::result::Result { + if !binary.is_empty() { + let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); + return T::deserialize(&mut deserializer) + .map_err(|err| DiskError::other(format!("decode {value_name} msgpack failed: {err}"))); + } + + serde_json::from_str(json).map_err(|err| DiskError::other(format!("decode {value_name} failed: {err}"))) +} + +fn encode_msgpack(value: &T, value_name: &str) -> std::result::Result, DiskError> { + let mut serializer = rmp_serde::Serializer::new(Vec::new()); + value + .serialize(&mut serializer) + .map_err(|err| DiskError::other(format!("encode {value_name} msgpack failed: {err}")))?; + Ok(serializer.into_inner()) +} impl NodeService { pub(super) async fn handle_disk_info(&self, request: Request) -> Result, Status> { @@ -86,32 +106,44 @@ impl NodeService { ) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let read_multiple_req = match serde_json::from_str::(&request.read_multiple_req) { + let read_multiple_req = match decode_msgpack_or_json::( + &request.read_multiple_req_bin, + &request.read_multiple_req, + "ReadMultipleReq", + ) { Ok(read_multiple_req) => read_multiple_req, Err(err) => { return Ok(Response::new(ReadMultipleResponse { success: false, read_multiple_resps: Vec::new(), + read_multiple_resps_bin: Vec::new(), error: Some(DiskError::other(format!("decode ReadMultipleReq failed: {err}")).into()), })); } }; match disk.read_multiple(read_multiple_req).await { Ok(read_multiple_resps) => { - let read_multiple_resps = read_multiple_resps + let read_multiple_resps: Vec = read_multiple_resps .into_iter() .filter_map(|read_multiple_resp| serde_json::to_string(&read_multiple_resp).ok()) .collect(); + let read_multiple_resps_bin = read_multiple_resps + .iter() + .filter_map(|json_str| serde_json::from_str::(json_str).ok()) + .filter_map(|resp| encode_msgpack(&resp, "ReadMultipleResp").ok()) + .collect(); Ok(Response::new(ReadMultipleResponse { success: true, read_multiple_resps, + read_multiple_resps_bin, error: None, })) } Err(err) => Ok(Response::new(ReadMultipleResponse { success: false, read_multiple_resps: Vec::new(), + read_multiple_resps_bin: Vec::new(), error: Some(err.into()), })), } @@ -119,6 +151,7 @@ impl NodeService { Ok(Response::new(ReadMultipleResponse { success: false, read_multiple_resps: Vec::new(), + read_multiple_resps_bin: Vec::new(), error: Some(DiskError::other("can not find disk".to_string()).into()), })) } @@ -239,21 +272,34 @@ impl NodeService { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { match disk.read_xl(&request.volume, &request.path, request.read_data).await { - Ok(raw_file_info) => match serde_json::to_string(&raw_file_info) { - Ok(raw_file_info) => Ok(Response::new(ReadXlResponse { - success: true, - raw_file_info, - error: None, - })), - Err(err) => Ok(Response::new(ReadXlResponse { - success: false, - raw_file_info: String::new(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), - }, + Ok(raw_file_info) => { + let raw_file_info_json = serde_json::to_string(&raw_file_info); + let raw_file_info_bin = encode_msgpack(&raw_file_info, "RawFileInfo"); + match (raw_file_info_json, raw_file_info_bin) { + (Ok(raw_file_info), Ok(raw_file_info_bin)) => Ok(Response::new(ReadXlResponse { + success: true, + raw_file_info, + raw_file_info_bin, + error: None, + })), + (Err(err), _) => Ok(Response::new(ReadXlResponse { + success: false, + raw_file_info: String::new(), + raw_file_info_bin: Vec::new(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })), + (_, Err(err)) => Ok(Response::new(ReadXlResponse { + success: false, + raw_file_info: String::new(), + raw_file_info_bin: Vec::new(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })), + } + } Err(err) => Ok(Response::new(ReadXlResponse { success: false, raw_file_info: String::new(), + raw_file_info_bin: Vec::new(), error: Some(err.into()), })), } @@ -261,6 +307,7 @@ impl NodeService { Ok(Response::new(ReadXlResponse { success: false, raw_file_info: String::new(), + raw_file_info_bin: Vec::new(), error: Some(DiskError::other("can not find disk".to_string()).into()), })) } @@ -272,12 +319,13 @@ impl NodeService { ) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let opts = match serde_json::from_str::(&request.opts) { + let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "ReadOptions") { Ok(options) => options, Err(err) => { return Ok(Response::new(ReadVersionResponse { success: false, file_info: String::new(), + file_info_bin: Vec::new(), error: Some(DiskError::other(format!("decode ReadOptions failed: {err}")).into()), })); } @@ -286,21 +334,34 @@ impl NodeService { .read_version("", &request.volume, &request.path, &request.version_id, &opts) .await { - Ok(file_info) => match serde_json::to_string(&file_info) { - Ok(file_info) => Ok(Response::new(ReadVersionResponse { - success: true, - file_info, - error: None, - })), - Err(err) => Ok(Response::new(ReadVersionResponse { - success: false, - file_info: String::new(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), - }, + Ok(file_info) => { + let file_info_json = serde_json::to_string(&file_info); + let file_info_bin = encode_msgpack(&file_info, "FileInfo"); + match (file_info_json, file_info_bin) { + (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse { + success: true, + file_info, + file_info_bin, + error: None, + })), + (Err(err), _) => Ok(Response::new(ReadVersionResponse { + success: false, + file_info: String::new(), + file_info_bin: Vec::new(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })), + (_, Err(err)) => Ok(Response::new(ReadVersionResponse { + success: false, + file_info: String::new(), + file_info_bin: Vec::new(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })), + } + } Err(err) => Ok(Response::new(ReadVersionResponse { success: false, file_info: String::new(), + file_info_bin: Vec::new(), error: Some(err.into()), })), } @@ -308,6 +369,7 @@ impl NodeService { Ok(Response::new(ReadVersionResponse { success: false, file_info: String::new(), + file_info_bin: Vec::new(), error: Some(DiskError::other("can not find disk".to_string()).into()), })) } @@ -319,7 +381,7 @@ impl NodeService { ) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let file_info = match serde_json::from_str::(&request.file_info) { + let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info, Err(err) => { return Ok(Response::new(WriteMetadataResponse { @@ -352,7 +414,7 @@ impl NodeService { ) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let file_info = match serde_json::from_str::(&request.file_info) { + let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info, Err(err) => { return Ok(Response::new(UpdateMetadataResponse { @@ -361,7 +423,8 @@ impl NodeService { })); } }; - let opts = match serde_json::from_str::(&request.opts) { + let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "UpdateMetadataOpts") + { Ok(opts) => opts, Err(err) => { return Ok(Response::new(UpdateMetadataResponse { @@ -910,3 +973,42 @@ impl NodeService { } } } + +#[cfg(test)] +mod tests { + use super::*; + use serde::{Deserialize, Serialize}; + + #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] + struct SamplePayload { + name: String, + count: u32, + } + + #[test] + fn decode_msgpack_or_json_prefers_binary_payload() { + let payload = SamplePayload { + name: "rustfs".to_string(), + count: 3, + }; + + let binary = encode_msgpack(&payload, "SamplePayload").unwrap(); + let decoded = + decode_msgpack_or_json::(&binary, r#"{"name":"ignored","count":1}"#, "SamplePayload").unwrap(); + + assert_eq!(decoded, payload); + } + + #[test] + fn decode_msgpack_or_json_falls_back_to_json() { + let decoded = decode_msgpack_or_json::(&[], r#"{"name":"compat","count":7}"#, "SamplePayload").unwrap(); + + assert_eq!( + decoded, + SamplePayload { + name: "compat".to_string(), + count: 7, + } + ); + } +} diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs new file mode 100644 index 000000000..48e3713a1 --- /dev/null +++ b/rustfs/src/storage/rpc/http_service.rs @@ -0,0 +1,404 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::server::RPC_PREFIX; +use bytes::{Bytes, BytesMut}; +use futures_util::TryStreamExt; +use http::{HeaderMap, Method, Request, Response, StatusCode, Uri}; +use http_body_util::{BodyExt, Limited}; +use hyper::body::Incoming; +use rustfs_common::internode_metrics::global_internode_metrics; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_ecstore::disk::{DiskAPI, WalkDirOptions}; +use rustfs_ecstore::rpc::verify_rpc_signature; +use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; +use rustfs_ecstore::store::find_local_disk_by_ref; +use rustfs_utils::net::bytes_stream; +use s3s::Body; +use s3s::dto::StreamingBlob; +use serde::de::DeserializeOwned; +use serde_urlencoded::from_bytes; +use std::future::Future; +use std::pin::Pin; +use std::task::{Context, Poll}; +use tokio::io::{self, AsyncWriteExt}; +use tokio_util::io::ReaderStream; +use tower::Service; +use tracing::warn; + +type BoxError = Box; +type RpcErrorResponse = Box>; +const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; +const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; + +#[derive(Clone)] +pub struct InternodeRpcService { + inner: S, +} + +impl InternodeRpcService { + pub fn new(inner: S) -> Self { + Self { inner } + } +} + +#[derive(Debug, Default, serde::Deserialize)] +struct ReadFileQuery { + disk: String, + volume: String, + path: String, + offset: usize, + length: usize, +} + +#[derive(Debug, Default, serde::Deserialize)] +struct WalkDirQuery { + disk: String, +} + +#[derive(Debug, Default, serde::Deserialize)] +struct PutFileQuery { + disk: String, + volume: String, + path: String, + append: bool, + size: i64, +} + +impl Service> for InternodeRpcService +where + S: Service, Response = Response> + Clone + Send + 'static, + S::Future: Send + 'static, + S::Error: Into + Send + 'static, +{ + type Response = Response; + type Error = S::Error; + type Future = Pin> + Send>>; + + fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + self.inner.poll_ready(cx) + } + + fn call(&mut self, req: Request) -> Self::Future { + if !is_internode_rpc_path(req.uri().path()) { + let mut inner = self.inner.clone(); + return Box::pin(async move { inner.call(req).await }); + } + + Box::pin(async move { Ok(handle_internode_rpc(req).await) }) + } +} + +fn is_internode_rpc_path(path: &str) -> bool { + path.starts_with(RPC_PREFIX) +} + +async fn handle_internode_rpc(req: Request) -> Response { + if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) { + global_internode_metrics().record_error(); + return *response; + } + + let method = req.method().clone(); + let path = req.uri().path(); + + let response = match (method, path) { + (Method::GET, READ_FILE_STREAM_PATH) | (Method::HEAD, READ_FILE_STREAM_PATH) => handle_read_file(req).await, + (Method::GET, WALK_DIR_PATH) | (Method::HEAD, WALK_DIR_PATH) => handle_walk_dir(req).await, + (Method::PUT, PUT_FILE_STREAM_PATH) => handle_put_file(req).await, + _ => response_with_status(StatusCode::NOT_FOUND, "internode rpc route not found"), + }; + + if !response.status().is_success() { + global_internode_metrics().record_error(); + } + + response +} + +fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMap) -> Result<(), RpcErrorResponse> { + if method == Method::HEAD { + return Ok(()); + } + + verify_rpc_signature(&uri.to_string(), method, headers).map_err(|e| { + Box::new(response_with_status( + StatusCode::FORBIDDEN, + format!("rpc signature verification failed: {e}"), + )) + }) +} + +async fn handle_read_file(req: Request) -> Response { + if req.method() == Method::HEAD { + return empty_ok(); + } + + let query = match parse_query::(&req) { + Ok(query) => query, + Err(response) => return *response, + }; + + let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); + }; + + let file = match disk + .read_file_stream(&query.volume, &query.path, query.offset, query.length) + .await + { + Ok(file) => file, + Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")), + }; + + global_internode_metrics().record_incoming_request(); + let stream = read_file_body_stream(file, query.length); + + Response::builder() + .status(StatusCode::OK) + .body(Body::from(StreamingBlob::wrap(stream))) + .expect("failed to build read file stream response") +} + +fn read_file_body_stream(reader: R, length: usize) -> Pin> + Send + Sync>> +where + R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static, +{ + let metrics = global_internode_metrics().clone(); + let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { + metrics.record_sent_bytes(bytes.len()); + bytes + }); + + if length == 0 { + Box::pin(stream) + } else { + Box::pin(bytes_stream(stream, length)) + } +} + +async fn handle_walk_dir(req: Request) -> Response { + if req.method() == Method::HEAD { + return empty_ok(); + } + + let query = match parse_query::(&req) { + Ok(query) => query, + Err(response) => return *response, + }; + + let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); + }; + + let body = match Limited::new(req.into_body(), MAX_ADMIN_REQUEST_BODY_SIZE).collect().await { + Ok(body) => body.to_bytes(), + Err(e) => return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, format!("read body err {e}")), + }; + + let args: WalkDirOptions = match serde_json::from_slice(&body) { + Ok(args) => args, + Err(e) => return response_with_status(StatusCode::BAD_REQUEST, format!("unmarshal body err {e}")), + }; + + let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE); + tokio::spawn(async move { + if let Err(e) = disk.walk_dir(args, &mut wd).await { + warn!("walk dir err {}", e); + } + }); + + global_internode_metrics().record_incoming_request(); + let metrics = global_internode_metrics().clone(); + let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { + metrics.record_sent_bytes(bytes.len()); + bytes + }); + + Response::builder() + .status(StatusCode::OK) + .body(Body::from(StreamingBlob::wrap(stream))) + .expect("failed to build walk dir response") +} + +async fn handle_put_file(req: Request) -> Response { + let query = match parse_query::(&req) { + Ok(query) => query, + Err(response) => return *response, + }; + + let Some(disk) = find_local_disk_by_ref(&query.disk).await else { + return response_with_status(StatusCode::BAD_REQUEST, "disk not found"); + }; + + let mut file = if query.append { + match disk.append_file(&query.volume, &query.path).await { + Ok(file) => file, + Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("append file err {e}")), + } + } else { + match disk.create_file("", &query.volume, &query.path, query.size).await { + Ok(file) => file, + Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("create file err {e}")), + } + }; + + let copied = match write_body_chunks_to_writer(req.into_body().into_data_stream(), &mut file).await { + Ok(copied) => copied, + Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")), + }; + + global_internode_metrics().record_incoming_request(); + global_internode_metrics().record_recv_bytes(copied as usize); + + if let Err(e) = file.flush().await { + return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")); + } + + empty_ok() +} + +async fn write_body_chunks_to_writer(body: S, writer: &mut W) -> io::Result +where + S: futures::TryStream + Unpin, + E: Into, + W: tokio::io::AsyncWrite + Unpin, +{ + let mut body = body; + let mut copied = 0_u64; + let mut pending = BytesMut::with_capacity(DEFAULT_READ_BUFFER_SIZE); + + while let Some(bytes) = body.try_next().await.map_err(io::Error::other)? { + copied += bytes.len() as u64; + pending.extend_from_slice(&bytes); + + if pending.len() >= DEFAULT_READ_BUFFER_SIZE { + writer.write_all(&pending).await?; + pending.clear(); + } + } + + if !pending.is_empty() { + writer.write_all(&pending).await?; + } + + Ok(copied) +} + +fn parse_query(req: &Request) -> Result +where + T: DeserializeOwned + Default, +{ + match req.uri().query() { + Some(query) => from_bytes(query.as_bytes()) + .map_err(|e| Box::new(response_with_status(StatusCode::BAD_REQUEST, format!("get query failed {e}")))), + None => Ok(T::default()), + } +} + +fn empty_ok() -> Response { + Response::builder() + .status(StatusCode::OK) + .body(Body::empty()) + .expect("failed to build empty ok response") +} + +fn response_with_status(status: StatusCode, message: impl Into) -> Response { + Response::builder() + .status(status) + .header(http::header::CONTENT_TYPE, "text/plain; charset=utf-8") + .body(Body::from(Bytes::from(message.into()))) + .expect("failed to build rpc error response") +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio_stream::StreamExt; + use tokio_stream::iter; + + #[test] + fn internode_rpc_path_matches_rpc_prefix() { + assert!(is_internode_rpc_path("/rustfs/rpc/read_file_stream")); + assert!(is_internode_rpc_path("/rustfs/rpc/walk_dir")); + assert!(!is_internode_rpc_path("/rustfs/admin/v3/info")); + } + + #[test] + fn rpc_head_signature_verification_is_skipped() { + let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri"); + let headers = HeaderMap::new(); + assert!(verify_internode_rpc_signature(&uri, &Method::HEAD, &headers).is_ok()); + } + + #[test] + fn rpc_get_request_requires_signature() { + let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri"); + let headers = HeaderMap::new(); + let response = verify_internode_rpc_signature(&uri, &Method::GET, &headers).expect_err("response"); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + } + + #[tokio::test] + async fn write_body_chunks_to_writer_streams_all_chunks() { + let (mut reader, mut writer) = tokio::io::duplex(64); + let body = iter(vec![ + Ok::(Bytes::from_static(b"hello ")), + Ok(Bytes::from_static(b"world")), + ]); + + let copied = write_body_chunks_to_writer(body, &mut writer).await.expect("copy succeeds"); + drop(writer); + + let mut out = Vec::new(); + reader.read_to_end(&mut out).await.expect("read succeeds"); + + assert_eq!(copied, 11); + assert_eq!(out, b"hello world"); + } + + #[tokio::test] + async fn read_file_body_stream_keeps_full_stream_when_length_is_zero() { + let (reader, mut writer) = tokio::io::duplex(64); + tokio::spawn(async move { + writer.write_all(b"hello world").await.expect("write succeeds"); + }); + + let mut stream = read_file_body_stream(reader, 0); + let mut out = Vec::new(); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.expect("chunk succeeds")); + } + + assert_eq!(out, b"hello world"); + } + + #[tokio::test] + async fn read_file_body_stream_truncates_to_requested_length() { + let (reader, mut writer) = tokio::io::duplex(64); + tokio::spawn(async move { + writer.write_all(b"hello world").await.expect("write succeeds"); + }); + + let mut stream = read_file_body_stream(reader, 5); + let mut out = Vec::new(); + while let Some(chunk) = stream.next().await { + out.extend_from_slice(&chunk.expect("chunk succeeds")); + } + + assert_eq!(out, b"hello"); + } +} diff --git a/rustfs/src/storage/rpc/mod.rs b/rustfs/src/storage/rpc/mod.rs index c4e0c59c5..bef5bb854 100644 --- a/rustfs/src/storage/rpc/mod.rs +++ b/rustfs/src/storage/rpc/mod.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod http_service; pub mod node_service; +pub use http_service::InternodeRpcService; pub use node_service::{NodeService, make_server}; diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index ec2088922..35355a40b 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -21,14 +21,14 @@ use rustfs_ecstore::{ admin_server_info::get_local_server_property, bucket::{metadata::load_bucket_metadata, metadata_sys}, disk::{ - DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts, - error::DiskError, + DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadMultipleResp, ReadOptions, + UpdateMetadataOpts, error::DiskError, }, get_global_lock_client, metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics}, new_object_layer_fn, rpc::{LocalPeerS3Client, PeerS3Client}, - store::{all_local_disk_path, find_local_disk}, + store::{all_local_disk_path, find_local_disk_by_ref}, store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI}, }; use rustfs_filemeta::{FileInfo, MetacacheReader}; @@ -74,8 +74,8 @@ pub fn make_server() -> NodeService { } impl NodeService { - async fn find_disk(&self, disk_path: &String) -> Option { - find_local_disk(disk_path).await + async fn find_disk(&self, disk_path: &str) -> Option { + find_local_disk_by_ref(disk_path).await } async fn all_disk(&self) -> Vec { @@ -1353,6 +1353,8 @@ mod tests { path: "test-path".to_string(), file_info: "{}".to_string(), opts: "{}".to_string(), + file_info_bin: Vec::new(), + opts_bin: Vec::new(), }); let response = service.update_metadata(request).await; @@ -1373,6 +1375,8 @@ mod tests { path: "test-path".to_string(), file_info: "invalid json".to_string(), opts: "{}".to_string(), + file_info_bin: Vec::new(), + opts_bin: Vec::new(), }); let response = service.update_metadata(request).await; @@ -1393,6 +1397,8 @@ mod tests { path: "test-path".to_string(), file_info: "{}".to_string(), opts: "invalid json".to_string(), + file_info_bin: Vec::new(), + opts_bin: Vec::new(), }); let response = service.update_metadata(request).await; @@ -1412,6 +1418,7 @@ mod tests { volume: "test-volume".to_string(), path: "test-path".to_string(), file_info: "{}".to_string(), + file_info_bin: Vec::new(), }); let response = service.write_metadata(request).await; @@ -1431,6 +1438,7 @@ mod tests { volume: "test-volume".to_string(), path: "test-path".to_string(), file_info: "invalid json".to_string(), + file_info_bin: Vec::new(), }); let response = service.write_metadata(request).await; @@ -1451,6 +1459,7 @@ mod tests { path: "test-path".to_string(), version_id: "version1".to_string(), opts: "{}".to_string(), + opts_bin: Vec::new(), }); let response = service.read_version(request).await; @@ -1472,6 +1481,7 @@ mod tests { path: "test-path".to_string(), version_id: "version1".to_string(), opts: "invalid json".to_string(), + opts_bin: Vec::new(), }); let response = service.read_version(request).await; @@ -1629,6 +1639,7 @@ mod tests { let request = Request::new(ReadMultipleRequest { disk: "invalid-disk-path".to_string(), read_multiple_req: "{}".to_string(), + read_multiple_req_bin: Vec::new(), }); let response = service.read_multiple(request).await; @@ -1647,6 +1658,7 @@ mod tests { let request = Request::new(ReadMultipleRequest { disk: "invalid-disk-path".to_string(), read_multiple_req: "invalid json".to_string(), + read_multiple_req_bin: Vec::new(), }); let response = service.read_multiple(request).await; @@ -2195,7 +2207,7 @@ mod tests { #[tokio::test] async fn test_find_disk_method() { let service = create_test_node_service(); - let disk = service.find_disk(&"non-existent-disk".to_string()).await; + let disk = service.find_disk("non-existent-disk").await; // Should return None for non-existent disk assert!(disk.is_none()); }