[PATCH proxmox-backup 5/6] fix #7024: cli: add more information inside the backup progress logs

[Shan Shaji <s.shaji@proxmox.com>]

previously, the progress logs only reported the processed stream size
and uploaded size. To improve visibility additional information from
`PxarArchiverProgressStats` and `UploadCounters` has been included in
the logs.

To access the upload statistics, a new callback field has been added to
the UploadOptions struct that accepts UploadCounters and TimeSpan as
parameters. Additionally, to include pxar metadata in the logs, the
`PxarArchiverProgressStats` instance is now created earlier and shared
between the archiver and the progress logger callback.

Fixes: https://bugzilla.proxmox.com/show_bug.cgi?id=7024

Signed-off-by: Shan Shaji <s.shaji@proxmox.com>
---
 pbs-client/src/backup_stats.rs                |  4 +-
 pbs-client/src/backup_writer.rs               | 24 ++++-----
 pbs-client/src/lib.rs                         |  2 +-
 pbs-client/src/pxar/create.rs                 | 17 +++---
 pbs-client/src/pxar/mod.rs                    |  3 +-
 pbs-client/src/pxar_backup_stream.rs          | 13 ++++-
 proxmox-backup-client/src/main.rs             | 53 +++++++++++++++++--
 .../src/proxmox_restore_daemon/api.rs         |  1 +
 pxar-bin/src/main.rs                          |  1 +
 tests/catar.rs                                |  1 +
 10 files changed, 89 insertions(+), 30 deletions(-)

