use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, RwLock};
use std::time::Duration;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Metrics {
pub timestamp: DateTime<Utc>,
pub cpu_usage: f64,
pub memory_usage: f64,
pub network_io: u64,
pub disk_io: u64,
pub active_connections: u64,
pub request_rate: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeFrame {
pub start_time: DateTime<Utc>,
pub end_time: DateTime<Utc>,
pub duration: Duration,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourcePrediction {
pub timeframe: TimeFrame,
pub predicted_cpu: f64,
pub predicted_memory: f64,
pub predicted_connections: u64,
pub confidence: f64,
pub factors: Vec<PredictionFactor>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PredictionFactor {
pub name: String,
pub impact: f64,
pub description: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TrendAnalysis {
pub trend_direction: TrendDirection,
pub growth_rate: f64,
pub seasonality: Option<SeasonalityPattern>,
pub anomalies: Vec<Anomaly>,
pub forecast_accuracy: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum TrendDirection {
Increasing,
Decreasing,
Stable,
Volatile,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SeasonalityPattern {
pub period: Duration,
pub amplitude: f64,
pub phase: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Anomaly {
pub timestamp: DateTime<Utc>,
pub metric_type: String,
pub deviation_score: f64,
pub description: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScalingStrategy {
pub trigger_metric: String,
pub threshold: f64,
pub action: ScalingAction,
pub cooldown_period: Duration,
pub max_instances: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScalingAction {
pub action_type: ActionType,
pub instance_count: u32,
pub resources: ResourceAllocation,
pub execution_window: TimeFrame,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum ActionType {
ScaleUp,
ScaleDown,
ScaleOut,
ScaleIn,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceAllocation {
pub cpu_cores: f64,
pub memory_gb: f64,
pub storage_gb: f64,
pub network_bandwidth_mbps: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScalingResult {
pub success: bool,
pub instances_before: u32,
pub instances_after: u32,
pub resources_before: ResourceAllocation,
pub resources_after: ResourceAllocation,
pub performance_impact: f64,
pub cost_impact: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Task {
pub id: String,
pub name: String,
pub priority: TaskPriority,
pub estimated_duration: Duration,
pub resource_requirements: ResourceRequirements,
pub dependencies: Vec<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
pub enum TaskPriority {
Critical,
High,
Medium,
Low,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceRequirements {
pub cpu: f64,
pub memory: f64,
pub estimated_duration: Duration,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Schedule {
pub tasks: Vec<ScheduledTask>,
pub total_duration: Duration,
pub resource_utilization: HashMap<String, f64>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduledTask {
pub task: Task,
pub scheduled_start: DateTime<Utc>,
pub scheduled_end: DateTime<Utc>,
pub allocated_resources: ResourceAllocation,
}
#[derive(Debug, Clone)]
pub struct PredictiveScaler {
predictor: Arc<RwLock<ResourcePredictor>>,
analyzer: Arc<RwLock<TrendAnalyzer>>,
scaler: Arc<RwLock<AutoScaler>>,
}
#[derive(Debug, Clone)]
pub struct ResourcePredictor {
model: Arc<PredictionModel>,
historical_data: Arc<RwLock<Vec<Metrics>>>,
}
#[derive(Debug, Clone)]
pub struct TrendAnalyzer {
patterns: Arc<RwLock<HashMap<String, SeasonalityPattern>>>,
anomaly_detector: Arc<AnomalyDetector>,
}
#[derive(Debug, Clone)]
pub struct AutoScaler {
current_capacity: Arc<RwLock<ResourceAllocation>>,
scaling_history: Arc<RwLock<Vec<ScalingAction>>>,
}
#[derive(Debug, Clone)]
pub struct PredictionModel {
model_type: ModelType,
parameters: HashMap<String, f64>,
}
#[derive(Debug, Clone)]
pub enum ModelType {
LinearRegression,
ARIMA,
ExponentialSmoothing,
Prophet,
}
#[derive(Debug, Clone)]
pub struct AnomalyDetector {
threshold: f64,
window_size: usize,
}
impl PredictiveScaler {
pub fn new() -> Self {
let predictor = Arc::new(RwLock::new(ResourcePredictor::new()));
let analyzer = Arc::new(RwLock::new(TrendAnalyzer::new()));
let scaler = Arc::new(RwLock::new(AutoScaler::new()));
Self {
predictor,
analyzer,
scaler,
}
}
pub async fn predict_resource_usage(
&self,
timeframe: TimeFrame,
) -> Result<ResourcePrediction, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(60)).await;
let predictor: _ = self.predictor.read().await;
let prediction: _ = predictor.predict(&timeframe).await?;
Ok(prediction)
}
pub async fn analyze_trends(
&self,
historical_data: &[Metrics],
) -> Result<TrendAnalysis, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(70)).await;
let analyzer: _ = self.analyzer.read().await;
let analysis: _ = analyzer.analyze(historical_data).await?;
Ok(analysis)
}
pub async fn suggest_scaling(
&self,
prediction: &ResourcePrediction,
) -> Result<ScalingStrategy, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
let mut strategy = ScalingStrategy {
trigger_metric: "cpu_usage".to_string(),
threshold: 70.0,
action: ScalingAction {
action_type: ActionType::ScaleOut,
instance_count: 2,
resources: ResourceAllocation {
cpu_cores: 2.0,
memory_gb: 4.0,
storage_gb: 100.0,
network_bandwidth_mbps: 1000,
},
execution_window: prediction.timeframe.clone(),
},
cooldown_period: Duration::minutes(5),
max_instances: 10,
};
if prediction.predicted_cpu > 80.0 {
strategy.action = ScalingAction {
action_type: ActionType::ScaleOut,
instance_count: ((prediction.predicted_cpu - 80.0) / 10.0).ceil() as u32,
resources: ResourceAllocation {
cpu_cores: 2.0,
memory_gb: 4.0,
storage_gb: 100.0,
network_bandwidth_mbps: 1000,
},
execution_window: prediction.timeframe.clone(),
};
strategy.threshold = 70.0;
} else if prediction.predicted_cpu < 30.0 {
strategy.action = ScalingAction {
action_type: ActionType::ScaleIn,
instance_count: 1,
resources: ResourceAllocation {
cpu_cores: 1.0,
memory_gb: 2.0,
storage_gb: 50.0,
network_bandwidth_mbps: 500,
},
execution_window: prediction.timeframe.clone(),
};
strategy.threshold = 20.0;
}
Ok(strategy)
}
pub async fn auto_scale(
&self,
strategy: &ScalingStrategy,
) -> Result<ScalingResult, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let scaler: _ = self.scaler.read().await;
let result: _ = scaler.execute(strategy).await?;
Ok(result)
}
pub async fn optimize_schedule(
&self,
tasks: &[Task],
) -> Result<Schedule, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(80)).await;
let mut sorted_tasks = tasks.to_vec();
sorted_tasks.sort_by(|a, b| b.priority.cmp(&a.priority));
let mut scheduled_tasks = Vec::new();
let mut current_time = Utc::now();
let mut total_duration = Duration::seconds(0);
for task in sorted_tasks {
let scheduled_start: _ = current_time;
let scheduled_end: _ = current_time + task.estimated_duration;
scheduled_tasks.push(ScheduledTask {
task: task.clone(),
scheduled_start,
scheduled_end,
allocated_resources: ResourceAllocation {
cpu_cores: task.resource_requirements.cpu,
memory_gb: task.resource_requirements.memory,
storage_gb: 10.0,
network_bandwidth_mbps: 100,
},
});
current_time = scheduled_end;
total_duration = total_duration + task.estimated_duration;
}
let mut resource_utilization = HashMap::new();
resource_utilization.insert("cpu".to_string(), 0.75);
resource_utilization.insert("memory".to_string(), 0.65);
Ok(Schedule {
tasks: scheduled_tasks,
total_duration,
resource_utilization,
})
}
pub async fn predict_execution_time(
&self,
task: &Task,
) -> Result<Duration, Box<dyn std::error::Error>> {
tokio::time::sleep(std::time::Duration::from_millis(30)).await;
let base_time: _ = task.estimated_duration.num_seconds() as f64;
let cpu_factor: _ = 1.0 / task.resource_requirements.cpu;
let memory_factor: _ = 1.0 / task.resource_requirements.memory;
let predicted_time: _ = base_time * cpu_factor * memory_factor;
let predicted_duration: _ = Duration::seconds(predicted_time.round() as i64);
Ok(predicted_duration)
}
}
impl ResourcePredictor {
pub fn new() -> Self {
Self {
model: Arc::new(PredictionModel::new(ModelType::LinearRegression)),
historical_data: Arc::new(RwLock::new(Vec::new())),
}
}
pub async fn predict(
&self,
timeframe: &TimeFrame,
) -> Result<ResourcePrediction, Box<dyn std::error::Error>> {
let historical_data: _ = self.historical_data.read().await;
if historical_data.is_empty() {
return Err("没有历史数据用于预测".into());
}
let last_data: _ = &historical_data[historical_data.len() - 1];
let predicted_cpu: _ = (last_data.cpu_usage * 1.1).min(100.0);
let predicted_memory: _ = (last_data.memory_usage * 1.05).min(100.0);
let predicted_connections: _ = (last_data.active_connections as f64 * 1.2) as u64;
let factors: _ = vec![
PredictionFactor {
name: "历史趋势".to_string(),
impact: 0.6,
description: "基于历史数据趋势".to_string(),
},
PredictionFactor {
name: "季节性".to_string(),
impact: 0.3,
description: "时间季节性影响".to_string(),
},
PredictionFactor {
name: "外部因素".to_string(),
impact: 0.1,
description: "外部负载变化".to_string(),
},
];
Ok(ResourcePrediction {
timeframe: timeframe.clone(),
predicted_cpu,
predicted_memory,
predicted_connections,
confidence: 0.82,
factors,
})
}
pub async fn add_historical_data(&self, metrics: Metrics) {
let mut data = self.historical_data.write().await;
data.push(metrics);
if data.len() > 1000 {
data.remove(0);
}
}
}
impl TrendAnalyzer {
pub fn new() -> Self {
Self {
patterns: Arc::new(RwLock::new(HashMap::new())),
anomaly_detector: Arc::new(AnomalyDetector::new(2.0, 50)),
}
}
pub async fn analyze(
&self,
data: &[Metrics],
) -> Result<TrendAnalysis, Box<dyn std::error::Error>> {
if data.len() < 2 {
return Err("数据点不足，无法分析趋势".into());
}
let first_cpu: _ = data.first().unwrap().cpu_usage;
let last_cpu: _ = data.last().unwrap().cpu_usage;
let trend_direction: _ = if last_cpu > first_cpu * 1.1 {
TrendDirection::Increasing
} else if last_cpu < first_cpu * 0.9 {
TrendDirection::Decreasing
} else {
TrendDirection::Stable
};
let time_span: _ =
(data.last().unwrap().timestamp - data.first().unwrap().timestamp).num_seconds() as f64;
let growth_rate: _ = if time_span > 0.0 {
(last_cpu - first_cpu) / time_span * 3600.0 } else {
0.0
};
let seasonality: _ = self.detect_seasonality(data);
let anomalies: _ = self.anomaly_detector.detect(data);
Ok(TrendAnalysis {
trend_direction,
growth_rate,
seasonality,
anomalies,
forecast_accuracy: 0.85,
})
}
fn detect_seasonality(&self, data: &[Metrics]) -> Option<SeasonalityPattern> {
if data.len() > 24 {
Some(SeasonalityPattern {
period: Duration::hours(24),
amplitude: 10.0,
phase: 0.0,
})
} else {
None
}
}
}
impl AutoScaler {
pub fn new() -> Self {
Self {
current_capacity: Arc::new(RwLock::new(ResourceAllocation {
cpu_cores: 2.0,
memory_gb: 4.0,
storage_gb: 100.0,
network_bandwidth_mbps: 1000,
})),
scaling_history: Arc::new(RwLock::new(Vec::new())),
}
}
pub async fn execute(
&self,
strategy: &ScalingStrategy,
) -> Result<ScalingResult, Box<dyn std::error::Error>> {
let mut capacity = self.current_capacity.write().await;
let mut history = self.scaling_history.write().await;
let instances_before: _ = (capacity.cpu_cores / 2.0) as u32;
let resources_before: _ = capacity.clone();
match strategy.action.action_type {
ActionType::ScaleOut | ActionType::ScaleUp => {
capacity.cpu_cores += strategy.action.resources.cpu_cores;
capacity.memory_gb += strategy.action.resources.memory_gb;
capacity.storage_gb += strategy.action.resources.storage_gb;
capacity.network_bandwidth_mbps += strategy.action.resources.network_bandwidth_mbps;
}
ActionType::ScaleIn | ActionType::ScaleDown => {
capacity.cpu_cores =
(capacity.cpu_cores - strategy.action.resources.cpu_cores).max(1.0);
capacity.memory_gb =
(capacity.memory_gb - strategy.action.resources.memory_gb).max(1.0);
capacity.storage_gb =
(capacity.storage_gb - strategy.action.resources.storage_gb).max(10.0);
capacity.network_bandwidth_mbps = (capacity.network_bandwidth_mbps
- strategy.action.resources.network_bandwidth_mbps)
.max(100);
}
}
let instances_after: _ = (capacity.cpu_cores / 2.0) as u32;
history.push(strategy.action.clone());
Ok(ScalingResult {
success: true,
instances_before,
instances_after,
resources_before,
resources_after: capacity.clone(),
performance_impact: 15.0,
cost_impact: 10.0,
})
}
}
impl PredictionModel {
pub fn new(model_type: ModelType) -> Self {
Self {
model_type,
parameters: HashMap::new(),
}
}
}
impl AnomalyDetector {
pub fn new(threshold: f64, window_size: usize) -> Self {
Self {
threshold,
window_size,
}
}
pub fn detect(&self, data: &[Metrics]) -> Vec<Anomaly> {
let mut anomalies = Vec::new();
if data.len() < self.window_size {
return anomalies;
}
for i in self.window_size..data.len() {
let window: _ = &data[i - self.window_size..i];
let current: _ = &data[i];
let avg_cpu: f64 =
window.iter().map(|m| m.cpu_usage).sum::<f64>() / window.len() as f64;
let std_cpu: _ = (window
.iter()
.map(|m| {
let diff: _ = m.cpu_usage - avg_cpu;
diff * diff
})
.sum::<f64>()
/ window.len() as f64)
.sqrt();
if (current.cpu_usage - avg_cpu).abs() > self.threshold * std_cpu {
anomalies.push(Anomaly {
timestamp: current.timestamp,
metric_type: "cpu_usage".to_string(),
deviation_score: (current.cpu_usage - avg_cpu) / std_cpu,
description: format!("CPU 使用率异常偏离平均值 {:.2}", current.cpu_usage),
});
}
}
anomalies
}
}
#[cfg(test)]
mod tests {
#[tokio::test]
async fn test_predictive_scaler_creation() {
let scaler: _ = PredictiveScaler::new();
}
#[tokio::test]
async fn test_resource_prediction() {
let scaler: _ = PredictiveScaler::new();
let timeframe: _ = TimeFrame {
start_time: Utc::now(),
end_time: Utc::now() + Duration::hours(1),
duration: Duration::hours(1),
};
{
let predictor: _ = scaler.predictor.read().await;
let mut data = predictor.historical_data.write().await;
data.push(Metrics {
timestamp: Utc::now(),
cpu_usage: 50.0,
memory_usage: 60.0,
network_io: 1000,
disk_io: 500,
active_connections: 100,
request_rate: 10.0,
});
}
let prediction: _ = scaler.predict_resource_usage(timeframe).await.unwrap();
assert!(prediction.predicted_cpu > 0.0);
assert!(prediction.confidence > 0.0);
assert!(!prediction.factors.is_empty());
}
#[tokio::test]
async fn test_trend_analysis() {
let scaler: _ = PredictiveScaler::new();
let mut historical_data = Vec::new();
for i in 0..10 {
historical_data.push(Metrics {
timestamp: Utc::now() + Duration::minutes(i as i64),
cpu_usage: 50.0 + i as f64 * 5.0,
memory_usage: 60.0 + i as f64 * 3.0,
network_io: 1000 + i as u64 * 100,
disk_io: 500 + i as u64 * 50,
active_connections: 100 + i as u64 * 10,
request_rate: 10.0 + i as f64,
});
}
let analysis: _ = scaler.analyze_trends(&historical_data).await.unwrap();
assert_eq!(analysis.trend_direction, TrendDirection::Increasing);
assert!(analysis.growth_rate > 0.0);
assert!(analysis.forecast_accuracy > 0.0);
}
#[tokio::test]
async fn test_schedule_optimization() {
let scaler: _ = PredictiveScaler::new();
let tasks: _ = vec![
Task {
id: "1".to_string(),
name: "任务1".to_string(),
priority: TaskPriority::High,
estimated_duration: Duration::minutes(30),
resource_requirements: ResourceRequirements {
cpu: 2.0,
memory: 4.0,
estimated_duration: Duration::minutes(30),
},
dependencies: vec![],
},
Task {
id: "2".to_string(),
name: "任务2".to_string(),
priority: TaskPriority::Medium,
estimated_duration: Duration::minutes(20),
resource_requirements: ResourceRequirements {
cpu: 1.0,
memory: 2.0,
estimated_duration: Duration::minutes(20),
},
dependencies: vec![],
},
];
let schedule: _ = scaler.optimize_schedule(&tasks).await.unwrap();
assert!(!schedule.tasks.is_empty());
assert_eq!(schedule.tasks.len(), tasks.len());
assert!(schedule.total_duration.num_seconds() > 0);
}
}