use std::collections::VecDeque;
use super::{
CorrelationResult, EventSample, EventType, InterferenceResult, IsolationAction,
IsolationRecommendation, PerformanceSample, SystemSnapshot,
};
#[derive(Debug)]
pub struct CorrelationAnalyzer {
perf_samples: VecDeque<PerformanceSample>,
event_samples: VecDeque<EventSample>,
snapshots: VecDeque<SystemSnapshot>,
max_samples: usize,
spike_threshold: f64,
window_sec: f64,
}
impl Default for CorrelationAnalyzer {
fn default() -> Self {
Self {
perf_samples: VecDeque::new(),
event_samples: VecDeque::new(),
snapshots: VecDeque::new(),
max_samples: 1000,
spike_threshold: 15.0,
window_sec: 60.0,
}
}
}
impl CorrelationAnalyzer {
pub fn new() -> Self {
Self::default()
}
pub fn with_max_samples(mut self, max: usize) -> Self {
self.max_samples = max;
self
}
pub fn with_spike_threshold(mut self, percent: f64) -> Self {
self.spike_threshold = percent;
self
}
pub fn with_window(mut self, seconds: f64) -> Self {
self.window_sec = seconds;
self
}
pub fn add_perf_sample(&mut self, sample: PerformanceSample) {
if self.perf_samples.len() >= self.max_samples {
self.perf_samples.pop_front();
}
self.perf_samples.push_back(sample);
}
pub fn add_event_sample(&mut self, sample: EventSample) {
if self.event_samples.len() >= self.max_samples {
self.event_samples.pop_front();
}
self.event_samples.push_back(sample);
}
pub fn add_snapshot(&mut self, snapshot: SystemSnapshot) {
if self.snapshots.len() >= self.max_samples / 10 {
self.snapshots.pop_front();
}
self.snapshots.push_back(snapshot);
}
pub fn perf_sample_count(&self) -> usize {
self.perf_samples.len()
}
pub fn event_sample_count(&self) -> usize {
self.event_samples.len()
}
pub fn correlate_events(&self, event_type: EventType) -> Option<CorrelationResult> {
let events: Vec<_> = self
.event_samples
.iter()
.filter(|e| e.event_type == event_type)
.collect();
if events.len() < 5 || self.perf_samples.len() < 5 {
return None;
}
let pearson_r = self.compute_correlation(&events);
let is_significant = pearson_r.abs() > 0.3 && events.len() >= 10;
Some(CorrelationResult {
event_type,
pearson_r,
sample_count: events.len().min(self.perf_samples.len()),
is_significant,
optimal_lag: 0.0,
})
}
fn compute_correlation(&self, events: &[&EventSample]) -> f64 {
let mut paired: Vec<(f64, f64)> = Vec::new();
for event in events {
if let Some(perf) = self.find_closest_perf(event.timestamp) {
paired.push((event.value, perf.cv_percent));
}
}
if paired.len() < 3 {
return 0.0;
}
let n = paired.len() as f64;
let sum_x: f64 = paired.iter().map(|(x, _)| x).sum();
let sum_y: f64 = paired.iter().map(|(_, y)| y).sum();
let sum_xy: f64 = paired.iter().map(|(x, y)| x * y).sum();
let sum_xx: f64 = paired.iter().map(|(x, _)| x * x).sum();
let sum_yy: f64 = paired.iter().map(|(_, y)| y * y).sum();
let numerator = n * sum_xy - sum_x * sum_y;
let denominator = ((n * sum_xx - sum_x * sum_x) * (n * sum_yy - sum_y * sum_y)).sqrt();
if denominator.abs() < 1e-10 {
0.0
} else {
numerator / denominator
}
}
fn find_closest_perf(&self, timestamp: f64) -> Option<&PerformanceSample> {
self.perf_samples
.iter()
.min_by(|a, b| {
(a.timestamp - timestamp)
.abs()
.partial_cmp(&(b.timestamp - timestamp).abs())
.expect("values should be comparable")
})
.filter(|p| (p.timestamp - timestamp).abs() < self.window_sec)
}
pub fn detect_interference(&self) -> Option<InterferenceResult> {
let event_types = [
EventType::Interrupt,
EventType::DiskIo,
EventType::Network,
EventType::ContextSwitch,
EventType::PageFault,
EventType::ProcessCpu,
];
let mut correlations: Vec<(EventType, f64)> = Vec::new();
for event_type in event_types {
if let Some(result) = self.correlate_events(event_type) {
if result.pearson_r.abs() > 0.1 {
correlations.push((event_type, result.pearson_r));
}
}
}
if correlations.is_empty() {
return None;
}
correlations.sort_by(|a, b| {
b.1.abs()
.partial_cmp(&a.1.abs())
.expect("values should be comparable")
});
let (primary_source, correlation) = correlations[0];
let secondary_sources: Vec<_> = correlations.into_iter().skip(1).collect();
let confidence = if correlation.abs() > 0.5 && self.perf_samples.len() > 20 {
0.9
} else if correlation.abs() > 0.3 {
0.7
} else {
0.5
};
Some(InterferenceResult {
primary_source,
correlation,
confidence,
secondary_sources,
})
}
pub fn recommend_isolation(&self) -> IsolationRecommendation {
if let Some(interference) = self.detect_interference() {
let (action, expected_improvement) = match interference.primary_source {
EventType::Interrupt => (IsolationAction::CpuPin, 20.0),
EventType::ContextSwitch => (IsolationAction::RealtimePriority, 15.0),
EventType::ProcessCpu => (IsolationAction::CpuPin, 25.0),
EventType::DiskIo => (IsolationAction::MemoryIsolation, 10.0),
EventType::Network => (IsolationAction::NetworkIsolation, 15.0),
EventType::PageFault => (IsolationAction::MemoryIsolation, 20.0),
};
let reason = format!(
"{} correlation ({:.2}) with {} detected",
interference.category().name(),
interference.correlation,
interference.primary_source.name()
);
IsolationRecommendation {
action,
reason,
expected_improvement: expected_improvement * interference.correlation.abs(),
confidence: interference.confidence,
}
} else {
IsolationRecommendation {
action: IsolationAction::None,
reason: "No significant interference detected".to_string(),
expected_improvement: 0.0,
confidence: 0.9,
}
}
}
pub fn capture_system_state(&mut self, timestamp: f64) -> SystemSnapshot {
let snapshot = SystemSnapshot::new(timestamp);
self.add_snapshot(snapshot.clone());
snapshot
}
pub fn get_spikes(&self) -> Vec<&PerformanceSample> {
self.perf_samples
.iter()
.filter(|s| s.is_spike(self.spike_threshold))
.collect()
}
pub fn clear(&mut self) {
self.perf_samples.clear();
self.event_samples.clear();
self.snapshots.clear();
}
}