use std::collections::{HashSet, HashMap};
-use std::ffi::OsStr;
use std::io::{self, Write, Seek, SeekFrom};
use std::os::unix::fs::OpenOptionsExt;
-use std::os::unix::io::RawFd;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::{Arc, Mutex};
-use std::task::{Context, Poll};
+use std::task::Context;
use anyhow::{bail, format_err, Error};
use chrono::{Local, DateTime, Utc, TimeZone};
use futures::future::FutureExt;
-use futures::select;
use futures::stream::{StreamExt, TryStreamExt};
-use nix::unistd::{fork, ForkResult, pipe};
use serde_json::{json, Value};
-use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::mpsc;
use xdg::BaseDirectories;
use proxmox::api::schema::*;
use proxmox::api::cli::*;
use proxmox::api::api;
+use pxar::accessor::{MaybeReady, ReadAt, ReadAtOperation};
use proxmox_backup::tools;
use proxmox_backup::api2::types::*;
Shell,
};
+mod proxmox_backup_client;
+use proxmox_backup_client::*;
+
const ENV_VAR_PBS_FINGERPRINT: &str = "PBS_FINGERPRINT";
const ENV_VAR_PBS_PASSWORD: &str = "PBS_PASSWORD";
-const REPO_URL_SCHEMA: Schema = StringSchema::new("Repository URL.")
+pub const REPO_URL_SCHEMA: Schema = StringSchema::new("Repository URL.")
.format(&BACKUP_REPO_URL)
.max_length(256)
.schema();
-const KEYFILE_SCHEMA: Schema = StringSchema::new(
+pub const KEYFILE_SCHEMA: Schema = StringSchema::new(
"Path to encryption key. All data will be encrypted using this key.")
.schema();
std::env::var("PBS_REPOSITORY").ok()
}
-fn extract_repository_from_value(
+pub fn extract_repository_from_value(
param: &Value,
) -> Result<BackupRepository, Error> {
let _ = replace_file(path, new_data.to_string().as_bytes(), CreateOptions::new());
}
-fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
+pub fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
let mut result = vec![];
Ok(result["data"].take())
}
-async fn api_datastore_latest_snapshot(
+pub async fn api_datastore_latest_snapshot(
client: &HttpClient,
store: &str,
group: BackupGroup,
Ok((group.backup_type().to_owned(), group.backup_id().to_owned(), backup_time))
}
-
async fn backup_directory<P: AsRef<Path>>(
client: &BackupWriter,
+ previous_manifest: Option<Arc<BackupManifest>>,
dir_path: P,
archive_name: &str,
chunk_size: Option<usize>,
device_set: Option<HashSet<u64>>,
verbose: bool,
skip_lost_and_found: bool,
- crypt_config: Option<Arc<CryptConfig>>,
catalog: Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
exclude_pattern: Vec<MatchEntry>,
entries_max: usize,
});
let stats = client
- .upload_stream(archive_name, stream, "dynamic", None, crypt_config)
+ .upload_stream(previous_manifest, archive_name, stream, "dynamic", None)
.await?;
Ok(stats)
async fn backup_image<P: AsRef<Path>>(
client: &BackupWriter,
+ previous_manifest: Option<Arc<BackupManifest>>,
image_path: P,
archive_name: &str,
image_size: u64,
chunk_size: Option<usize>,
_verbose: bool,
- crypt_config: Option<Arc<CryptConfig>>,
) -> Result<BackupStats, Error> {
let path = image_path.as_ref().to_owned();
let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
let stats = client
- .upload_stream(archive_name, stream, "fixed", Some(image_size), crypt_config)
+ .upload_stream(previous_manifest, archive_name, stream, "fixed", Some(image_size))
.await?;
Ok(stats)
}
fn spawn_catalog_upload(
- client: Arc<BackupWriter>,
- crypt_config: Option<Arc<CryptConfig>>,
+ client: Arc<BackupWriter>
) -> Result<
(
Arc<Mutex<CatalogWriter<crate::tools::StdChannelWriter>>>,
tokio::spawn(async move {
let catalog_upload_result = client
- .upload_stream(CATALOG_NAME, catalog_chunk_stream, "dynamic", None, crypt_config)
+ .upload_stream(None, CATALOG_NAME, catalog_chunk_stream, "dynamic", None)
.await;
if let Err(ref err) = catalog_upload_result {
let client = BackupWriter::start(
client,
+ crypt_config.clone(),
repo.store(),
backup_type,
&backup_id,
verbose,
).await?;
+ let previous_manifest = if let Ok(previous_manifest) = client.download_previous_manifest().await {
+ Some(Arc::new(previous_manifest))
+ } else {
+ None
+ };
+
let snapshot = BackupDir::new(backup_type, backup_id, backup_time.timestamp());
let mut manifest = BackupManifest::new(snapshot);
BackupSpecificationType::CONFIG => {
println!("Upload config file '{}' to '{:?}' as {}", filename, repo, target);
let stats = client
- .upload_blob_from_file(&filename, &target, crypt_config.clone(), true)
+ .upload_blob_from_file(&filename, &target, true, Some(true))
.await?;
manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
}
BackupSpecificationType::LOGFILE => { // fixme: remove - not needed anymore ?
println!("Upload log file '{}' to '{:?}' as {}", filename, repo, target);
let stats = client
- .upload_blob_from_file(&filename, &target, crypt_config.clone(), true)
+ .upload_blob_from_file(&filename, &target, true, Some(true))
.await?;
manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
}
BackupSpecificationType::PXAR => {
// start catalog upload on first use
if catalog.is_none() {
- let (cat, res) = spawn_catalog_upload(client.clone(), crypt_config.clone())?;
+ let (cat, res) = spawn_catalog_upload(client.clone())?;
catalog = Some(cat);
catalog_result_tx = Some(res);
}
catalog.lock().unwrap().start_directory(std::ffi::CString::new(target.as_str())?.as_c_str())?;
let stats = backup_directory(
&client,
+ previous_manifest.clone(),
&filename,
&target,
chunk_size_opt,
devices.clone(),
verbose,
skip_lost_and_found,
- crypt_config.clone(),
catalog.clone(),
pattern_list.clone(),
entries_max as usize,
println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
let stats = backup_image(
&client,
- &filename,
+ previous_manifest.clone(),
+ &filename,
&target,
size,
chunk_size_opt,
verbose,
- crypt_config.clone(),
).await?;
manifest.add_file(target, stats.size, stats.csum, is_encrypted)?;
}
let target = "rsa-encrypted.key";
println!("Upload RSA encoded key to '{:?}' as {}", repo, target);
let stats = client
- .upload_blob_from_data(rsa_encrypted_key, target, None, false, false)
+ .upload_blob_from_data(rsa_encrypted_key, target, false, None)
.await?;
manifest.add_file(format!("{}.blob", target), stats.size, stats.csum, is_encrypted)?;
println!("Upload index.json to '{:?}'", repo);
let manifest = serde_json::to_string_pretty(&manifest)?.into();
client
- .upload_blob_from_data(manifest, MANIFEST_BLOB_NAME, crypt_config.clone(), true, true)
+ .upload_blob_from_data(manifest, MANIFEST_BLOB_NAME, true, Some(true))
.await?;
client.finish().await?;
result
}
-fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+pub fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
proxmox_backup::tools::runtime::main(async { complete_group_or_snapshot_do(arg, param).await })
}
.collect()
}
-fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+pub fn complete_pxar_archive_name(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
complete_server_file_name(arg, param)
.iter()
.filter_map(|v| {
.insert("change-passphrase", key_change_passphrase_cmd_def)
}
-fn mount(
- param: Value,
- _info: &ApiMethod,
- _rpcenv: &mut dyn RpcEnvironment,
-) -> Result<Value, Error> {
- let verbose = param["verbose"].as_bool().unwrap_or(false);
- if verbose {
- // This will stay in foreground with debug output enabled as None is
- // passed for the RawFd.
- return proxmox_backup::tools::runtime::main(mount_do(param, None));
- }
-
- // Process should be deamonized.
- // Make sure to fork before the async runtime is instantiated to avoid troubles.
- let pipe = pipe()?;
- match fork() {
- Ok(ForkResult::Parent { .. }) => {
- nix::unistd::close(pipe.1).unwrap();
- // Blocks the parent process until we are ready to go in the child
- let _res = nix::unistd::read(pipe.0, &mut [0]).unwrap();
- Ok(Value::Null)
- }
- Ok(ForkResult::Child) => {
- nix::unistd::close(pipe.0).unwrap();
- nix::unistd::setsid().unwrap();
- proxmox_backup::tools::runtime::main(mount_do(param, Some(pipe.1)))
- }
- Err(_) => bail!("failed to daemonize process"),
- }
-}
use proxmox_backup::client::RemoteChunkReader;
/// This is a workaround until we have cleaned up the chunk/reader/... infrastructure for better
/// Ideally BufferedDynamicReader gets replaced so the LruCache maps to `BroadcastFuture<Chunk>`,
/// so that we can properly access it from multiple threads simultaneously while not issuing
/// duplicate simultaneous reads over http.
-struct BufferedDynamicReadAt {
+pub struct BufferedDynamicReadAt {
inner: Mutex<BufferedDynamicReader<RemoteChunkReader>>,
}
}
}
-impl pxar::accessor::ReadAt for BufferedDynamicReadAt {
- fn poll_read_at(
- self: Pin<&Self>,
+impl ReadAt for BufferedDynamicReadAt {
+ fn start_read_at<'a>(
+ self: Pin<&'a Self>,
_cx: &mut Context,
- buf: &mut [u8],
+ buf: &'a mut [u8],
offset: u64,
- ) -> Poll<io::Result<usize>> {
+ ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
use std::io::Read;
- tokio::task::block_in_place(move || {
+ MaybeReady::Ready(tokio::task::block_in_place(move || {
let mut reader = self.inner.lock().unwrap();
reader.seek(SeekFrom::Start(offset))?;
- Poll::Ready(Ok(reader.read(buf)?))
- })
+ Ok(reader.read(buf)?)
+ }))
}
-}
-async fn mount_do(param: Value, pipe: Option<RawFd>) -> Result<Value, Error> {
- let repo = extract_repository_from_value(¶m)?;
- let archive_name = tools::required_string_param(¶m, "archive-name")?;
- let target = tools::required_string_param(¶m, "target")?;
- let client = connect(repo.host(), repo.user())?;
-
- record_repository(&repo);
-
- let path = tools::required_string_param(¶m, "snapshot")?;
- let (backup_type, backup_id, backup_time) = if path.matches('/').count() == 1 {
- let group: BackupGroup = path.parse()?;
- api_datastore_latest_snapshot(&client, repo.store(), group).await?
- } else {
- let snapshot: BackupDir = path.parse()?;
- (snapshot.group().backup_type().to_owned(), snapshot.group().backup_id().to_owned(), snapshot.backup_time())
- };
-
- let keyfile = param["keyfile"].as_str().map(PathBuf::from);
- let crypt_config = match keyfile {
- None => None,
- Some(path) => {
- let (key, _) = load_and_decrypt_key(&path, &get_encryption_key_password)?;
- Some(Arc::new(CryptConfig::new(key)?))
- }
- };
-
- let server_archive_name = if archive_name.ends_with(".pxar") {
- format!("{}.didx", archive_name)
- } else {
- bail!("Can only mount pxar archives.");
- };
-
- let client = BackupReader::start(
- client,
- crypt_config.clone(),
- repo.store(),
- &backup_type,
- &backup_id,
- backup_time,
- true,
- ).await?;
-
- let manifest = client.download_manifest().await?;
-
- if server_archive_name.ends_with(".didx") {
- let index = client.download_dynamic_index(&manifest, &server_archive_name).await?;
- let most_used = index.find_most_used_chunks(8);
- let chunk_reader = RemoteChunkReader::new(client.clone(), crypt_config, most_used);
- let reader = BufferedDynamicReader::new(index, chunk_reader);
- let archive_size = reader.archive_size();
- let reader: proxmox_backup::pxar::fuse::Reader =
- Arc::new(BufferedDynamicReadAt::new(reader));
- let decoder = proxmox_backup::pxar::fuse::Accessor::new(reader, archive_size).await?;
- let options = OsStr::new("ro,default_permissions");
-
- let session = proxmox_backup::pxar::fuse::Session::mount(
- decoder,
- &options,
- false,
- Path::new(target),
- )
- .map_err(|err| format_err!("pxar mount failed: {}", err))?;
-
- if let Some(pipe) = pipe {
- nix::unistd::chdir(Path::new("/")).unwrap();
- // Finish creation of daemon by redirecting filedescriptors.
- let nullfd = nix::fcntl::open(
- "/dev/null",
- nix::fcntl::OFlag::O_RDWR,
- nix::sys::stat::Mode::empty(),
- ).unwrap();
- nix::unistd::dup2(nullfd, 0).unwrap();
- nix::unistd::dup2(nullfd, 1).unwrap();
- nix::unistd::dup2(nullfd, 2).unwrap();
- if nullfd > 2 {
- nix::unistd::close(nullfd).unwrap();
- }
- // Signal the parent process that we are done with the setup and it can
- // terminate.
- nix::unistd::write(pipe, &[0u8])?;
- nix::unistd::close(pipe).unwrap();
- }
-
- let mut interrupt = signal(SignalKind::interrupt())?;
- select! {
- res = session.fuse() => res?,
- _ = interrupt.recv().fuse() => {
- // exit on interrupted
- }
- }
- } else {
- bail!("unknown archive file extension (expected .pxar)");
+ fn poll_complete<'a>(
+ self: Pin<&'a Self>,
+ _op: ReadAtOperation<'a>,
+ ) -> MaybeReady<io::Result<usize>, ReadAtOperation<'a>> {
+ panic!("LocalDynamicReadAt::start_read_at returned Pending");
}
-
- Ok(Value::Null)
}
+
#[api(
input: {
properties: {
.completion_cb("keyfile", tools::complete_file_name)
.completion_cb("chunk-size", complete_chunk_size);
+ let benchmark_cmd_def = CliCommand::new(&API_METHOD_BENCHMARK)
+ .completion_cb("repository", complete_repository)
+ .completion_cb("keyfile", tools::complete_file_name);
+
let upload_log_cmd_def = CliCommand::new(&API_METHOD_UPLOAD_LOG)
.arg_param(&["snapshot", "logfile"])
.completion_cb("snapshot", complete_backup_snapshot)
let logout_cmd_def = CliCommand::new(&API_METHOD_API_LOGOUT)
.completion_cb("repository", complete_repository);
- #[sortable]
- const API_METHOD_MOUNT: ApiMethod = ApiMethod::new(
- &ApiHandler::Sync(&mount),
- &ObjectSchema::new(
- "Mount pxar archive.",
- &sorted!([
- ("snapshot", false, &StringSchema::new("Group/Snapshot path.").schema()),
- ("archive-name", false, &StringSchema::new("Backup archive name.").schema()),
- ("target", false, &StringSchema::new("Target directory path.").schema()),
- ("repository", true, &REPO_URL_SCHEMA),
- ("keyfile", true, &StringSchema::new("Path to encryption key.").schema()),
- ("verbose", true, &BooleanSchema::new("Verbose output.").default(false).schema()),
- ]),
- )
- );
-
- let mount_cmd_def = CliCommand::new(&API_METHOD_MOUNT)
- .arg_param(&["snapshot", "archive-name", "target"])
- .completion_cb("repository", complete_repository)
- .completion_cb("snapshot", complete_group_or_snapshot)
- .completion_cb("archive-name", complete_pxar_archive_name)
- .completion_cb("target", tools::complete_file_name);
-
-
let cmd_def = CliCommandMap::new()
.insert("backup", backup_cmd_def)
.insert("upload-log", upload_log_cmd_def)
.insert("files", files_cmd_def)
.insert("status", status_cmd_def)
.insert("key", key_mgmt_cli())
- .insert("mount", mount_cmd_def)
+ .insert("mount", mount_cmd_def())
.insert("catalog", catalog_mgmt_cli())
- .insert("task", task_mgmt_cli());
+ .insert("task", task_mgmt_cli())
+ .insert("benchmark", benchmark_cmd_def);
let rpcenv = CliEnvironment::new();
run_cli_command(cmd_def, rpcenv, Some(|future| {
projects / proxmox-backup.git / blobdiff