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 fn status(
 198     param: Value,
 199     _info: &ApiMethod,
 200     _rpcenv: &mut dyn RpcEnvironment,
 201 ) -> Result<Value, Error> {
 202
 203     let store = param["store"].as_str().unwrap();
 204
 205     let datastore = DataStore::lookup_datastore(store)?;
 206
 207     let base_path = datastore.base_path();
 208
 209     let mut stat: libc::statfs64 = unsafe { std::mem::zeroed() };
 210
 211     use nix::NixPath;
 212
 213     let res = base_path.with_nix_path(|cstr| unsafe { libc::statfs64(cstr.as_ptr(), &mut stat) })?;
 214     nix::errno::Errno::result(res)?;
 215
 216     let bsize = stat.f_bsize as u64;
 217     Ok(json!({
 218         "total": stat.f_blocks*bsize,
 219         "used": (stat.f_blocks-stat.f_bfree)*bsize,
 220         "avail": stat.f_bavail*bsize,
 221     }))
 222 }
 223
 224 #[macro_export]
 225 macro_rules! add_common_prune_prameters {
 226     ( [ $( $list1:tt )* ] ) => {
 227         add_common_prune_prameters!([$( $list1 )* ] ,  [])
 228     };
 229     ( [ $( $list1:tt )* ] ,  [ $( $list2:tt )* ] ) => {
 230         [
 231             $( $list1 )*
 232             (
 233                 "keep-daily",
 234                 true,
 235                 &IntegerSchema::new("Number of daily backups to keep.")
 236                     .minimum(1)
 237                     .schema()
 238             ),
 239             (
 240                 "keep-last",
 241                 true,
 242                 &IntegerSchema::new("Number of backups to keep.")
 243                     .minimum(1)
 244                     .schema()
 245             ),
 246             (
 247                 "keep-monthly",
 248                 true,
 249                 &IntegerSchema::new("Number of monthly backups to keep.")
 250                     .minimum(1)
 251                     .schema()
 252             ),
 253             (
 254                 "keep-weekly",
 255                 true,
 256                 &IntegerSchema::new("Number of weekly backups to keep.")
 257                     .minimum(1)
 258                     .schema()
 259             ),
 260             (
 261                 "keep-yearly",
 262                 true,
 263                 &IntegerSchema::new("Number of yearly backups to keep.")
 264                     .minimum(1)
 265                     .schema()
 266             ),
 267             $( $list2 )*
 268         ]
 269     }
 270 }
 271
 272 const API_METHOD_STATUS: ApiMethod = ApiMethod::new(
 273     &ApiHandler::Sync(&status),
 274     &ObjectSchema::new(
 275         "Get datastore status.",
 276         &add_common_prune_prameters!([],[
 277             ("store", false, &StringSchema::new("Datastore name.").schema()),
 278         ]),
 279     )
 280 );
 281
 282 fn prune(
 283     param: Value,
 284     _info: &ApiMethod,
 285     _rpcenv: &mut dyn RpcEnvironment,
 286 ) -> Result<Value, Error> {
 287
 288     let store = param["store"].as_str().unwrap();
 289
 290     let backup_type = tools::required_string_param(&param, "backup-type")?;
 291     let backup_id = tools::required_string_param(&param, "backup-id")?;
 292
 293     let group = BackupGroup::new(backup_type, backup_id);
 294
 295     let datastore = DataStore::lookup_datastore(store)?;
 296
 297     let prune_options = PruneOptions {
 298         keep_last: param["keep-last"].as_u64(),
 299         keep_daily: param["keep-daily"].as_u64(),
 300         keep_weekly: param["keep-weekly"].as_u64(),
 301         keep_monthly: param["keep-monthly"].as_u64(),
 302         keep_yearly: param["keep-yearly"].as_u64(),
 303     };
 304
 305     let worker = WorkerTask::new("prune", Some(store.to_owned()), "root@pam", true)?;
 306     let result = try_block! {
 307         if !prune_options.keeps_something() {
 308             worker.log("No prune selection - keeping all files.");
 309             return Ok(());
 310         } else {
 311             worker.log(format!("Starting prune on store {}", store));
 312         }
 313
 314         let list = group.list_backups(&datastore.base_path())?;
 315
 316         let mut prune_info = compute_prune_info(list, &prune_options)?;
 317
 318         prune_info.reverse(); // delete older snapshots first
 319
 320         for (info, keep) in prune_info {
 321             if keep {
 322                 worker.log(format!("keep {:?}", info.backup_dir.relative_path()));
 323             } else {
 324                 worker.log(format!("remove {:?}", info.backup_dir.relative_path()));
 325                 datastore.remove_backup_dir(&info.backup_dir)?;
 326             }
 327         }
 328
 329         Ok(())
 330     };
 331
 332     worker.log_result(&result);
 333
 334     if let Err(err) = result {
 335         bail!("prune failed - {}", err);
 336     }
 337
 338     Ok(json!(null))
 339 }
 340
 341 const API_METHOD_PRUNE: ApiMethod = ApiMethod::new(
 342     &ApiHandler::Sync(&prune),
 343     &ObjectSchema::new(
 344         "Prune the datastore.",
 345         &add_common_prune_prameters!([
 346             ("backup-id", false, &BACKUP_ID_SCHEMA),
 347             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 348         ],[
 349             ("store", false, &StringSchema::new("Datastore name.").schema()),
 350         ])
 351     )
 352 );
 353
 354 fn start_garbage_collection(
 355     param: Value,
 356     _info: &ApiMethod,
 357     rpcenv: &mut dyn RpcEnvironment,
 358 ) -> Result<Value, Error> {
 359
 360     let store = param["store"].as_str().unwrap().to_string();
 361
 362     let datastore = DataStore::lookup_datastore(&store)?;
 363
 364     println!("Starting garbage collection on store {}", store);
 365
 366     let to_stdout = if rpcenv.env_type() == RpcEnvironmentType::CLI { true } else { false };
 367
 368     let upid_str = WorkerTask::new_thread(
 369         "garbage_collection", Some(store.clone()), "root@pam", to_stdout, move |worker|
 370         {
 371             worker.log(format!("starting garbage collection on store {}", store));
 372             datastore.garbage_collection(worker)
 373         })?;
 374
 375     Ok(json!(upid_str))
 376 }
 377
 378 #[sortable]
 379 pub const API_METHOD_START_GARBAGE_COLLECTION: ApiMethod = ApiMethod::new(
 380     &ApiHandler::Sync(&start_garbage_collection),
 381     &ObjectSchema::new(
 382         "Start garbage collection.",
 383         &sorted!([
 384             ("store", false, &StringSchema::new("Datastore name.").schema()),
 385         ])
 386     )
 387 );
 388
 389 fn garbage_collection_status(
 390     param: Value,
 391     _info: &ApiMethod,
 392     _rpcenv: &mut dyn RpcEnvironment,
 393 ) -> Result<Value, Error> {
 394
 395     let store = param["store"].as_str().unwrap();
 396
 397     let datastore = DataStore::lookup_datastore(&store)?;
 398
 399     println!("Garbage collection status on store {}", store);
 400
 401     let status = datastore.last_gc_status();
 402
 403     Ok(serde_json::to_value(&status)?)
 404 }
 405
 406 #[sortable]
 407 pub const API_METHOD_GARBAGE_COLLECTION_STATUS: ApiMethod = ApiMethod::new(
 408     &ApiHandler::Sync(&garbage_collection_status),
 409     &ObjectSchema::new(
 410         "Garbage collection status.",
 411         &sorted!([
 412             ("store", false, &StringSchema::new("Datastore name.").schema()),
 413         ])
 414     )
 415 );
 416
 417 fn get_datastore_list(
 418     _param: Value,
 419     _info: &ApiMethod,
 420     _rpcenv: &mut dyn RpcEnvironment,
 421 ) -> Result<Value, Error> {
 422
 423     let config = datastore::config()?;
 424
 425     Ok(config.convert_to_array("store"))
 426 }
 427
 428
 429 fn download_file(
 430     _parts: Parts,
 431     _req_body: Body,
 432     param: Value,
 433     _info: &ApiMethod,
 434     _rpcenv: Box<dyn RpcEnvironment>,
 435 ) -> ApiFuture {
 436
 437     async move {
 438         let store = tools::required_string_param(&param, "store")?;
 439
 440         let datastore = DataStore::lookup_datastore(store)?;
 441
 442         let file_name = tools::required_string_param(&param, "file-name")?.to_owned();
 443
 444         let backup_type = tools::required_string_param(&param, "backup-type")?;
 445         let backup_id = tools::required_string_param(&param, "backup-id")?;
 446         let backup_time = tools::required_integer_param(&param, "backup-time")?;
 447
 448         println!("Download {} from {} ({}/{}/{}/{})", file_name, store,
 449                  backup_type, backup_id, Local.timestamp(backup_time, 0), file_name);
 450
 451         let backup_dir = BackupDir::new(backup_type, backup_id, backup_time);
 452
 453         let mut path = datastore.base_path();
 454         path.push(backup_dir.relative_path());
 455         path.push(&file_name);
 456
 457         let file = tokio::fs::File::open(path)
 458             .map_err(|err| http_err!(BAD_REQUEST, format!("File open failed: {}", err)))
 459             .await?;
 460
 461         let payload = tokio::codec::FramedRead::new(file, tokio::codec::BytesCodec::new())
 462             .map_ok(|bytes| hyper::Chunk::from(bytes.freeze()));
 463         let body = Body::wrap_stream(payload);
 464
 465         // fixme: set other headers ?
 466         Ok(Response::builder()
 467            .status(StatusCode::OK)
 468            .header(header::CONTENT_TYPE, "application/octet-stream")
 469            .body(body)
 470            .unwrap())
 471     }.boxed()
 472 }
 473
 474 #[sortable]
 475 pub const API_METHOD_DOWNLOAD_FILE: ApiMethod = ApiMethod::new(
 476     &ApiHandler::AsyncHttp(&download_file),
 477     &ObjectSchema::new(
 478         "Download single raw file from backup snapshot.",
 479         &sorted!([
 480             ("store", false, &StringSchema::new("Datastore name.").schema()),
 481             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 482             ("backup-id", false,  &BACKUP_ID_SCHEMA),
 483             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 484             ("file-name", false, &StringSchema::new("Raw file name.")
 485              .format(&FILENAME_FORMAT)
 486              .schema()
 487             ),
 488         ]),
 489     )
 490 );
 491
 492 fn upload_backup_log(
 493     _parts: Parts,
 494     req_body: Body,
 495     param: Value,
 496     _info: &ApiMethod,
 497     _rpcenv: Box<dyn RpcEnvironment>,
 498 ) -> ApiFuture {
 499
 500     async move {
 501         let store = tools::required_string_param(&param, "store")?;
 502
 503         let datastore = DataStore::lookup_datastore(store)?;
 504
 505         let file_name = "client.log.blob";
 506
 507         let backup_type = tools::required_string_param(&param, "backup-type")?;
 508         let backup_id = tools::required_string_param(&param, "backup-id")?;
 509         let backup_time = tools::required_integer_param(&param, "backup-time")?;
 510
 511         let backup_dir = BackupDir::new(backup_type, backup_id, backup_time);
 512
 513         let mut path = datastore.base_path();
 514         path.push(backup_dir.relative_path());
 515         path.push(&file_name);
 516
 517         if path.exists() {
 518             bail!("backup already contains a log.");
 519         }
 520
 521         println!("Upload backup log to {}/{}/{}/{}/{}", store,
 522                  backup_type, backup_id, BackupDir::backup_time_to_string(backup_dir.backup_time()), file_name);
 523
 524         let data = req_body
 525             .map_err(Error::from)
 526             .try_fold(Vec::new(), |mut acc, chunk| {
 527                 acc.extend_from_slice(&*chunk);
 528                 future::ok::<_, Error>(acc)
 529             })
 530             .await?;
 531
 532         let blob = DataBlob::from_raw(data)?;
 533         // always verify CRC at server side
 534         blob.verify_crc()?;
 535         let raw_data = blob.raw_data();
 536         file_set_contents(&path, raw_data, None)?;
 537
 538         // fixme: use correct formatter
 539         Ok(crate::server::formatter::json_response(Ok(Value::Null)))
 540     }.boxed()
 541 }
 542
 543 #[sortable]
 544 pub const API_METHOD_UPLOAD_BACKUP_LOG: ApiMethod = ApiMethod::new(
 545     &ApiHandler::AsyncHttp(&upload_backup_log),
 546     &ObjectSchema::new(
 547         "Download single raw file from backup snapshot.",
 548         &sorted!([
 549             ("store", false, &StringSchema::new("Datastore name.").schema()),
 550             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 551             ("backup-id", false, &BACKUP_ID_SCHEMA),
 552             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 553         ]),
 554     )
 555 );
 556
 557 const STORE_SCHEMA: Schema = StringSchema::new("Datastore name.").schema();
 558
 559 #[sortable]
 560 const DATASTORE_INFO_SUBDIRS: SubdirMap = &[
 561     (
 562         "download",
 563         &Router::new()
 564             .download(&API_METHOD_DOWNLOAD_FILE)
 565     ),
 566     (
 567         "files",
 568         &Router::new()
 569             .get(
 570                 &ApiMethod::new(
 571                     &ApiHandler::Sync(&list_snapshot_files),
 572                     &ObjectSchema::new(
 573                         "List snapshot files.",
 574                         &sorted!([
 575                             ("store", false, &STORE_SCHEMA),
 576                             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 577                             ("backup-id", false, &BACKUP_ID_SCHEMA),
 578                             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 579                         ]),
 580                     )
 581                 )
 582             )
 583     ),
 584     (
 585         "gc",
 586         &Router::new()
 587             .get(&API_METHOD_GARBAGE_COLLECTION_STATUS)
 588             .post(&API_METHOD_START_GARBAGE_COLLECTION)
 589     ),
 590     (
 591         "groups",
 592         &Router::new()
 593             .get(
 594                 &ApiMethod::new(
 595                     &ApiHandler::Sync(&list_groups),
 596                     &ObjectSchema::new(
 597                         "List backup groups.",
 598                         &sorted!([ ("store", false, &STORE_SCHEMA) ]),
 599                     )
 600                 )
 601             )
 602     ),
 603     (
 604         "prune",
 605         &Router::new()
 606             .post(&API_METHOD_PRUNE)
 607     ),
 608     (
 609         "snapshots",
 610         &Router::new()
 611             .get(
 612                 &ApiMethod::new(
 613                     &ApiHandler::Sync(&list_snapshots),
 614                     &ObjectSchema::new(
 615                         "List backup groups.",
 616                         &sorted!([
 617                             ("store", false, &STORE_SCHEMA),
 618                             ("backup-type", true, &BACKUP_TYPE_SCHEMA),
 619                             ("backup-id", true, &BACKUP_ID_SCHEMA),
 620                         ]),
 621                     )
 622                 )
 623             )
 624             .delete(
 625                 &ApiMethod::new(
 626                     &ApiHandler::Sync(&delete_snapshots),
 627                     &ObjectSchema::new(
 628                         "Delete backup snapshot.",
 629                         &sorted!([
 630                             ("store", false, &STORE_SCHEMA),
 631                             ("backup-type", false, &BACKUP_TYPE_SCHEMA),
 632                             ("backup-id", false, &BACKUP_ID_SCHEMA),
 633                             ("backup-time", false, &BACKUP_TIME_SCHEMA),
 634                         ]),
 635                     )
 636                 )
 637             )
 638     ),
 639     (
 640         "status",
 641         &Router::new()
 642             .get(&API_METHOD_STATUS)
 643     ),
 644     (
 645         "upload-backup-log",
 646         &Router::new()
 647             .upload(&API_METHOD_UPLOAD_BACKUP_LOG)
 648     ),
 649 ];
 650
 651 const DATASTORE_INFO_ROUTER: Router = Router::new()
 652     .get(&list_subdirs_api_method!(DATASTORE_INFO_SUBDIRS))
 653     .subdirs(DATASTORE_INFO_SUBDIRS);
 654
 655
 656 pub const ROUTER: Router = Router::new()
 657     .get(
 658         &ApiMethod::new(
 659             &ApiHandler::Sync(&get_datastore_list),
 660             &ObjectSchema::new("Directory index.", &[])
 661         )
 662     )
 663     .match_all("store", &DATASTORE_INFO_ROUTER);