systemprompt_analytics/services/
anomaly_detection.rs

1//! In-memory threshold and spike detection for live metrics.
2//!
3//! [`AnomalyDetectionService`] tracks per-metric warning/critical thresholds
4//! and a rolling one-hour event window, classifying each value as
5//! [`AnomalyLevel::Normal`], `Warning`, or `Critical` and detecting trend
6//! spikes relative to the recent average. State is held in memory behind
7//! `RwLock`s, not persisted.
8//!
9//! Copyright (c) systemprompt.io — Business Source License 1.1.
10//! See <https://systemprompt.io> for licensing details.
11
12use std::collections::HashMap;
13use std::sync::Arc;
14
15use chrono::{DateTime, Duration, Utc};
16use tokio::sync::RwLock;
17
18#[derive(Debug, Clone)]
19pub struct AnomalyThresholdConfig {
20    pub warning_threshold: f64,
21    pub critical_threshold: f64,
22}
23
24impl Copy for AnomalyThresholdConfig {}
25
26#[derive(Debug, Clone)]
27pub struct AnomalyEvent {
28    pub timestamp: DateTime<Utc>,
29    pub value: f64,
30}
31
32impl Copy for AnomalyEvent {}
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35pub enum AnomalyLevel {
36    Normal,
37    Warning,
38    Critical,
39}
40
41#[derive(Debug, Clone)]
42pub struct AnomalyCheckResult {
43    pub metric_name: String,
44    pub current_value: f64,
45    pub level: AnomalyLevel,
46    pub message: Option<String>,
47}
48
49#[derive(Clone, Debug)]
50pub struct AnomalyDetectionService {
51    thresholds: Arc<RwLock<HashMap<String, AnomalyThresholdConfig>>>,
52    recent_events: Arc<RwLock<HashMap<String, Vec<AnomalyEvent>>>>,
53}
54
55impl AnomalyDetectionService {
56    pub fn new() -> Self {
57        Self {
58            thresholds: Arc::new(RwLock::new(Self::default_thresholds())),
59            recent_events: Arc::new(RwLock::new(HashMap::new())),
60        }
61    }
62
63    fn default_thresholds() -> HashMap<String, AnomalyThresholdConfig> {
64        let mut thresholds = HashMap::new();
65
66        thresholds.insert(
67            "requests_per_minute".into(),
68            AnomalyThresholdConfig {
69                warning_threshold: 15.0,
70                critical_threshold: 30.0,
71            },
72        );
73
74        thresholds.insert(
75            "session_count_per_fingerprint".into(),
76            AnomalyThresholdConfig {
77                warning_threshold: 5.0,
78                critical_threshold: 10.0,
79            },
80        );
81
82        thresholds.insert(
83            "error_rate".into(),
84            AnomalyThresholdConfig {
85                warning_threshold: 0.1,
86                critical_threshold: 0.25,
87            },
88        );
89
90        thresholds
91    }
92
93    pub async fn check_anomaly(&self, metric_name: &str, value: f64) -> AnomalyCheckResult {
94        let level = {
95            let thresholds = self.thresholds.read().await;
96            thresholds
97                .get(metric_name)
98                .map_or(AnomalyLevel::Normal, |t| {
99                    if value >= t.critical_threshold {
100                        AnomalyLevel::Critical
101                    } else if value >= t.warning_threshold {
102                        AnomalyLevel::Warning
103                    } else {
104                        AnomalyLevel::Normal
105                    }
106                })
107        };
108
109        let message = match level {
110            AnomalyLevel::Critical => Some(format!(
111                "CRITICAL: {metric_name} = {value:.2} exceeds critical threshold"
112            )),
113            AnomalyLevel::Warning => Some(format!(
114                "WARNING: {metric_name} = {value:.2} exceeds warning threshold"
115            )),
116            AnomalyLevel::Normal => None,
117        };
118
119        AnomalyCheckResult {
120            metric_name: metric_name.to_owned(),
121            current_value: value,
122            level,
123            message,
124        }
125    }
126
127    pub async fn record_event(&self, metric_name: &str, value: f64) {
128        let now = Utc::now();
129        let cutoff = now - Duration::hours(1);
130        let key = metric_name.to_owned();
131        let event = AnomalyEvent {
132            timestamp: now,
133            value,
134        };
135
136        let mut events = self.recent_events.write().await;
137        events.entry(key).or_default().push(event);
138        if let Some(metric_events) = events.get_mut(metric_name) {
139            metric_events.retain(|e| e.timestamp > cutoff);
140        }
141    }
142
143    pub async fn check_trend_anomaly(
144        &self,
145        metric_name: &str,
146        window_minutes: i64,
147    ) -> Option<AnomalyCheckResult> {
148        let metric_events = {
149            let events = self.recent_events.read().await;
150            events.get(metric_name).cloned()?
151        };
152
153        if metric_events.len() < 2 {
154            return None;
155        }
156
157        let cutoff = Utc::now() - Duration::minutes(window_minutes);
158        let recent: Vec<_> = metric_events
159            .iter()
160            .filter(|e| e.timestamp > cutoff)
161            .collect();
162
163        if recent.is_empty() {
164            return None;
165        }
166
167        let avg: f64 = recent.iter().map(|e| e.value).sum::<f64>() / recent.len() as f64;
168        let latest = recent.last()?.value;
169        let spike_ratio = if avg > 0.0 { latest / avg } else { 1.0 };
170
171        if spike_ratio > 3.0 {
172            Some(AnomalyCheckResult {
173                metric_name: metric_name.to_owned(),
174                current_value: latest,
175                level: AnomalyLevel::Critical,
176                message: Some(format!(
177                    "Spike: {metric_name} jumped {spike_ratio:.1}x above average"
178                )),
179            })
180        } else if spike_ratio > 2.0 {
181            Some(AnomalyCheckResult {
182                metric_name: metric_name.to_owned(),
183                current_value: latest,
184                level: AnomalyLevel::Warning,
185                message: Some(format!(
186                    "Elevated: {metric_name} is {spike_ratio:.1}x above average"
187                )),
188            })
189        } else {
190            None
191        }
192    }
193
194    pub async fn update_threshold(&self, metric_name: &str, warning: f64, critical: f64) {
195        let mut thresholds = self.thresholds.write().await;
196        thresholds.insert(
197            metric_name.to_owned(),
198            AnomalyThresholdConfig {
199                warning_threshold: warning,
200                critical_threshold: critical,
201            },
202        );
203    }
204
205    pub async fn get_recent_events(&self, metric_name: &str) -> Vec<AnomalyEvent> {
206        let events = self.recent_events.read().await;
207        events.get(metric_name).cloned().unwrap_or_else(Vec::new)
208    }
209}
210
211impl Default for AnomalyDetectionService {
212    fn default() -> Self {
213        Self::new()
214    }
215}
systemprompt_analytics/services/anomaly_detection.rs

systemprompt_analytics/services/
anomaly_detection.rs
