Designing an Anomaly Detection Engine for High-Cardinality Metrics Using Holt-Winters in Rust — code snippets

snippet-1.txt

// The core math engine for additive Holt-Winters triple exponential smoothing.
// Maintains rolling level, trend, and seasonal components in a memory-efficient manner.

pub struct HoltWinters {
    alpha: f64,       // Level smoothing parameter (0 < alpha < 1)
    beta: f64,        // Trend smoothing parameter (0 < beta < 1)
    gamma: f64,       // Seasonal smoothing parameter (0 < gamma < 1)
    period: usize,     // Seasonality period length (e.g. 288 steps for 24h at 5m resolution)
    level: f64,       // Current estimate of the base level
    trend: f64,       // Current estimate of the trend direction
    seasonal: Vec<f64>, // Circular vector storing seasonal components
    steps_observed: usize,
}

impl HoltWinters {
    pub fn new(alpha: f64, beta: f64, gamma: f64, period: usize, initial_value: f64) -> Self {
        Self {
            alpha,
            beta,
            gamma,
            period,
            level: initial_value,
            trend: 0.0,
            seasonal: vec![0.0; period],
            steps_observed: 0,
        }
    }

    /// Feeds a new raw metric sample and returns the forecasted value for the *next* step.
    pub fn update(&mut self, value: f64) -> f64 {
        let idx = self.steps_observed % self.period;
        let prev_level = self.level;
        let prev_trend = self.trend;
        let prev_seasonal = self.seasonal[idx];

        // Bootstrap period: fill seasonality baseline with simple exponential smoothing
        if self.steps_observed < self.period {
            self.steps_observed += 1;
            self.level = self.alpha * value + (1.0 - self.alpha) * prev_level;
            // Trend is set as the difference of levels in the early phase
            self.trend = self.beta * (self.level - prev_level) + (1.0 - self.beta) * prev_trend;
            // Seasonal component is initially assumed flat
            self.seasonal[idx] = self.gamma * (value - self.level);
            return self.level + self.trend;
        }

        // Standard additive Holt-Winters update steps
        self.level = self.alpha * (value - prev_seasonal) + (1.0 - self.alpha) * (prev_level + prev_trend);
        self.trend = self.beta * (self.level - prev_level) + (1.0 - self.beta) * prev_trend;
        self.seasonal[idx] = self.gamma * (value - self.level) + (1.0 - self.gamma) * prev_seasonal;

        self.steps_observed += 1;

        // Forecast for t + 1
        let next_idx = self.steps_observed % self.period;
        self.level + self.trend + self.seasonal[next_idx]
    }

    pub fn steps_observed(&self) -> usize {
        self.steps_observed
    }
}

snippet-2.txt

// Low-overhead memory representation for high-cardinality metric states.
// By using 64-bit SipHash hashes instead of full labels and strings,
// we reduce the memory overhead per timeseries to a minimum.

use std::hash::{Hash, Hasher};

/// A high-speed, non-cryptographic label hasher based on SipHash 1-3.
pub fn hash_metric(metric_name: &str, labels: &[(&str, &str)]) -> u64 {
    let mut hasher = siphasher::sip::SipHasher13::new();
    hasher.write(metric_name.as_bytes());
    for &(k, v) in labels {
        hasher.write(k.as_bytes());
        hasher.write(v.as_bytes());
    }
    hasher.finish()
}

/// Dynamic metric state containing the forecasting model and rolling variance.
/// Designed for high alignment density, avoiding heap allocations in hot paths.
pub struct CompactTimeSeriesState {
    pub model: HoltWinters,
    pub variance: f64,       // Rolling variance calculated via Welford's algorithm
    pub last_timestamp: u64, // Last epoch time in seconds
}

impl CompactTimeSeriesState {
    pub fn new(model: HoltWinters, initial_timestamp: u64) -> Self {
        Self {
            model,
            variance: 0.0,
            last_timestamp: initial_timestamp,
        }
    }
}

snippet-3.txt

// Lock-free consistent hashing router designed to distribute ingested metrics
// to worker threads. This ensures each worker thread operates on a thread-local
// slice of metric states, eliminating lock contention completely.

use crossbeam_channel::{bounded, Receiver, Sender, TrySendError};

pub struct MetricPayload {
    pub key_hash: u64,
    pub timestamp: u64,
    pub value: f64,
}

pub struct MetricRouter {
    senders: Vec<Sender<MetricPayload>>,
}

