src/server/worker_task.rs

   1 use failure::*;
   2 use lazy_static::lazy_static;
   3 use chrono::Local;
   4
   5 use tokio::sync::oneshot;
   6 use futures::*;
   7 use std::sync::{Arc, Mutex};
   8 use std::collections::HashMap;
   9 use std::sync::atomic::{AtomicBool, Ordering};
  10 use std::io::{BufRead, BufReader};
  11 use std::fs::File;
  12 use std::panic::UnwindSafe;
  13
  14 use serde_json::{json, Value};
  15
  16 use proxmox::tools::{
  17     try_block,
  18     fs::{create_path, replace_file, CreateOptions},
  19 };
  20
  21 use super::UPID;
  22
  23 use crate::tools::FileLogger;
  24
  25 macro_rules! PROXMOX_BACKUP_VAR_RUN_DIR_M { () => ("/run/proxmox-backup") }
  26 macro_rules! PROXMOX_BACKUP_LOG_DIR_M { () => ("/var/log/proxmox-backup") }
  27 macro_rules! PROXMOX_BACKUP_TASK_DIR_M { () => (concat!( PROXMOX_BACKUP_LOG_DIR_M!(), "/tasks")) }
  28
  29 pub const PROXMOX_BACKUP_VAR_RUN_DIR: &str = PROXMOX_BACKUP_VAR_RUN_DIR_M!();
  30 pub const PROXMOX_BACKUP_LOG_DIR: &str = PROXMOX_BACKUP_LOG_DIR_M!();
  31 pub const PROXMOX_BACKUP_TASK_DIR: &str = PROXMOX_BACKUP_TASK_DIR_M!();
  32 pub const PROXMOX_BACKUP_TASK_LOCK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/.active.lock");
  33 pub const PROXMOX_BACKUP_ACTIVE_TASK_FN: &str = concat!(PROXMOX_BACKUP_TASK_DIR_M!(), "/active");
  34
  35 lazy_static! {
  36     static ref WORKER_TASK_LIST: Mutex<HashMap<usize, Arc<WorkerTask>>> = Mutex::new(HashMap::new());
  37
  38     static ref MY_PID: i32 = unsafe { libc::getpid() };
  39     static ref MY_PID_PSTART: u64 = proxmox::sys::linux::procfs::read_proc_pid_stat(*MY_PID).unwrap().starttime;
  40 }
  41
  42 /// Test if the task is still running
  43 pub fn worker_is_active(upid: &UPID) -> bool {
  44
  45     if (upid.pid == *MY_PID) && (upid.pstart == *MY_PID_PSTART) {
  46         WORKER_TASK_LIST.lock().unwrap().contains_key(&upid.task_id)
  47     } else {
  48         use proxmox::sys::linux::procfs;
  49         procfs::check_process_running_pstart(upid.pid, upid.pstart).is_some()
  50     }
  51 }
  52
  53 pub fn create_task_control_socket() -> Result<(), Error> {
  54
  55     let socketname = format!(
  56         "\0{}/proxmox-task-control-{}.sock", PROXMOX_BACKUP_VAR_RUN_DIR, *MY_PID);
  57
  58     let control_future = super::create_control_socket(socketname, |param| {
  59         let param = param.as_object()
  60             .ok_or_else(|| format_err!("unable to parse parameters (expected json object)"))?;
  61         if param.keys().count() != 2 { bail!("wrong number of parameters"); }
  62
  63         let command = param.get("command")
  64             .ok_or_else(|| format_err!("unable to parse parameters (missing command)"))?;
  65
  66         // this is the only command for now
  67         if command != "abort-task" { bail!("got unknown command '{}'", command); }
  68
  69         let upid_str = param["upid"].as_str()
  70             .ok_or_else(|| format_err!("unable to parse parameters (missing upid)"))?;
  71
  72         let upid = upid_str.parse::<UPID>()?;
  73
  74         if !((upid.pid == *MY_PID) && (upid.pstart == *MY_PID_PSTART)) {
  75             bail!("upid does not belong to this process");
  76         }
  77
  78         let hash = WORKER_TASK_LIST.lock().unwrap();
  79         if let Some(ref worker) = hash.get(&upid.task_id) {
  80             worker.request_abort();
  81         } else {
  82             // assume task is already stopped
  83         }
  84         Ok(Value::Null)
  85     })?;
  86
  87     tokio::spawn(control_future);
  88
  89     Ok(())
  90 }
  91
  92 pub fn abort_worker_async(upid: UPID) {
  93     tokio::spawn(async move {
  94         if let Err(err) = abort_worker(upid).await {
  95             eprintln!("abort worker failed - {}", err);
  96         }
  97     });
  98 }
  99
 100 pub fn abort_worker(upid: UPID) -> impl Future<Output = Result<(), Error>> {
 101
 102     let target_pid = upid.pid;
 103
 104     let socketname = format!(
 105         "\0{}/proxmox-task-control-{}.sock", PROXMOX_BACKUP_VAR_RUN_DIR, target_pid);
 106
 107     let cmd = json!({
 108         "command": "abort-task",
 109         "upid": upid.to_string(),
 110     });
 111
 112     super::send_command(socketname, cmd).map_ok(|_| ())
 113 }
 114
 115 fn parse_worker_status_line(line: &str) -> Result<(String, UPID, Option<(i64, String)>), Error> {
 116
 117     let data = line.splitn(3, ' ').collect::<Vec<&str>>();
 118
 119     let len = data.len();
 120
 121     match len {
 122         1 => Ok((data[0].to_owned(), data[0].parse::<UPID>()?, None)),
 123         3 => {
 124             let endtime = i64::from_str_radix(data[1], 16)?;
 125             Ok((data[0].to_owned(), data[0].parse::<UPID>()?, Some((endtime, data[2].to_owned()))))
 126         }
 127         _ => bail!("wrong number of components"),
 128     }
 129 }
 130
 131 /// Create task log directory with correct permissions
 132 pub fn create_task_log_dirs() -> Result<(), Error> {
 133
 134     try_block!({
 135         let (backup_uid, backup_gid) = crate::tools::getpwnam_ugid("backup")?;
 136         let opts = CreateOptions::new()
 137             .owner(nix::unistd::Uid::from_raw(backup_uid))
 138             .group(nix::unistd::Gid::from_raw(backup_gid));
 139
 140         create_path(PROXMOX_BACKUP_LOG_DIR, None, Some(opts.clone()))?;
 141         create_path(PROXMOX_BACKUP_TASK_DIR, None, Some(opts.clone()))?;
 142         create_path(PROXMOX_BACKUP_VAR_RUN_DIR, None, Some(opts))?;
 143         Ok(())
 144     }).map_err(|err: Error| format_err!("unable to create task log dir - {}", err))?;
 145
 146     Ok(())
 147 }
 148
 149 /// Read exits status from task log file
 150 pub fn upid_read_status(upid: &UPID) -> Result<String, Error> {
 151     let mut status = String::from("unknown");
 152
 153     let path = upid.log_path();
 154
 155     let mut file = File::open(path)?;
 156
 157     /// speedup - only read tail
 158     use std::io::Seek;
 159     use std::io::SeekFrom;
 160     let _ = file.seek(SeekFrom::End(-8192)); // ignore errors
 161
 162     let reader = BufReader::new(file);
 163
 164     for line in reader.lines() {
 165         let line = line?;
 166
 167         let mut iter = line.splitn(2, ": TASK ");
 168         if iter.next() == None { continue; }
 169         match iter.next() {
 170             None => continue,
 171             Some(rest) => {
 172                 if rest == "OK" {
 173                     status = String::from(rest);
 174                 } else if rest.starts_with("ERROR: ") {
 175                     status = String::from(&rest[7..]);
 176                 }
 177             }
 178         }
 179     }
 180
 181     Ok(status)
 182 }
 183
 184 /// Task details including parsed UPID
 185 ///
 186 /// If there is no `state`, the task is still running.
 187 #[derive(Debug)]
 188 pub struct TaskListInfo {
 189     /// The parsed UPID
 190     pub upid: UPID,
 191     /// UPID string representation
 192     pub upid_str: String,
 193     /// Task `(endtime, status)` if already finished
 194     ///
 195     /// The `status` ise iether `unknown`, `OK`, or `ERROR: ...`
 196     pub state: Option<(i64, String)>, // endtime, status
 197 }
 198
 199 // atomically read/update the task list, update status of finished tasks
 200 // new_upid is added to the list when specified.
 201 // Returns a sorted list of known tasks,
 202 fn update_active_workers(new_upid: Option<&UPID>) -> Result<Vec<TaskListInfo>, Error> {
 203
 204     let (backup_uid, backup_gid) = crate::tools::getpwnam_ugid("backup")?;
 205     let uid = nix::unistd::Uid::from_raw(backup_uid);
 206     let gid = nix::unistd::Gid::from_raw(backup_gid);
 207
 208     let lock = crate::tools::open_file_locked(PROXMOX_BACKUP_TASK_LOCK_FN, std::time::Duration::new(10, 0))?;
 209     nix::unistd::chown(PROXMOX_BACKUP_TASK_LOCK_FN, Some(uid), Some(gid))?;
 210
 211     let reader = match File::open(PROXMOX_BACKUP_ACTIVE_TASK_FN) {
 212         Ok(f) => Some(BufReader::new(f)),
 213         Err(err) => {
 214             if err.kind() ==  std::io::ErrorKind::NotFound {
 215                  None
 216             } else {
 217                 bail!("unable to open active worker {:?} - {}", PROXMOX_BACKUP_ACTIVE_TASK_FN, err);
 218             }
 219         }
 220     };
 221
 222     let mut active_list = vec![];
 223     let mut finish_list = vec![];
 224
 225     if let Some(lines) = reader.map(|r| r.lines()) {
 226
 227         for line in lines {
 228             let line = line?;
 229             match parse_worker_status_line(&line) {
 230                 Err(err) => bail!("unable to parse active worker status '{}' - {}", line, err),
 231                 Ok((upid_str, upid, state)) => {
 232
 233                     let running = worker_is_active(&upid);
 234
 235                     if running {
 236                         active_list.push(TaskListInfo { upid, upid_str, state: None });
 237                     } else {
 238                         match state {
 239                             None => {
 240                                 println!("Detected stoped UPID {}", upid_str);
 241                                 let status = upid_read_status(&upid)
 242                                     .unwrap_or_else(|_| String::from("unknown"));
 243                                 finish_list.push(TaskListInfo {
 244                                     upid, upid_str, state: Some((Local::now().timestamp(), status))
 245                                 });
 246                             }
 247                             Some((endtime, status)) => {
 248                                 finish_list.push(TaskListInfo {
 249                                     upid, upid_str, state: Some((endtime, status))
 250                                 })
 251                             }
 252                         }
 253                     }
 254                 }
 255             }
 256         }
 257     }
 258
 259     if let Some(upid) = new_upid {
 260         active_list.push(TaskListInfo { upid: upid.clone(), upid_str: upid.to_string(), state: None });
 261     }
 262
 263     // assemble list without duplicates
 264     // we include all active tasks,
 265     // and fill up to 1000 entries with finished tasks
 266
 267     let max = 1000;
 268
 269     let mut task_hash = HashMap::new();
 270
 271     for info in active_list {
 272         task_hash.insert(info.upid_str.clone(), info);
 273     }
 274
 275     for info in finish_list {
 276         if task_hash.len() > max { break; }
 277         if !task_hash.contains_key(&info.upid_str) {
 278             task_hash.insert(info.upid_str.clone(), info);
 279         }
 280     }
 281
 282     let mut task_list: Vec<TaskListInfo> = vec![];
 283     for (_, info) in task_hash { task_list.push(info); }
 284
 285     task_list.sort_unstable_by(|b, a| { // lastest on top
 286         match (&a.state, &b.state) {
 287             (Some(s1), Some(s2)) => s1.0.cmp(&s2.0),
 288             (Some(_), None) => std::cmp::Ordering::Less,
 289             (None, Some(_)) => std::cmp::Ordering::Greater,
 290             _ => a.upid.starttime.cmp(&b.upid.starttime),
 291         }
 292     });
 293
 294     let mut raw = String::new();
 295     for info in &task_list {
 296         if let Some((endtime, status)) = &info.state {
 297             raw.push_str(&format!("{} {:08X} {}\n", info.upid_str, endtime, status));
 298         } else {
 299             raw.push_str(&info.upid_str);
 300             raw.push('\n');
 301         }
 302     }
 303
 304     replace_file(
 305         PROXMOX_BACKUP_ACTIVE_TASK_FN,
 306         raw.as_bytes(),
 307         CreateOptions::new()
 308             .owner(uid)
 309             .group(gid),
 310     )?;
 311
 312     drop(lock);
 313
 314     Ok(task_list)
 315 }
 316
 317 /// Returns a sorted list of known tasks
 318 ///
 319 /// The list is sorted by `(starttime, endtime)` in ascending order
 320 pub fn read_task_list() -> Result<Vec<TaskListInfo>, Error> {
 321     update_active_workers(None)
 322 }
 323
 324 /// Launch long running worker tasks.
 325 ///
 326 /// A worker task can either be a whole thread, or a simply tokio
 327 /// task/future. Each task can `log()` messages, which are stored
 328 /// persistently to files. Task should poll the `abort_requested`
 329 /// flag, and stop execution when requested.
 330 #[derive(Debug)]
 331 pub struct WorkerTask {
 332     upid: UPID,
 333     data: Mutex<WorkerTaskData>,
 334     abort_requested: AtomicBool,
 335 }
 336
 337 impl std::fmt::Display for WorkerTask {
 338
 339     fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
 340         self.upid.fmt(f)
 341     }
 342 }
 343
 344 #[derive(Debug)]
 345 struct WorkerTaskData {
 346     logger: FileLogger,
 347     progress: f64, // 0..1
 348     pub abort_listeners: Vec<oneshot::Sender<()>>,
 349 }
 350
 351 impl Drop for WorkerTask {
 352
 353     fn drop(&mut self) {
 354         println!("unregister worker");
 355     }
 356 }
 357
 358 impl WorkerTask {
 359
 360     pub fn new(worker_type: &str, worker_id: Option<String>, username: &str, to_stdout: bool) -> Result<Arc<Self>, Error> {
 361         println!("register worker");
 362
 363         let upid = UPID::new(worker_type, worker_id, username)?;
 364         let task_id = upid.task_id;
 365
 366         let mut path = std::path::PathBuf::from(PROXMOX_BACKUP_TASK_DIR);
 367
 368         path.push(format!("{:02X}", upid.pstart % 256));
 369
 370         let (backup_uid, backup_gid) = crate::tools::getpwnam_ugid("backup")?;
 371         let uid = nix::unistd::Uid::from_raw(backup_uid);
 372         let gid = nix::unistd::Gid::from_raw(backup_gid);
 373
 374         create_path(&path, None, Some(CreateOptions::new().owner(uid).group(gid)))?;
 375
 376         path.push(upid.to_string());
 377
 378         println!("FILE: {:?}", path);
 379
 380         let logger = FileLogger::new(&path, to_stdout)?;
 381         nix::unistd::chown(&path, Some(uid), Some(gid))?;
 382
 383         update_active_workers(Some(&upid))?;
 384
 385         let worker = Arc::new(Self {
 386             upid,
 387             abort_requested: AtomicBool::new(false),
 388             data: Mutex::new(WorkerTaskData {
 389                 logger,
 390                 progress: 0.0,
 391                 abort_listeners: vec![],
 392             }),
 393         });
 394
 395         let mut hash = WORKER_TASK_LIST.lock().unwrap();
 396
 397         hash.insert(task_id, worker.clone());
 398         super::set_worker_count(hash.len());
 399
 400         Ok(worker)
 401     }
 402
 403     /// Spawn a new tokio task/future.
 404     pub fn spawn<F, T>(
 405         worker_type: &str,
 406         worker_id: Option<String>,
 407         username: &str,
 408         to_stdout: bool,
 409         f: F,
 410     ) -> Result<String, Error>
 411         where F: Send + 'static + FnOnce(Arc<WorkerTask>) -> T,
 412               T: Send + 'static + Future<Output = Result<(), Error>>,
 413     {
 414         let worker = WorkerTask::new(worker_type, worker_id, username, to_stdout)?;
 415         let upid_str = worker.upid.to_string();
 416         let f = f(worker.clone());
 417         tokio::spawn(async move {
 418             let result = f.await;
 419             worker.log_result(&result);
 420         });
 421
 422         Ok(upid_str)
 423     }
 424
 425     /// Create a new worker thread.
 426     pub fn new_thread<F>(
 427         worker_type: &str,
 428         worker_id: Option<String>,
 429         username: &str,
 430         to_stdout: bool,
 431         f: F,
 432     ) -> Result<String, Error>
 433         where F: Send + UnwindSafe + 'static + FnOnce(Arc<WorkerTask>) -> Result<(), Error>
 434     {
 435         println!("register worker thread");
 436
 437         let (p, c) = oneshot::channel::<()>();
 438
 439         let worker = WorkerTask::new(worker_type, worker_id, username, to_stdout)?;
 440         let upid_str = worker.upid.to_string();
 441
 442         let _child = std::thread::spawn(move || {
 443             let worker1 = worker.clone();
 444             let result = match std::panic::catch_unwind(move || f(worker1)) {
 445                 Ok(r) => r,
 446                 Err(panic) => {
 447                     match panic.downcast::<&str>() {
 448                         Ok(panic_msg) => {
 449                             Err(format_err!("worker panicked: {}", panic_msg))
 450                         }
 451                         Err(_) => {
 452                             Err(format_err!("worker panicked: unknown type."))
 453                         }
 454                     }
 455                 }
 456             };
 457
 458             worker.log_result(&result);
 459             p.send(()).unwrap();
 460         });
 461
 462         tokio::spawn(c.map(|_| ()));
 463
 464         Ok(upid_str)
 465     }
 466
 467     /// Log task result, remove task from running list
 468     pub fn log_result(&self, result: &Result<(), Error>) {
 469
 470         if let Err(err) = result {
 471             self.log(&format!("TASK ERROR: {}", err));
 472         } else {
 473             self.log("TASK OK");
 474         }
 475
 476         WORKER_TASK_LIST.lock().unwrap().remove(&self.upid.task_id);
 477         let _ = update_active_workers(None);
 478         super::set_worker_count(WORKER_TASK_LIST.lock().unwrap().len());
 479     }
 480
 481     /// Log a message.
 482     pub fn log<S: AsRef<str>>(&self, msg: S) {
 483         let mut data = self.data.lock().unwrap();
 484         data.logger.log(msg);
 485     }
 486
 487     /// Set progress indicator
 488     pub fn progress(&self, progress: f64) {
 489         if progress >= 0.0 && progress <= 1.0 {
 490             let mut data = self.data.lock().unwrap();
 491             data.progress = progress;
 492         } else {
 493            // fixme:  log!("task '{}': ignoring strange value for progress '{}'", self.upid, progress);
 494         }
 495     }
 496
 497     /// Request abort
 498     pub fn request_abort(&self) {
 499         eprintln!("set abort flag for worker {}", self.upid);
 500         self.abort_requested.store(true, Ordering::SeqCst);
 501         // noitify listeners
 502         let mut data = self.data.lock().unwrap();
 503         loop {
 504             match data.abort_listeners.pop() {
 505                 None => { break; },
 506                 Some(ch) => {
 507                     let _ = ch.send(()); // ignore erros here
 508                 },
 509             }
 510         }
 511     }
 512
 513     /// Test if abort was requested.
 514     pub fn abort_requested(&self) -> bool {
 515         self.abort_requested.load(Ordering::SeqCst)
 516     }
 517
 518     /// Fail if abort was requested.
 519     pub fn fail_on_abort(&self) -> Result<(), Error> {
 520         if self.abort_requested() {
 521             bail!("task '{}': abort requested - aborting task", self.upid);
 522         }
 523         Ok(())
 524     }
 525
 526     /// Get a future which resolves on task abort
 527     pub fn abort_future(&self) ->  oneshot::Receiver<()> {
 528         let (tx, rx) = oneshot::channel::<()>();
 529
 530         let mut data = self.data.lock().unwrap();
 531         if self.abort_requested() {
 532             let _ = tx.send(());
 533         } else {
 534             data.abort_listeners.push(tx);
 535         }
 536         rx
 537     }
 538 }