ottyl/src/rate.rs

//! Ticket 040: the single windowed-rate/count implementation in the codebase, shared by
//! PromQL's `rate()` (`crate::engine::evaluate_promql`, via the `windowed_rate_series` adapter
//! there) and alert `Threshold`/`Ratio` conditions (`crate::alerts`). Previously each computed
//! its own trailing window independently — PromQL scanning indexed bucket data, alerts scanning
//! raw log-line matches with a hand-rolled `VecDeque` — and could disagree for equivalent-looking
//! rules simply because the window boundary was implemented twice.
//!
//! `count_in_window` is the one place that decides what "falls in the trailing window" means;
//! everything else in this module is built on top of it, either by batch-rescanning a full point
//! series (`rate_at`/`windowed_rate_series`, fine for PromQL's historical/one-shot evaluation) or
//! incrementally (`RateWindow`, for alerts, which see one point at a time and need cost
//! proportional to the window rather than a full rescan per ingested line).

use std::collections::VecDeque;

use crate::timestamp::Timestamp;

/// How many `points` fall in the trailing `(now - window_ms, now]` window. Points are occurrence
/// markers rather than a continuous reading, so only their timestamps matter — the boundary is
/// exclusive on the old end and inclusive of `now` itself, matching the original
/// `windowed_rate_series` behavior this replaces.
fn count_in_window(points: &[(Timestamp, f64)], now: Timestamp, window_ms: i64) -> usize {
    let now_ms = now.as_millis();
    let window_start = now_ms - window_ms;
    points
        .iter()
        .filter(|(ts, _)| {
            let ms = ts.as_millis();
            ms > window_start && ms <= now_ms
        })
        .count()
}

/// Trailing-window occurrence rate (points per second) ending at `now`: how many `points` fall in
/// `(now - window_ms, now]`, divided by the window width in seconds.
pub fn rate_at(points: &[(Timestamp, f64)], now: Timestamp, window_ms: i64) -> f64 {
    let window_seconds = window_ms as f64 / 1000.0;
    count_in_window(points, now, window_ms) as f64 / window_seconds
}

/// Evaluates `rate_at` at every one of `points`' own timestamps, turning an occurrence series
/// into a windowed rate series — used by PromQL's `rate()` so it plots as a real line rather than
/// one trailing point.
pub fn windowed_rate_series(points: &[(Timestamp, f64)], window_ms: i64) -> Vec<(Timestamp, f64)> {
    points.iter().map(|(ts, _)| (*ts, rate_at(points, *ts, window_ms))).collect()
}

/// Incrementally maintained trailing window: for a caller that sees points arrive one at a time
/// (alert `Threshold`/`Ratio` conditions, evaluated per ingested line) and needs cost proportional
/// to the window rather than `count_in_window`'s full rescan per call. Assumes points are pushed
/// in non-decreasing timestamp order, true for ingested log lines.
#[derive(Debug, Default, Clone)]
pub struct RateWindow {
    timestamps: VecDeque<Timestamp>,
}

impl RateWindow {
    /// Builds a window pre-populated from already-seen history (e.g. currently-ingested logs
    /// matching an alert's query), so a freshly-created window starts with whatever of that
    /// history already falls in the trailing window instead of empty.
    pub fn backfill(timestamps: impl IntoIterator<Item = Timestamp>, now: Timestamp, window_ms: i64) -> Self {
        let mut window = Self::default();
        for ts in timestamps {
            window.push(ts, window_ms);
        }
        window.prune(now, window_ms);
        window
    }

    /// Records one occurrence at `timestamp` and drops anything that has aged out of the
    /// resulting trailing window.
    pub fn push(&mut self, timestamp: Timestamp, window_ms: i64) {
        self.timestamps.push_back(timestamp);
        self.prune(timestamp, window_ms);
    }

    /// Same boundary as `count_in_window`: drops entries at or before `now - window_ms`.
    fn prune(&mut self, now: Timestamp, window_ms: i64) {
        let window_start = now.as_millis() - window_ms;
        while let Some(&front) = self.timestamps.front() {
            if front.as_millis() <= window_start {
                self.timestamps.pop_front();
            } else {
                break;
            }
        }
    }

    pub fn count(&self) -> usize {
        self.timestamps.len()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn rate_at_matches_hand_computed_window() {
        let points = vec![
            (Timestamp::from_millis(0), 1.0),
            (Timestamp::from_millis(1000), 1.0),
            (Timestamp::from_millis(2000), 1.0),
        ];
        // (1000, 2000]: only the point at 2000.
        assert_eq!(rate_at(&points, Timestamp::from_millis(2000), 1000), 1.0);
        // (0, 2000]: points at 1000 and 2000 -> 2 points / 2s.
        assert_eq!(rate_at(&points, Timestamp::from_millis(2000), 2000), 1.0);
    }

    #[test]
    fn windowed_rate_series_matches_rate_at_at_each_point() {
        let points = vec![
            (Timestamp::from_millis(0), 1.0),
            (Timestamp::from_millis(500), 1.0),
            (Timestamp::from_millis(1000), 1.0),
        ];
        let series = windowed_rate_series(&points, 1000);
        for (ts, rate) in &series {
            assert_eq!(*rate, rate_at(&points, *ts, 1000));
        }
    }

    #[test]
    fn rate_window_incremental_count_matches_batch_count_in_window() {
        let mut window = RateWindow::default();
        let millis = [0, 200, 400, 1200];
        for ms in millis {
            window.push(Timestamp::from_millis(ms), 1000);
        }
        let points: Vec<(Timestamp, f64)> =
            millis.iter().map(|ms| (Timestamp::from_millis(*ms), 1.0)).collect();
        assert_eq!(window.count(), count_in_window(&points, Timestamp::from_millis(1200), 1000));
    }

    #[test]
    fn rate_window_backfill_seeds_from_history_within_the_window() {
        let history = [0, 500, 2000, 2500].map(Timestamp::from_millis);
        // now=2500ms, window=1000ms -> window (1500, 2500]: only 2000 and 2500 qualify.
        let window = RateWindow::backfill(history, Timestamp::from_millis(2500), 1000);
        assert_eq!(window.count(), 2);
    }
}