[pbs-devel] [PATCH proxmox-backup 1/3] cleanup: merge endtime into TaskState

[pbs-devel] [PATCH proxmox-backup 1/3] cleanup: merge endtime into TaskState

[Dominik Csapak]

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/admin/sync.rs    |   4 +-
 src/api2/node/tasks.rs    |   7 +--
 src/api2/types/mod.rs     |   2 +-
 src/config/jobstate.rs    |  12 ++---
 src/server/worker_task.rs | 110 ++++++++++++++++++++++----------------
 5 files changed, 76 insertions(+), 59 deletions(-)

diff --git a/src/api2/admin/sync.rs b/src/api2/admin/sync.rs
index c09bea4f..aafd808f 100644
--- a/src/api2/admin/sync.rs
+++ b/src/api2/admin/sync.rs
@@ -42,9 +42,9 @@ pub fn list_sync_jobs(
                 let parsed_upid: UPID = upid.parse()?;
                 (Some(upid), None, None, parsed_upid.starttime)
             },
-            JobState::Finished { upid, endtime, state } => {
+            JobState::Finished { upid, state } => {
                 let parsed_upid: UPID = upid.parse()?;
-                (Some(upid), Some(endtime), Some(state.to_string()), parsed_upid.starttime)
+                (Some(upid), Some(state.endtime()), Some(state.to_string()), parsed_upid.starttime)
             },
         };
 
