From mboxrd@z Thu Jan 1 00:00:00 1970 Return-Path: Received: from gate001.proxmox.com (gate001.proxmox.com [45.144.208.40]) by lore.proxmox.com (Postfix) with ESMTPS id AB9D51FF0C0 for ; Wed, 21 Oct 2026 22:56:02 +0200 (CEST) Received: from gate001.proxmox.com (localhost.localdomain [127.0.0.1]) by gate001.proxmox.com (Proxmox) with ESMTP id D6A4C215D4; Wed, 21 Oct 2026 22:55:58 +0200 (CEST) From: Thomas Lamprecht To: pbs-devel@lists.proxmox.com Subject: [PATCH 3/7] client: job: add job runner and scheduled backup daemon Date: Wed, 14 Oct 2026 12:51:32 +0200 Message-ID: <20261014105208.2747208-4-t.lamprecht@proxmox.com> X-Mailer: git-send-email 2.47.3 In-Reply-To: <20261014105208.2747208-1-t.lamprecht@proxmox.com> References: <20261014105208.2747208-1-t.lamprecht@proxmox.com> MIME-Version: 1.0 Content-Transfer-Encoding: 8bit X-Bm-Milter-Handled: 55990f41-d878-4baa-be0a-ee34c49e34d2 X-Bm-Transport-Timestamp: 1792616142104 X-SPAM-LEVEL: Spam detection results: 0 AWL -1.534 Adjusted score from AWL reputation of From: address DATE_IN_PAST_96_XX 2.07 Date: is 96 hours or more before Received: date DMARC_MISSING 0.1 Missing DMARC policy KAM_DMARC_STATUS 0.01 Test Rule for DKIM or SPF Failure with Strict Alignment (newer systems) KAM_MAILER 2 Automated Mailer Tag Left in Email RCVD_IN_DNSWL_MED -2.3 Sender listed at https://www.dnswl.org/, medium trust SPF_HELO_NONE 0.001 SPF: HELO does not publish an SPF Record SPF_PASS -0.001 SPF: sender matches SPF record Message-ID-Hash: AQJJIU2RBLVJQ3NECJSEFG5DW7YFHQBR X-Message-ID-Hash: AQJJIU2RBLVJQ3NECJSEFG5DW7YFHQBR X-MailFrom: t.lamprecht@proxmox.com X-Mailman-Rule-Misses: dmarc-mitigation; no-senders; approved; loop; banned-address; emergency; member-moderation; nonmember-moderation; administrivia; implicit-dest; max-recipients; max-size; news-moderation; no-subject; digests; suspicious-header X-Mailman-Version: 3.3.10 Precedence: list List-Id: Proxmox Backup Server development discussion List-Help: List-Owner: List-Post: List-Subscribe: List-Unsubscribe: Add `job run ` to run one job, and `job scheduler`, a per-user daemon that starts due jobs. The daemon plans at least once a minute rather than sleeping until the next event, as timers stop during suspend and wall-clock deadlines move with NTP and DST. Each run is a separate `job run` process without stdin, which keeps the secrets of jobs apart and avoids prompts on a shared terminal. At most two backups run at once. A run holds the config lock until it has the job lock, like `job remove`, and a scheduled run then checks that the job is still due. A run whose outcome cannot be recorded fails, as the job would otherwise stay due and run again. The pre-run hook proceeds on exit 0, skips on 1 and fails the run otherwise, so that a broken hook is not mistaken for a skip. It gets a minute, no secrets, and its own process group, so a timeout also stops what it started. If the config does not parse, the daemon refuses to start, and a running one starts no new runs until it is fixed. Signed-off-by: Thomas Lamprecht --- proxmox-backup-client/Cargo.toml | 3 +- proxmox-backup-client/src/job/mod.rs | 22 + proxmox-backup-client/src/job/runner.rs | 502 +++++++++++++++++++++ proxmox-backup-client/src/job/scheduler.rs | 327 ++++++++++++++ proxmox-backup-client/src/main.rs | 4 +- proxmox-backup-client/tests/job.rs | 263 +++++++++++ 6 files changed, 1119 insertions(+), 2 deletions(-) create mode 100644 proxmox-backup-client/src/job/mod.rs create mode 100644 proxmox-backup-client/src/job/runner.rs create mode 100644 proxmox-backup-client/src/job/scheduler.rs create mode 100644 proxmox-backup-client/tests/job.rs diff --git a/proxmox-backup-client/Cargo.toml b/proxmox-backup-client/Cargo.toml index 5f0140e78..e3a5474d9 100644 --- a/proxmox-backup-client/Cargo.toml +++ b/proxmox-backup-client/Cargo.toml @@ -13,7 +13,7 @@ nix.workspace = true openssl.workspace = true serde.workspace = true serde_json.workspace = true -tokio = { workspace = true, features = [ "rt", "rt-multi-thread" ] } +tokio = { workspace = true, features = [ "io-util", "macros", "process", "rt", "rt-multi-thread", "signal", "time" ] } tokio-stream.workspace = true tokio-util = { workspace = true, features = [ "codec" ] } xdg.workspace = true @@ -29,6 +29,7 @@ proxmox-log.workspace = true proxmox-io.workspace = true proxmox-router = { workspace = true, features = [ "cli" ] } proxmox-schema = { workspace = true, features = [ "api-macro" ] } +proxmox-section-config.workspace = true proxmox-sortable-macro.workspace = true proxmox-sys.workspace = true proxmox-systemd.workspace = true diff --git a/proxmox-backup-client/src/job/mod.rs b/proxmox-backup-client/src/job/mod.rs new file mode 100644 index 000000000..ffde90ad1 --- /dev/null +++ b/proxmox-backup-client/src/job/mod.rs @@ -0,0 +1,22 @@ +//! Scheduled backup jobs: the `job` subcommands, the [`scheduler`] daemon and the [`runner`]. + +use proxmox_router::cli::*; + +use pbs_client::client_job::complete_job_id; + +mod runner; +mod scheduler; + +pub fn cli() -> CliCommandMap { + CliCommandMap::new() + .insert( + "run", + CliCommand::new(&runner::API_METHOD_RUN_JOB) + .arg_param(&["id"]) + .completion_cb("id", complete_job_id), + ) + .insert( + "scheduler", + CliCommand::new(&scheduler::API_METHOD_RUN_SCHEDULER), + ) +} diff --git a/proxmox-backup-client/src/job/runner.rs b/proxmox-backup-client/src/job/runner.rs new file mode 100644 index 000000000..a3bcefeed --- /dev/null +++ b/proxmox-backup-client/src/job/runner.rs @@ -0,0 +1,502 @@ +//! Executing a single backup job, as `proxmox-backup-client job run ` does. + +use anyhow::{Error, bail, format_err}; +use serde_json::Value; + +use proxmox_router::cli::*; +use proxmox_schema::api; + +use pbs_api_types::{BackupNamespace, ClientRateLimitConfig, JOB_ID_SCHEMA}; +use pbs_client::client_job::{self, ClientBackupJobConfig}; + +#[api( + input: { + properties: { + id: { + schema: JOB_ID_SCHEMA, + }, + "dry-run": { + type: Boolean, + description: "Only show what the job would back up, without uploading \ + anything or recording the run.", + optional: true, + default: false, + }, + scheduled: { + type: Boolean, + description: "Only run the job if it is due, as the scheduler does.", + optional: true, + default: false, + }, + }, + }, +)] +/// Run a backup job now. +pub(super) async fn run_job(id: String, dry_run: bool, scheduled: bool) -> Result<(), Error> { + // Hold the config lock until the job lock is taken, like `job remove`, so that a removal + // cannot slip in between. + let config_lock = client_job::lock_config()?; + + let (config, _digest) = client_job::config()?; + let job_config = match client_job::find_job(&config, &id)? { + Some(job_config) => job_config, + None if scheduled => { + log::info!("backup job '{id}' no longer exists, not running it"); + return Ok(()); + } + None => bail!("job '{id}' does not exist"), + }; + + // even a dry run, so that it cannot overlap with a real one + let job = client_job::ClientJob::new(&id)?; + drop(config_lock); + + // the daemon decided before the lock was taken, and things may have changed since + if scheduled && !still_due(&job_config)? { + return Ok(()); + } + + run_backup_job_once(job, job_config, dry_run).await +} + +/// Whether a job the scheduler picked is still due. +fn still_due(job_config: &ClientBackupJobConfig) -> Result { + let id = &job_config.id; + + if job_config.disable.unwrap_or(false) { + log::info!("backup job '{id}' is disabled, not running it"); + return Ok(false); + } + + let event: proxmox_time::CalendarEvent = job_config.schedule.parse()?; + match client_job::next_run_after(&event, id)? { + Some(next) if next <= proxmox_time::epoch_i64() => Ok(true), + _ => { + log::info!("backup job '{id}' is no longer due, not running it"); + Ok(false) + } + } +} + +/// Run a backup job and record its outcome. +async fn run_backup_job_once( + mut job: client_job::ClientJob, + job_config: ClientBackupJobConfig, + dry_run: bool, +) -> Result<(), Error> { + let id = &job_config.id; + if !dry_run { + job.start()?; + } + + let result = run_backup_job_inner(&job_config, dry_run).await; + + // Persist the outcome without replacing a backup error with a state-write error. + let record = match &result { + Ok(JobOutcome::Completed) if dry_run => { + log::info!("dry run of backup job '{id}' finished"); + Ok(()) + } + // log success only once it is recorded + Ok(JobOutcome::Completed) => job.finish(true, None).inspect(|()| { + log::info!("backup job '{id}' finished successfully"); + }), + Ok(JobOutcome::SkippedByHook { code }) => { + if dry_run { + log::info!("backup job '{id}' was skipped by its pre-run hook"); + Ok(()) + } else { + job.skip(*code).inspect(|()| { + log::info!("backup job '{id}' was skipped by its pre-run hook"); + }) + } + } + Err(_) if dry_run => Ok(()), + Err(err) => job.finish(false, Some(truncated_message(err))), + }; + // An unrecorded outcome fails the run, as the job would otherwise stay due and run again. + match (result, record) { + (Err(err), record) => { + if let Err(record_err) = record { + log::error!("backup job '{id}': cannot record its outcome: {record_err}"); + } + bail!("backup job '{id}' failed: {err:#}") + } + (Ok(_), Err(err)) => { + bail!("backup job '{id}' ran, but its outcome was not recorded: {err}") + } + (Ok(_), Ok(())) => Ok(()), + } +} + +/// The error for the state file, capped, as it can carry a password helper's output. +fn truncated_message(err: &Error) -> String { + const MAX_LEN: usize = 1024; + + let mut text = format!("{err:#}"); + if text.len() > MAX_LEN { + let end = (0..=MAX_LEN) + .rev() + .find(|&idx| text.is_char_boundary(idx)) + .unwrap_or(0); + text.truncate(end); + text.push_str("[...]"); + } + text +} + +#[derive(Debug)] +enum JobOutcome { + Completed, + SkippedByHook { code: i32 }, +} + +/// Everything that can fail once the run is recorded as started. +async fn run_backup_job_inner( + job_config: &ClientBackupJobConfig, + dry_run: bool, +) -> Result { + // a hand-edited job never went through `job create` or `job update` + job_config.validate()?; + + if let Some(hook) = &job_config.pre_run_hook { + if let Some(outcome) = run_pre_run_hook(&job_config.id, hook).await? { + return Ok(outcome); + } + } + + populate_secret_env(job_config); + + log::info!("starting backup job '{}'", job_config.id); + do_create_backup_from_job(job_config, dry_run).await?; + Ok(JobOutcome::Completed) +} + +/// Environment variables that can carry a secret into the client. +const SECRET_ENV_VARS: [&str; 8] = [ + "PBS_PASSWORD", + "PBS_PASSWORD_FD", + "PBS_PASSWORD_FILE", + "PBS_PASSWORD_CMD", + "PBS_ENCRYPTION_PASSWORD", + "PBS_ENCRYPTION_PASSWORD_FD", + "PBS_ENCRYPTION_PASSWORD_FILE", + "PBS_ENCRYPTION_PASSWORD_CMD", +]; + +/// A hook that never returns would keep the job locked, so it gets a deadline. +const PRE_RUN_HOOK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60); + +/// Any other non-zero exit status means a broken hook, not a skip. +const PRE_RUN_HOOK_SKIP_CODE: i32 = 1; + +/// Run a job's pre-run hook: `Ok(None)` proceeds, `Ok(Some(..))` skips, an error fails the run. +async fn run_pre_run_hook(job_id: &str, hook: &str) -> Result, Error> { + run_pre_run_hook_timeout(job_id, hook, PRE_RUN_HOOK_TIMEOUT).await +} + +async fn run_pre_run_hook_timeout( + job_id: &str, + hook: &str, + timeout: std::time::Duration, +) -> Result, Error> { + use std::os::unix::process::ExitStatusExt; + + log::info!("running pre-run hook for job '{job_id}'"); + let mut command = tokio::process::Command::new(hook); + command + .env("PBS_JOB_ID", job_id) + .current_dir("/") + .stdin(std::process::Stdio::null()) + // its own process group, so that a timeout also stops what it started + .process_group(0) + .kill_on_drop(true); + // third-party code has no business seeing the backup's secrets + for var in SECRET_ENV_VARS { + command.env_remove(var); + } + command.env_remove("CREDENTIALS_DIRECTORY"); + + let mut child = command + .spawn() + .map_err(|err| format_err!("cannot run pre-run hook '{hook}': {err}"))?; + + let status = match tokio::time::timeout(timeout, child.wait()).await { + Ok(status) => status.map_err(|err| format_err!("pre-run hook '{hook}' failed: {err}"))?, + Err(_) => { + kill_hook_process_group(&mut child).await; + bail!( + "pre-run hook '{hook}' did not finish within {} seconds", + timeout.as_secs(), + ); + } + }; + + if let Some(signal) = status.signal() { + bail!("pre-run hook '{hook}' was killed by signal {signal}"); + } + match status.code() { + Some(0) => Ok(None), + Some(PRE_RUN_HOOK_SKIP_CODE) => Ok(Some(JobOutcome::SkippedByHook { + code: PRE_RUN_HOOK_SKIP_CODE, + })), + Some(code) => bail!( + "pre-run hook '{hook}' exited with status {code}, expected 0 to proceed or \ + {PRE_RUN_HOOK_SKIP_CODE} to skip the run" + ), + None => bail!("pre-run hook '{hook}' did not exit normally"), + } +} + +/// Stop a hook that ran out of time, together with the processes it started. +async fn kill_hook_process_group(child: &mut tokio::process::Child) { + use nix::sys::signal::{Signal, killpg}; + use nix::unistd::Pid; + + if let Some(pid) = child.id() { + // the hook leads its own group, so the group ID is its PID + if let Err(err) = killpg(Pid::from_raw(pid as i32), Signal::SIGKILL) { + log::warn!("cannot stop the pre-run hook's process group: {err}"); + } + } + let _ = child.wait().await; +} + +/// Hand the job's secret sources and fingerprint to the client through the usual `PBS_*` variables. +/// +/// Inherited variables for a secret the job configures are cleared first, as they would win. +fn populate_secret_env(job: &pbs_client::client_job::ClientBackupJobConfig) { + let set = |name: &str, value: Option<&String>| { + if let Some(value) = value { + std::env::set_var(name, value); + } + }; + + if job.password_file.is_some() || job.password_cmd.is_some() { + for var in SECRET_ENV_VARS + .iter() + .filter(|var| !var.contains("ENCRYPTION")) + { + std::env::remove_var(var); + } + } + if job.encryption_password_file.is_some() || job.encryption_password_cmd.is_some() { + for var in SECRET_ENV_VARS + .iter() + .filter(|var| var.contains("ENCRYPTION")) + { + std::env::remove_var(var); + } + } + + set("PBS_PASSWORD_FILE", job.password_file.as_ref()); + set("PBS_PASSWORD_CMD", job.password_cmd.as_ref()); + set( + "PBS_ENCRYPTION_PASSWORD_FILE", + job.encryption_password_file.as_ref(), + ); + set( + "PBS_ENCRYPTION_PASSWORD_CMD", + job.encryption_password_cmd.as_ref(), + ); + set("PBS_FINGERPRINT", job.fingerprint.as_ref()); +} + +/// Insert a job option if set, so that `create_backup` applies its own default otherwise. +fn insert_opt( + param: &mut serde_json::Map, + key: &str, + value: Option<&T>, +) -> Result<(), Error> { + if let Some(value) = value { + param.insert(key.to_string(), serde_json::to_value(value)?); + } + Ok(()) +} + +/// Run the backup of a job through the regular `create_backup` command. +async fn do_create_backup_from_job( + job: &ClientBackupJobConfig, + dry_run: bool, +) -> Result<(), Error> { + let repo = job.resolve_repository()?; + + let mut param = serde_json::Map::new(); + param.insert("repository".into(), repo.to_string().into()); + // always set, or `PBS_NAMESPACE` from the environment would apply + let ns = job.ns.clone().unwrap_or_else(BackupNamespace::root); + param.insert("ns".into(), serde_json::to_value(ns)?); + param.insert("backupspec".into(), serde_json::to_value(&job.backupspec)?); + + insert_opt(&mut param, "backup-id", job.backup_id.as_ref())?; + insert_opt(&mut param, "backup-type", job.backup_type.as_ref())?; + insert_opt(&mut param, "keyfile", job.keyfile.as_ref())?; + insert_opt( + &mut param, + "master-pubkey-file", + job.master_pubkey_file.as_ref(), + )?; + insert_opt(&mut param, "crypt-mode", job.crypt_mode.as_ref())?; + insert_opt(&mut param, "exclude", job.exclude.as_ref())?; + insert_opt(&mut param, "include-dev", job.include_dev.as_ref())?; + insert_opt( + &mut param, + "all-file-systems", + job.all_file_systems.as_ref(), + )?; + insert_opt( + &mut param, + "skip-lost-and-found", + job.skip_lost_and_found.as_ref(), + )?; + insert_opt( + &mut param, + "change-detection-mode", + job.change_detection_mode.as_ref(), + )?; + insert_opt(&mut param, "chunk-size", job.chunk_size.as_ref())?; + insert_opt(&mut param, "entries-max", job.entries_max.as_ref())?; + insert_opt( + &mut param, + "skip-e2big-xattr", + job.skip_e2big_xattr.as_ref(), + )?; + + let param = Value::Object(param); + let all_file_systems = job.all_file_systems.unwrap_or(false); + let skip_lost_and_found = job.skip_lost_and_found.unwrap_or(false); + let skip_e2big_xattr = job.skip_e2big_xattr.unwrap_or(false); + let change_detection_mode = job.change_detection_mode.clone(); + // `ClientRateLimitConfig` has private fields, so build it through serde like the CLI does + let mut limit = serde_json::Map::new(); + insert_opt(&mut limit, "rate", job.rate.as_ref())?; + insert_opt(&mut limit, "burst", job.burst.as_ref())?; + let limit: ClientRateLimitConfig = serde_json::from_value(Value::Object(limit))?; + + let rpcenv = &mut CliEnvironment::new() as &mut dyn proxmox_router::RpcEnvironment; + + crate::create_backup( + param, + all_file_systems, + skip_lost_and_found, + change_detection_mode, + dry_run, + skip_e2big_xattr, + job.no_cache.unwrap_or(false), + limit, + &crate::API_METHOD_CREATE_BACKUP, + rpcenv, + ) + .await?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use proxmox_schema::{ApiType, ObjectSchemaType}; + + use pbs_client::client_job::ClientBackupJobConfig; + + use super::*; + + /// Write an executable hook script into a fresh directory and return its path. + fn write_hook(name: &str, body: &str) -> std::path::PathBuf { + use std::os::unix::fs::PermissionsExt; + + let dir = std::env::temp_dir().join(format!("pbs-hook-test-{}-{name}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let path = dir.join(format!("{name}.sh")); + std::fs::write(&path, body).unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + + /// A hook that runs out of time is killed together with the processes it started. + #[tokio::test] + async fn a_hook_that_times_out_takes_its_process_group_with_it() { + let dir = std::env::temp_dir().join(format!("pbs-hook-test-{}-group", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + let pidfile = dir.join("grandchild.pid"); + let _ = std::fs::remove_file(&pidfile); + + let hook = write_hook( + "hangs", + &format!( + "#!/bin/sh\nsh -c 'echo $$ > {}; exec sleep 60' &\nsleep 60\n", + pidfile.display(), + ), + ); + let hook = hook.to_str().unwrap(); + + // long enough for the helper to record itself, even on a busy machine + let err = run_pre_run_hook_timeout("group", hook, std::time::Duration::from_secs(2)) + .await + .expect_err("a hook that sleeps past its timeout must fail the run"); + assert!(err.to_string().contains("did not finish within"), "{err}"); + + let mut recorded = None; + for _ in 0..50 { + if let Some(pid) = std::fs::read_to_string(&pidfile) + .ok() + .and_then(|text| text.trim().parse::().ok()) + { + recorded = Some(pid); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + let grandchild = recorded.expect("the hook records the process it started"); + + // the group may take a moment to go away + let mut alive = true; + for _ in 0..50 { + if !std::path::Path::new(&format!("/proc/{grandchild}")).exists() { + alive = false; + break; + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } + assert!( + !alive, + "process {grandchild} outlived the hook that started it" + ); + + let _ = std::fs::remove_file(&pidfile); + } + + /// Backup options a job deliberately lacks, each with the reason. + const NOT_IN_JOBS: [(&str, &str); 4] = [ + ( + "keyfd", + "a file descriptor cannot be stored and reopened later", + ), + ("master-pubkey-fd", "same, the job uses master-pubkey-file"), + ("backup-time", "each run picks its own time"), + ("dry-run", "per invocation, `job run` has its own flag"), + ]; + + /// A job must cover every option of a one-shot backup, or list it in `NOT_IN_JOBS`. + #[test] + fn job_config_covers_the_backup_options() { + let job = ClientBackupJobConfig::API_SCHEMA + .any_object() + .expect("job config schema is an object schema"); + + let mut missing = Vec::new(); + for (name, _optional, _schema) in crate::API_METHOD_CREATE_BACKUP.parameters.properties() { + if job.lookup(name).is_some() { + continue; + } + if NOT_IN_JOBS.iter().any(|(excluded, _)| excluded == name) { + continue; + } + missing.push(*name); + } + assert!( + missing.is_empty(), + "backup options missing from the job config: {missing:?}. Add them, or list them in \ + NOT_IN_JOBS with the reason they cannot be stored.", + ); + } +} diff --git a/proxmox-backup-client/src/job/scheduler.rs b/proxmox-backup-client/src/job/scheduler.rs new file mode 100644 index 000000000..e8eaa7d0f --- /dev/null +++ b/proxmox-backup-client/src/job/scheduler.rs @@ -0,0 +1,327 @@ +//! The per-user scheduler daemon: it watches the job config and starts due jobs, each as its own +//! `proxmox-backup-client job run ` process. + +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; + +use anyhow::{Error, bail, format_err}; + +use proxmox_schema::api; +use proxmox_section_config::SectionConfigData; + +use nix::fcntl::{Flock, FlockArg}; +use pbs_client::client_job::{self, ClientBackupJobConfig}; + +/// Argv prefix passed to a re-exec of this binary to run a single job. +/// Kept in sync with the CLI command tree wired up in [`cli`]. +const JOB_RUN_ARGV: [&str; 2] = ["job", "run"]; + +#[api( + input: { + properties: {}, + }, +)] +/// Run the scheduler daemon that starts due backup jobs. +pub(super) async fn run_scheduler() -> Result<(), Error> { + use tokio::signal::unix::{SignalKind, signal}; + + // before the PID file exists, as SIGHUP terminates a process without a handler + let mut sighup = signal(SignalKind::hangup())?; + let mut sigterm = signal(SignalKind::terminate())?; + + // held for the daemon's lifetime, which keeps a second scheduler from starting + let _pid_lock = acquire_scheduler_pidfile(&scheduler_pid_path()?)?; + // a broken config at startup would look healthy to systemd + client_job::config()?; + + let self_exe = std::env::current_exe()?; + let mut children: Vec<(String, tokio::process::Child)> = Vec::new(); + let mut last_started: HashMap = HashMap::new(); + // jobs that found no free slot, so that each wait is logged once + let mut waiting: HashSet = HashSet::new(); + // config problems are logged once per config, not on every pass + let mut reported_digest = None; + let mut reported_load_error = false; + + log::info!("backup job scheduler started (PID {})", std::process::id()); + + loop { + reap_finished(&mut children); + let in_flight: HashSet = children.iter().map(|(id, _)| id.clone()).collect(); + + // every run reads the config itself, so a broken one starts no runs at all + let plan = match client_job::config() { + Ok((config, digest)) => { + reported_load_error = false; + let report = reported_digest.replace(digest) != Some(digest); + plan_jobs(&config, &in_flight, &last_started, report) + } + Err(err) => { + if !std::mem::replace(&mut reported_load_error, true) { + log::error!( + "cannot read the job config, starting no runs until it is fixed: {err}" + ); + } + Plan { + due: Vec::new(), + next_wake: now_plus(MAX_SLEEP), + } + } + }; + let deferred = start_due_jobs(&self_exe, plan.due, &mut children, &mut last_started); + for job_id in deferred.difference(&waiting) { + log::info!( + "backup job '{job_id}' is queued, {MAX_CONCURRENT_RUNS} backups are already \ + running" + ); + } + waiting = deferred; + + let delay = (plan.next_wake - proxmox_time::epoch_i64()).clamp(0, MAX_SLEEP); + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(delay as u64)) => {} + // a finished run frees a slot; `Child::wait` is cancel safe + _ = any_child_finished(&mut children) => {} + _ = sighup.recv() => log::info!("received SIGHUP, reloading configuration"), + _ = sigterm.recv() => { + log::info!("received SIGTERM, shutting down"); + break; + } + } + } + + reap_finished(&mut children); + if !children.is_empty() { + log::info!("waiting for {} running backup(s) to finish", children.len()); + let wait_all = async { + for (job_id, child) in children.iter_mut() { + log_child_result(job_id, child.wait().await); + } + }; + // a second SIGTERM stops the wait + tokio::select! { + _ = wait_all => {} + _ = sigterm.recv() => { + log::warn!("received second SIGTERM, exiting without waiting for running backups"); + } + } + } + + log::info!("backup job scheduler stopped"); + Ok(()) +} + +/// What one planning pass concluded. +struct Plan { + /// IDs of the jobs to start now, the longest-waiting one first. + due: Vec, + /// Wall-clock time by which the loop has to look again. + next_wake: i64, +} + +/// Work out which jobs are due, skipping disabled and running ones. `report` enables the log +/// messages about the config itself, so that they do not repeat on every pass. +fn plan_jobs( + config: &SectionConfigData, + in_flight: &HashSet, + last_started: &HashMap, + report: bool, +) -> Plan { + use proxmox_time::CalendarEvent; + + let now = proxmox_time::epoch_i64(); + let mut next_wake = now_plus(MAX_SLEEP); + let mut due = Vec::new(); + + for job_id in client_job::job_ids(config) { + let job: ClientBackupJobConfig = match config.lookup(client_job::JOB_SECTION_TYPE, job_id) { + Ok(job) => job, + Err(err) => { + if report { + log::error!("cannot parse job '{job_id}': {err}"); + } + continue; + } + }; + + if job.disable.unwrap_or(false) { + continue; + } + + let event: CalendarEvent = match job.schedule.parse() { + Ok(event) => event, + Err(err) => { + if report { + log::error!("job '{job_id}': invalid schedule '{}': {err}", job.schedule); + } + continue; + } + }; + + let next = match client_job::next_run_after(&event, job_id) { + Ok(Some(next)) => next, + Ok(None) => { + if report { + log::warn!( + "job '{job_id}': schedule '{}' has no future event, the job will not run \ + again", + job.schedule, + ); + } + continue; + } + Err(err) => { + if report { + log::error!("job '{job_id}': cannot compute next run: {err}"); + } + continue; + } + }; + + if next > now { + next_wake = next_wake.min(next); + continue; + } + if in_flight.contains(job_id) || client_job::is_running(job_id) { + continue; + } + // a run that records nothing leaves its job due, so do not restart it on every wakeup + if let Some(started) = last_started.get(job_id) { + let earliest = started + MIN_RESTART_INTERVAL; + if earliest > now { + next_wake = next_wake.min(earliest); + continue; + } + } + due.push((next, job_id.clone())); + } + + // longest waiting first, rather than in config order + due.sort_unstable(); + + Plan { + due: due.into_iter().map(|(_next, job_id)| job_id).collect(), + next_wake, + } +} + +/// Start the due jobs, up to the concurrency limit, and return the ones that have to wait. +fn start_due_jobs( + self_exe: &Path, + due: Vec, + children: &mut Vec<(String, tokio::process::Child)>, + last_started: &mut HashMap, +) -> HashSet { + let mut deferred = HashSet::new(); + for job_id in due { + if children.len() >= MAX_CONCURRENT_RUNS { + deferred.insert(job_id); + continue; + } + match tokio::process::Command::new(self_exe) + .args(JOB_RUN_ARGV) + .arg(&job_id) + // the run checks again that the job is still due + .arg("--scheduled") + // never prompt on a terminal, the same as under systemd + .stdin(std::process::Stdio::null()) + .spawn() + { + Ok(child) => { + match child.id() { + Some(pid) => log::info!("spawned backup job '{job_id}' (PID {pid})"), + None => log::info!("spawned backup job '{job_id}'"), + } + last_started.insert(job_id.clone(), proxmox_time::epoch_i64()); + children.push((job_id, child)); + } + Err(err) => log::error!("cannot spawn backup job '{job_id}': {err}"), + } + } + deferred +} + +/// Upper bound for one sleep. Timers stop during suspend and wall-clock deadlines move with NTP +/// and DST, so look again at least once a minute, like the proxy does. +const MAX_SLEEP: i64 = 60; + +/// Shortest time between two starts of the same job, for runs that record no outcome. +const MIN_RESTART_INTERVAL: i64 = MAX_SLEEP; + +/// Backups running at the same time, so that due jobs do not all compete for disk and uplink. +const MAX_CONCURRENT_RUNS: usize = 2; + +fn now_plus(secs: i64) -> i64 { + proxmox_time::epoch_i64().saturating_add(secs) +} + +/// Wait until any one of the running jobs finishes, or forever if none is running. +async fn any_child_finished(children: &mut [(String, tokio::process::Child)]) { + if children.is_empty() { + std::future::pending::<()>().await; + return; + } + let waits = children + .iter_mut() + .map(|(_, child)| Box::pin(child.wait())) + .collect::>(); + let _ = futures::future::select_all(waits).await; +} + +/// Reap the runs that exited, so that they no longer count as in flight. +fn reap_finished(children: &mut Vec<(String, tokio::process::Child)>) { + children.retain_mut(|(job_id, child)| match child.try_wait() { + Ok(None) => true, + Ok(Some(status)) => { + log_child_result(job_id, Ok(status)); + false + } + Err(err) => { + log_child_result(job_id, Err(err)); + false + } + }); +} + +/// Log a failed run's exit status. Runs log their own outcome, but a killed one cannot. +fn log_child_result(job_id: &str, result: std::io::Result) { + match result { + Ok(status) if status.success() => (), + Ok(status) => log::warn!("backup job '{job_id}' failed ({status})"), + Err(err) => log::error!("backup job '{job_id}': cannot determine outcome: {err}"), + } +} + +fn scheduler_pid_path() -> Result { + Ok(pbs_client::tools::base_directories()? + .get_state_home() + .join("scheduler.pid")) +} + +/// Lock the PID file and write our PID to it. The kernel drops the lock when the process exits. +fn acquire_scheduler_pidfile(path: &Path) -> Result, Error> { + use std::io::Write; + + if let Some(dir) = path.parent() { + proxmox_sys::fs::create_path(dir, None, None)?; + } + // not truncated before the lock is held, the file may belong to a running scheduler + let file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(path) + .map_err(|err| format_err!("cannot open {}: {err}", path.display()))?; + let mut locked = match Flock::lock(file, FlockArg::LockExclusiveNonblock) { + Ok(locked) => locked, + Err((_, nix::errno::Errno::EWOULDBLOCK)) => bail!( + "another backup job scheduler is already running ({} is locked)", + path.display(), + ), + Err((_, errno)) => bail!("cannot lock {}: {errno}", path.display()), + }; + locked.set_len(0)?; + write!(locked, "{}", std::process::id())?; + Ok(locked) +} diff --git a/proxmox-backup-client/src/main.rs b/proxmox-backup-client/src/main.rs index 5238a75ea..5979377a6 100644 --- a/proxmox-backup-client/src/main.rs +++ b/proxmox-backup-client/src/main.rs @@ -65,6 +65,7 @@ use pbs_key_config::{KeyConfig, decrypt_key, rsa_encrypt_key_config}; use pbs_tools::crypt_config::CryptConfig; use pbs_tools::json; +pub mod job; pub mod key; pub mod namespace; @@ -799,7 +800,7 @@ fn spawn_catalog_upload( )] /// Create (host) backup. #[allow(clippy::too_many_arguments)] -async fn create_backup( +pub(crate) async fn create_backup( param: Value, all_file_systems: bool, skip_lost_and_found: bool, @@ -2294,6 +2295,7 @@ fn main() { .insert("benchmark", benchmark_cmd_def) .insert("change-owner", change_owner_cmd_def) .insert("namespace", namespace::cli_map()) + .insert("job", job::cli()) .insert("group", group_mgmt_cli()) .alias(&["files"], &["snapshot", "files"]) .alias(&["forget"], &["snapshot", "forget"]) diff --git a/proxmox-backup-client/tests/job.rs b/proxmox-backup-client/tests/job.rs new file mode 100644 index 000000000..bb165434a --- /dev/null +++ b/proxmox-backup-client/tests/job.rs @@ -0,0 +1,263 @@ +//! End-to-end checks of `job run` against the built binary, for the locks, the job state and the +//! config on disk. + +use std::fs::File; +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use nix::errno::Errno; +use nix::fcntl::{Flock, FlockArg}; + +/// A private home directory, so that the jobs of whoever runs the tests stay untouched. +struct Home { + dir: PathBuf, +} + +impl Home { + fn new(name: &str) -> Self { + let dir = Path::new(env!("CARGO_TARGET_TMPDIR")) + .join(format!("job-{name}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + Self { dir } + } + + fn config_dir(&self) -> PathBuf { + self.dir.join("config/proxmox-backup") + } + + fn state_dir(&self) -> PathBuf { + self.dir.join("state/proxmox-backup/jobstates") + } + + fn state_file(&self, id: &str) -> PathBuf { + self.state_dir().join(format!("backup-job-{id}.json")) + } + + /// Write a job config holding a single job, with `extra` appended to its section. + fn write_job(&self, id: &str, extra: &str) { + std::fs::create_dir_all(self.config_dir()).unwrap(); + std::fs::write( + self.config_dir().join("jobs.cfg"), + format!( + "backup-job: {id}\n\tbackupspec etc.pxar:/etc\n\tdatastore store1\n\ + \tschedule daily\n{extra}" + ), + ) + .unwrap(); + } + + /// Record that the job was created just now, so that its next event lies ahead. + fn write_created_now(&self, id: &str) { + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs(); + std::fs::create_dir_all(self.state_dir()).unwrap(); + std::fs::write( + self.state_file(id), + format!("{{\"created\":{{\"time\":{now}}}}}"), + ) + .unwrap(); + } + + /// Write an executable hook script and return its path. + fn hook(&self, name: &str, body: &str) -> PathBuf { + use std::os::unix::fs::PermissionsExt; + + let path = self.dir.join(name); + std::fs::write(&path, body).unwrap(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)).unwrap(); + path + } + + fn client(&self) -> Command { + let mut command = Command::new(env!("CARGO_BIN_EXE_proxmox-backup-client")); + // no stray PBS_* variable from the caller + command + .env_clear() + .env("PATH", "/usr/bin:/bin") + .env("HOME", &self.dir) + .env("XDG_CONFIG_HOME", self.dir.join("config")) + .env("XDG_STATE_HOME", self.dir.join("state")); + command + } + + fn run(&self, args: &[&str]) -> Output { + self.client() + .arg("job") + .arg("run") + .args(args) + .output() + .unwrap() + } +} + +impl Drop for Home { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.dir); + } +} + +fn stderr(output: &Output) -> String { + String::from_utf8_lossy(&output.stderr).into_owned() +} + +/// A run the scheduler starts must check again that the job is due, a manual one must not. +#[test] +fn a_scheduled_run_starts_only_a_job_that_is_still_due() { + let home = Home::new("scheduled"); + let marker = home.dir.join("hook-ran"); + let hook = home.hook( + "mark.sh", + &format!("#!/bin/sh\ntouch '{}'\nexit 1\n", marker.display()), + ); + let hook_line = format!("\tpre-run-hook {}\n", hook.display()); + + // just created, so its next event lies ahead + home.write_job("nightly", &hook_line); + home.write_created_now("nightly"); + let out = home.run(&["nightly", "--scheduled"]); + assert!(out.status.success(), "{}", stderr(&out)); + assert!(!marker.exists(), "ran a job that is not due"); + assert!(stderr(&out).contains("no longer due"), "{}", stderr(&out)); + + // due, having never run, but disabled + std::fs::remove_file(home.state_file("nightly")).unwrap(); + home.write_job("nightly", &format!("{hook_line}\tdisable true\n")); + let out = home.run(&["nightly", "--scheduled"]); + assert!(out.status.success(), "{}", stderr(&out)); + assert!(!marker.exists(), "ran a disabled job"); + assert!(stderr(&out).contains("disabled"), "{}", stderr(&out)); + + // removed after the daemon picked it + let out = home.run(&["removed", "--scheduled"]); + assert!(out.status.success(), "{}", stderr(&out)); + assert!( + stderr(&out).contains("no longer exists"), + "{}", + stderr(&out) + ); + + // and a job that is due does run, or the checks above would prove nothing + home.write_job("nightly", &hook_line); + let out = home.run(&["nightly", "--scheduled"]); + assert!(out.status.success(), "{}", stderr(&out)); + assert!(marker.exists(), "did not run a job that is due"); + + // a manual run is not bound by the schedule + std::fs::remove_file(&marker).unwrap(); + home.write_created_now("nightly"); + let out = home.run(&["nightly"]); + assert!(out.status.success(), "{}", stderr(&out)); + assert!(marker.exists(), "a manual run did not run the job"); +} + +/// A run whose outcome cannot be recorded must not report success. +#[test] +fn a_run_whose_outcome_cannot_be_recorded_fails() { + let home = Home::new("unrecorded"); + let state = home.state_file("nightly"); + + // the hook puts a directory where the outcome goes, then asks to skip the run + let hook = home.hook( + "block.sh", + &format!( + "#!/bin/sh\nrm -f '{0}'\nmkdir '{0}'\nexit 1\n", + state.display() + ), + ); + home.write_job("nightly", &format!("\tpre-run-hook {}\n", hook.display())); + let out = home.run(&["nightly"]); + assert!(!out.status.success(), "reported a skip it did not record"); + assert!( + stderr(&out).contains("outcome was not recorded"), + "{}", + stderr(&out) + ); + assert!( + !stderr(&out).contains("was skipped by its pre-run hook"), + "{}", + stderr(&out) + ); + + // a failed run keeps its own error, and also says that it could not record it + std::fs::remove_dir(&state).unwrap(); + let hook = home.hook( + "block-and-fail.sh", + &format!( + "#!/bin/sh\nrm -f '{0}'\nmkdir '{0}'\nexit 2\n", + state.display() + ), + ); + home.write_job("nightly", &format!("\tpre-run-hook {}\n", hook.display())); + let out = home.run(&["nightly"]); + assert!(!out.status.success(), "{}", stderr(&out)); + assert!( + stderr(&out).contains("exited with status 2"), + "{}", + stderr(&out) + ); + assert!( + stderr(&out).contains("cannot record its outcome"), + "{}", + stderr(&out) + ); +} + +/// Try to take a lock that someone else might hold, reporting whether it was free. +fn lock_is_free(path: &Path) -> bool { + let file = File::options().read(true).write(true).open(path).unwrap(); + match Flock::lock(file, FlockArg::LockExclusiveNonblock) { + Ok(_lock) => true, + Err((_, Errno::EWOULDBLOCK)) => false, + Err((_, errno)) => panic!("cannot probe {}: {errno}", path.display()), + } +} + +/// A run holds the config lock until it has the job lock, so that `job remove` cannot slip in. +#[test] +fn a_run_holds_the_config_lock_while_it_waits_for_the_job() { + let home = Home::new("lock-order"); + home.write_job("nightly", ""); + + // hold the job lock, as a run of the same job would + std::fs::create_dir_all(home.state_dir()).unwrap(); + let job_lock_path = home.state_dir().join(".backup-job-nightly.lck"); + let job_lock = Flock::lock( + File::create(&job_lock_path).unwrap(), + FlockArg::LockExclusive, + ) + .map_err(|(_, errno)| errno) + .unwrap(); + + let config_lock = home.config_dir().join(".jobs.lck"); + let run = home + .client() + .args(["job", "run", "nightly"]) + .stderr(std::process::Stdio::piped()) + .spawn() + .unwrap(); + + // still held a second later, well into the wait, not just while reading the config + let started = Instant::now(); + while !config_lock.exists() || lock_is_free(&config_lock) { + assert!( + started.elapsed() < Duration::from_secs(2), + "the run never took the config lock" + ); + std::thread::sleep(Duration::from_millis(20)); + } + std::thread::sleep(Duration::from_secs(1)); + assert!( + !lock_is_free(&config_lock), + "the run let go of the config lock while it waited for the job" + ); + + // keep the job locked until the run gives up + let out = run.wait_with_output().unwrap(); + drop(job_lock); + assert!(!out.status.success()); + assert!(stderr(&out).contains("already running"), "{}", stderr(&out)); +} -- 2.47.3