src/api2/admin/datastore.rs

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