]> git.proxmox.com Git - proxmox-backup.git/blobdiff - src/bin/proxmox-backup-proxy.rs
server: implement access log rotation with re-open via command socket
[proxmox-backup.git] / src / bin / proxmox-backup-proxy.rs
index abc7b4ac8de7a124dc80aa15532c3b60d1b92f66..060752d4e51875bc86789c974267e798132760e2 100644 (file)
@@ -1,6 +1,6 @@
-use std::sync::Arc;
-use std::ffi::OsString;
-use std::path::Path;
+use std::sync::{Arc};
+use std::path::{Path, PathBuf};
+use std::os::unix::io::AsRawFd;
 
 use anyhow::{bail, format_err, Error};
 use futures::*;
@@ -9,21 +9,62 @@ use openssl::ssl::{SslMethod, SslAcceptor, SslFiletype};
 
 use proxmox::try_block;
 use proxmox::api::RpcEnvironmentType;
-use proxmox::sys::linux::procfs::mountinfo::{Device, MountInfo};
 
+use proxmox_backup::{
+    backup::DataStore,
+    server::{
+        UPID,
+        WorkerTask,
+        ApiConfig,
+        rest::*,
+        jobstate::{
+            self,
+            Job,
+        },
+        rotate_task_log_archive,
+    },
+    tools::systemd::time::{
+        parse_calendar_event,
+        compute_next_event,
+    },
+};
+
+
+use proxmox_backup::api2::types::Authid;
 use proxmox_backup::configdir;
 use proxmox_backup::buildcfg;
 use proxmox_backup::server;
-use proxmox_backup::tools::daemon;
-use proxmox_backup::server::{ApiConfig, rest::*};
 use proxmox_backup::auth_helpers::*;
