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