src/server/jobstate.rs

   1 //! Generic JobState handling
   2 //!
   3 //! A 'Job' can have 3 states
   4 //!  - Created, when a schedule was created but never executed
   5 //!  - Started, when a job is running right now
   6 //!  - Finished, when a job was running in the past
   7 //!
   8 //! and is identified by 2 values: jobtype and jobname (e.g. 'syncjob' and 'myfirstsyncjob')
   9 //!
  10 //! This module Provides 2 helper structs to handle those coniditons
  11 //! 'Job' which handles locking and writing to a file
  12 //! 'JobState' which is the actual state
  13 //!
  14 //! an example usage would be
  15 //! ```no_run
  16 //! # use anyhow::{bail, Error};
  17 //! # use proxmox_backup::server::TaskState;
  18 //! # use proxmox_backup::server::jobstate::*;
  19 //! # fn some_code() -> TaskState { TaskState::OK { endtime: 0 } }
  20 //! # fn code() -> Result<(), Error> {
  21 //! // locks the correct file under /var/lib
  22 //! // or fails if someone else holds the lock
  23 //! let mut job = match Job::new("jobtype", "jobname") {
  24 //!     Ok(job) => job,
  25 //!     Err(err) => bail!("could not lock jobstate"),
  26 //! };
  27 //!
  28 //! // job holds the lock, we can start it
  29 //! job.start("someupid")?;
  30 //! // do something
  31 //! let task_state = some_code();
  32 //! job.finish(task_state)?;
  33 //!
  34 //! // release the lock
  35 //! drop(job);
  36 //! # Ok(())
  37 //! # }
  38 //!
  39 //! ```
  40 use std::path::{Path, PathBuf};
  41
  42 use anyhow::{bail, format_err, Error};
  43 use serde::{Deserialize, Serialize};
  44
  45 use proxmox::tools::fs::{
  46     create_path, file_read_optional_string, replace_file, CreateOptions,
  47 };
  48
  49 use pbs_systemd::time::{compute_next_event, parse_calendar_event};
  50
  51 use crate::{
  52     backup::{open_backup_lockfile, BackupLockGuard},
  53     api2::types::JobScheduleStatus,
  54     server::{
  55         UPID,
  56         TaskState,
  57         upid_read_status,
  58         worker_is_active_local,
  59     },
  60 };
  61
  62 #[derive(Serialize, Deserialize)]
  63 #[serde(rename_all = "kebab-case")]
  64 /// Represents the State of a specific Job
  65 pub enum JobState {
  66     /// A job was created at 'time', but never started/finished
  67     Created { time: i64 },
  68     /// The Job was last started in 'upid',
  69     Started { upid: String },
  70     /// The Job was last started in 'upid', which finished with 'state', and was last updated at 'updated'
  71     Finished {
  72         upid: String,
  73         state: TaskState,
  74         updated: Option<i64>,
  75     },
  76 }
  77
  78 /// Represents a Job and holds the correct lock
  79 pub struct Job {
  80     jobtype: String,
  81     jobname: String,
  82     /// The State of the job
  83     pub state: JobState,
  84     _lock: BackupLockGuard,
  85 }
  86
  87 const JOB_STATE_BASEDIR: &str = "/var/lib/proxmox-backup/jobstates";
  88
  89 /// Create jobstate stat dir with correct permission
  90 pub fn create_jobstate_dir() -> Result<(), Error> {
  91     let backup_user = crate::backup::backup_user()?;
  92     let opts = CreateOptions::new()
  93         .owner(backup_user.uid)
  94         .group(backup_user.gid);
  95
  96     create_path(JOB_STATE_BASEDIR, None, Some(opts))
  97         .map_err(|err: Error| format_err!("unable to create rrdb stat dir - {}", err))?;
  98
  99     Ok(())
 100 }
 101
 102 fn get_path(jobtype: &str, jobname: &str) -> PathBuf {
 103     let mut path = PathBuf::from(JOB_STATE_BASEDIR);
 104     path.push(format!("{}-{}.json", jobtype, jobname));
 105     path
 106 }
 107
 108 fn get_lock<P>(path: P) -> Result<BackupLockGuard, Error>
 109 where
 110     P: AsRef<Path>,
 111 {
 112     let mut path = path.as_ref().to_path_buf();
 113     path.set_extension("lck");
 114     open_backup_lockfile(&path, None, true)
 115 }
 116
 117 /// Removes the statefile of a job, this is useful if we delete a job
 118 pub fn remove_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
 119     let mut path = get_path(jobtype, jobname);
 120     let _lock = get_lock(&path)?;
 121     std::fs::remove_file(&path).map_err(|err| {
 122         format_err!(
 123             "cannot remove statefile for {} - {}: {}",
 124             jobtype,
 125             jobname,
 126             err
 127         )
 128     })?;
 129     path.set_extension("lck");
 130     // ignore errors
 131     let _ = std::fs::remove_file(&path).map_err(|err| {
 132         format_err!(
 133             "cannot remove lockfile for {} - {}: {}",
 134             jobtype,
 135             jobname,
 136             err
 137         )
 138     });
 139     Ok(())
 140 }
 141
 142 /// Creates the statefile with the state 'Created'
 143 /// overwrites if it exists already
 144 pub fn create_state_file(jobtype: &str, jobname: &str) -> Result<(), Error> {
 145     let mut job = Job::new(jobtype, jobname)?;
 146     job.write_state()
 147 }
 148
 149 /// Tries to update the state file with the current time
 150 /// if the job is currently running, does nothing.
 151 /// Intended for use when the schedule changes.
 152 pub fn update_job_last_run_time(jobtype: &str, jobname: &str) -> Result<(), Error> {
 153     let mut job = match Job::new(jobtype, jobname) {
 154         Ok(job) => job,
 155         Err(_) => return Ok(()), // was locked (running), so do not update
 156     };
 157     let time = proxmox::tools::time::epoch_i64();
 158
 159     job.state = match JobState::load(jobtype, jobname)? {
 160         JobState::Created { .. } => JobState::Created { time },
 161         JobState::Started { .. } => return Ok(()), // currently running (without lock?)
 162         JobState::Finished {
 163             upid,
 164             state,
 165             updated: _,
 166         } => JobState::Finished {
 167             upid,
 168             state,
 169             updated: Some(time),
 170         },
 171     };
 172     job.write_state()
 173 }
 174
 175 /// Returns the last run time of a job by reading the statefile
 176 /// Note that this is not locked
 177 pub fn last_run_time(jobtype: &str, jobname: &str) -> Result<i64, Error> {
 178     match JobState::load(jobtype, jobname)? {
 179         JobState::Created { time } => Ok(time),
 180         JobState::Finished {
 181             updated: Some(time),
 182             ..
 183         } => Ok(time),
 184         JobState::Started { upid }
 185         | JobState::Finished {
 186             upid,
 187             state: _,
 188             updated: None,
 189         } => {
 190             let upid: UPID = upid
 191                 .parse()
 192                 .map_err(|err| format_err!("could not parse upid from state: {}", err))?;
 193             Ok(upid.starttime)
 194         }
 195     }
 196 }
 197
 198 impl JobState {
 199     /// Loads and deserializes the jobstate from type and name.
 200     /// When the loaded state indicates a started UPID,
 201     /// we go and check if it has already stopped, and
 202     /// returning the correct state.
 203     ///
 204     /// This does not update the state in the file.
 205     pub fn load(jobtype: &str, jobname: &str) -> Result<Self, Error> {
 206         if let Some(state) = file_read_optional_string(get_path(jobtype, jobname))? {
 207             match serde_json::from_str(&state)? {
 208                 JobState::Started { upid } => {
 209                     let parsed: UPID = upid
 210                         .parse()
 211                         .map_err(|err| format_err!("error parsing upid: {}", err))?;
 212
 213                     if !worker_is_active_local(&parsed) {
 214                         let state = upid_read_status(&parsed)
 215                             .map_err(|err| format_err!("error reading upid log status: {}", err))?;
 216
 217                         Ok(JobState::Finished {
 218                             upid,
 219                             state,
 220                             updated: None,
 221                         })
 222                     } else {
 223                         Ok(JobState::Started { upid })
 224                     }
 225                 }
 226                 other => Ok(other),
 227             }
 228         } else {
 229             Ok(JobState::Created {
 230                 time: proxmox::tools::time::epoch_i64() - 30,
 231             })
 232         }
 233     }
 234 }
 235
 236 impl Job {
 237     /// Creates a new instance of a job with the correct lock held
 238     /// (will be hold until the job is dropped again).
 239     ///
 240     /// This does not load the state from the file, to do that,
 241     /// 'load' must be called
 242     pub fn new(jobtype: &str, jobname: &str) -> Result<Self, Error> {
 243         let path = get_path(jobtype, jobname);
 244
 245         let _lock = get_lock(&path)?;
 246
 247         Ok(Self {
 248             jobtype: jobtype.to_string(),
 249             jobname: jobname.to_string(),
 250             state: JobState::Created {
 251                 time: proxmox::tools::time::epoch_i64(),
 252             },
 253             _lock,
 254         })
 255     }
 256
 257     /// Start the job and update the statefile accordingly
 258     /// Fails if the job was already started
 259     pub fn start(&mut self, upid: &str) -> Result<(), Error> {
 260         if let JobState::Started { .. } = self.state {
 261             bail!("cannot start job that is started!");
 262         }
 263
 264         self.state = JobState::Started {
 265             upid: upid.to_string(),
 266         };
 267
 268         self.write_state()
 269     }
 270
 271     /// Finish the job and update the statefile accordingly with the given taskstate
 272     /// Fails if the job was not yet started
 273     pub fn finish(&mut self, state: TaskState) -> Result<(), Error> {
 274         let upid = match &self.state {
 275             JobState::Created { .. } => bail!("cannot finish when not started"),
 276             JobState::Started { upid } => upid,
 277             JobState::Finished { upid, .. } => upid,
 278         }
 279         .to_string();
 280
 281         self.state = JobState::Finished {
 282             upid,
 283             state,
 284             updated: None,
 285         };
 286
 287         self.write_state()
 288     }
 289
 290     pub fn jobtype(&self) -> &str {
 291         &self.jobtype
 292     }
 293
 294     pub fn jobname(&self) -> &str {
 295         &self.jobname
 296     }
 297
 298     fn write_state(&mut self) -> Result<(), Error> {
 299         let serialized = serde_json::to_string(&self.state)?;
 300         let path = get_path(&self.jobtype, &self.jobname);
 301
 302         let backup_user = crate::backup::backup_user()?;
 303         let mode = nix::sys::stat::Mode::from_bits_truncate(0o0644);
 304         // set the correct owner/group/permissions while saving file
 305         // owner(rw) = backup, group(r)= backup
 306         let options = CreateOptions::new()
 307             .perm(mode)
 308             .owner(backup_user.uid)
 309             .group(backup_user.gid);
 310
 311         replace_file(path, serialized.as_bytes(), options)
 312     }
 313 }
 314
 315 pub fn compute_schedule_status(
 316     job_state: &JobState,
 317     schedule: Option<&str>,
 318 ) -> Result<JobScheduleStatus, Error> {
 319     let (upid, endtime, state, last) = match job_state {
 320         JobState::Created { time } => (None, None, None, *time),
 321         JobState::Started { upid } => {
 322             let parsed_upid: UPID = upid.parse()?;
 323             (Some(upid), None, None, parsed_upid.starttime)
 324         }
 325         JobState::Finished {
 326             upid,
 327             state,
 328             updated,
 329         } => {
 330             let last = updated.unwrap_or_else(|| state.endtime());
 331             (
 332                 Some(upid),
 333                 Some(state.endtime()),
 334                 Some(state.to_string()),
 335                 last,
 336             )
 337         }
 338     };
 339
 340     let mut status = JobScheduleStatus::default();
 341     status.last_run_upid = upid.map(String::from);
 342     status.last_run_state = state;
 343     status.last_run_endtime = endtime;
 344
 345     if let Some(schedule) = schedule {
 346         if let Ok(event) = parse_calendar_event(&schedule) {
 347             // ignore errors
 348             status.next_run = compute_next_event(&event, last, false).unwrap_or(None);
 349         }
 350     }
 351
 352     Ok(status)
 353 }