gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-16 18:08:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ecstore): expose read version stage metrics (#6073)

Browse Source 
Record local read_version path resolution, path length check, xl.meta read, and metadata decode durations through the existing GET stage metrics channel. The new samples are gated by GET stage metrics so metrics-off reads avoid timer and recorder work.

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-14 00:33:46 +08:00
committed by GitHub
parent 9b66040a02
commit 1ac28d6459
3 changed files with 157 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/diagnostics/get.rs
+8
View File
@@ -84,6 +84,10 @@ pub(crate) const GET_STAGE_READER_STREAM_FIRST_READ: &str = "reader_stream_first
pub(crate) const GET_STAGE_READER_TASK_BITROT_READER_INIT: &str = "reader_task_bitrot_reader_init";
pub(crate) const GET_STAGE_READER_TASK_FILE_OPEN: &str = "reader_task_file_open";
pub(crate) const GET_STAGE_READER_TASK_READER_CONSTRUCTION: &str = "reader_task_reader_construction";
pub(crate) const GET_STAGE_READ_VERSION_DECODE: &str = "read_version_decode";
pub(crate) const GET_STAGE_READ_VERSION_PATH_CHECK: &str = "read_version_path_check";
pub(crate) const GET_STAGE_READ_VERSION_PATH_RESOLVE: &str = "read_version_path_resolve";
pub(crate) const GET_STAGE_READ_VERSION_XLMETA_READ: &str = "read_version_xlmeta_read";
pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct";
pub(crate) const GET_STAGE_RESPONSE_HANDOFF: &str = "response_handoff";
pub(crate) const GET_STAGE_SLOWEST_METADATA_RESPONSE: &str = "slowest_metadata_response";
@@ -442,6 +446,10 @@ mod tests {
assert_eq!(GET_STAGE_QUORUM_REACHED, "quorum_reached");
assert_eq!(GET_STAGE_RANGE, "range");
assert_eq!(GET_STAGE_READER_SETUP, "reader_setup");
assert_eq!(GET_STAGE_READ_VERSION_DECODE, "read_version_decode");
assert_eq!(GET_STAGE_READ_VERSION_PATH_CHECK, "read_version_path_check");
assert_eq!(GET_STAGE_READ_VERSION_PATH_RESOLVE, "read_version_path_resolve");
assert_eq!(GET_STAGE_READ_VERSION_XLMETA_READ, "read_version_xlmeta_read");
assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct");
assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff");
assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response");
crates/ecstore/src/disk/local.rs
+145 -23
View File
@@ -15,6 +15,11 @@
use crate::config::storageclass::DEFAULT_INLINE_BLOCK;
use crate::crash_inject::{self, CrashPoint};
use crate::data_usage::local_snapshot::ensure_data_usage_layout;
use crate::diagnostics::get::{
GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_READ_VERSION_DECODE,
GET_STAGE_READ_VERSION_PATH_CHECK, GET_STAGE_READ_VERSION_PATH_RESOLVE, GET_STAGE_READ_VERSION_XLMETA_READ,
get_stage_timer_if_enabled, record_get_stage_duration_if_enabled,
};
#[cfg(test)]
use crate::disk::HEALING_MARKER_PATH;
use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
@@ -9840,6 +9845,12 @@ impl DiskAPI for LocalDisk {
opts: &ReadOptions,
) -> Result<FileInfo> {
crate::hp_guard!("LocalDisk::read_version");
let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
let metrics_path = if stage_metrics_enabled && crate::bucket::utils::is_meta_bucketname(volume) {
GET_OBJECT_PATH_INTERNAL_META
} else {
GET_OBJECT_PATH_LEGACY_DUPLEX
};
if !org_volume.is_empty() {
let org_volume_path = self.io_get_bucket_path(org_volume)?;
if !skip_access_checks(org_volume) {
@@ -9849,37 +9860,46 @@ impl DiskAPI for LocalDisk {
}
}
let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let file_path = self.io_get_object_path(volume, path)?;
let volume_dir = self.io_get_bucket_path(volume)?;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start);
let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled);
check_path_length(file_path.to_string_lossy().as_ref())?;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_CHECK, path_check_start);
let read_data = opts.read_data;
let (data, _) = self
.read_raw(volume, volume_dir.clone(), file_path, read_data)
.await
.map_err(|e| {
if e == DiskError::FileNotFound && !version_id.is_empty() {
DiskError::FileVersionNotFound
} else {
e
}
})?;
let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await;
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start);
let (data, _) = raw_read_result.map_err(|e| {
if e == DiskError::FileNotFound && !version_id.is_empty() {
DiskError::FileVersionNotFound
} else {
e
}
})?;
let mut fi = get_file_info(
&data,
volume,
path,
version_id,
FileInfoOpts {
data: read_data,
include_free_versions: opts.incl_free_versions,
include_part_checksums: false,
},
)?;
fi.validate_for_metadata_read()?;
let decode_start = get_stage_timer_if_enabled(stage_metrics_enabled);
let file_info_result: Result<FileInfo> = (|| {
let fi = get_file_info(
&data,
volume,
path,
version_id,
FileInfoOpts {
data: read_data,
include_free_versions: opts.incl_free_versions,
include_part_checksums: false,
},
)?;
fi.validate_for_metadata_read()?;
Ok(fi)
})();
record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_DECODE, decode_start);
let mut fi = file_info_result?;
if fi.is_canonical_delete_marker() {
return Ok(fi);
}
@@ -10562,6 +10582,108 @@ mod test {
meta.marshal_msg().expect("test metadata should encode")
}
#[test]
#[serial_test::serial]
fn read_version_records_local_metadata_stage_breakdown() {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("test runtime should be created");
let recorder = crate::test_metrics::CapturingRecorder::default();
let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
rustfs_io_metrics::set_get_stage_metrics_enabled(true);
metrics::with_local_recorder(&recorder, || {
runtime.block_on(async {
let dir = tempfile::tempdir().expect("temp dir should be created");
let endpoint =
Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
let bucket = "bucket";
let object = "stage-breakdown";
ensure_test_volume(&disk, bucket).await;
let object_dir = dir.path().join(bucket).join(object);
fs::create_dir_all(&object_dir)
.await
.expect("object directory should be created");
fs::write(
object_dir.join(STORAGE_FORMAT_FILE),
test_meta(test_file_info(object, Uuid::new_v4(), None, Some(Bytes::from_static(b"inline")))),
)
.await
.expect("object metadata should be written");
disk.read_version(
"",
bucket,
object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("read_version should succeed");
let meta_object = "stage-breakdown-meta";
let meta_object_dir = dir.path().join(RUSTFS_META_BUCKET).join(meta_object);
fs::create_dir_all(&meta_object_dir)
.await
.expect("internal metadata object directory should be created");
fs::write(
meta_object_dir.join(STORAGE_FORMAT_FILE),
test_meta(test_file_info(meta_object, Uuid::new_v4(), None, Some(Bytes::from_static(b"meta")))),
)
.await
.expect("internal metadata should be written");
disk.read_version(
"",
RUSTFS_META_BUCKET,
meta_object,
"",
&ReadOptions {
read_data: true,
..Default::default()
},
)
.await
.expect("internal metadata read_version should succeed");
});
});
rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
for stage in [
GET_STAGE_READ_VERSION_PATH_RESOLVE,
GET_STAGE_READ_VERSION_PATH_CHECK,
GET_STAGE_READ_VERSION_XLMETA_READ,
GET_STAGE_READ_VERSION_DECODE,
] {
assert_eq!(
recorder
.histogram_values(
"rustfs_io_get_object_stage_duration_seconds",
&[("path", GET_OBJECT_PATH_LEGACY_DUPLEX), ("stage", stage)]
)
.len(),
1,
"{stage} should be recorded once for user-bucket LocalDisk::read_version"
);
assert_eq!(
recorder
.histogram_values(
"rustfs_io_get_object_stage_duration_seconds",
&[("path", GET_OBJECT_PATH_INTERNAL_META), ("stage", stage)]
)
.len(),
1,
"{stage} should be recorded once for internal-meta LocalDisk::read_version"
);
}
}
#[test]
fn inline_metadata_rollback_dir_avoids_real_data_dir_collision() {
let target_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("version id should parse");
crates/io-metrics/src/lib.rs
+4
View File
@@ -2703,6 +2703,10 @@ mod tests {
record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
record_get_object_metadata_fanout_duration("legacy_duplex", 0.001);
record_get_object_stage_duration("legacy_duplex", "read_version_path_resolve", 0.0001);
record_get_object_stage_duration("legacy_duplex", "read_version_path_check", 0.0001);
record_get_object_stage_duration("legacy_duplex", "read_version_xlmeta_read", 0.0005);
record_get_object_stage_duration("legacy_duplex", "read_version_decode", 0.0002);
record_get_object_first_metadata_response_latency("legacy_duplex", 0.001);
record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001);
record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003);