-use proxmox_backup::tools::disks::{ DiskManage, zfs::zfs_pool_stats };
-
-fn main() {
-    if let Err(err) = proxmox_backup::tools::runtime::main(run()) {
-        eprintln!("Error: {}", err);
-        std::process::exit(-1);
+use proxmox_backup::tools::{
+    daemon,
+    disks::{
+        DiskManage,
+        zfs_pool_stats,
+    },
+    logrotate::LogRotate,
+    socket::{
+        set_tcp_keepalive,
+        PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
+    },
+};
+
+use proxmox_backup::api2::pull::do_sync_job;
+use proxmox_backup::server::do_verification_job;
+use proxmox_backup::server::do_prune_job;
+
+fn main() -> Result<(), Error> {
+    proxmox_backup::tools::setup_safe_path_env();
+
+    let backup_uid = proxmox_backup::backup::backup_user()?.uid;
+    let backup_gid = proxmox_backup::backup::backup_group()?.gid;
+    let running_uid = nix::unistd::Uid::effective();
+    let running_gid = nix::unistd::Gid::effective();
+
+    if running_uid != backup_uid || running_gid != backup_gid {
+        bail!("proxy not running as backup user or group (got uid {} gid {})", running_uid, running_gid);
     }
+
+    proxmox_backup::tools::runtime::main(run())
 }
 
 async fn run() -> Result<(), Error> {
@@ -40,26 +81,31 @@ async fn run() -> Result<(), Error> {
     let mut config = ApiConfig::new(
         buildcfg::JS_DIR, &proxmox_backup::api2::ROUTER, RpcEnvironmentType::PUBLIC)?;
 
-    // add default dirs which includes jquery and bootstrap
-    // my $base = '/usr/share/libpve-http-server-perl';
-    // add_dirs($self->{dirs}, '/css/' => "$base/css/");
-    // add_dirs($self->{dirs}, '/js/' => "$base/js/");
-    // add_dirs($self->{dirs}, '/fonts/' => "$base/fonts/");
     config.add_alias("novnc", "/usr/share/novnc-pve");
     config.add_alias("extjs", "/usr/share/javascript/extjs");
     config.add_alias("fontawesome", "/usr/share/fonts-font-awesome");
     config.add_alias("xtermjs", "/usr/share/pve-xtermjs");
+    config.add_alias("locale", "/usr/share/pbs-i18n");
     config.add_alias("widgettoolkit", "/usr/share/javascript/proxmox-widget-toolkit");
     config.add_alias("css", "/usr/share/javascript/proxmox-backup/css");
     config.add_alias("docs", "/usr/share/doc/proxmox-backup/html");
 
+    let mut indexpath = PathBuf::from(buildcfg::JS_DIR);
+    indexpath.push("index.hbs");
+    config.register_template("index", &indexpath)?;
+    config.register_template("console", "/usr/share/pve-xtermjs/index.html.hbs")?;
+
+    let mut commando_sock = server::CommandoSocket::new(server::our_ctrl_sock());
+
+    config.enable_file_log(buildcfg::API_ACCESS_LOG_FN, &mut commando_sock)?;
+
     let rest_server = RestServer::new(config);
 
     //openssl req -x509 -newkey rsa:4096 -keyout /etc/proxmox-backup/proxy.key -out /etc/proxmox-backup/proxy.pem -nodes
     let key_path = configdir!("/proxy.key");
     let cert_path = configdir!("/proxy.pem");
 
-    let mut acceptor = SslAcceptor::mozilla_intermediate(SslMethod::tls()).unwrap();
+    let mut acceptor = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls()).unwrap();
     acceptor.set_private_key_file(key_path, SslFiletype::PEM)
         .map_err(|err| format_err!("unable to read proxy key {} - {}", key_path, err))?;
     acceptor.set_certificate_chain_file(cert_path)
@@ -77,8 +123,9 @@ async fn run() -> Result<(), Error> {
                     let acceptor = Arc::clone(&acceptor);
                     async move {
                         sock.set_nodelay(true).unwrap();
-                        sock.set_send_buffer_size(1024*1024).unwrap();
-                        sock.set_recv_buffer_size(1024*1024).unwrap();
+
+                        let _ = set_tcp_keepalive(sock.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
+
                         Ok(tokio_openssl::accept(&acceptor, sock)
                             .await
                             .ok() // handshake errors aren't be fatal, so return None to filter
@@ -99,10 +146,12 @@ async fn run() -> Result<(), Error> {
         },
     );
 
+    server::write_pid(buildcfg::PROXMOX_BACKUP_PROXY_PID_FN)?;
     daemon::systemd_notify(daemon::SystemdNotify::Ready)?;
 
     let init_result: Result<(), Error> = try_block!({
-        server::create_task_control_socket()?;
+        server::register_task_control_commands(&mut commando_sock)?;
+        commando_sock.spawn()?;
         server::server_state_init()?;
         Ok(())
     });
@@ -136,11 +185,12 @@ fn start_task_scheduler() {
     tokio::spawn(task.map(|_| ()));
 }
 
-use std::time:: {Instant, Duration, SystemTime, UNIX_EPOCH};
+use std::time::{SystemTime, Instant, Duration, UNIX_EPOCH};
 
 fn next_minute() -> Result<Instant, Error> {
-    let epoch_now = SystemTime::now().duration_since(UNIX_EPOCH)?;
-    let epoch_next = Duration::from_secs((epoch_now.as_secs()/60 + 1)*60);
+    let now = SystemTime::now();
+    let epoch_now = now.duration_since(UNIX_EPOCH)?;
+    let epoch_next = Duration::from_secs((epoch_now.as_secs()/60  + 1)*60);
     Ok(Instant::now() + epoch_next - epoch_now)
 }
 
@@ -188,46 +238,20 @@ async fn schedule_tasks() -> Result<(), Error> {
     schedule_datastore_garbage_collection().await;
     schedule_datastore_prune().await;
     schedule_datastore_sync_jobs().await;
+    schedule_datastore_verify_jobs().await;
+    schedule_task_log_rotate().await;
 
     Ok(())
 }
 
-fn lookup_last_worker(worker_type: &str, worker_id: &str) -> Result<Option<server::UPID>, Error> {
-
-    let list = proxmox_backup::server::read_task_list()?;
-
-    let mut last: Option<&server::UPID> = None;
-
-    for entry in list.iter() {
-        if entry.upid.worker_type == worker_type {
-            if let Some(ref id) = entry.upid.worker_id {
-                if id == worker_id {
-                    match last {
-                        Some(ref upid) => {
-                            if upid.starttime < entry.upid.starttime {
-                                last = Some(&entry.upid)
-                            }
-                        }
-                        None => {
-                            last = Some(&entry.upid)
-                        }
-                    }
-                }
-            }
-        }
-    }
-
-    Ok(last.cloned())
-}
-
-
 async fn schedule_datastore_garbage_collection() {
 
-    use proxmox_backup::backup::DataStore;
-    use proxmox_backup::server::{UPID, WorkerTask};
-    use proxmox_backup::config::datastore::{self, DataStoreConfig};
-    use proxmox_backup::tools::systemd::time::{
-        parse_calendar_event, compute_next_event};
+    use proxmox_backup::config::{
+        datastore::{
+            self,
+            DataStoreConfig,
+        },
+    };
 
     let config = match datastore::config() {
         Err(err) => {
@@ -281,58 +305,52 @@ async fn schedule_datastore_garbage_collection() {
                 }
             }
         } else {
-            match lookup_last_worker(worker_type, &store) {
-                Ok(Some(upid)) => upid.starttime,
-                Ok(None) => 0,
+            match jobstate::last_run_time(worker_type, &store) {
+                Ok(time) => time,
                 Err(err) => {
-                    eprintln!("lookup_last_job_start failed: {}", err);
+                    eprintln!("could not get last run time of {} {}: {}", worker_type, store, err);
                     continue;
                 }
             }
         };
 
         let next = match compute_next_event(&event, last, false) {
-            Ok(next) => next,
+            Ok(Some(next)) => next,
+            Ok(None) => continue,
             Err(err) => {
                 eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
                 continue;
             }
         };
-        let now = match SystemTime::now().duration_since(UNIX_EPOCH) {
-            Ok(epoch_now) => epoch_now.as_secs() as i64,
-            Err(err) => {
-                eprintln!("query system time failed - {}", err);
-                continue;
-            }
-        };
+
+        let now = proxmox::tools::time::epoch_i64();
+
         if next > now  { continue; }
 
-        let store2 = store.clone();
-
-        if let Err(err) = WorkerTask::new_thread(
-            worker_type,
-            Some(store.clone()),
-            "backup@pam",
-            false,
-            move |worker| {
-                worker.log(format!("starting garbage collection on store {}", store));
-                worker.log(format!("task triggered by schedule '{}'", event_str));
-                datastore.garbage_collection(&worker)
-            }
-        ) {
-            eprintln!("unable to start garbage collection on store {} - {}", store2, err);
+        let job = match Job::new(worker_type, &store) {
+            Ok(job) => job,
+            Err(_) => continue, // could not get lock
+        };
+
+        let auth_id = Authid::backup_auth_id();
+
+        if let Err(err) = crate::server::do_garbage_collection_job(job, datastore, auth_id, Some(event_str)) {
+            eprintln!("unable to start garbage collection job on datastore {} - {}", store, err);
         }
     }
 }
 
 async fn schedule_datastore_prune() {
 
-    use proxmox_backup::backup::{
-        PruneOptions, DataStore, BackupGroup, BackupDir, compute_prune_info};
-    use proxmox_backup::server::{WorkerTask};
-    use proxmox_backup::config::datastore::{self, DataStoreConfig};
-    use proxmox_backup::tools::systemd::time::{
-        parse_calendar_event, compute_next_event};
+    use proxmox_backup::{
+        backup::{
+            PruneOptions,
+        },
+        config::datastore::{
+            self,
+            DataStoreConfig,
+        },
+    };
 
     let config = match datastore::config() {
         Err(err) => {
@@ -343,13 +361,6 @@ async fn schedule_datastore_prune() {
     };
 
     for (store, (_, store_config)) in config.sections {
-        let datastore = match DataStore::lookup_datastore(&store) {
-            Ok(datastore) => datastore,
-            Err(err) => {
-                eprintln!("lookup_datastore '{}' failed - {}", store, err);
-                continue;
-            }
-        };
 
         let store_config: DataStoreConfig = match serde_json::from_value(store_config) {
             Ok(c) => c,
@@ -377,99 +388,26 @@ async fn schedule_datastore_prune() {
             continue;
         }
 
-        let event = match parse_calendar_event(&event_str) {
-            Ok(event) => event,
-            Err(err) => {
-                eprintln!("unable to parse schedule '{}' - {}", event_str, err);
-                continue;
-            }
-        };
-
         let worker_type = "prune";
+        if check_schedule(worker_type, &event_str, &store) {
+            let job = match Job::new(worker_type, &store) {
+                Ok(job) => job,
+                Err(_) => continue, // could not get lock
+            };
 
-        let last = match lookup_last_worker(worker_type, &store) {
-            Ok(Some(upid)) => {
-                if proxmox_backup::server::worker_is_active_local(&upid) {
-                    continue;
-                }
-                upid.starttime
-            }
-            Ok(None) => 0,
-            Err(err) => {
-                eprintln!("lookup_last_job_start failed: {}", err);
-                continue;
-            }
-        };
-
-        let next = match compute_next_event(&event, last, false) {
-            Ok(next) => next,
-            Err(err) => {
-                eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
-                continue;
-            }
-        };
-
-        let now = match SystemTime::now().duration_since(UNIX_EPOCH) {
-            Ok(epoch_now) => epoch_now.as_secs() as i64,
-            Err(err) => {
-                eprintln!("query system time failed - {}", err);
-                continue;
+            let auth_id = Authid::backup_auth_id().clone();
+            if let Err(err) = do_prune_job(job, prune_options, store.clone(), &auth_id, Some(event_str)) {
+                eprintln!("unable to start datastore prune job {} - {}", &store, err);
             }
         };
-        if next > now  { continue; }
-
-        let store2 = store.clone();
-
-        if let Err(err) = WorkerTask::new_thread(
-            worker_type,
-            Some(store.clone()),
-            "backup@pam",
-            false,
-            move |worker| {
-                worker.log(format!("Starting datastore prune on store \"{}\"", store));
-                worker.log(format!("task triggered by schedule '{}'", event_str));
-                worker.log(format!("retention options: {}", prune_options.cli_options_string()));
-
-                let base_path = datastore.base_path();
-
-                let groups = BackupGroup::list_groups(&base_path)?;
-                for group in groups {
-                    let list = group.list_backups(&base_path)?;
-                    let mut prune_info = compute_prune_info(list, &prune_options)?;
-                    prune_info.reverse(); // delete older snapshots first
-
-                    worker.log(format!("Starting prune on store \"{}\" group \"{}/{}\"",
-                                       store, group.backup_type(), group.backup_id()));
-
-                    for (info, keep) in prune_info {
-                        worker.log(format!(
-                            "{} {}/{}/{}",
-                            if keep { "keep" } else { "remove" },
-                            group.backup_type(), group.backup_id(),
-                            BackupDir::backup_time_to_string(info.backup_dir.backup_time())));
-
-                        if !keep {
-                            datastore.remove_backup_dir(&info.backup_dir)?;
-                        }
-                    }
-                }
-
-                Ok(())
-            }
-        ) {
-            eprintln!("unable to start datastore prune on store {} - {}", store2, err);
-        }
     }
 }
 
 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} },
-        tools::systemd::time::{ parse_calendar_event, compute_next_event },
+    use proxmox_backup::config::sync::{
+        self,
+        SyncJobConfig,
     };
 
     let config = match sync::config() {
@@ -480,14 +418,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,
@@ -502,100 +432,152 @@ async fn schedule_datastore_sync_jobs() {
             None => continue,
         };
 
-        let event = match parse_calendar_event(&event_str) {
-            Ok(event) => event,
-            Err(err) => {
-                eprintln!("unable to parse schedule '{}' - {}", event_str, err);
-                continue;
-            }
-        };
-
         let worker_type = "syncjob";
+        if check_schedule(worker_type, &event_str, &job_id) {
+            let job = match Job::new(worker_type, &job_id) {
+                Ok(job) => job,
+                Err(_) => continue, // could not get lock
+            };
 
-        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,
-            Err(err) => {
-                eprintln!("lookup_last_job_start failed: {}", err);
-                continue;
+            let auth_id = Authid::backup_auth_id().clone();
+            if let Err(err) = do_sync_job(job, job_config, &auth_id, Some(event_str)) {
+                eprintln!("unable to start datastore sync job {} - {}", &job_id, err);
             }
         };
+    }
+}
 
-        let next = match compute_next_event(&event, last, false) {
-            Ok(next) => next,
+async fn schedule_datastore_verify_jobs() {
+
+    use proxmox_backup::config::verify::{
+        self,
+        VerificationJobConfig,
+    };
+
+    let config = match verify::config() {
+        Err(err) => {
+            eprintln!("unable to read verification job config - {}", err);
+            return;
+        }
+        Ok((config, _digest)) => config,
+    };
+    for (job_id, (_, job_config)) in config.sections {
+        let job_config: VerificationJobConfig = match serde_json::from_value(job_config) {
+            Ok(c) => c,
             Err(err) => {
-                eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
+                eprintln!("verification job config from_value failed - {}", err);
                 continue;
             }
         };
+        let event_str = match job_config.schedule {
+            Some(ref event_str) => event_str.clone(),
+            None => continue,
+        };
 
-        let now = match SystemTime::now().duration_since(UNIX_EPOCH) {
-            Ok(epoch_now) => epoch_now.as_secs() as i64,
-            Err(err) => {
-                eprintln!("query system time failed - {}", err);
-                continue;
+        let worker_type = "verificationjob";
+        let auth_id = Authid::backup_auth_id().clone();
+        if check_schedule(worker_type, &event_str, &job_id) {
+            let job = match Job::new(&worker_type, &job_id) {
+                Ok(job) => job,
+                Err(_) => continue, // could not get lock
+            };
+            if let Err(err) = do_verification_job(job, job_config, &auth_id, Some(event_str)) {
+                eprintln!("unable to start datastore verification job {} - {}", &job_id, err);
             }
         };
-        if next > now  { continue; }
+    }
+}
 
+async fn schedule_task_log_rotate() {
 
-        let job_id2 = job_id.clone();
+    let worker_type = "logrotate";
+    let job_id = "task_archive";
 
-        let tgt_store = match DataStore::lookup_datastore(&job_config.store) {
-            Ok(datastore) => datastore,
-            Err(err) => {
-                eprintln!("lookup_datastore '{}' failed - {}", job_config.store, err);
-                continue;
-            }
-        };
+    // schedule daily at 00:00 like normal logrotate
+    let schedule = "00:00";
 
-        let remote: Remote = match remote_config.lookup("remote", &job_config.remote) {
-            Ok(remote) => remote,
-            Err(err) => {
-                eprintln!("remote_config lookup failed: {}", err);
-                continue;
-            }
-        };
+    if !check_schedule(worker_type, schedule, job_id) {
+        // if we never ran the rotation, schedule instantly
+        match jobstate::JobState::load(worker_type, job_id) {
+            Ok(state) => match state {
+                jobstate::JobState::Created { .. } => {},
+                _ => return,
+            },
+            _ => return,
+        }
+    }
 
-        let username = String::from("backup@pam");
+    let mut job = match Job::new(worker_type, job_id) {
+        Ok(job) => job,
+        Err(_) => return, // could not get lock
+    };
 
-        let delete = job_config.remove_vanished.unwrap_or(true);
+    if let Err(err) = WorkerTask::new_thread(
+        worker_type,
+        Some(job_id.to_string()),
+        Authid::backup_auth_id().clone(),
+        false,
+        move |worker| {
+            job.start(&worker.upid().to_string())?;
+            worker.log(format!("starting task log rotation"));
+
+            let result = try_block!({
+                // rotate task log archive
+                let max_size = 500000; // a normal entry has about 100b, so ~ 5000 entries/file
+                let max_files = 20; // times twenty files gives at least 100000 task entries
+                let has_rotated = rotate_task_log_archive(max_size, true, Some(max_files))?;
+                if has_rotated {
+                    worker.log(format!("task log archive was rotated"));
+                } else {
+                    worker.log(format!("task log archive was not rotated"));
+                }
 
-        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 max_size = 32 * 1024 * 1024 - 1;
+                let max_files = 14;
+                let mut logrotate = LogRotate::new(buildcfg::API_ACCESS_LOG_FN, true)
+                        .ok_or_else(|| format_err!("could not get API access log file names"))?;
 
-                let options = HttpClientOptions::new()
-                    .password(Some(remote.password.clone()))
-                    .fingerprint(remote.fingerprint.clone());
+                let has_rotated = logrotate.rotate(max_size, None, Some(max_files))?;
+                if has_rotated {
+                    println!("rotated access log, telling daemons to re-open log file");
+                    proxmox_backup::tools::runtime::block_on(command_reopen_logfiles())?;
 
-                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))?;
+                    worker.log(format!("API access log was rotated"));
+                } else {
+                    worker.log(format!("API access log was not rotated"));
+                }
 
-                let src_repo = BackupRepository::new(Some(remote.userid), Some(remote.host), job_config.remote_store);
+                Ok(())
+            });
 
-                pull_store(&worker, &client, &src_repo, tgt_store, delete, username).await?;
+            let status = worker.create_state(&result);
 
-                Ok(())
+            if let Err(err) = job.finish(status) {
+                eprintln!("could not finish job state for {}: {}", worker_type, err);
             }
-        ) {
-            eprintln!("unable to start datastore sync job {} - {}", job_id2, err);
-        }
+
+            result
+        },
+    ) {
+        eprintln!("unable to start task log rotation: {}", err);
     }
+
+}
+
+async fn command_reopen_logfiles() -> Result<(), Error> {
+    // only care about the most recent daemon instance for each, proxy & api, as other older ones
+    // should not respond to new requests anyway, but only finish their current one and then exit.
+    let sock = server::our_ctrl_sock();
+    server::send_command(sock, serde_json::json!({
+        "command": "api-access-log-reopen",
+    })).await?;
+
+    let pid = server::read_pid(buildcfg::PROXMOX_BACKUP_API_PID_FN)?;
+    let sock = server::ctrl_sock_from_pid(pid);
+    server::send_command(sock, serde_json::json!({
+        "command": "api-access-log-reopen",
+    })).await?;
+    Ok(())
 }
 
 async fn run_stat_generator() {
@@ -603,7 +585,7 @@ async fn run_stat_generator() {
     let mut count = 0;
     loop {
         count += 1;
-        let save = if count > 6 { count = 0; true } else { false };
+        let save = if count >= 6 { count = 0; true } else { false };
 
         let delay_target = Instant::now() +  Duration::from_secs(10);
 
@@ -710,14 +692,44 @@ async fn generate_host_stats(save: bool) {
     });
 }
 
+fn check_schedule(worker_type: &str, event_str: &str, id: &str) -> bool {
+    let event = match parse_calendar_event(event_str) {
+        Ok(event) => event,
+        Err(err) => {
+            eprintln!("unable to parse schedule '{}' - {}", event_str, err);
+            return false;
+        }
+    };
+
+    let last = match jobstate::last_run_time(worker_type, &id) {
+        Ok(time) => time,
+        Err(err) => {
+            eprintln!("could not get last run time of {} {}: {}", worker_type, id, err);
+            return false;
+        }
+    };
+
+    let next = match compute_next_event(&event, last, false) {
+        Ok(Some(next)) => next,
+        Ok(None) => return false,
+        Err(err) => {
+            eprintln!("compute_next_event for '{}' failed - {}", event_str, err);
+            return false;
+        }
+    };
+
+    let now = proxmox::tools::time::epoch_i64();
+    next <= now
+}
+
 fn gather_disk_stats(disk_manager: Arc<DiskManage>, path: &Path, rrd_prefix: &str, save: bool) {
 
     match proxmox_backup::tools::disks::disk_usage(path) {
-        Ok((total, used, _avail)) => {
+        Ok(status) => {
             let rrd_key = format!("{}/total", rrd_prefix);
-            rrd_update_gauge(&rrd_key, total as f64, save);
+            rrd_update_gauge(&rrd_key, status.total as f64, save);
             let rrd_key = format!("{}/used", rrd_prefix);
-            rrd_update_gauge(&rrd_key, used as f64, save);
+            rrd_update_gauge(&rrd_key, status.used as f64, save);
         }
         Err(err) => {
             eprintln!("read disk_usage on {:?} failed - {}", path, err);