src/api2/admin/datastore.rs

   1 use failure::*;
   2
   3 use crate::tools;
   4 use crate::api_schema::*;
   5 use crate::api_schema::router::*;
   6 //use crate::server::rest::*;
   7 use serde_json::{json, Value};
   8 use std::collections::{HashSet, HashMap};
   9 use chrono::{DateTime, Datelike, Local};
  10 use std::path::PathBuf;
  11 use std::sync::Arc;
  12
  13 //use hyper::StatusCode;
  14 //use hyper::rt::{Future, Stream};
  15
  16 use crate::config::datastore;
  17
  18 use crate::backup::*;
  19
  20 mod catar;
  21 mod upload;
  22
  23 fn group_backups(backup_list: Vec<BackupInfo>) -> HashMap<String, Vec<BackupInfo>> {
  24
  25     let mut group_hash = HashMap::new();
  26
  27     for info in backup_list {
  28         let group_id = info.backup_dir.group().group_path().to_str().unwrap().to_owned();
  29         let time_list = group_hash.entry(group_id).or_insert(vec![]);
  30         time_list.push(info);
  31     }
  32
  33     group_hash
  34 }
  35
  36 fn mark_selections<F: Fn(DateTime<Local>, &BackupInfo) -> String> (
  37     mark: &mut HashSet<PathBuf>,
  38     list: &Vec<BackupInfo>,
  39     keep: usize,
  40     select_id: F,
  41 ){
  42     let mut hash = HashSet::new();
  43     for info in list {
  44         let local_time = info.backup_dir.backup_time().with_timezone(&Local);
  45         if hash.len() >= keep as usize { break; }
  46         let backup_id = info.backup_dir.relative_path();
  47         let sel_id: String = select_id(local_time, &info);
  48         if !hash.contains(&sel_id) {
  49             hash.insert(sel_id);
  50             //println!(" KEEP ID {} {}", backup_id, local_time.format("%c"));
  51             mark.insert(backup_id);
  52         }
  53     }
  54 }
  55
  56 fn list_groups(
  57     param: Value,
  58     _info: &ApiMethod,
  59     _rpcenv: &mut RpcEnvironment,
  60 ) -> Result<Value, Error> {
  61
  62     let store = param["store"].as_str().unwrap();
  63
  64     let datastore = DataStore::lookup_datastore(store)?;
  65
  66     let backup_list =  datastore.list_backups()?;
  67
  68     let group_hash = group_backups(backup_list);
  69
  70     let mut groups = vec![];
  71
  72     for (_group_id, mut list) in group_hash {
  73
  74         BackupInfo::sort_list(&mut list, false);
  75
  76         let info = &list[0];
  77         let group = info.backup_dir.group();
  78
  79         groups.push(json!({
  80             "backup-type": group.backup_type(),
  81             "backup-id": group.backup_id(),
  82             "last-backup": info.backup_dir.backup_time().timestamp(),
  83             "backup-count": list.len() as u64,
  84             "files": info.files,
  85         }));
  86     }
  87
  88     Ok(json!(groups))
  89 }
  90
  91 fn list_snapshot_files (
  92     param: Value,
  93     _info: &ApiMethod,
  94     _rpcenv: &mut RpcEnvironment,
  95 ) -> Result<Value, Error> {
  96
  97     let store = tools::required_string_param(&param, "store")?;
  98     let backup_type = tools::required_string_param(&param, "backup-type")?;
  99     let backup_id = tools::required_string_param(&param, "backup-id")?;
 100     let backup_time = tools::required_integer_param(&param, "backup-time")?;
 101
 102     let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
 103
 104     let datastore = DataStore::lookup_datastore(store)?;
 105
 106     let files = datastore.list_files(&snapshot)?;
 107
 108     Ok(json!(files))
 109 }
 110
 111 fn delete_snapshots (
 112     param: Value,
 113     _info: &ApiMethod,
 114     _rpcenv: &mut RpcEnvironment,
 115 ) -> Result<Value, Error> {
 116
 117     let store = tools::required_string_param(&param, "store")?;
 118     let backup_type = tools::required_string_param(&param, "backup-type")?;
 119     let backup_id = tools::required_string_param(&param, "backup-id")?;
 120     let backup_time = tools::required_integer_param(&param, "backup-time")?;
 121
 122     let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
 123
 124     let datastore = DataStore::lookup_datastore(store)?;
 125
 126     datastore.remove_backup_dir(&snapshot)?;
 127
 128     Ok(Value::Null)
 129 }
 130
 131 fn list_snapshots (
 132     param: Value,
 133     _info: &ApiMethod,
 134     _rpcenv: &mut RpcEnvironment,
 135 ) -> Result<Value, Error> {
 136
 137     let store = tools::required_string_param(&param, "store")?;
 138     let backup_type = tools::required_string_param(&param, "backup-type")?;
 139     let backup_id = tools::required_string_param(&param, "backup-id")?;
 140
 141     let group = BackupGroup::new(backup_type, backup_id);
 142
 143     let datastore = DataStore::lookup_datastore(store)?;
 144
 145     let backup_list = datastore.list_backups()?;
 146
 147     let mut group_hash = group_backups(backup_list);
 148
 149     let group_id = group.group_path().to_str().unwrap().to_owned();
 150
 151     let group_snapshots = match group_hash.get_mut(&group_id) {
 152         Some(data) => {
 153             // new backups first
 154             BackupInfo::sort_list(data, false);
 155             data
 156         }
 157         None => bail!("Backup group '{}' does not exists.", group_id),
 158     };
 159
 160     let mut snapshots = vec![];
 161
 162     for info in group_snapshots {
 163
 164         let group = info.backup_dir.group();
 165
 166         snapshots.push(json!({
 167             "backup-type": group.backup_type(),
 168             "backup-id": group.backup_id(),
 169             "backup-time": info.backup_dir.backup_time().timestamp(),
 170             "files": info.files,
 171         }));
 172     }
 173
 174     Ok(json!(snapshots))
 175 }
 176
 177 fn prune(
 178     param: Value,
 179     _info: &ApiMethod,
 180     _rpcenv: &mut RpcEnvironment,
 181 ) -> Result<Value, Error> {
 182
 183     let store = param["store"].as_str().unwrap();
 184
 185     let datastore = DataStore::lookup_datastore(store)?;
 186
 187     println!("Starting prune on store {}", store);
 188
 189     let backup_list =  datastore.list_backups()?;
 190
 191     let group_hash = group_backups(backup_list);
 192
 193     for (_group_id, mut list) in group_hash {
 194
 195         let mut mark = HashSet::new();
 196
 197         BackupInfo::sort_list(&mut list, false);
 198
 199         if let Some(keep_last) = param["keep-last"].as_u64() {
 200             list.iter().take(keep_last as usize).for_each(|info| {
 201                 mark.insert(info.backup_dir.relative_path());
 202             });
 203         }
 204
 205         if let Some(keep_daily) = param["keep-daily"].as_u64() {
 206             mark_selections(&mut mark, &list, keep_daily as usize, |local_time, _info| {
 207                 format!("{}/{}/{}", local_time.year(), local_time.month(), local_time.day())
 208             });
 209         }
 210
 211         if let Some(keep_weekly) = param["keep-weekly"].as_u64() {
 212             mark_selections(&mut mark, &list, keep_weekly as usize, |local_time, _info| {
 213                 format!("{}/{}", local_time.year(), local_time.iso_week().week())
 214             });
 215         }
 216
 217         if let Some(keep_monthly) = param["keep-monthly"].as_u64() {
 218             mark_selections(&mut mark, &list, keep_monthly as usize, |local_time, _info| {
 219                 format!("{}/{}", local_time.year(), local_time.month())
 220             });
 221         }
 222
 223         if let Some(keep_yearly) = param["keep-yearly"].as_u64() {
 224             mark_selections(&mut mark, &list, keep_yearly as usize, |local_time, _info| {
 225                 format!("{}/{}", local_time.year(), local_time.year())
 226             });
 227         }
 228
 229         let mut remove_list: Vec<BackupInfo> = list.into_iter()
 230             .filter(|info| !mark.contains(&info.backup_dir.relative_path())).collect();
 231
 232         BackupInfo::sort_list(&mut remove_list, true);
 233
 234         for info in remove_list {
 235             datastore.remove_backup_dir(&info.backup_dir)?;
 236         }
 237     }
 238
 239     Ok(json!(null))
 240 }
 241
 242 pub fn add_common_prune_prameters(schema: ObjectSchema) -> ObjectSchema  {
 243
 244     schema
 245         .optional(
 246             "keep-last",
 247             IntegerSchema::new("Number of backups to keep.")
 248                 .minimum(1)
 249         )
 250         .optional(
 251             "keep-daily",
 252             IntegerSchema::new("Number of daily backups to keep.")
 253                 .minimum(1)
 254         )
 255         .optional(
 256             "keep-weekly",
 257             IntegerSchema::new("Number of weekly backups to keep.")
 258                 .minimum(1)
 259         )
 260         .optional(
 261             "keep-monthly",
 262             IntegerSchema::new("Number of monthly backups to keep.")
 263                 .minimum(1)
 264         )
 265         .optional(
 266             "keep-yearly",
 267             IntegerSchema::new("Number of yearly backups to keep.")
 268                 .minimum(1)
 269         )
 270 }
 271
 272 fn api_method_prune() -> ApiMethod {
 273     ApiMethod::new(
 274         prune,
 275         add_common_prune_prameters(
 276             ObjectSchema::new("Prune the datastore.")
 277                 .required(
 278                     "store",
 279                     StringSchema::new("Datastore name.")
 280                 )
 281         )
 282     )
 283 }
 284
 285 // this is just a test for mutability/mutex handling  - will remove later
 286 fn start_garbage_collection(
 287     param: Value,
 288     _info: &ApiMethod,
 289     _rpcenv: &mut RpcEnvironment,
 290 ) -> Result<Value, Error> {
 291
 292     let store = param["store"].as_str().unwrap();
 293
 294     let datastore = DataStore::lookup_datastore(store)?;
 295
 296     println!("Starting garbage collection on store {}", store);
 297
 298     datastore.garbage_collection()?;
 299
 300     Ok(json!(null))
 301 }
 302
 303 pub fn api_method_start_garbage_collection() -> ApiMethod {
 304     ApiMethod::new(
 305         start_garbage_collection,
 306         ObjectSchema::new("Start garbage collection.")
 307             .required("store", StringSchema::new("Datastore name."))
 308     )
 309 }
 310
 311 fn garbage_collection_status(
 312     param: Value,
 313     _info: &ApiMethod,
 314     _rpcenv: &mut RpcEnvironment,
 315 ) -> Result<Value, Error> {
 316
 317     let store = param["store"].as_str().unwrap();
 318
 319     println!("Garbage collection status on store {}", store);
 320
 321     Ok(json!(null))
 322
 323 }
 324
 325 pub fn api_method_garbage_collection_status() -> ApiMethod {
 326     ApiMethod::new(
 327         garbage_collection_status,
 328         ObjectSchema::new("Garbage collection status.")
 329             .required("store", StringSchema::new("Datastore name."))
 330     )
 331 }
 332
 333 fn get_backup_list(
 334     param: Value,
 335     _info: &ApiMethod,
 336     _rpcenv: &mut RpcEnvironment,
 337 ) -> Result<Value, Error> {
 338
 339     //let config = datastore::config()?;
 340
 341     let store = param["store"].as_str().unwrap();
 342
 343     let datastore = DataStore::lookup_datastore(store)?;
 344
 345     let mut list = vec![];
 346
 347     for info in datastore.list_backups()? {
 348         list.push(json!({
 349             "backup-type": info.backup_dir.group().backup_type(),
 350             "backup-id": info.backup_dir.group().backup_id(),
 351             "backup-time": info.backup_dir.backup_time().timestamp(),
 352             "files": info.files,
 353         }));
 354     }
 355
 356     let result = json!(list);
 357
 358     Ok(result)
 359 }
 360
 361 fn get_datastore_list(
 362     _param: Value,
 363     _info: &ApiMethod,
 364     _rpcenv: &mut RpcEnvironment,
 365 ) -> Result<Value, Error> {
 366
 367     let config = datastore::config()?;
 368
 369     Ok(config.convert_to_array("store"))
 370 }
 371
 372
 373 pub fn router() -> Router {
 374
 375     let store_schema: Arc<Schema> = Arc::new(
 376         StringSchema::new("Datastore name.").into()
 377     );
 378
 379     let datastore_info = Router::new()
 380         .get(ApiMethod::new(
 381             |_,_,_| Ok(json!([
 382                 {"subdir": "backups" },
 383                 {"subdir": "catar" },
 384                 {"subdir": "gc" },
 385                 {"subdir": "groups" },
 386                 {"subdir": "snapshots" },
 387                 {"subdir": "status" },
 388                 {"subdir": "prune" },
 389            ])),
 390             ObjectSchema::new("Directory index.")
 391                 .required("store", store_schema.clone()))
 392         )
 393         .subdir(
 394             "backups",
 395             Router::new()
 396                 .get(ApiMethod::new(
 397                     get_backup_list,
 398                     ObjectSchema::new("List backups.")
 399                         .required("store", store_schema.clone()))))
 400         .subdir(
 401             "catar",
 402             Router::new()
 403                 .download(catar::api_method_download_catar())
 404                 .upload(catar::api_method_upload_catar()))
 405         .subdir(
 406             "test-upload",
 407             Router::new()
 408                 .upgrade(upload::api_method_upgrade_upload()))
 409         .subdir(
 410             "gc",
 411             Router::new()
 412                 .get(api_method_garbage_collection_status())
 413                 .post(api_method_start_garbage_collection()))
 414         .subdir(
 415             "files",
 416             Router::new()
 417                 .get(
 418                     ApiMethod::new(
 419                         list_snapshot_files,
 420                         ObjectSchema::new("List snapshot files.")
 421                             .required("store", store_schema.clone())
 422                             .required("backup-type", StringSchema::new("Backup type."))
 423                             .required("backup-id", StringSchema::new("Backup ID."))
 424                             .required("backup-time", IntegerSchema::new("Backup time (Unix epoch.)")
 425                                       .minimum(1547797308))
 426                     )
 427                 )
 428         )
 429         .subdir(
 430             "groups",
 431             Router::new()
 432                 .get(ApiMethod::new(
 433                     list_groups,
 434                     ObjectSchema::new("List backup groups.")
 435                         .required("store", store_schema.clone()))))
 436         .subdir(
 437             "snapshots",
 438             Router::new()
 439                 .get(
 440                     ApiMethod::new(
 441                         list_snapshots,
 442                         ObjectSchema::new("List backup groups.")
 443                             .required("store", store_schema.clone())
 444                             .required("backup-type", StringSchema::new("Backup type."))
 445                             .required("backup-id", StringSchema::new("Backup ID."))
 446                     )
 447                 )
 448                 .delete(
 449                     ApiMethod::new(
 450                         delete_snapshots,
 451                         ObjectSchema::new("Delete backup snapshot.")
 452                             .required("store", store_schema.clone())
 453                             .required("backup-type", StringSchema::new("Backup type."))
 454                             .required("backup-id", StringSchema::new("Backup ID."))
 455                             .required("backup-time", IntegerSchema::new("Backup time (Unix epoch.)")
 456                                       .minimum(1547797308))
 457                     )
 458                 )
 459         )
 460         .subdir(
 461             "prune",
 462             Router::new()
 463                 .post(api_method_prune()));
 464
 465
 466
 467     let route = Router::new()
 468         .get(ApiMethod::new(
 469             get_datastore_list,
 470             ObjectSchema::new("Directory index.")))
 471         .match_all("store", datastore_info);
 472
 473
 474
 475     route
 476 }