mirror of
https://github.com/Portabase/agent.git
synced 2026-10-20 03:11:34 +00:00
fix: stats and logs
This commit is contained in:
+3
-8
@@ -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"
|
||||
}
|
||||
}
|
||||
]
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<String> {
|
||||
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),
|
||||
))
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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<String> {
|
||||
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");
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user