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

[Christian Ebner <c.ebner@proxmox.com>]

Unfortunately passing the pxar progress state along to the upload stream 
for logging via the UploadOptions does not lead to nice code here. I 
think this needs revisiting.

What if we refactor the progress logging logic to be passed to the 
upload stream via e.g. an optional callback method provided via the 
UploadOptions?

By this we would have a cleaner interface and no need to mix pxar 
archiver related counters with the upload counters AFAICT.


Maybe something along the lines of (untested and incomplete):

diff --git a/pbs-client/src/backup_writer.rs 
b/pbs-client/src/backup_writer.rs
index d0246689c..fe48a3f0e 100644
--- a/pbs-client/src/backup_writer.rs
+++ b/pbs-client/src/backup_writer.rs
@@ -48,6 +48,8 @@ impl Drop for BackupWriter {
      }
  }

+type ProgressLoggerCallback = Box<dyn Fn(&UploadCounters) + Send + Sync 
+ 'static>;
+
  /// Options for uploading blobs/streams to the server
  #[derive(Default, Clone)]
  pub struct UploadOptions {
@@ -55,7 +57,7 @@ pub struct UploadOptions {
      pub compress: bool,
      pub encrypt: bool,
      pub index_type: IndexType,
-    pub is_metadata_mode: bool,
+    pub(crate) progress_logging_callback: 
Option<Arc<ProgressLoggerCallback>>,
  }

  /// Index type for upload options.
@@ -375,12 +377,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.is_metadata_mode,
+            options.progress_logging_callback,
          )
          .await?;

@@ -482,9 +483,8 @@ impl BackupWriter {
              },
              options.compress,
              injections,
-            archive_name,
              pxar_archive_progress_stats,
-            options.is_metadata_mode,
+            options.progress_logging_callback,
          )
          .await?;

@@ -776,9 +776,8 @@ impl BackupWriter {
          crypt_config: Option<Arc<CryptConfig>>,
          compress: bool,
          injections: Option<std::sync::mpsc::Receiver<InjectChunks>>,
-        archive: &BackupArchiveName,
          pxar_archive_progress_stats: 
Option<Arc<PxarArchiverProgressStats>>,
-        is_metadata_mode: bool,
+        progress_logging_callback: Option<Arc<ProgressLoggerCallback>>,
      ) -> impl Future<Output = Result<UploadStats, Error>> {
          let mut counters = 
UploadCounters::new(pxar_archive_progress_stats);
          let counters_readonly = counters.clone();
@@ -859,24 +858,22 @@ impl BackupWriter {
          Self::upload_merged_chunk_stream(
              h2,
              wid,
-            archive,
              prefix,
              stream,
              index_csum_2,
              counters_readonly,
-            is_metadata_mode,
+            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,
-        is_metadata_mode: bool,
+        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");
@@ -887,42 +884,13 @@ impl BackupWriter {
          let (upload_queue, upload_result) =
              Self::append_chunk_queue(h2.clone(), wid, 
append_chunk_path, uploaded_len.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();
+            let callback = Arc::clone(&callback);
+
              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(uploaded_len.load(Ordering::SeqCst));
-                    let elapsed = TimeSpan::from(start_time.elapsed());
-
-                    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}");
-                    }
+                    callback(&counters);
                  }
              }))
          } else {

On 6/10/26 8:02 PM, Shan Shaji wrote:
> 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>>,

no need for this to be pub, the UploadCounters are pub(crate) only anyways.

>   }
>   
>   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,

nit: can be a simplified s/pxar_progress: 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")