use super::config::*;
use super::optimizer::{Adaptation, AdaptationPriority, AdaptationType};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
#[derive(Debug, Clone, Serialize)]
pub struct ResourceUsage {
pub memory_usage_mb: usize,
pub cpu_usage_percent: f64,
pub gpu_usage_percent: Option<f64>,
pub network_io_mbps: f64,
pub disk_io_mbps: f64,
pub active_threads: usize,
#[serde(skip)]
pub timestamp: Instant,
}
#[derive(Debug, Clone)]
pub struct ResourceBudget {
pub memory_budget: MemoryBudget,
pub cpu_budget: CpuBudget,
pub network_budget: NetworkBudget,
pub time_budget: TimeBudget,
pub enforcement_strategy: BudgetEnforcementStrategy,
pub flexibility: f64,
}
#[derive(Debug, Clone)]
pub struct MemoryBudget {
pub max_allocation_mb: usize,
pub soft_limit_mb: usize,
pub cleanup_threshold: f64,
pub enable_compression: bool,
pub priority_levels: Vec<MemoryPriority>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum MemoryPriority {
Critical,
High,
Normal,
Low,
Temporary,
}
#[derive(Debug, Clone)]
pub struct CpuBudget {
pub max_utilization: f64,
pub target_utilization: f64,
pub max_threads: usize,
pub thread_priority: ThreadPriorityConfig,
pub cpu_affinity: Option<Vec<usize>>,
}
#[derive(Debug, Clone)]
pub struct ThreadPriorityConfig {
pub high_priority_threads: usize,
pub normal_priority_threads: usize,
pub background_threads: usize,
pub dynamic_priority: bool,
}
#[derive(Debug, Clone)]
pub struct NetworkBudget {
pub max_bandwidth_mbps: f64,
pub priority_allocation: HashMap<String, f64>,
pub enable_traffic_shaping: bool,
pub qos_settings: QoSSettings,
}
#[derive(Debug, Clone)]
pub struct QoSSettings {
pub max_latency_ms: u64,
pub jitter_tolerance_ms: u64,
pub packet_loss_tolerance: f64,
pub traffic_classes: Vec<TrafficClass>,
}
#[derive(Debug, Clone)]
pub struct TrafficClass {
pub name: String,
pub priority: u8,
pub bandwidth_guarantee: f64,
pub max_bandwidth: f64,
}
#[derive(Debug, Clone)]
pub struct TimeBudget {
pub max_batch_processing_time: Duration,
pub target_batch_processing_time: Duration,
pub operation_timeout: Duration,
pub deadline_enforcement: DeadlineEnforcement,
}
#[derive(Debug, Clone)]
pub enum DeadlineEnforcement {
Strict,
Soft,
BestEffort,
Adaptive,
}
#[derive(Debug, Clone)]
pub enum BudgetEnforcementStrategy {
Strict,
Throttling,
LoadShedding,
GracefulDegradation,
Adaptive,
}
pub struct ResourceManager {
config: ResourceConfig,
current_usage: Arc<Mutex<ResourceUsage>>,
usage_history: Arc<Mutex<VecDeque<ResourceUsage>>>,
budget: ResourceBudget,
allocations: Arc<Mutex<HashMap<String, ResourceAllocation>>>,
monitoring_handle: Option<std::thread::JoinHandle<()>>,
predictor: ResourcePredictor,
optimizer: ResourceOptimizer,
alert_system: ResourceAlertSystem,
}
#[derive(Debug, Clone)]
pub struct ResourceAllocation {
pub component_name: String,
pub allocated_memory_mb: usize,
pub allocated_cpu_percent: f64,
pub allocated_bandwidth_mbps: f64,
pub priority: ResourcePriority,
pub allocation_time: Instant,
pub last_access: Instant,
pub usage_stats: ComponentUsageStats,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum ResourcePriority {
Critical = 0,
High = 1,
Normal = 2,
Low = 3,
Temporary = 4,
}
#[derive(Debug, Clone)]
pub struct ComponentUsageStats {
pub peak_memory_mb: usize,
pub avg_memory_mb: usize,
pub peak_cpu_percent: f64,
pub avg_cpu_percent: f64,
pub total_processing_time: Duration,
pub operation_count: u64,
pub efficiency_score: f64,
}
pub struct ResourcePredictor {
usage_patterns: VecDeque<ResourceUsage>,
prediction_horizon: usize,
prediction_accuracy: HashMap<String, f64>,
seasonal_patterns: HashMap<String, Vec<f64>>,
trend_analysis: ResourceTrendAnalysis,
}
#[derive(Debug, Clone)]
pub struct ResourceTrendAnalysis {
pub memory_trend: TrendDirection,
pub cpu_trend: TrendDirection,
pub network_trend: TrendDirection,
pub trend_confidence: f64,
pub trend_stability: f64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TrendDirection {
Increasing,
Decreasing,
Stable,
Oscillating,
Unknown,
}
pub struct ResourceOptimizer {
strategy: ResourceOptimizationStrategy,
optimization_history: VecDeque<OptimizationEvent>,
performance_impact: HashMap<String, f64>,
constraints: OptimizationConstraints,
}
#[derive(Debug, Clone)]
pub enum ResourceOptimizationStrategy {
Conservative,
Aggressive,
Balanced,
PowerEfficient,
LatencyOptimized,
ThroughputOptimized,
}
#[derive(Debug, Clone)]
pub struct OptimizationEvent {
pub timestamp: Instant,
pub optimization_type: String,
pub affected_resources: Vec<String>,
pub resource_deltas: HashMap<String, f64>,
pub performance_impact: f64,
pub success: bool,
}
#[derive(Debug, Clone)]
pub struct OptimizationConstraints {
pub min_guarantees: HashMap<String, f64>,
pub max_limits: HashMap<String, f64>,
pub change_rate_limits: HashMap<String, f64>,
pub stability_requirements: StabilityRequirements,
}
#[derive(Debug, Clone)]
pub struct StabilityRequirements {
pub min_stable_period: Duration,
pub max_change_frequency: f64,
pub prevent_oscillation: bool,
pub hysteresis_factor: f64,
}
pub struct ResourceAlertSystem {
thresholds: ResourceThresholds,
active_alerts: VecDeque<ResourceAlert>,
alert_history: VecDeque<ResourceAlert>,
alert_handlers: Vec<Box<dyn AlertHandler>>,
}
#[derive(Debug, Clone)]
pub struct ResourceThresholds {
pub memory_thresholds: ThresholdSet,
pub cpu_thresholds: ThresholdSet,
pub network_thresholds: ThresholdSet,
pub response_time_thresholds: ThresholdSet,
}
#[derive(Debug, Clone)]
pub struct ThresholdSet {
pub warning: f64,
pub critical: f64,
pub emergency: f64,
pub recovery: f64,
}
#[derive(Debug, Clone)]
pub struct ResourceAlert {
pub id: String,
pub timestamp: Instant,
pub severity: AlertSeverity,
pub resource_type: String,
pub current_value: f64,
pub threshold_value: f64,
pub message: String,
pub suggested_actions: Vec<String>,
pub auto_resolution_attempts: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum AlertSeverity {
Info,
Warning,
Error,
Critical,
Emergency,
}
pub trait AlertHandler: Send + Sync {
fn handle_alert(&self, alert: &ResourceAlert) -> Result<(), String>;
fn priority(&self) -> u32;
fn can_handle(&self, alert: &ResourceAlert) -> bool;
}
impl ResourceManager {
pub fn new(config: &StreamingConfig) -> Result<Self, String> {
let resource_config = config.resource_config.clone();
let budget = ResourceBudget {
memory_budget: MemoryBudget {
max_allocation_mb: resource_config.max_memory_mb,
soft_limit_mb: (resource_config.max_memory_mb as f64 * 0.8) as usize,
cleanup_threshold: resource_config.cleanup_threshold,
enable_compression: true,
priority_levels: vec![
MemoryPriority::Critical,
MemoryPriority::High,
MemoryPriority::Normal,
MemoryPriority::Low,
],
},
cpu_budget: CpuBudget {
max_utilization: resource_config.max_cpu_percent,
target_utilization: resource_config.max_cpu_percent * 0.8,
max_threads: num_cpus::get(),
thread_priority: ThreadPriorityConfig {
high_priority_threads: 2,
normal_priority_threads: num_cpus::get() - 2,
background_threads: 1,
dynamic_priority: true,
},
cpu_affinity: None,
},
network_budget: NetworkBudget {
max_bandwidth_mbps: 100.0, priority_allocation: HashMap::new(),
enable_traffic_shaping: false,
qos_settings: QoSSettings {
max_latency_ms: 100,
jitter_tolerance_ms: 10,
packet_loss_tolerance: 0.1,
traffic_classes: Vec::new(),
},
},
time_budget: TimeBudget {
max_batch_processing_time: Duration::from_secs(30),
target_batch_processing_time: Duration::from_secs(10),
operation_timeout: Duration::from_secs(60),
deadline_enforcement: DeadlineEnforcement::Soft,
},
enforcement_strategy: match resource_config.allocation_strategy {
ResourceAllocationStrategy::Static => BudgetEnforcementStrategy::Strict,
ResourceAllocationStrategy::Dynamic => BudgetEnforcementStrategy::Throttling,
ResourceAllocationStrategy::Adaptive => BudgetEnforcementStrategy::Adaptive,
_ => BudgetEnforcementStrategy::GracefulDegradation,
},
flexibility: 0.2,
};
let predictor = ResourcePredictor::new();
let optimizer = ResourceOptimizer::new(ResourceOptimizationStrategy::Balanced);
let alert_system = ResourceAlertSystem::new();
Ok(Self {
config: resource_config,
current_usage: Arc::new(Mutex::new(ResourceUsage::default())),
usage_history: Arc::new(Mutex::new(VecDeque::with_capacity(1000))),
budget,
allocations: Arc::new(Mutex::new(HashMap::new())),
monitoring_handle: None,
predictor,
optimizer,
alert_system,
})
}
pub fn start_monitoring(&mut self) -> Result<(), String> {
if self.monitoring_handle.is_some() {
return Ok(()); }
let current_usage = Arc::clone(&self.current_usage);
let usage_history = Arc::clone(&self.usage_history);
let monitoring_frequency = self.config.monitoring_frequency;
let handle = std::thread::spawn(move || {
loop {
if let Ok(usage) = Self::collect_resource_usage() {
{
let mut current = current_usage.lock().expect("lock poisoned");
*current = usage.clone();
}
{
let mut history = usage_history.lock().expect("lock poisoned");
if history.len() >= 1000 {
history.pop_front();
}
history.push_back(usage);
}
}
std::thread::sleep(monitoring_frequency);
}
});
self.monitoring_handle = Some(handle);
Ok(())
}
fn collect_resource_usage() -> Result<ResourceUsage, String> {
let mut usage = ResourceUsage {
timestamp: Instant::now(),
..Default::default()
};
let info = sysinfo::System::new_all().used_memory();
usage.memory_usage_mb = (info / 1024 / 1024) as usize;
usage.cpu_usage_percent = 50.0;
usage.network_io_mbps = 1.0;
usage.disk_io_mbps = 5.0;
usage.active_threads = std::thread::available_parallelism()
.map(|n| n.get())
.unwrap_or(1);
Ok(usage)
}
pub fn allocate_resources(
&mut self,
component_name: &str,
memory_mb: usize,
cpu_percent: f64,
priority: ResourcePriority,
) -> Result<(), String> {
self.check_budget_constraints(memory_mb, cpu_percent)?;
let allocation = ResourceAllocation {
component_name: component_name.to_string(),
allocated_memory_mb: memory_mb,
allocated_cpu_percent: cpu_percent,
allocated_bandwidth_mbps: 0.0, priority,
allocation_time: Instant::now(),
last_access: Instant::now(),
usage_stats: ComponentUsageStats {
peak_memory_mb: 0,
avg_memory_mb: 0,
peak_cpu_percent: 0.0,
avg_cpu_percent: 0.0,
total_processing_time: Duration::ZERO,
operation_count: 0,
efficiency_score: 1.0,
},
};
let mut allocations = self.allocations.lock().expect("lock poisoned");
allocations.insert(component_name.to_string(), allocation);
Ok(())
}
fn check_budget_constraints(&self, memory_mb: usize, cpu_percent: f64) -> Result<(), String> {
let allocations = self.allocations.lock().expect("lock poisoned");
let total_memory: usize = allocations
.values()
.map(|a| a.allocated_memory_mb)
.sum::<usize>()
+ memory_mb;
let total_cpu: f64 = allocations
.values()
.map(|a| a.allocated_cpu_percent)
.sum::<f64>()
+ cpu_percent;
if total_memory > self.budget.memory_budget.max_allocation_mb {
return Err(format!(
"Memory allocation would exceed budget: {} MB > {} MB",
total_memory, self.budget.memory_budget.max_allocation_mb
));
}
if total_cpu > self.budget.cpu_budget.max_utilization {
return Err(format!(
"CPU allocation would exceed budget: {:.2}% > {:.2}%",
total_cpu, self.budget.cpu_budget.max_utilization
));
}
Ok(())
}
pub fn update_utilization(&mut self) -> Result<(), String> {
let current_usage = self.current_usage.lock().expect("lock poisoned").clone();
let alerts = self.alert_system.check_thresholds(¤t_usage)?;
for alert in alerts {
self.alert_system.handle_alert(alert)?;
}
self.predictor.update(¤t_usage)?;
if self.config.enable_dynamic_allocation {
self.optimizer
.check_optimization_opportunities(¤t_usage, &self.allocations)?;
}
Ok(())
}
pub fn has_sufficient_resources_for_processing(&self) -> Result<bool, String> {
let current_usage = self.current_usage.lock().expect("lock poisoned");
let memory_available = current_usage.memory_usage_mb
< (self.budget.memory_budget.soft_limit_mb as f64 * 0.9) as usize;
let cpu_available =
current_usage.cpu_usage_percent < self.budget.cpu_budget.target_utilization * 0.9;
Ok(memory_available && cpu_available)
}
pub fn compute_allocation_adaptation(&mut self) -> Result<Option<Adaptation<f32>>, String> {
let current_usage = self.current_usage.lock().expect("lock poisoned");
if current_usage.memory_usage_mb > self.budget.memory_budget.soft_limit_mb {
let adaptation = Adaptation {
adaptation_type: AdaptationType::ResourceAllocation,
magnitude: -0.2, target_component: "memory_manager".to_string(),
parameters: std::collections::HashMap::new(),
priority: AdaptationPriority::High,
timestamp: Instant::now(),
};
return Ok(Some(adaptation));
}
if current_usage.cpu_usage_percent > self.budget.cpu_budget.target_utilization {
let adaptation = Adaptation {
adaptation_type: AdaptationType::ResourceAllocation,
magnitude: -0.15, target_component: "cpu_manager".to_string(),
parameters: std::collections::HashMap::new(),
priority: AdaptationPriority::High,
timestamp: Instant::now(),
};
return Ok(Some(adaptation));
}
Ok(None)
}
pub fn apply_allocation_adaptation(
&mut self,
adaptation: &Adaptation<f32>,
) -> Result<(), String> {
if adaptation.adaptation_type == AdaptationType::ResourceAllocation {
match adaptation.target_component.as_str() {
"memory_manager" => {
let factor = 1.0 + adaptation.magnitude;
let mut allocations = self.allocations.lock().expect("lock poisoned");
for allocation in allocations.values_mut() {
if allocation.priority >= ResourcePriority::Normal {
allocation.allocated_memory_mb =
((allocation.allocated_memory_mb as f32) * factor) as usize;
}
}
}
"cpu_manager" => {
let factor = 1.0 + adaptation.magnitude;
let mut allocations = self.allocations.lock().expect("lock poisoned");
for allocation in allocations.values_mut() {
if allocation.priority >= ResourcePriority::Normal {
allocation.allocated_cpu_percent *= factor as f64;
}
}
}
_ => {
}
}
}
Ok(())
}
pub fn current_usage(&self) -> Result<ResourceUsage, String> {
Ok(self.current_usage.lock().expect("lock poisoned").clone())
}
pub fn get_usage_history(&self, count: usize) -> Vec<ResourceUsage> {
let history = self.usage_history.lock().expect("lock poisoned");
history.iter().rev().take(count).cloned().collect()
}
pub fn get_diagnostics(&self) -> ResourceDiagnostics {
let current_usage = self.current_usage.lock().expect("lock poisoned");
let allocations = self.allocations.lock().expect("lock poisoned");
ResourceDiagnostics {
current_usage: current_usage.clone(),
total_allocations: allocations.len(),
memory_utilization: (current_usage.memory_usage_mb as f64
/ self.budget.memory_budget.max_allocation_mb as f64)
* 100.0,
cpu_utilization: current_usage.cpu_usage_percent,
active_alerts: self.alert_system.active_alerts.len(),
budget_violations: 0, }
}
}
impl ResourcePredictor {
fn new() -> Self {
Self {
usage_patterns: VecDeque::with_capacity(1000),
prediction_horizon: 10,
prediction_accuracy: HashMap::new(),
seasonal_patterns: HashMap::new(),
trend_analysis: ResourceTrendAnalysis {
memory_trend: TrendDirection::Unknown,
cpu_trend: TrendDirection::Unknown,
network_trend: TrendDirection::Unknown,
trend_confidence: 0.0,
trend_stability: 0.0,
},
}
}
fn update(&mut self, usage: &ResourceUsage) -> Result<(), String> {
if self.usage_patterns.len() >= 1000 {
self.usage_patterns.pop_front();
}
self.usage_patterns.push_back(usage.clone());
if self.usage_patterns.len() >= 10 {
self.update_trend_analysis()?;
}
Ok(())
}
fn update_trend_analysis(&mut self) -> Result<(), String> {
let recent_patterns: Vec<_> = self.usage_patterns.iter().rev().take(10).collect();
let memory_values: Vec<f64> = recent_patterns
.iter()
.map(|u| u.memory_usage_mb as f64)
.collect();
self.trend_analysis.memory_trend = self.analyze_trend(&memory_values);
let cpu_values: Vec<f64> = recent_patterns
.iter()
.map(|u| u.cpu_usage_percent)
.collect();
self.trend_analysis.cpu_trend = self.analyze_trend(&cpu_values);
self.trend_analysis.trend_confidence =
self.calculate_trend_confidence(&memory_values, &cpu_values);
Ok(())
}
fn analyze_trend(&self, values: &[f64]) -> TrendDirection {
if values.len() < 3 {
return TrendDirection::Unknown;
}
let first_half: f64 =
values.iter().take(values.len() / 2).sum::<f64>() / (values.len() / 2) as f64;
let second_half: f64 = values.iter().skip(values.len() / 2).sum::<f64>()
/ (values.len() - values.len() / 2) as f64;
let change_threshold = 0.05; let relative_change = (second_half - first_half) / first_half.max(1.0);
if relative_change > change_threshold {
TrendDirection::Increasing
} else if relative_change < -change_threshold {
TrendDirection::Decreasing
} else {
TrendDirection::Stable
}
}
fn calculate_trend_confidence(&self, memory_values: &[f64], cpu_values: &[f64]) -> f64 {
let memory_variance = self.calculate_variance(memory_values);
let cpu_variance = self.calculate_variance(cpu_values);
let memory_confidence = 1.0 / (1.0 + memory_variance / 100.0);
let cpu_confidence = 1.0 / (1.0 + cpu_variance / 100.0);
(memory_confidence + cpu_confidence) / 2.0
}
fn calculate_variance(&self, values: &[f64]) -> f64 {
if values.len() < 2 {
return 0.0;
}
let mean = values.iter().sum::<f64>() / values.len() as f64;
let variance = values.iter().map(|x| (x - mean).powi(2)).sum::<f64>() / values.len() as f64;
variance
}
}
impl ResourceOptimizer {
fn new(strategy: ResourceOptimizationStrategy) -> Self {
Self {
strategy,
optimization_history: VecDeque::with_capacity(100),
performance_impact: HashMap::new(),
constraints: OptimizationConstraints {
min_guarantees: HashMap::new(),
max_limits: HashMap::new(),
change_rate_limits: HashMap::new(),
stability_requirements: StabilityRequirements {
min_stable_period: Duration::from_secs(60),
max_change_frequency: 0.1, prevent_oscillation: true,
hysteresis_factor: 0.1,
},
},
}
}
fn check_optimization_opportunities(
&mut self,
current_usage: &ResourceUsage,
allocations: &Arc<Mutex<HashMap<String, ResourceAllocation>>>,
) -> Result<(), String> {
match self.strategy {
ResourceOptimizationStrategy::Balanced => {
self.check_balanced_optimization(current_usage, allocations)?;
}
ResourceOptimizationStrategy::PowerEfficient => {
self.check_power_optimization(current_usage, allocations)?;
}
ResourceOptimizationStrategy::LatencyOptimized => {
self.check_latency_optimization(current_usage, allocations)?;
}
_ => {
}
}
Ok(())
}
fn check_balanced_optimization(
&mut self,
current_usage: &ResourceUsage,
_allocations: &Arc<Mutex<HashMap<String, ResourceAllocation>>>,
) -> Result<(), String> {
let memory_utilization = current_usage.memory_usage_mb as f64 / 1024.0; let cpu_utilization = current_usage.cpu_usage_percent;
if (memory_utilization > 80.0 && cpu_utilization < 40.0)
|| (cpu_utilization > 80.0 && memory_utilization < 40.0)
{
let optimization_event = OptimizationEvent {
timestamp: Instant::now(),
optimization_type: "resource_rebalancing".to_string(),
affected_resources: vec!["memory".to_string(), "cpu".to_string()],
resource_deltas: HashMap::new(),
performance_impact: 0.05, success: false, };
if self.optimization_history.len() >= 100 {
self.optimization_history.pop_front();
}
self.optimization_history.push_back(optimization_event);
}
Ok(())
}
fn check_power_optimization(
&mut self,
_current_usage: &ResourceUsage,
_allocations: &Arc<Mutex<HashMap<String, ResourceAllocation>>>,
) -> Result<(), String> {
Ok(())
}
fn check_latency_optimization(
&mut self,
_current_usage: &ResourceUsage,
_allocations: &Arc<Mutex<HashMap<String, ResourceAllocation>>>,
) -> Result<(), String> {
Ok(())
}
}
impl ResourceAlertSystem {
fn new() -> Self {
Self {
thresholds: ResourceThresholds {
memory_thresholds: ThresholdSet {
warning: 70.0,
critical: 85.0,
emergency: 95.0,
recovery: 65.0,
},
cpu_thresholds: ThresholdSet {
warning: 75.0,
critical: 90.0,
emergency: 98.0,
recovery: 70.0,
},
network_thresholds: ThresholdSet {
warning: 80.0,
critical: 95.0,
emergency: 99.0,
recovery: 75.0,
},
response_time_thresholds: ThresholdSet {
warning: 1000.0, critical: 5000.0, emergency: 10000.0, recovery: 500.0, },
},
active_alerts: VecDeque::new(),
alert_history: VecDeque::with_capacity(1000),
alert_handlers: Vec::new(),
}
}
fn check_thresholds(&mut self, usage: &ResourceUsage) -> Result<Vec<ResourceAlert>, String> {
let mut alerts = Vec::new();
let memory_percent = (usage.memory_usage_mb as f64 / 1024.0) * 100.0; if let Some(alert) =
self.check_threshold("memory", memory_percent, &self.thresholds.memory_thresholds)?
{
alerts.push(alert);
}
if let Some(alert) = self.check_threshold(
"cpu",
usage.cpu_usage_percent,
&self.thresholds.cpu_thresholds,
)? {
alerts.push(alert);
}
Ok(alerts)
}
fn check_threshold(
&self,
resource_type: &str,
current_value: f64,
thresholds: &ThresholdSet,
) -> Result<Option<ResourceAlert>, String> {
let severity = if current_value >= thresholds.emergency {
AlertSeverity::Emergency
} else if current_value >= thresholds.critical {
AlertSeverity::Critical
} else if current_value >= thresholds.warning {
AlertSeverity::Warning
} else {
return Ok(None);
};
let threshold_value = match severity {
AlertSeverity::Emergency => thresholds.emergency,
AlertSeverity::Critical => thresholds.critical,
AlertSeverity::Warning => thresholds.warning,
_ => thresholds.warning,
};
let suggested_actions = self.generate_suggested_actions(resource_type, &severity);
let alert = ResourceAlert {
id: format!("{}_{}", resource_type, Instant::now().elapsed().as_nanos()),
timestamp: Instant::now(),
severity,
resource_type: resource_type.to_string(),
current_value,
threshold_value,
message: format!(
"{} usage is {:.2}% (threshold: {:.2}%)",
resource_type, current_value, threshold_value
),
suggested_actions,
auto_resolution_attempts: 0,
};
Ok(Some(alert))
}
fn generate_suggested_actions(
&self,
resource_type: &str,
severity: &AlertSeverity,
) -> Vec<String> {
match (resource_type, severity) {
("memory", AlertSeverity::Critical | AlertSeverity::Emergency) => vec![
"Reduce buffer sizes".to_string(),
"Clear caches".to_string(),
"Reduce batch sizes".to_string(),
],
("memory", AlertSeverity::Warning) => vec![
"Monitor memory usage trends".to_string(),
"Consider reducing buffer sizes".to_string(),
],
("cpu", AlertSeverity::Critical | AlertSeverity::Emergency) => vec![
"Reduce processing frequency".to_string(),
"Lower thread count".to_string(),
"Defer non-critical operations".to_string(),
],
("cpu", AlertSeverity::Warning) => vec![
"Monitor CPU usage patterns".to_string(),
"Consider load balancing".to_string(),
],
_ => vec!["Monitor resource usage".to_string()],
}
}
fn handle_alert(&mut self, alert: ResourceAlert) -> Result<(), String> {
self.active_alerts.push_back(alert.clone());
if self.alert_history.len() >= 1000 {
self.alert_history.pop_front();
}
self.alert_history.push_back(alert.clone());
for handler in &self.alert_handlers {
if handler.can_handle(&alert) {
handler.handle_alert(&alert)?;
}
}
Ok(())
}
}
#[derive(Debug, Clone)]
pub struct ResourceDiagnostics {
pub current_usage: ResourceUsage,
pub total_allocations: usize,
pub memory_utilization: f64,
pub cpu_utilization: f64,
pub active_alerts: usize,
pub budget_violations: usize,
}
impl Default for ResourceUsage {
fn default() -> Self {
Self {
memory_usage_mb: 0,
cpu_usage_percent: 0.0,
gpu_usage_percent: None,
network_io_mbps: 0.0,
disk_io_mbps: 0.0,
active_threads: 0,
timestamp: Instant::now(),
}
}
}