[RFC PATCH proxmox-backup 4/6] fix #7024: cli: show number of processed files during backup

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

earlier, the progress logs only reported the processed stream size and
uploaded size. To improve this, `PxarArchiverProgressStats` is now
created earlier and shared between the archiver and the upload stream.

Signed-off-by: Shan Shaji <s.shaji@proxmox.com>
---
 pbs-client/src/backup_stats.rs                |  5 ++-
 pbs-client/src/backup_writer.rs               | 39 +++++++++++++++++--
 pbs-client/src/pxar/create.rs                 |  9 +++--
 pbs-client/src/pxar/mod.rs                    |  3 +-
 pbs-client/src/pxar_backup_stream.rs          | 13 ++++++-
 proxmox-backup-client/src/main.rs             | 27 +++++++++++--
 .../src/proxmox_restore_daemon/api.rs         |  1 +
 pxar-bin/src/main.rs                          |  1 +
 tests/catar.rs                                |  1 +
 9 files changed, 85 insertions(+), 14 deletions(-)

diff --git a/pbs-client/src/backup_stats.rs b/pbs-client/src/backup_stats.rs
index d7c13784b..aa5ae4889 100644
--- a/pbs-client/src/backup_stats.rs
+++ b/pbs-client/src/backup_stats.rs
@@ -4,6 +4,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::time::Duration;
 
+use crate::pxar::PxarArchiverProgressStats;
 use crate::pxar::create::ReusableDynamicEntry;
 
 /// Basic backup run statistics and archive checksum
@@ -50,11 +51,12 @@ pub(crate) struct UploadCounters {
     injected_stream_len: Arc<AtomicUsize>,
     reused_stream_len: Arc<AtomicUsize>,
     total_stream_len: Arc<AtomicUsize>,
+    pub pxar_progress: Option<Arc<PxarArchiverProgressStats>>,
 }
 
 impl UploadCounters {
     /// Create and zero init new upload counters
-    pub(crate) fn new() -> Self {
+    pub(crate) fn new(pxar_progress: Option<Arc<PxarArchiverProgressStats>>) -> Self {
         Self {
             total_chunk_count: Arc::new(AtomicUsize::new(0)),
             injected_chunk_count: Arc::new(AtomicUsize::new(0)),
@@ -63,6 +65,7 @@ impl UploadCounters {
             injected_stream_len: Arc::new(AtomicUsize::new(0)),
             reused_stream_len: Arc::new(AtomicUsize::new(0)),
             total_stream_len: Arc::new(AtomicUsize::new(0)),
+            pxar_progress: pxar_progress,
         }
     }
 
diff --git a/pbs-client/src/backup_writer.rs b/pbs-client/src/backup_writer.rs
index f0744ecaf..d0246689c 100644
--- a/pbs-client/src/backup_writer.rs
+++ b/pbs-client/src/backup_writer.rs
@@ -28,6 +28,8 @@ use proxmox_human_byte::HumanByte;
 use proxmox_log::{Level, debug, enabled, info, trace, warn};
 use proxmox_time::TimeSpan;
 
+use crate::pxar::PxarArchiverProgressStats;
+
 use super::backup_stats::{BackupStats, UploadCounters, UploadStats};
 use super::inject_reused_chunks::{InjectChunks, InjectReusedChunks, InjectedChunksInfo};
 use super::merge_known_chunks::{MergeKnownChunks, MergedChunkInfo};
@@ -53,6 +55,7 @@ pub struct UploadOptions {
     pub compress: bool,
     pub encrypt: bool,
     pub index_type: IndexType,
+    pub is_metadata_mode: bool,
 }
 
 /// Index type for upload options.
@@ -327,7 +330,7 @@ impl BackupWriter {
             .as_u64()
             .unwrap();
 
-        let mut counters = UploadCounters::new();
+        let mut counters = UploadCounters::new(None);
         let counters_readonly = counters.clone();
 
         let is_fixed_chunk_size = prefix == "fixed";
@@ -377,6 +380,7 @@ impl BackupWriter {
             stream,
             index_csum_2,
             counters_readonly,
+            options.is_metadata_mode,
         )
         .await?;
 
@@ -400,6 +404,7 @@ impl BackupWriter {
         stream: impl Stream<Item = Result<bytes::BytesMut, Error>>,
         options: UploadOptions,
         injections: Option<std::sync::mpsc::Receiver<InjectChunks>>,
+        pxar_archive_progress_stats: Option<Arc<PxarArchiverProgressStats>>,
     ) -> Result<BackupStats, Error> {
         let known_chunks = Arc::new(Mutex::new(HashSet::new()));
 
@@ -478,6 +483,8 @@ impl BackupWriter {
             options.compress,
             injections,
             archive_name,
+            pxar_archive_progress_stats,
+            options.is_metadata_mode,
         )
         .await?;
 
@@ -770,8 +777,10 @@ impl BackupWriter {
         compress: bool,
         injections: Option<std::sync::mpsc::Receiver<InjectChunks>>,
         archive: &BackupArchiveName,
+        pxar_archive_progress_stats: Option<Arc<PxarArchiverProgressStats>>,
+        is_metadata_mode: bool,
     ) -> impl Future<Output = Result<UploadStats, Error>> {
-        let mut counters = UploadCounters::new();
+        let mut counters = UploadCounters::new(pxar_archive_progress_stats);
         let counters_readonly = counters.clone();
 
         let is_fixed_chunk_size = prefix == "fixed";
@@ -855,6 +864,7 @@ impl BackupWriter {
             stream,
             index_csum_2,
             counters_readonly,
+            is_metadata_mode,
         )
     }
 
@@ -866,6 +876,7 @@ impl BackupWriter {
         stream: impl Stream<Item = Result<MergedChunkInfo, Error>>,
         index_csum: Arc<Mutex<Option<Sha256>>>,
         counters: UploadCounters,
+        is_metadata_mode: bool,
     ) -> impl Future<Output = Result<UploadStats, Error>> {
         let append_chunk_path = format!("{prefix}_index");
         let upload_chunk_path = format!("{prefix}_chunk");
@@ -889,7 +900,29 @@ impl BackupWriter {
                     let size_uploaded = HumanByte::from(uploaded_len.load(Ordering::SeqCst));
                     let elapsed = TimeSpan::from(start_time.elapsed());
 
-                    info!("processed {size} in {elapsed}, uploaded {size_uploaded}");
+                    if let Some(pxar_progress) = &counters.pxar_progress {
+                        let n_files = pxar_progress.total_files_count();
+                        let reuse_payload_size = pxar_progress.total_reused_payload_size();
+                        let total_of_logical_file_size = HumanByte::from(
+                            pxar_progress.total_reencoded_size() + reuse_payload_size,
+                        );
+                        let total_reuse_payload_size = HumanByte::from(reuse_payload_size);
+                        let reuse_file_count = pxar_progress.files_reused_count();
+
+                        if is_metadata_mode {
+                            info!(
+                                "scanned {n_files} files with {total_of_logical_file_size} (reusing {reuse_file_count} files with \
+                                {total_reuse_payload_size}) of which processed {size} in {elapsed} and uploaded {size_uploaded}"
+                            );
+                        } else {
+                            info!(
+                                "scanned {n_files} files with {total_of_logical_file_size} of which processed {size} \
+                                in {elapsed} and uploaded {size_uploaded}"
+                            );
+                        }
+                    } else {
+                        info!("processed {size} in {elapsed}, uploaded {size_uploaded}");
+                    }
                 }
             }))
         } else {
diff --git a/pbs-client/src/pxar/create.rs b/pbs-client/src/pxar/create.rs
index 1246f67d9..2d2dfbb3f 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,
@@ -223,7 +223,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,
 }
 
@@ -251,6 +251,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,
@@ -334,7 +335,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,
     };
 
@@ -2133,7 +2134,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 2680add78..1036ec7e9 100644
--- a/proxmox-backup-client/src/main.rs
+++ b/proxmox-backup-client/src/main.rs
@@ -31,7 +31,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,
@@ -268,6 +270,7 @@ async fn backup_directory<P: AsRef<Path>>(
         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(),
@@ -275,6 +278,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);
@@ -289,7 +293,13 @@ async fn backup_directory<P: AsRef<Path>>(
         }
     });
 
