gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(logging): reduce listing cancellation error noise (#4372)

Browse Source 
* fix(logging): reduce listing cancellation error noise

* fix(ecstore): preserve filemeta io error kind

* fix(ecstore): avoid redundant clone lint in test

---------

Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: overtrue <anzhengchao@gmail.com>
This commit is contained in:
cxymds
2026-07-08 09:35:24 +08:00
committed by GitHub
parent 3c3113619e
commit d25ddb0e1e
4 changed files with 240 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/cache_value/metacache_set.rs
+55 -1
View File
@@ -280,6 +280,13 @@ async fn list_path_raw_inner(
);
}
Err(err) => {
if err.is_metacache_output_stream_closed() {
rustfs_io_metrics::record_stage_duration(
"metacache_walk_dir_primary",
primary_walk_started.elapsed().as_secs_f64() * 1000.0,
);
return Ok(());
}
rustfs_io_metrics::record_stage_duration(
"metacache_walk_dir_primary_failed",
primary_walk_started.elapsed().as_secs_f64() * 1000.0,
@@ -427,6 +434,13 @@ async fn list_path_raw_inner(
last_err = None;
}
Err(err) => {
if err.is_metacache_output_stream_closed() {
rustfs_io_metrics::record_stage_duration(
"metacache_walk_dir_fallback",
fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
);
return Ok(());
}
rustfs_io_metrics::record_stage_duration(
"metacache_walk_dir_fallback_failed",
fallback_walk_started.elapsed().as_secs_f64() * 1000.0,
@@ -490,7 +504,7 @@ async fn list_path_raw_inner(
// );
if rx.is_cancelled() {
return Err(DiskError::other("canceled"));
return Ok(());
}
let mut top_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];
@@ -780,6 +794,9 @@ async fn list_path_raw_inner(
match result {
Ok(Ok(())) => {}
Ok(Err(err)) => {
if err.is_metacache_output_stream_closed() {
continue;
}
if is_missing_path_error(&err) {
debug!(
event = EVENT_METACACHE_LISTING,
@@ -1084,6 +1101,43 @@ mod tests {
listing.expect("unresponsive producer within failure budget should not fail quorum EOF");
}
#[tokio::test]
async fn list_path_raw_treats_external_cancel_as_successful_shutdown() {
let cancel = CancellationToken::new();
cancel.cancel();
list_path_raw(
cancel,
ListPathRawOptions {
disks: vec![None],
min_disks: 1,
test_reader_behaviors: vec![TestReaderBehavior::IgnoreCancel],
..Default::default()
},
)
.await
.expect("external cancellation should stop listing without a synthetic canceled error");
}
#[tokio::test]
async fn list_path_raw_ignores_closed_output_stream_after_quorum_eof() {
list_path_raw(
CancellationToken::new(),
ListPathRawOptions {
disks: vec![None, None, None],
min_disks: 2,
test_reader_behaviors: vec![
TestReaderBehavior::Eof,
TestReaderBehavior::Eof,
TestReaderBehavior::ProducerError(DiskError::metacache_output_stream_closed()),
],
..Default::default()
},
)
.await
.expect("closed metacache output after quorum EOF should not fail listing");
}
#[tokio::test]
async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() {
let seen = Arc::new(Mutex::new(Vec::new()));
crates/ecstore/src/disk/error.rs
+66 -1
View File
@@ -13,6 +13,7 @@
// limitations under the License.
use rustfs_rio::{InternodeHttpError, InternodeHttpErrorKind};
use std::error::Error as StdError;
use std::hash::{Hash, Hasher};
use std::io::{self};
use std::path::PathBuf;
@@ -20,6 +21,8 @@ use std::path::PathBuf;
pub type Error = DiskError;
pub type Result<T> = core::result::Result<T, Error>;
const METACACHE_OUTPUT_STREAM_CLOSED: &str = "metacache output stream closed";
// DiskError == StorageErr
#[derive(Debug, thiserror::Error)]
pub enum DiskError {
@@ -158,6 +161,26 @@ impl DiskError {
DiskError::Io(std::io::Error::other(error))
}
pub(crate) fn metacache_output_stream_closed() -> Self {
DiskError::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, METACACHE_OUTPUT_STREAM_CLOSED))
}
pub(crate) fn is_metacache_output_stream_closed(&self) -> bool {
matches!(
self,
DiskError::Io(io_error)
if io_error.kind() == std::io::ErrorKind::BrokenPipe
&& io_error.to_string() == METACACHE_OUTPUT_STREAM_CLOSED
)
}
pub(crate) fn contains_io_error_kind(&self, kind: std::io::ErrorKind) -> bool {
match self {
DiskError::Io(io_error) => io_error_chain_contains_kind(io_error, kind),
_ => false,
}
}
pub fn is_all_not_found(errs: &[Option<DiskError>]) -> bool {
for err in errs.iter() {
if let Some(err) = err {
@@ -250,7 +273,7 @@ impl DiskError {
impl From<rustfs_filemeta::Error> for DiskError {
fn from(e: rustfs_filemeta::Error) -> Self {
match e {
rustfs_filemeta::Error::Io(e) => DiskError::other(e),
rustfs_filemeta::Error::Io(e) => DiskError::Io(e),
rustfs_filemeta::Error::FileNotFound => DiskError::FileNotFound,
rustfs_filemeta::Error::FileVersionNotFound => DiskError::FileVersionNotFound,
rustfs_filemeta::Error::FileCorrupt => DiskError::FileCorrupt,
@@ -260,6 +283,29 @@ impl From<rustfs_filemeta::Error> for DiskError {
}
}
fn io_error_chain_contains_kind(io_error: &std::io::Error, kind: std::io::ErrorKind) -> bool {
if io_error.kind() == kind {
return true;
}
let mut source = StdError::source(io_error);
while let Some(err) = source {
if let Some(io_error) = err.downcast_ref::<std::io::Error>()
&& io_error.kind() == kind
{
return true;
}
if let Some(disk_error) = err.downcast_ref::<DiskError>()
&& disk_error.contains_io_error_kind(kind)
{
return true;
}
source = err.source();
}
false
}
impl From<std::io::Error> for DiskError {
fn from(e: std::io::Error) -> Self {
match e.downcast::<DiskError>() {
@@ -900,6 +946,25 @@ mod tests {
assert!(converted_io.to_string().contains("broken pipe"));
}
#[test]
fn test_wrapped_io_error_kind_detection() {
let filemeta_error = rustfs_filemeta::Error::Io(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "broken pipe"));
let disk_error = DiskError::from(filemeta_error);
assert!(disk_error.contains_io_error_kind(std::io::ErrorKind::BrokenPipe));
assert!(!disk_error.contains_io_error_kind(std::io::ErrorKind::TimedOut));
}
#[test]
fn test_metacache_output_stream_closed_classification_survives_clone() {
let disk_error = DiskError::metacache_output_stream_closed();
let cloned_error = disk_error.clone();
assert!(disk_error.is_metacache_output_stream_closed());
assert!(cloned_error.is_metacache_output_stream_closed());
assert_eq!(disk_error, cloned_error);
}
#[test]
fn test_error_display_preservation() {
let disk_errors = vec![
crates/ecstore/src/disk/local.rs
+105 -33
View File
@@ -834,6 +834,22 @@ fn is_bitrot_size_mismatch_error(err: &std::io::Error) -> bool {
err.to_string().contains("bitrot shard file size mismatch")
}
fn metacache_write_error(err: rustfs_filemeta::Error) -> DiskError {
let err = DiskError::from(err);
if err.contains_io_error_kind(ErrorKind::BrokenPipe) {
DiskError::metacache_output_stream_closed()
} else {
err
}
}
async fn write_metacache_obj<W>(out: &mut MetacacheWriter<W>, obj: &MetaCacheEntry) -> Result<()>
where
W: AsyncWrite + Unpin,
{
out.write_obj(obj).await.map_err(metacache_write_error)
}
impl FileCacheReclaimReader {
fn new(inner: File, reclaim_offset: u64, reclaim_len: usize, reclaim_on_drop: bool) -> Self {
#[cfg(target_os = "macos")]
@@ -3005,11 +3021,14 @@ impl LocalDisk {
*objs_returned += 1;
}
out.write_obj(&MetaCacheEntry {
name: name.clone(),
metadata: metadata.to_vec(),
..Default::default()
})
write_metacache_obj(
out,
&MetaCacheEntry {
name: name.clone(),
metadata: metadata.to_vec(),
..Default::default()
},
)
.await?;
continue;
@@ -3065,10 +3084,13 @@ impl LocalDisk {
&& *last_name < name
{
let (pop, skip_object, dir_to_skip) = dir_stack.pop().expect("operation should succeed");
out.write_obj(&MetaCacheEntry {
name: pop.clone(),
..Default::default()
})
write_metacache_obj(
out,
&MetaCacheEntry {
name: pop.clone(),
..Default::default()
},
)
.await?;
let scan_path = pop.clone();
@@ -3076,15 +3098,17 @@ impl LocalDisk {
&& let Err(er) =
Box::pin(self.scan_dir(pop, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await
{
error!(
event = EVENT_DISK_LOCAL_SCAN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
path = %scan_path,
operation = "scan_dir",
error = ?er,
"Disk local scan failed"
);
if !er.is_metacache_output_stream_closed() {
error!(
event = EVENT_DISK_LOCAL_SCAN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
path = %scan_path,
operation = "scan_dir",
error = ?er,
"Disk local scan failed"
);
}
return Err(er);
}
}
@@ -3114,7 +3138,7 @@ impl LocalDisk {
meta.metadata = res.to_vec();
out.write_obj(&meta).await?;
write_metacache_obj(out, &meta).await?;
let file_meta = if opts.limit > 0 || opts.recursive {
FileMeta::load(&res).ok()
@@ -3183,10 +3207,13 @@ impl LocalDisk {
return Ok(());
}
out.write_obj(&MetaCacheEntry {
name: dir.clone(),
..Default::default()
})
write_metacache_obj(
out,
&MetaCacheEntry {
name: dir.clone(),
..Default::default()
},
)
.await?;
let scan_path = dir.clone();
@@ -3194,15 +3221,17 @@ impl LocalDisk {
&& let Err(er) =
Box::pin(self.scan_dir(dir, prefix.clone(), opts, out, objs_returned, skip_object, dir_to_skip)).await
{
error!(
event = EVENT_DISK_LOCAL_SCAN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
path = %scan_path,
operation = "scan_dir",
error = ?er,
"Disk local recursive scan failed"
);
if !er.is_metacache_output_stream_closed() {
error!(
event = EVENT_DISK_LOCAL_SCAN_FAILED,
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_DISK_LOCAL,
path = %scan_path,
operation = "scan_dir",
error = ?er,
"Disk local recursive scan failed"
);
}
return Err(er);
}
}
@@ -4201,7 +4230,7 @@ impl DiskAPI for LocalDisk {
metadata: data.to_vec(),
..Default::default()
};
out.write_obj(&meta).await?;
write_metacache_obj(&mut out, &meta).await?;
objs_returned += 1;
} else {
let fpath =
@@ -6284,6 +6313,49 @@ mod test {
assert!(!wait.await.expect("operation should succeed"));
}
#[test]
fn metacache_write_obj_classifies_closed_output_stream() {
struct BrokenPipeWriter;
impl AsyncWrite for BrokenPipeWriter {
fn poll_write(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
_buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
std::task::Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::BrokenPipe, "closed")))
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::task::Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::task::Poll::Ready(Ok(()))
}
}
let mut writer = BrokenPipeWriter;
let mut out = MetacacheWriter::new(&mut writer);
let err = futures::executor::block_on(write_metacache_obj(
&mut out,
&MetaCacheEntry {
name: "object".to_string(),
..Default::default()
},
))
.expect_err("closed metacache output stream should fail");
assert!(err.is_metacache_output_stream_closed());
}
#[tokio::test]
async fn test_scan_dir_includes_nested_object_dirs() {
use rustfs_filemeta::MetacacheReader;
rustfs/src/app/bucket_usecase.rs
+14 -7
View File
@@ -1766,14 +1766,21 @@ impl DefaultBucketUsecase {
let replication_configuration = match metadata_sys::get_replication_config(&bucket).await {
Ok((cfg, _created)) => cfg,
Err(StorageError::ConfigNotFound) => {
return Err(S3Error::with_message(
S3ErrorCode::ReplicationConfigurationNotFoundError,
"replication not found".to_string(),
));
}
Err(err) => {
error!("get_replication_config err {:?}", err);
if err == StorageError::ConfigNotFound {
return Err(S3Error::with_message(
S3ErrorCode::ReplicationConfigurationNotFoundError,
"replication not found".to_string(),
));
}
warn!(
component = LOG_COMPONENT_APP,
subsystem = LOG_SUBSYSTEM_BUCKET,
event = "bucket_replication_config_load_failed",
bucket = %bucket,
error = ?err,
"Failed to load bucket replication configuration"
);
return Err(ApiError::from(err).into());
}
};