use std::sync::atomic::{AtomicU64, Ordering};
use crate::{
CapacityDimension, DatabaseDisposition, Health, LifecycleState, OutboundResult, Stage,
StageOutcome,
};
pub const LATENCY_BUCKET_UPPER_MS: [u64; 11] =
[1, 5, 10, 25, 50, 100, 250, 500, 1_000, 5_000, u64::MAX];
pub(crate) struct Metrics {
requests: [AtomicU64; 4],
stage_latency: [[AtomicU64; 11]; 7],
capacity_accepted: [AtomicU64; 5],
capacity_rejected: [AtomicU64; 5],
capacity_used: [AtomicU64; 5],
database: [AtomicU64; 3],
outbound: [AtomicU64; 4],
logger_dropped: AtomicU64,
logger_output_failed: AtomicU64,
lifecycle: AtomicU64,
health: AtomicU64,
ready: AtomicU64,
}
impl Default for Metrics {
fn default() -> Self {
Self {
requests: std::array::from_fn(|_| AtomicU64::new(0)),
stage_latency: std::array::from_fn(|_| std::array::from_fn(|_| AtomicU64::new(0))),
capacity_accepted: std::array::from_fn(|_| AtomicU64::new(0)),
capacity_rejected: std::array::from_fn(|_| AtomicU64::new(0)),
capacity_used: std::array::from_fn(|_| AtomicU64::new(0)),
database: std::array::from_fn(|_| AtomicU64::new(0)),
outbound: std::array::from_fn(|_| AtomicU64::new(0)),
logger_dropped: AtomicU64::new(0),
logger_output_failed: AtomicU64::new(0),
lifecycle: AtomicU64::new(LifecycleState::Starting as u64),
health: AtomicU64::new(Health::Healthy as u64),
ready: AtomicU64::new(0),
}
}
}
impl Metrics {
pub(crate) fn stage_finished(&self, stage: Stage, outcome: StageOutcome, elapsed_ms: u64) {
let bucket = LATENCY_BUCKET_UPPER_MS
.iter()
.position(|upper| elapsed_ms <= *upper)
.unwrap_or(LATENCY_BUCKET_UPPER_MS.len() - 1);
increment(&self.stage_latency[stage as usize][bucket], 1);
if stage == Stage::Response {
increment(&self.requests[outcome as usize], 1);
}
}
pub(crate) fn capacity(&self, dimension: Option<CapacityDimension>, used: u64, rejected: bool) {
let index = dimension.map_or(4, |value| value as usize);
let counters = if rejected {
&self.capacity_rejected
} else {
&self.capacity_accepted
};
increment(&counters[index], 1);
self.capacity_used[index].store(used, Ordering::Relaxed);
}
pub(crate) fn database(&self, disposition: DatabaseDisposition) {
increment(&self.database[disposition as usize], 1);
}
pub(crate) fn outbound(&self, result: OutboundResult) {
increment(&self.outbound[result as usize], 1);
}
pub(crate) fn logger_output(&self, output_failed: bool) {
if output_failed {
increment(&self.logger_output_failed, 1);
}
}
pub(crate) fn logger_dropped(&self, count: u64) {
increment(&self.logger_dropped, count);
}
pub(crate) fn lifecycle(&self, state: LifecycleState, health: Health) {
self.lifecycle.store(state as u64, Ordering::Relaxed);
self.health.store(health as u64, Ordering::Relaxed);
self.ready.store(
u64::from(state == LifecycleState::Running && health == Health::Healthy),
Ordering::Release,
);
}
pub(crate) fn snapshot(&self) -> MetricsSnapshot {
MetricsSnapshot {
requests: load_array(&self.requests),
stage_latency: std::array::from_fn(|stage| load_array(&self.stage_latency[stage])),
capacity_accepted: load_array(&self.capacity_accepted),
capacity_rejected: load_array(&self.capacity_rejected),
capacity_used: load_array(&self.capacity_used),
database: load_array(&self.database),
outbound: load_array(&self.outbound),
logger_dropped: self.logger_dropped.load(Ordering::Relaxed),
logger_output_failed: self.logger_output_failed.load(Ordering::Relaxed),
lifecycle: self.lifecycle.load(Ordering::Relaxed),
health: self.health.load(Ordering::Relaxed),
ready: self.ready.load(Ordering::Acquire) != 0,
}
}
}
fn load_array<const N: usize>(values: &[AtomicU64; N]) -> [u64; N] {
std::array::from_fn(|index| values[index].load(Ordering::Relaxed))
}
fn increment(value: &AtomicU64, count: u64) {
let _ = value.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
Some(current.saturating_add(count))
});
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct MetricsSnapshot {
requests: [u64; 4],
stage_latency: [[u64; 11]; 7],
capacity_accepted: [u64; 5],
capacity_rejected: [u64; 5],
capacity_used: [u64; 5],
database: [u64; 3],
outbound: [u64; 4],
logger_dropped: u64,
logger_output_failed: u64,
lifecycle: u64,
health: u64,
ready: bool,
}
impl MetricsSnapshot {
pub fn requests(&self, outcome: StageOutcome) -> u64 {
self.requests[outcome as usize]
}
pub fn stage_latency(&self, stage: Stage) -> &[u64; 11] {
&self.stage_latency[stage as usize]
}
pub fn capacity_accepted(&self, dimension: Option<CapacityDimension>) -> u64 {
self.capacity_accepted[dimension.map_or(4, |value| value as usize)]
}
pub fn capacity_rejected(&self, dimension: Option<CapacityDimension>) -> u64 {
self.capacity_rejected[dimension.map_or(4, |value| value as usize)]
}
pub fn capacity_used(&self, dimension: Option<CapacityDimension>) -> u64 {
self.capacity_used[dimension.map_or(4, |value| value as usize)]
}
pub fn database(&self, disposition: DatabaseDisposition) -> u64 {
self.database[disposition as usize]
}
pub fn outbound(&self, result: OutboundResult) -> u64 {
self.outbound[result as usize]
}
pub const fn logger_dropped(&self) -> u64 {
self.logger_dropped
}
pub const fn logger_output_failed(&self) -> u64 {
self.logger_output_failed
}
pub fn lifecycle(&self) -> LifecycleState {
decode_lifecycle(self.lifecycle)
}
pub fn health(&self) -> Health {
decode_health(self.health)
}
pub const fn ready(&self) -> bool {
self.ready
}
}
fn decode_lifecycle(value: u64) -> LifecycleState {
match value {
1 => LifecycleState::Running,
2 => LifecycleState::Draining,
3 => LifecycleState::Stopped,
_ => LifecycleState::Starting,
}
}
fn decode_health(value: u64) -> Health {
match value {
1 => Health::Degraded,
2 => Health::Failed,
_ => Health::Healthy,
}
}
#[cfg(test)]
mod tests {
use std::{mem::size_of, sync::Arc, thread};
use super::*;
#[test]
fn concurrent_updates_are_non_blocking_and_exact() {
let metrics = Arc::new(Metrics::default());
let mut workers = Vec::new();
for _ in 0..8 {
let metrics = metrics.clone();
workers.push(thread::spawn(move || {
for _ in 0..10_000 {
metrics.stage_finished(Stage::Response, StageOutcome::Success, 7);
}
}));
}
for worker in workers {
worker.join().unwrap();
}
let snapshot = metrics.snapshot();
assert_eq!(snapshot.requests(StageOutcome::Success), 80_000);
assert_eq!(snapshot.stage_latency(Stage::Response)[2], 80_000);
}
#[test]
fn storage_and_snapshot_have_compile_time_fixed_size() {
assert!(size_of::<Metrics>() <= 1_024);
assert!(size_of::<MetricsSnapshot>() <= 1_024);
assert_eq!(LATENCY_BUCKET_UPPER_MS.last(), Some(&u64::MAX));
}
}