diff --git a/pbs-client/src/backup_stats.rs b/pbs-client/src/backup_stats.rs
index 6aceee331..a1dc7ea7b 100644
--- a/pbs-client/src/backup_stats.rs
+++ b/pbs-client/src/backup_stats.rs
@@ -42,7 +42,7 @@ impl UploadStats {
 
 /// Atomic counters for accounting upload stream progress information
 #[derive(Clone)]
-pub(crate) struct UploadCounters {
+pub struct UploadCounters {
     injected_chunk_count: Arc<AtomicUsize>,
     known_chunk_count: Arc<AtomicUsize>,
     total_chunk_count: Arc<AtomicUsize>,
@@ -99,7 +99,7 @@ impl UploadCounters {
     }
 
     #[inline(always)]
-    pub(crate) fn total_stream_len(&self) -> usize {
+    pub fn total_stream_len(&self) -> usize {
         self.total_stream_len.load(Ordering::SeqCst)
     }
 
diff --git a/pbs-client/src/backup_writer.rs b/pbs-client/src/backup_writer.rs
index 34d8db90b..8ae0cffe8 100644
--- a/pbs-client/src/backup_writer.rs
+++ b/pbs-client/src/backup_writer.rs
@@ -45,6 +45,8 @@ impl Drop for BackupWriter {
     }
 }
 
+type ProgressLoggerCallback = Box<dyn Fn(&UploadCounters, &TimeSpan) + Send + Sync>;
+
 /// Options for uploading blobs/streams to the server
 #[derive(Default, Clone)]
 pub struct UploadOptions {
@@ -52,6 +54,7 @@ pub struct UploadOptions {
     pub compress: bool,
     pub encrypt: bool,
     pub index_type: IndexType,
+    pub progress_logging_callback: Option<Arc<ProgressLoggerCallback>>,
 }
 
 /// Index type for upload options.
@@ -371,11 +374,11 @@ impl BackupWriter {
         let upload_stats = Self::upload_merged_chunk_stream(
             self.h2.clone(),
             wid,
-            archive_name,
             prefix,
             stream,
             index_csum_2,
             counters_readonly,
+            options.progress_logging_callback,
         )
         .await?;
 
@@ -476,7 +479,7 @@ impl BackupWriter {
             },
             options.compress,
             injections,
-            archive_name,
+            options.progress_logging_callback,
         )
         .await?;
 
@@ -768,7 +771,7 @@ impl BackupWriter {
         crypt_config: Option<Arc<CryptConfig>>,
         compress: bool,
         injections: Option<std::sync::mpsc::Receiver<InjectChunks>>,
-        archive: &BackupArchiveName,
+        progress_logging_callback: Option<Arc<ProgressLoggerCallback>>,
     ) -> impl Future<Output = Result<UploadStats, Error>> {
         let mut counters = UploadCounters::new();
         let counters_readonly = counters.clone();
@@ -849,22 +852,22 @@ impl BackupWriter {
         Self::upload_merged_chunk_stream(
             h2,
             wid,
-            archive,
             prefix,
             stream,
             index_csum_2,
             counters_readonly,
+            progress_logging_callback,
         )
     }
 
     fn upload_merged_chunk_stream(
         h2: H2Client,
         wid: u64,
-        archive: &BackupArchiveName,
         prefix: &str,
         stream: impl Stream<Item = Result<MergedChunkInfo, Error>>,
         index_csum: Arc<Mutex<Option<Sha256>>>,
         counters: UploadCounters,
+        progress_logging_callback: Option<Arc<ProgressLoggerCallback>>,
     ) -> impl Future<Output = Result<UploadStats, Error>> {
         let append_chunk_path = format!("{prefix}_index");
         let upload_chunk_path = format!("{prefix}_chunk");
@@ -874,20 +877,13 @@ impl BackupWriter {
         let (upload_queue, upload_result) =
             Self::append_chunk_queue(h2.clone(), wid, append_chunk_path, counters.clone());
 
-        let progress_handle = if archive.ends_with(".img.fidx")
-            || archive.ends_with(".pxar.didx")
-            || archive.ends_with(".ppxar.didx")
-        {
+        let progress_handle = if let Some(callback) = progress_logging_callback {
             let counters = counters.clone();
             Some(tokio::spawn(async move {
                 loop {
                     tokio::time::sleep(tokio::time::Duration::from_secs(60)).await;
-
-                    let size = HumanByte::from(counters.total_stream_len());
-                    let size_uploaded = HumanByte::from(counters.uploaded_len());
                     let elapsed = TimeSpan::from(start_time.elapsed());
-
-                    info!("processed {size} in {elapsed}, uploaded {size_uploaded}");
+                    callback(&counters, &elapsed);
                 }
             }))
         } else {
diff --git a/pbs-client/src/lib.rs b/pbs-client/src/lib.rs
index 139f4f4b5..6bf6cc4e6 100644
--- a/pbs-client/src/lib.rs
+++ b/pbs-client/src/lib.rs
@@ -43,7 +43,7 @@ mod chunk_stream;
 pub use chunk_stream::{ChunkStream, FixedChunkStream, InjectionData};
 
 mod backup_stats;
-pub use backup_stats::BackupStats;
+pub use backup_stats::{BackupStats, UploadCounters};
 
 mod log_throttle;
 pub use log_throttle::LogThrottle;
diff --git a/pbs-client/src/pxar/create.rs b/pbs-client/src/pxar/create.rs
index ed7ca92f2..7012a88dd 100644
--- a/pbs-client/src/pxar/create.rs
+++ b/pbs-client/src/pxar/create.rs
@@ -151,7 +151,7 @@ pub(crate) struct HardLinkInfo {
 }
 
 #[derive(Default)]
-struct PxarArchiverProgressStats {
+pub struct PxarArchiverProgressStats {
     files_reused_count: AtomicU64,
     files_hardlink_count: AtomicU64,
     files_reencoded_count: AtomicU64,
@@ -169,17 +169,17 @@ impl PxarArchiverProgressStats {
     }
 
     #[inline(always)]
-    pub(crate) fn total_reencoded_size(&self) -> u64 {
+    pub fn total_reencoded_size(&self) -> u64 {
         self.total_reencoded_size.load(Ordering::Acquire)
     }
 
     #[inline(always)]
-    pub(crate) fn files_reused_count(&self) -> u64 {
+    pub fn files_reused_count(&self) -> u64 {
         self.files_reused_count.load(Ordering::Acquire)
     }
 
     #[inline(always)]
-    pub(crate) fn total_reused_payload_size(&self) -> u64 {
+    pub fn total_reused_payload_size(&self) -> u64 {
         self.total_reused_payload_size.load(Ordering::Acquire)
     }
 
@@ -197,7 +197,7 @@ impl PxarArchiverProgressStats {
     }
 
     #[inline(always)]
-    pub(crate) fn total_files_processed(&self) -> u64 {
+    pub fn total_files_processed(&self) -> u64 {
         self.files_reused_count() + self.total_hardlink_count() + self.files_reencoded_count()
     }
 
@@ -269,7 +269,7 @@ struct Archiver {
     // an older client without the encode-time check is detected here and the affected file is
     // re-encoded instead of reused.
     last_reusable_offset: Option<u64>,
-    progress_stats: PxarArchiverProgressStats,
+    progress_stats: Arc<PxarArchiverProgressStats>,
     split_archive: bool,
 }
 
@@ -297,6 +297,7 @@ pub async fn create_archive<T, F>(
     options: PxarCreateOptions,
     forced_boundaries: Option<mpsc::Sender<InjectChunks>>,
     suggested_boundaries: Option<mpsc::Sender<u64>>,
+    archiver_progress_stats: Option<Arc<PxarArchiverProgressStats>>,
 ) -> Result<(), Error>
 where
     T: SeqWrite + Send,
@@ -380,7 +381,7 @@ where
         previous_payload_index,
         cache: PxarLookaheadCache::new(options.max_cache_size),
         last_reusable_offset: None,
-        progress_stats: PxarArchiverProgressStats::default(),
+        progress_stats: archiver_progress_stats.unwrap_or_default(),
         split_archive,
     };
 
@@ -2144,7 +2145,7 @@ mod tests {
                 suggested_boundaries: Some(suggested_boundaries),
                 cache: PxarLookaheadCache::new(None),
                 last_reusable_offset: None,
-                progress_stats: PxarArchiverProgressStats::default(),
+                progress_stats: Arc::new(PxarArchiverProgressStats::default()),
                 split_archive: true,
             };
 
diff --git a/pbs-client/src/pxar/mod.rs b/pbs-client/src/pxar/mod.rs
index b88fac33f..d6820d01a 100644
--- a/pbs-client/src/pxar/mod.rs
+++ b/pbs-client/src/pxar/mod.rs
@@ -58,7 +58,8 @@ mod flags;
 pub use flags::Flags;
 
 pub use create::{
-    MetadataArchiveReader, PxarCreateOptions, PxarPrevRef, PxarWriters, create_archive,
+    MetadataArchiveReader, PxarArchiverProgressStats, PxarCreateOptions, PxarPrevRef, PxarWriters,
+    create_archive,
 };
 pub use extract::{
     ErrorHandler, OverwriteFlags, PxarExtractContext, PxarExtractOptions, create_tar, create_zip,
diff --git a/pbs-client/src/pxar_backup_stream.rs b/pbs-client/src/pxar_backup_stream.rs
index 430eee01f..d1b46cf8a 100644
--- a/pbs-client/src/pxar_backup_stream.rs
+++ b/pbs-client/src/pxar_backup_stream.rs
@@ -20,6 +20,7 @@ use proxmox_log::debug;
 use pbs_datastore::catalog::{BackupCatalogWriter, CatalogWriter};
 
 use crate::inject_reused_chunks::InjectChunks;
+use crate::pxar::PxarArchiverProgressStats;
 use crate::pxar::create::PxarWriters;
 
 /// Stream implementation to encode and upload .pxar archives.
@@ -50,6 +51,7 @@ impl PxarBackupStream {
         options: crate::pxar::PxarCreateOptions,
         boundaries: Option<mpsc::Sender<InjectChunks>>,
         separate_payload_stream: bool,
+        archiver_progress_stats: Option<Arc<PxarArchiverProgressStats>>,
     ) -> Result<(Self, Option<Self>), Error> {
         let buffer_size = 256 * 1024;
 
@@ -102,6 +104,7 @@ impl PxarBackupStream {
                 options,
                 boundaries,
                 suggested_boundaries_tx,
+                archiver_progress_stats,
             )
             .await
             {
@@ -145,10 +148,18 @@ impl PxarBackupStream {
         options: crate::pxar::PxarCreateOptions,
         boundaries: Option<mpsc::Sender<InjectChunks>>,
         separate_payload_stream: bool,
+        archiver_progress_stats: Option<Arc<PxarArchiverProgressStats>>,
     ) -> Result<(Self, Option<Self>), Error> {
         let dir = nix::dir::Dir::open(dirname, OFlag::O_DIRECTORY, Mode::empty())?;
 
-        Self::new(dir, catalog, options, boundaries, separate_payload_stream)
+        Self::new(
+            dir,
+            catalog,
+            options,
+            boundaries,
+            separate_payload_stream,
+            archiver_progress_stats,
+        )
     }
 }
 
diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs
index 5c52b7917..8ce5540d5 100644
--- a/proxmox-backup-client/src/main.rs
+++ b/proxmox-backup-client/src/main.rs
@@ -33,7 +33,9 @@ use pbs_api_types::{
     RateLimitConfig, ServerIdentity, SnapshotListItem, StorageStatus,
 };
 use pbs_client::catalog_shell::Shell;
-use pbs_client::pxar::{ErrorHandler as PxarErrorHandler, MetadataArchiveReader, PxarPrevRef};
+use pbs_client::pxar::{
+    ErrorHandler as PxarErrorHandler, MetadataArchiveReader, PxarArchiverProgressStats, PxarPrevRef,
+};
 use pbs_client::tools::{
     CHUNK_SIZE_SCHEMA, complete_archive_name, complete_auth_id, complete_backup_group,
     complete_backup_snapshot, complete_backup_source, complete_chunk_size,
@@ -50,8 +52,8 @@ use pbs_client::{
     BACKUP_SOURCE_SCHEMA, BackupDetectionMode, BackupReader, BackupRepository,
     BackupRepositoryArgs, BackupSpecificationType, BackupStats, BackupTargetArgs, BackupWriter,
     BackupWriterOptions, ChunkStream, FixedChunkStream, HttpClient, IndexType, InjectionData,
-    LogThrottle, PxarBackupStream, RemoteChunkReader, UploadOptions, delete_ticket_info,
-    parse_backup_specification, view_task_result,
+    LogThrottle, PxarBackupStream, RemoteChunkReader, UploadCounters, UploadOptions,
+    delete_ticket_info, parse_backup_specification, view_task_result,
 };
 use pbs_datastore::catalog::{BackupCatalogWriter, CatalogReader, CatalogWriter};
 use pbs_datastore::chunk_store::verify_chunk_size;
@@ -265,11 +267,13 @@ async fn backup_directory<P: AsRef<Path>>(
     catalog: Option<Arc<Mutex<Catalog>>>,
     pxar_create_options: pbs_client::pxar::PxarCreateOptions,
     upload_options: UploadOptions,
+    is_metadata_mode: bool,
 ) -> Result<(BackupStats, Option<BackupStats>), Error> {
     if let IndexType::Fixed(_) = upload_options.index_type {
         bail!("cannot backup directory with fixed chunk size!");
     }
 
+    let pxar_archiver_progress_stats = Arc::new(PxarArchiverProgressStats::default());
     let (payload_boundaries_tx, payload_boundaries_rx) = std::sync::mpsc::channel();
     let (pxar_stream, payload_stream) = PxarBackupStream::open(
         dir_path.as_ref(),
@@ -277,6 +281,7 @@ async fn backup_directory<P: AsRef<Path>>(
         pxar_create_options,
         Some(payload_boundaries_tx),
         payload_target.is_some(),
+        Some(pxar_archiver_progress_stats.clone()),
     )?;
 
     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size, None, None);
@@ -293,7 +298,40 @@ async fn backup_directory<P: AsRef<Path>>(
 
     let stats = client.upload_stream(archive_name, stream, upload_options.clone(), None);
 
+    let progress_logger = move |counters: &UploadCounters, elapsed: &TimeSpan| {
+        let size = HumanByte::from(counters.total_stream_len());
+        let size_uploaded = HumanByte::from(counters.uploaded_len());
+
+        let total_files_processed = pxar_archiver_progress_stats.total_files_processed();
+        let total_reencoded_size = pxar_archiver_progress_stats.total_reencoded_size();
+
+        if is_metadata_mode {
+            let reuse_file_count = pxar_archiver_progress_stats.files_reused_count();
+            let reuse_payload_size = pxar_archiver_progress_stats.total_reused_payload_size();
+
+            let total_reuse_payload_size = HumanByte::from(reuse_payload_size);
+            let total_logical_file_size =
+                HumanByte::from(total_reencoded_size + reuse_payload_size);
+            log::info!(
+                "scanned {total_files_processed} files with {total_logical_file_size} \
+                       (reusing {reuse_file_count} files with {total_reuse_payload_size}) \
+                       of which processed {size} in {elapsed} and uploaded {size_uploaded}"
+            );
+        } else {
+            let total_logical_file_size = HumanByte::from(total_reencoded_size);
+            log::info!(
+                "scanned {total_files_processed} files with {total_logical_file_size} \
+                       of which processed {size} in {elapsed} and uploaded {size_uploaded}"
+            );
+        }
+    };
+
     if let Some(mut payload_stream) = payload_stream {
+        let upload_options = UploadOptions {
+            progress_logging_callback: Some(Arc::new(Box::new(progress_logger))),
+            ..upload_options
+        };
+
         let payload_target = payload_target
             .ok_or_else(|| format_err!("got payload stream, but no target archive name"))?;
 
@@ -1248,6 +1286,7 @@ async fn create_backup(
                     catalog.as_ref().cloned(),
                     pxar_options,
                     upload_options,
+                    detection_mode.is_metadata(),
                 )
                 .await?;
 
@@ -1268,6 +1307,12 @@ async fn create_backup(
             (BackupSpecificationType::IMAGE, false) => {
                 log_file("image", &filename, target.as_ref());
 
+                let progress_logger = |counters: &UploadCounters, elapsed: &TimeSpan| {
+                    let size = HumanByte::from(counters.total_stream_len());
+                    let size_uploaded = HumanByte::from(counters.uploaded_len());
+                    log::info!("processed {size} in {elapsed}, uploaded {size_uploaded}");
+                };
+
                 // 0 means fifo pipe with unknown size
                 let image_file_size = (size != 0).then_some(size);
                 let upload_options = UploadOptions {
@@ -1275,6 +1320,8 @@ async fn create_backup(
                     index_type: IndexType::Fixed(image_file_size),
                     compress: true,
                     encrypt: crypto.mode == CryptMode::Encrypt,
+                    progress_logging_callback: Some(Arc::new(Box::new(progress_logger))),
+                    ..UploadOptions::default()
                 };
 
                 let stats =
diff --git a/proxmox-restore-daemon/src/proxmox_restore_daemon/api.rs b/proxmox-restore-daemon/src/proxmox_restore_daemon/api.rs
index ff21c9bee..b7f4fd141 100644
--- a/proxmox-restore-daemon/src/proxmox_restore_daemon/api.rs
+++ b/proxmox-restore-daemon/src/proxmox_restore_daemon/api.rs
@@ -375,6 +375,7 @@ fn extract(
                         options,
                         None,
                         None,
+                        None
                     )
                     .await
                 }
diff --git a/pxar-bin/src/main.rs b/pxar-bin/src/main.rs
index 4b14507ca..953c6e756 100644
--- a/pxar-bin/src/main.rs
+++ b/pxar-bin/src/main.rs
@@ -450,6 +450,7 @@ async fn create_archive(
         options,
         None,
         None,
+        None,
     )
     .await?;
 
diff --git a/tests/catar.rs b/tests/catar.rs
index aed23f866..81ca9d8db 100644
--- a/tests/catar.rs
+++ b/tests/catar.rs
@@ -41,6 +41,7 @@ fn run_test(dir_name: &str) -> Result<(), Error> {
         options,
         None,
         None,
+        None,
     ))?;
 
     Command::new("cmp")
-- 
2.47.3