diff --git a/crates/core/src/metrics/accumulator.rs b/crates/core/src/metrics/accumulator.rs new file mode 100644 index 000000000..ffdec3344 --- /dev/null +++ b/crates/core/src/metrics/accumulator.rs @@ -0,0 +1,201 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Per-key point storage for `MetricsCollector`. +//! +//! Points are kept in chunks of at most [`CHUNK_CAP`]. Taking the old points +//! out moves whole chunks and splits only the chunks that straddle the cutoff, +//! and a `record` call reallocates at most one chunk. + +use std::collections::VecDeque; +use std::time::{Instant, SystemTime}; + +use super::collector::window_cutoff; +use super::types::TimeWindow; + +/// Most points one chunk holds: the largest reallocation a `record` call can +/// cause, and the most points a drain splits one by one per straddling chunk. +pub(super) const CHUNK_CAP: usize = 1024; + +/// Capacity of a key's first chunk; most keys never fill one. +const FIRST_CHUNK_CAP: usize = 64; + +/// A single data point recorded for a metric. +#[derive(Debug, Clone)] +pub(super) struct DataPoint { + pub(super) value: f64, + pub(super) timestamp: Instant, + /// Wall-clock time for DB persistence. Truncated to minute boundary on flush. + pub(super) wall_time: SystemTime, +} + +/// Points in record order, with the range of their timestamps. +#[derive(Debug)] +struct Chunk { + points: Vec, + oldest: Instant, + newest: Instant, +} + +impl Chunk { + fn new(point: DataPoint, capacity: usize) -> Self { + let mut points = Vec::with_capacity(capacity); + let (oldest, newest) = (point.timestamp, point.timestamp); + points.push(point); + Self { + points, + oldest, + newest, + } + } + + fn push(&mut self, point: DataPoint) { + self.oldest = self.oldest.min(point.timestamp); + self.newest = self.newest.max(point.timestamp); + self.points.push(point); + } + + /// Moves the points that `is_old` selects out of the chunk, keeping order. + fn split_off_old(&mut self, is_old: impl Fn(Instant) -> bool) -> Vec { + let (old, young): (Vec<_>, Vec<_>) = + self.points.drain(..).partition(|p| is_old(p.timestamp)); + self.points = young; + if let Some(first) = self.points.first() { + let start = (first.timestamp, first.timestamp); + (self.oldest, self.newest) = self.points.iter().fold(start, |(lo, hi), p| { + (lo.min(p.timestamp), hi.max(p.timestamp)) + }); + } + old + } +} + +/// Accumulator for a single metric+dimension combination. +#[derive(Debug)] +pub(super) struct Accumulator { + chunks: VecDeque, +} + +impl Accumulator { + pub(super) fn new() -> Self { + Self { + chunks: VecDeque::new(), + } + } + + pub(super) fn record(&mut self, value: f64, now: Instant, wall_time: SystemTime) { + let point = DataPoint { + value, + timestamp: now, + wall_time, + }; + match self.chunks.back_mut() { + Some(chunk) if chunk.points.len() < CHUNK_CAP => chunk.push(point), + Some(_) => self.chunks.push_back(Chunk::new(point, CHUNK_CAP)), + None => self.chunks.push_back(Chunk::new(point, FIRST_CHUNK_CAP)), + } + } + + pub(super) fn is_empty(&self) -> bool { + self.chunks.is_empty() + } + + /// All points, in record order. + pub(super) fn points(&self) -> impl Iterator { + self.chunks.iter().flat_map(|c| c.points.iter()) + } + + /// Number of chunks held. + pub(super) fn chunk_count(&self) -> usize { + self.chunks.len() + } + + /// Appends the points of `newer`, recorded after every point held here. + /// Moves chunk headers only: amortized O(chunks of `newer`). + pub(super) fn append(&mut self, mut newer: Accumulator) { + self.chunks.append(&mut newer.chunks); + } + + /// Removes and returns the points at or before `cutoff`, in record order. + pub(super) fn take_expired(&mut self, cutoff: Instant) -> Vec> { + self.take_old(|t| t <= cutoff) + } + + /// Prune points older than the retention window (1 day). + pub(super) fn prune(&mut self, cutoff: Instant) { + self.take_old(|t| t < cutoff); + } + + /// Moves out every point that `is_old` selects. Chunks whose newest point + /// is old move whole; only chunks holding both old and young points are + /// split. Points arrive almost in time order, so that is about one chunk. + fn take_old(&mut self, is_old: impl Fn(Instant) -> bool) -> Vec> { + let mut out = Vec::new(); + while self.chunks.front().is_some_and(|c| is_old(c.newest)) { + if let Some(chunk) = self.chunks.pop_front() { + out.push(chunk.points); + } + } + let mut emptied = false; + for chunk in self.chunks.iter_mut().filter(|c| is_old(c.oldest)) { + if is_old(chunk.newest) { + out.push(std::mem::take(&mut chunk.points)); + } else { + out.push(chunk.split_off_old(&is_old)); + } + emptied |= chunk.points.is_empty(); + } + if emptied { + self.chunks.retain(|c| !c.points.is_empty()); + } + out + } + + pub(super) fn snapshot(&self, window: TimeWindow, now: Instant) -> Option { + let cutoff = window_cutoff(window, now); + let values: Vec = match cutoff { + Some(c) => self + .points() + .filter(|p| p.timestamp >= c) + .map(|p| p.value) + .collect(), + None => self.points().map(|p| p.value).collect(), + }; + + if values.is_empty() { + return None; + } + + let sum: f64 = values.iter().sum(); + #[allow(clippy::cast_possible_truncation)] + let count = values.len() as u64; + let min = values.iter().copied().fold(f64::INFINITY, f64::min); + let max = values.iter().copied().fold(f64::NEG_INFINITY, f64::max); + + Some(AccumulatorSnapshot { + sum, + count, + min, + max, + values, + }) + } + + /// Chunk lengths and capacities, for tests of the storage bound. + #[cfg(test)] + pub(super) fn chunk_shapes(&self) -> Vec<(usize, usize)> { + self.chunks + .iter() + .map(|c| (c.points.len(), c.points.capacity())) + .collect() + } +} + +/// Snapshot of an accumulator's data for a given time window. +pub(super) struct AccumulatorSnapshot { + pub(super) sum: f64, + pub(super) count: u64, + pub(super) min: f64, + pub(super) max: f64, + pub(super) values: Vec, +} diff --git a/crates/core/src/metrics/collector.rs b/crates/core/src/metrics/collector.rs index 56d416a25..689b900a5 100755 --- a/crates/core/src/metrics/collector.rs +++ b/crates/core/src/metrics/collector.rs @@ -5,87 +5,12 @@ //! per-key aggregation. use std::collections::HashMap; -use std::sync::RwLock; +use std::sync::{Mutex, RwLock}; use std::time::{Duration, Instant, SystemTime}; +use super::accumulator::Accumulator; use super::types::{LatencySegments, MetricName, TimeWindow}; -/// A single data point recorded for a metric. -#[derive(Debug, Clone)] -pub(super) struct DataPoint { - pub(super) value: f64, - pub(super) timestamp: Instant, - /// Wall-clock time for DB persistence. Truncated to minute boundary on flush. - pub(super) wall_time: SystemTime, -} - -/// Accumulator for a single metric+dimension combination. -#[derive(Debug)] -pub(super) struct Accumulator { - pub(super) points: Vec, -} - -impl Accumulator { - fn new() -> Self { - Self { - points: Vec::with_capacity(64), - } - } - - fn record(&mut self, value: f64, now: Instant, wall_time: SystemTime) { - self.points.push(DataPoint { - value, - timestamp: now, - wall_time, - }); - } - - /// Prune points older than the retention window (1 day). - fn prune(&mut self, cutoff: Instant) { - self.points.retain(|p| p.timestamp >= cutoff); - } - - pub(super) fn snapshot(&self, window: TimeWindow, now: Instant) -> Option { - let cutoff = window_cutoff(window, now); - let values: Vec = match cutoff { - Some(c) => self - .points - .iter() - .filter(|p| p.timestamp >= c) - .map(|p| p.value) - .collect(), - None => self.points.iter().map(|p| p.value).collect(), - }; - - if values.is_empty() { - return None; - } - - let sum: f64 = values.iter().sum(); - #[allow(clippy::cast_possible_truncation)] - let count = values.len() as u64; - let min = values.iter().copied().fold(f64::INFINITY, f64::min); - let max = values.iter().copied().fold(f64::NEG_INFINITY, f64::max); - - Some(AccumulatorSnapshot { - sum, - count, - min, - max, - values, - }) - } -} - -/// Snapshot of an accumulator's data for a given time window. -pub(super) struct AccumulatorSnapshot { - pub(super) sum: f64, - pub(super) count: u64, - pub(super) min: f64, - pub(super) max: f64, - pub(super) values: Vec, -} - /// Returns `None` for `AllTime` (no cutoff), or `Some(cutoff)` for bounded windows. pub(super) fn window_cutoff(window: TimeWindow, now: Instant) -> Option { let dur = match window { @@ -115,6 +40,9 @@ pub struct MetricsCollector { pub(super) data: RwLock>, /// Per-operation latency segment breakdowns for the console deep-dive. pub(super) segments: RwLock>, + /// Serializes drains, so each one takes and keeps points in record order. + /// `record` and `prune` never take it. + pub(super) drain_lock: Mutex<()>, } /// A single latency segment data point with metadata. @@ -134,6 +62,7 @@ impl MetricsCollector { Self { data: RwLock::new(HashMap::new()), segments: RwLock::new(Vec::with_capacity(256)), + drain_lock: Mutex::new(()), } } @@ -152,8 +81,16 @@ impl MetricsCollector { index_name: index_name.map(ToOwned::to_owned), operation: operation.map(ToOwned::to_owned), }; - let now = Instant::now(); - let wall_time = SystemTime::now(); + self.record_point(key, value, Instant::now(), SystemTime::now()); + } + + pub(super) fn record_point( + &self, + key: MetricKey, + value: f64, + now: Instant, + wall_time: SystemTime, + ) { // Poisoned lock: silently skip rather than panic (async safety rule). let Ok(mut map) = self.data.write() else { return; @@ -409,7 +346,7 @@ impl MetricsCollector { for acc in map.values_mut() { acc.prune(cutoff); } - map.retain(|_, acc| !acc.points.is_empty()); + map.retain(|_, acc| !acc.is_empty()); drop(map); // Prune segment points too. if let Ok(mut segs) = self.segments.write() { diff --git a/crates/core/src/metrics/collector_query.rs b/crates/core/src/metrics/collector_query.rs index c932ccde3..7ce9af6ff 100755 --- a/crates/core/src/metrics/collector_query.rs +++ b/crates/core/src/metrics/collector_query.rs @@ -6,11 +6,14 @@ //! Split from `collector.rs` to keep both files under the 500-line limit. use std::collections::HashMap; +use std::collections::hash_map::Entry; use std::time::{Duration, Instant, SystemTime}; -use super::collector::{MetricsCollector, window_cutoff}; +use super::accumulator::{Accumulator, DataPoint}; +use super::collector::{MetricKey, MetricsCollector, window_cutoff}; use super::types::{ - Dimension, MetricName, MetricSnapshot, MetricsQuery, OperationSegments, Percentiles, TimeWindow, + Dimension, FlushBucket, MetricName, MetricSnapshot, MetricsQuery, OperationSegments, + Percentiles, TimeWindow, }; /// Accumulator for per-operation segment sums: (auth, authz, throttle, dispatch, response, total, count). @@ -18,6 +21,9 @@ type SegmentAccum = (f64, f64, f64, f64, f64, f64, u64); impl MetricsCollector { /// Query metrics and return snapshots. + /// + /// While a [`drain`](Self::drain) splits its points, the buffered points + /// are out of the map and this query does not see them. #[must_use] pub fn query(&self, params: &MetricsQuery) -> Vec { let now = Instant::now(); @@ -91,68 +97,47 @@ impl MetricsCollector { /// `FlushBucket`s for DB persistence. Points newer than `age` are kept /// in memory for the next flush cycle. /// - /// The write lock is held only for the partition step; aggregation into - /// buckets happens outside the lock to avoid blocking `record_*` calls. - pub fn drain(&self, age: Duration) -> Vec { + /// The drain takes the write lock twice: to swap the map out, and to put + /// the younger points back ahead of the points recorded meanwhile. Neither + /// step touches the points held, so `record_*` calls wait a bounded time + /// however many points are buffered. Between the two steps an in-memory + /// `query` does not see the buffered points; the server reads metrics + /// from its store. The swap, split, and put-back of concurrent drains run + /// one at a time; their aggregation can overlap. The split and the + /// aggregation are CPU-bound: async callers should run the drain on a + /// blocking thread. + pub fn drain(&self, age: Duration) -> Vec { let cutoff = Instant::now().checked_sub(age).unwrap_or(Instant::now()); + aggregate(self.drain_points(cutoff, |_| {}, |_| {}).0) + } - // Phase 1: under write lock, partition old points out and collect them. - let drained: Vec<( - super::collector::MetricKey, - Vec, - )> = { + /// The points of a drain, and the chunks its second locked step moved. + /// `before_split` and `after_split` see the map taken by the first locked + /// step on each side of the split, without the write lock. + pub(super) fn drain_points( + &self, + cutoff: Instant, + before_split: impl FnOnce(&PointMap), + after_split: impl FnOnce(&PointMap), + ) -> (Drained, usize) { + let _serial = self + .drain_lock + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let mut held = { let Ok(mut map) = self.data.write() else { - return Vec::new(); + return (Vec::new(), 0); }; - let mut result = Vec::new(); - for (key, acc) in map.iter_mut() { - let (old, new): (Vec<_>, Vec<_>) = - acc.points.drain(..).partition(|p| p.timestamp <= cutoff); - acc.points = new; - if !old.is_empty() { - result.push((key.clone(), old)); - } - } - map.retain(|_, acc| !acc.points.is_empty()); - result + std::mem::take(&mut *map) }; - - // Phase 2: aggregate outside the lock. - let mut buckets: HashMap<(super::collector::MetricKey, i64), (f64, u64, f64, f64)> = - HashMap::new(); - for (key, points) in drained { - for dp in points { - let minute = truncate_to_minute(dp.wall_time); - let e = buckets.entry((key.clone(), minute)).or_insert(( - 0.0, - 0, - f64::INFINITY, - f64::NEG_INFINITY, - )); - e.0 += dp.value; - e.1 += 1; - e.2 = e.2.min(dp.value); - e.3 = e.3.max(dp.value); - } - } - - buckets - .into_iter() - .map( - |((key, minute), (sum, count, min, max))| super::types::FlushBucket { - bucket: SystemTime::UNIX_EPOCH - + Duration::from_secs(u64::try_from(minute).unwrap_or(0)), - metric: key.metric, - table_name: key.table_name.unwrap_or_default(), - index_name: key.index_name.unwrap_or_default(), - operation: key.operation.unwrap_or_default(), - sum, - count, - min, - max, - }, - ) - .collect() + before_split(&held); + let drained = take_expired(&mut held, cutoff); + after_split(&held); + let Ok(mut map) = self.data.write() else { + return (drained, 0); + }; + let moved = reattach(&mut map, held); + (drained, moved) } /// Query average latency segments for the console deep-dive. @@ -238,3 +223,78 @@ fn truncate_to_minute(t: SystemTime) -> i64 { let secs_i64 = secs as i64; secs_i64 - (secs_i64 % 60) } + +/// Drained points per key, in record order. +type Drained = Vec<(MetricKey, Vec>)>; + +pub(super) type PointMap = HashMap; + +/// Removes every point at or before `cutoff` from `held`, which the caller +/// has taken out of the collector, and drops the keys left empty. +pub(super) fn take_expired(held: &mut PointMap, cutoff: Instant) -> Drained { + let mut drained = Vec::new(); + for (key, acc) in held.iter_mut() { + let old = acc.take_expired(cutoff); + if !old.is_empty() { + drained.push((key.clone(), old)); + } + } + held.retain(|_, acc| !acc.is_empty()); + drained +} + +/// The second locked step of a drain: makes `young` the map again and appends +/// the points recorded since the first step. Returns the chunks it moved, which +/// are only the chunks of those new points. +pub(super) fn reattach(map: &mut PointMap, young: PointMap) -> usize { + let recorded = std::mem::replace(map, young); + let mut moved = 0; + for (key, acc) in recorded { + moved += acc.chunk_count(); + match map.entry(key) { + Entry::Occupied(mut e) => e.get_mut().append(acc), + Entry::Vacant(e) => { + e.insert(acc); + } + } + } + moved +} + +/// Sums drained points into 1-minute buckets per key, in record order. +pub(super) fn aggregate(drained: Drained) -> Vec { + let mut out = Vec::new(); + for (key, chunks) in drained { + let mut by_minute: HashMap = HashMap::new(); + for dp in chunks.iter().flatten() { + let e = by_minute + .entry(truncate_to_minute(dp.wall_time)) + .or_insert((0.0, 0, f64::INFINITY, f64::NEG_INFINITY)); + e.0 += dp.value; + e.1 += 1; + e.2 = e.2.min(dp.value); + e.3 = e.3.max(dp.value); + } + out.extend( + by_minute + .into_iter() + .map(|(minute, (sum, count, min, max))| FlushBucket { + bucket: SystemTime::UNIX_EPOCH + + Duration::from_secs(u64::try_from(minute).unwrap_or(0)), + metric: key.metric, + table_name: key.table_name.clone().unwrap_or_default(), + index_name: key.index_name.clone().unwrap_or_default(), + operation: key.operation.clone().unwrap_or_default(), + sum, + count, + min, + max, + }), + ); + } + out +} + +#[cfg(test)] +#[path = "drain_tests.rs"] +mod tests; diff --git a/crates/core/src/metrics/drain_tests.rs b/crates/core/src/metrics/drain_tests.rs new file mode 100644 index 000000000..2945d339c --- /dev/null +++ b/crates/core/src/metrics/drain_tests.rs @@ -0,0 +1,424 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! Tests for the drain: what it moves under the write lock, and that it drains +//! and keeps exactly the points a plain partition by cutoff would. + +use super::*; +use crate::metrics::accumulator::CHUNK_CAP; +use crate::metrics::types::{MetricsQuery, TimeWindow}; + +/// Small deterministic generator (xorshift64) for timestamp jitter. +struct Rng(u64); + +impl Rng { + fn next(&mut self) -> u64 { + self.0 ^= self.0 << 13; + self.0 ^= self.0 >> 7; + self.0 ^= self.0 << 17; + self.0 + } +} + +/// One recorded point, as the reference model sees it. +#[derive(Debug, Clone, Copy, PartialEq)] +struct Pt { + value: f64, + timestamp: Instant, + wall_time: SystemTime, +} + +impl From<&DataPoint> for Pt { + fn from(p: &DataPoint) -> Self { + Self { + value: p.value, + timestamp: p.timestamp, + wall_time: p.wall_time, + } + } +} + +fn keys(n: usize) -> Vec { + (0..n) + .map(|i| MetricKey { + metric: MetricName::SuccessfulRequestLatency, + table_name: Some(format!("table-{i}")), + index_name: None, + operation: Some("GetItem".to_owned()), + }) + .collect() +} + +/// Records `per_key` points for every key, interleaved across keys like the +/// request path does. Point `i` is at `base + i` microseconds, moved by up to +/// `jitter_us` either way, and its wall time advances 1 ms per point. +/// Returns the points per key in record order. +fn fill( + c: &MetricsCollector, + keys: &[MetricKey], + per_key: usize, + base: Instant, + jitter_us: u64, + rng: &mut Rng, +) -> Vec> { + let wall0 = SystemTime::UNIX_EPOCH + Duration::from_secs(1_800_000_000); + let mut recorded = vec![Vec::with_capacity(per_key); keys.len()]; + for i in 0..per_key { + for (k, key) in keys.iter().enumerate() { + let shift = if jitter_us == 0 { + 0 + } else { + rng.next() % (2 * jitter_us + 1) + }; + let micros = (i as u64 + jitter_us).saturating_sub(shift); + let p = Pt { + value: (i * keys.len() + k) as f64, + timestamp: base + Duration::from_micros(micros), + wall_time: wall0 + Duration::from_millis(i as u64), + }; + c.record_point(key.clone(), p.value, p.timestamp, p.wall_time); + recorded[k].push(p); + } + } + recorded +} + +fn drain_at(c: &MetricsCollector, cutoff: Instant) -> Drained { + c.drain_points(cutoff, |_| {}, |_| {}).0 +} + +fn kept(c: &MetricsCollector, key: &MetricKey) -> Vec { + let map = c.data.read().expect("lock"); + map.get(key) + .map(|acc| acc.points().map(Pt::from).collect()) + .unwrap_or_default() +} + +fn drained_of(drained: &Drained, key: &MetricKey) -> Vec { + drained + .iter() + .filter(|(k, _)| k == key) + .flat_map(|(_, chunks)| chunks.iter().flatten().map(Pt::from)) + .collect() +} + +/// Under the lock a drain moves no buffered point: the first step swaps the +/// map out and releases the lock before the split, the lock stays free and the +/// drain mutex stays held until the split is done, and the second step moves +/// back only the chunks recorded in between, the same count with 100 times +/// more points held. `moved` is reported by the code under test, so this +/// proves the structure of the second step, not its time. +#[test] +fn drain_moves_only_newly_recorded_chunks_under_the_lock() { + let keys = keys(7); + let mut moved_by_size = Vec::new(); + for per_key in [4_000, 400_000] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 0, &mut Rng(1)); + let half = (per_key / 2) as u64; + let late = base + Duration::from_secs(10); + let fresh = MetricKey { + operation: Some("PutItem".to_owned()), + ..keys[0].clone() + }; + let wall = SystemTime::UNIX_EPOCH; + let (drained, moved) = c.drain_points( + base + Duration::from_micros(half), + |held| { + // Every point is still in the taken map (the split has not + // run) while the write lock is free and the collector's map is + // empty. + let unsplit: usize = held.values().map(|acc| acc.points().count()).sum(); + assert_eq!( + unsplit, + keys.len() * per_key, + "split ran before the write lock was released" + ); + assert!(c.data.try_write().is_ok(), "lock held before the split"); + assert!(c.data.read().expect("lock").is_empty()); + assert!(c.drain_lock.try_lock().is_err(), "drains not serialized"); + for v in [1.0, 2.0] { + c.record_point(keys[0].clone(), -v, late, wall); + } + c.record_point(fresh.clone(), -4.0, late, wall); + }, + |held| { + // The split is done: only the younger points are left. The + // write lock is still free and the drain mutex still held. + let young: usize = held.values().map(|acc| acc.points().count()).sum(); + assert_eq!(young, keys.len() * (per_key - per_key / 2 - 1)); + assert!(c.data.try_write().is_ok(), "lock held over the split"); + assert!( + c.drain_lock.try_lock().is_err(), + "drain mutex released before the split ended" + ); + c.record_point(keys[0].clone(), -3.0, late, wall); + }, + ); + moved_by_size.push(moved); + + let drained_points: usize = drained + .iter() + .flat_map(|(_, chunks)| chunks.iter()) + .map(Vec::len) + .sum(); + assert_eq!(drained_points, keys.len() * (per_key / 2 + 1)); + // The points recorded during the drain follow the kept ones. + let mut want = recorded[0][per_key / 2 + 1..].to_vec(); + for v in [1.0, 2.0, 3.0] { + want.push(Pt { + value: -v, + timestamp: late, + wall_time: wall, + }); + } + assert_eq!(kept(&c, &keys[0]), want); + assert_eq!(kept(&c, &fresh).len(), 1); + assert_eq!(kept(&c, &keys[1]), recorded[1][per_key / 2 + 1..].to_vec()); + } + assert_eq!(moved_by_size, vec![2, 2], "chunks moved under the lock"); +} + +/// Points recorded out of order (the request path reads the clock before it +/// takes the lock) are drained and kept exactly as a partition would. +#[test] +fn drain_matches_a_partition_with_out_of_order_points() { + let keys = keys(3); + let per_key = 3 * CHUNK_CAP + 17; + for (seed, cut) in [(7, 0), (11, 500), (13, CHUNK_CAP), (17, 2_000), (19, 5_000)] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 300, &mut Rng(seed)); + let cutoff = base + Duration::from_micros(cut as u64); + let drained = drain_at(&c, cutoff); + + for (k, key) in keys.iter().enumerate() { + let (old, young): (Vec, Vec) = + recorded[k].iter().partition(|p| p.timestamp <= cutoff); + assert_eq!(drained_of(&drained, key), old, "drained, cutoff {cut}"); + assert_eq!(kept(&c, key), young, "kept, cutoff {cut}"); + } + } +} + +/// A second drain at a later cutoff takes the next points, and a drain of +/// everything leaves no key behind. +#[test] +fn successive_drains_take_every_point_once() { + let keys = keys(2); + let per_key = 5 * CHUNK_CAP; + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 50, &mut Rng(3)); + + let mut seen = vec![Vec::new(); keys.len()]; + for cut in [1_000_u64, 1_001, 2_500, 4_000, 10_000] { + let drained = drain_at(&c, base + Duration::from_micros(cut)); + for (k, key) in keys.iter().enumerate() { + seen[k].extend(drained_of(&drained, key)); + } + } + for k in 0..keys.len() { + let mut want = recorded[k].clone(); + want.sort_by_key(|p| p.timestamp); + seen[k].sort_by_key(|p| p.timestamp); + assert_eq!(seen[k], want); + } + assert!(c.data.read().expect("lock").is_empty()); +} + +type BucketRow = ( + String, + String, + String, + String, + SystemTime, + u64, + u64, + u64, + u64, +); + +fn rows(buckets: Vec) -> Vec { + let mut rows: Vec = buckets + .into_iter() + .map(|b| { + ( + b.metric.to_string(), + b.table_name, + b.index_name, + b.operation, + b.bucket, + b.sum.to_bits(), + b.count, + b.min.to_bits(), + b.max.to_bits(), + ) + }) + .collect(); + rows.sort_by(|a, b| (&a.0, &a.1, &a.2, &a.3, a.4).cmp(&(&b.0, &b.1, &b.2, &b.3, b.4))); + rows +} + +/// The flush buckets are the per-minute sum, count, min, and max of the +/// drained points, summed in record order. +#[test] +fn drain_buckets_match_a_per_minute_reference() { + let keys = keys(3); + let per_key = 150_000; // 150 s of wall time per key: three minute buckets + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 200, &mut Rng(5)); + let cutoff = base + Duration::from_micros(100_000); + let drained = drain_at(&c, cutoff); + + let mut want = Vec::new(); + for (k, key) in keys.iter().enumerate() { + let mut by_minute: Vec<(i64, (f64, u64, f64, f64))> = Vec::new(); + for p in recorded[k].iter().filter(|p| p.timestamp <= cutoff) { + let minute = truncate_to_minute(p.wall_time); + if by_minute.last().is_none_or(|(m, _)| *m != minute) { + by_minute.push((minute, (0.0, 0, f64::INFINITY, f64::NEG_INFINITY))); + } + let e = &mut by_minute.last_mut().expect("bucket").1; + e.0 += p.value; + e.1 += 1; + e.2 = e.2.min(p.value); + e.3 = e.3.max(p.value); + } + for (minute, (sum, count, min, max)) in by_minute { + want.push(FlushBucket { + bucket: SystemTime::UNIX_EPOCH + Duration::from_secs(minute as u64), + metric: key.metric, + table_name: key.table_name.clone().unwrap_or_default(), + index_name: String::new(), + operation: key.operation.clone().unwrap_or_default(), + sum, + count, + min, + max, + }); + } + } + assert_eq!(rows(aggregate(drained)), rows(want)); +} + +/// `drain(ZERO)`, the shutdown flush, drains every point recorded before it. +#[test] +fn drain_zero_takes_everything() { + let c = MetricsCollector::new(); + for i in 0..(2 * CHUNK_CAP + 5) { + c.record_latency(Some("T"), "GetItem", i as f64); + c.record_request_count("GetItem"); + } + let buckets = c.drain(Duration::ZERO); + let count: u64 = buckets.iter().map(|b| b.count).sum(); + assert_eq!(count, 2 * (2 * CHUNK_CAP as u64 + 5)); + assert!(c.data.read().expect("lock").is_empty()); +} + +/// `record` never grows a chunk past `CHUNK_CAP`, so its largest reallocation +/// under the lock is one chunk, not every point of the key. +#[test] +fn record_keeps_every_chunk_within_the_cap() { + let c = MetricsCollector::new(); + for i in 0..(10 * CHUNK_CAP + 3) { + c.record_latency(Some("T"), "GetItem", i as f64); + } + let map = c.data.read().expect("lock"); + let acc = map.values().next().expect("one key"); + let shapes = acc.chunk_shapes(); + assert_eq!(shapes.len(), 11); + for (len, capacity) in &shapes { + assert!( + *len <= CHUNK_CAP && *capacity <= CHUNK_CAP, + "{len}/{capacity}" + ); + } + let values: Vec = acc.points().map(|p| p.value).collect(); + let want: Vec = (0..(10 * CHUNK_CAP + 3)).map(|i| i as f64).collect(); + assert_eq!(values, want, "points out of record order"); +} + +/// The prune keeps exactly the points at or after its cutoff. +#[test] +fn prune_matches_a_retain_with_out_of_order_points() { + let keys = keys(2); + let per_key = 4 * CHUNK_CAP; + for cut in [0_u64, 777, 2_048, 3_333, 9_999] { + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, per_key, base, 100, &mut Rng(23)); + let cutoff = base + Duration::from_micros(cut); + for acc in c.data.write().expect("lock").values_mut() { + acc.prune(cutoff); + } + for (k, key) in keys.iter().enumerate() { + let want: Vec = recorded[k] + .iter() + .copied() + .filter(|p| p.timestamp >= cutoff) + .collect(); + assert_eq!(kept(&c, key), want, "cutoff {cut}"); + } + } +} + +/// An in-memory query after a drain reports the same sum, count, min, and +/// max as the kept points, summed in record order. +#[test] +fn query_after_drain_reports_the_kept_points() { + let keys = keys(1); + let c = MetricsCollector::new(); + let base = Instant::now(); + let recorded = fill(&c, &keys, 3 * CHUNK_CAP, base, 40, &mut Rng(29)); + let cutoff = base + Duration::from_micros(1_500); + drain_at(&c, cutoff); + + let young: Vec = recorded[0] + .iter() + .filter(|p| p.timestamp > cutoff) + .map(|p| p.value) + .collect(); + let snaps = c.query(&MetricsQuery { + window: Some(TimeWindow::AllTime), + ..Default::default() + }); + assert_eq!(snaps.len(), 1); + let s = &snaps[0]; + assert_eq!(s.count, young.len() as u64); + assert_eq!(s.sum.to_bits(), young.iter().sum::().to_bits()); + assert_eq!(s.min, young.iter().copied().fold(f64::INFINITY, f64::min)); + assert_eq!( + s.max, + young.iter().copied().fold(f64::NEG_INFINITY, f64::max) + ); +} + +/// A fully old chunk behind a chunk that holds a young point is taken whole, +/// and the result still matches a partition. +#[test] +fn drain_takes_a_fully_old_chunk_behind_a_young_point() { + let key = keys(1).remove(0); + let c = MetricsCollector::new(); + let base = Instant::now(); + let wall = SystemTime::UNIX_EPOCH; + let mut recorded = Vec::new(); + for i in 0..(2 * CHUNK_CAP) { + // One late-read young point at the head of the first chunk. + let micros = if i == 0 { 10_000 } else { i as u64 }; + let p = Pt { + value: i as f64, + timestamp: base + Duration::from_micros(micros), + wall_time: wall, + }; + c.record_point(key.clone(), p.value, p.timestamp, p.wall_time); + recorded.push(p); + } + let cutoff = base + Duration::from_micros(5_000); + let drained = drain_at(&c, cutoff); + let (old, young): (Vec, Vec) = recorded.iter().partition(|p| p.timestamp <= cutoff); + assert_eq!(drained_of(&drained, &key), old); + assert_eq!(kept(&c, &key), young); +} diff --git a/crates/core/src/metrics/mod.rs b/crates/core/src/metrics/mod.rs index 4db7c40fd..bf1ff0272 100755 --- a/crates/core/src/metrics/mod.rs +++ b/crates/core/src/metrics/mod.rs @@ -10,6 +10,7 @@ //! //! REQ-OBS-005: Metrics endpoint. +mod accumulator; mod collector; mod collector_query; mod types; diff --git a/crates/server/src/workers.rs b/crates/server/src/workers.rs index 372fdb45a..f519f91c3 100755 --- a/crates/server/src/workers.rs +++ b/crates/server/src/workers.rs @@ -198,7 +198,19 @@ pub(crate) async fn metrics_flush_worker( Duration::ZERO }; let cycle_start = std::time::Instant::now(); - let buckets = metrics.drain(drain_age); + // The drain aggregates every point it takes: keep it off the runtime + // workers so they keep polling sockets while it runs. + let drain_metrics = Arc::clone(&metrics); + let mut errored = false; + let buckets = + match tokio::task::spawn_blocking(move || drain_metrics.drain(drain_age)).await { + Ok(buckets) => buckets, + Err(e) => { + tracing::warn!("Metrics drain task failed: {e}"); + errored = true; + Vec::new() + } + }; if !buckets.is_empty() { let rows: Vec = buckets .iter() @@ -240,13 +252,13 @@ pub(crate) async fn metrics_flush_worker( let _ = store.insert_metrics(&rows).await; } // Prune old DB rows. - let mut errored = false; if let Err(e) = store.prune_metrics(RETENTION).await { tracing::warn!("Failed to prune old metrics from DB: {e:?}"); - metrics.record_worker_error(QuerySource::MetricsFlush); errored = true; } - if !errored { + if errored { + metrics.record_worker_error(QuerySource::MetricsFlush); + } else { #[allow(clippy::cast_precision_loss)] let cycle_us = cycle_start.elapsed().as_micros() as f64; metrics.record_worker_success(QuerySource::MetricsFlush, cycle_us);