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

refactor(logging): standardize heal and scanner events (#3414)

Browse Source 
* refactor(logging): standardize heal and scanner events

* chore(git): untrack local logging governance note

* chore(git): ignore local logging governance note
This commit is contained in:
houseme
2026-06-14 01:47:39 +08:00
committed by GitHub
parent 9059a9c68d
commit 7da10db852
14 changed files with 3425 additions and 363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+1
View File
@@ -47,6 +47,7 @@ __pycache__/
docs/*
!docs/architecture/
!docs/architecture/**
docs/heal-scanner-logging-governance.md
# nix stuff
result*
crates/heal/src/heal/channel.rs
+113 -15
View File
@@ -28,6 +28,12 @@ use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, error, info};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_CHANNEL: &str = "channel";
const EVENT_HEAL_CHANNEL_STATE: &str = "heal_channel_state";
const EVENT_HEAL_CHANNEL_REQUEST: &str = "heal_channel_request";
const EVENT_HEAL_CHANNEL_RESPONSE: &str = "heal_channel_response";
/// Heal channel processor
pub struct HealChannelProcessor {
/// Heal manager
@@ -57,7 +63,14 @@ impl HealChannelProcessor {
/// Start processing heal channel requests
pub async fn start(&mut self, mut receiver: HealChannelReceiver) -> Result<()> {
info!("Starting heal channel processor");
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "started",
"Heal channel processor state updated"
);
loop {
tokio::select! {
@@ -65,11 +78,26 @@ impl HealChannelProcessor {
match command {
Some(command) => {
if let Err(e) = self.process_command(command).await {
error!("Failed to process heal command: {}", e);
error!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "process_failed",
error = %e,
"Heal channel request processing failed"
);
}
}
None => {
debug!("Heal channel receiver closed, stopping processor");
debug!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "receiver_closed",
"Heal channel processor state updated"
);
break;
}
}
@@ -77,13 +105,29 @@ impl HealChannelProcessor {
response = self.response_receiver.recv() => {
if let Some(response) = response {
// Handle response if needed
info!("Received heal response for request: {}", response.request_id);
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_RESPONSE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %response.request_id,
success = response.success,
state = "received_local",
"Heal channel response observed"
);
}
}
}
}
info!("Heal channel processor stopped");
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "stopped",
"Heal channel processor state updated"
);
Ok(())
}
@@ -111,10 +155,15 @@ impl HealChannelProcessor {
response_tx: oneshot::Sender<std::result::Result<HealAdmissionResult, String>>,
) -> Result<()> {
info!(
"Processing heal start request: {} for bucket: {}/{}",
request.id,
request.bucket,
request.object_prefix.as_deref().unwrap_or("")
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %request.id,
bucket = %request.bucket,
object_prefix = %request.object_prefix.as_deref().unwrap_or(""),
state = "start_received",
"Heal channel start request received"
);
// Convert channel request to heal request
@@ -137,9 +186,14 @@ impl HealChannelProcessor {
match self.heal_manager.submit_heal_request(heal_request).await {
Ok(admission) => {
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %request.id,
admission = admission.result_label(),
"Heal request admission decision completed"
state = "admission_decided",
"Heal channel admission decision completed"
);
let _ = response_tx.send(Ok(admission));
@@ -163,7 +217,16 @@ impl HealChannelProcessor {
}
Err(e) => {
let error_text = e.to_string();
error!("Failed to submit heal request: {} - {}", request.id, error_text);
error!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %request.id,
state = "submit_failed",
error = %error_text,
"Heal channel start request failed"
);
let _ = response_tx.send(Err(error_text.clone()));
// Send error response
@@ -188,7 +251,16 @@ impl HealChannelProcessor {
client_token: String,
response_tx: oneshot::Sender<std::result::Result<HealChannelResponse, String>>,
) -> Result<()> {
info!("Processing heal query request for path: {}", heal_path);
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %client_token,
heal_path = %heal_path,
state = "query_received",
"Heal channel query request received"
);
let (summary, detail, items) = match self.heal_manager.get_task_report_for_path(&heal_path, &client_token).await {
Ok(HealTaskReport {
@@ -260,7 +332,16 @@ impl HealChannelProcessor {
client_token: String,
response_tx: oneshot::Sender<std::result::Result<HealChannelResponse, String>>,
) -> Result<()> {
info!("Processing heal cancel request for path: {}", heal_path);
info!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %client_token,
heal_path = %heal_path,
state = "cancel_received",
"Heal channel cancel request received"
);
let request_id = if client_token.is_empty() {
heal_path.clone()
@@ -362,12 +443,29 @@ impl HealChannelProcessor {
fn publish_response(&self, response: HealChannelResponse) {
// Try to send to local channel first, but don't block broadcast on failure
if let Err(e) = self.response_sender.send(response.clone()) {
error!("Failed to enqueue heal response locally: {}", e);
error!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_RESPONSE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
request_id = %response.request_id,
state = "enqueue_local_failed",
error = %e,
"Heal channel response enqueue failed"
);
}
// Always attempt to broadcast, even if local send failed
// Use the original response for broadcast; local send uses a clone
if let Err(e) = publish_heal_response(response) {
error!("Failed to broadcast heal response: {}", e);
error!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_RESPONSE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "broadcast_failed",
error = %e,
"Heal channel response broadcast failed"
);
}
}
crates/heal/src/heal/erasure_healer.rs
+297 -33
View File
@@ -29,6 +29,12 @@ use std::sync::{
use tokio::sync::{RwLock, Semaphore};
use tracing::{error, info, warn};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_ERASURE_HEALER: &str = "erasure_healer";
const EVENT_HEAL_ERASURE_RESUME_STATE: &str = "heal_erasure_resume_state";
const EVENT_HEAL_ERASURE_BUCKET_STATE: &str = "heal_erasure_bucket_state";
const EVENT_HEAL_ERASURE_OBJECT_STATE: &str = "heal_erasure_object_state";
/// Erasure Set Healer
pub struct ErasureSetHealer {
storage: Arc<dyn HealStorageAPI>,
@@ -93,7 +99,16 @@ impl ErasureSetHealer {
/// execute erasure set heal with resume
#[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
info!("Starting erasure set heal");
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket_count = buckets.len(),
state = "started",
"Erasure set heal started"
);
// 1. generate or get task id
let task_id = self.get_or_create_task_id(set_disk_id).await?;
@@ -109,10 +124,28 @@ impl ErasureSetHealer {
// 4. cleanup resume state
if result.is_ok() {
if let Err(e) = resume_manager.cleanup().await {
warn!("Failed to cleanup resume state: {}", e);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "resume_cleanup_failed",
error = %e,
"Erasure set resume cleanup failed"
);
}
if let Err(e) = checkpoint_manager.cleanup().await {
warn!("Failed to cleanup checkpoint: {}", e);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "checkpoint_cleanup_failed",
error = %e,
"Erasure set checkpoint cleanup failed"
);
}
}
@@ -129,19 +162,47 @@ impl ErasureSetHealer {
Ok(manager) => {
let state = manager.get_state().await;
if state.set_disk_id == set_disk_id && ResumeUtils::can_resume_task(&self.disk, &task_id).await {
info!("Found resumable task: {} for set {}", task_id, set_disk_id);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
task_id,
set_disk_id,
state = "resume_found",
"Erasure set resume state selected"
);
return Ok(task_id);
}
}
Err(e) => {
warn!("Failed to load resume state for task {}: {}", task_id, e);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
task_id,
set_disk_id,
state = "resume_load_failed",
error = %e,
"Erasure set resume state load failed"
);
}
}
}
// create new task id
let task_id = format!("{}_{}", set_disk_id, ResumeUtils::generate_task_id());
info!("Created new heal task: {}", task_id);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
task_id,
set_disk_id,
state = "resume_created",
"Erasure set resume state created"
);
Ok(task_id)
}
@@ -154,7 +215,16 @@ impl ErasureSetHealer {
) -> Result<(ResumeManager, CheckpointManager)> {
// check if resume state exists
if ResumeManager::has_resume_state(&self.disk, task_id).await {
info!("Loading existing resume state for task: {}", task_id);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
task_id,
set_disk_id,
state = "loading_existing",
"Erasure set resume state loading"
);
let resume_manager = ResumeManager::load_from_disk(self.disk.clone(), task_id).await?;
let checkpoint_manager = if CheckpointManager::has_checkpoint(&self.disk, task_id).await {
@@ -165,7 +235,16 @@ impl ErasureSetHealer {
Ok((resume_manager, checkpoint_manager))
} else {
info!("Creating new resume state for task: {}", task_id);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
task_id,
set_disk_id,
state = "creating_new",
"Erasure set resume state creating"
);
let resume_manager = ResumeManager::new(
self.disk.clone(),
@@ -195,8 +274,15 @@ impl ErasureSetHealer {
let checkpoint = checkpoint_manager.get_checkpoint().await;
info!(
"Resuming from bucket {} object {}",
checkpoint.current_bucket_index, checkpoint.current_object_index
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
current_bucket_index = checkpoint.current_bucket_index,
current_object_index = checkpoint.current_object_index,
state = "resuming",
"Erasure set heal resumed from checkpoint"
);
// 2. initialize progress
@@ -247,7 +333,15 @@ impl ErasureSetHealer {
// check cancel status
if self.cancel_token.is_cancelled() {
warn!("Heal task cancelled");
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "cancelled",
"Erasure set heal cancelled"
);
return Err(Error::TaskCancelled);
}
@@ -255,10 +349,29 @@ impl ErasureSetHealer {
match bucket_result {
Ok(_) => {
resume_manager.complete_bucket(bucket).await?;
info!("Completed heal for bucket: {}", bucket);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
state = "completed",
"Erasure set bucket heal completed"
);
}
Err(e) => {
error!("Failed to heal bucket {}: {}", bucket, e);
error!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
state = "failed",
error = %e,
"Erasure set bucket heal failed"
);
// continue to next bucket, do not interrupt the whole process
}
}
@@ -270,7 +383,15 @@ impl ErasureSetHealer {
// 5. mark task completed
resume_manager.mark_completed().await?;
info!("Erasure set heal completed successfully");
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
state = "completed",
"Erasure set heal completed"
);
Ok(())
}
@@ -290,13 +411,33 @@ impl ErasureSetHealer {
resume_manager: &ResumeManager,
checkpoint_manager: &CheckpointManager,
) -> Result<()> {
info!(target: "rustfs:heal:heal_bucket_with_resume" ,"Starting heal for bucket from object index {}", current_object_index);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
bucket_index,
current_object_index = *current_object_index,
state = "started",
"Erasure set bucket heal started"
);
// 1. get bucket info
let _bucket_info = match self.storage.get_bucket_info(bucket).await? {
Some(info) => info,
None => {
warn!("Bucket {} not found, skipping", bucket);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
state = "missing",
"Erasure set bucket heal skipped because bucket is missing"
);
return Ok(());
}
};
@@ -443,14 +584,31 @@ impl ErasureSetHealer {
Ok(true) => {
*successful_objects += 1;
checkpoint_manager.add_processed_object(object.clone()).await?;
info!("Successfully healed object {}/{}", bucket, object);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
object = %object,
state = "healed",
"Erasure set object heal completed"
);
}
Ok(false) => {
checkpoint_manager.add_processed_object(object.clone()).await?;
*successful_objects += 1;
info!(
target: "rustfs:heal:heal_bucket_with_resume" ,"Object {}/{} no longer exists, skipping heal (likely deleted intentionally)",
bucket, object
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
object = %object,
state = "missing_treated_as_ok",
"Erasure set object heal skipped because object disappeared"
);
}
Err(Error::TaskCancelled) => {
@@ -465,14 +623,33 @@ impl ErasureSetHealer {
*skipped_objects += 1;
checkpoint_manager.add_skipped_object(object.clone()).await?;
warn!(
"Skipping heal for object {}/{} due to transient existence check error: {}",
bucket, object, message
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
object = %object,
state = "transient_skip",
error = %message,
"Erasure set object heal skipped due to transient error"
);
}
Err(err) => {
*failed_objects += 1;
checkpoint_manager.add_failed_object(object.clone()).await?;
warn!("Error healing object {}/{}: {}", bucket, object, err);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
object = %object,
state = "failed",
error = %err,
"Erasure set object heal failed"
);
}
}
@@ -501,7 +678,16 @@ impl ErasureSetHealer {
continuation_token = next_token;
if continuation_token.is_none() {
warn!("List is truncated but no continuation token provided for {}", bucket);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
set_disk_id,
bucket,
state = "missing_continuation_token",
"Erasure set bucket listing truncated without continuation token"
);
break;
}
}
@@ -557,13 +743,29 @@ impl ErasureSetHealer {
bucket: &str,
progress: &Arc<RwLock<HealProgress>>,
) -> Result<()> {
info!("Starting heal for bucket: {}", bucket);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "started",
"Erasure set bucket heal started"
);
// 1. get bucket info
let _bucket_info = match storage.get_bucket_info(bucket).await? {
Some(info) => info,
None => {
warn!("Bucket {} not found, skipping", bucket);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "missing",
"Erasure set bucket heal skipped because bucket is missing"
);
return Ok(());
}
};
@@ -626,7 +828,15 @@ impl ErasureSetHealer {
continuation_token = next_token;
if continuation_token.is_none() {
warn!("List is truncated but no continuation token provided for {}", bucket);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
state = "missing_continuation_token",
"Erasure set bucket listing truncated without continuation token"
);
break;
}
}
@@ -638,8 +848,16 @@ impl ErasureSetHealer {
}
info!(
"Completed heal for bucket {}: {} success, {} failures (total scanned: {})",
bucket, total_success, total_failed, total_scanned
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_BUCKET_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
total_success,
total_failed,
total_scanned,
state = "completed",
"Erasure set bucket heal completed"
);
Ok(())
@@ -672,15 +890,44 @@ impl ErasureSetHealer {
match storage.heal_object(&bucket, &object, None, &heal_opts).await {
Ok((_result, None)) => {
info!("Successfully healed object {}/{}", bucket, object);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "healed",
"Erasure set object heal completed"
);
Ok(())
}
Ok((_, Some(err))) => {
warn!("Failed to heal object {}/{}: {}", bucket, object, err);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "failed",
error = %err,
"Erasure set object heal failed"
);
Err(Error::other(err))
}
Err(err) => {
warn!("Error healing object {}/{}: {}", bucket, object, err);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_OBJECT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
bucket,
object = %object,
state = "failed",
error = %err,
"Erasure set object heal failed"
);
Err(err)
}
}
@@ -701,10 +948,27 @@ impl ErasureSetHealer {
let total = success_count + failure_count;
info!("Erasure set heal completed: {}/{} buckets successful", success_count, total);
info!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
success_count,
total,
state = "summary",
"Erasure set heal summary recorded"
);
if failure_count > 0 {
warn!("{} buckets failed to heal", failure_count);
warn!(
target: "rustfs::heal::erasure_healer",
event = EVENT_HEAL_ERASURE_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_ERASURE_HEALER,
failure_count,
state = "summary_failed",
"Erasure set heal summary recorded failures"
);
return Err(Error::other(format!("{failure_count} buckets failed to heal")));
}
crates/heal/src/heal/manager.rs
+294 -42
View File
@@ -37,6 +37,16 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
const KEEP_HEAL_TASK_STATUS_DURATION: Duration = Duration::from_secs(10 * 60);
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_DISK_SCANNER: &str = "disk_scanner";
const LOG_SUBSYSTEM_MANAGER: &str = "manager";
const EVENT_HEAL_AUTO_SCAN_STATE: &str = "heal_auto_scan_state";
const EVENT_HEAL_AUTO_SCAN_DISK: &str = "heal_auto_scan_disk";
const EVENT_HEAL_AUTO_SCAN_ENQUEUE: &str = "heal_auto_scan_enqueue";
const EVENT_HEAL_MANAGER_STATE: &str = "heal_manager_state";
const EVENT_HEAL_QUEUE_ADMISSION: &str = "heal_queue_admission";
const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state";
const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state";
/// Priority queue wrapper for heal requests
/// Uses BinaryHeap for priority-based ordering while maintaining FIFO for same-priority items
@@ -542,13 +552,27 @@ impl HealManager {
pub async fn start(&self) -> Result<()> {
let mut state = self.state.write().await;
if state.is_running {
warn!("HealManager is already running");
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "already_running",
"Heal manager state unchanged"
);
return Ok(());
}
state.is_running = true;
drop(state);
info!("Starting HealManager");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "starting",
"Heal manager state updated"
);
// start scheduler
self.start_scheduler().await?;
@@ -556,13 +580,27 @@ impl HealManager {
// start auto disk scanner to heal unformatted disks
self.start_auto_disk_scanner().await?;
info!("HealManager started successfully");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "running",
"Heal manager state updated"
);
Ok(())
}
/// Stop HealManager
pub async fn stop(&self) -> Result<()> {
info!("Stopping HealManager");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "stopping",
"Heal manager state updated"
);
// cancel all tasks
self.cancel_token.cancel();
@@ -571,7 +609,16 @@ impl HealManager {
let mut active_heals = self.active_heals.lock().await;
for task in active_heals.values() {
if let Err(e) = task.cancel().await {
warn!("Failed to cancel task {}: {}", task.id, e);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "task_cancel_failed",
task_id = %task.id,
error = %e,
"Heal manager failed to cancel active task"
);
}
}
active_heals.clear();
@@ -583,7 +630,14 @@ impl HealManager {
let mut state = self.state.write().await;
state.is_running = false;
info!("HealManager stopped successfully");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "stopped",
"Heal manager state updated"
);
Ok(())
}
@@ -605,14 +659,28 @@ impl HealManager {
match admission {
HealAdmissionResult::Merged => {
info!("Heal request already queued (duplicate merged): {}", request.id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
priority = ?request.priority,
result = "merged_duplicate",
"Heal queue admission decided"
);
}
HealAdmissionResult::Dropped(reason) => {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
priority = ?request.priority,
reason = reason.as_str(),
"Dropping duplicate heal request due to admission policy"
result = "dropped_duplicate",
"Heal queue admission decided"
);
}
HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
@@ -628,13 +696,18 @@ impl HealManager {
if let Some(displaced) = queue.push_displacing_lower_priority(request) {
publish_heal_queue_length(&queue);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request_id,
priority = ?priority,
displaced_request_id = %displaced.id,
displaced_priority = ?displaced.priority,
queue_len,
queue_capacity,
"Admitted heal request by displacing lower-priority queued work"
result = "accepted_by_displacement",
"Heal queue admission decided"
);
drop(queue);
if config.event_driven_scheduler_enable {
@@ -644,11 +717,16 @@ impl HealManager {
}
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request_id,
priority = ?priority,
queue_len,
queue_capacity,
"Heal queue was full and lower-priority work was unavailable for displacement"
result = "full_no_displacement_candidate",
"Heal queue admission decided"
);
return Ok(HealAdmissionResult::Full);
}
@@ -657,21 +735,31 @@ impl HealManager {
match admission {
HealAdmissionResult::Dropped(reason) => {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
priority = ?request.priority,
queue_len,
queue_capacity,
reason = reason.as_str(),
"Dropping heal request because the queue is full"
result = "dropped_full",
"Heal queue admission decided"
);
}
HealAdmissionResult::Full => {
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
priority = ?request.priority,
queue_len,
queue_capacity,
"Rejecting heal request because the queue is full"
result = "rejected_full",
"Heal queue admission decided"
);
}
HealAdmissionResult::Accepted | HealAdmissionResult::Merged => {}
@@ -683,10 +771,15 @@ impl HealManager {
let capacity_threshold = (queue_capacity as f64 * 0.8) as usize;
if queue_len >= capacity_threshold {
warn!(
"Heal queue is {}% full ({}/{}). Consider increasing queue size or processing capacity.",
(queue_len * 100) / queue_capacity,
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
queue_len,
queue_capacity
queue_capacity,
queue_usage_pct = (queue_len * 100) / queue_capacity,
result = "queue_pressure_high",
"Heal queue pressure increased"
);
}
@@ -701,19 +794,35 @@ impl HealManager {
if matches!(priority, HealPriority::High | HealPriority::Urgent) {
let stats = queue.get_priority_stats();
info!(
"Heal queue stats after adding {:?} priority request: total={}, urgent={}, high={}, normal={}, low={}",
priority,
queue_len + 1,
stats.get(&HealPriority::Urgent).unwrap_or(&0),
stats.get(&HealPriority::High).unwrap_or(&0),
stats.get(&HealPriority::Normal).unwrap_or(&0),
stats.get(&HealPriority::Low).unwrap_or(&0)
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request_id,
priority = ?priority,
queue_len = queue_len + 1,
urgent = *stats.get(&HealPriority::Urgent).unwrap_or(&0),
high = *stats.get(&HealPriority::High).unwrap_or(&0),
normal = *stats.get(&HealPriority::Normal).unwrap_or(&0),
low = *stats.get(&HealPriority::Low).unwrap_or(&0),
result = "accepted",
"Heal queue snapshot recorded"
);
}
drop(queue);
info!("Submitted heal request: {} with priority: {:?}", request_id, priority);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request_id,
priority = ?priority,
queue_len = queue_len + 1,
result = "accepted",
"Heal queue admission decided"
);
if config.event_driven_scheduler_enable {
self.notify.notify_one();
}
@@ -884,7 +993,15 @@ impl HealManager {
task.cancel().await?;
active_heals.remove(task_id);
publish_active_heal_count(&active_heals);
info!("Cancelled active heal task: {}", task_id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "cancelled_active_task",
"Heal manager cancelled active task"
);
return Ok(());
}
}
@@ -892,7 +1009,15 @@ impl HealManager {
let mut queue = self.heal_queue.lock().await;
if queue.remove_request_id(task_id).is_some() {
publish_heal_queue_length(&queue);
info!("Cancelled queued heal task: {}", task_id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
state = "cancelled_queued_task",
"Heal manager cancelled queued task"
);
return Ok(());
}
@@ -938,7 +1063,16 @@ impl HealManager {
});
}
info!("Cancelled {} heal task(s) for path: {}", cancelled, heal_path);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_MANAGER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
heal_path,
cancelled,
state = "cancelled_path_tasks",
"Heal manager cancelled tasks for path"
);
Ok(cancelled)
}
@@ -979,7 +1113,14 @@ impl HealManager {
let event_driven_scheduler_enable = config.read().await.event_driven_scheduler_enable;
tokio::select! {
_ = cancel_token.cancelled() => {
info!("Heal scheduler received shutdown signal");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
state = "shutdown",
"Heal scheduler state updated"
);
break;
}
_ = notify.notified(), if event_driven_scheduler_enable => {
@@ -1010,7 +1151,15 @@ impl HealManager {
if duration < Duration::from_secs(10) {
duration = Duration::from_secs(10);
}
info!("start_auto_disk_scanner: Starting auto disk scanner with interval: {:?}", duration);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "started",
interval = ?duration,
"Heal auto disk scanner state updated"
);
tokio::spawn(async move {
let mut interval = interval(duration);
@@ -1018,7 +1167,14 @@ impl HealManager {
loop {
tokio::select! {
_ = cancel_token.cancelled() => {
info!("start_auto_disk_scanner: Auto disk scanner received shutdown signal");
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "shutdown",
"Heal auto disk scanner state updated"
);
break;
}
_ = interval.tick() => {
@@ -1029,12 +1185,28 @@ impl HealManager {
// detect unformatted disk via get_disk_id()
match disk.get_disk_id().await {
Err(DiskError::UnformattedDisk) => {
info!("start_auto_disk_scanner: Detected unformatted disk: {}", disk.endpoint());
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %disk.endpoint(),
disk_state = "unformatted",
"Heal auto disk scanner detected candidate disk"
);
endpoints.push(disk.endpoint());
}
Err(e) => {
// Log other errors for debugging
tracing::warn!("start_auto_disk_scanner: Disk {} check failed: {:?}", disk.endpoint(), e);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_DISK,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %disk.endpoint(),
disk_state = "check_failed",
error = ?e,
"Heal auto disk scanner failed to inspect disk"
);
}
Ok(_) => {
// Disk is formatted, no action needed
@@ -1044,7 +1216,14 @@ impl HealManager {
}
if endpoints.is_empty() {
debug!("start_auto_disk_scanner: No endpoints need healing");
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "idle",
"Heal auto disk scanner found no candidate disks"
);
continue;
}
@@ -1052,7 +1231,15 @@ impl HealManager {
let buckets = match storage.list_buckets().await {
Ok(buckets) => buckets.iter().map(|b| b.name.clone()).collect::<Vec<String>>(),
Err(e) => {
error!("start_auto_disk_scanner: Failed to get bucket list for auto healing: {}", e);
error!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
state = "bucket_list_failed",
error = %e,
"Heal auto disk scanner failed to list buckets"
);
continue;
}
};
@@ -1062,7 +1249,15 @@ impl HealManager {
let Some(set_disk_id) =
crate::heal::utils::format_set_disk_id_from_i32(ep.pool_idx, ep.set_idx)
else {
warn!("start_auto_disk_scanner: Skipping endpoint {} without valid pool/set index", ep);
warn!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %ep,
result = "skipped_invalid_set_disk_id",
"Heal auto disk scanner skipped enqueue"
);
continue;
};
// skip if already queued or healing
@@ -1088,7 +1283,16 @@ impl HealManager {
}
if skip {
info!("start_auto_disk_scanner: Skipping auto erasure set heal for endpoint: {} (set_disk_id: {}) because it is already queued or healing", ep, set_disk_id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %ep,
set_disk_id,
result = "skipped_duplicate",
"Heal auto disk scanner skipped enqueue"
);
continue;
}
@@ -1108,7 +1312,17 @@ impl HealManager {
if config.event_driven_scheduler_enable {
notify.notify_one();
}
info!("start_auto_disk_scanner: Enqueued auto erasure set heal for endpoint: {} (set_disk_id: {})", ep, set_disk_id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_AUTO_SCAN_ENQUEUE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_DISK_SCANNER,
endpoint = %ep,
set_disk_id,
bucket_count = buckets.len(),
result = "enqueued",
"Heal auto disk scanner enqueued task"
);
}
}
}
@@ -1192,16 +1406,45 @@ impl HealManager {
// start heal task
tokio::spawn(async move {
info!(
"Starting heal task: {} with priority: {:?}, type: {}, set: {}",
task_id, task_priority, task_type_label_for_spawn, task_set_label_for_spawn
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
priority = ?task_priority,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_started",
"Heal scheduler started task"
);
let result = task.execute().await;
match result {
Ok(_) => {
info!("Heal task completed successfully: {}", task_id);
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_completed",
"Heal scheduler finished task"
);
}
Err(e) => {
error!("Heal task failed: {} - {}", task_id, e);
error!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_SCHEDULER_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
task_id,
heal_type = %task_type_label_for_spawn,
set = %task_set_label_for_spawn,
state = "task_failed",
error = %e,
"Heal scheduler observed task failure"
);
}
}
let mut active_heals_guard = active_heals_clone.lock().await;
@@ -1249,7 +1492,16 @@ impl HealManager {
if !queue.is_empty() {
let remaining = queue.len();
if remaining > 10 {
info!("Heal queue has {} pending requests, {} tasks active", remaining, active_heals_guard.len());
info!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
queue_len = remaining,
active_tasks = active_heals_guard.len(),
state = "backlog_high",
"Heal queue backlog summary recorded"
);
}
}
}
crates/heal/src/heal/resume.rs
+107 -11
View File
@@ -23,6 +23,11 @@ use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use uuid::Uuid;
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_RESUME: &str = "resume";
const EVENT_HEAL_RESUME_STATE: &str = "heal_resume_state";
const EVENT_HEAL_CHECKPOINT_STATE: &str = "heal_checkpoint_state";
/// resume state file constants
const RESUME_STATE_FILE: &str = "ahm_resume_state.json";
const RESUME_PROGRESS_FILE: &str = "ahm_progress.json";
@@ -182,7 +187,15 @@ impl ResumeManager {
// save initial state
if let Err(e) = manager.save_state().await {
warn!("Failed to save initial resume state: {}", e);
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
state = "initial_save_failed",
error = %e,
"Heal resume state persistence failed"
);
}
Ok(manager)
}
@@ -286,7 +299,15 @@ impl ResumeManager {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
info!("Cleaned up resume state for task: {}", task_id);
info!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
state = "cleaned",
"Heal resume state cleaned"
);
Ok(())
}
@@ -302,7 +323,15 @@ impl ResumeManager {
let path_str = path_to_str(&file_path)?;
if let Err(e) = self.disk.write_all(RUSTFS_META_BUCKET, path_str, state_data.into()).await {
if matches!(e, DiskError::UnformattedDisk) {
warn!("Cannot save resume state: unformatted disk");
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id = %state.task_id,
state = "skipped_unformatted_disk",
"Heal resume state persistence skipped on unformatted disk"
);
return Ok(());
}
return Err(Error::TaskExecutionFailed {
@@ -310,7 +339,15 @@ impl ResumeManager {
});
}
debug!("Saved resume state for task: {}", state.task_id);
debug!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id = %state.task_id,
state = "saved",
"Heal resume state persisted"
);
Ok(())
}
@@ -402,7 +439,15 @@ impl CheckpointManager {
// save initial checkpoint
if let Err(e) = manager.save_checkpoint().await {
warn!("Failed to save initial checkpoint: {}", e);
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_CHECKPOINT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
state = "initial_save_failed",
error = %e,
"Heal checkpoint persistence failed"
);
}
Ok(manager)
}
@@ -479,7 +524,15 @@ impl CheckpointManager {
let _ = self.disk.delete(RUSTFS_META_BUCKET, path_str, Default::default()).await;
}
info!("Cleaned up checkpoint for task: {}", task_id);
info!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_CHECKPOINT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
state = "cleaned",
"Heal checkpoint cleaned"
);
Ok(())
}
@@ -500,7 +553,15 @@ impl CheckpointManager {
message: format!("Failed to save checkpoint: {e}"),
})?;
debug!("Saved checkpoint for task: {}", checkpoint.task_id);
debug!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_CHECKPOINT_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id = %checkpoint.task_id,
state = "saved",
"Heal checkpoint persisted"
);
Ok(())
}
@@ -538,7 +599,15 @@ impl ResumeUtils {
let entries = match disk.list_dir("", RUSTFS_META_BUCKET, BUCKET_META_PREFIX, -1).await {
Ok(entries) => entries,
Err(e) => {
debug!("Failed to list resume state files: {}", e);
debug!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
state = "list_failed",
error = %e,
"Heal resume state listing failed"
);
return Ok(Vec::new());
}
};
@@ -557,7 +626,15 @@ impl ResumeUtils {
}
}
debug!("Found {} resumable tasks: {:?}", task_ids.len(), task_ids);
debug!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_count = task_ids.len(),
state = "listed",
"Heal resume states listed"
);
Ok(task_ids)
}
@@ -572,9 +649,28 @@ impl ResumeUtils {
let age_hours = (current_time - state.last_update) / 3600;
if age_hours > max_age_hours {
info!("Cleaning up expired resume state for task: {} (age: {} hours)", task_id, age_hours);
info!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
age_hours,
state = "expired_cleanup_started",
"Heal resume state cleanup started"
);
if let Err(e) = resume_manager.cleanup().await {
warn!("Failed to cleanup expired resume state for task {}: {}", task_id, e);
warn!(
target: "rustfs::heal::resume",
event = EVENT_HEAL_RESUME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RESUME,
task_id,
age_hours,
state = "expired_cleanup_failed",
error = %e,
"Heal resume state cleanup failed"
);
}
}
}
crates/heal/src/heal/storage.rs
+654 -60
View File
File diff suppressed because it is too large Load Diff
crates/heal/src/heal/task.rs
+823 -82
View File
File diff suppressed because it is too large Load Diff
crates/heal/src/lib.rs
+21 -2
View File
@@ -22,6 +22,10 @@ use std::sync::{Arc, OnceLock};
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
const EVENT_HEAL_RUNTIME_STATE: &str = "heal_runtime_state";
// Global cancellation token for heal and related services
static GLOBAL_AHM_SERVICES_CANCEL_TOKEN: OnceLock<CancellationToken> = OnceLock::new();
@@ -92,12 +96,27 @@ pub async fn init_heal_manager(
if let Some(processor_guard) = GLOBAL_HEAL_CHANNEL_PROCESSOR.get() {
let mut processor = processor_guard.lock().await;
if let Err(e) = processor.start(receiver).await {
error!("Heal channel processor failed: {}", e);
error!(
target: "rustfs::heal",
event = EVENT_HEAL_RUNTIME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "channel_processor_failed",
error = %e,
"Heal runtime channel processor failed"
);
}
}
});
info!("Heal manager with channel processor initialized successfully");
info!(
target: "rustfs::heal",
event = EVENT_HEAL_RUNTIME_STATE,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "initialized",
"Heal runtime initialized"
);
Ok(heal_manager)
}
crates/scanner/src/data_usage_define.rs
+25 -2
View File
@@ -54,6 +54,10 @@ const METRIC_CACHE_SAVE_ATTEMPT_TOTAL: &str = "rustfs_scanner_cache_save_attempt
const METRIC_CACHE_SAVE_TIMEOUT_TOTAL: &str = "rustfs_scanner_cache_save_timeout_total";
const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_total";
const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_CACHE: &str = "cache";
const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
@@ -651,7 +655,16 @@ impl DataUsageCache {
}
if retries == 5 {
warn!("maximum retry reached to load the data usage cache `{}`", name);
warn!(
target: "rustfs::scanner::data_usage",
event = EVENT_SCANNER_CACHE_LOAD_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_CACHE,
cache_name = %name,
retries,
state = "max_retries_reached",
"Scanner cache load reached retry limit"
);
}
Ok(())
@@ -891,7 +904,17 @@ impl DataUsageCache {
Self::save_path_with_retry(store, &backup_path, &buf, backup_timeout_duration, DATA_USAGE_CACHE_BACKUP_SAVE_RETRIES)
.await
{
warn!("Failed to save data usage cache backup: {e}");
warn!(
target: "rustfs::scanner::data_usage",
event = EVENT_SCANNER_CACHE_SAVE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_CACHE,
cache_name = %name,
backup_path = %backup_path,
state = "backup_save_failed",
error = %e,
"Scanner cache backup save failed"
);
}
Ok(())
}
crates/scanner/src/runtime_config.rs
+21 -2
View File
@@ -39,6 +39,10 @@ use std::sync::{LazyLock, RwLock};
use std::time::Duration;
use tracing::warn;
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_RUNTIME_CONFIG: &str = "runtime_config";
const EVENT_SCANNER_RUNTIME_CONFIG_PARSE: &str = "scanner_runtime_config_parse";
const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0;
const MAX_SCANNER_DELAY_FACTOR: f64 = 10_000.0;
@@ -274,7 +278,17 @@ fn parse_env_delay(key: &'static str, value: String, fallback: f64) -> f64 {
match parse_config_delay(key, value.clone()) {
Ok(parsed) => parsed,
Err(_) => {
warn!(env = key, value, fallback, "Invalid scanner delay config, using derived value");
warn!(
target: "rustfs::scanner::runtime_config",
event = EVENT_SCANNER_RUNTIME_CONFIG_PARSE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME_CONFIG,
env = key,
value,
fallback,
state = "invalid_delay",
"Scanner runtime config used derived delay fallback"
);
fallback
}
}
@@ -294,10 +308,15 @@ fn parse_env_bitrot_cycle(value: String) -> Option<Duration> {
"false" | "off" | "no" | "disabled" => None,
value => value.parse::<u64>().ok().map(Duration::from_secs).or_else(|| {
warn!(
target: "rustfs::scanner::runtime_config",
event = EVENT_SCANNER_RUNTIME_CONFIG_PARSE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME_CONFIG,
env = ENV_SCANNER_BITROT_CYCLE_SECS,
value,
default_secs = DEFAULT_HEAL_BITROT_CYCLE_SECS,
"Invalid scanner bitrot cycle, using default"
state = "invalid_bitrot_cycle",
"Scanner runtime config used default bitrot cycle"
);
Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS))
}),
crates/scanner/src/scanner.rs
+290 -33
View File
@@ -54,6 +54,14 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};
const SINGLE_DISK_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
const LOG_SUBSYSTEM_BACKGROUND_HEAL: &str = "background_heal";
const EVENT_SCANNER_CYCLE_STATE: &str = "scanner_cycle_state";
const EVENT_SCANNER_LOCK_STATE: &str = "scanner_lock_state";
const EVENT_SCANNER_PERSIST_STATE: &str = "scanner_persist_state";
const EVENT_SCANNER_RUNTIME_CONFIG: &str = "scanner_runtime_config";
const EVENT_SCANNER_BACKGROUND_HEAL_STATE: &str = "scanner_background_heal_state";
#[cfg(test)]
const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
@@ -82,7 +90,15 @@ fn resolve_scanner_runtime_config() -> crate::runtime_config::ScannerRuntimeConf
match lookup_scanner_runtime_config(config.as_ref()) {
Ok(config) => config,
Err(err) => {
warn!(error = %err, "Failed to resolve scanner runtime config, using last applied config");
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "resolve_failed",
error = %err,
"Scanner runtime config resolution failed; using last applied config"
);
current_scanner_runtime_config()
}
}
@@ -156,8 +172,14 @@ async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc<ECStore>) -> b
Ok(data) => data,
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "startup_inspect_failed",
error = %err,
"Failed to inspect persisted data usage cache; keeping configured scanner startup delay"
"Scanner startup cache inspection failed; keeping configured startup delay"
);
return false;
}
@@ -169,8 +191,14 @@ async fn persisted_usage_cache_is_cold_for_startup(storeapi: &Arc<ECStore>) -> b
Ok(info) => data_usage_info_is_cold(&info),
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "startup_decode_failed",
error = %err,
"Failed to decode persisted data usage cache; running initial scanner cycle without startup delay"
"Scanner startup cache decode failed; skipping startup delay"
);
true
}
@@ -188,8 +216,13 @@ async fn initial_scanner_startup_usage_state(storeapi: &Arc<ECStore>) -> (bool,
Ok(buckets) => !buckets.is_empty(),
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "startup_bucket_inspect_failed",
error = %err,
"Failed to inspect buckets for scanner startup delay; keeping configured scanner startup delay"
"Scanner startup bucket inspection failed; keeping configured startup delay"
);
false
}
@@ -207,7 +240,15 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
// Force init global sleeper so config is read once at startup.
let _ = &*SCANNER_SLEEPER;
if let Err(err) = refresh_scanner_runtime_config_from_global() {
warn!(error = %err, "Failed to apply scanner runtime config at startup");
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "startup_apply_failed",
error = %err,
"Scanner runtime config apply failed at startup"
);
}
let ctx_clone = ctx;
@@ -220,7 +261,15 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
has_buckets,
);
if sleep_time.is_zero() {
info!("Skipping initial data scanner delay because persisted data usage cache is cold");
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "startup_delay_skipped",
reason = "usage_cache_cold",
"Scanner startup delay skipped"
);
} else {
tokio::time::sleep(sleep_time).await;
}
@@ -231,7 +280,15 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
}
if let Err(e) = run_data_scanner(ctx_clone.clone(), storeapi_clone.clone()).await {
error!("Failed to run data scanner: {e}");
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "run_failed",
error = %e,
"Scanner runtime iteration failed"
);
}
// Backoff before retrying after lock contention or scanner-level failures.
// Keep this cancellation-aware so shutdown is not delayed by backoff sleep.
@@ -276,8 +333,13 @@ async fn detect_scanner_maintenance_features(storeapi: &Arc<ECStore>) -> Scanner
Ok(buckets) => buckets,
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "maintenance_feature_inspect_failed",
error = %err,
"Failed to inspect scanner maintenance features; preserving speed-based scanner cycle"
"Scanner maintenance feature inspection failed; preserving speed-based cycle"
);
features.inspection_failed = true;
return features;
@@ -293,9 +355,14 @@ async fn detect_scanner_maintenance_features(storeapi: &Arc<ECStore>) -> Scanner
Err(EcstoreError::ConfigNotFound) => {}
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
bucket = %bucket.name,
state = "lifecycle_inspect_failed",
error = %err,
"Failed to inspect bucket lifecycle config; preserving speed-based scanner cycle"
"Scanner lifecycle inspection failed; preserving speed-based cycle"
);
features.inspection_failed = true;
}
@@ -310,9 +377,14 @@ async fn detect_scanner_maintenance_features(storeapi: &Arc<ECStore>) -> Scanner
Err(EcstoreError::ConfigNotFound) => {}
Err(err) => {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
bucket = %bucket.name,
state = "replication_inspect_failed",
error = %err,
"Failed to inspect bucket replication config; preserving speed-based scanner cycle"
"Scanner replication inspection failed; preserving speed-based cycle"
);
features.inspection_failed = true;
}
@@ -334,6 +406,10 @@ async fn configure_scanner_defaults(storeapi: &Arc<ECStore>) {
set_scanner_default_speed(ScannerSpeed::Slowest);
set_scanner_default_cycle_secs(default_cycle_secs);
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
env_speed = ENV_SCANNER_SPEED,
env_cycle = ENV_SCANNER_CYCLE,
env_start_delay = ENV_SCANNER_START_DELAY_SECS,
@@ -341,7 +417,8 @@ async fn configure_scanner_defaults(storeapi: &Arc<ECStore>) {
lifecycle_active = features.lifecycle,
replication_active = features.replication,
feature_inspection_failed = features.inspection_failed,
"Using slower scanner defaults for single-disk deployments; regular scanner cycles are preserved when maintenance features are active"
state = "single_disk_defaults_applied",
"Scanner defaults updated for single-disk deployment"
);
} else {
set_scanner_default_speed(ScannerSpeed::Default);
@@ -477,13 +554,31 @@ pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> BackgroundHeal
// Get last healing information
match read_config(storeapi, &BACKGROUND_HEAL_INFO_PATH).await {
Ok(buf) => serde_json::from_slice::<BackgroundHealInfo>(&buf).unwrap_or_else(|e| {
error!("Failed to unmarshal background heal info from {}: {}", &*BACKGROUND_HEAL_INFO_PATH, e);
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "decode_failed",
error = %e,
"Scanner background heal state decode failed"
);
BackgroundHealInfo::default()
}),
Err(e) => {
// Only log if it's not a ConfigNotFound error
if e != EcstoreError::ConfigNotFound {
warn!("Failed to read background heal info from {}: {}", &*BACKGROUND_HEAL_INFO_PATH, e);
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "read_failed",
error = %e,
"Scanner background heal state read failed"
);
}
BackgroundHealInfo::default()
}
@@ -502,14 +597,32 @@ pub async fn save_background_heal_info(storeapi: Arc<ECStore>, info: BackgroundH
let data = match serde_json::to_vec(&info) {
Ok(data) => data,
Err(e) => {
error!("Failed to marshal background heal info: {}", e);
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "encode_failed",
error = %e,
"Scanner background heal state encode failed"
);
return;
}
};
// Save configuration
if let Err(e) = save_config(storeapi, &BACKGROUND_HEAL_INFO_PATH, data).await {
warn!("Failed to save background heal info to {}: {}", &*BACKGROUND_HEAL_INFO_PATH, e);
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_BACKGROUND_HEAL_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_BACKGROUND_HEAL,
path = %&*BACKGROUND_HEAL_INFO_PATH,
state = "save_failed",
error = %e,
"Scanner background heal state save failed"
);
}
}
@@ -530,7 +643,15 @@ async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) {
async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>, cycle_info: &mut CurrentCycle) {
let _activity_guard = ScannerActivityGuard::new();
if let Err(err) = refresh_scanner_runtime_config_from_global() {
warn!(error = %err, "Failed to refresh scanner runtime config, using last applied config");
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_RUNTIME_CONFIG,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "refresh_failed",
error = %err,
"Scanner runtime config refresh failed; using last applied config"
);
}
let configured_cycle_interval = scanner_cycle_interval();
let configured_bitrot_cycle = scanner_bitrot_cycle();
@@ -542,7 +663,6 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
cycle_budget_config.max_objects,
cycle_budget_config.max_directories,
);
info!("Start run data scanner cycle");
cycle_info.current = cycle_info.next;
let now = Instant::now();
cycle_info.started = Utc::now();
@@ -557,6 +677,16 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
background_heal_info.bitrot_start_time,
configured_bitrot_cycle,
);
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
scan_mode = ?scan_mode,
state = "started",
"Scanner cycle state updated"
);
let _scan_mode_guard = ScannerScanModeGuard::new(scan_mode);
if let Some(new_heal_info) = background_heal_info_for_scan_start(
background_heal_info.clone(),
@@ -589,12 +719,18 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
global_metrics().finish_scan_cycle_work(cycle_work_start);
if budget_elapsed {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
duration = ?now.elapsed(),
reason = ?cycle_budget.reason(),
max_duration = ?cycle_budget.max_duration(),
max_objects = ?cycle_budget.max_objects(),
max_directories = ?cycle_budget.max_directories(),
"Data scanner cycle stopped after reaching its cycle budget"
state = "budget_reached",
"Scanner cycle stopped after reaching budget"
);
let budget_reason = cycle_budget.reason();
emit_scan_cycle_partial_with_source(
@@ -605,7 +741,18 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
mark_scan_cycle_idle(cycle_info).await;
return;
}
error!(duration = ?now.elapsed(), "Fail run data scanner cycle: {e}");
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
scan_mode = ?scan_mode,
state = "failed",
duration = ?now.elapsed(),
error = %e,
"Scanner cycle state updated"
);
emit_scan_cycle_complete(false, cycle_start.elapsed());
if let Some(new_heal_info) = background_heal_info_for_scan_complete(background_heal_info.clone(), scan_mode) {
save_background_heal_info(storeapi.clone(), new_heal_info).await;
@@ -614,12 +761,18 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
}
if cycle_budget.budget_elapsed() && !ctx.is_cancelled() {
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
duration = ?now.elapsed(),
reason = ?cycle_budget.reason(),
max_duration = ?cycle_budget.max_duration(),
max_objects = ?cycle_budget.max_objects(),
max_directories = ?cycle_budget.max_directories(),
"Data scanner cycle stopped after reaching its cycle budget"
state = "budget_reached",
"Scanner cycle stopped after reaching budget"
);
global_metrics().finish_scan_cycle_work(cycle_work_start);
let budget_reason = cycle_budget.reason();
@@ -643,7 +796,18 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
cycle_info.cycle_completed.push(Utc::now());
global_metrics().clear_current_scan_mode();
info!(duration = ?now.elapsed(), cycles_total=cycle_info.cycle_completed.len(), "Success run data scanner cycle");
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
scan_mode = ?scan_mode,
state = "completed",
duration = ?now.elapsed(),
cycles_total = cycle_info.cycle_completed.len(),
"Scanner cycle state updated"
);
retain_recent_cycle_completions(&mut cycle_info.cycle_completed);
global_metrics().set_cycle(Some(cycle_info.clone())).await;
@@ -655,9 +819,26 @@ async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc<ECStore>
buf.extend_from_slice(&cycle_info_buf);
if let Err(e) = save_config(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf).await {
error!("Failed to save data usage bloom name to {}: {}", &*DATA_USAGE_BLOOM_NAME_PATH, e);
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "failed",
error = %e,
"Scanner state persistence failed"
);
} else {
info!("Data usage bloom name saved successfully");
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "saved",
"Scanner state persisted"
);
}
}
@@ -666,16 +847,42 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
let _guard = match storeapi.new_ns_lock(RUSTFS_META_BUCKET, "leader.lock").await {
Ok(ns_lock) => match ns_lock.get_write_lock_quiet(get_lock_acquire_timeout()).await {
Ok(guard) => {
debug!("run_data_scanner: acquired leader write lock");
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_LOCK_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
lock_name = "leader.lock",
state = "acquired",
"Scanner leader lock state updated"
);
guard
}
Err(e) => {
debug!("run_data_scanner: other node is running, failed to acquire leader write lock: {:?}", e);
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_LOCK_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
lock_name = "leader.lock",
state = "contended",
error = ?e,
"Scanner leader lock state updated"
);
return Ok(());
}
},
Err(e) => {
error!("run_data_scanner: failed to create namespace lock: {e}");
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_LOCK_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
lock_name = "leader.lock",
state = "create_failed",
error = %e,
"Scanner leader lock state updated"
);
return Ok(());
}
};
@@ -689,7 +896,16 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
} else if buf.len() > 8 {
cycle_info.next = u64::from_le_bytes(buf[0..8].try_into().unwrap_or_default());
if let Err(e) = cycle_info.unmarshal(&buf[8..]) {
warn!("Failed to unmarshal cycle info: {e}");
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %&*DATA_USAGE_BLOOM_NAME_PATH,
state = "cycle_decode_failed",
error = %e,
"Scanner cycle state decode failed"
);
}
}
@@ -714,7 +930,14 @@ pub async fn run_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) ->
global_metrics().set_cycle(None).await;
debug!("Data scanner done");
debug!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "stopped",
"Scanner runtime stopped"
);
Ok(())
}
@@ -755,8 +978,15 @@ pub async fn store_data_usage_in_backend(
&& new_ts <= existing_ts
{
info!(
"Skip persisting data usage: incoming last_update {:?} <= existing {:?}",
new_ts, existing_ts
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
incoming_last_update = ?new_ts,
existing_last_update = ?existing_ts,
state = "skip_stale_update",
"Scanner data usage persistence skipped stale update"
);
continue;
}
@@ -765,7 +995,16 @@ pub async fn store_data_usage_in_backend(
let data = match serde_json::to_vec(&data_usage_info) {
Ok(data) => data,
Err(e) => {
error!("Failed to marshal data usage info: {}", e);
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "encode_failed",
error = %e,
"Scanner data usage persistence encode failed"
);
continue;
}
};
@@ -775,7 +1014,16 @@ pub async fn store_data_usage_in_backend(
let backup_path = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str());
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = save_config(storeapi.clone(), &backup_path, data.clone()).await {
warn!("Failed to save data usage backup to {}: {}", backup_path, e);
warn!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %backup_path,
state = "backup_save_failed",
error = %e,
"Scanner data usage backup save failed"
);
}
done_save();
attempts = 1;
@@ -784,7 +1032,16 @@ pub async fn store_data_usage_in_backend(
// Save main configuration
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = save_config(storeapi.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str(), data).await {
error!("Failed to save data usage info to {}: {e}", DATA_USAGE_OBJ_NAME_PATH.as_str());
error!(
target: "rustfs::scanner",
event = EVENT_SCANNER_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
path = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
state = "save_failed",
error = %e,
"Scanner data usage persistence failed"
);
} else {
rustfs_ecstore::data_usage::replace_bucket_usage_memory_from_info(&data_usage_info).await;
}
crates/scanner/src/scanner_folder.rs
+417 -47
View File
@@ -68,6 +68,16 @@ use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_FOLDER: &str = "folder";
const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle";
const LOG_SUBSYSTEM_HEAL: &str = "heal";
const EVENT_SCANNER_FOLDER_STATE: &str = "scanner_folder_state";
const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action";
const EVENT_SCANNER_HEAL_ADMISSION: &str = "scanner_heal_admission";
const EVENT_SCANNER_ALERT_STATE: &str = "scanner_alert_state";
const EVENT_SCANNER_COMPAT_STATE: &str = "scanner_compat_state";
const DATA_USAGE_UPDATE_DIR_CYCLES: u32 = 16;
const DATA_SCANNER_COMPACT_LEAST_OBJECT: usize = 500;
const DATA_SCANNER_COMPACT_AT_CHILDREN: usize = 10000;
@@ -361,8 +371,13 @@ fn warn_inline_heal_compat_requested() {
SCANNER_INLINE_HEAL_WARN_ONCE.call_once(|| {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_COMPAT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
env = rustfs_config::ENV_SCANNER_INLINE_HEAL_ENABLE,
"Inline scanner heal rollback is no longer supported; continuing to enqueue heal candidates asynchronously"
state = "inline_heal_rollback_unsupported",
"Scanner inline-heal rollback is unsupported; using async heal admission"
);
});
}
@@ -541,13 +556,18 @@ async fn send_required_scanner_heal_request(
record_high_priority_heal_escalation(candidate_type, priority, result);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_HEAL_ADMISSION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
candidate_type,
bucket,
object = object.unwrap_or(""),
priority = heal_priority_label(priority),
admission = result.result_label(),
reason = result.reason_label(),
"High-priority heal request was not admitted; escalating to scanner failure"
state = "high_priority_not_admitted",
"Scanner high-priority heal admission failed"
);
Err(build_high_priority_heal_admission_error(candidate_type, bucket, object, priority, result))
}
@@ -601,15 +621,39 @@ impl ScannerItem {
size_summary: &mut SizeSummary,
) {
if object_infos.is_empty() {
debug!("apply_actions: no object infos for object: {}", self.object_path());
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
object_path = %self.object_path(),
state = "no_object_versions",
"Scanner lifecycle action skipped"
);
return;
}
debug!("apply_actions: applying actions for object: {}", self.object_path());
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
object_path = %self.object_path(),
state = "started",
"Scanner lifecycle action evaluation started"
);
let versioning_config = match BucketVersioningSys::get(&self.bucket).await {
Ok(versioning_config) => versioning_config,
Err(_) => {
warn!("apply_actions: Failed to get versioning configuration for bucket {}", self.bucket);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
state = "versioning_lookup_failed",
"Scanner lifecycle action skipped"
);
return;
}
};
@@ -620,7 +664,16 @@ impl ScannerItem {
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!("apply_actions: Failed to get actual size for object {}", oi.name);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
continue;
}
};
@@ -634,7 +687,15 @@ impl ScannerItem {
self.alert_excessive_versions(object_infos.len(), cumulative_size);
debug!("apply_actions: done for now no lifecycle config");
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
object_path = %self.object_path(),
state = "no_lifecycle_config",
"Scanner lifecycle action finished without lifecycle rules"
);
return;
};
@@ -651,7 +712,16 @@ impl ScannerItem {
{
Ok(events) => events,
Err(e) => {
warn!("apply_actions: Failed to evaluate lifecycle for object: {}", e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
object_path = %self.object_path(),
state = "evaluate_failed",
error = %e,
"Scanner lifecycle action evaluation failed"
);
return;
}
};
@@ -666,7 +736,16 @@ impl ScannerItem {
let actual_size = match oi.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!("apply_actions: Failed to get actual size for object {}", oi.name);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
}
};
@@ -676,7 +755,17 @@ impl ScannerItem {
match event.action {
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action);
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
action = %event.action,
state = "apply_expiry_rule",
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
@@ -693,7 +782,16 @@ impl ScannerItem {
let retained_size = match retained.get_actual_size() {
Ok(size) => size,
Err(_) => {
warn!("apply_actions: Failed to get actual size for object {}", retained.name);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %retained.name,
state = "size_lookup_failed",
"Scanner lifecycle action used fallback size"
);
0
}
};
@@ -709,7 +807,17 @@ impl ScannerItem {
}
IlmAction::DeleteAction | IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => {
debug!("apply_actions: applying expiry rule for object: {} {}", oi.name, event.action);
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
action = %event.action,
state = "apply_expiry_rule",
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let queued = apply_expiry_rule(event, &LcEventSrc::Scanner, oi).await;
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
@@ -737,7 +845,17 @@ impl ScannerItem {
noncurrent_events.push(event.clone());
}
IlmAction::TransitionAction | IlmAction::TransitionVersionAction => {
debug!("apply_actions: applying transition rule for object: {} {}", oi.name, event.action);
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_LIFECYCLE_ACTION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_LIFECYCLE,
bucket = %self.bucket,
object = %oi.name,
action = %event.action,
state = "apply_transition_rule",
"Scanner lifecycle action dispatched"
);
let done_ilm = Metrics::time_ilm(event.action);
let queued = apply_transition_rule(event, &LcEventSrc::Scanner, oi).await;
if record_scanner_ilm_action_if_queued(global_metrics(), event.action, 1, queued) {
@@ -853,10 +971,15 @@ impl ScannerItem {
async fn enqueue_heal(&mut self, oi: &ObjectInfo) {
let done_heal = Metrics::time(Metric::HealAbandonedObject);
debug!(
"enqueue_heal: bucket: {}, object_path: {}, version_id: {}",
self.bucket,
self.object_path(),
oi.version_id.unwrap_or_default()
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_HEAL_ADMISSION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket = %self.bucket,
object = %self.object_path(),
version_id = %oi.version_id.unwrap_or_default(),
state = "request_started",
"Scanner heal admission started"
);
let now = OffsetDateTime::now_utc();
@@ -868,6 +991,10 @@ impl ScannerItem {
age.whole_seconds().max(0)
});
info!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_HEAL_ADMISSION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket = %self.bucket,
object = %self.object_path(),
version_id = %oi.version_id.unwrap_or_default(),
@@ -875,7 +1002,8 @@ impl ScannerItem {
cooldown_secs = cooldown.as_secs(),
original_scan_mode = %HealScanMode::Deep.as_str(),
effective_scan_mode = %scan_mode.as_str(),
"scanner deep heal downgraded to normal during new-object cooldown"
state = "downgraded_to_normal",
"Scanner heal admission downgraded deep scan during cooldown"
);
}
@@ -898,13 +1026,28 @@ impl ScannerItem {
Ok(HealAdmissionResult::Accepted | HealAdmissionResult::Merged) => {}
Ok(result @ (HealAdmissionResult::Full | HealAdmissionResult::Dropped(_))) => {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_HEAL_ADMISSION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket = %self.bucket,
object = %self.object_path(),
admission = %describe_heal_admission(result),
"enqueue_heal: low-priority heal request was not admitted"
state = "not_admitted",
"Scanner heal admission rejected low-priority request"
);
}
Err(e) => warn!("enqueue_heal: failed to submit heal request: {}", e),
Err(e) => warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_HEAL_ADMISSION,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_HEAL,
bucket = %self.bucket,
object = %self.object_path(),
state = "submit_failed",
error = %e,
"Scanner heal admission submission failed"
),
}
if admitted {
done_heal();
@@ -922,11 +1065,16 @@ impl ScannerItem {
)
.increment(1);
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(),
versions = remaining_versions,
threshold = scanner_excess_versions_threshold(),
"scanner detected object with excessive retained versions"
state = "excess_versions",
"Scanner alert recorded excessive retained versions"
);
}
if too_large_versions {
@@ -937,12 +1085,17 @@ impl ScannerItem {
)
.increment(1);
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(),
versions = remaining_versions,
cumulative_size,
threshold = scanner_excess_version_size_threshold(),
"scanner detected object with excessive retained version size"
state = "excess_version_size",
"Scanner alert recorded excessive retained version size"
);
}
}
@@ -1077,11 +1230,16 @@ impl FolderScanner {
)
.increment(1);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_ALERT_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
root = %self.root,
folder,
folders = total_folders,
threshold,
"scanner detected folder with excessive direct subfolders"
state = "excess_folders",
"Scanner alert recorded excessive direct subfolders"
);
}
@@ -1140,7 +1298,16 @@ impl FolderScanner {
{
// Try to send without blocking
if let Err(e) = updates.send(flat.clone()).await {
error!("send_update: failed to send update: {}", e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
root = %self.new_cache.info.name,
state = "update_send_failed",
error = %e,
"Scanner folder update send failed"
);
}
self.last_update = SystemTime::now();
}
@@ -1190,7 +1357,16 @@ impl FolderScanner {
abandoned_children = self.old_cache.find_children_copy(this_hash.clone());
}
debug!("scan_folder : {}/{}", &self.root, &folder.name);
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
root = %self.root,
folder = %folder.name,
state = "scan_started",
"Scanner folder state updated"
);
let (_, prefix) = path2_bucket_object_with_base_path(&self.root, &folder.name);
let active_life_cycle = if self
@@ -1224,12 +1400,29 @@ impl FolderScanner {
let dir_path = path_join_buf(&[&self.root, &folder.name]);
debug!("scan_folder: dir_path: {:?}", dir_path);
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = ?dir_path,
state = "dir_open",
"Scanner folder state updated"
);
let mut dir_reader = match tokio::fs::read_dir(&dir_path).await {
Ok(dir_reader) => dir_reader,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: directory disappeared before read {}: {}", dir_path, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_missing_before_read",
error = %e,
"Scanner folder state updated"
);
return Ok(());
}
Err(e) => return Err(ScannerError::Io(e)),
@@ -1240,11 +1433,29 @@ impl FolderScanner {
Ok(Some(entry)) => entry,
Ok(None) => break,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: directory disappeared during iteration {}: {}", dir_path, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_missing_during_iteration",
error = %e,
"Scanner folder state updated"
);
break;
}
Err(e) if e.kind() == ErrorKind::NotADirectory => {
warn!("scan_folder: path became non-directory during iteration {}: {}", dir_path, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
dir_path = %dir_path,
state = "dir_became_non_directory",
error = %e,
"Scanner folder state updated"
);
break;
}
Err(e) => return Err(ScannerError::Io(e)),
@@ -1269,11 +1480,29 @@ impl FolderScanner {
let mut entry_type = match entry.file_type().await {
Ok(entry_type) => entry_type,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: entry disappeared before type lookup {}: {}", entry_name, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
entry = %entry_name,
state = "entry_missing_before_type_lookup",
error = %e,
"Scanner folder state updated"
);
continue;
}
Err(e) if e.kind() == ErrorKind::TooManyLinks => {
warn!("scan_folder: entry hit symlink loop before type lookup {}: {}", entry_name, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
entry = %entry_name,
state = "entry_symlink_loop_before_type_lookup",
error = %e,
"Scanner folder state updated"
);
continue;
}
Err(e) => return Err(ScannerError::Io(e)),
@@ -1283,18 +1512,44 @@ impl FolderScanner {
let metadata = match tokio::fs::metadata(&file_path).await {
Ok(metadata) => metadata,
Err(e) if e.kind() == ErrorKind::NotFound => {
warn!("scan_folder: symlink target disappeared before metadata lookup {}: {}", file_path, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
file_path = %file_path,
state = "symlink_target_missing_before_metadata",
error = %e,
"Scanner folder state updated"
);
continue;
}
Err(e) if e.kind() == ErrorKind::TooManyLinks => {
warn!("scan_folder: symlink target hit loop before metadata lookup {}: {}", file_path, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
file_path = %file_path,
state = "symlink_target_loop_before_metadata",
error = %e,
"Scanner folder state updated"
);
continue;
}
Err(e) => return Err(ScannerError::Io(e)),
};
if metadata.is_dir() {
warn!("scan_folder: ignoring symlinked directory {}", file_path);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
file_path = %file_path,
state = "symlink_directory_ignored",
"Scanner folder state updated"
);
continue;
}
@@ -1384,8 +1639,15 @@ impl FolderScanner {
if should_log_failed_object(into.failed_objects) {
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
path = %item.path,
failed_objects = into.failed_objects,
"scan_folder: failed to get size for item {}: {}", item.path, e
state = "get_size_failed",
error = %e,
"Scanner folder failed to get object size"
);
}
}
@@ -1426,7 +1688,15 @@ impl FolderScanner {
if found_objects && is_erasure().await {
// If we found an object in erasure mode, we skip subdirs (only datadirs)...
info!("scan_folder: done for now found an object in erasure mode");
info!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
state = "erasure_object_found",
"Scanner folder stopped descending after finding erasure object"
);
break;
}
@@ -1445,7 +1715,16 @@ impl FolderScanner {
existing_folders.clear();
if self.data_usage_scanner_debug {
debug!("scan_folder: Preemptively compacting: {}, entries: {}", folder.name, new_folders.len());
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
entry_count = new_folders.len(),
state = "preemptive_compaction",
"Scanner folder switched to compacted mode"
);
}
}
@@ -1570,7 +1849,17 @@ impl FolderScanner {
if ctx.is_cancelled() {
return Err(e);
}
warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
child_folder = %folder_item.name,
state = "child_scan_failed",
error = %e,
"Scanner child folder scan failed"
);
continue;
}
tokio::task::yield_now().await;
@@ -1596,13 +1885,31 @@ impl FolderScanner {
// Scan for healing
if abandoned_children.is_empty() || !self.should_heal().await {
debug!("scan_folder: done for now abandoned children are empty or we are not healing");
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
state = "heal_skip_no_abandoned_children",
"Scanner folder skipped heal scan for abandoned children"
);
// If we are not heal scanning, return now.
break;
}
if self.disks.is_empty() || self.disks_quorum == 0 {
debug!("scan_folder: done for now disks are empty or quorum is 0");
debug!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
disks = self.disks.len(),
quorum = self.disks_quorum,
state = "heal_skip_no_quorum",
"Scanner folder skipped heal scan because quorum is unavailable"
);
break;
}
@@ -1660,7 +1967,16 @@ impl FolderScanner {
let agreed_tx = agreed_tx.clone();
Box::pin(async move {
if let Err(e) = agreed_tx.send(entry_name).await {
error!("scan_folder: list_path_raw: failed to send entry name: {}: {}", entry.name, e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
entry = %entry.name,
state = "list_path_agreed_send_failed",
error = %e,
"Scanner list_path_raw agreed callback failed"
);
}
})
})),
@@ -1668,7 +1984,15 @@ impl FolderScanner {
let partial_tx = partial_tx.clone();
Box::pin(async move {
if let Err(e) = partial_tx.send(entries).await {
error!("scan_folder: list_path_raw: failed to send partial err: {}", e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
state = "list_path_partial_send_failed",
error = %e,
"Scanner list_path_raw partial callback failed"
);
}
})
})),
@@ -1677,7 +2001,15 @@ impl FolderScanner {
let errs_clone = errs.to_vec();
Box::pin(async move {
if let Err(e) = finished_tx.send(errs_clone).await {
error!("scan_folder: list_path_raw: failed to send finished errs: {}", e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
state = "list_path_finished_send_failed",
error = %e,
"Scanner list_path_raw finished callback failed"
);
}
})
})),
@@ -1686,7 +2018,17 @@ impl FolderScanner {
)
.await
{
error!("scan_folder: failed to list path: {}/{}: {}", bucket_clone, prefix_clone, e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %bucket_clone,
prefix = %prefix_clone,
state = "list_path_failed",
error = %e,
"Scanner list_path_raw failed"
);
}
});
@@ -1745,7 +2087,17 @@ impl FolderScanner {
let fivs = match entry.file_info_versions(&bucket) {
Ok(fivs) => fivs,
Err(e) => {
error!("scan_folder: list_path_raw: failed to get file info versions: {}", e);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
bucket = %bucket,
entry = %entry.name,
state = "file_info_versions_failed",
error = %e,
"Scanner list_path_raw failed to resolve file versions"
);
send_required_scanner_heal_request(
"object",
&bucket,
@@ -1789,7 +2141,15 @@ impl FolderScanner {
finished_closed = true;
continue;
};
error!("scan_folder: list_path_raw: failed to get finished errs: {:?}", errs);
error!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
state = "list_path_finished_with_errors",
errors = ?errs,
"Scanner list_path_raw finished with disk errors"
);
child_ctx.cancel();
}
_ = child_ctx.cancelled() => {
@@ -1820,7 +2180,17 @@ impl FolderScanner {
if ctx.is_cancelled() {
return Err(e);
}
warn!("scan_folder: failed to scan child folder {}: {}", folder_item.name, e);
warn!(
target: "rustfs::scanner::folder",
event = EVENT_SCANNER_FOLDER_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_FOLDER,
folder = %folder.name,
child_folder = %folder_item.name,
state = "heal_child_scan_failed",
error = %e,
"Scanner heal child folder scan failed"
);
continue;
}
tokio::task::yield_now().await;
crates/scanner/src/scanner_io.rs
+328 -34
View File
@@ -59,6 +59,12 @@ use tokio_util::sync::CancellationToken;
use tracing::{debug, error, warn};
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_IO: &str = "io";
const EVENT_SCANNER_DISK_BUCKET_STATE: &str = "scanner_disk_bucket_state";
const EVENT_SCANNER_DATA_USAGE_STREAM: &str = "scanner_data_usage_stream";
const EVENT_SCANNER_CACHE_PERSIST_STATE: &str = "scanner_cache_persist_state";
const EVENT_SCANNER_SET_STATE: &str = "scanner_set_state";
const METRIC_SCANNER_SET_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_set_scan_concurrency_limit";
const METRIC_SCANNER_DISK_SCAN_CONCURRENCY_LIMIT: &str = "rustfs_scanner_disk_scan_concurrency_limit";
@@ -305,12 +311,30 @@ async fn persist_and_publish_cache_snapshot<S: ObjectIO>(
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = cache_snapshot.save(store, DATA_USAGE_CACHE_NAME).await {
error!("Failed to save data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
cache_name = DATA_USAGE_CACHE_NAME,
state = "save_failed",
error = %e,
"Scanner cache snapshot persistence failed"
);
}
done_save();
if let Err(e) = updates.send(cache_snapshot).await {
error!("Failed to send data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
cache_name = DATA_USAGE_CACHE_NAME,
state = "publish_failed",
error = %e,
"Scanner cache snapshot publish failed"
);
}
last_update
@@ -379,7 +403,15 @@ impl ScannerIO for ECStore {
if all_buckets.is_empty() {
reset_set_scan_gauges();
if let Err(e) = updates.send(DataUsageInfo::default()).await {
error!("Failed to send data usage info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "empty_bucket_publish_failed",
error = %e,
"Scanner set state update failed"
);
}
return Ok(());
}
@@ -389,7 +421,15 @@ impl ScannerIO for ECStore {
total_results += pool.disk_set.len();
}
if total_results == 0 {
warn!("nsscanner: no disk sets available for non-empty bucket list");
warn!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket_count = all_buckets.len(),
state = "no_disk_sets",
"Scanner set state update detected missing disk sets"
);
reset_set_scan_gauges();
return Ok(());
}
@@ -397,9 +437,14 @@ impl ScannerIO for ECStore {
let set_scan_limit = scanner_max_concurrent_set_scans(total_results);
record_set_scan_concurrency_limit(set_scan_limit);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
total_sets = total_results,
concurrency_limit = set_scan_limit,
"nsscanner: scanner set concurrency budget"
state = "concurrency_budget",
"Scanner set concurrency budget resolved"
);
let set_scan_semaphore = Arc::new(Semaphore::new(set_scan_limit));
let queued_set_scans = Arc::new(AtomicUsize::new(total_results));
@@ -495,10 +540,15 @@ impl ScannerIO for ECStore {
)
.increment(1);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
pool = %pool_label,
set = %set_label,
error = %e,
"Failed to scan set; continuing scanner cycle"
state = "set_scan_failed",
"Scanner set scan failed; continuing cycle"
);
let mut first_err = first_err_mutex_clone.lock().await;
record_set_scan_failure(&mut first_err, e);
@@ -540,7 +590,15 @@ impl ScannerIO for ECStore {
if all_merged.root().is_some() && (!has_sent_once || merged_last_update > last_update) {
let dui = all_merged.dui(&all_merged.info.name, &all_buckets_clone);
if let Err(e) = updates.send(dui).await {
error!("Failed to send data usage info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "send_merged_failed",
error = %e,
"Scanner merged data usage publish failed"
);
}
}
break;
@@ -559,7 +617,15 @@ impl ScannerIO for ECStore {
if all_merged.root().is_some() && (!has_sent_once || merged_last_update > last_update) {
let dui = all_merged.dui(&all_merged.info.name, &all_buckets_clone);
if let Err(e) = updates.send(dui).await {
error!("Failed to send data usage info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "send_merged_failed",
error = %e,
"Scanner merged data usage publish failed"
);
}
has_sent_once = true;
last_update = merged_last_update;
@@ -604,18 +670,32 @@ impl ScannerIOCache for SetDisks {
let (disks, healing) = self.get_online_disks_with_healing(false).await;
if disks.is_empty() {
debug!("nsscanner_cache: no online disks available for set");
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
pool = self.pool_index,
set = self.set_index,
state = "no_online_disks",
"Scanner set state found no online disks"
);
reset_disk_bucket_scan_gauges(&pool_label, &set_label);
return Ok(());
}
let disk_scan_limit = scanner_max_concurrent_disk_scans(disks.len());
record_disk_scan_concurrency_limit(&pool_label, &set_label, disk_scan_limit);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
pool = self.pool_index,
set = self.set_index,
online_disks = disks.len(),
concurrency_limit = disk_scan_limit,
"nsscanner_cache: scanner disk concurrency budget"
state = "disk_concurrency_budget",
"Scanner disk concurrency budget resolved"
);
let disk_scan_semaphore = Arc::new(Semaphore::new(disk_scan_limit));
let queued_disk_bucket_scans = Arc::new(AtomicUsize::new(buckets.len()));
@@ -646,7 +726,16 @@ impl ScannerIOCache for SetDisks {
if old_cache.find(&bucket.name).is_none()
&& let Err(e) = bucket_tx.send(bucket.clone()).await
{
error!("Failed to send bucket info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "send_bucket_failed",
error = %e,
"Scanner bucket dispatch failed"
);
}
}
@@ -658,7 +747,16 @@ impl ScannerIOCache for SetDisks {
}
if let Err(e) = bucket_tx.send(bucket.clone()).await {
error!("Failed to send bucket info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_SET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "send_bucket_failed",
error = %e,
"Scanner bucket dispatch failed"
);
}
}
}
@@ -795,13 +893,31 @@ impl ScannerIOCache for SetDisks {
set_label_clone.clone(),
);
debug!("nsscanner_disk: got bucket: {}", bucket.name);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "scan_started",
"Scanner disk bucket state updated"
);
let cache_name = path_join_buf(&[&bucket.name, DATA_USAGE_CACHE_NAME]);
let mut cache = DataUsageCache::default();
if let Err(e) = cache.load(store_clone_clone.clone(), &cache_name).await {
error!("Failed to load data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "cache_load_failed",
error = %e,
"Scanner disk bucket state updated"
);
}
if cache.info.name.is_empty() {
@@ -818,7 +934,16 @@ impl ScannerIOCache for SetDisks {
};
}
debug!("nsscanner_disk: cache.info.name: {:?}", cache.info.name);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = ?cache.info.name,
state = "cache_ready",
"Scanner disk bucket state updated"
);
let (updates_tx, mut updates_rx) = mpsc::channel::<DataUsageEntry>(1);
@@ -841,7 +966,16 @@ impl ScannerIOCache for SetDisks {
})
.await
{
error!("Failed to send data usage entry info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket_name_clone,
state = "send_failed",
error = %e,
"Scanner data usage stream failed"
);
}
}
});
@@ -855,9 +989,27 @@ impl ScannerIOCache for SetDisks {
Ok(scan_outcome) => scan_outcome,
Err(e) => {
if ctx_clone.is_cancelled() {
debug!("Scanner disk scan stopped after cancellation: {}", e);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "cancelled",
error = %e,
"Scanner disk bucket state updated"
);
} else {
error!("Failed to scan disk: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "scan_failed",
error = %e,
"Scanner disk bucket state updated"
);
}
if let (Some(last_update), Some(before_update)) = (cache.info.last_update, before)
@@ -865,13 +1017,32 @@ impl ScannerIOCache for SetDisks {
{
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = cache.save(store_clone_clone.clone(), cache_name.as_str()).await {
error!("Failed to save data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "save_failed",
error = %e,
"Scanner bucket cache save failed"
);
}
done_save();
}
if let Err(e) = update_fut.await {
error!("Failed to update data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "update_join_failed",
error = %e,
"Scanner bucket update task failed"
);
}
continue;
}
@@ -882,40 +1053,114 @@ impl ScannerIOCache for SetDisks {
ScannerDiskScanOutcome::Partial(cache) => {
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = cache.save(store_clone_clone.clone(), cache_name.as_str()).await {
error!("Failed to save partial data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "partial_save_failed",
error = %e,
"Scanner partial bucket cache save failed"
);
}
done_save();
if let Err(e) = update_fut.await {
error!("Failed to update partial data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "partial_update_join_failed",
error = %e,
"Scanner partial bucket update task failed"
);
}
if let Err(e) = send_cache_root_entry_info(&bucket_result_tx_clone_clone, &cache).await {
error!("Failed to send partial data usage entry info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "send_partial_root_failed",
error = %e,
"Scanner partial root entry publish failed"
);
}
continue;
}
};
debug!("nsscanner_disk: got cache: {}", cache.info.name);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache.info.name,
state = "scan_completed",
"Scanner disk bucket state updated"
);
if let Err(e) = update_fut.await {
error!("nsscanner_disk: Failed to update data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "update_join_failed",
error = %e,
"Scanner bucket update task failed"
);
}
if ctx_clone.is_cancelled() {
break;
}
debug!("nsscanner_disk: sending data usage entry info: {}", cache.info.name);
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache.info.name,
state = "send_root_entry",
"Scanner data usage stream progress updated"
);
if let Err(e) = send_cache_root_entry_info(&bucket_result_tx_clone_clone, &cache).await {
error!("nsscanner_disk: Failed to send data usage entry info: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "send_root_failed",
error = %e,
"Scanner root entry publish failed"
);
}
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = cache.save(store_clone_clone.clone(), &cache_name).await {
error!("nsscanner_disk: Failed to save data usage cache: {}", e);
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "save_failed",
error = %e,
"Scanner bucket cache save failed"
);
}
done_save();
}
@@ -931,7 +1176,14 @@ impl ScannerIOCache for SetDisks {
send_update_fut.await?;
debug!("nsscanner_cache: done");
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "set_scan_completed",
"Scanner set-level disk scan completed"
);
Ok(())
}
@@ -966,7 +1218,17 @@ impl ScannerIODisk for Disk {
let fivs = match meta.get_file_info_versions(item.bucket.as_str(), item.object_path().as_str(), false) {
Ok(versions) => versions,
Err(e) => {
error!("Failed to get file info versions: {}/{}, err: {e}", item.bucket, item.object_path());
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %item.bucket,
object = %item.object_path(),
state = "file_info_versions_failed",
error = %e,
"Scanner disk bucket failed to resolve file info versions"
);
return Err(StorageError::other(SCANNER_SKIP_FILE_ERROR.to_string()));
}
};
@@ -1072,28 +1334,60 @@ impl ScannerIODisk for Disk {
// TODO: object lock
let Some(ecstore) = new_object_layer_fn() else {
error!("ECStore not available");
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket,
state = "ecstore_unavailable",
"Scanner disk bucket missing object layer"
);
return Err(StorageError::other("ECStore not available".to_string()));
};
let disk_location = self.get_disk_location();
let (Some(pool_idx), Some(set_idx)) = (disk_location.pool_idx, disk_location.set_idx) else {
error!("Disk location not available");
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket,
state = "disk_location_unavailable",
"Scanner disk bucket missing disk location"
);
return Err(StorageError::other("Disk location not available".to_string()));
};
let disks_result = StorageAdminApi::disk_set_inventory(ecstore.as_ref(), DiskSetSelector::new(pool_idx, set_idx)).await?;
let Some(disk_idx) = disk_location.disk_idx else {
error!("Disk index not available");
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket,
state = "disk_index_unavailable",
"Scanner disk bucket missing disk index"
);
return Err(StorageError::other("Disk index not available".to_string()));
};
let local_disk = if let Some(Some(local_disk)) = disks_result.get(disk_idx) {
local_disk.clone()
} else {
error!("Local disk not available");
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DISK_BUCKET_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket,
state = "local_disk_unavailable",
"Scanner disk bucket missing local disk"
);
return Err(StorageError::other("Local disk not available".to_string()));
};
scripts/check_logging_guardrails.sh
+34
View File
@@ -18,6 +18,15 @@ checked_files=(
"crates/ecstore/src/store/peer.rs"
"crates/ecstore/src/store/init.rs"
"crates/ecstore/src/tier/tier.rs"
"crates/heal/src/heal/manager.rs"
"crates/heal/src/heal/storage.rs"
"crates/heal/src/heal/task.rs"
"crates/heal/src/heal/erasure_healer.rs"
"crates/heal/src/heal/resume.rs"
"crates/heal/src/heal/channel.rs"
"crates/scanner/src/scanner.rs"
"crates/scanner/src/scanner_io.rs"
"crates/scanner/src/scanner_folder.rs"
"crates/concurrency/src/workers.rs"
"crates/concurrency/src/manager.rs"
"crates/concurrency/src/lock.rs"
@@ -126,6 +135,31 @@ forbidden_patterns=(
'warn!("KMS initialization skipped: {e}")'
'warn!("Audit system: {e}")'
'warn!("notification system: {e}")'
'info!("Starting HealManager")'
'info!("Stopping HealManager")'
'info!("HealManager started successfully")'
'info!("HealManager stopped successfully")'
'info!("Healing MRF: {}")'
'info!("Step 1: Performing MRF heal using ecstore")'
'info!("Skipping initial data scanner delay because persisted data usage cache is cold")'
'error!("Failed to run data scanner: {e}")'
'warn!("Failed to read background heal info from {}: {}")'
'error!("Failed to marshal background heal info: {}")'
'warn!("Failed to save background heal info to {}: {}")'
'debug!("nsscanner_cache: no online disks available for set")'
'debug!("scan_folder: Preemptively compacting: {}, entries: {}")'
'warn!("scan_folder: failed to scan child folder {}: {}")'
'error!("scan_folder: failed to list path: {}/{}: {}")'
'info!("Cancelled active heal task: {}")'
'info!("Cancelled queued heal task: {}")'
'info!("Cancelled {} heal task(s) for path: {}")'
'info!("Heal scheduler received shutdown signal")'
'info!("Heal queue has {} pending requests, {} tasks active")'
'error!("Heal channel processor failed: {}")'
'info!("Heal manager with channel processor initialized successfully")'
'"scanner deep heal downgraded to normal during new-object cooldown"'
'"scanner detected folder with excessive direct subfolders"'
'"scan_folder: failed to get size for item {}: {}"'
'info!("worker take, {}", *available)'
'info!("worker give, {}", *available)'
'info!("worker wait end")'