[pbs-devel] [PATCH proxmox-backup 0/8] improve syncjob handling

[pbs-devel] [PATCH proxmox-backup 0/8] improve syncjob handling

[Dominik Csapak]

by saving the state in a separate file instead of parsing the task log
should be faster and more correct, since we cannot phase out
the last sync if too many tasks are started

we have to do the same for all other tasks that can be scheduled

Dominik Csapak (8):
  worker_task: refactor log text generator
  worker_task: add getter for upid
  config: add JobState helper
  api2/pull: add do_syn_job helper
  syncjob: use do_sync_job for manual and scheduled sync jobs
  syncjob: use JobState for determining when to run next scheduled sync
  api2/admin/sync: use JobState for faster access to state info
  ui: syncjob: use the Task text directly

 src/api2/admin/sync.rs          |  57 ++++-----------
 src/api2/config/sync.rs         |   2 +
 src/api2/pull.rs                |  70 ++++++++++++++++++
 src/bin/proxmox-backup-api.rs   |   1 +
 src/bin/proxmox-backup-proxy.rs |  98 +++++++------------------
 src/config.rs                   |   1 +
 src/config/jobstate.rs          | 126 ++++++++++++++++++++++++++++++++
 src/server/worker_task.rs       |  20 +++--
 www/config/SyncView.js          |   2 +-
 9 files changed, 255 insertions(+), 122 deletions(-)
 create mode 100644 src/config/jobstate.rs

-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 1/8] worker_task: refactor log text generator

[Dominik Csapak]

we will need this elsewhere, so pull it out

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/server/worker_task.rs | 16 +++++++++++-----
 1 file changed, 11 insertions(+), 5 deletions(-)

diff --git a/src/server/worker_task.rs b/src/server/worker_task.rs
index af6686fd..f6882d13 100644
--- a/src/server/worker_task.rs
+++ b/src/server/worker_task.rs
@@ -502,17 +502,23 @@ impl WorkerTask {
         Ok(upid_str)
     }
 
-    /// Log task result, remove task from running list
-    pub fn log_result(&self, result: &Result<(), Error>) {
+    /// get the Text of the result
+    pub fn get_log_text(&self, result: &Result<(), Error>) -> String {
 
         let warn_count = self.data.lock().unwrap().warn_count;
+
         if let Err(err) = result {
-            self.log(&format!("TASK ERROR: {}", err));
+            format!("ERROR: {}", err)
         } else if warn_count > 0 {
-            self.log(format!("TASK WARNINGS: {}", warn_count));
+            format!("WARNINGS: {}", warn_count)
         } else {
-            self.log("TASK OK");
+            "OK".to_string()
         }
+    }
+
+    /// Log task result, remove task from running list
+    pub fn log_result(&self, result: &Result<(), Error>) {
+        self.log(format!("TASK {}", self.get_log_text(result)));
 
         WORKER_TASK_LIST.lock().unwrap().remove(&self.upid.task_id);
         let _ = update_active_workers(None);
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 2/8] worker_task: add getter for upid

[Dominik Csapak]

sometimes we need the upid inside the worker itself, so give a
possibilty to get it

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/server/worker_task.rs | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/src/server/worker_task.rs b/src/server/worker_task.rs
index f6882d13..fdbbacbe 100644
--- a/src/server/worker_task.rs
+++ b/src/server/worker_task.rs
@@ -589,4 +589,8 @@ impl WorkerTask {
         }
         rx
     }
+
+    pub fn upid(&self) -> &UPID {
+        &self.upid
+    }
 }
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 3/8] config: add JobState helper

[Dominik Csapak]

this is intended to be a generic helper to (de)serialize job states
(e.g., sync, verify, and so on)

writes a json file into '/var/lib/proxmox-backup/jobstates/TYPE-ID.json'

the api creates the directory with the correct permissions, like
the rrd directory

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/bin/proxmox-backup-api.rs |   1 +
 src/config.rs                 |   1 +
 src/config/jobstate.rs        | 126 ++++++++++++++++++++++++++++++++++
 3 files changed, 128 insertions(+)
 create mode 100644 src/config/jobstate.rs

