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

feat(scanner): define replication boundary contract (#3630)

Browse Source 
* feat(scanner): define replication boundary contract

* fix(scanner): propagate replication boundary metrics

* fix(scanner): preserve repair metadata on merge

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-06-23 21:35:36 +08:00
committed by GitHub
parent eff656e086
commit 7bdb25ae9d
6 changed files with 282 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/common/src/metrics.rs
+130
View File
@@ -319,6 +319,36 @@ pub enum ScannerReplicationRepairKind {
SiteActiveResync,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScannerReplicationRepairRole {
RepairAdmission,
BoundarySignal,
}
impl ScannerReplicationRepairRole {
pub fn as_str(self) -> &'static str {
match self {
Self::RepairAdmission => "repair_admission",
Self::BoundarySignal => "boundary_signal",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ScannerReplicationExecutionOwner {
BucketReplicationQueue,
SiteReplicationRuntime,
}
impl ScannerReplicationExecutionOwner {
pub fn as_str(self) -> &'static str {
match self {
Self::BucketReplicationQueue => "bucket_replication_queue",
Self::SiteReplicationRuntime => "site_replication_runtime",
}
}
}
impl ScannerReplicationRepairKind {
const ALL: [Self; 6] = [
Self::BucketObject,
@@ -338,6 +368,24 @@ impl ScannerReplicationRepairKind {
}
}
pub fn scanner_role(self) -> ScannerReplicationRepairRole {
match self {
Self::BucketObject | Self::BucketDeleteMarker | Self::BucketVersionPurge | Self::BucketExistingObject => {
ScannerReplicationRepairRole::RepairAdmission
}
Self::SitePassiveRequeue | Self::SiteActiveResync => ScannerReplicationRepairRole::BoundarySignal,
}
}
pub fn execution_owner(self) -> ScannerReplicationExecutionOwner {
match self {
Self::BucketObject | Self::BucketDeleteMarker | Self::BucketVersionPurge | Self::BucketExistingObject => {
ScannerReplicationExecutionOwner::BucketReplicationQueue
}
Self::SitePassiveRequeue | Self::SiteActiveResync => ScannerReplicationExecutionOwner::SiteReplicationRuntime,
}
}
pub fn as_str(self) -> &'static str {
match self {
Self::BucketObject => "object",
@@ -515,6 +563,8 @@ impl ScannerSourceWorkValues {
ScannerReplicationRepairSnapshot {
source: kind.source().as_str().to_string(),
kind: kind.as_str().to_string(),
scanner_role: kind.scanner_role().as_str().to_string(),
execution_owner: kind.execution_owner().as_str().to_string(),
checked: self.checked,
queued: self.queued,
executed: self.executed,
@@ -896,6 +946,10 @@ pub struct ScannerSourceWorkSnapshot {
pub struct ScannerReplicationRepairSnapshot {
pub source: String,
pub kind: String,
#[serde(default)]
pub scanner_role: String,
#[serde(default)]
pub execution_owner: String,
pub checked: u64,
pub queued: u64,
pub executed: u64,
@@ -3407,6 +3461,14 @@ mod tests {
ScannerReplicationRepairKind::BucketExistingObject,
ScannerSourceWorkUpdate::queued(4),
);
metrics.record_scanner_replication_repair_work(
ScannerReplicationRepairKind::SitePassiveRequeue,
ScannerSourceWorkUpdate::missed(5),
);
metrics.record_scanner_replication_repair_work(
ScannerReplicationRepairKind::SiteActiveResync,
ScannerSourceWorkUpdate::queued(6),
);
let report = metrics.report().await;
let object = report
@@ -3414,6 +3476,8 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "object")
.expect("bucket object repair work should be visible");
assert_eq!(object.scanner_role, "repair_admission");
assert_eq!(object.execution_owner, "bucket_replication_queue");
assert_eq!(object.queued, 2);
let delete_marker = report
@@ -3421,6 +3485,8 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "delete_marker")
.expect("bucket delete-marker repair work should be visible");
assert_eq!(delete_marker.scanner_role, "repair_admission");
assert_eq!(delete_marker.execution_owner, "bucket_replication_queue");
assert_eq!(delete_marker.missed, 1);
let version_purge = report
@@ -3428,6 +3494,8 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "version_purge")
.expect("bucket version purge repair work should be visible");
assert_eq!(version_purge.scanner_role, "repair_admission");
assert_eq!(version_purge.execution_owner, "bucket_replication_queue");
assert_eq!(version_purge.skipped, 3);
let existing_object = report
@@ -3435,7 +3503,69 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "existing_object")
.expect("bucket existing object repair work should be visible");
assert_eq!(existing_object.scanner_role, "repair_admission");
assert_eq!(existing_object.execution_owner, "bucket_replication_queue");
assert_eq!(existing_object.queued, 4);
let passive_requeue = report
.replication_repair
.iter()
.find(|repair| repair.source == "site_replication" && repair.kind == "passive_requeue")
.expect("site passive requeue boundary work should be visible");
assert_eq!(passive_requeue.scanner_role, "boundary_signal");
assert_eq!(passive_requeue.execution_owner, "site_replication_runtime");
assert_eq!(passive_requeue.missed, 5);
let active_resync = report
.replication_repair
.iter()
.find(|repair| repair.source == "site_replication" && repair.kind == "active_resync")
.expect("site active resync boundary work should be visible");
assert_eq!(active_resync.scanner_role, "boundary_signal");
assert_eq!(active_resync.execution_owner, "site_replication_runtime");
assert_eq!(active_resync.queued, 6);
}
#[test]
fn scanner_replication_repair_kind_sources_are_stable() {
let bucket_kinds = [
ScannerReplicationRepairKind::BucketObject,
ScannerReplicationRepairKind::BucketDeleteMarker,
ScannerReplicationRepairKind::BucketVersionPurge,
ScannerReplicationRepairKind::BucketExistingObject,
];
for kind in bucket_kinds {
assert_eq!(kind.source(), ScannerWorkSource::BucketReplication);
assert_eq!(kind.scanner_role(), ScannerReplicationRepairRole::RepairAdmission);
assert_eq!(kind.execution_owner(), ScannerReplicationExecutionOwner::BucketReplicationQueue);
}
assert_eq!(
ScannerReplicationRepairKind::SitePassiveRequeue.source(),
ScannerWorkSource::SiteReplication
);
assert_eq!(
ScannerReplicationRepairKind::SiteActiveResync.source(),
ScannerWorkSource::SiteReplication
);
assert_eq!(
ScannerReplicationRepairKind::SitePassiveRequeue.scanner_role(),
ScannerReplicationRepairRole::BoundarySignal
);
assert_eq!(
ScannerReplicationRepairKind::SiteActiveResync.scanner_role(),
ScannerReplicationRepairRole::BoundarySignal
);
assert_eq!(
ScannerReplicationRepairKind::SitePassiveRequeue.execution_owner(),
ScannerReplicationExecutionOwner::SiteReplicationRuntime
);
assert_eq!(
ScannerReplicationRepairKind::SiteActiveResync.execution_owner(),
ScannerReplicationExecutionOwner::SiteReplicationRuntime
);
assert_eq!(ScannerReplicationRepairKind::SitePassiveRequeue.as_str(), "passive_requeue");
assert_eq!(ScannerReplicationRepairKind::SiteActiveResync.as_str(), "active_resync");
}
#[tokio::test]
crates/ecstore/src/metrics_realtime.rs
+18
View File
@@ -294,6 +294,8 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
.map(|repair| MadminScannerReplicationRepairSnapshot {
source: repair.source,
kind: repair.kind,
scanner_role: repair.scanner_role,
execution_owner: repair.execution_owner,
checked: repair.checked,
queued: repair.queued,
executed: repair.executed,
@@ -308,6 +310,8 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
.map(|repair| MadminScannerReplicationRepairSnapshot {
source: repair.source,
kind: repair.kind,
scanner_role: repair.scanner_role,
execution_owner: repair.execution_owner,
checked: repair.checked,
queued: repair.queued,
executed: repair.executed,
@@ -322,6 +326,8 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
.map(|repair| MadminScannerReplicationRepairSnapshot {
source: repair.source,
kind: repair.kind,
scanner_role: repair.scanner_role,
execution_owner: repair.execution_owner,
checked: repair.checked,
queued: repair.queued,
executed: repair.executed,
@@ -844,6 +850,8 @@ mod test {
replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
checked: 62,
queued: 63,
executed: 64,
@@ -854,6 +862,8 @@ mod test {
current_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "delete_marker".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
checked: 68,
queued: 69,
executed: 70,
@@ -864,6 +874,8 @@ mod test {
last_cycle_replication_repair: vec![rustfs_common::metrics::ScannerReplicationRepairSnapshot {
source: "site_replication".to_string(),
kind: "active_resync".to_string(),
scanner_role: "boundary_signal".to_string(),
execution_owner: "site_replication_runtime".to_string(),
checked: 74,
queued: 75,
executed: 76,
@@ -943,11 +955,17 @@ mod test {
assert_eq!(scanner.last_cycle_source_work[0].skipped, 60);
assert_eq!(scanner.replication_repair[0].source, "bucket_replication");
assert_eq!(scanner.replication_repair[0].kind, "object");
assert_eq!(scanner.replication_repair[0].scanner_role, "repair_admission");
assert_eq!(scanner.replication_repair[0].execution_owner, "bucket_replication_queue");
assert_eq!(scanner.replication_repair[0].missed, 67);
assert_eq!(scanner.current_cycle_replication_repair[0].kind, "delete_marker");
assert_eq!(scanner.current_cycle_replication_repair[0].scanner_role, "repair_admission");
assert_eq!(scanner.current_cycle_replication_repair[0].execution_owner, "bucket_replication_queue");
assert_eq!(scanner.current_cycle_replication_repair[0].queued, 69);
assert_eq!(scanner.last_cycle_replication_repair[0].source, "site_replication");
assert_eq!(scanner.last_cycle_replication_repair[0].kind, "active_resync");
assert_eq!(scanner.last_cycle_replication_repair[0].scanner_role, "boundary_signal");
assert_eq!(scanner.last_cycle_replication_repair[0].execution_owner, "site_replication_runtime");
assert_eq!(scanner.last_cycle_replication_repair[0].skipped, 78);
assert_eq!(scanner.partial_cycles, 80);
}
crates/madmin/src/metrics.rs
+107
View File
@@ -164,6 +164,10 @@ pub struct ScannerReplicationRepairSnapshot {
pub source: String,
#[serde(rename = "kind", default)]
pub kind: String,
#[serde(rename = "scanner_role", default)]
pub scanner_role: String,
#[serde(rename = "execution_owner", default)]
pub execution_owner: String,
#[serde(rename = "checked", default)]
pub checked: u64,
#[serde(rename = "queued", default)]
@@ -330,6 +334,12 @@ impl ScannerSourceWorkSnapshot {
impl ScannerReplicationRepairSnapshot {
fn merge(&mut self, other: &Self) {
if self.scanner_role.is_empty() && !other.scanner_role.is_empty() {
self.scanner_role = other.scanner_role.clone();
}
if self.execution_owner.is_empty() && !other.execution_owner.is_empty() {
self.execution_owner = other.execution_owner.clone();
}
self.checked = self.checked.saturating_add(other.checked);
self.queued = self.queued.saturating_add(other.queued);
self.executed = self.executed.saturating_add(other.executed);
@@ -1662,6 +1672,8 @@ mod tests {
replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
queued: 2,
..Default::default()
}],
@@ -1674,18 +1686,24 @@ mod tests {
ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
missed: 3,
..Default::default()
},
ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "delete_marker".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
skipped: 4,
..Default::default()
},
ScannerReplicationRepairSnapshot {
source: "site_replication".to_string(),
kind: "active_resync".to_string(),
scanner_role: "boundary_signal".to_string(),
execution_owner: "site_replication_runtime".to_string(),
queued: 5,
..Default::default()
},
@@ -1698,6 +1716,8 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "object")
.expect("bucket object repair snapshot should be merged");
assert_eq!(object.scanner_role, "repair_admission");
assert_eq!(object.execution_owner, "bucket_replication_queue");
assert_eq!(object.queued, 2);
assert_eq!(object.missed, 3);
@@ -1706,6 +1726,8 @@ mod tests {
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "delete_marker")
.expect("bucket delete-marker repair snapshot should be merged");
assert_eq!(delete_marker.scanner_role, "repair_admission");
assert_eq!(delete_marker.execution_owner, "bucket_replication_queue");
assert_eq!(delete_marker.skipped, 4);
let site_resync = scanner
@@ -1713,9 +1735,94 @@ mod tests {
.iter()
.find(|repair| repair.source == "site_replication" && repair.kind == "active_resync")
.expect("site active resync snapshot should remain separate");
assert_eq!(site_resync.scanner_role, "boundary_signal");
assert_eq!(site_resync.execution_owner, "site_replication_runtime");
assert_eq!(site_resync.queued, 5);
}
#[test]
fn scanner_metrics_merge_preserves_replication_repair_metadata_from_newer_nodes() {
let collected_at = Utc::now();
let mut scanner = ScannerMetrics {
collected_at,
replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
checked: 2,
..Default::default()
}],
current_cycle_replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "delete_marker".to_string(),
queued: 3,
..Default::default()
}],
last_cycle_replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "site_replication".to_string(),
kind: "active_resync".to_string(),
missed: 4,
..Default::default()
}],
..Default::default()
};
scanner.merge(&ScannerMetrics {
collected_at: collected_at + chrono::Duration::seconds(1),
replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "object".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
checked: 5,
..Default::default()
}],
current_cycle_replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "bucket_replication".to_string(),
kind: "delete_marker".to_string(),
scanner_role: "repair_admission".to_string(),
execution_owner: "bucket_replication_queue".to_string(),
queued: 7,
..Default::default()
}],
last_cycle_replication_repair: vec![ScannerReplicationRepairSnapshot {
source: "site_replication".to_string(),
kind: "active_resync".to_string(),
scanner_role: "boundary_signal".to_string(),
execution_owner: "site_replication_runtime".to_string(),
missed: 11,
..Default::default()
}],
..Default::default()
});
let object = scanner
.replication_repair
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "object")
.expect("mixed-version object repair snapshot should be merged");
assert_eq!(object.scanner_role, "repair_admission");
assert_eq!(object.execution_owner, "bucket_replication_queue");
assert_eq!(object.checked, 7);
let delete_marker = scanner
.current_cycle_replication_repair
.iter()
.find(|repair| repair.source == "bucket_replication" && repair.kind == "delete_marker")
.expect("mixed-version current-cycle repair snapshot should be merged");
assert_eq!(delete_marker.scanner_role, "repair_admission");
assert_eq!(delete_marker.execution_owner, "bucket_replication_queue");
assert_eq!(delete_marker.queued, 10);
let active_resync = scanner
.last_cycle_replication_repair
.iter()
.find(|repair| repair.source == "site_replication" && repair.kind == "active_resync")
.expect("mixed-version last-cycle repair snapshot should be merged");
assert_eq!(active_resync.scanner_role, "boundary_signal");
assert_eq!(active_resync.execution_owner, "site_replication_runtime");
assert_eq!(active_resync.missed, 15);
}
#[test]
fn scanner_metrics_merge_preserves_distributed_status_fields() {
let collected_at = Utc::now();
crates/scanner/src/scanner_folder.rs
+2
View File
@@ -2643,6 +2643,8 @@ mod tests {
.iter()
.find(|repair| repair.source == ScannerWorkSource::BucketReplication.as_str() && repair.kind == "object")
.expect("bucket object repair work should be visible");
assert_eq!(object_repair.scanner_role, "repair_admission");
assert_eq!(object_repair.execution_owner, "bucket_replication_queue");
assert_eq!(object_repair.skipped, 1);
assert_eq!(object_repair.queued, 1);
assert_eq!(object_repair.missed, 1);