impl MetricRouter {
    pub fn new(num_workers: usize, queue_capacity: usize) -> (Self, Vec<Receiver<MetricPayload>>) {
        let mut senders = Vec::with_capacity(num_workers);
        let mut receivers = Vec::with_capacity(num_workers);

        for _ in 0..num_workers {
            let (tx, rx) = bounded(queue_capacity);
            senders.push(tx);
            receivers.push(rx);
        }

        (Self { senders }, receivers)
    }

    /// Dispatches the payload using the hash of the metric labels.
    pub fn route(&self, payload: MetricPayload) -> Result<(), TrySendError<MetricPayload>> {
        let worker_idx = (payload.key_hash as usize) % self.senders.len();
        self.senders[worker_idx].try_send(payload)
    }
}

snippet-4.txt

// Thread-local worker loop. By executing inside a dedicated OS thread,
// it performs O(1) lookups in a standard `hashbrown::HashMap` without any lock overhead.

use hashbrown::HashMap;

pub struct Alert {
    pub key_hash: u64,
    pub timestamp: u64,
    pub actual: f64,
    pub predicted: f64,
    pub z_score: f64,
}

pub struct AnomalyEngineWorker {
    receiver: Receiver<MetricPayload>,
    alert_sender: Sender<Alert>,
    states: HashMap<u64, CompactTimeSeriesState>,
    alpha: f64,
    beta: f64,
    gamma: f64,
    period: usize,
}

impl AnomalyEngineWorker {
    pub fn new(
        receiver: Receiver<MetricPayload>,
        alert_sender: Sender<Alert>,
        alpha: f64,
        beta: f64,
        gamma: f64,
        period: usize,
    ) -> Self {
        Self {
            receiver,
            alert_sender,
            states: HashMap::with_capacity(100_000), // Pre-allocate to reduce rehashing cost
            alpha,
            beta,
            gamma,
            period,
        }
    }

    pub fn start(mut self) {
        for payload in self.receiver.iter() {
            let state = self.states.entry(payload.key_hash).or_insert_with(|| {
                let model = HoltWinters::new(self.alpha, self.beta, self.gamma, self.period, payload.value);
                CompactTimeSeriesState::new(model, payload.timestamp)
            });

            // Predict value using existing Holt-Winters components
            let predicted = state.model.update(payload.value);
            let diff = payload.value - predicted;

            // Exponentially Weighted Moving Variance (EWMV) update
            let prev_variance = state.variance;
            let decay = 0.05; // Smoothing factor for variance
            state.variance = (1.0 - decay) * prev_variance + decay * diff.powi(2);
            state.last_timestamp = payload.timestamp;

            let std_dev = state.variance.sqrt();
            if std_dev > 0.0 && state.model.steps_observed() > self.period {
                let z_score = diff.abs() / std_dev;
                
                // Alert if actual value exceeds 3 standard deviations from predicted
                if z_score > 3.0 {
                    let _ = self.alert_sender.send(Alert {
                        key_hash: payload.key_hash,
                        timestamp: payload.timestamp,
                        actual: payload.value,
                        predicted,
                        z_score,
                    });
                }
            }
        }
    }
}

snippet-5.txt

// Production-grade gap filling and step-forward forecasting logic.
// Simulates missing data points during networking failures to prevent model corruption.

impl AnomalyEngineWorker {
    pub fn process_with_gap_correction(
        &mut self,
        payload: MetricPayload,
        expected_interval_secs: u64,
    ) {
        let state = match self.states.get_mut(&payload.key_hash) {
            Some(s) => s,
            None => {
                let model = HoltWinters::new(self.alpha, self.beta, self.gamma, self.period, payload.value);
                self.states.insert(payload.key_hash, CompactTimeSeriesState::new(model, payload.timestamp));
                return;
            }
        };

        let elapsed = payload.timestamp.saturating_sub(state.last_timestamp);
        let missing_steps = elapsed / expected_interval_secs;

        // If gaps exceed 1 expected interval, step-forward model simulation is required
        if missing_steps > 1 {
            // Linear transition estimation for level smoothing over the missing gap
            let final_val = payload.value;
            let start_val = state.model.level;
            let slope = (final_val - start_val) / (missing_steps as f64);

            for step in 1..missing_steps {
                let simulated_value = start_val + (slope * step as f64);
                // Advance the Holt-Winters components without evaluating alerts
                state.model.update(simulated_value);
            }
        }

        // Finally, process the actual observation
        let predicted = state.model.update(payload.value);
        let diff = payload.value - predicted;
        
        let prev_variance = state.variance;
        state.variance = 0.95 * prev_variance + 0.05 * diff.powi(2);
        state.last_timestamp = payload.timestamp;

        // Perform standard alerting checks...
    }
}