add disk I/O stats

6 files changed, 138 insertions, 27 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 838cc0f..6c5e172 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -35,6 +35,18 @@ pub(crate) struct StatsResponse {
     /// The peers listed in /peers together (a forgotten peer's are gone).
     incidents: Incidents,
     propagation: PropagationStats,
+    disk: DiskStats,
+}
+
+/// Files this node reads and writes for itself (the state file, a backup to recover
+/// from), since it started.
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct DiskStats {
+    reads: u64,
+    read_bytes: u64,
+    writes: u64,
+    written_bytes: u64,
 }
 
 /// How long changes written elsewhere took from their commit there until applied here,
@@ -195,5 +207,11 @@ 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,
+        },
     }
 }
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index f68ca85..ea97bd2 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -297,6 +297,32 @@ 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,
+    );
+
     let propagation = &stats.propagation;
     out.family(
         "propagation_delay_seconds",
diff --git a/src/main.rs b/src/main.rs
index d6773eb..57c26b5 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -120,9 +120,14 @@ async fn main() -> Result<()> {
         Some(path) => Some(bootstrap::load_backup(path)?),
         None => None,
     };
-    let (initial_doc, recovered_epoch, recovered_swarm) = match recovered {
-        Some((doc, epoch, swarm_id)) => (doc, Some(epoch), Some(swarm_id)),
-        None => (AutoCommit::new(), 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_epoch = match (recovered_epoch, args.debug_epoch) {
         (Some(epoch), _) => epoch,
@@ -188,6 +193,14 @@ 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/swarm/bootstrap.rs b/src/swarm/bootstrap.rs
index 125321b..b725786 100644
--- a/src/swarm/bootstrap.rs
+++ b/src/swarm/bootstrap.rs
@@ -140,9 +140,18 @@ fn read_swarm_id(doc: &AutoCommit) -> Option<String> {
     }
 }
 
-/// A backup to recover from (`--recover-from`, as `GET /admin/document` saves it): the
-/// document under a fresh actor, the epoch it is on and the swarm it names.
-pub(crate) fn load_backup(path: &Path) -> Result<(AutoCommit, Epoch, String)> {
+/// A backup to recover from (`--recover-from`).
+pub(crate) struct Backup {
+    /// Under a fresh actor.
+    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<Backup> {
     let bytes =
         std::fs::read(path).with_context(|| format!("Reading the backup {}", path.display()))?;
     let mut doc = AutoCommit::load(&bytes)
@@ -156,7 +165,12 @@ pub(crate) fn load_backup(path: &Path) -> Result<(AutoCommit, Epoch, String)> {
     };
     let epoch = snapshot::epoch_of(&mut doc)
         .map_err(|reason| anyhow::anyhow!("The backup {}: {reason}", path.display()))?;
-    Ok((doc, epoch, swarm_id))
+    Ok(Backup {
+        doc,
+        epoch,
+        swarm_id,
+        bytes: bytes.len() as u64,
+    })
 }
 
 /// Resolves the seeds in `unresolved`, says HELLO to each that resolves, and keeps the
diff --git a/src/swarm/state_file.rs b/src/swarm/state_file.rs
index 0a9913d..79caec2 100644
--- a/src/swarm/state_file.rs
+++ b/src/swarm/state_file.rs
@@ -38,6 +38,8 @@ 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<u64>,
     // A std mutex: never held across an await.
     inner: Mutex<Inner>,
 }
@@ -53,24 +55,28 @@ 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<Remembered>) {
+        let mut read_bytes = None;
         let remembered = match std::fs::read(&path) {
-            Ok(bytes) => match serde_json::from_slice::<Remembered>(&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");
+            Ok(bytes) => {
+                read_bytes = Some(bytes.len() as u64);
+                match serde_json::from_slice::<Remembered>(&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");
+                        }
+                        None
                     }
-                    None
                 }
-            },
+            }
             Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
             Err(error) => {
                 error!(path = %path.display(), error = %error, "Couldn't read the state file; remembering nothing");
@@ -79,6 +85,7 @@ impl StateFile {
         };
         let file = Self {
             path,
+            read_bytes,
             inner: Mutex::new(Inner {
                 written: remembered.clone(),
                 ..Inner::default()
@@ -87,19 +94,24 @@ impl StateFile {
         (file, remembered)
     }
 
+    pub(crate) fn read_bytes(&self) -> Option<u64> {
+        self.read_bytes
+    }
+
     fn lock(&self) -> MutexGuard<'_, Inner> {
         self.inner
             .lock()
             .unwrap_or_else(|poisoned| poisoned.into_inner())
     }
 
-    fn write(&self, remembered: &Remembered) -> std::io::Result<()> {
-        let bytes = serde_json::to_vec_pretty(remembered)?;
+    /// Returns the bytes written.
+    fn write(&self, remembered: &Remembered) -> std::io::Result<u64> {
+        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.write_all(b"\n")?;
             file.sync_all()?;
         }
         std::fs::rename(&temporary, &self.path)?;
@@ -109,7 +121,7 @@ impl StateFile {
                 let _ = dir.sync_all();
             }
         }
-        Ok(())
+        Ok(bytes.len() as u64)
     }
 }
 
@@ -169,7 +181,8 @@ pub(crate) async fn update(state: &AppState, now: bool) {
     let writer = std::sync::Arc::clone(&file);
     let content = remembered.clone();
     match tokio::task::spawn_blocking(move || writer.write(&content)).await {
-        Ok(Ok(())) => {
+        Ok(Ok(bytes)) => {
+            state.stats.disk_written(bytes);
             info!(
                 path = %file.path.display(),
                 seeds = remembered.seeds.len(),
diff --git a/src/swarm/stats.rs b/src/swarm/stats.rs
index eb34276..ab5c4da 100644
--- a/src/swarm/stats.rs
+++ b/src/swarm/stats.rs
@@ -33,6 +33,17 @@ 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
@@ -107,6 +118,7 @@ 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 {
@@ -151,6 +163,20 @@ 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;
     }
@@ -205,6 +231,7 @@ impl NodeStats {
             max_heads: inner.max_heads,
             stale_hello_replies: inner.stale_hello_replies,
             propagation: inner.propagation,
+            disk: inner.disk,
         }
     }
 }