From 2139dd494efddda6d5b15ed39e39991d20da23c5 Mon Sep 17 00:00:00 2001 From: charles-gauthereau Date: Sun, 18 Oct 2026 19:32:20 +0200 Subject: [PATCH] fix: stats and logs --- databases.json | 11 +++------- src/domain/files/sync/rclone.rs | 33 ++++++++++++++++++++++-------- src/domain/files/sync/stats.rs | 21 +++++++++++++++++++ src/services/restore/downloader.rs | 11 +--------- src/tests/domain/files/sync.rs | 15 +++++++++++++- src/utils/text.rs | 10 +++++++++ 6 files changed, 74 insertions(+), 27 deletions(-) diff --git a/databases.json b/databases.json index f480f42..5e65499 100644 --- a/databases.json +++ b/databases.json @@ -138,9 +138,7 @@ "path": "/data/files-archive", "generated_id": "64aeaf80-e03b-4503-a5c1-00a1b829f0e5", "options": { - "method": "archive", - "exclude": ["node_modules", "*.tmp", "/cache"], - "one_file_system": false + "method": "archive" } }, { @@ -149,9 +147,7 @@ "path": "/data/files-snapshot", "generated_id": "9093ad85-3f67-489f-9759-aa369572b30d", "options": { - "method": "snapshot", - "exclude": ["node_modules", "*.log", "/build"], - "one_file_system": true + "method": "snapshot" } }, { @@ -160,8 +156,7 @@ "path": "/data/files-sync", "generated_id": "353ecc44-7c75-40e7-803e-ac07683c2c05", "options": { - "method": "sync", - "exclude": ["*.tmp", "/cache"] + "method": "sync" } } ] diff --git a/src/domain/files/sync/rclone.rs b/src/domain/files/sync/rclone.rs index f940c7a..6dad00c 100644 --- a/src/domain/files/sync/rclone.rs +++ b/src/domain/files/sync/rclone.rs @@ -1,5 +1,5 @@ use super::excludes::sync_excludes; -use super::stats::{SyncStats, parse_log}; +use super::stats::{SyncStats, parse_log, progress}; use crate::domain::files::matcher::{exclude_patterns, one_file_system, source_root}; use crate::services::backup::logger::JobLogger; use crate::services::config::DatabaseConfig; @@ -7,6 +7,7 @@ use crate::services::storage::providers::rclone::helpers::{remote_target, write_ use anyhow::{Context, Result, anyhow, bail}; use std::path::Path; use std::process::Stdio; +use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::process::Command; pub fn replica_target(remote_name: &str, base_path: &str, folder: &str, generated_id: &str) -> String { @@ -34,7 +35,7 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge .arg(&root) .arg(dest) .args([ - "--links", "--delete-excluded", "--use-json-log", "--stats", "1h", "--stats-log-level", "NOTICE", "--retries", "1", + "--links", "--delete-excluded", "--use-json-log", "--stats-log-level", "NOTICE", "--retries", "1", "--contimeout", "30s", "--low-level-retries", "3", ]); if one_file_system(cfg) { @@ -44,29 +45,45 @@ pub async fn sync_dir(config_text: &str, dest: &str, cfg: &DatabaseConfig, logge cmd.arg("--exclude").arg(pattern); } + let debug = tracing::enabled!(tracing::Level::DEBUG); + cmd.args(["--stats", if debug { "5s" } else { "1h" }]); + logger.log("info", format!("Syncing {} to {dest}", root.display())); - let out = cmd + let mut child = cmd .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) .kill_on_drop(true) - .output() - .await + .spawn() .map_err(|e| match e.kind() { std::io::ErrorKind::NotFound => anyhow!("rclone binary not found"), _ => anyhow!(e).context("failed to start rclone"), })?; + let mut lines = BufReader::new(child.stderr.take().context("rclone stderr unavailable")?).lines(); + let mut stderr = String::new(); + while let Some(line) = lines.next_line().await.context("failed to read rclone output")? { + if debug { + if let Some(progress) = progress(&line) { + tracing::debug!("{progress}"); + } + } + stderr.push_str(&line); + stderr.push('\n'); + } + let status = child.wait().await.context("failed to wait for rclone")?; - let log = parse_log(&String::from_utf8_lossy(&out.stderr)); + let log = parse_log(&stderr); for error in &log.errors { logger.log("warn", format!("rclone: {error}")); } - if !out.status.success() { + if !status.success() { let reason = log .stats .as_ref() .and_then(|s| s.last_error.clone()) .or_else(|| log.errors.last().cloned()) .unwrap_or_else(|| "no error message".to_string()); - bail!("rclone sync failed ({}): {reason}; the replica may be partially updated", out.status); + bail!("rclone sync failed ({status}): {reason}; the replica may be partially updated"); } let mut stats = log.stats.unwrap_or_default(); stats.replica_bytes = match remote_size(config.path(), dest).await { diff --git a/src/domain/files/sync/stats.rs b/src/domain/files/sync/stats.rs index 0772176..bfa925e 100644 --- a/src/domain/files/sync/stats.rs +++ b/src/domain/files/sync/stats.rs @@ -1,3 +1,4 @@ +use crate::utils::text::human_size; use serde::Deserialize; use serde_json::{Value, json}; @@ -62,3 +63,23 @@ pub fn parse_log(stderr: &str) -> SyncLog { } log } + +pub fn progress(line: &str) -> Option { + let value: Value = serde_json::from_str(line).ok()?; + let stats = value.get("stats")?; + let n = |key: &str| stats[key].as_u64().unwrap_or(0); + let (bytes, total) = (n("bytes"), n("totalBytes")); + let percent = if total > 0 { bytes * 100 / total } else { 100 }; + let eta = stats["eta"].as_u64().map(|s| format!(", ETA {s}s")).unwrap_or_default(); + Some(format!( + "rclone sync: {} / {} ({percent}%), {}/{} transferred, {}/{} checked, {} deleted, {}/s{eta}", + human_size(bytes), + human_size(total), + n("transfers"), + n("totalTransfers"), + n("checks"), + n("totalChecks"), + n("deletes"), + human_size(stats["speed"].as_f64().unwrap_or(0.0) as u64), + )) +} diff --git a/src/services/restore/downloader.rs b/src/services/restore/downloader.rs index bd70a98..c4ae1e0 100644 --- a/src/services/restore/downloader.rs +++ b/src/services/restore/downloader.rs @@ -2,6 +2,7 @@ use super::service::RestoreService; use crate::services::backup::logger::JobLogger; use crate::utils::retry::{RetryPolicy, retry}; +use crate::utils::text::human_size; use anyhow::Result; use futures::StreamExt; use reqwest::{Client, Url}; @@ -10,16 +11,6 @@ use std::sync::Arc; use std::time::Instant; use tokio::io::AsyncWriteExt; -fn human_size(bytes: u64) -> String { - if bytes >= 1024 * 1024 { - format!("{} MB", bytes / 1024 / 1024) - } else if bytes >= 1024 { - format!("{} KB", bytes / 1024) - } else { - format!("{bytes} B") - } -} - impl RestoreService { pub async fn download_backup( &self, diff --git a/src/tests/domain/files/sync.rs b/src/tests/domain/files/sync.rs index 7197a26..f1baea8 100644 --- a/src/tests/domain/files/sync.rs +++ b/src/tests/domain/files/sync.rs @@ -1,5 +1,5 @@ use crate::domain::files::sync::excludes::sync_excludes; -use crate::domain::files::sync::stats::{SyncStats, parse_log}; +use crate::domain::files::sync::stats::{SyncStats, parse_log, progress}; fn strings(items: &[&str]) -> Vec { items.iter().map(|s| s.to_string()).collect() @@ -71,6 +71,19 @@ fn parse_log_reads_a_successful_run() { assert!(log.errors.is_empty()); } +#[test] +fn progress_formats_periodic_stats_and_ignores_other_lines() { + let line = r#"{"level":"notice","msg":"stats","stats":{"bytes":536870912,"totalBytes":1073741824,"transfers":1200,"totalTransfers":2924,"checks":3,"totalChecks":10,"deletes":0,"speed":89128960.5,"eta":6}}"#; + assert_eq!( + progress(line).unwrap(), + "rclone sync: 512 MB / 1024 MB (50%), 1200/2924 transferred, 3/10 checked, 0 deleted, 85 MB/s, ETA 6s" + ); + let idle = r#"{"level":"notice","stats":{"bytes":0,"totalBytes":0,"speed":0,"eta":null}}"#; + assert_eq!(progress(idle).unwrap(), "rclone sync: 0 B / 0 B (100%), 0/0 transferred, 0/0 checked, 0 deleted, 0 B/s"); + assert!(progress(r#"{"level":"error","msg":"boom"}"#).is_none()); + assert!(progress("plain text").is_none()); +} + #[test] fn parse_log_keeps_plain_text_lines_as_errors() { let log = parse_log("Error: unknown flag: --bogus\n"); diff --git a/src/utils/text.rs b/src/utils/text.rs index a8814ea..e3ef593 100644 --- a/src/utils/text.rs +++ b/src/utils/text.rs @@ -53,3 +53,13 @@ fn convert_dow_part(part: &str) -> String { None => converted, } } + +pub fn human_size(bytes: u64) -> String { + if bytes >= 1024 * 1024 { + format!("{} MB", bytes / 1024 / 1024) + } else if bytes >= 1024 { + format!("{} KB", bytes / 1024) + } else { + format!("{bytes} B") + } +}