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

refact: step 2

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
dandan
2025-07-09 17:32:40 +08:00
committed by junxiang Mu
parent 1f11a3167b
commit b40ef147a9
33 changed files with 1188 additions and 14575 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ahm/src/heal/engine.rs
-438
View File
@@ -1,438 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{
collections::HashMap,
sync::Arc,
time::{Duration, Instant, SystemTime},
};
use tokio::{
sync::{mpsc, RwLock},
time::sleep,
};
use tracing::{error, info, warn};
use uuid::Uuid;
use crate::error::Result;
use super::{HealConfig, HealPriority, HealResult, HealStatistics, HealTask, Status};
/// Main healing engine that coordinates repair operations
pub struct HealEngine {
config: HealConfig,
status: Arc<RwLock<Status>>,
statistics: Arc<RwLock<HealStatistics>>,
task_queue: Arc<RwLock<Vec<HealTask>>>,
active_tasks: Arc<RwLock<HashMap<String, HealTask>>>,
completed_tasks: Arc<RwLock<Vec<HealResult>>>,
shutdown_tx: Option<mpsc::Sender<()>>,
}
impl HealEngine {
/// Create a new healing engine
pub fn new(config: HealConfig) -> Self {
Self {
config,
status: Arc::new(RwLock::new(Status::Initializing)),
statistics: Arc::new(RwLock::new(HealStatistics::default())),
task_queue: Arc::new(RwLock::new(Vec::new())),
active_tasks: Arc::new(RwLock::new(HashMap::new())),
completed_tasks: Arc::new(RwLock::new(Vec::new())),
shutdown_tx: None,
}
}
/// Start the healing engine
pub async fn start(&mut self) -> Result<()> {
info!("Starting heal engine");
let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
self.shutdown_tx = Some(shutdown_tx);
// Update status
{
let mut status = self.status.write().await;
*status = Status::Idle;
}
let config = self.config.clone();
let status = Arc::clone(&self.status);
let statistics = Arc::clone(&self.statistics);
let task_queue = Arc::clone(&self.task_queue);
let active_tasks = Arc::clone(&self.active_tasks);
let completed_tasks = Arc::clone(&self.completed_tasks);
// Start the main healing loop
tokio::spawn(async move {
let mut interval = tokio::time::interval(config.heal_interval);
loop {
tokio::select! {
_ = interval.tick() => {
if let Err(e) = Self::process_healing_cycle(
&config,
&status,
&statistics,
&task_queue,
&active_tasks,
&completed_tasks,
).await {
error!("Healing cycle failed: {}", e);
}
}
_ = shutdown_rx.recv() => {
info!("Shutdown signal received, stopping heal engine");
break;
}
}
}
// Update status to stopped
let mut status = status.write().await;
*status = Status::Stopped;
});
info!("Heal engine started successfully");
Ok(())
}
/// Stop the healing engine
pub async fn stop(&mut self) -> Result<()> {
info!("Stopping heal engine");
// Update status
{
let mut status = self.status.write().await;
*status = Status::Stopping;
}
// Send shutdown signal
if let Some(shutdown_tx) = &self.shutdown_tx {
let _ = shutdown_tx.send(()).await;
}
// Wait for engine to stop
let mut attempts = 0;
while attempts < 10 {
let status = self.status.read().await;
if *status == Status::Stopped {
break;
}
drop(status);
sleep(Duration::from_millis(100)).await;
attempts += 1;
}
info!("Heal engine stopped");
Ok(())
}
/// Add a healing task to the queue
pub async fn add_task(&self, task: HealTask) -> Result<()> {
let task_id = task.id.clone();
let queue = Arc::clone(&self.task_queue);
// Add task to priority queue
queue.write().await.push(task);
info!("Added healing task to queue: {}", task_id);
Ok(())
}
/// Get current engine status
pub async fn status(&self) -> Status {
self.status.read().await.clone()
}
/// Get current engine status (alias for status)
pub async fn get_status(&self) -> Status {
self.status.read().await.clone()
}
/// Get engine configuration
pub async fn get_config(&self) -> HealConfig {
self.config.clone()
}
/// Get healing statistics
pub async fn statistics(&self) -> HealStatistics {
self.statistics.read().await.clone()
}
/// Get completed healing results
pub async fn completed_results(&self) -> Vec<HealResult> {
self.completed_tasks.read().await.clone()
}
/// Process a single healing cycle
async fn process_healing_cycle(
config: &HealConfig,
status: &Arc<RwLock<Status>>,
statistics: &Arc<RwLock<HealStatistics>>,
task_queue: &Arc<RwLock<Vec<HealTask>>>,
active_tasks: &Arc<RwLock<HashMap<String, HealTask>>>,
completed_tasks: &Arc<RwLock<Vec<HealResult>>>,
) -> Result<()> {
// Update status to healing
{
let mut status = status.write().await;
*status = Status::Healing;
}
// Get ready tasks from queue
let mut queue = task_queue.write().await;
let mut ready_tasks = Vec::new();
let mut remaining_tasks = Vec::new();
for task in queue.drain(..) {
if task.is_ready() {
ready_tasks.push(task);
} else {
remaining_tasks.push(task);
}
}
// Sort ready tasks by priority
ready_tasks.sort_by(|a, b| a.priority.cmp(&b.priority));
// Process ready tasks
let active_count = active_tasks.read().await.len();
let max_concurrent = config.max_workers.saturating_sub(active_count);
for task in ready_tasks.into_iter().take(max_concurrent) {
if let Err(e) = Self::process_task(
config,
statistics,
active_tasks,
completed_tasks,
task,
).await {
error!("Failed to process healing task: {}", e);
}
}
// Put remaining tasks back in queue
queue.extend(remaining_tasks);
// Update statistics
{
let mut stats = statistics.write().await;
stats.queued_tasks = queue.len() as u64;
stats.active_workers = active_tasks.read().await.len() as u64;
}
// Update status back to idle
{
let mut status = status.write().await;
*status = Status::Idle;
}
Ok(())
}
/// Process a single healing task
async fn process_task(
config: &HealConfig,
statistics: &Arc<RwLock<HealStatistics>>,
active_tasks: &Arc<RwLock<HashMap<String, HealTask>>>,
completed_tasks: &Arc<RwLock<Vec<HealResult>>>,
task: HealTask,
) -> Result<()> {
let task_id = task.id.clone();
// Add task to active tasks
{
let mut active = active_tasks.write().await;
active.insert(task_id.clone(), task.clone());
}
// Update statistics
{
let mut stats = statistics.write().await;
stats.total_repairs += 1;
stats.active_workers = active_tasks.read().await.len() as u64;
}
info!("Processing healing task: {}", task_id);
// Simulate healing operation
let start_time = Instant::now();
let result = Self::perform_healing_operation(&task, config).await;
let duration = start_time.elapsed();
// Create heal result
let heal_result = HealResult {
success: result.is_ok(),
original_issue: task.issue.clone(),
repair_duration: duration,
retry_attempts: task.retry_count,
error_message: result.err().map(|e| e.to_string()),
metadata: None,
completed_at: SystemTime::now(),
};
// Update statistics
{
let mut stats = statistics.write().await;
if heal_result.success {
stats.successful_repairs += 1;
} else {
stats.failed_repairs += 1;
}
stats.total_repair_time += duration;
stats.average_repair_time = if stats.total_repairs > 0 {
Duration::from_secs_f64(
stats.total_repair_time.as_secs_f64() / stats.total_repairs as f64
)
} else {
Duration::ZERO
};
stats.last_repair_time = Some(SystemTime::now());
stats.total_retry_attempts += task.retry_count as u64;
}
// Add result to completed tasks
{
let mut completed = completed_tasks.write().await;
completed.push(heal_result.clone());
}
// Remove task from active tasks
{
let mut active = active_tasks.write().await;
active.remove(&task_id);
}
// Update statistics
{
let mut stats = statistics.write().await;
stats.active_workers = active_tasks.read().await.len() as u64;
}
if heal_result.success {
info!("Healing task completed successfully: {}", task_id);
} else {
warn!("Healing task failed: {}", task_id);
}
Ok(())
}
/// Perform the actual healing operation
async fn perform_healing_operation(task: &HealTask, _config: &HealConfig) -> Result<()> {
// Simulate healing operation based on issue type
match task.issue.issue_type {
crate::scanner::HealthIssueType::MissingReplica => {
// Simulate replica repair
sleep(Duration::from_millis(100)).await;
info!("Repaired missing replica for {}/{}", task.issue.bucket, task.issue.object);
}
crate::scanner::HealthIssueType::ChecksumMismatch => {
// Simulate checksum repair
sleep(Duration::from_millis(200)).await;
info!("Repaired checksum mismatch for {}/{}", task.issue.bucket, task.issue.object);
}
crate::scanner::HealthIssueType::DiskReadError => {
// Simulate disk error recovery
sleep(Duration::from_millis(300)).await;
info!("Recovered from disk read error for {}/{}", task.issue.bucket, task.issue.object);
}
_ => {
// Generic repair for other issue types
sleep(Duration::from_millis(150)).await;
info!("Performed generic repair for {}/{}", task.issue.bucket, task.issue.object);
}
}
// Simulate occasional failures for testing
if task.retry_count > 0 && task.retry_count % 3 == 0 {
return Err(crate::error::Error::Other(anyhow::anyhow!("Simulated healing failure")));
}
Ok(())
}
/// Start healing operations
pub async fn start_healing(&self) -> Result<()> {
let mut status = self.status.write().await;
*status = Status::Running;
info!("Healing operations started");
Ok(())
}
/// Stop healing operations
pub async fn stop_healing(&self) -> Result<()> {
let mut status = self.status.write().await;
*status = Status::Stopped;
info!("Healing operations stopped");
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::scanner::{HealthIssue, HealthIssueType, Severity};
#[tokio::test]
async fn test_heal_engine_creation() {
let config = HealConfig::default();
let engine = HealEngine::new(config);
assert_eq!(engine.status().await, Status::Initializing);
}
#[tokio::test]
async fn test_heal_engine_start_stop() {
let config = HealConfig::default();
let mut engine = HealEngine::new(config);
// Start engine
engine.start().await.unwrap();
sleep(Duration::from_millis(100)).await;
// Check status
let status = engine.status().await;
assert!(matches!(status, Status::Idle | Status::Healing));
// Stop engine
engine.stop().await.unwrap();
sleep(Duration::from_millis(100)).await;
// Check status
let status = engine.status().await;
assert_eq!(status, Status::Stopped);
}
#[tokio::test]
async fn test_add_healing_task() {
let config = HealConfig::default();
let engine = HealEngine::new(config);
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = HealTask::new(issue);
engine.add_task(task).await.unwrap();
let stats = engine.statistics().await;
assert_eq!(stats.queued_tasks, 1);
}
}
crates/ahm/src/heal/mod.rs
-360
View File
@@ -1,360 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Healing subsystem for the AHM system
//!
//! The heal subsystem provides intelligent repair capabilities:
//! - Priority-based healing queue
//! - Real-time and background healing modes
//! - Comprehensive repair validation
//! - Adaptive healing strategies
pub mod engine;
pub mod priority_queue;
pub mod repair_worker;
pub mod validation;
pub use engine::HealEngine;
pub use priority_queue::PriorityQueue;
pub use repair_worker::RepairWorker;
pub use validation::HealValidator;
use std::time::{Duration, SystemTime};
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use derive_builder::Builder;
use crate::scanner::{HealthIssue, HealthIssueType, Severity};
/// Configuration for the healing system
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealConfig {
/// Maximum number of concurrent repair workers
pub max_workers: usize,
/// Maximum number of tasks in the priority queue
pub max_queue_size: usize,
/// Timeout for individual repair operations
pub repair_timeout: Duration,
/// Interval between healing cycles
pub heal_interval: Duration,
/// Whether to enable automatic healing
pub auto_heal_enabled: bool,
/// Maximum number of retry attempts for failed repairs
pub max_retry_attempts: u32,
/// Backoff delay between retry attempts
pub retry_backoff_delay: Duration,
/// Whether to validate repairs after completion
pub validate_after_repair: bool,
}
impl Default for HealConfig {
fn default() -> Self {
Self {
max_workers: 4,
max_queue_size: 1000,
repair_timeout: Duration::from_secs(300), // 5 minutes
heal_interval: Duration::from_secs(60), // 1 minute
auto_heal_enabled: true,
max_retry_attempts: 3,
retry_backoff_delay: Duration::from_secs(30),
validate_after_repair: true,
}
}
}
/// Result of a healing operation
#[derive(Debug, Clone)]
pub struct HealResult {
/// Whether the healing operation was successful
pub success: bool,
/// The original health issue that was addressed
pub original_issue: HealthIssue,
/// Time taken to complete the repair
pub repair_duration: Duration,
/// Number of retry attempts made
pub retry_attempts: u32,
/// Error message if repair failed
pub error_message: Option<String>,
/// Additional metadata about the repair
pub metadata: Option<serde_json::Value>,
/// Timestamp when the repair was completed
pub completed_at: SystemTime,
}
/// Statistics for the healing system
#[derive(Debug, Clone, Default)]
pub struct HealStatistics {
/// Total number of repair tasks processed
pub total_repairs: u64,
/// Number of successful repairs
pub successful_repairs: u64,
/// Number of failed repairs
pub failed_repairs: u64,
/// Number of tasks currently in queue
pub queued_tasks: u64,
/// Number of active workers
pub active_workers: u64,
/// Total time spent on repairs
pub total_repair_time: Duration,
/// Average repair time
pub average_repair_time: Duration,
/// Last repair completion time
pub last_repair_time: Option<SystemTime>,
/// Number of retry attempts made
pub total_retry_attempts: u64,
}
/// Priority levels for healing tasks
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum HealPriority {
/// Critical issues that need immediate attention
Critical = 0,
/// High priority issues
High = 1,
/// Medium priority issues
Medium = 2,
/// Low priority issues
Low = 3,
}
impl From<Severity> for HealPriority {
fn from(severity: Severity) -> Self {
match severity {
Severity::Critical => HealPriority::Critical,
Severity::High => HealPriority::High,
Severity::Medium => HealPriority::Medium,
Severity::Low => HealPriority::Low,
}
}
}
/// A healing task to be processed
#[derive(Debug, Clone)]
pub struct HealTask {
/// Unique identifier for the task
pub id: String,
/// The health issue to be repaired
pub issue: HealthIssue,
/// Priority level for this task
pub priority: HealPriority,
/// When the task was created
pub created_at: SystemTime,
/// When the task should be processed (for delayed tasks)
pub scheduled_at: Option<SystemTime>,
/// Number of retry attempts made
pub retry_count: u32,
/// Maximum number of retry attempts allowed
pub max_retries: u32,
/// Additional context for the repair operation
pub context: Option<serde_json::Value>,
}
impl HealTask {
/// Create a new healing task
pub fn new(issue: HealthIssue) -> Self {
let priority = HealPriority::from(issue.severity);
Self {
id: uuid::Uuid::new_v4().to_string(),
issue,
priority,
created_at: SystemTime::now(),
scheduled_at: None,
retry_count: 0,
max_retries: 3,
context: None,
}
}
/// Create a delayed healing task
pub fn delayed(issue: HealthIssue, delay: Duration) -> Self {
let mut task = Self::new(issue);
task.scheduled_at = Some(SystemTime::now() + delay);
task
}
/// Check if the task is ready to be processed
pub fn is_ready(&self) -> bool {
if let Some(scheduled_at) = self.scheduled_at {
SystemTime::now() >= scheduled_at
} else {
true
}
}
/// Check if the task can be retried
pub fn can_retry(&self) -> bool {
self.retry_count < self.max_retries
}
/// Increment the retry count
pub fn increment_retry(&mut self) {
self.retry_count += 1;
}
}
/// Heal engine status
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum Status {
/// Heal engine is initializing
Initializing,
/// Heal engine is idle
Idle,
/// Heal engine is running normally
Running,
/// Heal engine is actively healing
Healing,
/// Heal engine is paused
Paused,
/// Heal engine is stopping
Stopping,
/// Heal engine has stopped
Stopped,
/// Heal engine encountered an error
Error(String),
}
/// Healing operation modes
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum HealMode {
/// Real-time healing during GET/PUT operations
RealTime,
/// Background healing during scheduled scans
Background,
/// On-demand healing triggered by admin
OnDemand,
/// Emergency healing for critical issues
Emergency,
}
/// Validation result for a repaired object
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValidationResult {
/// Type of validation performed
pub validation_type: ValidationType,
/// Whether validation passed
pub passed: bool,
/// Details about the validation
pub details: String,
/// Time taken for validation
pub duration: Duration,
}
/// Types of validation that can be performed
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum ValidationType {
/// Checksum verification
Checksum,
/// Shard count verification
ShardCount,
/// Data integrity check
DataIntegrity,
/// Metadata consistency check
MetadataConsistency,
/// Cross-shard redundancy check
RedundancyCheck,
}
/// Healing strategies for different scenarios
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum HealStrategy {
/// Repair using available data shards
DataShardRepair,
/// Repair using parity shards
ParityShardRepair,
/// Hybrid repair using both data and parity
HybridRepair,
/// Metadata-only repair
MetadataRepair,
/// Full object reconstruction
FullReconstruction,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_heal_priority_from_severity() {
assert_eq!(HealPriority::from(Severity::Critical), HealPriority::Critical);
assert_eq!(HealPriority::from(Severity::High), HealPriority::High);
assert_eq!(HealPriority::from(Severity::Medium), HealPriority::Medium);
assert_eq!(HealPriority::from(Severity::Low), HealPriority::Low);
}
#[test]
fn test_heal_task_creation() {
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = HealTask::new(issue.clone());
assert_eq!(task.priority, HealPriority::Critical);
assert_eq!(task.issue.bucket, issue.bucket);
assert_eq!(task.issue.object, issue.object);
assert_eq!(task.retry_count, 0);
assert_eq!(task.max_retries, 3);
assert!(task.is_ready());
}
#[test]
fn test_delayed_heal_task() {
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Medium,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let delay = Duration::from_secs(1);
let task = HealTask::delayed(issue, delay);
assert!(task.scheduled_at.is_some());
assert!(!task.is_ready()); // Should not be ready immediately
// Wait for the delay to pass
std::thread::sleep(delay + Duration::from_millis(100));
assert!(task.is_ready());
}
#[test]
fn test_heal_task_retry_logic() {
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Low,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let mut task = HealTask::new(issue);
assert!(task.can_retry());
task.increment_retry();
assert_eq!(task.retry_count, 1);
assert!(task.can_retry());
task.increment_retry();
task.increment_retry();
assert_eq!(task.retry_count, 3);
assert!(!task.can_retry());
}
}
crates/ahm/src/heal/priority_queue.rs
-413
View File
@@ -1,413 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{
collections::BinaryHeap,
sync::Arc,
time::{Duration, SystemTime},
};
use tokio::sync::RwLock;
use tracing::{debug, info, warn};
use crate::error::Result;
use super::{HealPriority, HealTask};
/// Priority queue for healing tasks
pub struct PriorityQueue {
tasks: Arc<RwLock<BinaryHeap<HealTask>>>,
max_size: usize,
statistics: Arc<RwLock<QueueStatistics>>,
}
/// Statistics for the priority queue
#[derive(Debug, Clone, Default)]
pub struct QueueStatistics {
/// Total number of tasks added to the queue
pub total_tasks_added: u64,
/// Total number of tasks removed from the queue
pub total_tasks_removed: u64,
/// Current number of tasks in the queue
pub current_queue_size: u64,
/// Maximum queue size reached
pub max_queue_size_reached: u64,
/// Number of tasks rejected due to queue being full
pub tasks_rejected: u64,
/// Average time tasks spend in queue
pub average_queue_time: Duration,
/// Total time all tasks have spent in queue
pub total_queue_time: Duration,
}
impl PriorityQueue {
/// Create a new priority queue
pub fn new(max_size: usize) -> Self {
Self {
tasks: Arc::new(RwLock::new(BinaryHeap::new())),
max_size,
statistics: Arc::new(RwLock::new(QueueStatistics::default())),
}
}
/// Add a task to the queue
pub async fn push(&self, task: HealTask) -> Result<()> {
let mut tasks = self.tasks.write().await;
let mut stats = self.statistics.write().await;
if tasks.len() >= self.max_size {
stats.tasks_rejected += 1;
warn!("Priority queue is full, rejecting task: {}", task.id);
return Err(crate::error::Error::Other(anyhow::anyhow!("Queue is full")));
}
let task_id = task.id.clone();
let priority = task.priority.clone();
tasks.push(task);
stats.total_tasks_added += 1;
stats.current_queue_size = tasks.len() as u64;
stats.max_queue_size_reached = stats.max_queue_size_reached.max(tasks.len() as u64);
debug!("Added task to priority queue: {} (priority: {:?})", task_id, priority);
Ok(())
}
/// Remove and return the highest priority task
pub async fn pop(&self) -> Option<HealTask> {
let mut tasks = self.tasks.write().await;
let mut stats = self.statistics.write().await;
if let Some(task) = tasks.pop() {
stats.total_tasks_removed += 1;
stats.current_queue_size = tasks.len() as u64;
// Update queue time statistics
let queue_time = SystemTime::now().duration_since(task.created_at).unwrap_or(Duration::ZERO);
stats.total_queue_time += queue_time;
stats.average_queue_time = if stats.total_tasks_removed > 0 {
Duration::from_secs_f64(
stats.total_queue_time.as_secs_f64() / stats.total_tasks_removed as f64
)
} else {
Duration::ZERO
};
debug!("Removed task from priority queue: {} (priority: {:?})", task.id, task.priority);
Some(task)
} else {
None
}
}
/// Peek at the highest priority task without removing it
pub async fn peek(&self) -> Option<HealTask> {
let tasks = self.tasks.read().await;
tasks.peek().cloned()
}
/// Get the current size of the queue
pub async fn len(&self) -> usize {
self.tasks.read().await.len()
}
/// Check if the queue is empty
pub async fn is_empty(&self) -> bool {
self.tasks.read().await.is_empty()
}
/// Get queue statistics
pub async fn statistics(&self) -> QueueStatistics {
self.statistics.read().await.clone()
}
/// Clear all tasks from the queue
pub async fn clear(&self) {
let mut tasks = self.tasks.write().await;
let mut stats = self.statistics.write().await;
let cleared_count = tasks.len();
tasks.clear();
stats.current_queue_size = 0;
info!("Cleared {} tasks from priority queue", cleared_count);
}
/// Get all tasks that are ready to be processed
pub async fn get_ready_tasks(&self, max_count: usize) -> Vec<HealTask> {
let mut tasks = self.tasks.write().await;
let mut ready_tasks = Vec::new();
let mut remaining_tasks = Vec::new();
while let Some(task) = tasks.pop() {
if task.is_ready() && ready_tasks.len() < max_count {
ready_tasks.push(task);
} else {
remaining_tasks.push(task);
}
}
// Put remaining tasks back
for task in remaining_tasks {
tasks.push(task);
}
ready_tasks
}
/// Remove a specific task by ID
pub async fn remove_task(&self, task_id: &str) -> bool {
let mut tasks = self.tasks.write().await;
let mut stats = self.statistics.write().await;
let mut temp_tasks = Vec::new();
let mut found = false;
while let Some(task) = tasks.pop() {
if task.id == task_id {
found = true;
stats.total_tasks_removed += 1;
debug!("Removed specific task from queue: {}", task_id);
} else {
temp_tasks.push(task);
}
}
// Put remaining tasks back
for task in temp_tasks {
tasks.push(task);
}
stats.current_queue_size = tasks.len() as u64;
found
}
/// Get tasks by priority level
pub async fn get_tasks_by_priority(&self, priority: HealPriority) -> Vec<HealTask> {
let mut tasks = self.tasks.write().await;
let mut matching_tasks = Vec::new();
let mut other_tasks = Vec::new();
while let Some(task) = tasks.pop() {
if task.priority == priority {
matching_tasks.push(task);
} else {
other_tasks.push(task);
}
}
// Put other tasks back
for task in other_tasks {
tasks.push(task);
}
matching_tasks
}
/// Update task priority
pub async fn update_priority(&self, task_id: &str, new_priority: HealPriority) -> bool {
let mut tasks = self.tasks.write().await;
let mut temp_tasks = Vec::new();
let mut found = false;
while let Some(mut task) = tasks.pop() {
if task.id == task_id {
task.priority = new_priority.clone();
found = true;
debug!("Updated task priority: {} -> {:?}", task_id, new_priority);
}
temp_tasks.push(task);
}
// Put all tasks back
for task in temp_tasks {
tasks.push(task);
}
found
}
}
// Implement Ord for HealTask to enable priority queue functionality
impl std::cmp::Ord for HealTask {
fn cmp(&self, other: &Self) -> std::cmp::Ordering {
// Higher priority (lower enum value) comes first
self.priority.cmp(&other.priority)
.then_with(|| self.created_at.cmp(&other.created_at))
}
}
impl std::cmp::PartialOrd for HealTask {
fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
Some(self.cmp(other))
}
}
impl std::cmp::PartialEq for HealTask {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl std::cmp::Eq for HealTask {}
#[cfg(test)]
mod tests {
use super::*;
use crate::scanner::{HealthIssue, HealthIssueType, Severity};
#[tokio::test]
async fn test_priority_queue_creation() {
let queue = PriorityQueue::new(100);
assert_eq!(queue.len().await, 0);
assert!(queue.is_empty().await);
}
#[tokio::test]
async fn test_priority_queue_push_pop() {
let queue = PriorityQueue::new(10);
let issue1 = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Low,
bucket: "bucket1".to_string(),
object: "object1".to_string(),
description: "Test issue 1".to_string(),
metadata: None,
};
let issue2 = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "bucket2".to_string(),
object: "object2".to_string(),
description: "Test issue 2".to_string(),
metadata: None,
};
let task1 = HealTask::new(issue1);
let task2 = HealTask::new(issue2);
// Add tasks
queue.push(task1.clone()).await.unwrap();
queue.push(task2.clone()).await.unwrap();
assert_eq!(queue.len().await, 2);
// Critical task should come first
let first_task = queue.pop().await.unwrap();
assert_eq!(first_task.priority, HealPriority::Critical);
assert_eq!(first_task.id, task2.id);
let second_task = queue.pop().await.unwrap();
assert_eq!(second_task.priority, HealPriority::Low);
assert_eq!(second_task.id, task1.id);
assert!(queue.is_empty().await);
}
#[tokio::test]
async fn test_priority_queue_full() {
let queue = PriorityQueue::new(1);
let issue1 = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Low,
bucket: "bucket1".to_string(),
object: "object1".to_string(),
description: "Test issue 1".to_string(),
metadata: None,
};
let issue2 = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "bucket2".to_string(),
object: "object2".to_string(),
description: "Test issue 2".to_string(),
metadata: None,
};
let task1 = HealTask::new(issue1);
let task2 = HealTask::new(issue2);
// First task should succeed
queue.push(task1).await.unwrap();
assert_eq!(queue.len().await, 1);
// Second task should fail
let result = queue.push(task2).await;
assert!(result.is_err());
assert_eq!(queue.len().await, 1);
let stats = queue.statistics().await;
assert_eq!(stats.tasks_rejected, 1);
}
#[tokio::test]
async fn test_priority_queue_remove_task() {
let queue = PriorityQueue::new(10);
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Medium,
bucket: "bucket1".to_string(),
object: "object1".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = HealTask::new(issue);
let task_id = task.id.clone();
queue.push(task).await.unwrap();
assert_eq!(queue.len().await, 1);
// Remove the task
let removed = queue.remove_task(&task_id).await;
assert!(removed);
assert_eq!(queue.len().await, 0);
// Try to remove non-existent task
let removed = queue.remove_task("non-existent").await;
assert!(!removed);
}
#[tokio::test]
async fn test_priority_queue_update_priority() {
let queue = PriorityQueue::new(10);
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Low,
bucket: "bucket1".to_string(),
object: "object1".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = HealTask::new(issue);
let task_id = task.id.clone();
queue.push(task).await.unwrap();
// Update priority
let updated = queue.update_priority(&task_id, HealPriority::Critical).await;
assert!(updated);
// Check that the task now has higher priority
let popped_task = queue.pop().await.unwrap();
assert_eq!(popped_task.priority, HealPriority::Critical);
assert_eq!(popped_task.id, task_id);
}
}
crates/ahm/src/heal/repair_worker.rs
-505
View File
@@ -1,505 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{
sync::Arc,
time::{Duration, Instant, SystemTime},
};
use tokio::{
sync::{mpsc, RwLock},
time::{sleep, timeout},
};
use tracing::{debug, error, info, warn};
use crate::error::Result;
use super::{HealConfig, HealResult, HealTask, Status};
/// Configuration for repair workers
#[derive(Debug, Clone)]
pub struct RepairWorkerConfig {
/// Worker ID
pub worker_id: String,
/// Maximum time to spend on a single repair operation
pub operation_timeout: Duration,
/// Whether to enable detailed logging
pub enable_detailed_logging: bool,
/// Maximum number of concurrent operations
pub max_concurrent_operations: usize,
/// Retry configuration
pub retry_config: RetryConfig,
}
/// Retry configuration for repair operations
#[derive(Debug, Clone)]
pub struct RetryConfig {
/// Maximum number of retry attempts
pub max_attempts: u32,
/// Initial backoff delay
pub initial_backoff: Duration,
/// Maximum backoff delay
pub max_backoff: Duration,
/// Backoff multiplier
pub backoff_multiplier: f64,
/// Whether to use exponential backoff
pub exponential_backoff: bool,
}
impl Default for RepairWorkerConfig {
fn default() -> Self {
Self {
worker_id: "worker-1".to_string(),
operation_timeout: Duration::from_secs(300), // 5 minutes
enable_detailed_logging: true,
max_concurrent_operations: 1,
retry_config: RetryConfig::default(),
}
}
}
impl Default for RetryConfig {
fn default() -> Self {
Self {
max_attempts: 3,
initial_backoff: Duration::from_secs(1),
max_backoff: Duration::from_secs(60),
backoff_multiplier: 2.0,
exponential_backoff: true,
}
}
}
/// Statistics for a repair worker
#[derive(Debug, Clone, Default)]
pub struct WorkerStatistics {
/// Total number of tasks processed
pub total_tasks_processed: u64,
/// Number of successful repairs
pub successful_repairs: u64,
/// Number of failed repairs
pub failed_repairs: u64,
/// Total time spent on repairs
pub total_repair_time: Duration,
/// Average repair time
pub average_repair_time: Duration,
/// Number of retry attempts made
pub total_retry_attempts: u64,
/// Current worker status
pub status: WorkerStatus,
/// Last task completion time
pub last_task_time: Option<SystemTime>,
}
/// Worker status
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum WorkerStatus {
/// Worker is idle
Idle,
/// Worker is processing a task
Processing,
/// Worker is retrying a failed task
Retrying,
/// Worker is stopping
Stopping,
/// Worker has stopped
Stopped,
/// Worker encountered an error
Error(String),
}
impl Default for WorkerStatus {
fn default() -> Self {
WorkerStatus::Idle
}
}
/// Repair worker that executes healing tasks
pub struct RepairWorker {
config: RepairWorkerConfig,
statistics: Arc<RwLock<WorkerStatistics>>,
status: Arc<RwLock<WorkerStatus>>,
result_tx: mpsc::Sender<HealResult>,
shutdown_tx: Option<mpsc::Sender<()>>,
}
impl RepairWorker {
/// Create a new repair worker
pub fn new(
config: RepairWorkerConfig,
result_tx: mpsc::Sender<HealResult>,
) -> Self {
Self {
config,
statistics: Arc::new(RwLock::new(WorkerStatistics::default())),
status: Arc::new(RwLock::new(WorkerStatus::Idle)),
result_tx,
shutdown_tx: None,
}
}
/// Start the repair worker
pub async fn start(&mut self) -> Result<()> {
info!("Starting repair worker: {}", self.config.worker_id);
let (_task_tx, task_rx) = mpsc::channel(100);
let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);
self.shutdown_tx = Some(shutdown_tx);
// Update status
{
let mut status = self.status.write().await;
*status = WorkerStatus::Idle;
}
let config = self.config.clone();
let statistics = Arc::clone(&self.statistics);
let status = Arc::clone(&self.status);
let result_tx = self.result_tx.clone();
// Start the worker loop
tokio::spawn(async move {
let mut task_rx = task_rx;
loop {
tokio::select! {
Some(task) = task_rx.recv() => {
if let Err(e) = Self::process_task(
&config,
&statistics,
&status,
&result_tx,
task,
).await {
error!("Failed to process task: {}", e);
}
}
_ = shutdown_rx.recv() => {
info!("Shutdown signal received, stopping worker: {}", config.worker_id);
break;
}
}
}
// Update status to stopped
let mut status = status.write().await;
*status = WorkerStatus::Stopped;
});
info!("Repair worker started: {}", self.config.worker_id);
Ok(())
}
/// Stop the repair worker
pub async fn stop(&mut self) -> Result<()> {
info!("Stopping repair worker: {}", self.config.worker_id);
// Update status
{
let mut status = self.status.write().await;
*status = WorkerStatus::Stopping;
}
// Send shutdown signal
if let Some(shutdown_tx) = &self.shutdown_tx {
let _ = shutdown_tx.send(()).await;
}
// Wait for worker to stop
let mut attempts = 0;
while attempts < 10 {
let status = self.status.read().await;
if *status == WorkerStatus::Stopped {
break;
}
drop(status);
sleep(Duration::from_millis(100)).await;
attempts += 1;
}
info!("Repair worker stopped: {}", self.config.worker_id);
Ok(())
}
/// Submit a task to the worker
pub async fn submit_task(&self, _task: HealTask) -> Result<()> {
// TODO: Implement task submission
Err(crate::error::Error::Other(anyhow::anyhow!("Task submission not implemented")))
}
/// Get worker statistics
pub async fn statistics(&self) -> WorkerStatistics {
self.statistics.read().await.clone()
}
/// Get worker status
pub async fn status(&self) -> WorkerStatus {
self.status.read().await.clone()
}
/// Process a single task
async fn process_task(
config: &RepairWorkerConfig,
statistics: &Arc<RwLock<WorkerStatistics>>,
status: &Arc<RwLock<WorkerStatus>>,
result_tx: &mpsc::Sender<HealResult>,
task: HealTask,
) -> Result<()> {
let task_id = task.id.clone();
// Update status to processing
{
let mut status = status.write().await;
*status = WorkerStatus::Processing;
}
// Update statistics
{
let mut stats = statistics.write().await;
stats.total_tasks_processed += 1;
stats.status = WorkerStatus::Processing;
}
info!("Processing repair task: {} (worker: {})", task_id, config.worker_id);
let start_time = Instant::now();
let mut attempt = 0;
let mut last_error = None;
// Retry loop
while attempt < config.retry_config.max_attempts {
attempt += 1;
if attempt > 1 {
// Update status to retrying
{
let mut status = status.write().await;
*status = WorkerStatus::Retrying;
}
// Calculate backoff delay
let backoff_delay = if config.retry_config.exponential_backoff {
let delay = config.retry_config.initial_backoff *
(config.retry_config.backoff_multiplier.powi((attempt - 1) as i32)) as u32;
delay.min(config.retry_config.max_backoff)
} else {
config.retry_config.initial_backoff
};
warn!("Retrying task {} (attempt {}/{}), waiting {:?}",
task_id, attempt, config.retry_config.max_attempts, backoff_delay);
sleep(backoff_delay).await;
}
// Attempt the repair operation
let result = timeout(
config.operation_timeout,
Self::perform_repair_operation(&task, config)
).await;
match result {
Ok(Ok(())) => {
// Success
let duration = start_time.elapsed();
let heal_result = HealResult {
success: true,
original_issue: task.issue.clone(),
repair_duration: duration,
retry_attempts: attempt - 1,
error_message: None,
metadata: None,
completed_at: SystemTime::now(),
};
// Send result
if let Err(e) = result_tx.send(heal_result).await {
error!("Failed to send heal result: {}", e);
}
// Update statistics
{
let mut stats = statistics.write().await;
stats.successful_repairs += 1;
stats.total_repair_time += duration;
stats.average_repair_time = if stats.total_tasks_processed > 0 {
Duration::from_secs_f64(
stats.total_repair_time.as_secs_f64() / stats.total_tasks_processed as f64
)
} else {
Duration::ZERO
};
stats.total_retry_attempts += (attempt - 1) as u64;
stats.last_task_time = Some(SystemTime::now());
stats.status = WorkerStatus::Idle;
}
info!("Successfully completed repair task: {} (worker: {})", task_id, config.worker_id);
return Ok(());
}
Ok(Err(e)) => {
// Operation failed
let error_msg = e.to_string();
last_error = Some(e);
warn!("Repair operation failed for task {} (attempt {}/{}): {}",
task_id, attempt, config.retry_config.max_attempts, error_msg);
}
Err(_) => {
// Operation timed out
last_error = Some(crate::error::Error::Other(anyhow::anyhow!("Operation timed out")));
warn!("Repair operation timed out for task {} (attempt {}/{})",
task_id, attempt, config.retry_config.max_attempts);
}
}
}
// All attempts failed
let duration = start_time.elapsed();
let heal_result = HealResult {
success: false,
original_issue: task.issue.clone(),
repair_duration: duration,
retry_attempts: attempt - 1,
error_message: last_error.map(|e| e.to_string()),
metadata: None,
completed_at: SystemTime::now(),
};
// Send result
if let Err(e) = result_tx.send(heal_result).await {
error!("Failed to send heal result: {}", e);
}
// Update statistics
{
let mut stats = statistics.write().await;
stats.failed_repairs += 1;
stats.total_repair_time += duration;
stats.average_repair_time = if stats.total_tasks_processed > 0 {
Duration::from_secs_f64(
stats.total_repair_time.as_secs_f64() / stats.total_tasks_processed as f64
)
} else {
Duration::ZERO
};
stats.total_retry_attempts += (attempt - 1) as u64;
stats.last_task_time = Some(SystemTime::now());
stats.status = WorkerStatus::Idle;
}
error!("Failed to complete repair task after {} attempts: {} (worker: {})",
attempt, task_id, config.worker_id);
Ok(())
}
/// Perform the actual repair operation
async fn perform_repair_operation(task: &HealTask, config: &RepairWorkerConfig) -> Result<()> {
if config.enable_detailed_logging {
debug!("Starting repair operation for task: {} (worker: {})", task.id, config.worker_id);
}
// Simulate repair operation based on issue type
match task.issue.issue_type {
crate::scanner::HealthIssueType::MissingReplica => {
// Simulate replica repair
sleep(Duration::from_millis(100)).await;
if config.enable_detailed_logging {
debug!("Repaired missing replica for {}/{}", task.issue.bucket, task.issue.object);
}
}
crate::scanner::HealthIssueType::ChecksumMismatch => {
// Simulate checksum repair
sleep(Duration::from_millis(200)).await;
if config.enable_detailed_logging {
debug!("Repaired checksum mismatch for {}/{}", task.issue.bucket, task.issue.object);
}
}
crate::scanner::HealthIssueType::DiskReadError => {
// Simulate disk error recovery
sleep(Duration::from_millis(300)).await;
if config.enable_detailed_logging {
debug!("Recovered from disk read error for {}/{}", task.issue.bucket, task.issue.object);
}
}
_ => {
// Generic repair for other issue types
sleep(Duration::from_millis(150)).await;
if config.enable_detailed_logging {
debug!("Performed generic repair for {}/{}", task.issue.bucket, task.issue.object);
}
}
}
// Simulate occasional failures for testing
if task.retry_count > 0 && task.retry_count % 3 == 0 {
return Err(crate::error::Error::Other(anyhow::anyhow!("Simulated repair failure")));
}
if config.enable_detailed_logging {
debug!("Completed repair operation for task: {} (worker: {})", task.id, config.worker_id);
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::scanner::{HealthIssue, HealthIssueType, Severity};
#[tokio::test]
async fn test_repair_worker_creation() {
let config = RepairWorkerConfig::default();
let (result_tx, _result_rx) = mpsc::channel(100);
let worker = RepairWorker::new(config, result_tx);
assert_eq!(worker.status().await, WorkerStatus::Idle);
}
#[tokio::test]
async fn test_repair_worker_start_stop() {
let config = RepairWorkerConfig::default();
let (result_tx, _result_rx) = mpsc::channel(100);
let mut worker = RepairWorker::new(config, result_tx);
// Start worker
worker.start().await.unwrap();
sleep(Duration::from_millis(100)).await;
// Check status
let status = worker.status().await;
assert_eq!(status, WorkerStatus::Idle);
// Stop worker
worker.stop().await.unwrap();
sleep(Duration::from_millis(100)).await;
// Check status
let status = worker.status().await;
assert_eq!(status, WorkerStatus::Stopped);
}
#[tokio::test]
async fn test_repair_worker_statistics() {
let config = RepairWorkerConfig::default();
let (result_tx, _result_rx) = mpsc::channel(100);
let worker = RepairWorker::new(config, result_tx);
let stats = worker.statistics().await;
assert_eq!(stats.total_tasks_processed, 0);
assert_eq!(stats.successful_repairs, 0);
assert_eq!(stats.failed_repairs, 0);
assert_eq!(stats.status, WorkerStatus::Idle);
}
}
crates/ahm/src/heal/validation.rs
-453
View File
@@ -1,453 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{
collections::HashMap,
sync::Arc,
time::{Duration, Instant, SystemTime},
};
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use crate::error::Result;
use super::{HealResult, HealTask};
/// Configuration for validation operations
#[derive(Debug, Clone)]
pub struct ValidationConfig {
/// Whether to enable validation after repair
pub enable_post_repair_validation: bool,
/// Timeout for validation operations
pub validation_timeout: Duration,
/// Whether to enable detailed validation logging
pub enable_detailed_logging: bool,
/// Maximum number of validation retries
pub max_validation_retries: u32,
/// Validation retry delay
pub validation_retry_delay: Duration,
}
impl Default for ValidationConfig {
fn default() -> Self {
Self {
enable_post_repair_validation: true,
validation_timeout: Duration::from_secs(60), // 1 minute
max_validation_retries: 3,
validation_retry_delay: Duration::from_secs(5),
enable_detailed_logging: true,
}
}
}
/// Validation result for a repair operation
#[derive(Debug, Clone)]
pub struct ValidationResult {
/// Whether validation passed
pub passed: bool,
/// Validation type
pub validation_type: ValidationType,
/// Detailed validation message
pub message: String,
/// Time taken for validation
pub duration: Duration,
/// Validation timestamp
pub timestamp: SystemTime,
/// Additional validation metadata
pub metadata: Option<serde_json::Value>,
}
/// Types of validation that can be performed
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ValidationType {
/// Checksum validation
Checksum,
/// File existence validation
FileExistence,
/// File size validation
FileSize,
/// File permissions validation
FilePermissions,
/// Metadata consistency validation
MetadataConsistency,
/// Replication status validation
ReplicationStatus,
/// Data integrity validation
DataIntegrity,
/// Custom validation
Custom(String),
}
/// Statistics for validation operations
#[derive(Debug, Clone, Default)]
pub struct ValidationStatistics {
/// Total number of validations performed
pub total_validations: u64,
/// Number of successful validations
pub successful_validations: u64,
/// Number of failed validations
pub failed_validations: u64,
/// Total time spent on validation
pub total_validation_time: Duration,
/// Average validation time
pub average_validation_time: Duration,
/// Number of validation retries
pub total_validation_retries: u64,
/// Last validation time
pub last_validation_time: Option<SystemTime>,
}
/// Validator for repair operations
pub struct HealValidator {
config: ValidationConfig,
statistics: Arc<RwLock<ValidationStatistics>>,
}
impl HealValidator {
/// Create a new validator
pub fn new(config: ValidationConfig) -> Self {
Self {
config,
statistics: Arc::new(RwLock::new(ValidationStatistics::default())),
}
}
/// Validate a repair operation
pub async fn validate_repair(&self, task: &HealTask, result: &HealResult) -> Result<Vec<ValidationResult>> {
if !self.config.enable_post_repair_validation {
return Ok(Vec::new());
}
let start_time = Instant::now();
let mut validation_results = Vec::new();
info!("Starting validation for repair task: {}", task.id);
// Perform different types of validation based on the issue type
match task.issue.issue_type {
crate::scanner::HealthIssueType::MissingReplica => {
validation_results.extend(self.validate_replica_repair(task, result).await?);
}
crate::scanner::HealthIssueType::ChecksumMismatch => {
validation_results.extend(self.validate_checksum_repair(task, result).await?);
}
crate::scanner::HealthIssueType::DiskReadError => {
validation_results.extend(self.validate_disk_repair(task, result).await?);
}
_ => {
validation_results.extend(self.validate_generic_repair(task, result).await?);
}
}
let duration = start_time.elapsed();
// Update statistics
{
let mut stats = self.statistics.write().await;
stats.total_validations += validation_results.len() as u64;
stats.total_validation_time += duration;
stats.average_validation_time = if stats.total_validations > 0 {
Duration::from_secs_f64(
stats.total_validation_time.as_secs_f64() / stats.total_validations as f64
)
} else {
Duration::ZERO
};
stats.last_validation_time = Some(SystemTime::now());
let successful_count = validation_results.iter().filter(|r| r.passed).count();
let failed_count = validation_results.len() - successful_count;
stats.successful_validations += successful_count as u64;
stats.failed_validations += failed_count as u64;
}
if self.config.enable_detailed_logging {
debug!("Validation completed for task {}: {} passed, {} failed",
task.id,
validation_results.iter().filter(|r| r.passed).count(),
validation_results.iter().filter(|r| !r.passed).count()
);
}
Ok(validation_results)
}
/// Validate replica repair
async fn validate_replica_repair(&self, task: &HealTask, _result: &HealResult) -> Result<Vec<ValidationResult>> {
let mut results = Vec::new();
// Validate file existence
let existence_result = self.validate_file_existence(&task.issue.bucket, &task.issue.object).await;
results.push(existence_result);
// Validate replication status
let replication_result = self.validate_replication_status(&task.issue.bucket, &task.issue.object).await;
results.push(replication_result);
Ok(results)
}
/// Validate checksum repair
async fn validate_checksum_repair(&self, task: &HealTask, _result: &HealResult) -> Result<Vec<ValidationResult>> {
let mut results = Vec::new();
// Validate checksum
let checksum_result = self.validate_checksum(&task.issue.bucket, &task.issue.object).await;
results.push(checksum_result);
// Validate data integrity
let integrity_result = self.validate_data_integrity(&task.issue.bucket, &task.issue.object).await;
results.push(integrity_result);
Ok(results)
}
/// Validate disk repair
async fn validate_disk_repair(&self, task: &HealTask, _result: &HealResult) -> Result<Vec<ValidationResult>> {
let mut results = Vec::new();
// Validate file existence
let existence_result = self.validate_file_existence(&task.issue.bucket, &task.issue.object).await;
results.push(existence_result);
// Validate file permissions
let permissions_result = self.validate_file_permissions(&task.issue.bucket, &task.issue.object).await;
results.push(permissions_result);
Ok(results)
}
/// Validate generic repair
async fn validate_generic_repair(&self, task: &HealTask, _result: &HealResult) -> Result<Vec<ValidationResult>> {
let mut results = Vec::new();
// Validate file existence
let existence_result = self.validate_file_existence(&task.issue.bucket, &task.issue.object).await;
results.push(existence_result);
// Validate metadata consistency
let metadata_result = self.validate_metadata_consistency(&task.issue.bucket, &task.issue.object).await;
results.push(metadata_result);
Ok(results)
}
/// Validate file existence
async fn validate_file_existence(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate file existence check
tokio::time::sleep(Duration::from_millis(10)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::FileExistence,
message: format!("File existence validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Validate checksum
async fn validate_checksum(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate checksum validation
tokio::time::sleep(Duration::from_millis(20)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::Checksum,
message: format!("Checksum validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Validate replication status
async fn validate_replication_status(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate replication status validation
tokio::time::sleep(Duration::from_millis(15)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::ReplicationStatus,
message: format!("Replication status validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Validate file permissions
async fn validate_file_permissions(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate file permissions validation
tokio::time::sleep(Duration::from_millis(5)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::FilePermissions,
message: format!("File permissions validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Validate metadata consistency
async fn validate_metadata_consistency(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate metadata consistency validation
tokio::time::sleep(Duration::from_millis(25)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::MetadataConsistency,
message: format!("Metadata consistency validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Validate data integrity
async fn validate_data_integrity(&self, bucket: &str, object: &str) -> ValidationResult {
let start_time = Instant::now();
// Simulate data integrity validation
tokio::time::sleep(Duration::from_millis(30)).await;
let duration = start_time.elapsed();
let passed = true; // Simulate successful validation
ValidationResult {
passed,
validation_type: ValidationType::DataIntegrity,
message: format!("Data integrity validation for {}/{}", bucket, object),
duration,
timestamp: SystemTime::now(),
metadata: None,
}
}
/// Get validation statistics
pub async fn statistics(&self) -> ValidationStatistics {
self.statistics.read().await.clone()
}
/// Reset validation statistics
pub async fn reset_statistics(&self) {
let mut stats = self.statistics.write().await;
*stats = ValidationStatistics::default();
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::scanner::{HealthIssue, HealthIssueType, Severity};
#[tokio::test]
async fn test_validator_creation() {
let config = ValidationConfig::default();
let validator = HealValidator::new(config);
let stats = validator.statistics().await;
assert_eq!(stats.total_validations, 0);
}
#[tokio::test]
async fn test_validate_repair() {
let config = ValidationConfig::default();
let validator = HealValidator::new(config);
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = super::HealTask::new(issue);
let result = super::HealResult {
success: true,
original_issue: task.issue.clone(),
repair_duration: Duration::from_secs(1),
retry_attempts: 0,
error_message: None,
metadata: None,
completed_at: SystemTime::now(),
};
let validation_results = validator.validate_repair(&task, &result).await.unwrap();
assert!(!validation_results.is_empty());
let stats = validator.statistics().await;
assert_eq!(stats.total_validations, validation_results.len() as u64);
}
#[tokio::test]
async fn test_validation_disabled() {
let mut config = ValidationConfig::default();
config.enable_post_repair_validation = false;
let validator = HealValidator::new(config);
let issue = HealthIssue {
issue_type: HealthIssueType::MissingReplica,
severity: Severity::Critical,
bucket: "test-bucket".to_string(),
object: "test-object".to_string(),
description: "Test issue".to_string(),
metadata: None,
};
let task = super::HealTask::new(issue);
let result = super::HealResult {
success: true,
original_issue: task.issue.clone(),
repair_duration: Duration::from_secs(1),
retry_attempts: 0,
error_message: None,
metadata: None,
completed_at: SystemTime::now(),
};
let validation_results = validator.validate_repair(&task, &result).await.unwrap();
assert!(validation_results.is_empty());
}
}