-    let stats = client.upload_stream(archive_name, stream, upload_options.clone(), None);
+    let stats = client.upload_stream(
+        archive_name,
+        stream,
+        upload_options.clone(),
+        None,
+        Some(pxar_archiver_progress_stats.clone()),
+    );
 
     if let Some(mut payload_stream) = payload_stream {
         let payload_target = payload_target
@@ -319,6 +329,7 @@ async fn backup_directory<P: AsRef<Path>>(
             stream,
             upload_options,
             Some(payload_injections_rx),
+            Some(pxar_archiver_progress_stats),
         );
 
         match futures::join!(stats, payload_stats) {
@@ -359,7 +370,7 @@ async fn backup_image<P: AsRef<Path>>(
     }
 
     let stats = client
-        .upload_stream(archive_name, stream, upload_options, None)
+        .upload_stream(archive_name, stream, upload_options, None, None)
         .await?;
 
     Ok(stats)
@@ -652,7 +663,13 @@ fn spawn_catalog_upload(
 
     tokio::spawn(async move {
         let catalog_upload_result = client
-            .upload_stream(&CATALOG_NAME, catalog_chunk_stream, upload_options, None)
+            .upload_stream(
+                &CATALOG_NAME,
+                catalog_chunk_stream,
+                upload_options,
+                None,
+                None,
+            )
             .await;
 
         if let Err(ref err) = catalog_upload_result {
@@ -1234,6 +1251,7 @@ async fn create_backup(
                     previous_manifest: previous_manifest.clone(),
                     compress: true,
                     encrypt: crypto.mode == CryptMode::Encrypt,
+                    is_metadata_mode: detection_mode.is_metadata(),
                     ..UploadOptions::default()
                 };
 
@@ -1273,6 +1291,7 @@ async fn create_backup(
                     index_type: IndexType::Fixed(image_file_size),
                     compress: true,
                     encrypt: crypto.mode == CryptMode::Encrypt,
+                    is_metadata_mode: false,
                 };
 
                 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