all lists on lists.proxmox.com
 help / color / mirror / Atom feed
* [PATCH proxmox-backup] sync: pull: use LogLineSender in d SyncSourceReader
@ 2026-04-24 13:35 Dominik Csapak
  2026-04-24 13:37 ` Dominik Csapak
  2026-04-26 21:40 ` applied: " Thomas Lamprecht
  0 siblings, 2 replies; 5+ messages in thread
From: Dominik Csapak @ 2026-04-24 13:35 UTC (permalink / raw)
  To: pbs-devel

otherwise some log messages are not rendered with the correct prefix
when using multiple worker threads.

For example:

Snapshot ct/122/2026-04-24T12:56:17Z: got backup log file client.log.blob

vs

[ct/122]: Snapshot ct/122/2026-04-24T13:00:02Z: got backup log file client.log.blob

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
there may be other places where we don't use the LogLineSender here,
but could not find any on first glance.

 src/server/pull.rs | 10 ++++---
 src/server/sync.rs | 67 +++++++++++++++++++++++++++++++++++-----------
 2 files changed, 59 insertions(+), 18 deletions(-)

diff --git a/src/server/pull.rs b/src/server/pull.rs
index 7fa273edb..2dff2a9f8 100644
--- a/src/server/pull.rs
+++ b/src/server/pull.rs
@@ -450,7 +450,7 @@ async fn pull_single_archive<'a>(
         .await?;
 
     reader
-        .load_file_into(archive_name, &tmp_path)
+        .load_file_into(archive_name, &tmp_path, Arc::clone(&log_sender))
         .await
         .with_context(|| archive_prefix.clone())?;
 
@@ -664,7 +664,11 @@ async fn pull_snapshot<'a>(
     let mut tmp_manifest_name = manifest_name.clone();
     tmp_manifest_name.set_extension("tmp");
     let Some(tmp_manifest_blob) = reader
-        .load_file_into(MANIFEST_BLOB_NAME.as_ref(), &tmp_manifest_name)
+        .load_file_into(
+            MANIFEST_BLOB_NAME.as_ref(),
+            &tmp_manifest_name,
+            Arc::clone(&log_sender),
+        )
         .await
         .with_context(|| prefix.clone())?
     else {
@@ -676,7 +680,7 @@ async fn pull_snapshot<'a>(
     let fetch_log = async || {
         if !client_log_name.exists() {
             reader
-                .try_download_client_log(&client_log_name)
+                .try_download_client_log(&client_log_name, Arc::clone(&log_sender))
                 .await
                 .with_context(|| prefix.clone())?;
             if client_log_name.exists() {
diff --git a/src/server/sync.rs b/src/server/sync.rs
index d25180341..8acbaf673 100644
--- a/src/server/sync.rs
+++ b/src/server/sync.rs
@@ -101,10 +101,19 @@ pub(crate) trait SyncSourceReader: Send + Sync {
     /// Asynchronously loads a file from the source into a local file.
     /// `filename` is the name of the file to load from the source.
     /// `into` is the path of the local file to load the source file into.
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error>;
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error>;
 
     /// Tries to download the client log from the source and save it into a local file.
-    async fn try_download_client_log(&self, to_path: &Path) -> Result<(), Error>;
+    async fn try_download_client_log(
+        &self,
+        to_path: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error>;
 
     fn skip_chunk_sync(&self, target_store_name: &str) -> bool;
 }
@@ -137,7 +146,12 @@ impl SyncSourceReader for RemoteSourceReader {
         Ok(Arc::new(chunk_reader))
     }
 
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error> {
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error> {
         let mut tmp_file = std::fs::OpenOptions::new()
             .write(true)
             .create(true)
@@ -149,10 +163,15 @@ impl SyncSourceReader for RemoteSourceReader {
             match err.downcast_ref::<HttpError>() {
                 Some(HttpError { code, message }) => match *code {
                     StatusCode::NOT_FOUND => {
-                        info!(
-                            "Snapshot {}: skipped because vanished since start of sync",
-                            &self.dir
-                        );
+                        log_sender
+                            .log(
+                                Level::INFO,
+                                format!(
+                                    "Snapshot {}: skipped because vanished since start of sync",
+                                    &self.dir
+                                ),
+                            )
+                            .await?;
                         return Ok(None);
                     }
                     _ => {
@@ -168,7 +187,11 @@ impl SyncSourceReader for RemoteSourceReader {
         Ok(DataBlob::load_from_reader(&mut tmp_file).ok())
     }
 
-    async fn try_download_client_log(&self, to_path: &Path) -> Result<(), Error> {
+    async fn try_download_client_log(
+        &self,
+        to_path: &Path,
+        log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error> {
         let mut tmp_path = to_path.to_owned();
         tmp_path.set_extension("tmp");
 
@@ -189,11 +212,16 @@ impl SyncSourceReader for RemoteSourceReader {
             if let Err(err) = std::fs::rename(&tmp_path, to_path) {
                 bail!("Atomic rename file {to_path:?} failed - {err}");
             }
-            info!(
-                "Snapshot {snapshot}: got backup log file {client_log_name}",
-                snapshot = &self.dir,
-                client_log_name = client_log_name.deref()
-            );
+            log_sender
+                .log(
+                    Level::INFO,
+                    format!(
+                        "Snapshot {snapshot}: got backup log file {client_log_name}",
+                        snapshot = &self.dir,
+                        client_log_name = client_log_name.deref()
+                    ),
+                )
+                .await?;
         }
 
         Ok(())
@@ -215,7 +243,12 @@ impl SyncSourceReader for LocalSourceReader {
         Ok(Arc::new(chunk_reader))
     }
 
-    async fn load_file_into(&self, filename: &str, into: &Path) -> Result<Option<DataBlob>, Error> {
+    async fn load_file_into(
+        &self,
+        filename: &str,
+        into: &Path,
+        _log_sender: Arc<LogLineSender>,
+    ) -> Result<Option<DataBlob>, Error> {
         let mut tmp_file = std::fs::OpenOptions::new()
             .write(true)
             .create(true)
@@ -229,7 +262,11 @@ impl SyncSourceReader for LocalSourceReader {
         Ok(DataBlob::load_from_reader(&mut tmp_file).ok())
     }
 
-    async fn try_download_client_log(&self, _to_path: &Path) -> Result<(), Error> {
+    async fn try_download_client_log(
+        &self,
+        _to_path: &Path,
+        _log_sender: Arc<LogLineSender>,
+    ) -> Result<(), Error> {
         Ok(())
     }
 
-- 
2.47.3





^ permalink raw reply related	[flat|nested] 5+ messages in thread
end of thread, other threads:[~2026-04-27  7:56 UTC | newest]

Thread overview: 5+ messages (download: mbox.gz follow: Atom feed
-- links below jump to the message on this page --
2026-04-24 13:35 [PATCH proxmox-backup] sync: pull: use LogLineSender in d SyncSourceReader Dominik Csapak
2026-04-24 13:37 ` Dominik Csapak
2026-04-26 21:40 ` applied: " Thomas Lamprecht
2026-04-27  6:37   ` Christian Ebner
2026-04-27  7:56     ` Dominik Csapak

This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.
Service provided by Proxmox Server Solutions GmbH | Privacy | Legal