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

refactor(heal): prune statistics label helpers (#6312)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-08-20 22:43:59 +08:00
committed by GitHub
parent 830e553a3c
commit 114bf5148c
5 changed files with 60 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/set_disk/mod.rs
+2 -2
View File
@@ -5717,8 +5717,8 @@ mod tests {
.filter(|(composite, _, _, _)| {
composite.key().name() == "rustfs_s3_put_object_stage_duration_ms"
&& composite.key().labels().any(|label| {
label.key().to_string() == "stage"
&& label.value().to_string() == rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_WAIT
label.key() == "stage"
&& label.value() == rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_WAIT
})
})
.map(|(_, _, _, value)| match value {
crates/heal/src/heal/manager/scheduler.rs
+4 -22
View File
@@ -567,35 +567,17 @@ impl HealManager {
pub(super) fn heal_request_set_key(request: &HealRequest) -> Option<String> {
match &request.heal_type {
HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),
HealType::Object { .. } => heal_options_set_key(&request.options),
_ => None,
}
}
pub(super) fn heal_options_set_key(options: &HealOptions) -> Option<String> {
match (options.pool_index, options.set_index) {
(Some(pool), Some(set)) => Some(format!("pool_{pool}_set_{set}")),
HealType::Object { .. } => request.options.set_key(),
_ => None,
}
}
pub(super) fn heal_request_type_label(request: &HealRequest) -> &'static str {
match &request.heal_type {
HealType::Cluster => "cluster",
HealType::Object { .. } => "object",
HealType::Bucket { .. } => "bucket",
HealType::Prefix { .. } => "prefix",
HealType::ErasureSet { .. } => "erasure_set",
HealType::Metadata { .. } => "metadata",
HealType::ECDecode { .. } => "ec_decode",
}
request.heal_type.kind_label()
}
pub(super) fn heal_request_set_metric_label(request: &HealRequest) -> String {
heal_request_set_key(request).unwrap_or_else(|| match (request.options.pool_index, request.options.set_index) {
(Some(pool), Some(set)) => format!("pool_{pool}_set_{set}"),
_ => "global".to_string(),
})
heal_request_set_key(request).unwrap_or_else(|| request.options.set_metric_label())
}
pub(super) fn record_scheduler_skip(set_label: &str) {
@@ -673,7 +655,7 @@ fn emit_mrf_repaired_events(targets: Vec<MrfRepairNoticeTarget>) {
pub(super) fn heal_request_set_key_for_task(task: &HealTask) -> Option<String> {
match &task.heal_type {
HealType::ErasureSet { set_disk_id, .. } => Some(set_disk_id.clone()),
HealType::Object { .. } => heal_options_set_key(&task.options),
HealType::Object { .. } => task.options.set_key(),
_ => None,
}
}
crates/heal/src/heal/manager/tests.rs
+31 -12
View File
@@ -744,10 +744,8 @@ fn test_priority_queue_pop_runnable_skips_blocked_erasure_set() {
let mut running = HashMap::new();
running.insert("pool_0_set_1".to_string(), 1);
let (popped, skipped_sets) = queue.pop_runnable_with_skips(
|request| can_schedule_request(request, &running, 1),
|request| heal_request_set_key(request),
);
let (popped, skipped_sets) =
queue.pop_runnable_with_skips(|request| can_schedule_request(request, &running, 1), heal_request_set_key);
let popped = popped.expect("should find runnable request");
assert_eq!(skipped_sets, vec!["pool_0_set_1".to_string()]);
@@ -788,10 +786,8 @@ fn test_priority_queue_pop_runnable_restores_all_blocked_items() {
running.insert("pool_0_set_2".to_string(), 1);
running.insert("pool_0_set_3".to_string(), 1);
let (popped, skipped_sets) = queue.pop_runnable_with_skips(
|request| can_schedule_request(request, &running, 1),
|request| heal_request_set_key(request),
);
let (popped, skipped_sets) =
queue.pop_runnable_with_skips(|request| can_schedule_request(request, &running, 1), heal_request_set_key);
assert!(popped.is_none());
assert_eq!(
@@ -843,10 +839,8 @@ fn test_priority_queue_pop_runnable_restores_deferred_with_tail() {
running.insert("pool_0_set_1".to_string(), 1);
running.insert("pool_0_set_2".to_string(), 1);
let (popped, skipped_sets) = queue.pop_runnable_with_skips(
|request| can_schedule_request(request, &running, 1),
|request| heal_request_set_key(request),
);
let (popped, skipped_sets) =
queue.pop_runnable_with_skips(|request| can_schedule_request(request, &running, 1), heal_request_set_key);
assert_eq!(skipped_sets, vec!["pool_0_set_1".to_string(), "pool_0_set_2".to_string()]);
assert!(matches!(
@@ -904,6 +898,31 @@ fn test_can_schedule_scoped_object_request_respects_per_set_limit() {
assert!(can_schedule_request(&request, &running, 2));
}
#[test]
fn test_heal_request_and_task_metric_labels_match() {
let request = HealRequest::new(
HealType::Object {
bucket: "bucket".to_string(),
object: "object".to_string(),
version_id: None,
},
HealOptions {
pool_index: Some(0),
set_index: Some(1),
..Default::default()
},
HealPriority::Normal,
);
assert_eq!(heal_request_type_label(&request), "object");
assert_eq!(heal_request_set_key(&request), Some("pool_0_set_1".to_string()));
assert_eq!(heal_request_set_metric_label(&request), "pool_0_set_1");
let task = HealTask::from_request(request, Arc::new(MockStorage));
assert_eq!(task.metric_type_label(), "object");
assert_eq!(task.metric_set_label(), "pool_0_set_1");
}
#[tokio::test]
async fn test_submit_heal_request_returns_merged_for_duplicate() {
let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
crates/heal/src/heal/progress.rs
-43
View File
@@ -218,15 +218,6 @@ impl HealStatistics {
self.total_bytes_healed += bytes;
self.last_update_time = SystemTime::now();
}
pub fn get_success_rate(&self) -> f64 {
let total = self.successful_tasks + self.failed_tasks;
if total > 0 {
(self.successful_tasks as f64 / total as f64) * 100.0
} else {
0.0
}
}
}
#[cfg(test)]
@@ -539,38 +530,4 @@ mod tests {
assert_eq!(stats.total_objects_healed, 8);
assert_eq!(stats.total_bytes_healed, 8192);
}
#[test]
fn test_heal_statistics_get_success_rate() {
let mut stats = HealStatistics::new();
stats.successful_tasks = 8;
stats.failed_tasks = 2;
// success_rate = 8 / (8 + 2) * 100 = 80%
assert!((stats.get_success_rate() - 80.0).abs() < 0.001);
}
#[test]
fn test_heal_statistics_get_success_rate_zero_total() {
let stats = HealStatistics::new();
assert_eq!(stats.get_success_rate(), 0.0);
}
#[test]
fn test_heal_statistics_get_success_rate_all_success() {
let mut stats = HealStatistics::new();
stats.successful_tasks = 10;
stats.failed_tasks = 0;
assert!((stats.get_success_rate() - 100.0).abs() < 0.001);
}
#[test]
fn test_heal_statistics_get_success_rate_all_failure() {
let mut stats = HealStatistics::new();
stats.successful_tasks = 0;
stats.failed_tasks = 5;
assert_eq!(stats.get_success_rate(), 0.0);
}
}
crates/heal/src/heal/task.rs
+23 -21
View File
@@ -109,7 +109,7 @@ pub enum HealType {
}
impl HealType {
fn log_kind(&self) -> &'static str {
pub(crate) fn kind_label(&self) -> &'static str {
match self {
Self::Cluster => "cluster",
Self::Object { .. } => "object",
@@ -227,6 +227,19 @@ impl Default for HealOptions {
}
}
impl HealOptions {
pub(crate) fn set_key(&self) -> Option<String> {
match (self.pool_index, self.set_index) {
(Some(pool), Some(set)) => Some(format!("pool_{pool}_set_{set}")),
_ => None,
}
}
pub(crate) fn set_metric_label(&self) -> String {
self.set_key().unwrap_or_else(|| "global".to_string())
}
}
/// Heal task status
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum HealTaskStatus {
@@ -491,15 +504,7 @@ impl HealTask {
}
pub fn metric_type_label(&self) -> &'static str {
match &self.heal_type {
HealType::Cluster => "cluster",
HealType::Object { .. } => "object",
HealType::Bucket { .. } => "bucket",
HealType::Prefix { .. } => "prefix",
HealType::ErasureSet { .. } => "erasure_set",
HealType::Metadata { .. } => "metadata",
HealType::ECDecode { .. } => "ec_decode",
}
self.heal_type.kind_label()
}
pub(crate) fn has_batch_failure(&self) -> bool {
@@ -520,10 +525,7 @@ impl HealTask {
pub fn metric_set_label(&self) -> String {
match &self.heal_type {
HealType::ErasureSet { set_disk_id, .. } => set_disk_id.clone(),
_ => match (self.options.pool_index, self.options.set_index) {
(Some(pool), Some(set)) => format!("pool_{pool}_set_{set}"),
_ => "global".to_string(),
},
_ => self.options.set_metric_label(),
}
}
@@ -532,7 +534,7 @@ impl HealTask {
let mut event = TraceEvent::new(TraceKind::Heal, TraceFunc::HealTask)
.with_duration(duration)
.with_attr("task_id", self.id.as_str())
.with_attr("heal_type", self.heal_type.log_kind())
.with_attr("heal_type", self.heal_type.kind_label())
.with_attr("state", state)
.with_attr("source", self.source.as_str())
.with_attr("priority", self.priority.as_str())
@@ -795,7 +797,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "started",
queue_delay = ?queue_delay,
"Heal task started"
@@ -836,7 +838,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "completed",
"Heal task completed"
});
@@ -850,7 +852,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "cancelled",
"Heal task cancelled"
);
@@ -863,7 +865,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "timed_out",
"Heal task timed out"
});
@@ -880,7 +882,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "failed",
error = %e,
"Heal task failed"
@@ -909,7 +911,7 @@ impl HealTask {
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
heal_type = self.heal_type.log_kind(),
heal_type = self.heal_type.kind_label(),
state = "cancelled",
source = "manual",
"Heal task cancellation requested"