use async_trait::async_trait;
use chrono::{DateTime, Datelike, Timelike, Utc};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, info, trace, warn};
use uuid::Uuid;
use super::stats::{
calculate_linear_trend, mean, percentage_change, std_dev, CircularBuffer,
SimpleMovingAverage,
};
use super::traits::{
Analyzer, AnalyzerConfig, AnalyzerError, AnalyzerResult, AnalyzerState,
};
use super::types::{
Action, ActionType, Alert, AlertStatus, AnalysisReport, AnalyzerEvent, AnalyzerStats,
Confidence, Evidence, EvidenceType, Impact, ImpactMetric, Insight, InsightCategory,
Priority, Recommendation, ReportSummary, RiskLevel, Severity, Threshold, ThresholdOperator,
};
#[derive(Debug, Clone)]
pub struct PatternAnalyzerConfig {
pub base: AnalyzerConfig,
pub burst_threshold_multiplier: f64,
pub growth_rate_threshold_pct: f64,
pub min_requests_for_patterns: usize,
pub moving_average_window: usize,
pub enable_forecasting: bool,
}
impl Default for PatternAnalyzerConfig {
fn default() -> Self {
Self {
base: AnalyzerConfig::default(),
burst_threshold_multiplier: 2.0,
growth_rate_threshold_pct: 20.0,
min_requests_for_patterns: 50,
moving_average_window: 10,
enable_forecasting: true,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PatternType {
PeakHours,
LowTraffic,
TrafficBurst,
SteadyGrowth,
SeasonalPattern,
}
impl PatternType {
fn as_str(&self) -> &str {
match self {
PatternType::PeakHours => "peak_hours",
PatternType::LowTraffic => "low_traffic",
PatternType::TrafficBurst => "traffic_burst",
PatternType::SteadyGrowth => "steady_growth",
PatternType::SeasonalPattern => "seasonal_pattern",
}
}
}
#[derive(Debug, Clone)]
struct TemporalMetrics {
hourly_counts: [u64; 24],
daily_counts: [u64; 7],
recent_requests: CircularBuffer<DateTime<Utc>>,
hourly_series: Vec<(f64, f64)>,
moving_average: SimpleMovingAverage,
total_requests: u64,
user_requests: HashMap<String, u64>,
endpoint_requests: HashMap<String, u64>,
model_requests: HashMap<String, u64>,
}
impl TemporalMetrics {
fn new(window_size: usize, history_size: usize) -> Self {
Self {
hourly_counts: [0; 24],
daily_counts: [0; 7],
recent_requests: CircularBuffer::new(history_size),
hourly_series: Vec::new(),
moving_average: SimpleMovingAverage::new(window_size),
total_requests: 0,
user_requests: HashMap::new(),
endpoint_requests: HashMap::new(),
model_requests: HashMap::new(),
}
}
fn record_request(&mut self, timestamp: DateTime<Utc>, metadata: &HashMap<String, String>) {
let hour = timestamp.hour() as usize;
let day = timestamp.weekday().num_days_from_monday() as usize;
self.hourly_counts[hour] += 1;
self.daily_counts[day] += 1;
self.recent_requests.push(timestamp);
self.total_requests += 1;
if let Some(user_id) = metadata.get("user_id") {
*self.user_requests.entry(user_id.clone()).or_insert(0) += 1;
}
if let Some(endpoint) = metadata.get("endpoint") {
*self.endpoint_requests.entry(endpoint.clone()).or_insert(0) += 1;
}
self.update_moving_average();
}
fn record_model(&mut self, model: &str) {
*self.model_requests.entry(model.to_string()).or_insert(0) += 1;
}
fn update_moving_average(&mut self) {
let now = Utc::now();
let one_minute_ago = now - chrono::Duration::minutes(1);
let recent_count = self
.recent_requests
.as_slice()
.iter()
.filter(|&&ts| ts > one_minute_ago)
.count();
self.moving_average.update(recent_count as f64);
}
fn get_peak_hours(&self, threshold_multiplier: f64) -> Vec<usize> {
if self.total_requests < 24 {
return Vec::new();
}
let avg = self.hourly_counts.iter().sum::<u64>() as f64 / 24.0;
let threshold = avg * threshold_multiplier;
self.hourly_counts
.iter()
.enumerate()
.filter(|(_, &count)| count as f64 > threshold)
.map(|(hour, _)| hour)
.collect()
}
fn get_low_traffic_hours(&self, threshold_multiplier: f64) -> Vec<usize> {
if self.total_requests < 24 {
return Vec::new();
}
let avg = self.hourly_counts.iter().sum::<u64>() as f64 / 24.0;
let threshold = avg / threshold_multiplier;
self.hourly_counts
.iter()
.enumerate()
.filter(|(_, &count)| count > 0 && (count as f64) < threshold)
.map(|(hour, _)| hour)
.collect()
}
fn detect_burst(&self, threshold_multiplier: f64) -> Option<f64> {
if let Some(avg) = self.moving_average.value() {
if avg == 0.0 {
return None;
}
let now = Utc::now();
let one_minute_ago = now - chrono::Duration::minutes(1);
let recent_count = self
.recent_requests
.as_slice()
.iter()
.filter(|&&ts| ts > one_minute_ago)
.count() as f64;
if recent_count > avg * threshold_multiplier {
return Some(recent_count / avg);
}
}
None
}
fn calculate_growth_rate(&self) -> Option<f64> {
if self.hourly_series.len() < 2 {
return None;
}
if let Some(trend) = calculate_linear_trend(&self.hourly_series) {
if self.hourly_series.first().map(|(_, v)| *v).unwrap_or(0.0) > 0.0 {
let baseline = self.hourly_series.first().unwrap().1;
let growth_pct = (trend.slope / baseline) * 100.0;
return Some(growth_pct);
}
}
None
}
fn detect_weekday_pattern(&self) -> Option<f64> {
if self.total_requests < 100 {
return None;
}
let weekday_total: u64 = self.daily_counts[0..5].iter().sum();
let weekend_total: u64 = self.daily_counts[5..7].iter().sum();
if weekend_total == 0 {
return None;
}
let weekday_avg = weekday_total as f64 / 5.0;
let weekend_avg = weekend_total as f64 / 2.0;
if weekend_avg > 0.0 {
let difference_pct = ((weekday_avg - weekend_avg) / weekend_avg) * 100.0;
Some(difference_pct)
} else {
None
}
}
fn get_top_users(&self, limit: usize) -> Vec<(String, u64)> {
let mut users: Vec<_> = self.user_requests.iter()
.map(|(k, v)| (k.clone(), *v))
.collect();
users.sort_by(|a, b| b.1.cmp(&a.1));
users.truncate(limit);
users
}
fn get_top_endpoints(&self, limit: usize) -> Vec<(String, u64)> {
let mut endpoints: Vec<_> = self.endpoint_requests.iter()
.map(|(k, v)| (k.clone(), *v))
.collect();
endpoints.sort_by(|a, b| b.1.cmp(&a.1));
endpoints.truncate(limit);
endpoints
}
}
pub struct PatternAnalyzer {
config: PatternAnalyzerConfig,
state: AnalyzerState,
metrics: Arc<RwLock<TemporalMetrics>>,
stats: Arc<RwLock<AnalyzerStats>>,
start_time: Instant,
last_report_time: Arc<RwLock<Instant>>,
}
impl PatternAnalyzer {
pub fn new(config: PatternAnalyzerConfig) -> Self {
let window_size = config.moving_average_window;
let history_size = 1000;
Self {
config,
state: AnalyzerState::Initialized,
metrics: Arc::new(RwLock::new(TemporalMetrics::new(window_size, history_size))),
stats: Arc::new(RwLock::new(AnalyzerStats {
analyzer: "pattern".to_string(),
..Default::default()
})),
start_time: Instant::now(),
last_report_time: Arc::new(RwLock::new(Instant::now())),
}
}
pub fn with_defaults() -> Self {
Self::new(PatternAnalyzerConfig::default())
}
async fn process_request(&mut self, event: AnalyzerEvent) -> AnalyzerResult<()> {
if let AnalyzerEvent::Request {
timestamp,
model,
metadata,
..
} = event
{
let mut metrics = self.metrics.write().await;
metrics.record_request(timestamp, &metadata);
metrics.record_model(&model);
trace!(
"Recorded pattern metrics: timestamp={}, model={}",
timestamp,
model
);
}
Ok(())
}
async fn process_response(&mut self, event: AnalyzerEvent) -> AnalyzerResult<()> {
if let AnalyzerEvent::Response {
timestamp,
model,
metadata,
..
} = event
{
let mut metrics = self.metrics.write().await;
metrics.record_request(timestamp, &metadata);
metrics.record_model(&model);
trace!(
"Recorded pattern metrics from response: timestamp={}, model={}",
timestamp,
model
);
}
Ok(())
}
async fn process_metric(&mut self, event: AnalyzerEvent) -> AnalyzerResult<()> {
if let AnalyzerEvent::Metric {
timestamp,
labels,
..
} = event
{
let mut metrics = self.metrics.write().await;
metrics.record_request(timestamp, &labels);
}
Ok(())
}
async fn analyze_patterns(&self) -> Vec<Insight> {
let metrics = self.metrics.read().await;
let mut insights = Vec::new();
if metrics.total_requests < self.config.min_requests_for_patterns as u64 {
debug!(
"Insufficient data for pattern analysis: {} requests (minimum: {})",
metrics.total_requests, self.config.min_requests_for_patterns
);
return insights;
}
let peak_hours = metrics.get_peak_hours(1.5);
if !peak_hours.is_empty() {
insights.push(self.create_peak_hours_insight(&peak_hours, &metrics));
}
let low_traffic_hours = metrics.get_low_traffic_hours(2.0);
if !low_traffic_hours.is_empty() {
insights.push(self.create_low_traffic_insight(&low_traffic_hours, &metrics));
}
if let Some(burst_ratio) = metrics.detect_burst(self.config.burst_threshold_multiplier) {
insights.push(self.create_burst_insight(burst_ratio));
}
if let Some(growth_rate) = metrics.calculate_growth_rate() {
if growth_rate.abs() > self.config.growth_rate_threshold_pct {
insights.push(self.create_growth_insight(growth_rate));
}
}
if let Some(weekday_diff) = metrics.detect_weekday_pattern() {
if weekday_diff.abs() > 30.0 {
insights.push(self.create_seasonal_insight(weekday_diff));
}
}
insights
}
fn create_peak_hours_insight(&self, peak_hours: &[usize], metrics: &TemporalMetrics) -> Insight {
let hours_str = if peak_hours.len() > 3 {
format!(
"{}:00-{}:00",
peak_hours.first().unwrap(),
peak_hours.last().unwrap()
)
} else {
peak_hours
.iter()
.map(|h| format!("{}:00", h))
.collect::<Vec<_>>()
.join(", ")
};
let peak_count: u64 = peak_hours.iter().map(|&h| metrics.hourly_counts[h]).sum();
let avg_count = metrics.hourly_counts.iter().sum::<u64>() as f64 / 24.0;
let peak_avg = peak_count as f64 / peak_hours.len() as f64;
Insight {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
severity: Severity::Info,
confidence: Confidence::new(0.90),
title: format!("Peak traffic hours: {}", hours_str),
description: format!(
"Traffic peaks during {} with {:.0}% higher volume than average. Peak hours average {:.0} requests/hour vs overall average of {:.0} requests/hour.",
hours_str,
((peak_avg - avg_count) / avg_count) * 100.0,
peak_avg,
avg_count
),
category: InsightCategory::Pattern,
evidence: vec![Evidence {
evidence_type: EvidenceType::Pattern,
description: format!("Hourly distribution analysis"),
data: serde_json::json!({
"peak_hours": peak_hours,
"peak_avg_requests": peak_avg,
"overall_avg_requests": avg_count,
"increase_pct": ((peak_avg - avg_count) / avg_count) * 100.0
}),
timestamp: Utc::now(),
}],
metrics: {
let mut m = HashMap::new();
m.insert("peak_avg_requests".to_string(), peak_avg);
m.insert("overall_avg_requests".to_string(), avg_count);
m
},
tags: vec!["temporal".to_string(), "peak_hours".to_string()],
}
}
fn create_low_traffic_insight(&self, low_hours: &[usize], metrics: &TemporalMetrics) -> Insight {
let hours_str = if low_hours.len() > 3 {
format!(
"{}:00-{}:00",
low_hours.first().unwrap(),
low_hours.last().unwrap()
)
} else {
low_hours
.iter()
.map(|h| format!("{}:00", h))
.collect::<Vec<_>>()
.join(", ")
};
let low_count: u64 = low_hours.iter().map(|&h| metrics.hourly_counts[h]).sum();
let low_avg = low_count as f64 / low_hours.len() as f64;
Insight {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
severity: Severity::Info,
confidence: Confidence::new(0.85),
title: format!("Low traffic period: {}", hours_str),
description: format!(
"Traffic is significantly lower during {} with an average of {:.0} requests/hour. Consider cost optimization during these periods.",
hours_str, low_avg
),
category: InsightCategory::Pattern,
evidence: vec![Evidence {
evidence_type: EvidenceType::Pattern,
description: "Low traffic period detected".to_string(),
data: serde_json::json!({
"low_traffic_hours": low_hours,
"avg_requests": low_avg
}),
timestamp: Utc::now(),
}],
metrics: {
let mut m = HashMap::new();
m.insert("low_traffic_avg_requests".to_string(), low_avg);
m
},
tags: vec!["temporal".to_string(), "low_traffic".to_string()],
}
}
fn create_burst_insight(&self, burst_ratio: f64) -> Insight {
let severity = if burst_ratio > 5.0 {
Severity::Warning
} else {
Severity::Info
};
Insight {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
severity,
confidence: Confidence::new(0.92),
title: format!("Traffic burst detected: {:.1}x normal", burst_ratio),
description: format!(
"Current traffic is {:.1}x higher than the moving average. This sudden spike may indicate increased user activity, viral content, or potential issues.",
burst_ratio
),
category: InsightCategory::Anomaly,
evidence: vec![Evidence {
evidence_type: EvidenceType::Anomaly,
description: "Burst detection via moving average".to_string(),
data: serde_json::json!({
"burst_ratio": burst_ratio,
"threshold": self.config.burst_threshold_multiplier
}),
timestamp: Utc::now(),
}],
metrics: {
let mut m = HashMap::new();
m.insert("burst_ratio".to_string(), burst_ratio);
m
},
tags: vec!["burst".to_string(), "traffic".to_string(), "anomaly".to_string()],
}
}
fn create_growth_insight(&self, growth_rate: f64) -> Insight {
let is_positive = growth_rate > 0.0;
let severity = if growth_rate.abs() > 50.0 {
Severity::Warning
} else {
Severity::Info
};
Insight {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
severity,
confidence: Confidence::new(0.82),
title: format!(
"{} trend: {:.1}% growth",
if is_positive { "Growth" } else { "Decline" },
growth_rate.abs()
),
description: format!(
"Traffic shows a {} trend with {:.1}% {} over the analyzed period. {} capacity planning to accommodate this trend.",
if is_positive { "positive growth" } else { "declining" },
growth_rate.abs(),
if is_positive { "increase" } else { "decrease" },
if is_positive { "Consider" } else { "Review" }
),
category: InsightCategory::Pattern,
evidence: vec![Evidence {
evidence_type: EvidenceType::TimeSeries,
description: "Linear trend analysis".to_string(),
data: serde_json::json!({
"growth_rate_pct": growth_rate,
"trend_type": if is_positive { "growth" } else { "decline" }
}),
timestamp: Utc::now(),
}],
metrics: {
let mut m = HashMap::new();
m.insert("growth_rate_pct".to_string(), growth_rate);
m
},
tags: vec![
"growth".to_string(),
"trend".to_string(),
if is_positive { "increasing".to_string() } else { "decreasing".to_string() }
],
}
}
fn create_seasonal_insight(&self, weekday_diff: f64) -> Insight {
let is_higher_weekday = weekday_diff > 0.0;
Insight {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
severity: Severity::Info,
confidence: Confidence::new(0.88),
title: format!(
"Weekly pattern: {:.0}% {} traffic on {}",
weekday_diff.abs(),
if is_higher_weekday { "higher" } else { "lower" },
if is_higher_weekday { "weekdays" } else { "weekends" }
),
description: format!(
"Traffic shows a clear weekly pattern with {:.0}% {} volume on {} compared to {}. This seasonal pattern can be used for capacity planning and cost optimization.",
weekday_diff.abs(),
if is_higher_weekday { "higher" } else { "lower" },
if is_higher_weekday { "weekdays (Mon-Fri)" } else { "weekends (Sat-Sun)" },
if is_higher_weekday { "weekends" } else { "weekdays" }
),
category: InsightCategory::Pattern,
evidence: vec![Evidence {
evidence_type: EvidenceType::Pattern,
description: "Weekday vs weekend comparison".to_string(),
data: serde_json::json!({
"weekday_weekend_diff_pct": weekday_diff,
"pattern_type": "weekly"
}),
timestamp: Utc::now(),
}],
metrics: {
let mut m = HashMap::new();
m.insert("weekday_weekend_diff_pct".to_string(), weekday_diff.abs());
m
},
tags: vec!["seasonal".to_string(), "weekly".to_string(), "pattern".to_string()],
}
}
fn generate_recommendations(&self, insights: &[Insight], metrics: &TemporalMetrics) -> Vec<Recommendation> {
let mut recommendations = Vec::new();
for insight in insights {
if insight.tags.contains(&"peak_hours".to_string()) {
recommendations.push(Recommendation {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
priority: Priority::Medium,
title: "Scale resources for peak hours".to_string(),
description: "Configure auto-scaling to handle peak traffic periods efficiently.".to_string(),
action: Action {
action_type: ActionType::Scale,
instructions: "1. Set up auto-scaling policies for peak hours\n2. Pre-warm resources before peak periods\n3. Configure rate limiting to protect during spikes".to_string(),
parameters: HashMap::new(),
estimated_effort: Some(3.0),
risk_level: RiskLevel::Low,
},
expected_impact: Impact {
performance: Some(ImpactMetric {
name: "latency_during_peaks".to_string(),
current: 0.0,
expected: 0.0,
unit: "ms".to_string(),
improvement_pct: 25.0,
}),
cost: Some(ImpactMetric {
name: "infrastructure_cost".to_string(),
current: 100.0,
expected: 110.0,
unit: "usd".to_string(),
improvement_pct: -10.0,
}),
quality: None,
description: "Better performance during peaks with slightly higher cost".to_string(),
},
confidence: Confidence::new(0.85),
related_insights: vec![insight.id.clone()],
tags: vec!["scaling".to_string(), "capacity".to_string()],
});
}
if insight.tags.contains(&"low_traffic".to_string()) {
recommendations.push(Recommendation {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
priority: Priority::Low,
title: "Optimize costs during low-traffic periods".to_string(),
description: "Scale down resources during consistently low-traffic hours to reduce costs.".to_string(),
action: Action {
action_type: ActionType::ConfigChange,
instructions: "1. Configure auto-scaling to reduce capacity during low-traffic hours\n2. Consider spot instances for non-critical workloads\n3. Schedule batch jobs during these periods".to_string(),
parameters: HashMap::new(),
estimated_effort: Some(2.0),
risk_level: RiskLevel::Low,
},
expected_impact: Impact {
performance: None,
cost: Some(ImpactMetric {
name: "infrastructure_cost".to_string(),
current: 100.0,
expected: 85.0,
unit: "usd".to_string(),
improvement_pct: 15.0,
}),
quality: None,
description: "15% cost savings with no performance impact".to_string(),
},
confidence: Confidence::new(0.80),
related_insights: vec![insight.id.clone()],
tags: vec!["cost_optimization".to_string(), "scaling".to_string()],
});
}
if insight.tags.contains(&"burst".to_string()) {
recommendations.push(Recommendation {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
priority: Priority::High,
title: "Implement burst handling strategies".to_string(),
description: "Deploy caching and rate limiting to handle traffic bursts efficiently.".to_string(),
action: Action {
action_type: ActionType::ConfigChange,
instructions: "1. Enable CDN caching for static responses\n2. Implement request coalescing for identical prompts\n3. Set up rate limiting per user/API key\n4. Configure burst capacity in load balancers".to_string(),
parameters: HashMap::new(),
estimated_effort: Some(4.0),
risk_level: RiskLevel::Medium,
},
expected_impact: Impact {
performance: Some(ImpactMetric {
name: "burst_handling".to_string(),
current: 0.0,
expected: 0.0,
unit: "requests".to_string(),
improvement_pct: 50.0,
}),
cost: Some(ImpactMetric {
name: "api_cost".to_string(),
current: 100.0,
expected: 80.0,
unit: "usd".to_string(),
improvement_pct: 20.0,
}),
quality: None,
description: "Better burst handling with 20% cost reduction via caching".to_string(),
},
confidence: Confidence::new(0.82),
related_insights: vec![insight.id.clone()],
tags: vec!["caching".to_string(), "rate_limiting".to_string()],
});
}
if insight.tags.contains(&"growth".to_string()) && insight.tags.contains(&"increasing".to_string()) {
recommendations.push(Recommendation {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
priority: Priority::High,
title: "Plan for continued growth".to_string(),
description: "Proactively scale infrastructure to accommodate growth trend.".to_string(),
action: Action {
action_type: ActionType::Review,
instructions: "1. Review capacity projections for next 30-90 days\n2. Negotiate volume discounts with LLM providers\n3. Consider reserved capacity for predictable baseline\n4. Implement aggressive caching strategies".to_string(),
parameters: HashMap::new(),
estimated_effort: Some(6.0),
risk_level: RiskLevel::Medium,
},
expected_impact: Impact {
performance: None,
cost: Some(ImpactMetric {
name: "future_cost".to_string(),
current: 100.0,
expected: 120.0,
unit: "usd".to_string(),
improvement_pct: -20.0,
}),
quality: None,
description: "Ensure capacity for growth with optimized pricing".to_string(),
},
confidence: Confidence::new(0.75),
related_insights: vec![insight.id.clone()],
tags: vec!["capacity_planning".to_string(), "growth".to_string()],
});
}
}
let top_endpoints = metrics.get_top_endpoints(5);
if !top_endpoints.is_empty() {
let top_endpoint_pct = (top_endpoints.first().unwrap().1 as f64 / metrics.total_requests as f64) * 100.0;
if top_endpoint_pct > 20.0 {
recommendations.push(Recommendation {
id: Uuid::new_v4().to_string(),
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
priority: Priority::Medium,
title: "Cache responses for common endpoints".to_string(),
description: format!(
"Top endpoint accounts for {:.1}% of traffic. Implement semantic caching.",
top_endpoint_pct
),
action: Action {
action_type: ActionType::Optimize,
instructions: "1. Implement semantic caching for similar prompts\n2. Cache responses with TTL based on content type\n3. Monitor cache hit rates and adjust strategies".to_string(),
parameters: HashMap::new(),
estimated_effort: Some(5.0),
risk_level: RiskLevel::Low,
},
expected_impact: Impact {
performance: Some(ImpactMetric {
name: "latency".to_string(),
current: 1000.0,
expected: 100.0,
unit: "ms".to_string(),
improvement_pct: 90.0,
}),
cost: Some(ImpactMetric {
name: "api_cost".to_string(),
current: 100.0,
expected: 70.0,
unit: "usd".to_string(),
improvement_pct: 30.0,
}),
quality: None,
description: "90% latency reduction and 30% cost savings via caching".to_string(),
},
confidence: Confidence::new(0.88),
related_insights: vec![],
tags: vec!["caching".to_string(), "optimization".to_string()],
});
}
}
recommendations
}
}
#[async_trait]
impl Analyzer for PatternAnalyzer {
fn name(&self) -> &str {
"pattern"
}
fn config(&self) -> &AnalyzerConfig {
&self.config.base
}
fn state(&self) -> AnalyzerState {
self.state
}
async fn start(&mut self) -> AnalyzerResult<()> {
if self.state != AnalyzerState::Initialized && self.state != AnalyzerState::Stopped {
return Err(AnalyzerError::InvalidState(format!(
"Cannot start from state: {:?}",
self.state
)));
}
self.state = AnalyzerState::Starting;
info!("Starting Pattern Analyzer");
self.state = AnalyzerState::Running;
info!("Pattern Analyzer started successfully");
Ok(())
}
async fn stop(&mut self) -> AnalyzerResult<()> {
if !matches!(
self.state,
AnalyzerState::Running | AnalyzerState::Draining
) {
return Err(AnalyzerError::InvalidState(format!(
"Cannot stop from state: {:?}",
self.state
)));
}
self.state = AnalyzerState::Draining;
info!("Stopping Pattern Analyzer");
self.state = AnalyzerState::Stopped;
info!("Pattern Analyzer stopped");
Ok(())
}
async fn process_event(&mut self, event: AnalyzerEvent) -> AnalyzerResult<()> {
if !self.state.can_accept_events() {
return Err(AnalyzerError::InvalidState(format!(
"Cannot process events in state: {:?}",
self.state
)));
}
let start = Instant::now();
let result = match &event {
AnalyzerEvent::Request { .. } => self.process_request(event).await,
AnalyzerEvent::Response { .. } => self.process_response(event).await,
AnalyzerEvent::Metric { .. } => self.process_metric(event).await,
_ => Ok(()), };
let mut stats = self.stats.write().await;
stats.events_processed += 1;
let elapsed = start.elapsed().as_micros() as u64;
stats.analysis_time_ms += elapsed / 1000;
stats.avg_analysis_time_us =
(stats.analysis_time_ms * 1000) as f64 / stats.events_processed as f64;
if result.is_err() {
stats.error_count += 1;
stats.last_error = result.as_ref().err().map(|e| e.to_string());
}
result
}
async fn generate_report(&self) -> AnalyzerResult<AnalysisReport> {
let metrics = self.metrics.read().await;
let stats = self.stats.read().await;
let last_report = *self.last_report_time.read().await;
let insights = self.analyze_patterns().await;
let recommendations = self.generate_recommendations(&insights, &metrics);
let alerts = Vec::new();
let mut report_metrics = HashMap::new();
report_metrics.insert("total_requests".to_string(), metrics.total_requests as f64);
if let Some(ma) = metrics.moving_average.value() {
report_metrics.insert("requests_per_minute".to_string(), ma);
}
for (hour, count) in metrics.hourly_counts.iter().enumerate() {
report_metrics.insert(format!("hour_{}", hour), *count as f64);
}
let days = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"];
for (day, count) in metrics.daily_counts.iter().enumerate() {
report_metrics.insert(format!("day_{}", days[day]), *count as f64);
}
let report = AnalysisReport {
analyzer: "pattern".to_string(),
timestamp: Utc::now(),
period_start: Utc::now() - chrono::Duration::from_std(last_report.elapsed()).unwrap(),
period_end: Utc::now(),
summary: ReportSummary {
events_processed: stats.events_processed,
events_per_second: stats.events_processed as f64
/ last_report.elapsed().as_secs_f64(),
insights_count: insights.len() as u64,
recommendations_count: recommendations.len() as u64,
alerts_count: alerts.len() as u64,
analysis_duration_ms: stats.analysis_time_ms,
highlights: vec![
format!("Total requests: {}", metrics.total_requests),
format!("Unique users: {}", metrics.user_requests.len()),
format!("Unique endpoints: {}", metrics.endpoint_requests.len()),
],
},
insights,
recommendations,
alerts,
metrics: report_metrics,
};
*self.last_report_time.write().await = Instant::now();
Ok(report)
}
fn get_stats(&self) -> AnalyzerStats {
AnalyzerStats {
analyzer: "pattern".to_string(),
uptime_seconds: self.start_time.elapsed().as_secs(),
..Default::default()
}
}
async fn reset(&mut self) -> AnalyzerResult<()> {
let mut metrics = self.metrics.write().await;
*metrics = TemporalMetrics::new(
self.config.moving_average_window,
1000,
);
let mut stats = self.stats.write().await;
*stats = AnalyzerStats {
analyzer: "pattern".to_string(),
..Default::default()
};
info!("Pattern Analyzer reset");
Ok(())
}
async fn health_check(&self) -> AnalyzerResult<()> {
let stats = self.stats.read().await;
if stats.events_processed > 0 {
let error_rate = stats.error_count as f64 / stats.events_processed as f64;
if error_rate > 0.1 {
return Err(AnalyzerError::InternalError(format!(
"High error rate: {:.2}%",
error_rate * 100.0
)));
}
}
if stats.memory_usage_bytes > self.config.base.max_memory_bytes {
return Err(AnalyzerError::ResourceExhausted(format!(
"Memory usage {} exceeds limit {}",
stats.memory_usage_bytes, self.config.base.max_memory_bytes
)));
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
fn create_test_request_event(hour: u32, metadata: HashMap<String, String>) -> AnalyzerEvent {
let timestamp = Utc::now()
.date_naive()
.and_hms_opt(hour, 0, 0)
.unwrap()
.and_utc();
AnalyzerEvent::Request {
timestamp,
request_id: Uuid::new_v4().to_string(),
model: "gpt-4".to_string(),
prompt_tokens: 100,
max_tokens: 1000,
temperature: 0.7,
metadata,
}
}
fn create_test_response_event(hour: u32) -> AnalyzerEvent {
let timestamp = Utc::now()
.date_naive()
.and_hms_opt(hour, 0, 0)
.unwrap()
.and_utc();
AnalyzerEvent::Response {
timestamp,
request_id: Uuid::new_v4().to_string(),
model: "gpt-4".to_string(),
completion_tokens: 100,
total_tokens: 200,
latency_ms: 500,
success: true,
error: None,
metadata: HashMap::new(),
}
}
#[tokio::test]
async fn test_analyzer_lifecycle() {
let mut analyzer = PatternAnalyzer::with_defaults();
assert_eq!(analyzer.state(), AnalyzerState::Initialized);
analyzer.start().await.unwrap();
assert_eq!(analyzer.state(), AnalyzerState::Running);
analyzer.stop().await.unwrap();
assert_eq!(analyzer.state(), AnalyzerState::Stopped);
}
#[tokio::test]
async fn test_process_request_event() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
let event = create_test_request_event(14, HashMap::new());
analyzer.process_event(event).await.unwrap();
let metrics = analyzer.metrics.read().await;
assert_eq!(metrics.total_requests, 1);
assert_eq!(metrics.hourly_counts[14], 1);
}
#[tokio::test]
async fn test_peak_hours_detection() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
for _ in 0..20 {
for hour in 0..24 {
let count = if (14..=16).contains(&hour) { 10 } else { 2 };
for _ in 0..count {
let event = create_test_response_event(hour);
analyzer.process_event(event).await.unwrap();
}
}
}
let insights = analyzer.analyze_patterns().await;
assert!(!insights.is_empty());
assert!(insights.iter().any(|i| i.title.contains("Peak traffic")));
}
#[tokio::test]
async fn test_low_traffic_detection() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
for _ in 0..20 {
for hour in 0..24 {
let count = if (0..=5).contains(&hour) { 1 } else { 10 };
for _ in 0..count {
let event = create_test_response_event(hour);
analyzer.process_event(event).await.unwrap();
}
}
}
let insights = analyzer.analyze_patterns().await;
assert!(insights.iter().any(|i| i.title.contains("Low traffic")));
}
#[tokio::test]
async fn test_burst_detection() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
for _ in 0..100 {
let event = create_test_response_event(12);
analyzer.process_event(event).await.unwrap();
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
}
for _ in 0..500 {
let event = create_test_response_event(12);
analyzer.process_event(event).await.unwrap();
}
let insights = analyzer.analyze_patterns().await;
assert!(insights.iter().any(|i| i.title.contains("burst")));
}
#[tokio::test]
async fn test_seasonal_pattern_detection() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
let mut metrics = analyzer.metrics.write().await;
for day in 0..5 {
metrics.daily_counts[day] = 100;
}
metrics.daily_counts[5] = 30;
metrics.daily_counts[6] = 30;
metrics.total_requests = 560;
drop(metrics);
let insights = analyzer.analyze_patterns().await;
assert!(insights.iter().any(|i| i.title.contains("Weekly pattern")));
}
#[tokio::test]
async fn test_user_tracking() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
let mut metadata = HashMap::new();
metadata.insert("user_id".to_string(), "user123".to_string());
for _ in 0..10 {
let event = create_test_request_event(12, metadata.clone());
analyzer.process_event(event).await.unwrap();
}
let metrics = analyzer.metrics.read().await;
assert_eq!(metrics.user_requests.get("user123"), Some(&10));
}
#[tokio::test]
async fn test_endpoint_tracking() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
let mut metadata = HashMap::new();
metadata.insert("endpoint".to_string(), "/api/chat".to_string());
for _ in 0..15 {
let event = create_test_request_event(12, metadata.clone());
analyzer.process_event(event).await.unwrap();
}
let metrics = analyzer.metrics.read().await;
assert_eq!(metrics.endpoint_requests.get("/api/chat"), Some(&15));
}
#[tokio::test]
async fn test_generate_report() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
for hour in 0..24 {
for _ in 0..5 {
let event = create_test_response_event(hour);
analyzer.process_event(event).await.unwrap();
}
}
let report = analyzer.generate_report().await.unwrap();
assert_eq!(report.analyzer, "pattern");
assert!(report.summary.events_processed > 0);
assert!(!report.metrics.is_empty());
}
#[tokio::test]
async fn test_reset() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
let event = create_test_response_event(12);
analyzer.process_event(event).await.unwrap();
analyzer.reset().await.unwrap();
let metrics = analyzer.metrics.read().await;
assert_eq!(metrics.total_requests, 0);
assert_eq!(metrics.hourly_counts.iter().sum::<u64>(), 0);
}
#[tokio::test]
async fn test_recommendations_generation() {
let mut analyzer = PatternAnalyzer::with_defaults();
analyzer.start().await.unwrap();
for _ in 0..20 {
for hour in 0..24 {
let count = if (14..=16).contains(&hour) { 10 } else { 2 };
for _ in 0..count {
let event = create_test_response_event(hour);
analyzer.process_event(event).await.unwrap();
}
}
}
let report = analyzer.generate_report().await.unwrap();
assert!(!report.recommendations.is_empty());
assert!(report.recommendations.iter().any(|r| r.title.contains("Scale")));
}
}