src/api2/admin/datastore.rs

   1 use std::collections::{HashSet, HashMap};
   2
   3 use chrono::{TimeZone, Local};
   4 use failure::*;
   5 use futures::*;
   6 use hyper::http::request::Parts;
   7 use hyper::{header, Body, Response, StatusCode};
   8 use serde_json::{json, Value};
   9
  10 use proxmox::{sortable, identity};
  11 use proxmox::api::{http_err, list_subdirs_api_method};
  12 use proxmox::api::{ApiFuture, ApiHandler, ApiMethod, Router, RpcEnvironment, RpcEnvironmentType};
  13 use proxmox::api::router::SubdirMap;
  14 use proxmox::api::schema::*;
  15 use proxmox::tools::{try_block, fs::file_get_contents, fs::file_set_contents};
  16
  17 use crate::api2::types::*;
  18 use crate::backup::*;
  19 use crate::config::datastore;
  20 use crate::server::WorkerTask;
  21 use crate::tools;
  22
  23 fn read_backup_index(store: &DataStore, backup_dir: &BackupDir) -> Result<Value, Error> {
  24
  25     let mut path = store.base_path();
  26     path.push(backup_dir.relative_path());
  27     path.push("index.json.blob");
  28
  29     let raw_data = file_get_contents(&path)?;
  30     let data = DataBlob::from_raw(raw_data)?.decode(None)?;
  31     let index_size = data.len();
  32     let mut result: Value = serde_json::from_reader(&mut &data[..])?;
  33
  34     let mut result = result["files"].take();
  35
  36     if result == Value::Null {
  37         bail!("missing 'files' property in backup index {:?}", path);
  38     }
  39
  40     result.as_array_mut().unwrap().push(json!({
  41         "filename": "index.json.blob",
  42         "size": index_size,
  43     }));
  44
  45     Ok(result)
  46 }
  47
  48 fn group_backups(backup_list: Vec<BackupInfo>) -> HashMap<String, Vec<BackupInfo>> {
  49
  50     let mut group_hash = HashMap::new();
  51
  52     for info in backup_list {
  53         let group_id = info.backup_dir.group().group_path().to_str().unwrap().to_owned();
  54         let time_list = group_hash.entry(group_id).or_insert(vec![]);
  55         time_list.push(info);
  56     }
  57
  58     group_hash
  59 }
  60
  61 fn list_groups(
  62     param: Value,
  63     _info: &ApiMethod,
  64     _rpcenv: &mut dyn RpcEnvironment,
  65 ) -> Result<Value, Error> {
  66
  67     let store = param["store"].as_str().unwrap();
  68
  69     let datastore = DataStore::lookup_datastore(store)?;
  70
  71     let backup_list = BackupInfo::list_backups(&datastore.base_path())?;
  72
  73     let group_hash = group_backups(backup_list);
  74
  75     let mut groups = vec![];
  76
  77     for (_group_id, mut list) in group_hash {
  78
  79         BackupInfo::sort_list(&mut list, false);
  80
  81         let info = &list[0];
  82         let group = info.backup_dir.group();
  83
  84         groups.push(json!({
  85             "backup-type": group.backup_type(),
  86             "backup-id": group.backup_id(),
  87             "last-backup": info.backup_dir.backup_time().timestamp(),
  88             "backup-count": list.len() as u64,
  89             "files": info.files,
  90         }));
  91     }
  92
  93     Ok(json!(groups))
  94 }
  95
  96 fn list_snapshot_files (
  97     param: Value,
  98     _info: &ApiMethod,
  99     _rpcenv: &mut dyn RpcEnvironment,
 100 ) -> Result<Value, Error> {
 101
 102     let store = tools::required_string_param(&param, "store")?;
 103     let backup_type = tools::required_string_param(&param, "backup-type")?;
 104     let backup_id = tools::required_string_param(&param, "backup-id")?;
 105     let backup_time = tools::required_integer_param(&param, "backup-time")?;
 106
 107     let datastore = DataStore::lookup_datastore(store)?;
 108     let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
 109
 110     let mut files = read_backup_index(&datastore, &snapshot)?;
 111
 112     let info = BackupInfo::new(&datastore.base_path(), snapshot)?;
 113
 114     let file_set = files.as_array().unwrap().iter().fold(HashSet::new(), |mut acc, item| {
 115         acc.insert(item["filename"].as_str().unwrap().to_owned());
 116         acc
 117     });
 118
 119     for file in info.files {
 120         if file_set.contains(&file) { continue; }
 121         files.as_array_mut().unwrap().push(json!({ "filename": file }));
 122     }
 123
 124     Ok(files)
 125 }
 126
 127 fn delete_snapshots (
 128     param: Value,
 129     _info: &ApiMethod,
 130     _rpcenv: &mut dyn RpcEnvironment,
 131 ) -> Result<Value, Error> {
 132
 133     let store = tools::required_string_param(&param, "store")?;
 134     let backup_type = tools::required_string_param(&param, "backup-type")?;
 135     let backup_id = tools::required_string_param(&param, "backup-id")?;
 136     let backup_time = tools::required_integer_param(&param, "backup-time")?;
 137
 138     let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
 139
 140     let datastore = DataStore::lookup_datastore(store)?;
 141
 142     datastore.remove_backup_dir(&snapshot)?;
 143
 144     Ok(Value::Null)
 145 }
 146
 147 fn list_snapshots (
 148     param: Value,
 149     _info: &ApiMethod,
 150     _rpcenv: &mut dyn RpcEnvironment,
 151 ) -> Result<Value, Error> {
 152
 153     let store = tools::required_string_param(&param, "store")?;
 154     let backup_type = param["backup-type"].as_str();
 155     let backup_id = param["backup-id"].as_str();
 156
 157     let datastore = DataStore::lookup_datastore(store)?;
 158
 159     let base_path = datastore.base_path();
 160
 161     let backup_list = BackupInfo::list_backups(&base_path)?;
 162
 163     let mut snapshots = vec![];
 164
 165     for info in backup_list {
 166         let group = info.backup_dir.group();
 167         if let Some(backup_type) = backup_type {
 168             if backup_type != group.backup_type() { continue; }
 169         }
 170         if let Some(backup_id) = backup_id {
 171             if backup_id != group.backup_id() { continue; }
 172         }
 173
 174         let mut result_item = json!({
 175             "backup-type": group.backup_type(),
 176             "backup-id": group.backup_id(),
 177             "backup-time": info.backup_dir.backup_time().timestamp(),
 178             "files": info.files,
 179         });
 180
 181         if let Ok(index) = read_backup_index(&datastore, &info.backup_dir) {
 182             let mut backup_size = 0;
 183             for item in index.as_array().unwrap().iter() {
 184                 if let Some(item_size) = item["size"].as_u64() {
 185                     backup_size += item_size;
 186                 }
 187             }
 188             result_item["size"] = backup_size.into();
 189         }
 190
 191         snapshots.push(result_item);
 192     }
 193
 194     Ok(json!(snapshots))
 195 }
 196
 197 #[sortable]
 198 const API_METHOD_STATUS: ApiMethod = ApiMethod::new(
 199     &ApiHandler::Sync(&status),
 200     &ObjectSchema::new(
 201         "Get datastore status.",
 202         &sorted!([
 203             ("store", false, &DATASTORE_SCHEMA),
 204         ]),
 205     )
 206 );
 207
 208 fn status(
 209     param: Value,
 210     _info: &ApiMethod,
 211     _rpcenv: &mut dyn RpcEnvironment,
 212 ) -> Result<Value, Error> {
 213
 214     let store = param["store"].as_str().unwrap();
 215
 216     let datastore = DataStore::lookup_datastore(store)?;
 217
 218     let base_path = datastore.base_path();
 219
 220     let mut stat: libc::statfs64 = unsafe { std::mem::zeroed() };
 221
 222     use nix::NixPath;
 223
 224     let res = base_path.with_nix_path(|cstr| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) })?;
 225     nix::errno::Errno::result(res)?;
 226
 227     let bsize = stat.f_bsize as u64;
 228     Ok(json!({
 229         "total": stat.f_blocks*bsize,
 230         "used": (stat.f_blocks-stat.f_bfree)*bsize,
 231         "avail": stat.f_bavail*bsize,
 232     }))
 233 }
 234
 235 #[macro_export]
 236 macro_rules! add_common_prune_prameters {
 237     ( [ $( $list1:tt )* ] ) => {
 238         add_common_prune_prameters!([$( $list1 )* ] ,  [])
 239     };
 240     ( [ $( $list1:tt )* ] ,  [ $( $list2:tt )* ] ) => {
 241         [
 242             $( $list1 )*
 243             (
 244                 "keep-daily",
 245                 true,
 246                 &IntegerSchema::new("Number of daily backups to keep.")
 247                     .minimum(1)
 248                     .schema()
 249             ),
 250             (
 251                 "keep-hourly",
 252                 true,
 253                 &IntegerSchema::new("Number of hourly backups to keep.")
 254                     .minimum(1)
 255                     .schema()
 256             ),
 257             (
 258                 "keep-last",
 259                 true,
 260                 &IntegerSchema::new("Number of backups to keep.")
 261                     .minimum(1)
 262                     .schema()
 263             ),
 264             (
 265                 "keep-monthly",
 266                 true,
 267                 &IntegerSchema::new("Number of monthly backups to keep.")
 268                     .minimum(1)
 269                     .schema()
 270             ),
 271             (
 272                 "keep-weekly",
 273                 true,
 274                 &IntegerSchema::new("Number of weekly backups to keep.")
 275                     .minimum(1)
 276                     .schema()
 277             ),
 278             (
 279                 "keep-yearly",
 280                 true,
 281                 &IntegerSchema::new("Number of yearly backups to keep.")
 282                     .minimum(1)
 283                     .schema()
 284             ),
 285             $( $list2 )*
 286         ]
 287     }
 288 }
 289
 290 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
 291     &ApiHandler::Sync(&prune),
 292     &ObjectSchema::new(
 293         "Prune the datastore.",
 294         &add_common_prune_prameters!([
 295             ("backup-id", false, &BACKUP_ID_SCHEMA),
 296             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 297             ("dry-run", true, &BooleanSchema::new(
 298                 "Just show what prune would do, but do not delete anything.")
 299              .schema()
 300             ),
 301         ],[
 302             ("store", false, &DATASTORE_SCHEMA),
 303         ])
 304     )
 305 );
 306
 307 fn prune(
 308     param: Value,
 309     _info: &ApiMethod,
 310     _rpcenv: &mut dyn RpcEnvironment,
 311 ) -> Result<Value, Error> {
 312
 313     let store = param["store"].as_str().unwrap();
 314
 315     let backup_type = tools::required_string_param(&param, "backup-type")?;
 316     let backup_id = tools::required_string_param(&param, "backup-id")?;
 317
 318     let dry_run = param["dry-run"].as_bool().unwrap_or(false);
 319
 320     let group = BackupGroup::new(backup_type, backup_id);
 321
 322     let datastore = DataStore::lookup_datastore(store)?;
 323
 324     let prune_options = PruneOptions {
 325         keep_last: param["keep-last"].as_u64(),
 326         keep_hourly: param["keep-hourly"].as_u64(),
 327         keep_daily: param["keep-daily"].as_u64(),
 328         keep_weekly: param["keep-weekly"].as_u64(),
 329         keep_monthly: param["keep-monthly"].as_u64(),
 330         keep_yearly: param["keep-yearly"].as_u64(),
 331     };
 332
 333     // We use a WorkerTask just to have a task log, but run synchrounously
 334     let worker = WorkerTask::new("prune", Some(store.to_owned()), "root@pam", true)?;
 335     let result = try_block! {
 336         if !prune_options.keeps_something() {
 337             worker.log("No prune selection - keeping all files.");
 338             return Ok(());
 339         } else {
 340             if dry_run {
 341                 worker.log(format!("Testing prune on store {}", store));
 342             } else {
 343                 worker.log(format!("Starting prune on store {}", store));
 344             }
 345         }
 346
 347         let list = group.list_backups(&datastore.base_path())?;
 348
 349         let mut prune_info = compute_prune_info(list, &prune_options)?;
 350
 351         prune_info.reverse(); // delete older snapshots first
 352
 353         for (info, keep) in prune_info {
 354             let backup_time = info.backup_dir.backup_time();
 355             let timestamp = BackupDir::backup_time_to_string(backup_time);
 356             let group = info.backup_dir.group();
 357
 358             let msg = format!(
 359                 "{}/{}/{} {}",
 360                 group.backup_type(),
 361                 group.backup_id(),
 362                 timestamp,
 363                 if keep { "keep" } else { "remove" },
 364             );
 365
 366             worker.log(msg);
 367
 368             if !(dry_run || keep) {
 369                 datastore.remove_backup_dir(&info.backup_dir)?;
 370             }
 371         }
 372
 373         Ok(())
 374     };
 375
 376     worker.log_result(&result);
 377
 378     if let Err(err) = result {
 379         bail!("prune failed - {}", err);
 380     }
 381
 382     Ok(json!(worker.to_string())) // return the UPID
 383 }
 384
 385 #[sortable]
 386 pub const API_METHOD_START_GARBAGE_COLLECTION: ApiMethod = ApiMethod::new(
 387     &ApiHandler::Sync(&start_garbage_collection),
 388     &ObjectSchema::new(
 389         "Start garbage collection.",
 390         &sorted!([
 391             ("store", false, &DATASTORE_SCHEMA),
 392         ])
 393     )
 394 );
 395
 396 fn start_garbage_collection(
 397     param: Value,
 398     _info: &ApiMethod,
 399     rpcenv: &mut dyn RpcEnvironment,
 400 ) -> Result<Value, Error> {
 401
 402     let store = param["store"].as_str().unwrap().to_string();
 403
 404     let datastore = DataStore::lookup_datastore(&store)?;
 405
 406     println!("Starting garbage collection on store {}", store);
 407
 408     let to_stdout = if rpcenv.env_type() == RpcEnvironmentType::CLI { true } else { false };
 409
 410     let upid_str = WorkerTask::new_thread(
 411         "garbage_collection", Some(store.clone()), "root@pam", to_stdout, move |worker|
 412         {
 413             worker.log(format!("starting garbage collection on store {}", store));
 414             datastore.garbage_collection(worker)
 415         })?;
 416
 417     Ok(json!(upid_str))
 418 }
 419
 420 #[sortable]
 421 pub const API_METHOD_GARBAGE_COLLECTION_STATUS: ApiMethod = ApiMethod::new(
 422     &ApiHandler::Sync(&garbage_collection_status),
 423     &ObjectSchema::new(
 424         "Garbage collection status.",
 425         &sorted!([
 426             ("store", false, &DATASTORE_SCHEMA),
 427         ])
 428     )
 429 );
 430
 431 fn garbage_collection_status(
 432     param: Value,
 433     _info: &ApiMethod,
 434     _rpcenv: &mut dyn RpcEnvironment,
 435 ) -> Result<Value, Error> {
 436
 437     let store = param["store"].as_str().unwrap();
 438
 439     let datastore = DataStore::lookup_datastore(&store)?;
 440
 441     println!("Garbage collection status on store {}", store);
 442
 443     let status = datastore.last_gc_status();
 444
 445     Ok(serde_json::to_value(&status)?)
 446 }
 447
 448
 449 fn get_datastore_list(
 450     _param: Value,
 451     _info: &ApiMethod,
 452     _rpcenv: &mut dyn RpcEnvironment,
 453 ) -> Result<Value, Error> {
 454
 455     let config = datastore::config()?;
 456
 457     Ok(config.convert_to_array("store"))
 458 }
 459
 460 #[sortable]
 461 pub const API_METHOD_DOWNLOAD_FILE: ApiMethod = ApiMethod::new(
 462     &ApiHandler::AsyncHttp(&download_file),
 463     &ObjectSchema::new(
 464         "Download single raw file from backup snapshot.",
 465         &sorted!([
 466             ("store", false, &DATASTORE_SCHEMA),
 467             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 468             ("backup-id", false,  &BACKUP_ID_SCHEMA),
 469             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 470             ("file-name", false, &StringSchema::new("Raw file name.")
 471              .format(&FILENAME_FORMAT)
 472              .schema()
 473             ),
 474         ]),
 475     )
 476 );
 477
 478 fn download_file(
 479     _parts: Parts,
 480     _req_body: Body,
 481     param: Value,
 482     _info: &ApiMethod,
 483     _rpcenv: Box<dyn RpcEnvironment>,
 484 ) -> ApiFuture {
 485
 486     async move {
 487         let store = tools::required_string_param(&param, "store")?;
 488
 489         let datastore = DataStore::lookup_datastore(store)?;
 490
 491         let file_name = tools::required_string_param(&param, "file-name")?.to_owned();
 492
 493         let backup_type = tools::required_string_param(&param, "backup-type")?;
 494         let backup_id = tools::required_string_param(&param, "backup-id")?;
 495         let backup_time = tools::required_integer_param(&param, "backup-time")?;
 496
 497         println!("Download {} from {} ({}/{}/{}/{})", file_name, store,
 498                  backup_type, backup_id, Local.timestamp(backup_time, 0), file_name);
 499
 500         let backup_dir = BackupDir::new(backup_type, backup_id, backup_time);
 501
 502         let mut path = datastore.base_path();
 503         path.push(backup_dir.relative_path());
 504         path.push(&file_name);
 505
 506         let file = tokio::fs::File::open(path)
 507             .map_err(|err| http_err!(BAD_REQUEST, format!("File open failed: {}", err)))
 508             .await?;
 509
 510         let payload = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 511             .map_ok(|bytes| hyper::body::Bytes::from(bytes.freeze()));
 512         let body = Body::wrap_stream(payload);
 513
 514         // fixme: set other headers ?
 515         Ok(Response::builder()
 516            .status(StatusCode::OK)
 517            .header(header::CONTENT_TYPE, "application/octet-stream")
 518            .body(body)
 519            .unwrap())
 520     }.boxed()
 521 }
 522
 523 #[sortable]
 524 pub const API_METHOD_UPLOAD_BACKUP_LOG: ApiMethod = ApiMethod::new(
 525     &ApiHandler::AsyncHttp(&upload_backup_log),
 526     &ObjectSchema::new(
 527         "Download single raw file from backup snapshot.",
 528         &sorted!([
 529             ("store", false, &DATASTORE_SCHEMA),
 530             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 531             ("backup-id", false, &BACKUP_ID_SCHEMA),
 532             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 533         ]),
 534     )
 535 );
 536
 537 fn upload_backup_log(
 538     _parts: Parts,
 539     req_body: Body,
 540     param: Value,
 541     _info: &ApiMethod,
 542     _rpcenv: Box<dyn RpcEnvironment>,
 543 ) -> ApiFuture {
 544
 545     async move {
 546         let store = tools::required_string_param(&param, "store")?;
 547
 548         let datastore = DataStore::lookup_datastore(store)?;
 549
 550         let file_name = "client.log.blob";
 551
 552         let backup_type = tools::required_string_param(&param, "backup-type")?;
 553         let backup_id = tools::required_string_param(&param, "backup-id")?;
 554         let backup_time = tools::required_integer_param(&param, "backup-time")?;
 555
 556         let backup_dir = BackupDir::new(backup_type, backup_id, backup_time);
 557
 558         let mut path = datastore.base_path();
 559         path.push(backup_dir.relative_path());
 560         path.push(&file_name);
 561
 562         if path.exists() {
 563             bail!("backup already contains a log.");
 564         }
 565
 566         println!("Upload backup log to {}/{}/{}/{}/{}", store,
 567                  backup_type, backup_id, BackupDir::backup_time_to_string(backup_dir.backup_time()), file_name);
 568
 569         let data = req_body
 570             .map_err(Error::from)
 571             .try_fold(Vec::new(), |mut acc, chunk| {
 572                 acc.extend_from_slice(&*chunk);
 573                 future::ok::<_, Error>(acc)
 574             })
 575             .await?;
 576
 577         let blob = DataBlob::from_raw(data)?;
 578         // always verify CRC at server side
 579         blob.verify_crc()?;
 580         let raw_data = blob.raw_data();
 581         file_set_contents(&path, raw_data, None)?;
 582
 583         // fixme: use correct formatter
 584         Ok(crate::server::formatter::json_response(Ok(Value::Null)))
 585     }.boxed()
 586 }
 587
 588 #[sortable]
 589 const DATASTORE_INFO_SUBDIRS: SubdirMap = &[
 590     (
 591         "download",
 592         &Router::new()
 593             .download(&API_METHOD_DOWNLOAD_FILE)
 594     ),
 595     (
 596         "files",
 597         &Router::new()
 598             .get(
 599                 &ApiMethod::new(
 600                     &ApiHandler::Sync(&list_snapshot_files),
 601                     &ObjectSchema::new(
 602                         "List snapshot files.",
 603                         &sorted!([
 604                             ("store", false, &DATASTORE_SCHEMA),
 605                             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 606                             ("backup-id", false, &BACKUP_ID_SCHEMA),
 607                             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 608                         ]),
 609                     )
 610                 )
 611             )
 612     ),
 613     (
 614         "gc",
 615         &Router::new()
 616             .get(&API_METHOD_GARBAGE_COLLECTION_STATUS)
 617             .post(&API_METHOD_START_GARBAGE_COLLECTION)
 618     ),
 619     (
 620         "groups",
 621         &Router::new()
 622             .get(
 623                 &ApiMethod::new(
 624                     &ApiHandler::Sync(&list_groups),
 625                     &ObjectSchema::new(
 626                         "List backup groups.",
 627                         &sorted!([ ("store", false, &DATASTORE_SCHEMA) ]),
 628                     )
 629                 )
 630             )
 631     ),
 632     (
 633         "prune",
 634         &Router::new()
 635             .post(&API_METHOD_PRUNE)
 636     ),
 637     (
 638         "snapshots",
 639         &Router::new()
 640             .get(
 641                 &ApiMethod::new(
 642                     &ApiHandler::Sync(&list_snapshots),
 643                     &ObjectSchema::new(
 644                         "List backup groups.",
 645                         &sorted!([
 646                             ("store", false, &DATASTORE_SCHEMA),
 647                             ("backup-type", true, &BACKUP_TYPE_SCHEMA),
 648                             ("backup-id", true, &BACKUP_ID_SCHEMA),
 649                         ]),
 650                     )
 651                 )
 652             )
 653             .delete(
 654                 &ApiMethod::new(
 655                     &ApiHandler::Sync(&delete_snapshots),
 656                     &ObjectSchema::new(
 657                         "Delete backup snapshot.",
 658                         &sorted!([
 659                             ("store", false, &DATASTORE_SCHEMA),
 660                             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 661                             ("backup-id", false, &BACKUP_ID_SCHEMA),
 662                             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 663                         ]),
 664                     )
 665                 )
 666             )
 667     ),
 668     (
 669         "status",
 670         &Router::new()
 671             .get(&API_METHOD_STATUS)
 672     ),
 673     (
 674         "upload-backup-log",
 675         &Router::new()
 676             .upload(&API_METHOD_UPLOAD_BACKUP_LOG)
 677     ),
 678 ];
 679
 680 const DATASTORE_INFO_ROUTER: Router = Router::new()
 681     .get(&list_subdirs_api_method!(DATASTORE_INFO_SUBDIRS))
 682     .subdirs(DATASTORE_INFO_SUBDIRS);
 683
 684
 685 pub const ROUTER: Router = Router::new()
 686     .get(
 687         &ApiMethod::new(
 688             &ApiHandler::Sync(&get_datastore_list),
 689             &ObjectSchema::new("Directory index.", &[])
 690         )
 691     )
 692     .match_all("store", &DATASTORE_INFO_ROUTER);