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