From f52ae6973299fa3c1d81758838951e39904e1d84 Mon Sep 17 00:00:00 2001 From: tv Date: Sun, 25 Oct 2026 13:38:59 +0100 Subject: add files module to consistently track stats --- src/api/stats.rs | 30 +++++--- src/api/stats/prometheus.rs | 53 +++++++------- src/files.rs | 174 ++++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 27 +++---- src/state.rs | 2 + src/swarm/bootstrap.rs | 12 +-- src/swarm/state_file.rs | 86 +++++++--------------- src/swarm/stats.rs | 27 ------- 8 files changed, 268 insertions(+), 143 deletions(-) create mode 100644 src/files.rs (limited to 'src') diff --git a/src/api/stats.rs b/src/api/stats.rs index 6c5e172..8899421 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -9,6 +9,8 @@ use axum::{ http::{HeaderMap, header}, response::{IntoResponse, Response}, }; +use std::collections::BTreeMap; + use serde::Serialize; use super::peers::EpochInfo; @@ -35,11 +37,11 @@ pub(crate) struct StatsResponse { /// The peers listed in /peers together (a forgotten peer's are gone). incidents: Incidents, propagation: PropagationStats, - disk: DiskStats, + /// This node's own files (`files`), by kind. + disk: BTreeMap<&'static str, DiskStats>, } -/// Files this node reads and writes for itself (the state file, a backup to recover -/// from), since it started. +/// Reads and writes of one kind of file, since the node started. #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] struct DiskStats { @@ -47,6 +49,7 @@ struct DiskStats { read_bytes: u64, writes: u64, written_bytes: u64, + errors: u64, } /// How long changes written elsewhere took from their commit there until applied here, @@ -207,11 +210,20 @@ async fn collect(state: &AppState) -> StatsResponse { .map(|(le_ms, count)| PropagationBucket { le_ms, count }) .collect(), }, - disk: DiskStats { - reads: node.disk.reads, - read_bytes: node.disk.read_bytes, - writes: node.disk.writes, - written_bytes: node.disk.written_bytes, - }, + disk: state + .files + .counts() + .into_iter() + .map(|(kind, counts)| { + let stats = DiskStats { + reads: counts.reads, + read_bytes: counts.read_bytes, + writes: counts.writes, + written_bytes: counts.written_bytes, + errors: counts.errors, + }; + (kind.as_str(), stats) + }) + .collect(), } } diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index ea97bd2..e267042 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -297,31 +297,34 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { stats.stale_hello_replies as f64, ); - let disk = &stats.disk; - out.single( - "disk_reads_total", - "counter", - "Files this node read for itself (state file, backup).", - disk.reads as f64, - ); - out.single( - "disk_read_bytes_total", - "counter", - "Bytes of those files.", - disk.read_bytes as f64, - ); - out.single( - "disk_writes_total", - "counter", - "Files this node wrote for itself (state file).", - disk.writes as f64, - ); - out.single( - "disk_written_bytes_total", - "counter", - "Bytes of those files.", - disk.written_bytes as f64, - ); + type Field = fn(&super::DiskStats) -> u64; + let disk: [(&str, &str, Field); 5] = [ + ("disk_reads_total", "Reads of this node's own files.", |c| { + c.reads + }), + ("disk_read_bytes_total", "Bytes read from them.", |c| { + c.read_bytes + }), + ( + "disk_writes_total", + "Writes of this node's own files.", + |c| c.writes, + ), + ("disk_written_bytes_total", "Bytes written to them.", |c| { + c.written_bytes + }), + ( + "disk_errors_total", + "Failed reads and writes of them.", + |c| c.errors, + ), + ]; + for (name, help, field) in disk { + out.family(name, "counter", help); + for (kind, counts) in &stats.disk { + out.sample(name, &[("file", *kind)], field(counts) as f64); + } + } let propagation = &stats.propagation; out.family( diff --git a/src/files.rs b/src/files.rs new file mode 100644 index 0000000..81c8b1e --- /dev/null +++ b/src/files.rs @@ -0,0 +1,174 @@ +//! The files this node reads and writes for itself, and how much it did (`/admin/stats`, +//! `disk`): the state file (`swarm::state_file`) and a backup to recover from +//! (`swarm::bootstrap`). Not the Mycelium daemon's files, configuration or logs. +//! +//! Writes replace a file atomically: a temporary file next to it, flushed to disk, then +//! renamed over it, so a crash leaves the old content or the new one. + +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, MutexGuard}; + +/// What a file is, for the counts. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) enum FileKind { + State, + Backup, +} + +impl FileKind { + pub(crate) const ALL: [FileKind; 2] = [FileKind::State, FileKind::Backup]; + + pub(crate) fn as_str(self) -> &'static str { + match self { + FileKind::State => "state", + FileKind::Backup => "backup", + } + } + + fn index(self) -> usize { + self as usize + } +} + +/// Counts for one kind of file, since the node started. +#[derive(Clone, Copy, Debug, Default)] +pub(crate) struct Counts { + pub(crate) reads: u64, + pub(crate) read_bytes: u64, + pub(crate) writes: u64, + pub(crate) written_bytes: u64, + /// Reads or writes that failed (a missing file read is no failure). + pub(crate) errors: u64, +} + +/// Reads and writes this node's files, counting per kind. +#[derive(Default)] +pub(crate) struct Files { + // A std mutex: every access is short and never spans an await. + counts: Mutex<[Counts; FileKind::ALL.len()]>, +} + +impl Files { + fn lock(&self) -> MutexGuard<'_, [Counts; FileKind::ALL.len()]> { + self.counts + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + } + + fn count(&self, kind: FileKind, f: impl FnOnce(&mut Counts)) { + f(&mut self.lock()[kind.index()]); + } + + /// The file's content; None if it doesn't exist. + pub(crate) fn read(&self, kind: FileKind, path: &Path) -> std::io::Result>> { + match std::fs::read(path) { + Ok(bytes) => { + self.count(kind, |counts| { + counts.reads += 1; + counts.read_bytes += bytes.len() as u64; + }); + Ok(Some(bytes)) + } + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(error) => { + self.count(kind, |counts| counts.errors += 1); + Err(error) + } + } + } + + /// Replaces the file's content atomically (see the module docs). + pub(crate) fn write_atomic( + &self, + kind: FileKind, + path: &Path, + bytes: &[u8], + ) -> std::io::Result<()> { + let result = write_atomic(path, bytes); + self.count(kind, |counts| match &result { + Ok(()) => { + counts.writes += 1; + counts.written_bytes += bytes.len() as u64; + } + Err(_) => counts.errors += 1, + }); + result + } + + /// Moves an unusable file aside (``), returning where to. + pub(crate) fn move_aside(&self, path: &Path, suffix: &str) -> std::io::Result { + let aside = with_suffix(path, suffix); + std::fs::rename(path, &aside)?; + Ok(aside) + } + + /// The counts per kind. + pub(crate) fn counts(&self) -> Vec<(FileKind, Counts)> { + let counts = *self.lock(); + FileKind::ALL + .into_iter() + .map(|kind| (kind, counts[kind.index()])) + .collect() + } +} + +fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> { + let temporary = with_suffix(path, ".tmp"); + { + let mut file = std::fs::File::create(&temporary)?; + file.write_all(bytes)?; + file.sync_all()?; + } + std::fs::rename(&temporary, path)?; + sync_parent(path); + Ok(()) +} + +/// Makes a rename in `path`'s directory durable (best effort: not every platform syncs +/// directories). +fn sync_parent(path: &Path) { + if let Some(dir) = path.parent().filter(|dir| !dir.as_os_str().is_empty()) { + if let Ok(dir) = std::fs::File::open(dir) { + let _ = dir.sync_all(); + } + } +} + +fn with_suffix(path: &Path, suffix: &str) -> PathBuf { + let mut name = path.as_os_str().to_owned(); + name.push(suffix); + PathBuf::from(name) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn writes_reads_and_counts() { + let dir = std::env::temp_dir().join(format!("continuum-files-{}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join("state.json"); + let files = Files::default(); + + assert_eq!(files.read(FileKind::State, &path).unwrap(), None); + files.write_atomic(FileKind::State, &path, b"one").unwrap(); + files + .write_atomic(FileKind::State, &path, b"three") + .unwrap(); + assert_eq!( + files.read(FileKind::State, &path).unwrap().as_deref(), + Some(&b"three"[..]) + ); + assert!(!with_suffix(&path, ".tmp").exists()); + + let (kind, state) = files.counts()[0]; + assert_eq!(kind, FileKind::State); + assert_eq!((state.writes, state.written_bytes), (2, 8)); + assert_eq!((state.reads, state.read_bytes, state.errors), (1, 5, 0)); + assert_eq!(files.counts()[1].1.reads, 0); + + std::fs::remove_dir_all(&dir).unwrap(); + } +} diff --git a/src/main.rs b/src/main.rs index 57c26b5..76e5ff0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,6 +18,7 @@ mod api; mod backoff; mod config; mod document; +mod files; mod logging; mod remake; mod state; @@ -86,10 +87,12 @@ async fn main() -> Result<()> { "Using Mycelium REST API" ); + let files = Arc::new(files::Files::default()); let mut seeds: Vec = args.seeds.clone(); let (state_file, remembered) = match &args.state_file { Some(path) => { - let (file, remembered) = swarm::state_file::StateFile::open(path.clone()); + let (file, remembered) = + swarm::state_file::StateFile::open(path.clone(), Arc::clone(&files)); (Some(Arc::new(file)), remembered) } None => (None, None), @@ -117,17 +120,12 @@ async fn main() -> Result<()> { let (change_events, _) = broadcast::channel(256); let recovered = match &args.recover_from { - Some(path) => Some(bootstrap::load_backup(path)?), + Some(path) => Some(bootstrap::load_backup(path, &files)?), None => None, }; - let (initial_doc, recovered_epoch, recovered_swarm, backup_bytes) = match recovered { - Some(backup) => ( - backup.doc, - Some(backup.epoch), - Some(backup.swarm_id), - Some(backup.bytes), - ), - None => (AutoCommit::new(), None, None, None), + let (initial_doc, recovered_epoch, recovered_swarm) = match recovered { + Some(backup) => (backup.doc, Some(backup.epoch), Some(backup.swarm_id)), + None => (AutoCommit::new(), None, None), }; let initial_epoch = match (recovered_epoch, args.debug_epoch) { (Some(epoch), _) => epoch, @@ -179,6 +177,7 @@ async fn main() -> Result<()> { blobs: Arc::new(swarm::blob::Blobs::new(args.blob_chunk_kib * 1024)), protocol_writers: Arc::new(args.protocol_writers.iter().cloned().collect()), state_file, + files, config: Arc::new(config.report()), neighbours: Default::default(), pulling: Default::default(), @@ -193,14 +192,6 @@ async fn main() -> Result<()> { trace!("Application state initialized"); - // Files read before the state existed. - let read = backup_bytes - .into_iter() - .chain(state.state_file.as_ref().and_then(|file| file.read_bytes())); - for bytes in read { - state.stats.disk_read(bytes); - } - // Join the remembered swarm and no other; HELLOs claim it. A backup to recover from // names its swarm itself, which wins. let start = match recovered_swarm { diff --git a/src/state.rs b/src/state.rs index 37bde4a..15652ab 100644 --- a/src/state.rs +++ b/src/state.rs @@ -602,6 +602,8 @@ pub(crate) struct AppState { /// HTTP API users who may write protocol data through the store API /// (`--protocol-writers`). pub(crate) protocol_writers: Arc>, + /// This node's own files, and how much it read and wrote them. + pub(crate) files: Arc, /// What this node remembers across restarts (`--state-file`), if configured. pub(crate) state_file: Option>, /// The effective configuration with sources, imported values left out diff --git a/src/swarm/bootstrap.rs b/src/swarm/bootstrap.rs index 3be462e..f41a00b 100644 --- a/src/swarm/bootstrap.rs +++ b/src/swarm/bootstrap.rs @@ -26,6 +26,7 @@ use uuid::Uuid; use crate::document::replica::Epoch; use crate::document::store; +use crate::files::{FileKind, Files}; use crate::remake::snapshot::{self, META_ROOT, SWARM}; use crate::state::AppState; use crate::swarm::peer::{remember_peer, send_hello_once}; @@ -148,14 +149,14 @@ pub(crate) struct Backup { pub(crate) doc: AutoCommit, pub(crate) epoch: Epoch, pub(crate) swarm_id: String, - /// The file's size (`swarm::stats`, disk). - pub(crate) bytes: u64, } /// Loads a backup to recover from, as `GET /admin/document` saves it. -pub(crate) fn load_backup(path: &Path) -> Result { - let bytes = - std::fs::read(path).with_context(|| format!("Reading the backup {}", path.display()))?; +pub(crate) fn load_backup(path: &Path, files: &Files) -> Result { + let bytes = files + .read(FileKind::Backup, path) + .with_context(|| format!("Reading the backup {}", path.display()))? + .with_context(|| format!("The backup {} doesn't exist", path.display()))?; let mut doc = AutoCommit::load(&bytes) .with_context(|| format!("Loading the backup {}", path.display()))? .with_actor(ActorId::random()); @@ -171,7 +172,6 @@ pub(crate) fn load_backup(path: &Path) -> Result { doc, epoch, swarm_id, - bytes: bytes.len() as u64, }) } diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs index 79caec2..62a6dbe 100644 --- a/src/swarm/state_file.rs +++ b/src/swarm/state_file.rs @@ -4,18 +4,17 @@ //! //! Written rarely: once the node is ready, then when the remembered members change //! (checked every few minutes, written at most once an hour), and at shutdown if they -//! changed since. Atomically: a temporary file next to it, flushed, then renamed over -//! it, so a crash leaves the old file or the new one. +//! changed since, atomically (`files`). -use std::io::Write; -use std::path::{Path, PathBuf}; -use std::sync::{Mutex, MutexGuard}; +use std::path::PathBuf; +use std::sync::{Arc, Mutex, MutexGuard}; use std::time::Duration; use serde::{Deserialize, Serialize}; use tokio::time::Instant; use tracing::{error, info, warn}; +use crate::files::{FileKind, Files}; use crate::state::AppState; use crate::swarm::members; @@ -38,8 +37,7 @@ pub(crate) struct Remembered { pub(crate) struct StateFile { path: PathBuf, - /// Bytes read when opened, if there was a file (`swarm::stats`, disk). - read_bytes: Option, + files: Arc, // A std mutex: never held across an await. inner: Mutex, } @@ -54,30 +52,27 @@ struct Inner { impl StateFile { /// Opens `path`, returning what it remembers. A missing file remembers nothing; an /// unreadable one is moved aside (`.bad`) and logged, and remembers nothing either. - pub(crate) fn open(path: PathBuf) -> (Self, Option) { - let mut read_bytes = None; - let remembered = match std::fs::read(&path) { - Ok(bytes) => { - read_bytes = Some(bytes.len() as u64); - match serde_json::from_slice::(&bytes) { - Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered), - Ok(_) => None, - Err(error) => { - let aside = with_suffix(&path, ".bad"); - error!( - path = %path.display(), - moved_to = %aside.display(), - error = %error, - "Unreadable state file; moving it aside and remembering nothing" - ); - if let Err(error) = std::fs::rename(&path, &aside) { - warn!(error = %error, "Couldn't move the state file aside"); + pub(crate) fn open(path: PathBuf, files: Arc) -> (Self, Option) { + let remembered = match files.read(FileKind::State, &path) { + Ok(Some(bytes)) => match serde_json::from_slice::(&bytes) { + Ok(remembered) if !remembered.swarm_id.is_empty() => Some(remembered), + Ok(_) => None, + Err(error) => { + error!( + path = %path.display(), + error = %error, + "Unreadable state file; moving it aside and remembering nothing" + ); + match files.move_aside(&path, ".bad") { + Ok(aside) => { + warn!(moved_to = %aside.display(), "Moved the state file aside") } - None + Err(error) => warn!(error = %error, "Couldn't move the state file aside"), } + None } - } - Err(error) if error.kind() == std::io::ErrorKind::NotFound => None, + }, + Ok(None) => None, Err(error) => { error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing"); None @@ -85,7 +80,7 @@ impl StateFile { }; let file = Self { path, - read_bytes, + files, inner: Mutex::new(Inner { written: remembered.clone(), ..Inner::default() @@ -94,43 +89,19 @@ impl StateFile { (file, remembered) } - pub(crate) fn read_bytes(&self) -> Option { - self.read_bytes - } - fn lock(&self) -> MutexGuard<'_, Inner> { self.inner .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) } - /// Returns the bytes written. - fn write(&self, remembered: &Remembered) -> std::io::Result { + fn write(&self, remembered: &Remembered) -> std::io::Result<()> { let mut bytes = serde_json::to_vec_pretty(remembered)?; bytes.push(b'\n'); - let temporary = with_suffix(&self.path, ".tmp"); - { - let mut file = std::fs::File::create(&temporary)?; - file.write_all(&bytes)?; - file.sync_all()?; - } - std::fs::rename(&temporary, &self.path)?; - // The rename itself, durably (best effort: not every platform syncs directories). - if let Some(dir) = self.path.parent().filter(|dir| !dir.as_os_str().is_empty()) { - if let Ok(dir) = std::fs::File::open(dir) { - let _ = dir.sync_all(); - } - } - Ok(bytes.len() as u64) + self.files.write_atomic(FileKind::State, &self.path, &bytes) } } -fn with_suffix(path: &Path, suffix: &str) -> PathBuf { - let mut name = path.as_os_str().to_owned(); - name.push(suffix); - PathBuf::from(name) -} - /// Brings the state file up to date if it is due: `now` for the first write and at /// shutdown, otherwise only every few minutes and at most once an hour. Called by the /// peer worker on its maintenance tick. @@ -178,11 +149,10 @@ pub(crate) async fn update(state: &AppState, now: bool) { } } - let writer = std::sync::Arc::clone(&file); + let writer = Arc::clone(&file); let content = remembered.clone(); match tokio::task::spawn_blocking(move || writer.write(&content)).await { - Ok(Ok(bytes)) => { - state.stats.disk_written(bytes); + Ok(Ok(())) => { info!( path = %file.path.display(), seeds = remembered.seeds.len(), diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs index ab5c4da..eb34276 100644 --- a/src/swarm/stats.rs +++ b/src/swarm/stats.rs @@ -33,17 +33,6 @@ struct Inner { /// The document version (`AppState::change_tx`) and epoch the saved size is of. measured_version: Option<(u64, Epoch)>, propagation: Propagation, - disk: Disk, -} - -/// Files this node reads and writes for itself: the state file, a backup to recover -/// from. Not the Mycelium daemon's, and not logs. -#[derive(Debug, Clone, Copy, Default)] -pub(crate) struct Disk { - pub(crate) reads: u64, - pub(crate) read_bytes: u64, - pub(crate) writes: u64, - pub(crate) written_bytes: u64, } /// Upper bounds (inclusive, milliseconds) of the propagation delay buckets; a last @@ -118,7 +107,6 @@ pub(crate) struct Snapshot { pub(crate) max_heads: usize, pub(crate) stale_hello_replies: u64, pub(crate) propagation: Propagation, - pub(crate) disk: Disk, } impl Default for NodeStats { @@ -163,20 +151,6 @@ impl NodeStats { } } - /// A file of `bytes` was read (`Disk`). - pub(crate) fn disk_read(&self, bytes: u64) { - let mut inner = self.lock(); - inner.disk.reads += 1; - inner.disk.read_bytes += bytes; - } - - /// A file of `bytes` was written (`Disk`). - pub(crate) fn disk_written(&self, bytes: u64) { - let mut inner = self.lock(); - inner.disk.writes += 1; - inner.disk.written_bytes += bytes; - } - pub(crate) fn stale_hello_reply(&self) { self.lock().stale_hello_replies += 1; } @@ -231,7 +205,6 @@ impl NodeStats { max_heads: inner.max_heads, stale_hello_replies: inner.stale_hello_replies, propagation: inner.propagation, - disk: inner.disk, } } } -- cgit v1.3.1