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

Merge branch 'main' into hs01-mrf-wiring

Browse Source
This commit is contained in:
houseme
2026-08-18 10:18:18 +08:00
committed by GitHub
parent e5ec4b95cf 9a2d06b370
commit ab91a05ef5
55 changed files with 3135 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/Cargo.toml
+3
View File
@@ -108,6 +108,9 @@ temp-env = { workspace = true }
tempfile = { workspace = true }
uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
tokio = { workspace = true, features = ["test-util", "fs", "rt-multi-thread"] }
# Test-only: pins the emitted scanner alert wire names against the canonical
# EventName string forms subscribers configure (rustfs/backlog#1868).
rustfs-s3-types.workspace = true
# Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
# the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
rustfs-ecstore = { workspace = true, features = ["test-util"] }
crates/scanner/src/scanner_folder.rs
+479 -4
View File
@@ -12,10 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::fs::FileType;
use std::io::ErrorKind;
use std::sync::{Arc, Once};
use std::sync::{Arc, Mutex, Once};
use std::time::{Duration, Instant, SystemTime};
use crate::ReplTargetSizeSummary;
@@ -32,6 +32,7 @@ use crate::scanner_io::{
SCANNER_SKIP_FILE_ERROR, ScannerIODisk as _, is_scanner_metadata_corrupt_error, is_scanner_metadata_transient_error,
};
use crate::sleeper::DynamicSleeper;
use crate::storage_api::owner::{EcstoreEventArgs, ecstore_send_event};
use metrics::{counter, describe_counter};
use rustfs_common::heal_channel::{
HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest,
@@ -41,6 +42,7 @@ use rustfs_common::metrics::{
CloseDiskGuard, IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource,
UpdateCurrentPathFn, current_path_updater, global_metrics,
};
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration};
@@ -97,6 +99,101 @@ const METRIC_SCANNER_EXCESS_FOLDERS_TOTAL: &str = "rustfs_scanner_excess_folders
const METRIC_SCANNER_PENDING_HEAL_PRUNE_TOTAL: &str = "rustfs_scanner_pending_heal_prune_total";
const METRIC_SCANNER_PENDING_HEAL_MALFORMED_TOTAL: &str = "rustfs_scanner_pending_heal_malformed_total";
const MAX_PENDING_SCANNER_HEAL_RETRIES_PER_BUCKET: usize = 128;
// --- scanner excess alerts as S3 notification events (rustfs/backlog#1868) --
//
// The excess-versions / excess-version-size / excess-folders alerts were
// metrics-and-logs only; subscribers (consoles, external auditors) had no way
// to hear them. MinIO emits s3:ObjectManyVersions / s3:ObjectLargeVersions /
// s3:PrefixManyFolders for the same conditions — RustFS carries those as
// EventName::Scanner* with the wire names below. Without a cooldown a single
// over-threshold object would re-emit on every scan cycle (~a minute), so
// emissions are edge-held per (kind, bucket, object) for 24h.
/// `s3:Scanner:ManyVersions` (MinIO `s3:ObjectManyVersions`).
pub const EVENT_SCANNER_MANY_VERSIONS: &str = "s3:Scanner:ManyVersions";
/// `s3:Scanner:LargeVersions` (MinIO `s3:ObjectLargeVersions`).
pub const EVENT_SCANNER_LARGE_VERSIONS: &str = "s3:Scanner:LargeVersions";
/// `s3:Scanner:BigPrefix` (MinIO `s3:PrefixManyFolders`).
pub const EVENT_SCANNER_BIG_PREFIX: &str = "s3:Scanner:BigPrefix";
const ENV_SCANNER_ALERT_COOLDOWN_SECS: &str = "RUSTFS_SCANNER_ALERT_COOLDOWN_SECS";
const DEFAULT_SCANNER_ALERT_COOLDOWN_SECS: u64 = 86_400;
/// Hard cap on distinct cooldown keys; a pathological number of over-threshold
/// objects clears the map wholesale instead of growing without bound (the
/// worst case is one re-emission per still-hot key per scan cycle).
const MAX_SCANNER_ALERT_COOLDOWN_KEYS: usize = 4096;
/// Distinct alert kinds sharing one cooldown map.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
enum ScannerAlertKind {
ManyVersions,
LargeVersions,
BigPrefix,
}
type ScannerAlertCooldownKey = (ScannerAlertKind, String, String);
type ScannerAlertCooldownMap = HashMap<ScannerAlertCooldownKey, Instant>;
static SCANNER_ALERT_EMISSION_COOLDOWN: Mutex<Option<ScannerAlertCooldownMap>> = Mutex::new(None);
fn scanner_alert_cooldown() -> Duration {
let raw = std::env::var(ENV_SCANNER_ALERT_COOLDOWN_SECS)
.ok()
.and_then(|v| v.parse::<u64>().ok());
Duration::from_secs(raw.unwrap_or(DEFAULT_SCANNER_ALERT_COOLDOWN_SECS))
}
/// Edge-held emission gate: returns `true` (and records the cooldown) only
/// when this (kind, bucket, object) last fired longer than the cooldown ago —
/// or never. Metrics and logs stay level-triggered every cycle; only the
/// notification events are held back.
fn scanner_alert_emission_allows(kind: ScannerAlertKind, bucket: &str, object: &str, cooldown: Duration) -> bool {
let key = (kind, bucket.to_string(), object.to_string());
let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner());
let guard = guard.get_or_insert_with(ScannerAlertCooldownMap::new);
let now = Instant::now();
// Expired entries leave first; the cap is still exceeded only when live
// keys alone overflow it, in which case a wholesale clear trades one
// extra emission per hot key for a hard memory bound.
if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
guard.retain(|_, fired_at| now.duration_since(*fired_at) < cooldown);
if guard.len() >= MAX_SCANNER_ALERT_COOLDOWN_KEYS {
guard.clear();
}
}
match guard.get(&key) {
Some(fired_at) if now.duration_since(*fired_at) < cooldown => false,
_ => {
guard.insert(key, now);
true
}
}
}
/// Emit a scanner alert as an S3 notification event through the standard
/// dispatch pipeline. Fire-and-forget: the notify layer owns delivery,
/// retry, and target filtering; the scanner never waits on it.
fn emit_scanner_alert_event(event_name: &str, bucket: &str, object: &str, size: i64, details: &[(&str, String)]) {
let mut req_params = HashMap::with_capacity(details.len());
for (key, value) in details {
req_params.insert((*key).to_string(), value.clone());
}
ecstore_send_event(EcstoreEventArgs {
event_name: event_name.to_string(),
bucket_name: bucket.to_string(),
object: crate::ScannerObjectInfo {
bucket: bucket.to_string(),
name: object.to_string(),
size,
..Default::default()
},
req_params,
user_agent: "Scanner".to_string(),
..Default::default()
});
}
const MAX_PENDING_SCANNER_HEALS_PER_BUCKET: usize = 10_000;
static SCANNER_INLINE_HEAL_WARN_ONCE: Once = Once::new();
@@ -430,6 +527,113 @@ fn non_negative_i64_to_u64(value: i64) -> u64 {
value.max(0) as u64
}
fn trace_start_instant() -> Option<Instant> {
(trace_subscriber_count() > 0).then(Instant::now)
}
fn emit_scanner_folder_trace(root: &str, folder: &str, objects: u64, started_at: Option<Instant>, state: &'static str) {
let Some(started_at) = started_at else {
return;
};
trace_emit(|| {
let (bucket, prefix) = path2_bucket_object_with_base_path(root, folder);
TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerFolder)
.with_bucket(bucket)
.with_object(prefix)
.with_duration(started_at.elapsed())
.with_attr("state", state)
.with_attr("objects", objects)
});
}
fn emit_scanner_ilm_action_trace(
bucket: &str,
object: &str,
action: IlmAction,
count: u64,
queued: bool,
started_at: Option<Instant>,
) {
let Some(started_at) = started_at else {
return;
};
let state = if queued { "queued" } else { "not_queued" };
trace_emit(|| {
TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerIlmAction)
.with_bucket(bucket)
.with_object(object)
.with_duration(started_at.elapsed())
.with_attr("state", state)
.with_attr("action", action.as_str())
.with_attr("count", count)
.with_attr("queued", queued)
});
}
struct ScannerHealCandidateTraceContext {
bucket: String,
object: Option<String>,
version_id: Option<String>,
scan_mode: Option<HealScanMode>,
started_at: Instant,
}
fn scanner_heal_candidate_trace_context(request: &HealChannelRequest) -> Option<ScannerHealCandidateTraceContext> {
let started_at = trace_start_instant()?;
Some(ScannerHealCandidateTraceContext {
bucket: request.bucket.clone(),
object: request.object_prefix.clone(),
version_id: request.object_version_id.clone(),
scan_mode: request.scan_mode,
started_at,
})
}
struct ScannerHealCandidateTrace<'a> {
candidate_type: &'static str,
bucket: &'a str,
object: Option<&'a str>,
version_id: Option<&'a str>,
priority: HealChannelPriority,
scan_mode: Option<HealScanMode>,
result: Result<HealAdmissionResult, &'a str>,
started_at: Instant,
}
fn emit_scanner_heal_candidate_trace(trace: ScannerHealCandidateTrace<'_>) {
trace_emit(|| {
let (state, admission, error) = match trace.result {
Ok(result) if result.is_admitted() => ("admitted", describe_heal_admission(result), None),
Ok(result) => ("not_admitted", describe_heal_admission(result), None),
Err(error) => ("submit_failed", "channel_error".to_string(), Some(error)),
};
let mut event = TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerHealCandidate)
.with_bucket(trace.bucket)
.with_duration(trace.started_at.elapsed())
.with_attr("state", state)
.with_attr("candidate_type", trace.candidate_type)
.with_attr("priority", heal_priority_label(trace.priority))
.with_attr("admission", admission);
if let Some(object) = trace.object {
event = event.with_object(object);
}
if let Some(version_id) = trace.version_id {
event = event.with_attr("version_id", version_id);
}
if let Some(scan_mode) = trace.scan_mode {
event = event.with_attr("scan_mode", scan_mode.as_str());
}
if let Some(error) = error {
event = event.with_attr("error", error);
}
event
});
}
fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary) {
into.size = into.size.saturating_add(summary.total_size);
into.versions = into.versions.saturating_add(summary.versions);
@@ -677,9 +881,22 @@ async fn send_scanner_heal_request(
request: HealChannelRequest,
) -> Result<HealAdmissionResult, ScannerError> {
let priority = request.priority;
let trace_context = scanner_heal_candidate_trace_context(&request);
match send_heal_request_with_admission(request).await {
Ok(result) => {
record_heal_candidate_admission(candidate_type, priority, result);
if let Some(trace_context) = trace_context.as_ref() {
emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace {
candidate_type,
bucket: &trace_context.bucket,
object: trace_context.object.as_deref(),
version_id: trace_context.version_id.as_deref(),
priority,
scan_mode: trace_context.scan_mode,
result: Ok(result),
started_at: trace_context.started_at,
});
}
Ok(result)
}
Err(err) => {
@@ -690,6 +907,18 @@ async fn send_scanner_heal_request(
"result" => "channel_error".to_string()
)
.increment(1);
if let Some(trace_context) = trace_context.as_ref() {
emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace {
candidate_type,
bucket: &trace_context.bucket,
object: trace_context.object.as_deref(),
version_id: trace_context.version_id.as_deref(),
priority,
scan_mode: trace_context.scan_mode,
result: Err(err.as_str()),
started_at: trace_context.started_at,
});
}
Err(ScannerError::Other(err))
}
}
@@ -905,7 +1134,9 @@ impl ScannerItem {
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
remaining_versions = 0;
@@ -957,7 +1188,9 @@ impl ScannerItem {
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
if !versioning_config.prefix_enabled(&self.object_path()) && event.action == IlmAction::DeleteAction {
@@ -995,7 +1228,9 @@ impl ScannerItem {
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let trace_started_at = trace_start_instant();
let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
emit_scanner_ilm_action_trace(&self.bucket, &oi.name, event.action, 1, queued, trace_started_at);
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
done_ilm(1)();
}
@@ -1019,7 +1254,21 @@ impl ScannerItem {
let action = event.action;
let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX);
let done_ilm = Metrics::time_ilm(action);
let trace_started_at = trace_start_instant();
let queued = enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await;
if let Some(trace_started_at) = trace_started_at {
let state = if queued { "queued" } else { "not_queued" };
trace_emit(|| {
TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerIlmAction)
.with_bucket(self.bucket.as_str())
.with_object(self.object_path())
.with_duration(trace_started_at.elapsed())
.with_attr("state", state)
.with_attr("action", action.as_str())
.with_attr("count", count)
.with_attr("queued", queued)
});
}
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
done_ilm(count)();
remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len());
@@ -1197,6 +1446,7 @@ impl ScannerItem {
fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) {
ensure_scanner_alert_metrics_registered();
let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size);
let object_path = self.object_path();
if too_many_versions {
global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1);
counter!(
@@ -1204,13 +1454,26 @@ impl ScannerItem {
"bucket" => self.bucket.clone()
)
.increment(1);
if scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, &self.bucket, &object_path, scanner_alert_cooldown())
{
emit_scanner_alert_event(
EVENT_SCANNER_MANY_VERSIONS,
&self.bucket,
&object_path,
cumulative_size,
&[
("versions", remaining_versions.to_string()),
("threshold", scanner_excess_versions_threshold().to_string()),
],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %self.bucket,
object = %self.object_path(),
object = %object_path,
versions = remaining_versions,
threshold = scanner_excess_versions_threshold(),
state = "excess_versions",
@@ -1224,13 +1487,31 @@ impl ScannerItem {
"bucket" => self.bucket.clone()
)
.increment(1);
if scanner_alert_emission_allows(
ScannerAlertKind::LargeVersions,
&self.bucket,
&object_path,
scanner_alert_cooldown(),
) {
emit_scanner_alert_event(
EVENT_SCANNER_LARGE_VERSIONS,
&self.bucket,
&object_path,
cumulative_size,
&[
("versions", remaining_versions.to_string()),
("cumulativeSize", cumulative_size.to_string()),
("threshold", scanner_excess_version_size_threshold().to_string()),
],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %self.bucket,
object = %self.object_path(),
object = %object_path,
versions = remaining_versions,
cumulative_size,
threshold = scanner_excess_version_size_threshold(),
@@ -1611,6 +1892,15 @@ impl FolderScanner {
"root" => self.root.clone()
)
.increment(1);
if scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, &self.root, folder, scanner_alert_cooldown()) {
emit_scanner_alert_event(
EVENT_SCANNER_BIG_PREFIX,
&self.root,
folder,
0,
&[("folders", total_folders.to_string()), ("threshold", threshold.to_string())],
);
}
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
@@ -1830,6 +2120,7 @@ impl FolderScanner {
into: &mut DataUsageEntry,
) -> Result<(), ScannerError> {
let done_folder = Metrics::time(Metric::ScanFolder);
let trace_started_at = trace_start_instant();
if ctx.is_cancelled() {
return Err(ScannerError::Other("Operation cancelled".to_string()));
@@ -2904,6 +3195,8 @@ impl FolderScanner {
}
done_folder();
let scanned_objects = u64::try_from(into.objects).unwrap_or(u64::MAX);
emit_scanner_folder_trace(&self.root, &folder.name, scanned_objects, trace_started_at, "completed");
Ok(())
}
@@ -3085,6 +3378,90 @@ mod tests {
#[cfg(unix)]
use std::os::unix::fs::{PermissionsExt, symlink};
use std::sync::Mutex;
/// Reset the process-global alert cooldown map; test-only.
fn reset_alert_cooldowns() {
*SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner()) = Some(ScannerAlertCooldownMap::new());
}
/// The emitted event-name strings must be exactly what `EventName`
/// serializes, or a bucket notification subscribed to the documented name
/// would silently never match (rustfs/backlog#1868).
#[test]
fn scanner_alert_wire_names_match_canonical_event_names() {
use rustfs_s3_types::EventName;
assert_eq!(EVENT_SCANNER_MANY_VERSIONS, EventName::ScannerManyVersions.to_string());
assert_eq!(EVENT_SCANNER_LARGE_VERSIONS, EventName::ScannerLargeVersions.to_string());
assert_eq!(EVENT_SCANNER_BIG_PREFIX, EventName::ScannerBigPrefix.to_string());
}
fn cooldown_map_len() -> usize {
SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner())
.as_ref()
.map(|map| map.len())
.unwrap_or(0)
}
/// Backdate every recorded cooldown so the next check fires again.
fn expire_all_alert_cooldowns(cooldown: Duration) {
let now = Instant::now();
let mut guard = SCANNER_ALERT_EMISSION_COOLDOWN
.lock()
.unwrap_or_else(|poison| poison.into_inner());
if let Some(map) = guard.as_mut() {
for fired_at in map.values_mut() {
if let Some(expired) = now.checked_sub(cooldown + Duration::from_secs(1)) {
*fired_at = expired;
}
}
}
}
/// The emission gate is the only thing standing between an over-threshold
/// object and one S3 event per scan cycle, so its edge semantics get
/// pinned directly. All scenarios share one #[test] because the cooldown
/// map is process-global and parallel tests would read each other's
/// firings.
#[test]
fn scanner_alert_emission_is_edge_held_per_key_and_bounded() {
reset_alert_cooldowns();
let cooldown = Duration::from_secs(3600);
// First firing allows, an immediate re-check is held.
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
assert!(!scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
// Different kind, object, and bucket are independent keys.
assert!(scanner_alert_emission_allows(ScannerAlertKind::LargeVersions, "bkt", "obj", cooldown));
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "other", cooldown));
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "other", "obj", cooldown));
assert_eq!(cooldown_map_len(), 4);
// After the cooldown elapses the same key fires again.
expire_all_alert_cooldowns(cooldown);
assert!(scanner_alert_emission_allows(ScannerAlertKind::ManyVersions, "bkt", "obj", cooldown));
// A zero cooldown degenerates to always-emit (operators may want that).
assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
assert!(scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", "dir", Duration::ZERO));
// Hard bound: overflow the cap with zero-cooldown keys and confirm the
// map clears rather than growing past it.
reset_alert_cooldowns();
for index in 0..=(MAX_SCANNER_ALERT_COOLDOWN_KEYS + 8) {
let _ = scanner_alert_emission_allows(ScannerAlertKind::BigPrefix, "bkt", &format!("dir-{index}"), Duration::ZERO);
}
assert!(
cooldown_map_len() <= MAX_SCANNER_ALERT_COOLDOWN_KEYS,
"cooldown map must stay bounded, got {}",
cooldown_map_len()
);
}
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use temp_env::{with_var, with_var_unset};
use tracing_subscriber::fmt::MakeWriter;
@@ -4409,6 +4786,104 @@ mod tests {
);
}
#[tokio::test]
async fn scanner_trace_helpers_emit_expected_events() {
let mut trace = rustfs_common::trace_bus::subscribe_trace_events();
emit_scanner_folder_trace(
"/tmp/rustfs-scanner-trace",
"/tmp/rustfs-scanner-trace/bucket-a/folder-a",
7,
Some(Instant::now()),
"completed",
);
let folder = recv_scanner_trace_event(
&mut trace,
TraceFunc::ScannerFolder,
Some("bucket-a"),
Some("folder-a"),
Some("completed"),
)
.await;
assert_eq!(trace_attr_string(&folder, "objects").as_deref(), Some("7"));
emit_scanner_ilm_action_trace("bucket-a", "object-a", IlmAction::DeleteAction, 2, true, Some(Instant::now()));
let ilm = recv_scanner_trace_event(
&mut trace,
TraceFunc::ScannerIlmAction,
Some("bucket-a"),
Some("object-a"),
Some("queued"),
)
.await;
assert_eq!(trace_attr_string(&ilm, "action").as_deref(), Some("delete"));
assert_eq!(trace_attr_string(&ilm, "count").as_deref(), Some("2"));
assert_eq!(trace_attr_string(&ilm, "queued").as_deref(), Some("true"));
emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace {
candidate_type: "object",
bucket: "bucket-a",
object: Some("object-a"),
version_id: Some("version-a"),
priority: HealChannelPriority::High,
scan_mode: Some(HealScanMode::Deep),
result: Ok(HealAdmissionResult::Merged),
started_at: Instant::now(),
});
let heal_candidate = recv_scanner_trace_event(
&mut trace,
TraceFunc::ScannerHealCandidate,
Some("bucket-a"),
Some("object-a"),
Some("admitted"),
)
.await;
assert_eq!(trace_attr_string(&heal_candidate, "candidate_type").as_deref(), Some("object"));
assert_eq!(trace_attr_string(&heal_candidate, "priority").as_deref(), Some("high"));
assert_eq!(trace_attr_string(&heal_candidate, "scan_mode").as_deref(), Some("deep"));
assert_eq!(trace_attr_string(&heal_candidate, "version_id").as_deref(), Some("version-a"));
assert_eq!(trace_attr_string(&heal_candidate, "admission").as_deref(), Some("merged"));
}
async fn recv_scanner_trace_event(
trace: &mut rustfs_common::trace_bus::TraceSubscription,
func: TraceFunc,
bucket: Option<&str>,
object: Option<&str>,
state: Option<&str>,
) -> TraceEvent {
for _ in 0..32 {
let event = tokio::time::timeout(Duration::from_secs(1), trace.recv())
.await
.expect("scanner trace event should arrive")
.expect("trace bus should stay open");
if event.kind == TraceKind::Scanner
&& event.func == func
&& event.bucket.as_deref() == bucket
&& event.object.as_deref() == object
&& state.is_none_or(|state| trace_attr_string(&event, "state").as_deref() == Some(state))
{
return (*event).clone();
}
}
panic!("expected scanner trace event {func:?} for bucket {bucket:?} object {object:?}");
}
fn trace_attr_string(event: &TraceEvent, key: &str) -> Option<String> {
event.attrs.iter().find_map(|attr| {
if attr.key != key {
return None;
}
Some(match &attr.value {
rustfs_common::trace_bus::TraceVal::Bool(value) => value.to_string(),
rustfs_common::trace_bus::TraceVal::U64(value) => value.to_string(),
rustfs_common::trace_bus::TraceVal::I64(value) => value.to_string(),
rustfs_common::trace_bus::TraceVal::Str(value) => value.to_string(),
})
})
}
#[test]
fn test_build_high_priority_heal_admission_error_contains_context() {
let err = build_high_priority_heal_admission_error(
crates/scanner/src/storage_api.rs
+4 -3
View File
@@ -78,6 +78,7 @@ pub(crate) use rustfs_ecstore::api::disk::{
pub(crate) use rustfs_ecstore::api::error::{
Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError,
};
pub(crate) use rustfs_ecstore::api::event::{EventArgs as EcstoreEventArgs, send_event as ecstore_send_event};
#[cfg(test)]
pub(crate) use rustfs_ecstore::api::layout::{
EndpointServerPools as EcstoreEndpointServerPools, Endpoints as EcstoreEndpoints, PoolEndpoints as EcstorePoolEndpoints,
@@ -110,8 +111,8 @@ pub(crate) mod owner {
ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS,
ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys,
EcstoreDisk, EcstoreDiskAPI, EcstoreDiskBytes, EcstoreDiskError, EcstoreDiskInfo, EcstoreDiskInfoOptions,
EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc,
EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs,
EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts,
EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard,
EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi,
ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule,
@@ -121,7 +122,7 @@ pub(crate) mod owner {
ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw,
ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path,
ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle,
ecstore_save_config, scanner_replication_config_for_lifecycle_eval,
ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval,
};
#[cfg(test)]