diff --git a/src/api2/node/tasks.rs b/src/api2/node/tasks.rs
index 1e9643ec..c8add6b4 100644
--- a/src/api2/node/tasks.rs
+++ b/src/api2/node/tasks.rs
@@ -105,7 +105,7 @@ async fn get_task_status(
     if crate::server::worker_is_active(&upid).await? {
         result["status"] = Value::from("running");
     } else {
-        let (_, exitstatus) = crate::server::upid_read_status(&upid).unwrap_or((0, TaskState::Unknown));
+        let exitstatus = crate::server::upid_read_status(&upid).unwrap_or(TaskState::Unknown { endtime: 0 });
         result["status"] = Value::from("stopped");
         result["exitstatus"] = Value::from(exitstatus.to_string());
     };
@@ -352,8 +352,9 @@ pub fn list_tasks(
 
         if let Some(ref state) = info.state {
             if running { continue; }
-            if errors && state.1 == crate::server::TaskState::OK {
-                continue;
+            match state {
+                crate::server::TaskState::OK { .. } if errors => continue,
+                _ => {},
             }
         }
 
diff --git a/src/api2/types/mod.rs b/src/api2/types/mod.rs
index a619810d..de7a5ca0 100644
--- a/src/api2/types/mod.rs
+++ b/src/api2/types/mod.rs
@@ -595,7 +595,7 @@ impl From<crate::server::TaskListInfo> for TaskListItem {
     fn from(info: crate::server::TaskListInfo) -> Self {
         let (endtime, status) = info
             .state
-            .map_or_else(|| (None, None), |(a,b)| (Some(a), Some(b.to_string())));
+            .map_or_else(|| (None, None), |a| (Some(a.endtime()), Some(a.to_string())));
 
         TaskListItem {
             upid: info.upid_str,
diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
index 45672cea..94566bb7 100644
--- a/src/config/jobstate.rs
+++ b/src/config/jobstate.rs
@@ -16,7 +16,7 @@
 //! # use anyhow::{bail, Error};
 //! # use proxmox_backup::server::TaskState;
 //! # use proxmox_backup::config::jobstate::*;
-//! # fn some_code() -> TaskState { TaskState::OK }
+//! # fn some_code() -> TaskState { TaskState::OK { endtime: 0 } }
 //! # fn code() -> Result<(), Error> {
 //! // locks the correct file under /var/lib
 //! // or fails if someone else holds the lock
@@ -62,8 +62,8 @@ pub enum JobState {
     Created { time: i64 },
     /// The Job was last started in 'upid',
     Started { upid: String },
-    /// The Job was last started in 'upid', which finished with 'state' at 'endtime'
-    Finished { upid: String, endtime: i64, state: TaskState }
+    /// The Job was last started in 'upid', which finished with 'state'
+    Finished { upid: String, state: TaskState }
 }
 
 /// Represents a Job and holds the correct lock
@@ -143,12 +143,11 @@ impl JobState {
                         .map_err(|err| format_err!("error parsing upid: {}", err))?;
 
                     if !worker_is_active_local(&parsed) {
-                        let (endtime, state) = upid_read_status(&parsed)
+                        let state = upid_read_status(&parsed)
                             .map_err(|err| format_err!("error reading upid log status: {}", err))?;
 
                         Ok(JobState::Finished {
                             upid,
-                            endtime,
                             state
                         })
                     } else {
@@ -225,11 +224,8 @@ impl Job {
             JobState::Finished { upid, .. } => upid,
         }.to_string();
 
-        let endtime: i64 = epoch_now_u64()? as i64;
-
         self.state = JobState::Finished {
             upid,
-            endtime,
             state,
         };
 
diff --git a/src/server/worker_task.rs b/src/server/worker_task.rs
index da1a877e..a9e4a36a 100644
--- a/src/server/worker_task.rs
+++ b/src/server/worker_task.rs
@@ -156,7 +156,7 @@ pub async fn abort_worker(upid: UPID) -> Result<(), Error> {
     super::send_command(socketname, cmd).map_ok(|_| ()).await
 }
 
-fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<(i64, String)>), Error> {
+fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<TaskState>), Error> {
 
     let data = line.splitn(3, ' ').collect::<Vec<&str>>();
 
@@ -166,7 +166,8 @@ fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<(i64, St
         1 => Ok((data[0].to_owned(), data[0].parse::<UPID>()?, None)),
         3 => {
             let endtime = i64::from_str_radix(data[1], 16)?;
-            Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some((endtime, data[2].to_owned()))))
+            let state = TaskState::from_endtime_and_message(endtime, data[2])?;
+            Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some(state)))
         }
         _ => bail!("wrong number of components"),
     }
@@ -193,9 +194,9 @@ pub fn create_task_log_dirs() -> Result<(), Error> {
 /// Read endtime (time of last log line) and exitstatus from task log file
 /// If there is not a single line with at valid datetime, we assume the
 /// starttime to be the endtime
-pub fn upid_read_status(upid: &UPID) -> Result<(i64, TaskState), Error> {
-    let mut status = TaskState::Unknown;
-    let mut time = upid.starttime;
+pub fn upid_read_status(upid: &UPID) -> Result<TaskState, Error> {
+    let mut endtime = upid.starttime;
+    let mut status = TaskState::Unknown { endtime };
 
     let path = upid.log_path();
 
@@ -213,7 +214,7 @@ pub fn upid_read_status(upid: &UPID) -> Result<(i64, TaskState), Error> {
 
         let mut iter = line.splitn(2, ": ");
         if let Some(time_str) = iter.next() {
-            time = chrono::DateTime::parse_from_rfc3339(time_str)
+            endtime = chrono::DateTime::parse_from_rfc3339(time_str)
                 .map_err(|err| format_err!("cannot parse '{}': {}", time_str, err))?
                 .timestamp();
         } else {
@@ -222,69 +223,86 @@ pub fn upid_read_status(upid: &UPID) -> Result<(i64, TaskState), Error> {
         match iter.next().and_then(|rest| rest.strip_prefix("TASK ")) {
             None => continue,
             Some(rest) => {
-                if let Ok(state) = rest.parse() {
+                if let Ok(state) = TaskState::from_endtime_and_message(endtime, rest) {
                     status = state;
                 }
             }
         }
     }
 
-    Ok((time, status))
+    Ok(status)
 }
 
 /// Task State
-#[derive(Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub enum TaskState {
     /// The Task ended with an undefined state
-    Unknown,
+    Unknown { endtime: i64 },
     /// The Task ended and there were no errors or warnings
-    OK,
+    OK { endtime: i64 },
     /// The Task had 'count' amount of warnings and no errors
-    Warning { count: u64 },
+    Warning { count: u64, endtime: i64 },
     /// The Task ended with the error described in 'message'
-    Error { message: String },
+    Error { message: String, endtime: i64 },
 }
 
 impl TaskState {
-    fn result_text(&self) -> String {
-        match self {
-            TaskState::Error { message } => format!("TASK ERROR: {}", message),
-            other => format!("TASK {}", other),
+    pub fn endtime(&self) -> i64 {
+        match *self {
+            TaskState::Unknown { endtime } => endtime,
+            TaskState::OK { endtime } => endtime,
+            TaskState::Warning { endtime, .. } => endtime,
+            TaskState::Error { endtime, .. } => endtime,
         }
     }
-}
 
-impl std::fmt::Display for TaskState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+    fn result_text(&self) -> String {
         match self {
-            TaskState::Unknown => write!(f, "unknown"),
-            TaskState::OK => write!(f, "OK"),
-            TaskState::Warning { count } => write!(f, "WARNINGS: {}", count),
-            TaskState::Error { message } => write!(f, "{}", message),
+            TaskState::Error { message, .. } => format!("TASK ERROR: {}", message),
+            other => format!("TASK {}", other),
         }
     }
-}
 
-impl std::str::FromStr for TaskState {
-    type Err = Error;
-
-    fn from_str(s: &str) -> Result<Self, Self::Err> {
+    fn from_endtime_and_message(endtime: i64, s: &str) -> Result<Self, Error> {
         if s == "unknown" {
-            Ok(TaskState::Unknown)
+            Ok(TaskState::Unknown { endtime })
         } else if s == "OK" {
-            Ok(TaskState::OK)
+            Ok(TaskState::OK { endtime })
         } else if s.starts_with("WARNINGS: ") {
             let count: u64 = s[10..].parse()?;
-            Ok(TaskState::Warning{ count })
+            Ok(TaskState::Warning{ count, endtime })
         } else if s.len() > 0 {
             let message = if s.starts_with("ERROR: ") { &s[7..] } else { s }.to_string();
-            Ok(TaskState::Error{ message })
+            Ok(TaskState::Error{ message, endtime })
         } else {
             bail!("unable to parse Task Status '{}'", s);
         }
     }
 }
 
+impl std::cmp::PartialOrd for TaskState {
+    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
+        Some(self.endtime().cmp(&other.endtime()))
+    }
+}
+
+impl std::cmp::Ord for TaskState {
+    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
+        self.endtime().cmp(&other.endtime())
+    }
+}
+
+impl std::fmt::Display for TaskState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            TaskState::Unknown { .. } => write!(f, "unknown"),
+            TaskState::OK { .. }=> write!(f, "OK"),
+            TaskState::Warning { count, .. } => write!(f, "WARNINGS: {}", count),
+            TaskState::Error { message, .. } => write!(f, "{}", message),
+        }
+    }
+}
+
 /// Task details including parsed UPID
 ///
 /// If there is no `state`, the task is still running.
@@ -295,7 +313,7 @@ pub struct TaskListInfo {
     /// UPID string representation
     pub upid_str: String,
     /// Task `(endtime, status)` if already finished
-    pub state: Option<(i64, TaskState)>, // endtime, status
+    pub state: Option<TaskState>, // endtime, status
 }
 
 // atomically read/update the task list, update status of finished tasks
@@ -334,15 +352,15 @@ fn update_active_workers(new_upid: Option<&UPID>) -> Result<Vec<TaskListInfo>, E
                     },
                     None => {
                         println!("Detected stopped UPID {}", upid_str);
-                        let (time, status) = upid_read_status(&upid)
-                            .unwrap_or_else(|_| (Local::now().timestamp(), TaskState::Unknown));
+                        let status = upid_read_status(&upid)
+                            .unwrap_or_else(|_| TaskState::Unknown { endtime: Local::now().timestamp() });
                         finish_list.push(TaskListInfo {
-                            upid, upid_str, state: Some((time, status))
+                            upid, upid_str, state: Some(status)
                         });
                     },
-                    Some((endtime, status)) => {
+                    Some(status) => {
                         finish_list.push(TaskListInfo {
-                            upid, upid_str, state: Some((endtime, status.parse()?))
+                            upid, upid_str, state: Some(status)
                         })
                     }
                 }
@@ -378,7 +396,7 @@ fn update_active_workers(new_upid: Option<&UPID>) -> Result<Vec<TaskListInfo>, E
 
     task_list.sort_unstable_by(|b, a| { // lastest on top
         match (&a.state, &b.state) {
-            (Some(s1), Some(s2)) => s1.0.cmp(&s2.0),
+            (Some(s1), Some(s2)) => s1.cmp(&s2),
             (Some(_), None) => std::cmp::Ordering::Less,
             (None, Some(_)) => std::cmp::Ordering::Greater,
             _ => a.upid.starttime.cmp(&b.upid.starttime),
@@ -387,8 +405,8 @@ fn update_active_workers(new_upid: Option<&UPID>) -> Result<Vec<TaskListInfo>, E
 
     let mut raw = String::new();
     for info in &task_list {
-        if let Some((endtime, status)) = &info.state {
-            raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, endtime, status));
+        if let Some(status) = &info.state {
+            raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, status.endtime(), status));
         } else {
             raw.push_str(&info.upid_str);
             raw.push('\n');
@@ -559,12 +577,14 @@ impl WorkerTask {
     pub fn create_state(&self, result: &Result<(), Error>) -> TaskState {
         let warn_count = self.data.lock().unwrap().warn_count;
 
+        let endtime = Local::now().timestamp();
+
         if let Err(err) = result {
-            TaskState::Error { message: err.to_string() }
+            TaskState::Error { message: err.to_string(), endtime }
         } else if warn_count > 0 {
-            TaskState::Warning { count: warn_count }
+            TaskState::Warning { count: warn_count, endtime }
         } else {
-            TaskState::OK
+            TaskState::OK { endtime }
         }
     }
 
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 2/3] cleanup: replace id from do_sync_job with info from job

[Dominik Csapak]

we already have it inside the job itself

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/admin/sync.rs          |  2 +-
 src/api2/pull.rs                | 11 +++++------
 src/bin/proxmox-backup-proxy.rs |  2 +-
 src/config/jobstate.rs          |  8 ++++++++
 4 files changed, 15 insertions(+), 8 deletions(-)

diff --git a/src/api2/admin/sync.rs b/src/api2/admin/sync.rs
index aafd808f..0985e26f 100644
--- a/src/api2/admin/sync.rs
+++ b/src/api2/admin/sync.rs
@@ -90,7 +90,7 @@ fn run_sync_job(
     let mut job = Job::new("syncjob", &id)?;
     job.load()?;
 
-    let upid_str = do_sync_job(&id, sync_job, &userid, None, job)?;
+    let upid_str = do_sync_job(job, sync_job, &userid, None)?;
 
     Ok(upid_str)
 }
diff --git a/src/api2/pull.rs b/src/api2/pull.rs
index e8eb35e1..fc13cf40 100644
--- a/src/api2/pull.rs
+++ b/src/api2/pull.rs
@@ -66,19 +66,18 @@ pub async fn get_pull_parameters(
 }
 
 pub fn do_sync_job(
-    id: &str,
+    mut job: Job,
     sync_job: SyncJobConfig,
     userid: &Userid,
     schedule: Option<String>,
-    mut job: Job,
 ) -> Result<String, Error> {
 
-    let job_id = id.to_string();
-    let worker_type = "syncjob";
+    let job_id = job.jobname().to_string();
+    let worker_type = job.jobtype().to_string();
 
     let upid_str = WorkerTask::spawn(
-        worker_type,
-        Some(id.to_string()),
+        &worker_type,
+        Some(job.jobname().to_string()),
         userid.clone(),
         false,
         move |worker| async move {
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index c7f8bc79..1a30bc6f 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -548,7 +548,7 @@ async fn schedule_datastore_sync_jobs() {
 
         let userid = Userid::backup_userid().clone();
 
-        if let Err(err) = do_sync_job(&job_id, job_config, &userid, Some(event_str), job) {
+        if let Err(err) = do_sync_job(job, job_config, &userid, Some(event_str)) {
             eprintln!("unable to start datastore sync job {} - {}", &job_id, err);
         }
     }
diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
index 94566bb7..8931563b 100644
--- a/src/config/jobstate.rs
+++ b/src/config/jobstate.rs
@@ -232,6 +232,14 @@ impl Job {
         self.write_state()
     }
 
+    pub fn jobtype(&self) -> &str {
+        &self.jobtype
+    }
+
+    pub fn jobname(&self) -> &str {
+        &self.jobname
+    }
+
     fn write_state(&mut self) -> Result<(), Error> {
         let serialized = serde_json::to_string(&self.state)?;
         let path = get_path(&self.jobtype, &self.jobname);
-- 
2.20.1

[pbs-devel] [PATCH proxmox-backup 3/3] config/jobstate: replace Job:load with create_state_file

[Dominik Csapak]

it really is not necessary, since the only time we are interested in
loading the state from the file is when we list it, and there
we use JobState::load directly to avoid the lock

we still need to create the file on syncjob creation though, so
that we have the correct time for the schedule

to do this we add a new create_state_file that overwrites it on creation
of a syncjob

for safety, we subtract 30 seconds from the in-memory state in case
the statefile is missing

since we call create_state_file from  proxmox-backup-api,
we have to chown the lock file after creating to the backup user,
else the sync job scheduling cannot aquire the lock

also we remove the lock file on statefile removal

Signed-off-by: Dominik Csapak <d.csapak@proxmox.com>
---
 src/api2/admin/sync.rs          |  3 +--
 src/api2/config/sync.rs         |  2 ++
 src/bin/proxmox-backup-proxy.rs |  8 +-----
 src/config/jobstate.rs          | 45 +++++++++++++++------------------
 4 files changed, 25 insertions(+), 33 deletions(-)

diff --git a/src/api2/admin/sync.rs b/src/api2/admin/sync.rs
index 0985e26f..ff1e5ebf 100644
--- a/src/api2/admin/sync.rs
+++ b/src/api2/admin/sync.rs
@@ -87,8 +87,7 @@ fn run_sync_job(
 
     let userid: Userid = rpcenv.get_user().unwrap().parse()?;
 
-    let mut job = Job::new("syncjob", &id)?;
-    job.load()?;
+    let job = Job::new("syncjob", &id)?;
 
     let upid_str = do_sync_job(job, sync_job, &userid, None)?;
 
diff --git a/src/api2/config/sync.rs b/src/api2/config/sync.rs
index 8b16192c..57192d90 100644
--- a/src/api2/config/sync.rs
+++ b/src/api2/config/sync.rs
@@ -83,6 +83,8 @@ pub fn create_sync_job(param: Value) -> Result<(), Error> {
 
     sync::save_config(&config)?;
 
+    crate::config::jobstate::create_state_file("syncjob", &sync_job.id)?;
+
     Ok(())
 }
 
diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 1a30bc6f..b75cc763 100644
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -536,13 +536,7 @@ async fn schedule_datastore_sync_jobs() {
         if next > now  { continue; }
 
         let job = match Job::new(worker_type, &job_id) {
-            Ok(mut job) => match job.load() {
-                Ok(_) => job,
-                Err(err) => {
-                    eprintln!("error loading jobstate for {} -  {}: {}", worker_type, &job_id, err);
-                    continue;
-                }
-            }
+            Ok(job) => job,
             Err(_) => continue, // could not get lock
         };
 
diff --git a/src/config/jobstate.rs b/src/config/jobstate.rs
index 8931563b..e5c701d7 100644
--- a/src/config/jobstate.rs
+++ b/src/config/jobstate.rs
@@ -25,13 +25,7 @@
 //!     Err(err) => bail!("could not lock jobstate"),
 //! };
 //!
-//! // job holds the lock
-//! match job.load() {
-//!     Ok(()) => {},
-//!     Err(err) => bail!("could not load state {}", err),
-//! }
-//!
-//! // now the job is loaded;
+//! // job holds the lock, we can start it
 //! job.start("someupid")?;
 //! // do something
 //! let task_state = some_code();
@@ -102,16 +96,32 @@ where
 {
     let mut path = path.as_ref().to_path_buf();
     path.set_extension("lck");
-    open_file_locked(path, Duration::new(10, 0))
+    let lock = open_file_locked(&path, Duration::new(10, 0))?;
+    let backup_user = crate::backup::backup_user()?;
+    nix::unistd::chown(&path, Some(backup_user.uid), Some(backup_user.gid))?;
+    Ok(lock)
 }
 
 /// Removes the statefile of a job, this is useful if we delete a job
 pub fn remove_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
-    let path = get_path(jobtype, jobname);
+    let mut path = get_path(jobtype, jobname);
     let _lock = get_lock(&path)?;
     std::fs::remove_file(&path).map_err(|err|
         format_err!("cannot remove statefile for {} - {}: {}", jobtype, jobname, err)
-    )
+    )?;
+    path.set_extension("lck");
+    // ignore errors
+    let _ = std::fs::remove_file(&path).map_err(|err|
+        format_err!("cannot remove lockfile for {} - {}: {}", jobtype, jobname, err)
+    );
+    Ok(())
+}
+
+/// Creates the statefile with the state 'Created'
+/// overwrites if it exists already
+pub fn create_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
+    let mut job = Job::new(jobtype, jobname)?;
+    job.write_state()
 }
 
 /// Returns the last run time of a job by reading the statefile
@@ -158,7 +168,7 @@ impl JobState {
             }
         } else {
             Ok(JobState::Created {
-                time: epoch_now_u64()? as i64
+                time: epoch_now_u64()? as i64 - 30,
             })
         }
     }
@@ -185,19 +195,6 @@ impl Job {
         })
     }
 
-    /// Loads the state from the statefile if it exists.
-    /// If not, it gets created. Updates 'Started' State to 'Finished'
-    /// if we detect the UPID already stopped
-    pub fn load(&mut self) -> Result<(), Error> {
-        self.state = JobState::load(&self.jobtype, &self.jobname)?;
-
-        if let Err(err) = self.write_state() {
-            bail!("could not write statefile: {}", err);
-        }
-
-        Ok(())
-    }
-
     /// Start the job and update the statefile accordingly
     /// Fails if the job was already started
     pub fn start(&mut self, upid: &str) -> Result<(), Error> {
-- 
2.20.1

[pbs-devel] applied: [PATCH proxmox-backup 1/3] cleanup: merge endtime into TaskState

[Dietmar Maurer]

applied all 3 patches