src/api2/tape/backup.rs

   1 use std::path::Path;
   2 use std::sync::{Mutex, Arc};
   3
   4 use anyhow::{bail, format_err, Error};
   5 use serde_json::Value;
   6
   7 use proxmox::{
   8     try_block,
   9     api::{
  10         api,
  11         RpcEnvironment,
  12         RpcEnvironmentType,
  13         Router,
  14         Permission,
  15     },
  16 };
  17
  18 use pbs_api_types::{
  19     Authid, Userid, TapeBackupJobConfig, TapeBackupJobSetup, TapeBackupJobStatus, MediaPoolConfig,
  20     UPID_SCHEMA, JOB_ID_SCHEMA, PRIV_DATASTORE_READ, PRIV_TAPE_AUDIT, PRIV_TAPE_WRITE,
  21 };
  22
  23 use pbs_datastore::{task_log, task_warn, StoreProgress};
  24 use pbs_datastore::backup_info::{BackupDir, BackupInfo};
  25 use pbs_datastore::task::TaskState;
  26 use pbs_config::CachedUserInfo;
  27
  28 use crate::{
  29     server::{
  30         lookup_user_email,
  31         TapeBackupJobSummary,
  32         jobstate::{
  33             Job,
  34             JobState,
  35             compute_schedule_status,
  36         },
  37     },
  38     backup::DataStore,
  39     server::WorkerTask,
  40     tape::{
  41         TAPE_STATUS_DIR,
  42         Inventory,
  43         PoolWriter,
  44         MediaPool,
  45         SnapshotReader,
  46         drive::{
  47             media_changer,
  48             lock_tape_device,
  49             TapeLockError,
  50             set_tape_device_state,
  51         },
  52         changer::update_changer_online_status,
  53     },
  54 };
  55
  56 const TAPE_BACKUP_JOB_ROUTER: Router = Router::new()
  57     .post(&API_METHOD_RUN_TAPE_BACKUP_JOB);
  58
  59 pub const ROUTER: Router = Router::new()
  60     .get(&API_METHOD_LIST_TAPE_BACKUP_JOBS)
  61     .post(&API_METHOD_BACKUP)
  62     .match_all("id", &TAPE_BACKUP_JOB_ROUTER);
  63
  64 fn check_backup_permission(
  65     auth_id: &Authid,
  66     store: &str,
  67     pool: &str,
  68     drive: &str,
  69 ) -> Result<(), Error> {
  70
  71     let user_info = CachedUserInfo::new()?;
  72
  73     let privs = user_info.lookup_privs(auth_id, &["datastore", store]);
  74     if (privs & PRIV_DATASTORE_READ) == 0 {
  75         bail!("no permissions on /datastore/{}", store);
  76     }
  77
  78     let privs = user_info.lookup_privs(auth_id, &["tape", "drive", drive]);
  79     if (privs & PRIV_TAPE_WRITE) == 0 {
  80         bail!("no permissions on /tape/drive/{}", drive);
  81     }
  82
  83     let privs = user_info.lookup_privs(auth_id, &["tape", "pool", pool]);
  84     if (privs & PRIV_TAPE_WRITE) == 0 {
  85         bail!("no permissions on /tape/pool/{}", pool);
  86     }
  87
  88     Ok(())
  89 }
  90
  91 #[api(
  92     returns: {
  93         description: "List configured thape backup jobs and their status",
  94         type: Array,
  95         items: { type: TapeBackupJobStatus },
  96     },
  97     access: {
  98         description: "List configured tape jobs filtered by Tape.Audit privileges",
  99         permission: &Permission::Anybody,
 100     },
 101 )]
 102 /// List all tape backup jobs
 103 pub fn list_tape_backup_jobs(
 104     _param: Value,
 105     mut rpcenv: &mut dyn RpcEnvironment,
 106 ) -> Result<Vec<TapeBackupJobStatus>, Error> {
 107     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
 108     let user_info = CachedUserInfo::new()?;
 109
 110     let (job_config, digest) = pbs_config::tape_job::config()?;
 111     let (pool_config, _pool_digest) = pbs_config::media_pool::config()?;
 112     let (drive_config, _digest) = pbs_config::drive::config()?;
 113
 114     let job_list_iter = job_config
 115         .convert_to_typed_array("backup")?
 116         .into_iter()
 117         .filter(|_job: &TapeBackupJobConfig| {
 118             // fixme: check access permission
 119             true
 120         });
 121
 122     let mut list = Vec::new();
 123     let status_path = Path::new(TAPE_STATUS_DIR);
 124     let current_time = proxmox::tools::time::epoch_i64();
 125
 126     for job in job_list_iter {
 127         let privs = user_info.lookup_privs(&auth_id, &["tape", "job", &job.id]);
 128         if (privs & PRIV_TAPE_AUDIT) == 0 {
 129             continue;
 130         }
 131
 132         let last_state = JobState::load("tape-backup-job", &job.id)
 133             .map_err(|err| format_err!("could not open statefile for {}: {}", &job.id, err))?;
 134
 135         let status = compute_schedule_status(&last_state, job.schedule.as_deref())?;
 136
 137         let next_run = status.next_run.unwrap_or(current_time);
 138
 139         let mut next_media_label = None;
 140
 141         if let Ok(pool) = pool_config.lookup::<MediaPoolConfig>("pool", &job.setup.pool) {
 142             let mut changer_name = None;
 143             if let Ok(Some((_, name))) = media_changer(&drive_config, &job.setup.drive) {
 144                 changer_name = Some(name);
 145             }
 146             if let Ok(mut pool) = MediaPool::with_config(status_path, &pool, changer_name, true) {
 147                 if pool.start_write_session(next_run, false).is_ok() {
 148                     if let Ok(media_id) = pool.guess_next_writable_media(next_run) {
 149                         next_media_label = Some(media_id.label.label_text);
 150                     }
 151                 }
 152             }
 153         }
 154
 155         list.push(TapeBackupJobStatus { config: job, status, next_media_label });
 156     }
 157
 158     rpcenv["digest"] = proxmox::tools::digest_to_hex(&digest).into();
 159
 160     Ok(list)
 161 }
 162
 163 pub fn do_tape_backup_job(
 164     mut job: Job,
 165     setup: TapeBackupJobSetup,
 166     auth_id: &Authid,
 167     schedule: Option<String>,
 168 ) -> Result<String, Error> {
 169
 170     let job_id = format!("{}:{}:{}:{}",
 171                          setup.store,
 172                          setup.pool,
 173                          setup.drive,
 174                          job.jobname());
 175
 176     let worker_type = job.jobtype().to_string();
 177
 178     let datastore = DataStore::lookup_datastore(&setup.store)?;
 179
 180     let (config, _digest) = pbs_config::media_pool::config()?;
 181     let pool_config: MediaPoolConfig = config.lookup("pool", &setup.pool)?;
 182
 183     let (drive_config, _digest) = pbs_config::drive::config()?;
 184
 185     // for scheduled jobs we acquire the lock later in the worker
 186     let drive_lock = if schedule.is_some() {
 187         None
 188     } else {
 189         Some(lock_tape_device(&drive_config, &setup.drive)?)
 190     };
 191
 192     let notify_user = setup.notify_user.as_ref().unwrap_or_else(|| &Userid::root_userid());
 193     let email = lookup_user_email(notify_user);
 194
 195     let upid_str = WorkerTask::new_thread(
 196         &worker_type,
 197         Some(job_id.clone()),
 198         auth_id.clone(),
 199         false,
 200         move |worker| {
 201             job.start(&worker.upid().to_string())?;
 202             let mut drive_lock = drive_lock;
 203
 204             let mut summary = Default::default();
 205             let job_result = try_block!({
 206                 if schedule.is_some() {
 207                     // for scheduled tape backup jobs, we wait indefinitely for the lock
 208                     task_log!(worker, "waiting for drive lock...");
 209                     loop {
 210                         worker.check_abort()?;
 211                         match lock_tape_device(&drive_config, &setup.drive) {
 212                             Ok(lock) => {
 213                                 drive_lock = Some(lock);
 214                                 break;
 215                             }
 216                             Err(TapeLockError::TimeOut) => continue,
 217                             Err(TapeLockError::Other(err)) => return Err(err),
 218                         }
 219                     }
 220                 }
 221                 set_tape_device_state(&setup.drive, &worker.upid().to_string())?;
 222
 223                 task_log!(worker,"Starting tape backup job '{}'", job_id);
 224                 if let Some(event_str) = schedule {
 225                     task_log!(worker,"task triggered by schedule '{}'", event_str);
 226                 }
 227
 228
 229                 backup_worker(
 230                     &worker,
 231                     datastore,
 232                     &pool_config,
 233                     &setup,
 234                     email.clone(),
 235                     &mut summary,
 236                     false,
 237                 )
 238             });
 239
 240             let status = worker.create_state(&job_result);
 241
 242             if let Some(email) = email {
 243                 if let Err(err) = crate::server::send_tape_backup_status(
 244                     &email,
 245                     Some(job.jobname()),
 246                     &setup,
 247                     &job_result,
 248                     summary,
 249                 ) {
 250                     eprintln!("send tape backup notification failed: {}", err);
 251                 }
 252             }
 253
 254             if let Err(err) = job.finish(status) {
 255                 eprintln!(
 256                     "could not finish job state for {}: {}",
 257                     job.jobtype().to_string(),
 258                     err
 259                 );
 260             }
 261
 262             if let Err(err) = set_tape_device_state(&setup.drive, "") {
 263                 eprintln!(
 264                     "could not unset drive state for {}: {}",
 265                     setup.drive,
 266                     err
 267                 );
 268             }
 269
 270             job_result
 271         }
 272     )?;
 273
 274     Ok(upid_str)
 275 }
 276
 277 #[api(
 278     input: {
 279         properties: {
 280             id: {
 281                 schema: JOB_ID_SCHEMA,
 282             },
 283         },
 284     },
 285     access: {
 286         // Note: parameters are from job config, so we need to test inside function body
 287         description: "The user needs Tape.Write privilege on /tape/pool/{pool} \
 288                       and /tape/drive/{drive}, Datastore.Read privilege on /datastore/{store}.",
 289         permission: &Permission::Anybody,
 290     },
 291 )]
 292 /// Runs a tape backup job manually.
 293 pub fn run_tape_backup_job(
 294     id: String,
 295     rpcenv: &mut dyn RpcEnvironment,
 296 ) -> Result<String, Error> {
 297     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
 298
 299     let (config, _digest) = pbs_config::tape_job::config()?;
 300     let backup_job: TapeBackupJobConfig = config.lookup("backup", &id)?;
 301
 302     check_backup_permission(
 303         &auth_id,
 304         &backup_job.setup.store,
 305         &backup_job.setup.pool,
 306         &backup_job.setup.drive,
 307     )?;
 308
 309     let job = Job::new("tape-backup-job", &id)?;
 310
 311     let upid_str = do_tape_backup_job(job, backup_job.setup, &auth_id, None)?;
 312
 313     Ok(upid_str)
 314 }
 315
 316 #[api(
 317     input: {
 318         properties: {
 319             setup: {
 320                 type: TapeBackupJobSetup,
 321                 flatten: true,
 322             },
 323             "force-media-set": {
 324                 description: "Ignore the allocation policy and start a new media-set.",
 325                 optional: true,
 326                 type: bool,
 327                 default: false,
 328             },
 329         },
 330     },
 331     returns: {
 332         schema: UPID_SCHEMA,
 333     },
 334     access: {
 335         // Note: parameters are no uri parameter, so we need to test inside function body
 336         description: "The user needs Tape.Write privilege on /tape/pool/{pool} \
 337                       and /tape/drive/{drive}, Datastore.Read privilege on /datastore/{store}.",
 338         permission: &Permission::Anybody,
 339     },
 340 )]
 341 /// Backup datastore to tape media pool
 342 pub fn backup(
 343     setup: TapeBackupJobSetup,
 344     force_media_set: bool,
 345     rpcenv: &mut dyn RpcEnvironment,
 346 ) -> Result<Value, Error> {
 347
 348     let auth_id: Authid = rpcenv.get_auth_id().unwrap().parse()?;
 349
 350     check_backup_permission(
 351         &auth_id,
 352         &setup.store,
 353         &setup.pool,
 354         &setup.drive,
 355     )?;
 356
 357     let datastore = DataStore::lookup_datastore(&setup.store)?;
 358
 359     let (config, _digest) = pbs_config::media_pool::config()?;
 360     let pool_config: MediaPoolConfig = config.lookup("pool", &setup.pool)?;
 361
 362     let (drive_config, _digest) = pbs_config::drive::config()?;
 363
 364     // early check/lock before starting worker
 365     let drive_lock = lock_tape_device(&drive_config, &setup.drive)?;
 366
 367     let to_stdout = rpcenv.env_type() == RpcEnvironmentType::CLI;
 368
 369     let job_id = format!("{}:{}:{}", setup.store, setup.pool, setup.drive);
 370
 371     let notify_user = setup.notify_user.as_ref().unwrap_or_else(|| &Userid::root_userid());
 372     let email = lookup_user_email(notify_user);
 373
 374     let upid_str = WorkerTask::new_thread(
 375         "tape-backup",
 376         Some(job_id),
 377         auth_id,
 378         to_stdout,
 379         move |worker| {
 380             let _drive_lock = drive_lock; // keep lock guard
 381             set_tape_device_state(&setup.drive, &worker.upid().to_string())?;
 382
 383             let mut summary = Default::default();
 384             let job_result = backup_worker(
 385                 &worker,
 386                 datastore,
 387                 &pool_config,
 388                 &setup,
 389                 email.clone(),
 390                 &mut summary,
 391                 force_media_set,
 392             );
 393
 394             if let Some(email) = email {
 395                 if let Err(err) = crate::server::send_tape_backup_status(
 396                     &email,
 397                     None,
 398                     &setup,
 399                     &job_result,
 400                     summary,
 401                 ) {
 402                     eprintln!("send tape backup notification failed: {}", err);
 403                 }
 404             }
 405
 406             // ignore errors
 407             let _ = set_tape_device_state(&setup.drive, "");
 408             job_result
 409         }
 410     )?;
 411
 412     Ok(upid_str.into())
 413 }
 414
 415 fn backup_worker(
 416     worker: &WorkerTask,
 417     datastore: Arc<DataStore>,
 418     pool_config: &MediaPoolConfig,
 419     setup: &TapeBackupJobSetup,
 420     email: Option<String>,
 421     summary: &mut TapeBackupJobSummary,
 422     force_media_set: bool,
 423 ) -> Result<(), Error> {
 424
 425     let status_path = Path::new(TAPE_STATUS_DIR);
 426     let start = std::time::Instant::now();
 427
 428     task_log!(worker, "update media online status");
 429     let changer_name = update_media_online_status(&setup.drive)?;
 430
 431     let pool = MediaPool::with_config(status_path, &pool_config, changer_name, false)?;
 432
 433     let mut pool_writer = PoolWriter::new(
 434         pool,
 435         &setup.drive,
 436         worker,
 437         email,
 438         force_media_set
 439     )?;
 440
 441     let mut group_list = BackupInfo::list_backup_groups(&datastore.base_path())?;
 442
 443     group_list.sort_unstable();
 444
 445     let group_count = group_list.len();
 446     task_log!(worker, "found {} groups", group_count);
 447
 448     let mut progress = StoreProgress::new(group_count as u64);
 449
 450     let latest_only = setup.latest_only.unwrap_or(false);
 451
 452     if latest_only {
 453         task_log!(worker, "latest-only: true (only considering latest snapshots)");
 454     }
 455
 456     let datastore_name = datastore.name();
 457
 458     let mut errors = false;
 459
 460     let mut need_catalog = false; // avoid writing catalog for empty jobs
 461
 462     for (group_number, group) in group_list.into_iter().enumerate() {
 463         progress.done_groups = group_number as u64;
 464         progress.done_snapshots = 0;
 465         progress.group_snapshots = 0;
 466
 467         let snapshot_list = group.list_backups(&datastore.base_path())?;
 468
 469         // filter out unfinished backups
 470         let mut snapshot_list: Vec<_> = snapshot_list
 471             .into_iter()
 472             .filter(|item| item.is_finished())
 473             .collect();
 474
 475         if snapshot_list.is_empty() {
 476             task_log!(worker, "group {} was empty", group);
 477             continue;
 478         }
 479
 480         BackupInfo::sort_list(&mut snapshot_list, true); // oldest first
 481
 482         if latest_only {
 483             progress.group_snapshots = 1;
 484             if let Some(info) = snapshot_list.pop() {
 485                 if pool_writer.contains_snapshot(datastore_name, &info.backup_dir.to_string()) {
 486                     task_log!(worker, "skip snapshot {}", info.backup_dir);
 487                     continue;
 488                 }
 489
 490                 need_catalog = true;
 491
 492                 let snapshot_name = info.backup_dir.to_string();
 493                 if !backup_snapshot(worker, &mut pool_writer, datastore.clone(), info.backup_dir)? {
 494                     errors = true;
 495                 } else {
 496                     summary.snapshot_list.push(snapshot_name);
 497                 }
 498                 progress.done_snapshots = 1;
 499                 task_log!(
 500                     worker,
 501                     "percentage done: {}",
 502                     progress
 503                 );
 504             }
 505         } else {
 506             progress.group_snapshots = snapshot_list.len() as u64;
 507             for (snapshot_number, info) in snapshot_list.into_iter().enumerate() {
 508                 if pool_writer.contains_snapshot(datastore_name, &info.backup_dir.to_string()) {
 509                     task_log!(worker, "skip snapshot {}", info.backup_dir);
 510                     continue;
 511                 }
 512
 513                 need_catalog = true;
 514
 515                 let snapshot_name = info.backup_dir.to_string();
 516                 if !backup_snapshot(worker, &mut pool_writer, datastore.clone(), info.backup_dir)? {
 517                     errors = true;
 518                 } else {
 519                     summary.snapshot_list.push(snapshot_name);
 520                 }
 521                 progress.done_snapshots = snapshot_number as u64 + 1;
 522                 task_log!(
 523                     worker,
 524                     "percentage done: {}",
 525                     progress
 526                 );
 527             }
 528         }
 529     }
 530
 531     pool_writer.commit()?;
 532
 533     if need_catalog {
 534         task_log!(worker, "append media catalog");
 535
 536         let uuid = pool_writer.load_writable_media(worker)?;
 537         let done = pool_writer.append_catalog_archive(worker)?;
 538         if !done {
 539             task_log!(worker, "catalog does not fit on tape, writing to next volume");
 540             pool_writer.set_media_status_full(&uuid)?;
 541             pool_writer.load_writable_media(worker)?;
 542             let done = pool_writer.append_catalog_archive(worker)?;
 543             if !done {
 544                 bail!("write_catalog_archive failed on second media");
 545             }
 546         }
 547     }
 548
 549     if setup.export_media_set.unwrap_or(false) {
 550         pool_writer.export_media_set(worker)?;
 551     } else if setup.eject_media.unwrap_or(false) {
 552         pool_writer.eject_media(worker)?;
 553     }
 554
 555     if errors {
 556         bail!("Tape backup finished with some errors. Please check the task log.");
 557     }
 558
 559     summary.duration = start.elapsed();
 560
 561     Ok(())
 562 }
 563
 564 // Try to update the the media online status
 565 fn update_media_online_status(drive: &str) -> Result<Option<String>, Error> {
 566
 567     let (config, _digest) = pbs_config::drive::config()?;
 568
 569     if let Ok(Some((mut changer, changer_name))) = media_changer(&config, drive) {
 570
 571          let label_text_list = changer.online_media_label_texts()?;
 572
 573         let status_path = Path::new(TAPE_STATUS_DIR);
 574         let mut inventory = Inventory::load(status_path)?;
 575
 576         update_changer_online_status(
 577             &config,
 578             &mut inventory,
 579             &changer_name,
 580             &label_text_list,
 581         )?;
 582
 583         Ok(Some(changer_name))
 584     } else {
 585         Ok(None)
 586     }
 587 }
 588
 589 pub fn backup_snapshot(
 590     worker: &WorkerTask,
 591     pool_writer: &mut PoolWriter,
 592     datastore: Arc<DataStore>,
 593     snapshot: BackupDir,
 594 ) -> Result<bool, Error> {
 595
 596     task_log!(worker, "backup snapshot {}", snapshot);
 597
 598     let snapshot_reader = match SnapshotReader::new(datastore.clone(), snapshot.clone()) {
 599         Ok(reader) => reader,
 600         Err(err) => {
 601             // ignore missing snapshots and continue
 602             task_warn!(worker, "failed opening snapshot '{}': {}", snapshot, err);
 603             return Ok(false);
 604         }
 605     };
 606
 607     let snapshot_reader = Arc::new(Mutex::new(snapshot_reader));
 608
 609     let (reader_thread, chunk_iter) = pool_writer.spawn_chunk_reader_thread(
 610         datastore.clone(),
 611         snapshot_reader.clone(),
 612     )?;
 613
 614     let mut chunk_iter = chunk_iter.peekable();
 615
 616     loop {
 617         worker.check_abort()?;
 618
 619         // test is we have remaining chunks
 620         match chunk_iter.peek() {
 621             None => break,
 622             Some(Ok(_)) => { /* Ok */ },
 623             Some(Err(err)) => bail!("{}", err),
 624         }
 625
 626         let uuid = pool_writer.load_writable_media(worker)?;
 627
 628         worker.check_abort()?;
 629
 630         let (leom, _bytes) = pool_writer.append_chunk_archive(worker, &mut chunk_iter, datastore.name())?;
 631
 632         if leom {
 633             pool_writer.set_media_status_full(&uuid)?;
 634         }
 635     }
 636
 637     if let Err(_) = reader_thread.join() {
 638         bail!("chunk reader thread failed");
 639     }
 640
 641     worker.check_abort()?;
 642
 643     let uuid = pool_writer.load_writable_media(worker)?;
 644
 645     worker.check_abort()?;
 646
 647     let snapshot_reader = snapshot_reader.lock().unwrap();
 648
 649     let (done, _bytes) = pool_writer.append_snapshot_archive(worker, &snapshot_reader)?;
 650
 651     if !done {
 652         // does not fit on tape, so we try on next volume
 653         pool_writer.set_media_status_full(&uuid)?;
 654
 655         worker.check_abort()?;
 656
 657         pool_writer.load_writable_media(worker)?;
 658         let (done, _bytes) = pool_writer.append_snapshot_archive(worker, &snapshot_reader)?;
 659
 660         if !done {
 661             bail!("write_snapshot_archive failed on second media");
 662         }
 663     }
 664
 665     task_log!(worker, "end backup {}:{}", datastore.name(), snapshot);
 666
 667     Ok(true)
 668 }