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