From 732eb9562324d663e5f61349360ed0243b659f77 Mon Sep 17 00:00:00 2001 From: Paul Mairo Date: Wed, 30 Sep 2026 10:09:14 +0200 Subject: [PATCH] halide-cache-server: add monitoring DISCLAIMER: Completely vibe-coded feel free to ignore. I didn't look at the code myself and I don't expect anyone to. I judge this one by output - /v1/history: per-minute counters and store size for 24 hours - /v1/clients: per-address counters, with the hostname the client now sends in X-Halide-Cache-Client - eviction ages in /v1/stats: a histogram of how long evicted entries had gone unused, the signal that --size is too small - /metrics in Prometheus format - a self-contained dashboard at / showing all of the above Totals, per-minute buckets and per-client rows share one Counters type behind the existing lock. --- Cargo.lock | 45 ++- halide-cache-server/Cargo.toml | 3 +- halide-cache-server/src/blobs.rs | 24 +- halide-cache-server/src/dashboard.html | 400 +++++++++++++++++++++++++ halide-cache-server/src/eviction.rs | 15 +- halide-cache-server/src/extract.rs | 32 +- halide-cache-server/src/main.rs | 19 +- halide-cache-server/src/metrics.rs | 310 ++++++++++++++++++- halide-cache-server/src/stats.rs | 256 +++++++++++++++- halide-cache-server/src/system.rs | 108 +++++++ halide-cache/Cargo.toml | 1 + halide-cache/src/remote.rs | 11 +- lager/src/lru.rs | 17 ++ 13 files changed, 1209 insertions(+), 32 deletions(-) create mode 100644 halide-cache-server/src/dashboard.html create mode 100644 halide-cache-server/src/system.rs diff --git a/Cargo.lock b/Cargo.lock index fabdd20..cccbd1f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -93,6 +93,7 @@ checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", "bytes", + "form_urlencoded", "futures-util", "http", "http-body", @@ -108,6 +109,7 @@ dependencies = [ "serde_core", "serde_json", "serde_path_to_error", + "serde_urlencoded", "sync_wrapper", "tokio", "tower", @@ -303,6 +305,15 @@ version = "0.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + [[package]] name = "fuchsia-cprng" version = "0.1.1" @@ -348,6 +359,16 @@ dependencies = [ "slab", ] +[[package]] +name = "gethostname" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bd49230192a3797a9a4d6abe9b3eed6f7fa4c8a8a4947977c6f80025f92cbd8" +dependencies = [ + "rustix", + "windows-link", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -379,6 +400,7 @@ dependencies = [ "blake3", "clap", "dirs", + "gethostname", "lager", "named-lock", "tar", @@ -400,6 +422,7 @@ dependencies = [ "clap", "futures-util", "lager", + "rustix", "serde", "tokio", "tokio-util", @@ -815,9 +838,9 @@ dependencies = [ [[package]] name = "rustix" -version = "1.1.4" +version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190" +checksum = "891efababe418670775f199f0d233d84843c227a0949a883ce15b37c78d6629d" dependencies = [ "bitflags", "errno", @@ -861,6 +884,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "same-file" version = "1.0.6" @@ -930,6 +959,18 @@ dependencies = [ "serde_core", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "sharded-slab" version = "0.1.7" diff --git a/halide-cache-server/Cargo.toml b/halide-cache-server/Cargo.toml index 882a4b9..9f4cebd 100644 --- a/halide-cache-server/Cargo.toml +++ b/halide-cache-server/Cargo.toml @@ -7,7 +7,7 @@ description = "HTTP server sharing halide-cache entries between machines" [dependencies] lager = { path = "../lager" } anyhow = "1.0.102" -axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio", "json"] } +axum = { version = "0.8.9", default-features = false, features = ["http1", "tokio", "json", "query"] } bytesize = "2.7.0" clap = { version = "4.5.53", features = ["derive"] } futures-util = { version = "0.3", default-features = false } @@ -15,4 +15,5 @@ serde = { version = "1", features = ["derive"] } tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "net", "fs", "signal", "time"] } tokio-util = { version = "0.7", features = ["io", "io-util"] } tracing = "0.1" +rustix = { version = "1.1.5", features = ["fs"] } tracing-subscriber = { version = "0.3.23", features = ["env-filter"] } diff --git a/halide-cache-server/src/blobs.rs b/halide-cache-server/src/blobs.rs index 627c610..905024f 100644 --- a/halide-cache-server/src/blobs.rs +++ b/halide-cache-server/src/blobs.rs @@ -1,6 +1,6 @@ use crate::AppState; -use crate::extract::BlobAddress; -use crate::metrics::Event; +use crate::extract::{BlobAddress, ClientId}; +use crate::metrics::{Event, Request as Timed}; use axum::{ body::Body, extract::{Request, State}, @@ -10,6 +10,7 @@ use axum::{ use futures_util::TryStreamExt; use lager::Address; use std::sync::Arc; +use std::time::Instant; use tokio_util::io::{ReaderStream, StreamReader, SyncIoBridge}; use tracing::{error, info, warn}; @@ -51,10 +52,16 @@ fn blob_headers(len: u64) -> HeaderMap { pub async fn get( State(state): State>, BlobAddress(address): BlobAddress, + ClientId(client): ClientId, ) -> Response { - match open_blob(&state, address).await { + let started = Instant::now(); + let found = open_blob(&state, address).await; + if found.is_ok() { + state.metrics.record_latency(Timed::Get, started.elapsed()); + } + match found { Ok(Some(blob)) => { - state.metrics.record(Event::Hit, blob.len); + state.metrics.record(Event::Hit, &client, blob.len); info!(%address, len = blob.len, "hit"); let stream = ReaderStream::new(tokio::fs::File::from_std(blob.file)); ( @@ -65,7 +72,7 @@ pub async fn get( .into_response() } Ok(None) => { - state.metrics.record(Event::Miss, 0); + state.metrics.record(Event::Miss, &client, 0); info!(%address, "miss"); StatusCode::NOT_FOUND.into_response() } @@ -76,8 +83,10 @@ pub async fn get( pub async fn put( State(state): State>, BlobAddress(address): BlobAddress, + ClientId(client): ClientId, request: Request, ) -> Response { + let started = Instant::now(); let body = request .into_body() .into_data_stream() @@ -100,11 +109,13 @@ pub async fn put( } else { (Event::Duplicate, StatusCode::OK, "replaced existing") }; - state.metrics.record(event, len); + state.metrics.record(event, &client, len); + state.metrics.record_latency(Timed::Put, started.elapsed()); info!(%address, len, what); status.into_response() } Ok(Err(lager::Error::Io(e))) if e.kind() == std::io::ErrorKind::FileTooLarge => { + state.metrics.record(Event::Rejected, &client, 0); warn!(%address, "upload rejected: larger than --max-blob-size"); ( StatusCode::PAYLOAD_TOO_LARGE, @@ -113,6 +124,7 @@ pub async fn put( .into_response() } Ok(Err(e)) => { + state.metrics.record(Event::Failed, &client, 0); warn!(%address, err = %e, "upload failed"); (StatusCode::BAD_REQUEST, "upload failed\n").into_response() } diff --git a/halide-cache-server/src/dashboard.html b/halide-cache-server/src/dashboard.html new file mode 100644 index 0000000..4033d52 --- /dev/null +++ b/halide-cache-server/src/dashboard.html @@ -0,0 +1,400 @@ + + + + + +halide-cache-server + + + +
+

halide-cache-server

+ loading… +
+ + + +
+
+ +
+ +
+
+
+
+
+
+
+
+
+ +
+

Clients

+

One row per source address seen since the server started. The hostname is what the client reported; behind a reverse proxy the first X-Forwarded-For entry is used as the address.

+
+ +
ClientHitsMissesHit rateUploadsServedReceivedFirst seenLast seen
+
+ + + + diff --git a/halide-cache-server/src/eviction.rs b/halide-cache-server/src/eviction.rs index 5d1116d..3d2e5fc 100644 --- a/halide-cache-server/src/eviction.rs +++ b/halide-cache-server/src/eviction.rs @@ -27,13 +27,22 @@ async fn evict(state: &AppState) { } else { Vec::new() }; - Ok::<_, lager::Error>((before, evicted, lru.lager_size(), lru.entries())) + Ok::<_, lager::Error>(( + before, + evicted, + lru.lager_size(), + lru.entries(), + lru.largest(), + lru.oldest(), + )) }) .await; match result { - Ok(Ok((before, evicted, size, entries))) => { - state.metrics.record_scan(size, entries, &evicted); + Ok(Ok((before, evicted, size, entries, largest, oldest))) => { + state + .metrics + .record_scan(size, entries, largest, oldest, &evicted); info!( size = %ByteSize::b(size), entries, diff --git a/halide-cache-server/src/extract.rs b/halide-cache-server/src/extract.rs index 7932b01..f80a015 100644 --- a/halide-cache-server/src/extract.rs +++ b/halide-cache-server/src/extract.rs @@ -1,8 +1,12 @@ +use crate::metrics::Client; use axum::{ - extract::{FromRequestParts, Path}, + extract::{ConnectInfo, FromRequestParts, Path}, http::{StatusCode, request::Parts}, }; use lager::Address; +use std::net::{IpAddr, SocketAddr}; + +const CLIENT_HEADER: &str = "x-halide-cache-client"; pub struct BlobAddress(pub Address); @@ -18,3 +22,29 @@ impl FromRequestParts for BlobAddress { .map_err(|_| StatusCode::BAD_REQUEST) } } + +pub struct ClientId(pub Client); + +impl FromRequestParts for ClientId { + type Rejection = StatusCode; + + async fn from_request_parts(parts: &mut Parts, state: &S) -> Result { + let ConnectInfo(peer) = ConnectInfo::::from_request_parts(parts, state) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + let ip: IpAddr = parts + .headers + .get("x-forwarded-for") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.split(',').next()) + .and_then(|v| v.trim().parse().ok()) + .unwrap_or(peer.ip()); + let hostname = parts + .headers + .get(CLIENT_HEADER) + .and_then(|v| v.to_str().ok()) + .map(|v| v.trim().chars().take(128).collect::()) + .filter(|v| !v.is_empty()); + Ok(ClientId(Client { ip, hostname })) + } +} diff --git a/halide-cache-server/src/main.rs b/halide-cache-server/src/main.rs index e9fa2ae..087ea3d 100644 --- a/halide-cache-server/src/main.rs +++ b/halide-cache-server/src/main.rs @@ -3,6 +3,7 @@ mod eviction; mod extract; mod metrics; mod stats; +mod system; use axum::{Router, routing::get}; use bytesize::ByteSize; @@ -62,14 +63,23 @@ async fn main() -> anyhow::Result<()> { started: Instant::now(), }); + tokio::spawn(system::sample_loop( + state.clone(), + args.data_dir.clone(), + Duration::from_secs(5), + )); tokio::spawn(eviction::eviction_loop( state.clone(), Duration::from_secs(args.evict_interval), )); let app = Router::new() + .route("/", get(stats::dashboard)) .route("/v1/blobs/{address}", get(blobs::get).put(blobs::put)) .route("/v1/stats", get(stats::stats)) + .route("/v1/history", get(stats::history)) + .route("/v1/clients", get(stats::clients)) + .route("/metrics", get(stats::prometheus)) .route("/healthz", get(|| async { "ok" })) .with_state(state); @@ -80,9 +90,12 @@ async fn main() -> anyhow::Result<()> { size = %args.size, "halide-cache-server listening" ); - axum::serve(listener, app) - .with_graceful_shutdown(shutdown_signal()) - .await?; + axum::serve( + listener, + app.into_make_service_with_connect_info::(), + ) + .with_graceful_shutdown(shutdown_signal()) + .await?; Ok(()) } diff --git a/halide-cache-server/src/metrics.rs b/halide-cache-server/src/metrics.rs index 0d6fc3a..576f638 100644 --- a/halide-cache-server/src/metrics.rs +++ b/halide-cache-server/src/metrics.rs @@ -1,6 +1,15 @@ +use crate::system::Host; use lager::Evicted; use serde::Serialize; +use std::collections::{HashMap, VecDeque}; +use std::net::IpAddr; use std::sync::Mutex; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +pub const BUCKET_SECONDS: u64 = 60; +const MAX_BUCKETS: usize = 24 * 60; +const MAX_CLIENTS: usize = 4096; +pub const AGE_BOUNDS: [u64; 6] = [3600, 6 * 3600, 24 * 3600, 3 * 86400, 7 * 86400, u64::MAX]; #[derive(Clone, Copy, Debug)] pub enum Event { @@ -8,14 +17,26 @@ pub enum Event { Miss, Upload, Duplicate, + Rejected, + Failed, } +#[derive(Clone, Copy, Debug)] +pub enum Request { + Get, + Put, +} + +const LATENCY_SAMPLES: usize = 1000; + #[derive(Default, Clone, Copy, Serialize)] pub struct Counters { pub hits: u64, pub misses: u64, pub uploads: u64, pub duplicates: u64, + pub rejected_uploads: u64, + pub failed_uploads: u64, pub bytes_served: u64, pub bytes_received: u64, } @@ -36,6 +57,8 @@ impl Counters { self.duplicates += 1; self.bytes_received += bytes; } + Event::Rejected => self.rejected_uploads += 1, + Event::Failed => self.failed_uploads += 1, } } } @@ -44,34 +67,213 @@ impl Counters { pub struct Store { pub size_bytes: u64, pub entries: usize, + pub largest_entry_bytes: u64, + pub oldest_entry_used: Option, pub evictions: u64, + pub evicted_bytes: u64, + pub eviction_ages: [u64; 6], +} + +#[derive(Default, Clone, Serialize)] +pub struct Bucket { + pub ts: u64, + #[serde(flatten)] + pub counters: Counters, + pub size_bytes: Option, + pub cpu_percent: Option, + pub memory_used_bytes: Option, +} + +#[derive(Clone, Copy, Serialize)] +pub struct LatencySummary { + pub median_ms: f64, + pub p95_ms: f64, + pub samples: usize, +} + +#[derive(Clone, Copy, Serialize)] +pub struct Latencies { + pub get: Option, + pub put: Option, +} + +#[derive(Default, Clone, Serialize)] +pub struct ClientStats { + pub ip: String, + pub hostname: Option, + #[serde(flatten)] + pub counters: Counters, + pub first_seen: u64, + pub last_seen: u64, +} + +pub struct Client { + pub ip: IpAddr, + pub hostname: Option, } #[derive(Default)] struct Inner { totals: Counters, store: Store, + host: Host, + get_latency: VecDeque, + put_latency: VecDeque, + history: VecDeque, + clients: HashMap, } #[derive(Default)] pub struct Metrics(Mutex); +pub fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + impl Metrics { - pub fn record(&self, event: Event, bytes: u64) { - self.0.lock().unwrap().totals.add(event, bytes); + pub fn record(&self, event: Event, client: &Client, bytes: u64) { + let ts = now(); + let mut m = self.0.lock().unwrap(); + m.totals.add(event, bytes); + current_bucket(&mut m.history, ts) + .counters + .add(event, bytes); + + if m.clients.len() >= MAX_CLIENTS + && !m.clients.contains_key(&client.ip) + && let Some(oldest) = m + .clients + .iter() + .min_by_key(|(_, c)| c.last_seen) + .map(|(ip, _)| *ip) + { + m.clients.remove(&oldest); + } + let c = m.clients.entry(client.ip).or_insert_with(|| ClientStats { + ip: client.ip.to_string(), + first_seen: ts, + ..Default::default() + }); + c.last_seen = ts; + if client.hostname.is_some() { + c.hostname = client.hostname.clone(); + } + c.counters.add(event, bytes); } - pub fn record_scan(&self, size: u64, entries: usize, evicted: &[Evicted]) { + pub fn record_scan( + &self, + size: u64, + entries: usize, + largest: u64, + oldest: Option, + evicted: &[Evicted], + ) { + let when = SystemTime::now(); let mut m = self.0.lock().unwrap(); m.store.size_bytes = size; m.store.entries = entries; - m.store.evictions += evicted.len() as u64; + m.store.largest_entry_bytes = largest; + m.store.oldest_entry_used = oldest + .and_then(|t| t.duration_since(UNIX_EPOCH).ok()) + .map(|d| d.as_secs()); + for e in evicted { + let age = when + .duration_since(e.last_used) + .map(|d| d.as_secs()) + .unwrap_or(0); + let bucket = AGE_BOUNDS.iter().position(|&b| age < b).unwrap_or(5); + m.store.eviction_ages[bucket] += 1; + m.store.evictions += 1; + m.store.evicted_bytes += e.size; + } + current_bucket(&mut m.history, now()).size_bytes = Some(size); } pub fn totals(&self) -> (Counters, Store) { let m = self.0.lock().unwrap(); (m.totals, m.store) } + + pub fn record_host(&self, host: Host) { + let mut m = self.0.lock().unwrap(); + m.host = host; + let bucket = current_bucket(&mut m.history, now()); + bucket.cpu_percent = host.cpu_percent; + bucket.memory_used_bytes = Some(host.memory_total_bytes - host.memory_available_bytes); + } + + pub fn record_latency(&self, request: Request, elapsed: Duration) { + let mut m = self.0.lock().unwrap(); + let samples = match request { + Request::Get => &mut m.get_latency, + Request::Put => &mut m.put_latency, + }; + if samples.len() == LATENCY_SAMPLES { + samples.pop_front(); + } + samples.push_back(elapsed); + } + + pub fn latencies(&self) -> Latencies { + let m = self.0.lock().unwrap(); + Latencies { + get: summarize(&m.get_latency), + put: summarize(&m.put_latency), + } + } + + pub fn host(&self) -> Host { + self.0.lock().unwrap().host + } + + pub fn history(&self, seconds: u64) -> Vec { + let since = now().saturating_sub(seconds); + let m = self.0.lock().unwrap(); + m.history + .iter() + .filter(|b| b.ts + BUCKET_SECONDS > since) + .cloned() + .collect() + } + + pub fn clients(&self) -> Vec { + let mut v: Vec = self.0.lock().unwrap().clients.values().cloned().collect(); + v.sort_by(|a, b| b.last_seen.cmp(&a.last_seen).then_with(|| a.ip.cmp(&b.ip))); + v + } +} + +fn summarize(samples: &VecDeque) -> Option { + if samples.is_empty() { + return None; + } + let mut sorted: Vec = samples.iter().copied().collect(); + sorted.sort_unstable(); + let at = + |q: f64| sorted[((sorted.len() - 1) as f64 * q).round() as usize].as_secs_f64() * 1000.0; + Some(LatencySummary { + median_ms: at(0.5), + p95_ms: at(0.95), + samples: sorted.len(), + }) +} + +fn current_bucket(history: &mut VecDeque, ts: u64) -> &mut Bucket { + let start = ts - ts % BUCKET_SECONDS; + if history.back().is_none_or(|b| b.ts != start) { + history.push_back(Bucket { + ts: start, + ..Default::default() + }); + while history.len() > MAX_BUCKETS { + history.pop_front(); + } + } + history.back_mut().unwrap() } #[cfg(test)] @@ -79,20 +281,102 @@ mod tests { use super::*; #[test] - fn records_totals_and_scans() { + fn records_totals_buckets_and_clients() { let m = Metrics::default(); - m.record(Event::Hit, 100); - m.record(Event::Miss, 0); - m.record(Event::Upload, 50); - m.record(Event::Duplicate, 50); - m.record_scan(10, 1, &[]); + let ip: IpAddr = "10.0.0.7".parse().unwrap(); + let anon = Client { ip, hostname: None }; + let named = Client { + ip, + hostname: Some("build-07".into()), + }; + m.record(Event::Hit, &anon, 100); + m.record(Event::Miss, &named, 0); + m.record(Event::Upload, &anon, 50); + m.record(Event::Duplicate, &anon, 50); - let (t, store) = m.totals(); + let (t, _) = m.totals(); assert_eq!((t.hits, t.misses, t.uploads, t.duplicates), (1, 1, 1, 1)); assert_eq!((t.bytes_served, t.bytes_received), (100, 100)); + + let h = m.history(3600); + assert_eq!(h.len(), 1); + assert_eq!((h[0].counters.hits, h[0].counters.uploads), (1, 1)); + + let c = m.clients(); + assert_eq!(c.len(), 1); + assert_eq!(c[0].ip, "10.0.0.7"); assert_eq!( - (store.size_bytes, store.entries, store.evictions), - (10, 1, 0) + c[0].hostname.as_deref(), + Some("build-07"), + "the name sticks" ); + assert_eq!(c[0].counters.bytes_received, 100); + } + + #[test] + fn eviction_ages_are_bucketed() { + let m = Metrics::default(); + let evicted = |size, age_secs| Evicted { + address: lager::Address::from([0u8; 64]), + last_used: SystemTime::now() - std::time::Duration::from_secs(age_secs), + size, + }; + m.record_scan( + 10, + 1, + 300, + None, + &[ + evicted(100, 60), + evicted(200, 7200), + evicted(300, 30 * 86400), + ], + ); + let (_, store) = m.totals(); + assert_eq!(store.eviction_ages, [1, 1, 0, 0, 0, 1]); + assert_eq!((store.evictions, store.evicted_bytes), (3, 600)); + assert_eq!((store.size_bytes, store.entries), (10, 1)); + assert_eq!(store.largest_entry_bytes, 300); + } + + #[test] + fn rejected_and_failed_uploads_are_counted() { + let m = Metrics::default(); + let client = Client { + ip: "10.0.0.8".parse().unwrap(), + hostname: None, + }; + m.record(Event::Rejected, &client, 0); + m.record(Event::Failed, &client, 0); + m.record(Event::Failed, &client, 0); + let (t, _) = m.totals(); + assert_eq!((t.rejected_uploads, t.failed_uploads, t.uploads), (1, 2, 0)); + assert_eq!(m.clients()[0].counters.failed_uploads, 2); + } + + #[test] + fn latency_keeps_the_last_samples_and_their_percentiles() { + let m = Metrics::default(); + assert!(m.latencies().get.is_none()); + for ms in 1..=(LATENCY_SAMPLES as u64 + 100) { + m.record_latency(Request::Get, Duration::from_millis(ms)); + } + let get = m.latencies().get.unwrap(); + assert_eq!(get.samples, LATENCY_SAMPLES); + assert_eq!(get.median_ms.round(), 601.0); + assert_eq!(get.p95_ms.round(), 1050.0); + assert!(m.latencies().put.is_none()); + } + + #[test] + fn history_is_bounded() { + let mut history = VecDeque::new(); + for i in 0..(MAX_BUCKETS as u64 + 10) { + current_bucket(&mut history, i * BUCKET_SECONDS) + .counters + .hits += 1; + } + assert_eq!(history.len(), MAX_BUCKETS); + assert_eq!(history.front().unwrap().ts, 10 * BUCKET_SECONDS); } } diff --git a/halide-cache-server/src/stats.rs b/halide-cache-server/src/stats.rs index da63e87..ebf624b 100644 --- a/halide-cache-server/src/stats.rs +++ b/halide-cache-server/src/stats.rs @@ -1,8 +1,17 @@ use crate::AppState; -use crate::metrics::{Counters, Store}; -use axum::{extract::State, response::Json}; +use crate::metrics::{self, AGE_BOUNDS, Counters, Latencies, Store}; +use crate::system::Host; +use axum::{ + extract::{Query, State}, + http::header, + response::{Html, IntoResponse, Json, Response}, +}; use std::sync::Arc; +pub async fn dashboard() -> Html<&'static str> { + Html(include_str!("dashboard.html")) +} + #[derive(serde::Serialize)] pub struct Stats { uptime_seconds: u64, @@ -12,6 +21,14 @@ pub struct Stats { totals: Counters, #[serde(flatten)] store: Store, + average_entry_bytes: u64, + eviction_age_bounds_seconds: [u64; 6], + host: Host, + latency: Latencies, +} + +fn average_entry_bytes(store: &Store) -> u64 { + store.size_bytes / (store.entries as u64).max(1) } pub async fn stats(State(state): State>) -> Json { @@ -27,5 +44,240 @@ pub async fn stats(State(state): State>) -> Json { }, totals, store, + average_entry_bytes: average_entry_bytes(&store), + eviction_age_bounds_seconds: AGE_BOUNDS, + host: state.metrics.host(), + latency: state.metrics.latencies(), + }) +} + +#[derive(serde::Deserialize)] +pub struct HistoryQuery { + #[serde(default = "default_window")] + seconds: u64, +} + +fn default_window() -> u64 { + 3600 +} + +#[derive(serde::Serialize)] +pub struct History { + bucket_seconds: u64, + now: u64, + buckets: Vec, +} + +pub async fn history( + State(state): State>, + Query(q): Query, +) -> Json { + Json(History { + bucket_seconds: metrics::BUCKET_SECONDS, + now: metrics::now(), + buckets: state.metrics.history(q.seconds.min(24 * 3600)), }) } + +pub async fn clients(State(state): State>) -> Json> { + Json(state.metrics.clients()) +} + +pub async fn prometheus(State(state): State>) -> Response { + let (t, store) = state.metrics.totals(); + + let gauges = [ + ( + "capacity_bytes", + "Configured maximum cache size", + state.max_size, + ), + ( + "size_bytes", + "Cache size at the last eviction pass", + store.size_bytes, + ), + ( + "entries", + "Entries at the last eviction pass", + store.entries as u64, + ), + ( + "largest_entry_bytes", + "Largest entry at the last eviction pass", + store.largest_entry_bytes, + ), + ( + "average_entry_bytes", + "Average entry size at the last eviction pass", + average_entry_bytes(&store), + ), + ( + "uptime_seconds", + "Seconds since start", + state.started.elapsed().as_secs(), + ), + ]; + let counters = [ + ("hits_total", "Downloads that found an entry", t.hits), + ("misses_total", "Downloads that found nothing", t.misses), + ("uploads_total", "New entries uploaded", t.uploads), + ( + "duplicate_uploads_total", + "Uploads replacing an existing entry", + t.duplicates, + ), + ( + "evictions_total", + "Entries removed by the LRU", + store.evictions, + ), + ( + "evicted_bytes_total", + "Bytes removed by the LRU", + store.evicted_bytes, + ), + ( + "served_bytes_total", + "Bytes sent to clients", + t.bytes_served, + ), + ( + "received_bytes_total", + "Bytes received from clients", + t.bytes_received, + ), + ( + "rejected_uploads_total", + "Uploads rejected for exceeding --max-blob-size", + t.rejected_uploads, + ), + ( + "failed_uploads_total", + "Uploads that failed, typically a dropped connection", + t.failed_uploads, + ), + ]; + + let mut out = String::new(); + for (kind, metrics) in [("gauge", &gauges[..]), ("counter", &counters[..])] { + for (name, help, value) in metrics { + out.push_str(&format!( + "# HELP halide_cache_{name} {help}\n# TYPE halide_cache_{name} {kind}\nhalide_cache_{name} {value}\n" + )); + } + } + + let host = state.metrics.host(); + let mut host_gauges = vec![ + ("host_cpus", "Logical CPUs of the host", host.cpus as f64), + ( + "host_load1", + "Host load average over 1 minute", + host.load[0], + ), + ( + "host_load5", + "Host load average over 5 minutes", + host.load[1], + ), + ( + "host_load15", + "Host load average over 15 minutes", + host.load[2], + ), + ( + "host_memory_total_bytes", + "Host memory", + host.memory_total_bytes as f64, + ), + ( + "host_memory_available_bytes", + "Host memory available for new allocations", + host.memory_available_bytes as f64, + ), + ]; + host_gauges.push(( + "disk_total_bytes", + "Size of the filesystem holding the store", + host.disk_total_bytes as f64, + )); + host_gauges.push(( + "disk_available_bytes", + "Free space on the filesystem holding the store", + host.disk_available_bytes as f64, + )); + if let Some(used) = store.oldest_entry_used { + host_gauges.push(( + "oldest_entry_age_seconds", + "Time since the least recently used entry was last used", + metrics::now().saturating_sub(used) as f64, + )); + } + let latency = state.metrics.latencies(); + if let Some(get) = latency.get { + host_gauges.push(( + "get_latency_median_ms", + "Median time to answer a download", + get.median_ms, + )); + host_gauges.push(( + "get_latency_p95_ms", + "95th percentile time to answer a download", + get.p95_ms, + )); + } + if let Some(put) = latency.put { + host_gauges.push(( + "put_latency_median_ms", + "Median time to store an upload", + put.median_ms, + )); + host_gauges.push(( + "put_latency_p95_ms", + "95th percentile time to store an upload", + put.p95_ms, + )); + } + if let Some(cpu) = host.cpu_percent { + host_gauges.push(( + "host_cpu_percent", + "Host CPU usage over the last 5 seconds", + cpu, + )); + } + for (name, help, value) in host_gauges { + out.push_str(&format!( + "# HELP halide_cache_{name} {help}\n# TYPE halide_cache_{name} gauge\nhalide_cache_{name} {value}\n" + )); + } + + out.push_str( + "# HELP halide_cache_eviction_age_seconds Seconds since last use when an entry was evicted\n\ + # TYPE halide_cache_eviction_age_seconds histogram\n", + ); + let mut cumulative = 0; + for (count, bound) in store.eviction_ages.iter().zip(AGE_BOUNDS) { + cumulative += count; + let le = if bound == u64::MAX { + "+Inf".to_string() + } else { + bound.to_string() + }; + out.push_str(&format!( + "halide_cache_eviction_age_seconds_bucket{{le=\"{le}\"}} {cumulative}\n" + )); + } + out.push_str(&format!( + "halide_cache_eviction_age_seconds_count {cumulative}\n" + )); + + ( + [( + header::CONTENT_TYPE, + "text/plain; version=0.0.4; charset=utf-8", + )], + out, + ) + .into_response() +} diff --git a/halide-cache-server/src/system.rs b/halide-cache-server/src/system.rs new file mode 100644 index 0000000..4e64971 --- /dev/null +++ b/halide-cache-server/src/system.rs @@ -0,0 +1,108 @@ +use crate::AppState; +use serde::Serialize; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +#[derive(Default, Clone, Copy, Serialize)] +pub struct Host { + pub cpus: usize, + pub cpu_percent: Option, + pub load: [f64; 3], + pub memory_total_bytes: u64, + pub memory_available_bytes: u64, + pub disk_total_bytes: u64, + pub disk_available_bytes: u64, +} + +pub async fn sample_loop(state: Arc, data_dir: PathBuf, interval: Duration) { + let cpus = std::thread::available_parallelism().map_or(0, |n| n.get()); + let mut previous = cpu_times(); + loop { + tokio::time::sleep(interval).await; + let current = cpu_times(); + let (memory_total_bytes, memory_available_bytes) = memory().unwrap_or_default(); + let (disk_total_bytes, disk_available_bytes) = disk(&data_dir).unwrap_or_default(); + state.metrics.record_host(Host { + cpus, + cpu_percent: previous.zip(current).and_then(cpu_percent), + load: load().unwrap_or_default(), + memory_total_bytes, + memory_available_bytes, + disk_total_bytes, + disk_available_bytes, + }); + previous = current; + } +} + +fn cpu_times() -> Option<(u64, u64)> { + let stat = std::fs::read_to_string("/proc/stat").ok()?; + let fields: Vec = stat + .lines() + .next()? + .split_whitespace() + .skip(1) + .take(8) + .filter_map(|f| f.parse().ok()) + .collect(); + if fields.len() < 5 { + return None; + } + let idle = fields[3] + fields[4]; + Some((idle, fields.iter().sum())) +} + +fn cpu_percent(((idle0, total0), (idle1, total1)): ((u64, u64), (u64, u64))) -> Option { + let total = total1.checked_sub(total0).filter(|&t| t > 0)?; + let idle = idle1.saturating_sub(idle0).min(total); + Some(100.0 * (total - idle) as f64 / total as f64) +} + +fn load() -> Option<[f64; 3]> { + let loadavg = std::fs::read_to_string("/proc/loadavg").ok()?; + let mut values = loadavg.split_whitespace().map(|v| v.parse().ok()); + Some([values.next()??, values.next()??, values.next()??]) +} + +fn memory() -> Option<(u64, u64)> { + let meminfo = std::fs::read_to_string("/proc/meminfo").ok()?; + let kib = |name: &str| { + meminfo.lines().find_map(|line| { + line.strip_prefix(name)? + .trim() + .strip_suffix("kB")? + .trim() + .parse::() + .ok() + }) + }; + Some((kib("MemTotal:")? * 1024, kib("MemAvailable:")? * 1024)) +} + +fn disk(path: &Path) -> Option<(u64, u64)> { + let fs = rustix::fs::statvfs(path).ok()?; + Some((fs.f_blocks * fs.f_frsize, fs.f_bavail * fs.f_frsize)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cpu_percent_is_the_busy_share_of_elapsed_time() { + assert_eq!(cpu_percent(((100, 1000), (150, 1100))), Some(50.0)); + assert_eq!(cpu_percent(((100, 1000), (200, 1100))), Some(0.0)); + assert_eq!(cpu_percent(((100, 1000), (100, 1000))), None); + } + + #[test] + fn reads_the_host() { + let (total, available) = memory().unwrap(); + assert!(total > 0 && available <= total); + assert!(load().is_some()); + assert!(cpu_times().is_some()); + let (disk_total, disk_available) = disk(Path::new("/")).unwrap(); + assert!(disk_total > 0 && disk_available <= disk_total); + } +} diff --git a/halide-cache/Cargo.toml b/halide-cache/Cargo.toml index af23560..af51ae8 100644 --- a/halide-cache/Cargo.toml +++ b/halide-cache/Cargo.toml @@ -14,6 +14,7 @@ anyhow = "1.0.102" thiserror = "2.0.17" dirs = "6.0.0" ureq = { version = "3.4.2", default-features = false, features = ["rustls"] } +gethostname = "1.1.0" which = "8.0.6" walkdir = "2.5.0" diff --git a/halide-cache/src/remote.rs b/halide-cache/src/remote.rs index 9d6a158..190b59b 100644 --- a/halide-cache/src/remote.rs +++ b/halide-cache/src/remote.rs @@ -1,9 +1,12 @@ use lager::{Address, Lager}; use std::time::Duration; +const CLIENT_HEADER: &str = "x-halide-cache-client"; + pub struct Remote { agent: ureq::Agent, base_url: String, + hostname: String, } #[derive(Debug, thiserror::Error)] @@ -29,6 +32,7 @@ impl Remote { Remote { agent, base_url: base_url.trim_end_matches('/').to_owned(), + hostname: gethostname::gethostname().to_string_lossy().into_owned(), } } @@ -37,7 +41,11 @@ impl Remote { } pub fn fetch_into(&self, address: &Address, lager: &Lager) -> Result { - let mut response = self.agent.get(self.url(address)).call()?; + let mut response = self + .agent + .get(self.url(address)) + .header(CLIENT_HEADER, &self.hostname) + .call()?; match response.status().as_u16() { 200 => {} 404 => return Ok(false), @@ -55,6 +63,7 @@ impl Remote { let response = self .agent .put(self.url(address)) + .header(CLIENT_HEADER, &self.hostname) .header("content-length", len.to_string()) .send(&file)?; match response.status().as_u16() { diff --git a/lager/src/lru.rs b/lager/src/lru.rs index 3ac5627..954012a 100644 --- a/lager/src/lru.rs +++ b/lager/src/lru.rs @@ -124,6 +124,14 @@ impl LRU { pub fn entries(&self) -> usize { self.heap.len() } + + pub fn largest(&self) -> u64 { + self.heap.iter().map(|item| item.size).max().unwrap_or(0) + } + + pub fn oldest(&self) -> Option { + self.heap.peek().map(|item| item.modified) + } } #[cfg(test)] @@ -162,6 +170,15 @@ mod tests { let evicted: Vec<_> = evicted.iter().map(|e| e.address).collect(); assert_eq!(evicted, vec![new], "only the untouched entry is removed"); assert_eq!(lru.entries(), 1, "the skipped entry is still counted"); + assert_eq!( + lru.largest(), + lru.lager_size(), + "the skipped entry is still the largest" + ); + assert!( + lru.oldest().is_some(), + "the skipped entry is still the oldest" + ); assert!(lager.retrieve(&old).is_ok(), "the touched entry survives"); }