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