diff --git a/src/bin/proxmox-backup-api.rs b/src/bin/proxmox-backup-api.rs
index 9dde46c0..ea306cf0 100644
--- a/src/bin/proxmox-backup-api.rs
+++ b/src/bin/proxmox-backup-api.rs
@@ -37,6 +37,7 @@ async fn run() -> Result<(), Error> {
     config::update_self_signed_cert(false)?;
 
     proxmox_backup::rrd::create_rrdb_dir()?;
+    proxmox_backup::config::jobstate::create_jobstate_dir()?;
 
     if let Err(err) = generate_auth_key() {
         bail!("unable to generate auth key - {}", err);
diff --git a/src/config.rs b/src/config.rs
index e805c38e..f9e08814 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -22,6 +22,7 @@ pub mod acl;
 pub mod cached_user_info;
 pub mod network;
 pub mod sync;
+pub mod jobstate;
 
 /// Check configuration directory permissions
 ///
diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
new file mode 100644
index 00000000..fb5a8cbf
--- /dev/null
+++ b/src/config/jobstate.rs
@@ -0,0 +1,126 @@
+use std::fs::File;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use serde::{Serialize, Deserialize};
+use anyhow::{bail, Error, format_err};
+use proxmox::tools::fs::{file_read_optional_string, replace_file, create_path, CreateOptions};
+
+use crate::tools::{epoch_now_u64, open_file_locked};
+
+#[serde(rename_all="kebab-case")]
+#[derive(Serialize,Deserialize)]
+pub struct JobState {
+    #[serde(skip_serializing_if="Option::is_none")]
+    pub upid: Option<String>,
+    pub starttime: i64,
+    #[serde(skip_serializing_if="Option::is_none")]
+    pub endtime: Option<i64>,
+    #[serde(skip_serializing_if="Option::is_none")]
+    pub state: Option<String>,
+}
+
+const JOB_STATE_BASEDIR: &str = "/var/lib/proxmox-backup/jobstates";
+
+/// Create jobstate stat dir with correct permission
+pub fn create_jobstate_dir() -> Result<(), Error> {
+
+    let backup_user = crate::backup::backup_user()?;
+    let opts = CreateOptions::new()
+        .owner(backup_user.uid)
+        .group(backup_user.gid);
+
+    create_path(JOB_STATE_BASEDIR, None, Some(opts))
+        .map_err(|err: Error| format_err!("unable to create rrdb stat dir - {}", err))?;
+
+    Ok(())
+}
+
+fn get_path(jobtype: &str, jobname: &str) -> Result<PathBuf, Error> {
+    let mut path = PathBuf::from(JOB_STATE_BASEDIR);
+    path.push(format!("{}-{}.json", jobtype, jobname));
+    Ok(path)
+}
+
+fn get_lock<P>(path: P) -> Result<File, Error>
+where
+    P: AsRef<Path>
+{
+    let mut path = path.as_ref().to_path_buf();
+    path.set_extension("lck");
+    let lock = open_file_locked(path, Duration::new(10, 0))?;
+    Ok(lock)
+}
+
+pub fn remove_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
+    let path = get_path(jobtype, jobname)?;
+    let _lock = get_lock(&path)?;
+    match std::fs::remove_file(&path) {
+        Ok(()) => Ok(()),
+        Err(err) => {
+            bail!("cannot remove statefile for {} - {}: {}", jobtype, jobname, err);
+        }
+    }
+}
+
+impl JobState {
+    pub fn new(upid: &str) -> Self {
+        Self{
+            upid: Some(upid.to_string()),
+            starttime: epoch_now_u64().unwrap_or_else(|_| 0) as i64,
+            endtime: None,
+            state: None,
+        }
+    }
+
+    pub fn finish(&mut self, state: &str) -> Result<(), Error> {
+        self.state = Some(state.to_string());
+        self.endtime = Some(epoch_now_u64()? as i64);
+        Ok(())
+    }
+
+    pub fn try_read_or_create(jobtype: &str, jobname: &str) -> Result<Self, Error> {
+        let path = get_path(jobtype, jobname)?;
+
+        let lock = get_lock(&path)?;
+
+        if let Some(state) = file_read_optional_string(path)? {
+            Ok(serde_json::from_str(&state)?)
+        } else {
+            let state = Self{
+                upid: None,
+                endtime: None,
+                state: None,
+                starttime: epoch_now_u64()? as i64,
+            };
+            state.write_state(jobtype, jobname, Some(lock))?;
+            Ok(state)
+        }
+    }
+
+    pub fn write_state(&self, jobtype: &str, jobname: &str, lock: Option<File>) -> Result<(), Error> {
+        let serialized = serde_json::to_string(&self)?;
+        let path = get_path(jobtype, jobname)?;
+
+        let _lock = if let Some(lock) = lock {
+            lock
+        } else {
+            get_lock(&path)?
+        };
+
+        let backup_user = crate::backup::backup_user()?;
+        let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
+        // set the correct owner/group/permissions while saving file
+        // owner(rw) = backup, group(r)= backup
+        let options = CreateOptions::new()
+            .perm(mode)
+            .owner(backup_user.uid)
+            .group(backup_user.gid);
+
+        replace_file(
+            path,
+            serialized.as_bytes(),
+            options,
+        )
+    }
+}
-- 
2.20.1

Re: [pbs-devel] [PATCH proxmox-backup 3/8] config: add JobState helper

[Dietmar Maurer]

comments inline

> On 07/31/2020 2:43 PM Dominik Csapak <d.csapak@proxmox.com> wrote:
> 
>  
> this is intended to be a generic helper to (de)serialize job states
> (e.g., sync, verify, and so on)
> 
> writes a json file into '/var/lib/proxmox-backup/jobstates/TYPE-ID.json'
> 
> the api creates the directory with the correct permissions, like
> the rrd directory
> 
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
>  src/bin/proxmox-backup-api.rs |   1 +
>  src/config.rs                 |   1 +
>  src/config/jobstate.rs        | 126 ++++++++++++++++++++++++++++++++++
>  3 files changed, 128 insertions(+)
>  create mode 100644 src/config/jobstate.rs
> 
> diff --git a/src/bin/proxmox-backup-api.rs b/src/bin/proxmox-backup-api.rs
> index 9dde46c0..ea306cf0 100644
> --- a/src/bin/proxmox-backup-api.rs
> +++ b/src/bin/proxmox-backup-api.rs
> @@ -37,6 +37,7 @@ async fn run() -> Result<(), Error> {
>      config::update_self_signed_cert(false)?;
>  
>      proxmox_backup::rrd::create_rrdb_dir()?;
> +    proxmox_backup::config::jobstate::create_jobstate_dir()?;
>  
>      if let Err(err) = generate_auth_key() {
>          bail!("unable to generate auth key - {}", err);
> diff --git a/src/config.rs b/src/config.rs
> index e805c38e..f9e08814 100644
> --- a/src/config.rs
> +++ b/src/config.rs
> @@ -22,6 +22,7 @@ pub mod acl;
>  pub mod cached_user_info;
>  pub mod network;
>  pub mod sync;
> +pub mod jobstate;
>  
>  /// Check configuration directory permissions
>  ///
> diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
> new file mode 100644
> index 00000000..fb5a8cbf
> --- /dev/null
> +++ b/src/config/jobstate.rs
> @@ -0,0 +1,126 @@
> +use std::fs::File;
> +use std::path::{Path, PathBuf};
> +use std::time::Duration;
> +
> +use serde::{Serialize, Deserialize};
> +use anyhow::{bail, Error, format_err};
> +use proxmox::tools::fs::{file_read_optional_string, replace_file, create_path, CreateOptions};
> +
> +use crate::tools::{epoch_now_u64, open_file_locked};
> +
> +#[serde(rename_all="kebab-case")]
> +#[derive(Serialize,Deserialize)]
> +pub struct JobState {
> +    #[serde(skip_serializing_if="Option::is_none")]
> +    pub upid: Option<String>,
> +    pub starttime: i64,


Why is upid optional, and starttime not? Also, Upid contains the startime, so
why do we store twice?

Also, please add documentation to pub structs and functions.

> +    #[serde(skip_serializing_if="Option::is_none")]
> +    pub endtime: Option<i64>,
> +    #[serde(skip_serializing_if="Option::is_none")]
> +    pub state: Option<String>,
> +}
> +
> +const JOB_STATE_BASEDIR: &str = "/var/lib/proxmox-backup/jobstates";
> +
> +/// Create jobstate stat dir with correct permission
> +pub fn create_jobstate_dir() -> Result<(), Error> {
> +
> +    let backup_user = crate::backup::backup_user()?;
> +    let opts = CreateOptions::new()
> +        .owner(backup_user.uid)
> +        .group(backup_user.gid);
> +
> +    create_path(JOB_STATE_BASEDIR, None, Some(opts))
> +        .map_err(|err: Error| format_err!("unable to create rrdb stat dir - {}", err))?;
> +
> +    Ok(())
> +}
> +
> +fn get_path(jobtype: &str, jobname: &str) -> Result<PathBuf, Error> {
> +    let mut path = PathBuf::from(JOB_STATE_BASEDIR);
> +    path.push(format!("{}-{}.json", jobtype, jobname));
> +    Ok(path)
> +}
> +
> +fn get_lock<P>(path: P) -> Result<File, Error>
> +where
> +    P: AsRef<Path>
> +{
> +    let mut path = path.as_ref().to_path_buf();
> +    path.set_extension("lck");
> +    let lock = open_file_locked(path, Duration::new(10, 0))?;
> +    Ok(lock)
> +}
> +
> +pub fn remove_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
> +    let path = get_path(jobtype, jobname)?;
> +    let _lock = get_lock(&path)?;
> +    match std::fs::remove_file(&path) {
> +        Ok(()) => Ok(()),
> +        Err(err) => {
> +            bail!("cannot remove statefile for {} - {}: {}", jobtype, jobname, err);
> +        }
> +    }
> +}
> +
> +impl JobState {
> +    pub fn new(upid: &str) -> Self {
> +        Self{
> +            upid: Some(upid.to_string()),
> +            starttime: epoch_now_u64().unwrap_or_else(|_| 0) as i64,

upid already contains a start time?

> +            endtime: None,
> +            state: None,
> +        }
> +    }
> +
> +    pub fn finish(&mut self, state: &str) -> Result<(), Error> {
> +        self.state = Some(state.to_string());
> +        self.endtime = Some(epoch_now_u64()? as i64);
> +        Ok(())
> +    }
> +
> +    pub fn try_read_or_create(jobtype: &str, jobname: &str) -> Result<Self, Error> {
> +        let path = get_path(jobtype, jobname)?;
> +
> +        let lock = get_lock(&path)?;
> +
> +        if let Some(state) = file_read_optional_string(path)? {
> +            Ok(serde_json::from_str(&state)?)
> +        } else {
> +            let state = Self{
> +                upid: None,
> +                endtime: None,
> +                state: None,
> +                starttime: epoch_now_u64()? as i64,
> +            };

why do we create a file with no info? Or is that now() relevant?

> +            state.write_state(jobtype, jobname, Some(lock))?;
> +            Ok(state)
> +        }
> +    }
> +
> +    pub fn write_state(&self, jobtype: &str, jobname: &str, lock: Option<File>) -> Result<(), Error> {

This is a strange pub API. I would remove parameter lock.

   fn write_state_no_lock(&self, path: &Path) -> Result<(), Error> {
        let serialized = serde_json::to_string(&self)?;
 
        let backup_user = crate::backup::backup_user()?;
        let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
        // set the correct owner/group/permissions while saving file
        // owner(rw) = backup, group(r)= backup
        let options = CreateOptions::new()
            .perm(mode)
            .owner(backup_user.uid)
            .group(backup_user.gid);

        replace_file(
            path,
            serialized.as_bytes(),
            options,
        )
    }

    pub fn write_state(&self, jobtype: &str, jobname: &str) -> Result<(), Error> {
        let path = get_path(jobtype, jobname)?;

        let _lock = get_lock(&path)?;

        self.write_state_no_lock(&path)
    }

> +        let serialized = serde_json::to_string(&self)?;
> +        let path = get_path(jobtype, jobname)?;
> +
> +        let _lock = if let Some(lock) = lock {
> +            lock
> +        } else {
> +            get_lock(&path)?
> +        };
> +
> +        let backup_user = crate::backup::backup_user()?;
> +        let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
> +        // set the correct owner/group/permissions while saving file
> +        // owner(rw) = backup, group(r)= backup
> +        let options = CreateOptions::new()
> +            .perm(mode)
> +            .owner(backup_user.uid)
> +            .group(backup_user.gid);
> +
> +        replace_file(
> +            path,
> +            serialized.as_bytes(),
> +            options,
> +        )
> +    }
> +}
> -- 
> 2.20.1
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

Re: [pbs-devel] [PATCH proxmox-backup 3/8] config: add JobState helper

[Dietmar Maurer]

suggested cleanups:

diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
index fb5a8cb..974b8e9 100644
--- a/src/config/jobstate.rs
+++ b/src/config/jobstate.rs
@@ -3,7 +3,7 @@ use std::path::{Path, PathBuf};
 use std::time::Duration;
 
 use serde::{Serialize, Deserialize};
-use anyhow::{bail, Error, format_err};
+use anyhow::{Error, format_err};
 use proxmox::tools::fs::{file_read_optional_string, replace_file, create_path, CreateOptions};
 
 use crate::tools::{epoch_now_u64, open_file_locked};
@@ -48,19 +48,15 @@ where
 {
     let mut path = path.as_ref().to_path_buf();
     path.set_extension("lck");
-    let lock = open_file_locked(path, Duration::new(10, 0))?;
-    Ok(lock)
+    open_file_locked(path, Duration::new(10, 0))
 }
 
 pub fn remove_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
     let path = get_path(jobtype, jobname)?;
     let _lock = get_lock(&path)?;
-    match std::fs::remove_file(&path) {
-        Ok(()) => Ok(()),
-        Err(err) => {
-            bail!("cannot remove statefile for {} - {}: {}", jobtype, jobname, err);
-        }
-    }
+    std::fs::remove_file(&path).map_err(|err| {
+        format_err!("cannot remove statefile for {} - {}: {}", jobtype, jobname, err)
+    })
 }

 impl JobState {

[pbs-devel] [PATCH proxmox-backup 4/8] api2/pull: add do_syn_job helper

[Dominik Csapak]

this helper is inspired by our code in proxmox-backup-proxy
with an added JobState writing

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/pull.rs | 70 ++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 70 insertions(+)

diff --git a/src/api2/pull.rs b/src/api2/pull.rs
index cf7de524..e788d20f 100644
--- a/src/api2/pull.rs
+++ b/src/api2/pull.rs
@@ -12,6 +12,8 @@ use crate::client::{HttpClient, HttpClientOptions, BackupRepository, pull::pull_
 use crate::api2::types::*;
 use crate::config::{
     remote,
+    sync::SyncJobConfig,
+    jobstate::JobState,
     acl::{PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ},
     cached_user_info::CachedUserInfo,
 };
@@ -62,6 +64,74 @@ pub async fn get_pull_parameters(
     Ok((client, src_repo, tgt_store))
 }
 
+pub fn do_sync_job(
+    id: &str,
+    sync_job: SyncJobConfig,
+    userid: &str,
+    schedule: Option<String>,
+) -> Result<String, Error> {
+
+    let job_id = id.to_string();
+    let job_id2 = id.to_string();
+    let worker_type = "syncjob";
+    let upid_str = WorkerTask::spawn(
+        worker_type,
+        Some(id.to_string()),
+        &userid.clone(),
+        false,
+        move |worker| async move {
+
+            let mut state = JobState::new(&worker.upid().to_string());
+            match state.write_state(worker_type, &job_id2, None) {
+                Ok(_) => {},
+                Err(err) => {
+                    eprintln!("could not write job state on start: {}", err);
+                }
+            }
+
+            let worker2 = worker.clone();
+
+            let res = async move {
+
+                let delete = sync_job.remove_vanished.unwrap_or(true);
+                let (client, src_repo, tgt_store) = get_pull_parameters(&sync_job.store, &sync_job.remote, &sync_job.remote_store).await?;
+
+                worker.log(format!("Starting datastore sync job '{}'", job_id));
+                if let Some(event_str) = schedule {
+                    worker.log(format!("task triggered by schedule '{}'", event_str));
+                }
+                worker.log(format!("Sync datastore '{}' from '{}/{}'",
+                        sync_job.store, sync_job.remote, sync_job.remote_store));
+
+                crate::client::pull::pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, String::from("backup@pam")).await?;
+
+                worker.log(format!("sync job '{}' end", &job_id));
+
+                Ok(())
+            }.await;
+
+            let status = worker2.get_log_text(&res);
+
+            match state.finish(&status) {
+                Ok(_) => {},
+                Err(err) => {
+                    eprintln!("could not finish job state: {}", err);
+                }
+            }
+
+            match state.write_state(&worker_type, &job_id2, None) {
+                Ok(_) => {},
+                Err(err) => {
+                    eprintln!("could not write job state on finish: {}", err);
+                }
+            }
+
+            res
+        })?;
+
+    Ok(upid_str)
+}
+
 #[api(
     input: {
         properties: {
-- 
2.20.1

Re: [pbs-devel] [PATCH proxmox-backup 4/8] api2/pull: add do_syn_job helper

[Dietmar Maurer]

Please can we first have a patch to split out do_syn_job from src/api2/admin/sync.rs?
 
> On 07/31/2020 2:43 PM Dominik Csapak <d.csapak@proxmox.com> wrote:
> 
>  
> this helper is inspired by our code in proxmox-backup-proxy
> with an added JobState writing
> 
> Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
> ---
>  src/api2/pull.rs | 70 ++++++++++++++++++++++++++++++++++++++++++++++++
>  1 file changed, 70 insertions(+)
> 
> diff --git a/src/api2/pull.rs b/src/api2/pull.rs
> index cf7de524..e788d20f 100644
> --- a/src/api2/pull.rs
> +++ b/src/api2/pull.rs
> @@ -12,6 +12,8 @@ use crate::client::{HttpClient, HttpClientOptions, BackupRepository, pull::pull_
>  use crate::api2::types::*;
>  use crate::config::{
>      remote,
> +    sync::SyncJobConfig,
> +    jobstate::JobState,
>      acl::{PRIV_DATASTORE_BACKUP, PRIV_DATASTORE_PRUNE, PRIV_REMOTE_READ},
>      cached_user_info::CachedUserInfo,
>  };
> @@ -62,6 +64,74 @@ pub async fn get_pull_parameters(
>      Ok((client, src_repo, tgt_store))
>  }
>  
> +pub fn do_sync_job(
> +    id: &str,
> +    sync_job: SyncJobConfig,
> +    userid: &str,
> +    schedule: Option<String>,
> +) -> Result<String, Error> {
> +
> +    let job_id = id.to_string();
> +    let job_id2 = id.to_string();
> +    let worker_type = "syncjob";
> +    let upid_str = WorkerTask::spawn(
> +        worker_type,
> +        Some(id.to_string()),
> +        &userid.clone(),
> +        false,
> +        move |worker| async move {
> +
> +            let mut state = JobState::new(&worker.upid().to_string());
> +            match state.write_state(worker_type, &job_id2, None) {
> +                Ok(_) => {},
> +                Err(err) => {
> +                    eprintln!("could not write job state on start: {}", err);
> +                }
> +            }
> +
> +            let worker2 = worker.clone();
> +
> +            let res = async move {
> +
> +                let delete = sync_job.remove_vanished.unwrap_or(true);
> +                let (client, src_repo, tgt_store) = get_pull_parameters(&sync_job.store, &sync_job.remote, &sync_job.remote_store).await?;
> +
> +                worker.log(format!("Starting datastore sync job '{}'", job_id));
> +                if let Some(event_str) = schedule {
> +                    worker.log(format!("task triggered by schedule '{}'", event_str));
> +                }
> +                worker.log(format!("Sync datastore '{}' from '{}/{}'",
> +                        sync_job.store, sync_job.remote, sync_job.remote_store));
> +
> +                crate::client::pull::pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, String::from("backup@pam")).await?;
> +
> +                worker.log(format!("sync job '{}' end", &job_id));
> +
> +                Ok(())
> +            }.await;
> +
> +            let status = worker2.get_log_text(&res);
> +
> +            match state.finish(&status) {
> +                Ok(_) => {},
> +                Err(err) => {
> +                    eprintln!("could not finish job state: {}", err);
> +                }
> +            }
> +
> +            match state.write_state(&worker_type, &job_id2, None) {
> +                Ok(_) => {},
> +                Err(err) => {
> +                    eprintln!("could not write job state on finish: {}", err);
> +                }
> +            }
> +
> +            res
> +        })?;
> +
> +    Ok(upid_str)
> +}
> +
>  #[api(
>      input: {
>          properties: {
> -- 
> 2.20.1
> 
> 
> 
> _______________________________________________
> pbs-devel mailing list
> pbs-devel@lists.proxmox.com
> https://lists.proxmox.com/cgi-bin/mailman/listinfo/pbs-devel

[pbs-devel] [PATCH proxmox-backup 5/8] syncjob: use do_sync_job for manual and scheduled sync jobs

[Dominik Csapak]

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/admin/sync.rs          | 18 ++-------
 src/bin/proxmox-backup-proxy.rs | 68 +++------------------------------
 2 files changed, 8 insertions(+), 78 deletions(-)

diff --git a/src/api2/admin/sync.rs b/src/api2/admin/sync.rs
index ac2858f4..e81a7e9f 100644
--- a/src/api2/admin/sync.rs
+++ b/src/api2/admin/sync.rs
@@ -83,7 +83,7 @@ pub fn list_sync_jobs(
     }
 )]
 /// Runs the sync jobs manually.
-async fn run_sync_job(
+fn run_sync_job(
     id: String,
     _info: &ApiMethod,
     rpcenv: &mut dyn RpcEnvironment,
@@ -92,21 +92,9 @@ async fn run_sync_job(
     let (config, _digest) = sync::config()?;
     let sync_job: SyncJobConfig = config.lookup("sync", &id)?;
 
-    let username = rpcenv.get_user().unwrap();
+    let userid = rpcenv.get_user().unwrap();
 
-    let delete = sync_job.remove_vanished.unwrap_or(true);
-    let (client, src_repo, tgt_store) = get_pull_parameters(&sync_job.store, &sync_job.remote, &sync_job.remote_store).await?;
-
-    let upid_str = WorkerTask::spawn("syncjob", Some(id.clone()), &username.clone(), false, move |worker| async move {
-
-        worker.log(format!("sync job '{}' start", &id));
-
-        crate::client::pull::pull_store(&worker, &client, &src_repo, tgt_store.clone(), delete, String::from("backup@pam")).await?;
-
-        worker.log(format!("sync job '{}' end", &id));
-
-        Ok(())
-    })?;
+    let upid_str = do_sync_job(&id, sync_job, &userid, None)?;
 
     Ok(upid_str)
 }
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index dc0d16d2..eea322fc 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -17,6 +17,8 @@ use proxmox_backup::server::{ApiConfig, rest::*};
 use proxmox_backup::auth_helpers::*;
 use proxmox_backup::tools::disks::{ DiskManage, zfs_pool_stats };
 
+use proxmox_backup::api2::pull::do_sync_job;
+
 fn main() {
     proxmox_backup::tools::setup_safe_path_env();
 
@@ -471,10 +473,7 @@ async fn schedule_datastore_prune() {
 async fn schedule_datastore_sync_jobs() {
 
     use proxmox_backup::{
-        backup::DataStore,
-        client::{ HttpClient, HttpClientOptions, BackupRepository, pull::pull_store },
-        server::{ WorkerTask },
-        config::{ sync::{self, SyncJobConfig}, remote::{self, Remote} },
+        config::{ sync::{self, SyncJobConfig}, jobstate::JobState },
         tools::systemd::time::{ parse_calendar_event, compute_next_event },
     };
 
@@ -486,14 +485,6 @@ async fn schedule_datastore_sync_jobs() {
         Ok((config, _digest)) => config,
     };
 
-    let remote_config = match remote::config() {
-        Err(err) => {
-            eprintln!("unable to read remote config - {}", err);
-            return;
-        }
-        Ok((config, _digest)) => config,
-    };
-
     for (job_id, (_, job_config)) in config.sections {
         let job_config: SyncJobConfig = match serde_json::from_value(job_config) {
             Ok(c) => c,
@@ -549,57 +540,8 @@ async fn schedule_datastore_sync_jobs() {
         };
         if next > now  { continue; }
 
-
-        let job_id2 = job_id.clone();
-
-        let tgt_store = match DataStore::lookup_datastore(&job_config.store) {
-            Ok(datastore) => datastore,
-            Err(err) => {
-                eprintln!("lookup_datastore '{}' failed - {}", job_config.store, err);
-                continue;
-            }
-        };
-
-        let remote: Remote = match remote_config.lookup("remote", &job_config.remote) {
-            Ok(remote) => remote,
-            Err(err) => {
-                eprintln!("remote_config lookup failed: {}", err);
-                continue;
-            }
-        };
-
-        let username = String::from("backup@pam");
-
-        let delete = job_config.remove_vanished.unwrap_or(true);
-
-        if let Err(err) = WorkerTask::spawn(
-            worker_type,
-            Some(job_id.clone()),
-            &username.clone(),
-            false,
-            move |worker| async move {
-                worker.log(format!("Starting datastore sync job '{}'", job_id));
-                worker.log(format!("task triggered by schedule '{}'", event_str));
-                worker.log(format!("Sync datastore '{}' from '{}/{}'",
-                                   job_config.store, job_config.remote, job_config.remote_store));
-
-                let options = HttpClientOptions::new()
-                    .password(Some(remote.password.clone()))
-                    .fingerprint(remote.fingerprint.clone());
-
-                let client = HttpClient::new(&remote.host, &remote.userid, options)?;
-                let _auth_info = client.login() // make sure we can auth
-                    .await
-                    .map_err(|err| format_err!("remote connection to '{}' failed - {}", remote.host, err))?;
-
-                let src_repo = BackupRepository::new(Some(remote.userid), Some(remote.host), job_config.remote_store);
-
-                pull_store(&worker, &client, &src_repo, tgt_store, delete, username).await?;
-
-                Ok(())
-            }
-        ) {
-            eprintln!("unable to start datastore sync job {} - {}", job_id2, err);
+        if let Err(err) = do_sync_job(&job_id, job_config, "backup@pam", Some(event_str)) {
+            eprintln!("unable to start datastore sync job {} - {}", &job_id, err);
         }
     }
 }
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 6/8] syncjob: use JobState for determining when to run next scheduled sync

[Dominik Csapak]

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/bin/proxmox-backup-proxy.rs | 30 +++++++++++++++++++++---------
 1 file changed, 21 insertions(+), 9 deletions(-)

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index eea322fc..2103dfef 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -509,20 +509,32 @@ async fn schedule_datastore_sync_jobs() {
 
         let worker_type = "syncjob";
 
-        let last = match lookup_last_worker(worker_type, &job_id) {
-            Ok(Some(upid)) => {
-                if proxmox_backup::server::worker_is_active_local(&upid) {
-                    continue;
-                }
-                upid.starttime
-            },
-            Ok(None) => 0,
+        let laststate = match JobState::try_read_or_create(worker_type, &job_id) {
+            Ok(state) => state,
             Err(err) => {
-                eprintln!("lookup_last_job_start failed: {}", err);
+                eprintln!("could not load last state: {}", err);
                 continue;
             }
         };
 
+        match laststate.upid {
+            Some(upid) => {
+                let upid: crate::server::UPID = match upid.parse() {
+                    Ok(upid) => upid,
+                    Err(err) => {
+                        eprintln!("could not parse upid from state: {}", err);
+                        continue;
+                    }
+                };
+                if proxmox_backup::server::worker_is_active_local(&upid) {
+                    continue;
+                }
+            }
+            None => {},
+        }
+
+        let last = laststate.starttime as i64;
+
         let next = match compute_next_event(&event, last, false) {
             Ok(next) => next,
             Err(err) => {
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 7/8] api2/admin/sync: use JobState for faster access to state info

[Dominik Csapak]

and delete the statefile again on syncjob removal

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/admin/sync.rs  | 39 ++++++++++-----------------------------
 src/api2/config/sync.rs |  2 ++
 2 files changed, 12 insertions(+), 29 deletions(-)

diff --git a/src/api2/admin/sync.rs b/src/api2/admin/sync.rs
index e81a7e9f..d432489d 100644
--- a/src/api2/admin/sync.rs
+++ b/src/api2/admin/sync.rs
@@ -1,15 +1,14 @@
 use anyhow::{Error};
 use serde_json::Value;
-use std::collections::HashMap;
 
 use proxmox::api::{api, ApiMethod, Router, RpcEnvironment};
 use proxmox::api::router::SubdirMap;
 use proxmox::{list_subdirs_api_method, sortable};
 
 use crate::api2::types::*;
-use crate::api2::pull::{get_pull_parameters};
+use crate::api2::pull::do_sync_job;
 use crate::config::sync::{self, SyncJobStatus, SyncJobConfig};
-use crate::server::{self, TaskListInfo, WorkerTask};
+use crate::config::jobstate::JobState;
 use crate::tools::systemd::time::{
     parse_calendar_event, compute_next_event};
 
@@ -33,33 +32,15 @@ pub fn list_sync_jobs(
 
     let mut list: Vec<SyncJobStatus> = config.convert_to_typed_array("sync")?;
 
-    let mut last_tasks: HashMap<String, &TaskListInfo> = HashMap::new();
-    let tasks = server::read_task_list()?;
-
-    for info in tasks.iter() {
-        let worker_id = match &info.upid.worker_id {
-            Some(id) => id,
-            _ => { continue; },
-        };
-        if let Some(last) = last_tasks.get(worker_id) {
-            if last.upid.starttime < info.upid.starttime {
-                last_tasks.insert(worker_id.to_string(), &info);
-            }
-        } else {
-            last_tasks.insert(worker_id.to_string(), &info);
-        }
-    }
-
     for job in &mut list {
-        let mut last = 0;
-        if let Some(task) = last_tasks.get(&job.id) {
-            job.last_run_upid = Some(task.upid_str.clone());
-            if let Some((endtime, status)) = &task.state {
-                job.last_run_state = Some(String::from(status));
-                job.last_run_endtime = Some(*endtime);
-                last = *endtime;
-            }
-        }
+        let last_state = JobState::try_read_or_create("syncjob", &job.id)?;
+
+        job.last_run_upid = last_state.upid;
+        job.last_run_state = last_state.state;
+        job.last_run_endtime = last_state.endtime;
+
+        let starttime = last_state.starttime;
+        let last = job.last_run_endtime.unwrap_or_else(|| starttime);
 
         job.next_run = (|| -> Option<i64> {
             let schedule = job.schedule.as_ref()?;
diff --git a/src/api2/config/sync.rs b/src/api2/config/sync.rs
index 860dcc36..ed1ccdad 100644
--- a/src/api2/config/sync.rs
+++ b/src/api2/config/sync.rs
@@ -263,6 +263,8 @@ pub fn delete_sync_job(id: String, digest: Option<String>) -> Result<(), Error>
 
     sync::save_config(&config)?;
 
+    crate::config::jobstate::remove_state_file("syncjob", &id)?;
+
     Ok(())
 }
 
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 8/8] ui: syncjob: use the Task text directly

[Dominik Csapak]

the text is here "ERROR: ..." now, so leave the 'Error' out

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 www/config/SyncView.js | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/www/config/SyncView.js b/www/config/SyncView.js
index 634977c4..fcdf57db 100644
--- a/www/config/SyncView.js
+++ b/www/config/SyncView.js
@@ -111,7 +111,7 @@ Ext.define('PBS.config.SyncJobView', {
 		return `<i class="fa fa-check good"></i> ${gettext("OK")}`;
 	    }
 
-	    return `<i class="fa fa-times critical"></i> ${gettext("Error")}:${value}`;
+	    return `<i class="fa fa-times critical"></i> ${value}`;
 	},
 
 	render_next_run: function(value, metadat, record) {
-- 
2.20.1