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