src/server/worker_task.rs

   1 use std::collections::{HashMap, VecDeque};
   2 use std::fs::File;
   3 use std::io::{Read, Write, BufRead, BufReader};
   4 use std::panic::UnwindSafe;
   5 use std::sync::atomic::{AtomicBool, Ordering};
   6 use std::sync::{Arc, Mutex};
   7
   8 use anyhow::{bail, format_err, Error};
   9 use futures::*;
  10 use lazy_static::lazy_static;
  11 use serde_json::{json, Value};
  12 use serde::{Serialize, Deserialize};
  13 use tokio::sync::oneshot;
  14
  15 use proxmox::sys::linux::procfs;
  16 use proxmox::try_block;
  17 use proxmox::tools::fs::{create_path, replace_file, CreateOptions};
  18
  19 use pbs_buildcfg;
  20 use pbs_tools::logrotate::{LogRotate, LogRotateFiles};
  21 use pbs_api_types::{Authid, TaskStateType, UPID};
  22 use pbs_config::{open_backup_lockfile, BackupLockGuard};
  23 use proxmox_rest_server::{CommandoSocket, FileLogger, FileLogOptions};
  24
  25 use super::UPIDExt;
  26
  27 macro_rules! taskdir {
  28     ($subdir:expr) => (concat!(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR_M!(), "/tasks", $subdir))
  29 }
  30 pub const PROXMOX_BACKUP_TASK_DIR: &str = taskdir!("/");
  31 pub const PROXMOX_BACKUP_TASK_LOCK_FN: &str = taskdir!("/.active.lock");
  32 pub const PROXMOX_BACKUP_ACTIVE_TASK_FN: &str = taskdir!("/active");
  33 pub const PROXMOX_BACKUP_INDEX_TASK_FN: &str = taskdir!("/index");
  34 pub const PROXMOX_BACKUP_ARCHIVE_TASK_FN: &str = taskdir!("/archive");
  35
  36 lazy_static! {
  37     static ref WORKER_TASK_LIST: Mutex<HashMap<usize, Arc<WorkerTask>>> = Mutex::new(HashMap::new());
  38 }
  39
  40 /// checks if the task UPID refers to a worker from this process
  41 fn is_local_worker(upid: &UPID) -> bool {
  42     upid.pid == crate::server::pid() && upid.pstart == crate::server::pstart()
  43 }
  44
  45 /// Test if the task is still running
  46 pub async fn worker_is_active(upid: &UPID) -> Result<bool, Error> {
  47     if is_local_worker(upid) {
  48         return Ok(WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id));
  49     }
  50
  51     if procfs::check_process_running_pstart(upid.pid, upid.pstart).is_none() {
  52         return Ok(false);
  53     }
  54
  55     let sock = crate::server::ctrl_sock_from_pid(upid.pid);
  56     let cmd = json!({
  57         "command": "worker-task-status",
  58         "args": {
  59             "upid": upid.to_string(),
  60         },
  61     });
  62     let status = proxmox_rest_server::send_command(sock, &cmd).await?;
  63
  64     if let Some(active) = status.as_bool() {
  65         Ok(active)
  66     } else {
  67         bail!("got unexpected result {:?} (expected bool)", status);
  68     }
  69 }
  70
  71 /// Test if the task is still running (fast but inaccurate implementation)
  72 ///
  73 /// If the task is spawned from a different process, we simply return if
  74 /// that process is still running. This information is good enough to detect
  75 /// stale tasks...
  76 pub fn worker_is_active_local(upid: &UPID) -> bool {
  77     if is_local_worker(upid) {
  78         WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id)
  79     } else {
  80         procfs::check_process_running_pstart(upid.pid, upid.pstart).is_some()
  81     }
  82 }
  83
  84 pub fn register_task_control_commands(
  85     commando_sock: &mut CommandoSocket,
  86 ) -> Result<(), Error> {
  87     fn get_upid(args: Option<&Value>) -> Result<UPID, Error> {
  88         let args = if let Some(args) = args { args } else { bail!("missing args") };
  89         let upid = match args.get("upid") {
  90             Some(Value::String(upid)) => upid.parse::<UPID>()?,
  91             None => bail!("no upid in args"),
  92             _ => bail!("unable to parse upid"),
  93         };
  94         if !is_local_worker(&upid) {
  95             bail!("upid does not belong to this process");
  96         }
  97         Ok(upid)
  98     }
  99
 100     commando_sock.register_command("worker-task-abort".into(), move |args| {
 101         let upid = get_upid(args)?;
 102
 103         if let Some(ref worker) = WORKER_TASK_LIST.lock().unwrap().get(&upid.task_id) {
 104             worker.request_abort();
 105         }
 106         Ok(Value::Null)
 107     })?;
 108     commando_sock.register_command("worker-task-status".into(), move |args| {
 109         let upid = get_upid(args)?;
 110
 111         let active = WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id);
 112
 113         Ok(active.into())
 114     })?;
 115
 116     Ok(())
 117 }
 118
 119 pub fn abort_worker_async(upid: UPID) {
 120     tokio::spawn(async move {
 121         if let Err(err) = abort_worker(upid).await {
 122             eprintln!("abort worker failed - {}", err);
 123         }
 124     });
 125 }
 126
 127 pub async fn abort_worker(upid: UPID) -> Result<(), Error> {
 128
 129     let sock = crate::server::ctrl_sock_from_pid(upid.pid);
 130     let cmd = json!({
 131         "command": "worker-task-abort",
 132         "args": {
 133             "upid": upid.to_string(),
 134         },
 135     });
 136     proxmox_rest_server::send_command(sock, &cmd).map_ok(|_| ()).await
 137 }
 138
 139 fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<TaskState>), Error> {
 140
 141     let data = line.splitn(3, ' ').collect::<Vec<&str>>();
 142
 143     let len = data.len();
 144
 145     match len {
 146         1 => Ok((data[0].to_owned(), data[0].parse::<UPID>()?, None)),
 147         3 => {
 148             let endtime = i64::from_str_radix(data[1], 16)?;
 149             let state = TaskState::from_endtime_and_message(endtime, data[2])?;
 150             Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some(state)))
 151         }
 152         _ => bail!("wrong number of components"),
 153     }
 154 }
 155
 156 /// Create task log directory with correct permissions
 157 pub fn create_task_log_dirs() -> Result<(), Error> {
 158
 159     try_block!({
 160         let backup_user = pbs_config::backup_user()?;
 161         let opts = CreateOptions::new()
 162             .owner(backup_user.uid)
 163             .group(backup_user.gid);
 164
 165         create_path(pbs_buildcfg::PROXMOX_BACKUP_LOG_DIR, None, Some(opts.clone()))?;
 166         create_path(PROXMOX_BACKUP_TASK_DIR, None, Some(opts.clone()))?;
 167         create_path(pbs_buildcfg::PROXMOX_BACKUP_RUN_DIR, None, Some(opts))?;
 168         Ok(())
 169     }).map_err(|err: Error| format_err!("unable to create task log dir - {}", err))?;
 170
 171     Ok(())
 172 }
 173
 174 /// Read endtime (time of last log line) and exitstatus from task log file
 175 /// If there is not a single line with at valid datetime, we assume the
 176 /// starttime to be the endtime
 177 pub fn upid_read_status(upid: &UPID) -> Result<TaskState, Error> {
 178
 179     let mut status = TaskState::Unknown { endtime: upid.starttime };
 180
 181     let path = upid.log_path();
 182
 183     let mut file = File::open(path)?;
 184
 185     /// speedup - only read tail
 186     use std::io::Seek;
 187     use std::io::SeekFrom;
 188     let _ = file.seek(SeekFrom::End(-8192)); // ignore errors
 189
 190     let mut data = Vec::with_capacity(8192);
 191     file.read_to_end(&mut data)?;
 192
 193     // strip newlines at the end of the task logs
 194     while data.last() == Some(&b'\n') {
 195         data.pop();
 196     }
 197
 198     let last_line = match data.iter().rposition(|c| *c == b'\n') {
 199         Some(start) if data.len() > (start+1) => &data[start+1..],
 200         Some(_) => &data, // should not happen, since we removed all trailing newlines
 201         None => &data,
 202     };
 203
 204     let last_line = std::str::from_utf8(last_line)
 205         .map_err(|err| format_err!("upid_read_status: utf8 parse failed: {}", err))?;
 206
 207     let mut iter = last_line.splitn(2, ": ");
 208     if let Some(time_str) = iter.next() {
 209         if let Ok(endtime) = proxmox::tools::time::parse_rfc3339(time_str) {
 210             // set the endtime even if we cannot parse the state
 211             status = TaskState::Unknown { endtime };
 212             if let Some(rest) = iter.next().and_then(|rest| rest.strip_prefix("TASK ")) {
 213                 if let Ok(state) = TaskState::from_endtime_and_message(endtime, rest) {
 214                     status = state;
 215                 }
 216             }
 217         }
 218     }
 219
 220     Ok(status)
 221 }
 222
 223 /// Task State
 224 #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)]
 225 pub enum TaskState {
 226     /// The Task ended with an undefined state
 227     Unknown { endtime: i64 },
 228     /// The Task ended and there were no errors or warnings
 229     OK { endtime: i64 },
 230     /// The Task had 'count' amount of warnings and no errors
 231     Warning { count: u64, endtime: i64 },
 232     /// The Task ended with the error described in 'message'
 233     Error { message: String, endtime: i64 },
 234 }
 235
 236 impl TaskState {
 237     pub fn endtime(&self) -> i64 {
 238         match *self {
 239             TaskState::Unknown { endtime } => endtime,
 240             TaskState::OK { endtime } => endtime,
 241             TaskState::Warning { endtime, .. } => endtime,
 242             TaskState::Error { endtime, .. } => endtime,
 243         }
 244     }
 245
 246     pub fn tasktype(&self) -> TaskStateType {
 247         match self {
 248             TaskState::OK { .. } => TaskStateType::OK,
 249             TaskState::Unknown { .. } => TaskStateType::Unknown,
 250             TaskState::Error { .. } => TaskStateType::Error,
 251             TaskState::Warning { .. } => TaskStateType::Warning,
 252         }
 253     }
 254
 255     fn result_text(&self) -> String {
 256         match self {
 257             TaskState::Error { message, .. } => format!("TASK ERROR: {}", message),
 258             other => format!("TASK {}", other),
 259         }
 260     }
 261
 262     fn from_endtime_and_message(endtime: i64, s: &str) -> Result<Self, Error> {
 263         if s == "unknown" {
 264             Ok(TaskState::Unknown { endtime })
 265         } else if s == "OK" {
 266             Ok(TaskState::OK { endtime })
 267         } else if let Some(warnings) = s.strip_prefix("WARNINGS: ") {
 268             let count: u64 = warnings.parse()?;
 269             Ok(TaskState::Warning{ count, endtime })
 270         } else if !s.is_empty() {
 271             let message = if let Some(err) = s.strip_prefix("ERROR: ") { err } else { s }.to_string();
 272             Ok(TaskState::Error{ message, endtime })
 273         } else {
 274             bail!("unable to parse Task Status '{}'", s);
 275         }
 276     }
 277 }
 278
 279 impl std::cmp::PartialOrd for TaskState {
 280     fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
 281         Some(self.endtime().cmp(&other.endtime()))
 282     }
 283 }
 284
 285 impl std::cmp::Ord for TaskState {
 286     fn cmp(&self, other: &Self) -> std::cmp::Ordering {
 287         self.endtime().cmp(&other.endtime())
 288     }
 289 }
 290
 291 impl std::fmt::Display for TaskState {
 292     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
 293         match self {
 294             TaskState::Unknown { .. } => write!(f, "unknown"),
 295             TaskState::OK { .. }=> write!(f, "OK"),
 296             TaskState::Warning { count, .. } => write!(f, "WARNINGS: {}", count),
 297             TaskState::Error { message, .. } => write!(f, "{}", message),
 298         }
 299     }
 300 }
 301
 302 /// Task details including parsed UPID
 303 ///
 304 /// If there is no `state`, the task is still running.
 305 #[derive(Debug)]
 306 pub struct TaskListInfo {
 307     /// The parsed UPID
 308     pub upid: UPID,
 309     /// UPID string representation
 310     pub upid_str: String,
 311     /// Task `(endtime, status)` if already finished
 312     pub state: Option<TaskState>, // endtime, status
 313 }
 314
 315 impl Into<pbs_api_types::TaskListItem> for TaskListInfo {
 316     fn into(self) -> pbs_api_types::TaskListItem {
 317         let (endtime, status) = self
 318             .state
 319             .map_or_else(|| (None, None), |a| (Some(a.endtime()), Some(a.to_string())));
 320
 321         pbs_api_types::TaskListItem {
 322             upid: self.upid_str,
 323             node: "localhost".to_string(),
 324             pid: self.upid.pid as i64,
 325             pstart: self.upid.pstart,
 326             starttime: self.upid.starttime,
 327             worker_type: self.upid.worker_type,
 328             worker_id: self.upid.worker_id,
 329             user: self.upid.auth_id,
 330             endtime,
 331             status,
 332         }
 333     }
 334 }
 335
 336 fn lock_task_list_files(exclusive: bool) -> Result<BackupLockGuard, Error> {
 337     open_backup_lockfile(PROXMOX_BACKUP_TASK_LOCK_FN, None, exclusive)
 338 }
 339
 340 /// checks if the Task Archive is bigger that 'size_threshold' bytes, and
 341 /// rotates it if it is
 342 pub fn rotate_task_log_archive(size_threshold: u64, compress: bool, max_files: Option<usize>) -> Result<bool, Error> {
 343     let _lock = lock_task_list_files(true)?;
 344
 345     let mut logrotate = LogRotate::new(PROXMOX_BACKUP_ARCHIVE_TASK_FN, compress)
 346         .ok_or_else(|| format_err!("could not get archive file names"))?;
 347
 348     logrotate.rotate(size_threshold, None, max_files)
 349 }
 350
 351 // atomically read/update the task list, update status of finished tasks
 352 // new_upid is added to the list when specified.
 353 fn update_active_workers(new_upid: Option<&UPID>) -> Result<(), Error> {
 354
 355     let backup_user = pbs_config::backup_user()?;
 356
 357     let lock = lock_task_list_files(true)?;
 358
 359     // TODO remove with 1.x
 360     let mut finish_list: Vec<TaskListInfo> = read_task_file_from_path(PROXMOX_BACKUP_INDEX_TASK_FN)?;
 361     let had_index_file = !finish_list.is_empty();
 362
 363     // We use filter_map because one negative case wants to *move* the data into `finish_list`,
 364     // clippy doesn't quite catch this!
 365     #[allow(clippy::unnecessary_filter_map)]
 366     let mut active_list: Vec<TaskListInfo> = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?
 367         .into_iter()
 368         .filter_map(|info| {
 369             if info.state.is_some() {
 370                 // this can happen when the active file still includes finished tasks
 371                 finish_list.push(info);
 372                 return None;
 373             }
 374
 375             if !worker_is_active_local(&info.upid) {
 376                 // println!("Detected stopped task '{}'", &info.upid_str);
 377                 let now = proxmox::tools::time::epoch_i64();
 378                 let status = upid_read_status(&info.upid).unwrap_or(TaskState::Unknown { endtime: now });
 379                 finish_list.push(TaskListInfo {
 380                     upid: info.upid,
 381                     upid_str: info.upid_str,
 382                     state: Some(status)
 383                 });
 384                 return None;
 385             }
 386
 387             Some(info)
 388         }).collect();
 389
 390     if let Some(upid) = new_upid {
 391         active_list.push(TaskListInfo { upid: upid.clone(), upid_str: upid.to_string(), state: None });
 392     }
 393
 394     let active_raw = render_task_list(&active_list);
 395
 396     replace_file(
 397         PROXMOX_BACKUP_ACTIVE_TASK_FN,
 398         active_raw.as_bytes(),
 399         CreateOptions::new()
 400             .owner(backup_user.uid)
 401             .group(backup_user.gid),
 402     )?;
 403
 404     finish_list.sort_unstable_by(|a, b| {
 405         match (&a.state, &b.state) {
 406             (Some(s1), Some(s2)) => s1.cmp(&s2),
 407             (Some(_), None) => std::cmp::Ordering::Less,
 408             (None, Some(_)) => std::cmp::Ordering::Greater,
 409             _ => a.upid.starttime.cmp(&b.upid.starttime),
 410         }
 411     });
 412
 413     if !finish_list.is_empty() {
 414         match std::fs::OpenOptions::new().append(true).create(true).open(PROXMOX_BACKUP_ARCHIVE_TASK_FN) {
 415             Ok(mut writer) => {
 416                 for info in &finish_list {
 417                     writer.write_all(render_task_line(&info).as_bytes())?;
 418                 }
 419             },
 420             Err(err) => bail!("could not write task archive - {}", err),
 421         }
 422
 423         nix::unistd::chown(PROXMOX_BACKUP_ARCHIVE_TASK_FN, Some(backup_user.uid), Some(backup_user.gid))?;
 424     }
 425
 426     // TODO Remove with 1.x
 427     // for compatibility, if we had an INDEX file, we do not need it anymore
 428     if had_index_file {
 429         let _ = nix::unistd::unlink(PROXMOX_BACKUP_INDEX_TASK_FN);
 430     }
 431
 432     drop(lock);
 433
 434     Ok(())
 435 }
 436
 437 fn render_task_line(info: &TaskListInfo) -> String {
 438     let mut raw = String::new();
 439     if let Some(status) = &info.state {
 440         raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, status.endtime(), status));
 441     } else {
 442         raw.push_str(&info.upid_str);
 443         raw.push('\n');
 444     }
 445
 446     raw
 447 }
 448
 449 fn render_task_list(list: &[TaskListInfo]) -> String {
 450     let mut raw = String::new();
 451     for info in list {
 452         raw.push_str(&render_task_line(&info));
 453     }
 454     raw
 455 }
 456
 457 // note this is not locked, caller has to make sure it is
 458 // this will skip (and log) lines that are not valid status lines
 459 fn read_task_file<R: Read>(reader: R) -> Result<Vec<TaskListInfo>, Error>
 460 {
 461     let reader = BufReader::new(reader);
 462     let mut list = Vec::new();
 463     for line in reader.lines() {
 464         let line = line?;
 465         match parse_worker_status_line(&line) {
 466             Ok((upid_str, upid, state)) => list.push(TaskListInfo {
 467                 upid_str,
 468                 upid,
 469                 state
 470             }),
 471             Err(err) => {
 472                 eprintln!("unable to parse worker status '{}' - {}", line, err);
 473                 continue;
 474             }
 475         };
 476     }
 477
 478     Ok(list)
 479 }
 480
 481 // note this is not locked, caller has to make sure it is
 482 fn read_task_file_from_path<P>(path: P) -> Result<Vec<TaskListInfo>, Error>
 483 where
 484     P: AsRef<std::path::Path> + std::fmt::Debug,
 485 {
 486     let file = match File::open(&path) {
 487         Ok(f) => f,
 488         Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
 489         Err(err) => bail!("unable to open task list {:?} - {}", path, err),
 490     };
 491
 492     read_task_file(file)
 493 }
 494
 495 pub struct TaskListInfoIterator {
 496     list: VecDeque<TaskListInfo>,
 497     end: bool,
 498     archive: Option<LogRotateFiles>,
 499     lock: Option<BackupLockGuard>,
 500 }
 501
 502 impl TaskListInfoIterator {
 503     pub fn new(active_only: bool) -> Result<Self, Error> {
 504         let (read_lock, active_list) = {
 505             let lock = lock_task_list_files(false)?;
 506             let active_list = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?;
 507
 508             let needs_update = active_list
 509                 .iter()
 510                 .any(|info| info.state.is_some() || !worker_is_active_local(&info.upid));
 511
 512             // TODO remove with 1.x
 513             let index_exists = std::path::Path::new(PROXMOX_BACKUP_INDEX_TASK_FN).is_file();
 514
 515             if needs_update || index_exists {
 516                 drop(lock);
 517                 update_active_workers(None)?;
 518                 let lock = lock_task_list_files(false)?;
 519                 let active_list = read_task_file_from_path(PROXMOX_BACKUP_ACTIVE_TASK_FN)?;
 520                 (lock, active_list)
 521             } else {
 522                 (lock, active_list)
 523             }
 524         };
 525
 526         let archive = if active_only {
 527             None
 528         } else {
 529             let logrotate = LogRotate::new(PROXMOX_BACKUP_ARCHIVE_TASK_FN, true)
 530                 .ok_or_else(|| format_err!("could not get archive file names"))?;
 531             Some(logrotate.files())
 532         };
 533
 534         let lock = if active_only { None } else { Some(read_lock) };
 535
 536         Ok(Self {
 537             list: active_list.into(),
 538             end: active_only,
 539             archive,
 540             lock,
 541         })
 542     }
 543 }
 544
 545 impl Iterator for TaskListInfoIterator {
 546     type Item = Result<TaskListInfo, Error>;
 547
 548     fn next(&mut self) -> Option<Self::Item> {
 549         loop {
 550             if let Some(element) = self.list.pop_back() {
 551                 return Some(Ok(element));
 552             } else if self.end {
 553                     return None;
 554             } else {
 555                 if let Some(mut archive) = self.archive.take() {
 556                     if let Some(file) = archive.next() {
 557                         let list = match read_task_file(file) {
 558                             Ok(list) => list,
 559                             Err(err) => return Some(Err(err)),
 560                         };
 561                         self.list.append(&mut list.into());
 562                         self.archive = Some(archive);
 563                         continue;
 564                     }
 565                 }
 566
 567                 self.end = true;
 568                 self.lock.take();
 569             }
 570         }
 571     }
 572 }
 573
 574 /// Launch long running worker tasks.
 575 ///
 576 /// A worker task can either be a whole thread, or a simply tokio
 577 /// task/future. Each task can `log()` messages, which are stored
 578 /// persistently to files. Task should poll the `abort_requested`
 579 /// flag, and stop execution when requested.
 580 pub struct WorkerTask {
 581     upid: UPID,
 582     data: Mutex<WorkerTaskData>,
 583     abort_requested: AtomicBool,
 584 }
 585
 586 impl std::fmt::Display for WorkerTask {
 587
 588     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
 589         self.upid.fmt(f)
 590     }
 591 }
 592
 593 struct WorkerTaskData {
 594     logger: FileLogger,
 595     progress: f64, // 0..1
 596     warn_count: u64,
 597     pub abort_listeners: Vec<oneshot::Sender<()>>,
 598 }
 599
 600 impl WorkerTask {
 601
 602     pub fn new(worker_type: &str, worker_id: Option<String>, auth_id: Authid, to_stdout: bool) -> Result<Arc<Self>, Error> {
 603         let upid = UPID::new(worker_type, worker_id, auth_id)?;
 604         let task_id = upid.task_id;
 605
 606         let mut path = std::path::PathBuf::from(PROXMOX_BACKUP_TASK_DIR);
 607
 608         path.push(format!("{:02X}", upid.pstart & 255));
 609
 610         let backup_user = pbs_config::backup_user()?;
 611
 612         create_path(&path, None, Some(CreateOptions::new().owner(backup_user.uid).group(backup_user.gid)))?;
 613
 614         path.push(upid.to_string());
 615
 616         let logger_options = FileLogOptions {
 617             to_stdout,
 618             exclusive: true,
 619             prefix_time: true,
 620             read: true,
 621             ..Default::default()
 622         };
 623         let logger = FileLogger::new(&path, logger_options)?;
 624         nix::unistd::chown(&path, Some(backup_user.uid), Some(backup_user.gid))?;
 625
 626         let worker = Arc::new(Self {
 627             upid: upid.clone(),
 628             abort_requested: AtomicBool::new(false),
 629             data: Mutex::new(WorkerTaskData {
 630                 logger,
 631                 progress: 0.0,
 632                 warn_count: 0,
 633                 abort_listeners: vec![],
 634             }),
 635         });
 636
 637         // scope to drop the lock again after inserting
 638         {
 639             let mut hash = WORKER_TASK_LIST.lock().unwrap();
 640             hash.insert(task_id, worker.clone());
 641             proxmox_rest_server::set_worker_count(hash.len());
 642         }
 643
 644         update_active_workers(Some(&upid))?;
 645
 646         Ok(worker)
 647     }
 648
 649     /// Spawn a new tokio task/future.
 650     pub fn spawn<F, T>(
 651         worker_type: &str,
 652         worker_id: Option<String>,
 653         auth_id: Authid,
 654         to_stdout: bool,
 655         f: F,
 656     ) -> Result<String, Error>
 657         where F: Send + 'static + FnOnce(Arc<WorkerTask>) -> T,
 658               T: Send + 'static + Future<Output = Result<(), Error>>,
 659     {
 660         let worker = WorkerTask::new(worker_type, worker_id, auth_id, to_stdout)?;
 661         let upid_str = worker.upid.to_string();
 662         let f = f(worker.clone());
 663         tokio::spawn(async move {
 664             let result = f.await;
 665             worker.log_result(&result);
 666         });
 667
 668         Ok(upid_str)
 669     }
 670
 671     /// Create a new worker thread.
 672     pub fn new_thread<F>(
 673         worker_type: &str,
 674         worker_id: Option<String>,
 675         auth_id: Authid,
 676         to_stdout: bool,
 677         f: F,
 678     ) -> Result<String, Error>
 679         where F: Send + UnwindSafe + 'static + FnOnce(Arc<WorkerTask>) -> Result<(), Error>
 680     {
 681         let worker = WorkerTask::new(worker_type, worker_id, auth_id, to_stdout)?;
 682         let upid_str = worker.upid.to_string();
 683
 684         let _child = std::thread::Builder::new().name(upid_str.clone()).spawn(move || {
 685             let worker1 = worker.clone();
 686             let result = match std::panic::catch_unwind(move || f(worker1)) {
 687                 Ok(r) => r,
 688                 Err(panic) => {
 689                     match panic.downcast::<&str>() {
 690                         Ok(panic_msg) => {
 691                             Err(format_err!("worker panicked: {}", panic_msg))
 692                         }
 693                         Err(_) => {
 694                             Err(format_err!("worker panicked: unknown type."))
 695                         }
 696                     }
 697                 }
 698             };
 699
 700             worker.log_result(&result);
 701         });
 702
 703         Ok(upid_str)
 704     }
 705
 706     /// create state from self and a result
 707     pub fn create_state(&self, result: &Result<(), Error>) -> TaskState {
 708         let warn_count = self.data.lock().unwrap().warn_count;
 709
 710         let endtime = proxmox::tools::time::epoch_i64();
 711
 712         if let Err(err) = result {
 713             TaskState::Error { message: err.to_string(), endtime }
 714         } else if warn_count > 0 {
 715             TaskState::Warning { count: warn_count, endtime }
 716         } else {
 717             TaskState::OK { endtime }
 718         }
 719     }
 720
 721     /// Log task result, remove task from running list
 722     pub fn log_result(&self, result: &Result<(), Error>) {
 723         let state = self.create_state(result);
 724         self.log(state.result_text());
 725
 726         WORKER_TASK_LIST.lock().unwrap().remove(&self.upid.task_id);
 727         let _ = update_active_workers(None);
 728         proxmox_rest_server::set_worker_count(WORKER_TASK_LIST.lock().unwrap().len());
 729     }
 730
 731     /// Log a message.
 732     pub fn log<S: AsRef<str>>(&self, msg: S) {
 733         let mut data = self.data.lock().unwrap();
 734         data.logger.log(msg);
 735     }
 736
 737     /// Log a message as warning.
 738     pub fn warn<S: AsRef<str>>(&self, msg: S) {
 739         let mut data = self.data.lock().unwrap();
 740         data.logger.log(format!("WARN: {}", msg.as_ref()));
 741         data.warn_count += 1;
 742     }
 743
 744     /// Set progress indicator
 745     pub fn progress(&self, progress: f64) {
 746         if progress >= 0.0 && progress <= 1.0 {
 747             let mut data = self.data.lock().unwrap();
 748             data.progress = progress;
 749         } else {
 750            // fixme:  log!("task '{}': ignoring strange value for progress '{}'", self.upid, progress);
 751         }
 752     }
 753
 754     /// Request abort
 755     pub fn request_abort(&self) {
 756         eprintln!("set abort flag for worker {}", self.upid);
 757
 758         let prev_abort = self.abort_requested.swap(true, Ordering::SeqCst);
 759         if !prev_abort { // log abort one time
 760             self.log(format!("received abort request ..."));
 761         }
 762         // noitify listeners
 763         let mut data = self.data.lock().unwrap();
 764         loop {
 765             match data.abort_listeners.pop() {
 766                 None => { break; },
 767                 Some(ch) => {
 768                     let _ = ch.send(()); // ignore errors here
 769                 },
 770             }
 771         }
 772     }
 773
 774     /// Test if abort was requested.
 775     pub fn abort_requested(&self) -> bool {
 776         self.abort_requested.load(Ordering::SeqCst)
 777     }
 778
 779     /// Fail if abort was requested.
 780     pub fn fail_on_abort(&self) -> Result<(), Error> {
 781         if self.abort_requested() {
 782             bail!("abort requested - aborting task");
 783         }
 784         Ok(())
 785     }
 786
 787     /// Get a future which resolves on task abort
 788     pub fn abort_future(&self) ->  oneshot::Receiver<()> {
 789         let (tx, rx) = oneshot::channel::<()>();
 790
 791         let mut data = self.data.lock().unwrap();
 792         if self.abort_requested() {
 793             let _ = tx.send(());
 794         } else {
 795             data.abort_listeners.push(tx);
 796         }
 797         rx
 798     }
 799
 800     pub fn upid(&self) -> &UPID {
 801         &self.upid
 802     }
 803 }
 804
 805 impl pbs_datastore::task::TaskState for WorkerTask {
 806     fn check_abort(&self) -> Result<(), Error> {
 807         self.fail_on_abort()
 808     }
 809
 810     fn log(&self, level: log::Level, message: &std::fmt::Arguments) {
 811         match level {
 812             log::Level::Error => self.warn(&message.to_string()),
 813             log::Level::Warn => self.warn(&message.to_string()),
 814             log::Level::Info => self.log(&message.to_string()),
 815             log::Level::Debug => self.log(&format!("DEBUG: {}", message)),
 816             log::Level::Trace => self.log(&format!("TRACE: {}", message)),
 817         }
 818     }
 819 }
 820
 821 /// Wait for a locally spanned worker task
 822 ///
 823 /// Note: local workers should print logs to stdout, so there is no
 824 /// need to fetch/display logs. We just wait for the worker to finish.
 825 pub async fn wait_for_local_worker(upid_str: &str) -> Result<(), Error> {
 826
 827     let upid: UPID = upid_str.parse()?;
 828
 829     let sleep_duration = core::time::Duration::new(0, 100_000_000);
 830
 831     loop {
 832         if worker_is_active_local(&upid) {
 833             tokio::time::sleep(sleep_duration).await;
 834         } else {
 835             break;
 836         }
 837     }
 838     Ok(())
 839 }