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

feat(get): add v2 metrics compatibility harness (#3913)

Browse Source 
* feat(get): add v2 metrics compatibility harness

* feat(get): observe metadata fanout quorum (#3915)

* feat(get): observe metadata fanout quorum

* fix(app): use storage ECStore type in app boundary

----
Co-Authored-By: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-06-26 20:09:09 +08:00
committed by GitHub
parent a010dce93c
commit 5a78a9c416
5 changed files with 916 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/get_diagnostics.rs
+37
View File
@@ -24,15 +24,25 @@ pub(crate) const GET_OBJECT_PATH_REMOTE_TRANSITION: &str = "remote_transition";
pub(crate) const GET_STAGE_DECODE: &str = "decode";
pub(crate) const GET_STAGE_EMIT: &str = "emit";
pub(crate) const GET_STAGE_FILL: &str = "fill";
pub(crate) const GET_STAGE_FIRST_BYTE: &str = "first_byte";
pub(crate) const GET_STAGE_FIRST_METADATA_RESPONSE: &str = "first_metadata_response";
pub(crate) const GET_STAGE_FIRST_VALID_METADATA_RESPONSE: &str = "first_valid_metadata_response";
pub(crate) const GET_STAGE_FIRST_SHARD_READ: &str = "first_shard_read";
pub(crate) const GET_STAGE_FULL_BODY: &str = "full_body";
pub(crate) const GET_STAGE_METADATA: &str = "metadata";
pub(crate) const GET_STAGE_METADATA_FANOUT: &str = "metadata_fanout";
pub(crate) const GET_STAGE_OUTPUT_LOCK_WAIT: &str = "output_lock_wait";
pub(crate) const GET_STAGE_OUTPUT_POLL: &str = "output_poll";
pub(crate) const GET_STAGE_QUORUM_REACHED: &str = "quorum_reached";
pub(crate) const GET_STAGE_RANGE: &str = "range";
pub(crate) const GET_STAGE_READER_SETUP: &str = "reader_setup";
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";
pub(crate) const GET_STAGE_STRIPE_READ: &str = "stripe_read";
pub(crate) const GET_STAGE_STRIPE_READ_FIRST_SHARD: &str = "stripe_read_first_shard";
pub(crate) const GET_STAGE_STRIPE_READ_QUORUM: &str = "stripe_read_quorum";
pub(crate) const GET_STAGE_BITROT_VERIFY: &str = "bitrot_verify";
pub(crate) const GET_READER_BUFFER_OUTPUT: &str = "output";
pub(crate) const GET_READER_BUFFER_PREFETCH: &str = "prefetch";
@@ -51,6 +61,15 @@ pub(crate) const GET_SHARD_READ_OUTCOME_SUCCESS: &str = "success";
pub(crate) const GET_SHARD_ROLE_DATA: &str = "data";
pub(crate) const GET_SHARD_ROLE_PARITY: &str = "parity";
pub(crate) const GET_METADATA_RESPONSE_CORRUPT: &str = "corrupt";
pub(crate) const GET_METADATA_RESPONSE_DISK_NOT_FOUND: &str = "disk_not_found";
pub(crate) const GET_METADATA_RESPONSE_ERROR: &str = "error";
pub(crate) const GET_METADATA_RESPONSE_IGNORED: &str = "ignored";
pub(crate) const GET_METADATA_RESPONSE_NOT_FOUND: &str = "not_found";
pub(crate) const GET_METADATA_RESPONSE_TIMEOUT: &str = "timeout";
pub(crate) const GET_METADATA_RESPONSE_VALID: &str = "valid";
pub(crate) const GET_METADATA_RESPONSE_VERSION_NOT_FOUND: &str = "version_not_found";
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum GetObjectFailureReason {
BitrotMismatch,
@@ -195,19 +214,37 @@ mod tests {
assert_eq!(GET_STAGE_DECODE, "decode");
assert_eq!(GET_STAGE_EMIT, "emit");
assert_eq!(GET_STAGE_FILL, "fill");
assert_eq!(GET_STAGE_FIRST_BYTE, "first_byte");
assert_eq!(GET_STAGE_FIRST_METADATA_RESPONSE, "first_metadata_response");
assert_eq!(GET_STAGE_FIRST_VALID_METADATA_RESPONSE, "first_valid_metadata_response");
assert_eq!(GET_STAGE_FIRST_SHARD_READ, "first_shard_read");
assert_eq!(GET_STAGE_FULL_BODY, "full_body");
assert_eq!(GET_STAGE_METADATA, "metadata");
assert_eq!(GET_STAGE_METADATA_FANOUT, "metadata_fanout");
assert_eq!(GET_STAGE_OUTPUT_LOCK_WAIT, "output_lock_wait");
assert_eq!(GET_STAGE_OUTPUT_POLL, "output_poll");
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_RECONSTRUCT, "reconstruct");
assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff");
assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response");
assert_eq!(GET_STAGE_STRIPE_READ, "stripe_read");
assert_eq!(GET_STAGE_STRIPE_READ_FIRST_SHARD, "stripe_read_first_shard");
assert_eq!(GET_STAGE_STRIPE_READ_QUORUM, "stripe_read_quorum");
assert_eq!(GET_STAGE_BITROT_VERIFY, "bitrot_verify");
assert_eq!(GET_SHARD_READ_OUTCOME_ERROR, "error");
assert_eq!(GET_SHARD_READ_OUTCOME_MISSING, "missing");
assert_eq!(GET_SHARD_READ_OUTCOME_SUCCESS, "success");
assert_eq!(GET_SHARD_ROLE_DATA, "data");
assert_eq!(GET_SHARD_ROLE_PARITY, "parity");
assert_eq!(GET_METADATA_RESPONSE_CORRUPT, "corrupt");
assert_eq!(GET_METADATA_RESPONSE_DISK_NOT_FOUND, "disk_not_found");
assert_eq!(GET_METADATA_RESPONSE_ERROR, "error");
assert_eq!(GET_METADATA_RESPONSE_IGNORED, "ignored");
assert_eq!(GET_METADATA_RESPONSE_NOT_FOUND, "not_found");
assert_eq!(GET_METADATA_RESPONSE_TIMEOUT, "timeout");
assert_eq!(GET_METADATA_RESPONSE_VALID, "valid");
assert_eq!(GET_METADATA_RESPONSE_VERSION_NOT_FOUND, "version_not_found");
}
}
crates/ecstore/src/set_disk/read.rs
+376 -13
View File
@@ -14,8 +14,11 @@
use super::*;
use crate::get_diagnostics::{
GET_OBJECT_PATH_CODEC_STREAMING, GET_STAGE_DECODE, GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason,
classify_disk_error, record_get_object_pipeline_failure, record_get_object_pipeline_failure_for_path,
GET_METADATA_RESPONSE_CORRUPT, GET_METADATA_RESPONSE_DISK_NOT_FOUND, GET_METADATA_RESPONSE_ERROR,
GET_METADATA_RESPONSE_IGNORED, GET_METADATA_RESPONSE_NOT_FOUND, GET_METADATA_RESPONSE_TIMEOUT, GET_METADATA_RESPONSE_VALID,
GET_METADATA_RESPONSE_VERSION_NOT_FOUND, GET_OBJECT_PATH_CODEC_STREAMING, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_DECODE,
GET_STAGE_RANGE, GET_STAGE_READER_SETUP, GetObjectFailureReason, classify_disk_error, record_get_object_pipeline_failure,
record_get_object_pipeline_failure_for_path,
};
use metrics::counter;
use rustfs_config::{DEFAULT_OBJECT_ZERO_COPY_ENABLE, ENV_OBJECT_ZERO_COPY_ENABLE};
@@ -35,6 +38,152 @@ const READ_REPAIR_HEAL_DEDUP_MAX_ENTRIES: usize = 4096;
static READ_REPAIR_HEAL_CACHE: OnceLock<RwLock<HashMap<ReadRepairHealCacheKey, Instant>>> = OnceLock::new();
#[derive(Clone, Copy, Debug)]
struct MetadataFanoutObservation {
outcome: &'static str,
elapsed: Duration,
valid: bool,
ignored: bool,
}
impl MetadataFanoutObservation {
fn from_file_info(file_info: &FileInfo, elapsed: Duration) -> Self {
if file_info.is_valid() {
Self {
outcome: GET_METADATA_RESPONSE_VALID,
elapsed,
valid: true,
ignored: false,
}
} else {
Self {
outcome: GET_METADATA_RESPONSE_ERROR,
elapsed,
valid: false,
ignored: false,
}
}
}
fn from_error(err: &DiskError, elapsed: Duration) -> Self {
Self {
outcome: classify_metadata_response_error(err),
elapsed,
valid: false,
ignored: is_metadata_fanout_ignored_error(err),
}
}
}
#[derive(Clone, Debug, Default)]
struct MetadataFanoutDiagnostics {
fanout_duration: Duration,
observations: Vec<MetadataFanoutObservation>,
}
impl MetadataFanoutDiagnostics {
fn new(fanout_duration: Duration, observations: Vec<MetadataFanoutObservation>) -> Self {
Self {
fanout_duration,
observations,
}
}
fn total_responses(&self) -> usize {
self.observations.len()
}
fn valid_responses(&self) -> usize {
self.observations.iter().filter(|observation| observation.valid).count()
}
fn ignored_responses(&self) -> usize {
self.observations.iter().filter(|observation| observation.ignored).count()
}
fn error_responses(&self) -> usize {
self.total_responses().saturating_sub(self.valid_responses())
}
fn first_response_latency(&self) -> Option<Duration> {
self.observations.iter().map(|observation| observation.elapsed).min()
}
fn first_valid_response_latency(&self) -> Option<Duration> {
self.observations
.iter()
.filter(|observation| observation.valid)
.map(|observation| observation.elapsed)
.min()
}
fn slowest_response_latency(&self) -> Option<Duration> {
self.observations.iter().map(|observation| observation.elapsed).max()
}
fn quorum_candidate_latency(&self, read_quorum: usize) -> Option<Duration> {
if read_quorum == 0 {
return Some(Duration::ZERO);
}
let mut valid_latencies = self
.observations
.iter()
.filter(|observation| observation.valid)
.map(|observation| observation.elapsed)
.collect::<Vec<_>>();
valid_latencies.sort_unstable();
valid_latencies.get(read_quorum.saturating_sub(1)).copied()
}
fn record(&self, path: &'static str) {
rustfs_io_metrics::record_get_object_metadata_fanout_duration(path, self.fanout_duration.as_secs_f64());
if let Some(latency) = self.first_response_latency() {
rustfs_io_metrics::record_get_object_first_metadata_response_latency(path, latency.as_secs_f64());
}
if let Some(latency) = self.first_valid_response_latency() {
rustfs_io_metrics::record_get_object_first_valid_metadata_response_latency(path, latency.as_secs_f64());
}
if let Some(latency) = self.slowest_response_latency() {
rustfs_io_metrics::record_get_object_slowest_metadata_response_latency(path, latency.as_secs_f64());
}
rustfs_io_metrics::record_get_object_metadata_fanout_shape(
path,
self.total_responses(),
self.valid_responses(),
self.ignored_responses(),
self.error_responses(),
);
for observation in &self.observations {
rustfs_io_metrics::record_get_object_metadata_response(path, observation.outcome);
}
}
fn record_quorum_candidate_latency(&self, path: &'static str, read_quorum: usize) {
if let Some(latency) = self.quorum_candidate_latency(read_quorum) {
rustfs_io_metrics::record_get_object_quorum_reached_latency(path, latency.as_secs_f64());
}
}
}
fn classify_metadata_response_error(err: &DiskError) -> &'static str {
match err {
DiskError::FileNotFound | DiskError::VolumeNotFound => GET_METADATA_RESPONSE_NOT_FOUND,
DiskError::FileVersionNotFound => GET_METADATA_RESPONSE_VERSION_NOT_FOUND,
DiskError::DiskNotFound => GET_METADATA_RESPONSE_DISK_NOT_FOUND,
DiskError::FileCorrupt | DiskError::CorruptedFormat | DiskError::CorruptedBackend | DiskError::OutdatedXLMeta => {
GET_METADATA_RESPONSE_CORRUPT
}
DiskError::Timeout => GET_METADATA_RESPONSE_TIMEOUT,
DiskError::FaultyDisk | DiskError::FaultyRemoteDisk => GET_METADATA_RESPONSE_IGNORED,
_ => GET_METADATA_RESPONSE_ERROR,
}
}
fn is_metadata_fanout_ignored_error(err: &DiskError) -> bool {
OBJECT_OP_IGNORED_ERRS.iter().any(|ignored| ignored == err)
}
#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct ReadRepairHealCacheKey {
bucket: String,
@@ -522,8 +671,62 @@ impl SetDisks {
healing: bool,
incl_free_versions: bool,
) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>)> {
let (ress, errors, _) = Self::read_all_fileinfo_inner(
disks,
org_bucket,
bucket,
object,
version_id,
read_data,
healing,
incl_free_versions,
false,
)
.await?;
Ok((ress, errors))
}
#[allow(clippy::too_many_arguments)]
async fn read_all_fileinfo_observed(
disks: &[Option<DiskStore>],
org_bucket: &str,
bucket: &str,
object: &str,
version_id: &str,
read_data: bool,
healing: bool,
incl_free_versions: bool,
) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>, MetadataFanoutDiagnostics)> {
Self::read_all_fileinfo_inner(
disks,
org_bucket,
bucket,
object,
version_id,
read_data,
healing,
incl_free_versions,
true,
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn read_all_fileinfo_inner(
disks: &[Option<DiskStore>],
org_bucket: &str,
bucket: &str,
object: &str,
version_id: &str,
read_data: bool,
healing: bool,
incl_free_versions: bool,
observe: bool,
) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>, MetadataFanoutDiagnostics)> {
let fanout_start = observe.then(Instant::now);
let mut ress = Vec::with_capacity(disks.len());
let mut errors = Vec::with_capacity(disks.len());
let mut observations = observe.then(|| Vec::with_capacity(disks.len()));
let opts = Arc::new(ReadOptions {
incl_free_versions,
read_data,
@@ -541,11 +744,14 @@ impl SetDisks {
let object = object.clone();
let version_id = version_id.clone();
tokio::spawn(async move {
if let Some(disk) = disk {
let response_start = observe.then(Instant::now);
let result = if let Some(disk) = disk {
disk.read_version(&org_bucket, &bucket, &object, &version_id, &opts).await
} else {
Err(DiskError::DiskNotFound)
}
};
let elapsed = response_start.map(|start| start.elapsed());
(result, elapsed)
})
});
@@ -554,23 +760,37 @@ impl SetDisks {
for result in results {
match result {
Ok(res) => match res {
Ok((res, elapsed)) => match res {
Ok(file_info) => {
if let (Some(observations), Some(elapsed)) = (&mut observations, elapsed) {
observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
}
ress.push(file_info);
errors.push(None);
}
Err(e) => {
if let (Some(observations), Some(elapsed)) = (&mut observations, elapsed) {
observations.push(MetadataFanoutObservation::from_error(&e, elapsed));
}
ress.push(FileInfo::default());
errors.push(Some(e));
}
},
Err(_) => {
let err = DiskError::Unexpected;
if let (Some(observations), Some(fanout_start)) = (&mut observations, fanout_start) {
observations.push(MetadataFanoutObservation::from_error(&err, fanout_start.elapsed()));
}
ress.push(FileInfo::default());
errors.push(Some(DiskError::Unexpected));
errors.push(Some(err));
}
}
}
Ok((ress, errors))
let diagnostics = match (fanout_start, observations) {
(Some(fanout_start), Some(observations)) => MetadataFanoutDiagnostics::new(fanout_start.elapsed(), observations),
_ => MetadataFanoutDiagnostics::default(),
};
Ok((ress, errors, diagnostics))
}
pub async fn read_version_optimized(
@@ -912,8 +1132,10 @@ impl SetDisks {
let vid = opts.version_id.clone().unwrap_or_default();
// TODO: optimize concurrency and break once enough slots are available
let (parts_metadata, errs) =
Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false, opts.incl_free_versions).await?;
let (parts_metadata, errs, metadata_fanout_diagnostics) =
Self::read_all_fileinfo_observed(&disks, "", bucket, object, vid.as_str(), read_data, false, opts.incl_free_versions)
.await?;
metadata_fanout_diagnostics.record(GET_OBJECT_PATH_LEGACY_DUPLEX);
// warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
// warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
@@ -928,15 +1150,18 @@ impl SetDisks {
return Err(e);
}
};
let read_quorum =
usize::try_from(read_quorum).map_err(|_| to_object_err(DiskError::ErasureReadQuorum.into(), vec![bucket, object]))?;
metadata_fanout_diagnostics.record_quorum_candidate_latency(GET_OBJECT_PATH_LEGACY_DUPLEX, read_quorum);
if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum as usize) {
if let Some(err) = reduce_read_quorum_errs(&errs, OBJECT_OP_IGNORED_ERRS, read_quorum) {
error!("reduce_read_quorum_errs: {:?}, bucket: {}, object: {}", &err, bucket, object);
return Err(to_object_err(err.into(), vec![bucket, object]));
}
let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum as usize);
let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum);
let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum as usize)?;
let fi = Self::pick_valid_fileinfo(&parts_metadata, mot_time, etag, read_quorum)?;
if errs.iter().any(|err| err.is_some()) {
let version_id = resolved_read_repair_version_id(&fi, opts.version_id.as_deref());
submit_read_repair_heal(
@@ -950,7 +1175,7 @@ impl SetDisks {
)
.await;
} else if use_metadata_cache {
self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum as usize)
self.cache_get_object_fileinfo(bucket, object, &fi, &parts_metadata, &op_online_disks, read_quorum)
.await;
}
// debug!("get_object_fileinfo pick fi {:?}", &fi);
@@ -1861,6 +2086,16 @@ mod metadata_cache_tests {
mod tests {
use super::*;
fn metadata_fanout_test_fileinfo(object: &str) -> FileInfo {
let mut fi = FileInfo::new(object, 2, 2);
fi.volume = "bucket".to_string();
fi.name = object.to_string();
fi.size = 1;
fi.erasure.index = 1;
fi.metadata.insert("etag".to_string(), "etag-1".to_string());
fi
}
fn codec_streaming_test_fileinfo(size: i64, part_count: usize) -> FileInfo {
let mut fi = FileInfo::new("object", 4, 2);
fi.volume = "bucket".to_string();
@@ -1888,6 +2123,134 @@ mod tests {
fi
}
#[test]
fn metadata_fanout_diagnostics_classifies_response_outcomes() {
let valid = metadata_fanout_test_fileinfo("object");
let invalid = FileInfo::default();
let observations = vec![
MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(3)),
MetadataFanoutObservation::from_file_info(&invalid, Duration::from_millis(4)),
MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(5)),
MetadataFanoutObservation::from_error(&DiskError::FileVersionNotFound, Duration::from_millis(6)),
MetadataFanoutObservation::from_error(&DiskError::DiskNotFound, Duration::from_millis(7)),
MetadataFanoutObservation::from_error(&DiskError::FileCorrupt, Duration::from_millis(8)),
MetadataFanoutObservation::from_error(&DiskError::Timeout, Duration::from_millis(9)),
MetadataFanoutObservation::from_error(&DiskError::FaultyDisk, Duration::from_millis(10)),
MetadataFanoutObservation::from_error(&DiskError::Unexpected, Duration::from_millis(11)),
];
let diagnostics = MetadataFanoutDiagnostics::new(Duration::from_millis(12), observations);
let outcomes = diagnostics
.observations
.iter()
.map(|observation| observation.outcome)
.collect::<Vec<_>>();
assert_eq!(
outcomes,
vec![
GET_METADATA_RESPONSE_VALID,
GET_METADATA_RESPONSE_ERROR,
GET_METADATA_RESPONSE_NOT_FOUND,
GET_METADATA_RESPONSE_VERSION_NOT_FOUND,
GET_METADATA_RESPONSE_DISK_NOT_FOUND,
GET_METADATA_RESPONSE_CORRUPT,
GET_METADATA_RESPONSE_TIMEOUT,
GET_METADATA_RESPONSE_IGNORED,
GET_METADATA_RESPONSE_ERROR,
]
);
assert_eq!(diagnostics.total_responses(), 9);
assert_eq!(diagnostics.valid_responses(), 1);
assert_eq!(diagnostics.error_responses(), 8);
}
#[test]
fn metadata_fanout_diagnostics_tracks_ignored_errors_without_hiding_outcomes() {
let diagnostics = MetadataFanoutDiagnostics::new(
Duration::from_millis(15),
vec![
MetadataFanoutObservation::from_error(&DiskError::DiskNotFound, Duration::from_millis(1)),
MetadataFanoutObservation::from_error(&DiskError::FaultyRemoteDisk, Duration::from_millis(2)),
MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(3)),
],
);
assert_eq!(diagnostics.ignored_responses(), 2);
assert_eq!(diagnostics.error_responses(), 3);
assert_eq!(diagnostics.observations[0].outcome, GET_METADATA_RESPONSE_DISK_NOT_FOUND);
assert_eq!(diagnostics.observations[1].outcome, GET_METADATA_RESPONSE_IGNORED);
assert_eq!(diagnostics.observations[2].outcome, GET_METADATA_RESPONSE_NOT_FOUND);
}
#[test]
fn metadata_fanout_quorum_candidate_latency_ignores_slow_trailing_valid_response() {
let valid = metadata_fanout_test_fileinfo("object");
let diagnostics = MetadataFanoutDiagnostics::new(
Duration::from_millis(250),
vec![
MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(2)),
MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(5)),
MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(7)),
MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(250)),
],
);
assert_eq!(diagnostics.first_response_latency(), Some(Duration::from_millis(2)));
assert_eq!(diagnostics.first_valid_response_latency(), Some(Duration::from_millis(5)));
assert_eq!(diagnostics.slowest_response_latency(), Some(Duration::from_millis(250)));
assert_eq!(diagnostics.quorum_candidate_latency(2), Some(Duration::from_millis(7)));
}
#[test]
fn metadata_fanout_quorum_candidate_latency_requires_enough_valid_responses() {
let valid = metadata_fanout_test_fileinfo("object");
let diagnostics = MetadataFanoutDiagnostics::new(
Duration::from_millis(8),
vec![
MetadataFanoutObservation::from_file_info(&valid, Duration::from_millis(5)),
MetadataFanoutObservation::from_error(&DiskError::FileNotFound, Duration::from_millis(6)),
],
);
assert_eq!(diagnostics.quorum_candidate_latency(0), Some(Duration::ZERO));
assert_eq!(diagnostics.quorum_candidate_latency(1), Some(Duration::from_millis(5)));
assert_eq!(diagnostics.quorum_candidate_latency(2), None);
}
#[test]
fn metadata_fanout_counts_delete_marker_and_conflicting_versions_as_valid_observations_only() {
let mut latest = metadata_fanout_test_fileinfo("object");
latest.version_id = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000001").expect("static uuid should parse"));
let mut historical = metadata_fanout_test_fileinfo("object");
historical.version_id = Some(Uuid::parse_str("00000000-0000-0000-0000-000000000002").expect("static uuid should parse"));
let mut delete_marker = metadata_fanout_test_fileinfo("object");
delete_marker.deleted = true;
delete_marker.version_id =
Some(Uuid::parse_str("00000000-0000-0000-0000-000000000003").expect("static uuid should parse"));
let diagnostics = MetadataFanoutDiagnostics::new(
Duration::from_millis(9),
vec![
MetadataFanoutObservation::from_file_info(&latest, Duration::from_millis(3)),
MetadataFanoutObservation::from_file_info(&historical, Duration::from_millis(4)),
MetadataFanoutObservation::from_file_info(&delete_marker, Duration::from_millis(5)),
],
);
assert_eq!(diagnostics.total_responses(), 3);
assert_eq!(diagnostics.valid_responses(), 3);
assert_eq!(diagnostics.error_responses(), 0);
assert!(
diagnostics
.observations
.iter()
.all(|observation| observation.outcome == GET_METADATA_RESPONSE_VALID)
);
}
fn codec_streaming_test_object_info(fi: &FileInfo) -> ObjectInfo {
ObjectInfo::from_file_info(fi, "bucket", "object", false)
}
crates/io-metrics/src/lib.rs
+154
View File
@@ -330,6 +330,7 @@ pub fn record_get_object_response_handoff(
"buffer_source" => buffer_source.to_string()
)
.record(duration_secs);
record_get_object_response_handoff_duration("s3_handler", duration_secs);
}
/// Record I/O queue congestion observation.
@@ -374,6 +375,109 @@ pub fn record_get_object_stage_duration(path: &'static str, stage: &'static str,
histogram!("rustfs_io_get_object_stage_duration_seconds", "path" => path, "stage" => stage).record(duration_secs);
}
/// Record GetObject metadata fanout duration.
#[inline(always)]
pub fn record_get_object_metadata_fanout_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "metadata_fanout", duration_secs);
}
/// Record latency until the first metadata response arrives.
#[inline(always)]
pub fn record_get_object_first_metadata_response_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "first_metadata_response", duration_secs);
}
/// Record latency until the first valid metadata response arrives.
#[inline(always)]
pub fn record_get_object_first_valid_metadata_response_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "first_valid_metadata_response", duration_secs);
}
/// Record latency of the slowest metadata response in a fanout.
#[inline(always)]
pub fn record_get_object_slowest_metadata_response_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "slowest_metadata_response", duration_secs);
}
/// Record latency until metadata quorum is reached.
#[inline(always)]
pub fn record_get_object_quorum_reached_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "quorum_reached", duration_secs);
}
/// Record one bounded metadata response outcome.
#[inline(always)]
pub fn record_get_object_metadata_response(path: &'static str, outcome: &'static str) {
if !get_stage_metrics_enabled() {
return;
}
counter!("rustfs_io_get_object_metadata_response_total", "path" => path, "outcome" => outcome).increment(1);
}
/// Record aggregate metadata fanout shape for one GetObject metadata read.
#[inline(always)]
pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize, valid: usize, ignored: usize, errors: usize) {
if !get_stage_metrics_enabled() {
return;
}
histogram!("rustfs_io_get_object_metadata_fanout_total_responses", "path" => path)
.record(metadata_fanout_count_to_f64(total));
histogram!("rustfs_io_get_object_metadata_fanout_valid_responses", "path" => path)
.record(metadata_fanout_count_to_f64(valid));
histogram!("rustfs_io_get_object_metadata_fanout_ignored_responses", "path" => path)
.record(metadata_fanout_count_to_f64(ignored));
histogram!("rustfs_io_get_object_metadata_fanout_error_responses", "path" => path)
.record(metadata_fanout_count_to_f64(errors));
}
/// Record GetObject reader setup duration.
#[inline(always)]
pub fn record_get_object_reader_setup_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "reader_setup", duration_secs);
}
/// Record latency until the first shard read completes.
#[inline(always)]
pub fn record_get_object_first_shard_read_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "first_shard_read", duration_secs);
}
/// Record GetObject bitrot verification duration.
#[inline(always)]
pub fn record_get_object_bitrot_verify_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "bitrot_verify", duration_secs);
}
/// Record GetObject reconstruct duration.
#[inline(always)]
pub fn record_get_object_reconstruct_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "reconstruct", duration_secs);
}
/// Record GetObject emit duration.
#[inline(always)]
pub fn record_get_object_emit_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "emit", duration_secs);
}
/// Record GetObject first-byte latency as observed by the caller that owns that boundary.
#[inline(always)]
pub fn record_get_object_first_byte_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "first_byte", duration_secs);
}
/// Record GetObject full-body latency as observed by the caller that owns that boundary.
#[inline(always)]
pub fn record_get_object_full_body_latency(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "full_body", duration_secs);
}
/// Record GetObject response handoff duration in the shared stage histogram.
#[inline(always)]
pub fn record_get_object_response_handoff_duration(path: &'static str, duration_secs: f64) {
record_get_object_stage_duration(path, "response_handoff", duration_secs);
}
/// Record the selected GetObject reader path.
#[inline(always)]
pub fn record_get_object_reader_path(path: &'static str) {
@@ -509,6 +613,11 @@ fn shard_read_fanout_to_f64(value: usize) -> f64 {
u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX))
}
#[inline(always)]
fn metadata_fanout_count_to_f64(value: usize) -> f64 {
u32::try_from(value).map(f64::from).unwrap_or(f64::from(u32::MAX))
}
/// Record per-stripe shard-read fanout shape for GetObject read-path attribution.
#[inline(always)]
pub fn record_get_object_shard_read_fanout(
@@ -1177,6 +1286,21 @@ mod tests {
record_get_object_reader_prefetch("codec_streaming", "stored");
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_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);
record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
record_get_object_metadata_response("legacy_duplex", "valid");
record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
record_get_object_reader_setup_duration("legacy_duplex", 0.003);
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
record_get_object_response_handoff_duration("s3_handler", 0.0001);
record_get_object_metadata_phase_duration(0.002);
record_get_object_shard_reader_setup_duration(0.003);
record_get_object_decode_duration(0.004);
@@ -1237,6 +1361,21 @@ mod tests {
record_get_object_reader_prefetch("codec_streaming", "stored");
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_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);
record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
record_get_object_metadata_response("legacy_duplex", "valid");
record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
record_get_object_reader_setup_duration("legacy_duplex", 0.003);
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
record_get_object_response_handoff_duration("s3_handler", 0.0001);
record_get_object_shard_reader_setup_duration(0.003);
record_get_object_decode_duration(0.004);
record_get_object_duplex_backpressure_duration(0.005);
@@ -1260,6 +1399,21 @@ mod tests {
record_get_object_reader_prefetch("codec_streaming", "stored");
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_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);
record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
record_get_object_metadata_response("legacy_duplex", "valid");
record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
record_get_object_reader_setup_duration("legacy_duplex", 0.003);
record_get_object_first_shard_read_duration("codec_streaming", 0.004);
record_get_object_bitrot_verify_duration("codec_streaming", 0.005);
record_get_object_reconstruct_duration("codec_streaming", 0.006);
record_get_object_emit_duration("codec_streaming", 0.007);
record_get_object_first_byte_latency("s3_handler", 0.008);
record_get_object_full_body_latency("s3_handler", 0.009);
record_get_object_response_handoff_duration("s3_handler", 0.0001);
record_get_object_shard_reader_setup_duration(0.003);
record_get_object_decode_duration(0.004);
record_get_object_duplex_backpressure_duration(0.005);