proxmox-backup-client/src/main.rs

   1 use std::collections::HashSet;
   2 use std::io::{self, Read, Write, Seek, SeekFrom};
   3 use std::path::{Path, PathBuf};
   4 use std::pin::Pin;
   5 use std::sync::{Arc, Mutex};
   6 use std::task::Context;
   7
   8 use anyhow::{bail, format_err, Error};
   9 use futures::stream::{StreamExt, TryStreamExt};
  10 use serde_json::{json, Value};
  11 use tokio::sync::mpsc;
  12 use tokio_stream::wrappers::ReceiverStream;
  13 use xdg::BaseDirectories;
  14
  15 use pathpatterns::{MatchEntry, MatchType, PatternFlag};
  16 use proxmox::tools::fs::{file_get_json, replace_file, CreateOptions, image_size};
  17 use proxmox_router::{ApiMethod, RpcEnvironment, cli::*};
  18 use proxmox_schema::api;
  19 use proxmox_time::{strftime_local, epoch_i64};
  20 use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
  21
  22 use pbs_api_types::{
  23     BACKUP_ID_SCHEMA, BACKUP_TIME_SCHEMA, BACKUP_TYPE_SCHEMA, Authid, CryptMode, GroupListItem,
  24     PruneListItem, SnapshotListItem, StorageStatus, Fingerprint, PruneOptions,
  25 };
  26 use pbs_client::{
  27     BACKUP_SOURCE_SCHEMA,
  28     BackupReader,
  29     BackupRepository,
  30     BackupSpecificationType,
  31     BackupStats,
  32     BackupWriter,
  33     ChunkStream,
  34     FixedChunkStream,
  35     HttpClient,
  36     PxarBackupStream,
  37     RemoteChunkReader,
  38     UploadOptions,
  39     delete_ticket_info,
  40     parse_backup_specification,
  41     view_task_result,
  42 };
  43 use pbs_client::catalog_shell::Shell;
  44 use pbs_client::tools::{
  45     complete_archive_name, complete_auth_id, complete_backup_group, complete_backup_snapshot,
  46     complete_backup_source, complete_chunk_size, complete_group_or_snapshot,
  47     complete_img_archive_name, complete_pxar_archive_name, complete_repository, connect,
  48     connect_rate_limited, extract_repository_from_value,
  49     key_source::{
  50         crypto_parameters, format_key_source, get_encryption_key_password, KEYFD_SCHEMA,
  51         KEYFILE_SCHEMA, MASTER_PUBKEY_FD_SCHEMA, MASTER_PUBKEY_FILE_SCHEMA,
  52     },
  53     CHUNK_SIZE_SCHEMA, REPO_URL_SCHEMA,
  54 };
  55 use pbs_config::key_config::{KeyConfig, decrypt_key, rsa_encrypt_key_config};
  56 use pbs_datastore::CATALOG_NAME;
  57 use pbs_datastore::backup_info::{BackupDir, BackupGroup};
  58 use pbs_datastore::catalog::{BackupCatalogWriter, CatalogReader, CatalogWriter};
  59 use pbs_datastore::chunk_store::verify_chunk_size;
  60 use pbs_datastore::dynamic_index::{BufferedDynamicReader, DynamicIndexReader};
  61 use pbs_datastore::fixed_index::FixedIndexReader;
  62 use pbs_datastore::index::IndexFile;
  63 use pbs_datastore::manifest::{
  64     ENCRYPTED_KEY_BLOB_NAME, MANIFEST_BLOB_NAME, ArchiveType, BackupManifest, archive_type,
  65 };
  66 use pbs_datastore::read_chunk::AsyncReadChunk;
  67 use pbs_tools::sync::StdChannelWriter;
  68 use pbs_tools::tokio::TokioWriterAdapter;
  69 use pbs_tools::json;
  70 use pbs_tools::crypt_config::CryptConfig;
  71
  72 mod benchmark;
  73 pub use benchmark::*;
  74 mod mount;
  75 pub use mount::*;
  76 mod task;
  77 pub use task::*;
  78 mod catalog;
  79 pub use catalog::*;
  80 mod snapshot;
  81 pub use snapshot::*;
  82 pub mod key;
  83
  84 fn record_repository(repo: &BackupRepository) {
  85
  86     let base = match BaseDirectories::with_prefix("proxmox-backup") {
  87         Ok(v) => v,
  88         _ => return,
  89     };
  90
  91     // usually $HOME/.cache/proxmox-backup/repo-list
  92     let path = match base.place_cache_file("repo-list") {
  93         Ok(v) => v,
  94         _ => return,
  95     };
  96
  97     let mut data = file_get_json(&path, None).unwrap_or_else(|_| json!({}));
  98
  99     let repo = repo.to_string();
 100
 101     data[&repo] = json!{ data[&repo].as_i64().unwrap_or(0) + 1 };
 102
 103     let mut map = serde_json::map::Map::new();
 104
 105     loop {
 106         let mut max_used = 0;
 107         let mut max_repo = None;
 108         for (repo, count) in data.as_object().unwrap() {
 109             if map.contains_key(repo) { continue; }
 110             if let Some(count) = count.as_i64() {
 111                 if count > max_used {
 112                     max_used = count;
 113                     max_repo = Some(repo);
 114                 }
 115             }
 116         }
 117         if let Some(repo) = max_repo {
 118             map.insert(repo.to_owned(), json!(max_used));
 119         } else {
 120             break;
 121         }
 122         if map.len() > 10 { // store max. 10 repos
 123             break;
 124         }
 125     }
 126
 127     let new_data = json!(map);
 128
 129     let _ = replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new(), false);
 130 }
 131
 132 async fn api_datastore_list_snapshots(
 133     client: &HttpClient,
 134     store: &str,
 135     group: Option<BackupGroup>,
 136 ) -> Result<Value, Error> {
 137
 138     let path = format!("api2/json/admin/datastore/{}/snapshots", store);
 139
 140     let mut args = json!({});
 141     if let Some(group) = group {
 142         args["backup-type"] = group.backup_type().into();
 143         args["backup-id"] = group.backup_id().into();
 144     }
 145
 146     let mut result = client.get(&path, Some(args)).await?;
 147
 148     Ok(result["data"].take())
 149 }
 150
 151 pub async fn api_datastore_latest_snapshot(
 152     client: &HttpClient,
 153     store: &str,
 154     group: BackupGroup,
 155 ) -> Result<(String, String, i64), Error> {
 156
 157     let list = api_datastore_list_snapshots(client, store, Some(group.clone())).await?;
 158     let mut list: Vec<SnapshotListItem> = serde_json::from_value(list)?;
 159
 160     if list.is_empty() {
 161         bail!("backup group {:?} does not contain any snapshots.", group.group_path());
 162     }
 163
 164     list.sort_unstable_by(|a, b| b.backup_time.cmp(&a.backup_time));
 165
 166     let backup_time = list[0].backup_time;
 167
 168     Ok((group.backup_type().to_owned(), group.backup_id().to_owned(), backup_time))
 169 }
 170
 171 async fn backup_directory<P: AsRef<Path>>(
 172     client: &BackupWriter,
 173     dir_path: P,
 174     archive_name: &str,
 175     chunk_size: Option<usize>,
 176     catalog: Arc<Mutex<CatalogWriter<TokioWriterAdapter<StdChannelWriter>>>>,
 177     pxar_create_options: pbs_client::pxar::PxarCreateOptions,
 178     upload_options: UploadOptions,
 179 ) -> Result<BackupStats, Error> {
 180
 181     let pxar_stream = PxarBackupStream::open(
 182         dir_path.as_ref(),
 183         catalog,
 184         pxar_create_options,
 185     )?;
 186     let mut chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
 187
 188     let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
 189
 190     let stream = ReceiverStream::new(rx)
 191         .map_err(Error::from);
 192
 193     // spawn chunker inside a separate task so that it can run parallel
 194     tokio::spawn(async move {
 195         while let Some(v) = chunk_stream.next().await {
 196             let _ = tx.send(v).await;
 197         }
 198     });
 199
 200     if upload_options.fixed_size.is_some() {
 201         bail!("cannot backup directory with fixed chunk size!");
 202     }
 203
 204     let stats = client
 205         .upload_stream(archive_name, stream, upload_options)
 206         .await?;
 207
 208     Ok(stats)
 209 }
 210
 211 async fn backup_image<P: AsRef<Path>>(
 212     client: &BackupWriter,
 213     image_path: P,
 214     archive_name: &str,
 215     chunk_size: Option<usize>,
 216     upload_options: UploadOptions,
 217 ) -> Result<BackupStats, Error> {
 218
 219     let path = image_path.as_ref().to_owned();
 220
 221     let file = tokio::fs::File::open(path).await?;
 222
 223     let stream = tokio_util::codec::FramedRead::new(file, tokio_util::codec::BytesCodec::new())
 224         .map_err(Error::from);
 225
 226     let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
 227
 228     if upload_options.fixed_size.is_none() {
 229         bail!("cannot backup image with dynamic chunk size!");
 230     }
 231
 232     let stats = client
 233         .upload_stream(archive_name, stream, upload_options)
 234         .await?;
 235
 236     Ok(stats)
 237 }
 238
 239 #[api(
 240    input: {
 241         properties: {
 242             repository: {
 243                 schema: REPO_URL_SCHEMA,
 244                 optional: true,
 245             },
 246             "output-format": {
 247                 schema: OUTPUT_FORMAT,
 248                 optional: true,
 249             },
 250         }
 251    }
 252 )]
 253 /// List backup groups.
 254 async fn list_backup_groups(param: Value) -> Result<Value, Error> {
 255
 256     let output_format = get_output_format(&param);
 257
 258     let repo = extract_repository_from_value(&param)?;
 259
 260     let client = connect(&repo)?;
 261
 262     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 263
 264     let mut result = client.get(&path, None).await?;
 265
 266     record_repository(&repo);
 267
 268     let render_group_path = |_v: &Value, record: &Value| -> Result<String, Error> {
 269         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 270         let group = BackupGroup::new(item.backup_type, item.backup_id);
 271         Ok(group.group_path().to_str().unwrap().to_owned())
 272     };
 273
 274     let render_last_backup = |_v: &Value, record: &Value| -> Result<String, Error> {
 275         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 276         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.last_backup)?;
 277         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
 278     };
 279
 280     let render_files = |_v: &Value, record: &Value| -> Result<String, Error> {
 281         let item: GroupListItem = serde_json::from_value(record.to_owned())?;
 282         Ok(pbs_tools::format::render_backup_file_list(&item.files))
 283     };
 284
 285     let options = default_table_format_options()
 286         .sortby("backup-type", false)
 287         .sortby("backup-id", false)
 288         .column(ColumnConfig::new("backup-id").renderer(render_group_path).header("group"))
 289         .column(
 290             ColumnConfig::new("last-backup")
 291                 .renderer(render_last_backup)
 292                 .header("last snapshot")
 293                 .right_align(false)
 294         )
 295         .column(ColumnConfig::new("backup-count"))
 296         .column(ColumnConfig::new("files").renderer(render_files));
 297
 298     let mut data: Value = result["data"].take();
 299
 300     let return_type = &pbs_api_types::ADMIN_DATASTORE_LIST_GROUPS_RETURN_TYPE;
 301
 302     format_and_print_result_full(&mut data, return_type, &output_format, &options);
 303
 304     Ok(Value::Null)
 305 }
 306
 307 #[api(
 308    input: {
 309         properties: {
 310             repository: {
 311                 schema: REPO_URL_SCHEMA,
 312                 optional: true,
 313             },
 314             group: {
 315                 type: String,
 316                 description: "Backup group.",
 317             },
 318             "new-owner": {
 319                 type: Authid,
 320             },
 321         }
 322    }
 323 )]
 324 /// Change owner of a backup group
 325 async fn change_backup_owner(group: String, mut param: Value) -> Result<(), Error> {
 326
 327     let repo = extract_repository_from_value(&param)?;
 328
 329     let mut client = connect(&repo)?;
 330
 331     param.as_object_mut().unwrap().remove("repository");
 332
 333     let group: BackupGroup = group.parse()?;
 334
 335     param["backup-type"] = group.backup_type().into();
 336     param["backup-id"] = group.backup_id().into();
 337
 338     let path = format!("api2/json/admin/datastore/{}/change-owner", repo.store());
 339     client.post(&path, Some(param)).await?;
 340
 341     record_repository(&repo);
 342
 343     Ok(())
 344 }
 345
 346 #[api(
 347    input: {
 348         properties: {
 349             repository: {
 350                 schema: REPO_URL_SCHEMA,
 351                 optional: true,
 352             },
 353         }
 354    }
 355 )]
 356 /// Try to login. If successful, store ticket.
 357 async fn api_login(param: Value) -> Result<Value, Error> {
 358
 359     let repo = extract_repository_from_value(&param)?;
 360
 361     let client = connect(&repo)?;
 362     client.login().await?;
 363
 364     record_repository(&repo);
 365
 366     Ok(Value::Null)
 367 }
 368
 369 #[api(
 370    input: {
 371         properties: {
 372             repository: {
 373                 schema: REPO_URL_SCHEMA,
 374                 optional: true,
 375             },
 376         }
 377    }
 378 )]
 379 /// Logout (delete stored ticket).
 380 fn api_logout(param: Value) -> Result<Value, Error> {
 381
 382     let repo = extract_repository_from_value(&param)?;
 383
 384     delete_ticket_info("proxmox-backup", repo.host(), repo.user())?;
 385
 386     Ok(Value::Null)
 387 }
 388
 389 #[api(
 390    input: {
 391         properties: {
 392             repository: {
 393                 schema: REPO_URL_SCHEMA,
 394                 optional: true,
 395             },
 396             "output-format": {
 397                 schema: OUTPUT_FORMAT,
 398                 optional: true,
 399             },
 400         }
 401    }
 402 )]
 403 /// Show client and optional server version
 404 async fn api_version(param: Value) -> Result<(), Error> {
 405
 406     let output_format = get_output_format(&param);
 407
 408     let mut version_info = json!({
 409         "client": {
 410             "version": pbs_buildcfg::PROXMOX_PKG_VERSION,
 411             "release": pbs_buildcfg::PROXMOX_PKG_RELEASE,
 412             "repoid": pbs_buildcfg::PROXMOX_PKG_REPOID,
 413         }
 414     });
 415
 416     let repo = extract_repository_from_value(&param);
 417     if let Ok(repo) = repo {
 418         let client = connect(&repo)?;
 419
 420         match client.get("api2/json/version", None).await {
 421             Ok(mut result) => version_info["server"] = result["data"].take(),
 422             Err(e) => eprintln!("could not connect to server - {}", e),
 423         }
 424     }
 425     if output_format == "text" {
 426         println!(
 427             "client version: {}.{}",
 428             pbs_buildcfg::PROXMOX_PKG_VERSION,
 429             pbs_buildcfg::PROXMOX_PKG_RELEASE,
 430         );
 431         if let Some(server) = version_info["server"].as_object() {
 432             let server_version = server["version"].as_str().unwrap();
 433             let server_release = server["release"].as_str().unwrap();
 434             println!("server version: {}.{}", server_version, server_release);
 435         }
 436     } else {
 437         format_and_print_result(&version_info, &output_format);
 438     }
 439
 440     Ok(())
 441 }
 442
 443 #[api(
 444     input: {
 445         properties: {
 446             repository: {
 447                 schema: REPO_URL_SCHEMA,
 448                 optional: true,
 449             },
 450             "output-format": {
 451                 schema: OUTPUT_FORMAT,
 452                 optional: true,
 453             },
 454         },
 455     },
 456 )]
 457 /// Start garbage collection for a specific repository.
 458 async fn start_garbage_collection(param: Value) -> Result<Value, Error> {
 459
 460     let repo = extract_repository_from_value(&param)?;
 461
 462     let output_format = get_output_format(&param);
 463
 464     let mut client = connect(&repo)?;
 465
 466     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 467
 468     let result = client.post(&path, None).await?;
 469
 470     record_repository(&repo);
 471
 472     view_task_result(&mut client, result, &output_format).await?;
 473
 474     Ok(Value::Null)
 475 }
 476
 477 struct CatalogUploadResult {
 478     catalog_writer: Arc<Mutex<CatalogWriter<TokioWriterAdapter<StdChannelWriter>>>>,
 479     result: tokio::sync::oneshot::Receiver<Result<BackupStats, Error>>,
 480 }
 481
 482 fn spawn_catalog_upload(
 483     client: Arc<BackupWriter>,
 484     encrypt: bool,
 485 ) -> Result<CatalogUploadResult, Error> {
 486     let (catalog_tx, catalog_rx) = std::sync::mpsc::sync_channel(10); // allow to buffer 10 writes
 487     let catalog_stream = pbs_tools::blocking::StdChannelStream(catalog_rx);
 488     let catalog_chunk_size = 512*1024;
 489     let catalog_chunk_stream = ChunkStream::new(catalog_stream, Some(catalog_chunk_size));
 490
 491     let catalog_writer = Arc::new(Mutex::new(CatalogWriter::new(TokioWriterAdapter::new(StdChannelWriter::new(catalog_tx)))?));
 492
 493     let (catalog_result_tx, catalog_result_rx) = tokio::sync::oneshot::channel();
 494
 495     let upload_options = UploadOptions {
 496         encrypt,
 497         compress: true,
 498         ..UploadOptions::default()
 499     };
 500
 501     tokio::spawn(async move {
 502         let catalog_upload_result = client
 503             .upload_stream(CATALOG_NAME, catalog_chunk_stream, upload_options)
 504             .await;
 505
 506         if let Err(ref err) = catalog_upload_result {
 507             eprintln!("catalog upload error - {}", err);
 508             client.cancel();
 509         }
 510
 511         let _ = catalog_result_tx.send(catalog_upload_result);
 512     });
 513
 514     Ok(CatalogUploadResult { catalog_writer, result: catalog_result_rx })
 515 }
 516
 517 #[api(
 518    input: {
 519        properties: {
 520            backupspec: {
 521                type: Array,
 522                description: "List of backup source specifications ([<label.ext>:<path>] ...)",
 523                items: {
 524                    schema: BACKUP_SOURCE_SCHEMA,
 525                }
 526            },
 527            repository: {
 528                schema: REPO_URL_SCHEMA,
 529                optional: true,
 530            },
 531            "include-dev": {
 532                description: "Include mountpoints with same st_dev number (see ``man fstat``) as specified files.",
 533                optional: true,
 534                items: {
 535                    type: String,
 536                    description: "Path to file.",
 537                }
 538            },
 539            "all-file-systems": {
 540                type: Boolean,
 541                description: "Include all mounted subdirectories.",
 542                optional: true,
 543            },
 544            keyfile: {
 545                schema: KEYFILE_SCHEMA,
 546                optional: true,
 547            },
 548            "keyfd": {
 549                schema: KEYFD_SCHEMA,
 550                optional: true,
 551            },
 552            "master-pubkey-file": {
 553                schema: MASTER_PUBKEY_FILE_SCHEMA,
 554                optional: true,
 555            },
 556            "master-pubkey-fd": {
 557                schema: MASTER_PUBKEY_FD_SCHEMA,
 558                optional: true,
 559            },
 560            "crypt-mode": {
 561                type: CryptMode,
 562                optional: true,
 563            },
 564            "skip-lost-and-found": {
 565                type: Boolean,
 566                description: "Skip lost+found directory.",
 567                optional: true,
 568            },
 569            "backup-type": {
 570                schema: BACKUP_TYPE_SCHEMA,
 571                optional: true,
 572            },
 573            "backup-id": {
 574                schema: BACKUP_ID_SCHEMA,
 575                optional: true,
 576            },
 577            "backup-time": {
 578                schema: BACKUP_TIME_SCHEMA,
 579                optional: true,
 580            },
 581            "chunk-size": {
 582                schema: CHUNK_SIZE_SCHEMA,
 583                optional: true,
 584            },
 585            rate: {
 586                type: u64,
 587                description: "Rate limit for TBF in bytes/second.",
 588                optional: true,
 589                minimum: 1,
 590            },
 591            burst: {
 592                type: u64,
 593                description: "Size of the TBF bucket, in bytes.",
 594                optional: true,
 595                minimum: 1,
 596            },
 597            "exclude": {
 598                type: Array,
 599                description: "List of paths or patterns for matching files to exclude.",
 600                optional: true,
 601                items: {
 602                    type: String,
 603                    description: "Path or match pattern.",
 604                 }
 605            },
 606            "entries-max": {
 607                type: Integer,
 608                description: "Max number of entries to hold in memory.",
 609                optional: true,
 610                default: pbs_client::pxar::ENCODER_MAX_ENTRIES as isize,
 611            },
 612            "verbose": {
 613                type: Boolean,
 614                description: "Verbose output.",
 615                optional: true,
 616            },
 617        }
 618    }
 619 )]
 620 /// Create (host) backup.
 621 async fn create_backup(
 622     param: Value,
 623     _info: &ApiMethod,
 624     _rpcenv: &mut dyn RpcEnvironment,
 625 ) -> Result<Value, Error> {
 626
 627     let repo = extract_repository_from_value(&param)?;
 628
 629     let backupspec_list = json::required_array_param(&param, "backupspec")?;
 630
 631     let all_file_systems = param["all-file-systems"].as_bool().unwrap_or(false);
 632
 633     let skip_lost_and_found = param["skip-lost-and-found"].as_bool().unwrap_or(false);
 634
 635     let verbose = param["verbose"].as_bool().unwrap_or(false);
 636
 637     let backup_time_opt = param["backup-time"].as_i64();
 638
 639     let chunk_size_opt = param["chunk-size"].as_u64().map(|v| (v*1024) as usize);
 640
 641     if let Some(size) = chunk_size_opt {
 642         verify_chunk_size(size)?;
 643     }
 644
 645     let rate_limit = param["rate"].as_u64();
 646     let bucket_size = param["burst"].as_u64();
 647
 648     let crypto = crypto_parameters(&param)?;
 649
 650     let backup_id = param["backup-id"].as_str().unwrap_or(&proxmox::tools::nodename());
 651
 652     let backup_type = param["backup-type"].as_str().unwrap_or("host");
 653
 654     let include_dev = param["include-dev"].as_array();
 655
 656     let entries_max = param["entries-max"].as_u64()
 657         .unwrap_or(pbs_client::pxar::ENCODER_MAX_ENTRIES as u64);
 658
 659     let empty = Vec::new();
 660     let exclude_args = param["exclude"].as_array().unwrap_or(&empty);
 661
 662     let mut pattern_list = Vec::with_capacity(exclude_args.len());
 663     for entry in exclude_args {
 664         let entry = entry.as_str().ok_or_else(|| format_err!("Invalid pattern string slice"))?;
 665         pattern_list.push(
 666             MatchEntry::parse_pattern(entry, PatternFlag::PATH_NAME, MatchType::Exclude)
 667                 .map_err(|err| format_err!("invalid exclude pattern entry: {}", err))?
 668         );
 669     }
 670
 671     let mut devices = if all_file_systems { None } else { Some(HashSet::new()) };
 672
 673     if let Some(include_dev) = include_dev {
 674         if all_file_systems {
 675             bail!("option 'all-file-systems' conflicts with option 'include-dev'");
 676         }
 677
 678         let mut set = HashSet::new();
 679         for path in include_dev {
 680             let path = path.as_str().unwrap();
 681             let stat = nix::sys::stat::stat(path)
 682                 .map_err(|err| format_err!("fstat {:?} failed - {}", path, err))?;
 683             set.insert(stat.st_dev);
 684         }
 685         devices = Some(set);
 686     }
 687
 688     let mut upload_list = vec![];
 689     let mut target_set = HashSet::new();
 690
 691     for backupspec in backupspec_list {
 692         let spec = parse_backup_specification(backupspec.as_str().unwrap())?;
 693         let filename = &spec.config_string;
 694         let target = &spec.archive_name;
 695
 696         if target_set.contains(target) {
 697             bail!("got target twice: '{}'", target);
 698         }
 699         target_set.insert(target.to_string());
 700
 701         use std::os::unix::fs::FileTypeExt;
 702
 703         let metadata = std::fs::metadata(filename)
 704             .map_err(|err| format_err!("unable to access '{}' - {}", filename, err))?;
 705         let file_type = metadata.file_type();
 706
 707         match spec.spec_type {
 708             BackupSpecificationType::PXAR => {
 709                 if !file_type.is_dir() {
 710                     bail!("got unexpected file type (expected directory)");
 711                 }
 712                 upload_list.push((BackupSpecificationType::PXAR, filename.to_owned(), format!("{}.didx", target), 0));
 713             }
 714             BackupSpecificationType::IMAGE => {
 715                 if !(file_type.is_file() || file_type.is_block_device()) {
 716                     bail!("got unexpected file type (expected file or block device)");
 717                 }
 718
 719                 let size = image_size(&PathBuf::from(filename))?;
 720
 721                 if size == 0 { bail!("got zero-sized file '{}'", filename); }
 722
 723                 upload_list.push((BackupSpecificationType::IMAGE, filename.to_owned(), format!("{}.fidx", target), size));
 724             }
 725             BackupSpecificationType::CONFIG => {
 726                 if !file_type.is_file() {
 727                     bail!("got unexpected file type (expected regular file)");
 728                 }
 729                 upload_list.push((BackupSpecificationType::CONFIG, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 730             }
 731             BackupSpecificationType::LOGFILE => {
 732                 if !file_type.is_file() {
 733                     bail!("got unexpected file type (expected regular file)");
 734                 }
 735                 upload_list.push((BackupSpecificationType::LOGFILE, filename.to_owned(), format!("{}.blob", target), metadata.len()));
 736             }
 737         }
 738     }
 739
 740     let backup_time = backup_time_opt.unwrap_or_else(epoch_i64);
 741
 742     let client = connect_rate_limited(&repo, rate_limit, bucket_size)?;
 743     record_repository(&repo);
 744
 745     println!("Starting backup: {}/{}/{}", backup_type, backup_id, BackupDir::backup_time_to_string(backup_time)?);
 746
 747     println!("Client name: {}", proxmox::tools::nodename());
 748
 749     let start_time = std::time::Instant::now();
 750
 751     println!("Starting backup protocol: {}", strftime_local("%c", epoch_i64())?);
 752
 753     let (crypt_config, rsa_encrypted_key) = match crypto.enc_key {
 754         None => (None, None),
 755         Some(key_with_source) => {
 756             println!(
 757                 "{}",
 758                 format_key_source(&key_with_source.source, "encryption")
 759             );
 760
 761             let (key, created, fingerprint) =
 762                 decrypt_key(&key_with_source.key, &get_encryption_key_password)?;
 763             println!("Encryption key fingerprint: {}", fingerprint);
 764
 765             let crypt_config = CryptConfig::new(key)?;
 766
 767             match crypto.master_pubkey {
 768                 Some(pem_with_source) => {
 769                     println!("{}", format_key_source(&pem_with_source.source, "master"));
 770
 771                     let rsa = openssl::rsa::Rsa::public_key_from_pem(&pem_with_source.key)?;
 772
 773                     let mut key_config = KeyConfig::without_password(key)?;
 774                     key_config.created = created; // keep original value
 775
 776                     let enc_key = rsa_encrypt_key_config(rsa, &key_config)?;
 777
 778                     (Some(Arc::new(crypt_config)), Some(enc_key))
 779                 },
 780                 _ => (Some(Arc::new(crypt_config)), None),
 781             }
 782         }
 783     };
 784
 785     let client = BackupWriter::start(
 786         client,
 787         crypt_config.clone(),
 788         repo.store(),
 789         backup_type,
 790         &backup_id,
 791         backup_time,
 792         verbose,
 793         false
 794     ).await?;
 795
 796     let download_previous_manifest = match client.previous_backup_time().await {
 797         Ok(Some(backup_time)) => {
 798             println!(
 799                 "Downloading previous manifest ({})",
 800                 strftime_local("%c", backup_time)?
 801             );
 802             true
 803         }
 804         Ok(None) => {
 805             println!("No previous manifest available.");
 806             false
 807         }
 808         Err(_) => {
 809             // Fallback for outdated server, TODO remove/bubble up with 2.0
 810             true
 811         }
 812     };
 813
 814     let previous_manifest = if download_previous_manifest {
 815         match client.download_previous_manifest().await {
 816             Ok(previous_manifest) => {
 817                 match previous_manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref)) {
 818                     Ok(()) => Some(Arc::new(previous_manifest)),
 819                     Err(err) => {
 820                         println!("Couldn't re-use previous manifest - {}", err);
 821                         None
 822                     }
 823                 }
 824             }
 825             Err(err) => {
 826                 println!("Couldn't download previous manifest - {}", err);
 827                 None
 828             }
 829         }
 830     } else {
 831         None
 832     };
 833
 834     let snapshot = BackupDir::new(backup_type, backup_id, backup_time)?;
 835     let mut manifest = BackupManifest::new(snapshot);
 836
 837     let mut catalog = None;
 838     let mut catalog_result_rx = None;
 839
 840     for (backup_type, filename, target, size) in upload_list {
 841         match backup_type {
 842             BackupSpecificationType::CONFIG => {
 843                 let upload_options = UploadOptions {
 844                     compress: true,
 845                     encrypt: crypto.mode == CryptMode::Encrypt,
 846                     ..UploadOptions::default()
 847                 };
 848
 849                 println!("Upload config file '{}' to '{}' as {}", filename, repo, target);
 850                 let stats = client
 851                     .upload_blob_from_file(&filename, &target, upload_options)
 852                     .await?;
 853                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 854             }
 855             BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
 856                 let upload_options = UploadOptions {
 857                     compress: true,
 858                     encrypt: crypto.mode == CryptMode::Encrypt,
 859                     ..UploadOptions::default()
 860                 };
 861
 862                 println!("Upload log file '{}' to '{}' as {}", filename, repo, target);
 863                 let stats = client
 864                     .upload_blob_from_file(&filename, &target, upload_options)
 865                     .await?;
 866                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 867             }
 868             BackupSpecificationType::PXAR => {
 869                 // start catalog upload on first use
 870                 if catalog.is_none() {
 871                     let catalog_upload_res = spawn_catalog_upload(client.clone(), crypto.mode == CryptMode::Encrypt)?;
 872                     catalog = Some(catalog_upload_res.catalog_writer);
 873                     catalog_result_rx = Some(catalog_upload_res.result);
 874                 }
 875                 let catalog = catalog.as_ref().unwrap();
 876
 877                 println!("Upload directory '{}' to '{}' as {}", filename, repo, target);
 878                 catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
 879
 880                 let pxar_options = pbs_client::pxar::PxarCreateOptions {
 881                     device_set: devices.clone(),
 882                     patterns: pattern_list.clone(),
 883                     entries_max: entries_max as usize,
 884                     skip_lost_and_found,
 885                     verbose,
 886                 };
 887
 888                 let upload_options = UploadOptions {
 889                     previous_manifest: previous_manifest.clone(),
 890                     compress: true,
 891                     encrypt: crypto.mode == CryptMode::Encrypt,
 892                     ..UploadOptions::default()
 893                 };
 894
 895                 let stats = backup_directory(
 896                     &client,
 897                     &filename,
 898                     &target,
 899                     chunk_size_opt,
 900                     catalog.clone(),
 901                     pxar_options,
 902                     upload_options,
 903                 ).await?;
 904                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 905                 catalog.lock().unwrap().end_directory()?;
 906             }
 907             BackupSpecificationType::IMAGE => {
 908                 println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
 909
 910                 let upload_options = UploadOptions {
 911                     previous_manifest: previous_manifest.clone(),
 912                     fixed_size: Some(size),
 913                     compress: true,
 914                     encrypt: crypto.mode == CryptMode::Encrypt,
 915                 };
 916
 917                 let stats = backup_image(
 918                     &client,
 919                     &filename,
 920                     &target,
 921                     chunk_size_opt,
 922                     upload_options,
 923                 ).await?;
 924                 manifest.add_file(target, stats.size, stats.csum, crypto.mode)?;
 925             }
 926         }
 927     }
 928
 929     // finalize and upload catalog
 930     if let Some(catalog) = catalog {
 931         let mutex = Arc::try_unwrap(catalog)
 932             .map_err(|_| format_err!("unable to get catalog (still used)"))?;
 933         let mut catalog = mutex.into_inner().unwrap();
 934
 935         catalog.finish()?;
 936
 937         drop(catalog); // close upload stream
 938
 939         if let Some(catalog_result_rx) = catalog_result_rx {
 940             let stats = catalog_result_rx.await??;
 941             manifest.add_file(CATALOG_NAME.to_owned(), stats.size, stats.csum, crypto.mode)?;
 942         }
 943     }
 944
 945     if let Some(rsa_encrypted_key) = rsa_encrypted_key {
 946         let target = ENCRYPTED_KEY_BLOB_NAME;
 947         println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
 948         let options = UploadOptions { compress: false, encrypt: false, ..UploadOptions::default() };
 949         let stats = client
 950             .upload_blob_from_data(rsa_encrypted_key, target, options)
 951             .await?;
 952         manifest.add_file(target.to_string(), stats.size, stats.csum, crypto.mode)?;
 953
 954     }
 955     // create manifest (index.json)
 956     // manifests are never encrypted, but include a signature
 957     let manifest = manifest.to_string(crypt_config.as_ref().map(Arc::as_ref))
 958         .map_err(|err| format_err!("unable to format manifest - {}", err))?;
 959
 960
 961     if verbose { println!("Upload index.json to '{}'", repo) };
 962     let options = UploadOptions { compress: true, encrypt: false, ..UploadOptions::default() };
 963     client
 964         .upload_blob_from_data(manifest.into_bytes(), MANIFEST_BLOB_NAME, options)
 965         .await?;
 966
 967     client.finish().await?;
 968
 969     let end_time = std::time::Instant::now();
 970     let elapsed = end_time.duration_since(start_time);
 971     println!("Duration: {:.2}s", elapsed.as_secs_f64());
 972
 973     println!("End Time: {}", strftime_local("%c", epoch_i64())?);
 974
 975     Ok(Value::Null)
 976 }
 977
 978 async fn dump_image<W: Write>(
 979     client: Arc<BackupReader>,
 980     crypt_config: Option<Arc<CryptConfig>>,
 981     crypt_mode: CryptMode,
 982     index: FixedIndexReader,
 983     mut writer: W,
 984     verbose: bool,
 985 ) -> Result<(), Error> {
 986
 987     let most_used = index.find_most_used_chunks(8);
 988
 989     let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, crypt_mode, most_used);
 990
 991     // Note: we avoid using BufferedFixedReader, because that add an additional buffer/copy
 992     // and thus slows down reading. Instead, directly use RemoteChunkReader
 993     let mut per = 0;
 994     let mut bytes = 0;
 995     let start_time = std::time::Instant::now();
 996
 997     for pos in 0..index.index_count() {
 998         let digest = index.index_digest(pos).unwrap();
 999         let raw_data = chunk_reader.read_chunk(&digest).await?;
1000         writer.write_all(&raw_data)?;
1001         bytes += raw_data.len();
1002         if verbose {
1003             let next_per = ((pos+1)*100)/index.index_count();
1004             if per != next_per {
1005                 eprintln!("progress {}% (read {} bytes, duration {} sec)",
1006                           next_per, bytes, start_time.elapsed().as_secs());
1007                 per = next_per;
1008             }
1009         }
1010     }
1011
1012     let end_time = std::time::Instant::now();
1013     let elapsed = end_time.duration_since(start_time);
1014     eprintln!("restore image complete (bytes={}, duration={:.2}s, speed={:.2}MB/s)",
1015               bytes,
1016               elapsed.as_secs_f64(),
1017               bytes as f64/(1024.0*1024.0*elapsed.as_secs_f64())
1018     );
1019
1020
1021     Ok(())
1022 }
1023
1024 fn parse_archive_type(name: &str) -> (String, ArchiveType) {
1025     if name.ends_with(".didx") || name.ends_with(".fidx") || name.ends_with(".blob") {
1026         (name.into(), archive_type(name).unwrap())
1027     } else if name.ends_with(".pxar") {
1028         (format!("{}.didx", name), ArchiveType::DynamicIndex)
1029     } else if name.ends_with(".img") {
1030         (format!("{}.fidx", name), ArchiveType::FixedIndex)
1031     } else {
1032         (format!("{}.blob", name), ArchiveType::Blob)
1033     }
1034 }
1035
1036 #[api(
1037    input: {
1038        properties: {
1039            repository: {
1040                schema: REPO_URL_SCHEMA,
1041                optional: true,
1042            },
1043            snapshot: {
1044                type: String,
1045                description: "Group/Snapshot path.",
1046            },
1047            "archive-name": {
1048                description: "Backup archive name.",
1049                type: String,
1050            },
1051            target: {
1052                type: String,
1053                description: r###"Target directory path. Use '-' to write to standard output.
1054
1055 We do not extract '.pxar' archives when writing to standard output.
1056
1057 "###
1058            },
1059            "allow-existing-dirs": {
1060                type: Boolean,
1061                description: "Do not fail if directories already exists.",
1062                optional: true,
1063            },
1064            keyfile: {
1065                schema: KEYFILE_SCHEMA,
1066                optional: true,
1067            },
1068            "keyfd": {
1069                schema: KEYFD_SCHEMA,
1070                optional: true,
1071            },
1072            "crypt-mode": {
1073                type: CryptMode,
1074                optional: true,
1075            },
1076        }
1077    }
1078 )]
1079 /// Restore backup repository.
1080 async fn restore(param: Value) -> Result<Value, Error> {
1081     let repo = extract_repository_from_value(&param)?;
1082
1083     let verbose = param["verbose"].as_bool().unwrap_or(false);
1084
1085     let allow_existing_dirs = param["allow-existing-dirs"].as_bool().unwrap_or(false);
1086
1087     let archive_name = json::required_string_param(&param, "archive-name")?;
1088
1089     let client = connect(&repo)?;
1090
1091     record_repository(&repo);
1092
1093     let path = json::required_string_param(&param, "snapshot")?;
1094
1095     let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
1096         let group: BackupGroup = path.parse()?;
1097         api_datastore_latest_snapshot(&client, repo.store(), group).await?
1098     } else {
1099         let snapshot: BackupDir = path.parse()?;
1100         (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
1101     };
1102
1103     let target = json::required_string_param(&param, "target")?;
1104     let target = if target == "-" { None } else { Some(target) };
1105
1106     let crypto = crypto_parameters(&param)?;
1107
1108     let crypt_config = match crypto.enc_key {
1109         None => None,
1110         Some(ref key) => {
1111             let (key, _, _) =
1112                 decrypt_key(&key.key, &get_encryption_key_password).map_err(|err| {
1113                     eprintln!("{}", format_key_source(&key.source, "encryption"));
1114                     err
1115                 })?;
1116             Some(Arc::new(CryptConfig::new(key)?))
1117         }
1118     };
1119
1120     let client = BackupReader::start(
1121         client,
1122         crypt_config.clone(),
1123         repo.store(),
1124         &backup_type,
1125         &backup_id,
1126         backup_time,
1127         true,
1128     ).await?;
1129
1130     let (archive_name, archive_type) = parse_archive_type(archive_name);
1131
1132     let (manifest, backup_index_data) = client.download_manifest().await?;
1133
1134     if archive_name == ENCRYPTED_KEY_BLOB_NAME && crypt_config.is_none() {
1135         eprintln!("Restoring encrypted key blob without original key - skipping manifest fingerprint check!")
1136     } else {
1137         if manifest.signature.is_some() {
1138             if let Some(key) = &crypto.enc_key {
1139                 eprintln!("{}", format_key_source(&key.source, "encryption"));
1140             }
1141             if let Some(config) = &crypt_config {
1142                 eprintln!("Fingerprint: {}", Fingerprint::new(config.fingerprint()));
1143             }
1144         }
1145         manifest.check_fingerprint(crypt_config.as_ref().map(Arc::as_ref))?;
1146     }
1147
1148     if archive_name == MANIFEST_BLOB_NAME {
1149         if let Some(target) = target {
1150             replace_file(target, &backup_index_data, CreateOptions::new(), false)?;
1151         } else {
1152             let stdout = std::io::stdout();
1153             let mut writer = stdout.lock();
1154             writer.write_all(&backup_index_data)
1155                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1156         }
1157
1158         return Ok(Value::Null);
1159     }
1160
1161     let file_info = manifest.lookup_file_info(&archive_name)?;
1162
1163     if archive_type == ArchiveType::Blob {
1164
1165         let mut reader = client.download_blob(&manifest, &archive_name).await?;
1166
1167         if let Some(target) = target {
1168            let mut writer = std::fs::OpenOptions::new()
1169                 .write(true)
1170                 .create(true)
1171                 .create_new(true)
1172                 .open(target)
1173                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?;
1174             std::io::copy(&mut reader, &mut writer)?;
1175         } else {
1176             let stdout = std::io::stdout();
1177             let mut writer = stdout.lock();
1178             std::io::copy(&mut reader, &mut writer)
1179                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1180         }
1181
1182     } else if archive_type == ArchiveType::DynamicIndex {
1183
1184         let index = client.download_dynamic_index(&manifest, &archive_name).await?;
1185
1186         let most_used = index.find_most_used_chunks(8);
1187
1188         let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, file_info.chunk_crypt_mode(), most_used);
1189
1190         let mut reader = BufferedDynamicReader::new(index, chunk_reader);
1191
1192         let options = pbs_client::pxar::PxarExtractOptions {
1193             match_list: &[],
1194             extract_match_default: true,
1195             allow_existing_dirs,
1196             on_error: None,
1197         };
1198
1199         if let Some(target) = target {
1200             pbs_client::pxar::extract_archive(
1201                 pxar::decoder::Decoder::from_std(reader)?,
1202                 Path::new(target),
1203                 pbs_client::pxar::Flags::DEFAULT,
1204                 |path| {
1205                     if verbose {
1206                         println!("{:?}", path);
1207                     }
1208                 },
1209                 options,
1210             )
1211             .map_err(|err| format_err!("error extracting archive - {}", err))?;
1212         } else {
1213             let mut writer = std::fs::OpenOptions::new()
1214                 .write(true)
1215                 .open("/dev/stdout")
1216                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?;
1217
1218             std::io::copy(&mut reader, &mut writer)
1219                 .map_err(|err| format_err!("unable to pipe data - {}", err))?;
1220         }
1221     } else if archive_type == ArchiveType::FixedIndex {
1222
1223         let index = client.download_fixed_index(&manifest, &archive_name).await?;
1224
1225         let mut writer = if let Some(target) = target {
1226             std::fs::OpenOptions::new()
1227                 .write(true)
1228                 .create(true)
1229                 .create_new(true)
1230                 .open(target)
1231                 .map_err(|err| format_err!("unable to create target file {:?} - {}", target, err))?
1232         } else {
1233             std::fs::OpenOptions::new()
1234                 .write(true)
1235                 .open("/dev/stdout")
1236                 .map_err(|err| format_err!("unable to open /dev/stdout - {}", err))?
1237         };
1238
1239         dump_image(client.clone(), crypt_config.clone(), file_info.chunk_crypt_mode(), index, &mut writer, verbose).await?;
1240     }
1241
1242     Ok(Value::Null)
1243 }
1244
1245 #[api(
1246     input: {
1247         properties: {
1248             "dry-run": {
1249                 type: bool,
1250                 optional: true,
1251                 description: "Just show what prune would do, but do not delete anything.",
1252             },
1253             group: {
1254                 type: String,
1255                 description: "Backup group",
1256             },
1257             "prune-options": {
1258                 type: PruneOptions,
1259                 flatten: true,
1260             },
1261             "output-format": {
1262                 schema: OUTPUT_FORMAT,
1263                 optional: true,
1264             },
1265             quiet: {
1266                 type: bool,
1267                 optional: true,
1268                 default: false,
1269                 description: "Minimal output - only show removals.",
1270             },
1271             repository: {
1272                 schema: REPO_URL_SCHEMA,
1273                 optional: true,
1274             },
1275         },
1276     },
1277 )]
1278 /// Prune a backup repository.
1279 async fn prune(
1280     dry_run: Option<bool>,
1281     group: String,
1282     prune_options: PruneOptions,
1283     quiet: bool,
1284     mut param: Value
1285 ) -> Result<Value, Error> {
1286     let repo = extract_repository_from_value(&param)?;
1287
1288     let mut client = connect(&repo)?;
1289
1290     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
1291
1292     let group: BackupGroup = group.parse()?;
1293
1294     let output_format = extract_output_format(&mut param);
1295
1296     let mut api_param = serde_json::to_value(prune_options)?;
1297     if let Some(dry_run) = dry_run {
1298         api_param["dry-run"] = dry_run.into();
1299     }
1300     api_param["backup-type"] = group.backup_type().into();
1301     api_param["backup-id"] = group.backup_id().into();
1302
1303     let mut result = client.post(&path, Some(api_param)).await?;
1304
1305     record_repository(&repo);
1306
1307     let render_snapshot_path = |_v: &Value, record: &Value| -> Result<String, Error> {
1308         let item: PruneListItem = serde_json::from_value(record.to_owned())?;
1309         let snapshot = BackupDir::new(item.backup_type, item.backup_id, item.backup_time)?;
1310         Ok(snapshot.relative_path().to_str().unwrap().to_owned())
1311     };
1312
1313     let render_prune_action = |v: &Value, _record: &Value| -> Result<String, Error> {
1314         Ok(match v.as_bool() {
1315             Some(true) => "keep",
1316             Some(false) => "remove",
1317             None => "unknown",
1318         }.to_string())
1319     };
1320
1321     let options = default_table_format_options()
1322         .sortby("backup-type", false)
1323         .sortby("backup-id", false)
1324         .sortby("backup-time", false)
1325         .column(ColumnConfig::new("backup-id").renderer(render_snapshot_path).header("snapshot"))
1326         .column(ColumnConfig::new("backup-time").renderer(pbs_tools::format::render_epoch).header("date"))
1327         .column(ColumnConfig::new("keep").renderer(render_prune_action).header("action"))
1328         ;
1329
1330     let return_type = &pbs_api_types::ADMIN_DATASTORE_PRUNE_RETURN_TYPE;
1331
1332     let mut data = result["data"].take();
1333
1334     if quiet {
1335         let list: Vec<Value> = data.as_array().unwrap().iter().filter(|item| {
1336             item["keep"].as_bool() == Some(false)
1337         }).cloned().collect();
1338         data = list.into();
1339     }
1340
1341     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1342
1343     Ok(Value::Null)
1344 }
1345
1346 #[api(
1347    input: {
1348        properties: {
1349            repository: {
1350                schema: REPO_URL_SCHEMA,
1351                optional: true,
1352            },
1353            "output-format": {
1354                schema: OUTPUT_FORMAT,
1355                optional: true,
1356            },
1357        }
1358    },
1359     returns: {
1360         type: StorageStatus,
1361     },
1362 )]
1363 /// Get repository status.
1364 async fn status(param: Value) -> Result<Value, Error> {
1365
1366     let repo = extract_repository_from_value(&param)?;
1367
1368     let output_format = get_output_format(&param);
1369
1370     let client = connect(&repo)?;
1371
1372     let path = format!("api2/json/admin/datastore/{}/status", repo.store());
1373
1374     let mut result = client.get(&path, None).await?;
1375     let mut data = result["data"].take();
1376
1377     record_repository(&repo);
1378
1379     let render_total_percentage = |v: &Value, record: &Value| -> Result<String, Error> {
1380         let v = v.as_u64().unwrap();
1381         let total = record["total"].as_u64().unwrap();
1382         let roundup = total/200;
1383         let per = ((v+roundup)*100)/total;
1384         let info = format!(" ({} %)", per);
1385         Ok(format!("{} {:>8}", v, info))
1386     };
1387
1388     let options = default_table_format_options()
1389         .noheader(true)
1390         .column(ColumnConfig::new("total").renderer(render_total_percentage))
1391         .column(ColumnConfig::new("used").renderer(render_total_percentage))
1392         .column(ColumnConfig::new("avail").renderer(render_total_percentage));
1393
1394     let return_type = &API_METHOD_STATUS.returns;
1395
1396     format_and_print_result_full(&mut data, return_type, &output_format, &options);
1397
1398     Ok(Value::Null)
1399 }
1400
1401 /// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
1402 /// async use!
1403 ///
1404 /// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
1405 /// so that we can properly access it from multiple threads simultaneously while not issuing
1406 /// duplicate simultaneous reads over http.
1407 pub struct BufferedDynamicReadAt {
1408     inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
1409 }
1410
1411 impl BufferedDynamicReadAt {
1412     fn new(inner: BufferedDynamicReader<RemoteChunkReader>) -> Self {
1413         Self {
1414             inner: Mutex::new(inner),
1415         }
1416     }
1417 }
1418
1419 impl ReadAt for BufferedDynamicReadAt {
1420     fn start_read_at<'a>(
1421         self: Pin<&'a Self>,
1422         _cx: &mut Context,
1423         buf: &'a mut [u8],
1424         offset: u64,
1425     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1426         MaybeReady::Ready(tokio::task::block_in_place(move || {
1427             let mut reader = self.inner.lock().unwrap();
1428             reader.seek(SeekFrom::Start(offset))?;
1429             Ok(reader.read(buf)?)
1430         }))
1431     }
1432
1433     fn poll_complete<'a>(
1434         self: Pin<&'a Self>,
1435         _op: ReadAtOperation<'a>,
1436     ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
1437         panic!("BufferedDynamicReadAt::start_read_at returned Pending");
1438     }
1439 }
1440
1441 fn main() {
1442
1443     let backup_cmd_def = CliCommand::new(&API_METHOD_CREATE_BACKUP)
1444         .arg_param(&["backupspec"])
1445         .completion_cb("repository", complete_repository)
1446         .completion_cb("backupspec", complete_backup_source)
1447         .completion_cb("keyfile", complete_file_name)
1448         .completion_cb("master-pubkey-file", complete_file_name)
1449         .completion_cb("chunk-size", complete_chunk_size);
1450
1451     let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
1452         .completion_cb("repository", complete_repository)
1453         .completion_cb("keyfile", complete_file_name);
1454
1455     let list_cmd_def = CliCommand::new(&API_METHOD_LIST_BACKUP_GROUPS)
1456         .completion_cb("repository", complete_repository);
1457
1458     let garbage_collect_cmd_def = CliCommand::new(&API_METHOD_START_GARBAGE_COLLECTION)
1459         .completion_cb("repository", complete_repository);
1460
1461     let restore_cmd_def = CliCommand::new(&API_METHOD_RESTORE)
1462         .arg_param(&["snapshot", "archive-name", "target"])
1463         .completion_cb("repository", complete_repository)
1464         .completion_cb("snapshot", complete_group_or_snapshot)
1465         .completion_cb("archive-name", complete_archive_name)
1466         .completion_cb("target", complete_file_name);
1467
1468     let prune_cmd_def = CliCommand::new(&API_METHOD_PRUNE)
1469         .arg_param(&["group"])
1470         .completion_cb("group", complete_backup_group)
1471         .completion_cb("repository", complete_repository);
1472
1473     let status_cmd_def = CliCommand::new(&API_METHOD_STATUS)
1474         .completion_cb("repository", complete_repository);
1475
1476     let login_cmd_def = CliCommand::new(&API_METHOD_API_LOGIN)
1477         .completion_cb("repository", complete_repository);
1478
1479     let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
1480         .completion_cb("repository", complete_repository);
1481
1482     let version_cmd_def = CliCommand::new(&API_METHOD_API_VERSION)
1483         .completion_cb("repository", complete_repository);
1484
1485     let change_owner_cmd_def = CliCommand::new(&API_METHOD_CHANGE_BACKUP_OWNER)
1486         .arg_param(&["group", "new-owner"])
1487         .completion_cb("group", complete_backup_group)
1488         .completion_cb("new-owner",  complete_auth_id)
1489         .completion_cb("repository", complete_repository);
1490
1491     let cmd_def = CliCommandMap::new()
1492         .insert("backup", backup_cmd_def)
1493         .insert("garbage-collect", garbage_collect_cmd_def)
1494         .insert("list", list_cmd_def)
1495         .insert("login", login_cmd_def)
1496         .insert("logout", logout_cmd_def)
1497         .insert("prune", prune_cmd_def)
1498         .insert("restore", restore_cmd_def)
1499         .insert("snapshot", snapshot_mgtm_cli())
1500         .insert("status", status_cmd_def)
1501         .insert("key", key::cli())
1502         .insert("mount", mount_cmd_def())
1503         .insert("map", map_cmd_def())
1504         .insert("unmap", unmap_cmd_def())
1505         .insert("catalog", catalog_mgmt_cli())
1506         .insert("task", task_mgmt_cli())
1507         .insert("version", version_cmd_def)
1508         .insert("benchmark", benchmark_cmd_def)
1509         .insert("change-owner", change_owner_cmd_def)
1510
1511         .alias(&["files"], &["snapshot", "files"])
1512         .alias(&["forget"], &["snapshot", "forget"])
1513         .alias(&["upload-log"], &["snapshot", "upload-log"])
1514         .alias(&["snapshots"], &["snapshot", "list"])
1515         ;
1516
1517     let rpcenv = CliEnvironment::new();
1518     run_cli_command(cmd_def, rpcenv, Some(|future| {
1519         pbs_runtime::main(future)
1520     }));
1521 }