use failure::*;
//use std::os::unix::io::AsRawFd;
-use chrono::{DateTime, Local, TimeZone};
+use chrono::{Local, TimeZone};
use std::path::{Path, PathBuf};
use std::collections::HashMap;
//use proxmox_backup::backup::datastore::*;
use serde_json::{json, Value};
-use hyper::Body;
+//use hyper::Body;
use std::sync::Arc;
use regex::Regex;
use xdg::BaseDirectories;
use lazy_static::lazy_static;
use futures::*;
+use tokio::sync::mpsc;
lazy_static! {
- static ref BACKUPSPEC_REGEX: Regex = Regex::new(r"^([a-zA-Z0-9_-]+\.(?:pxar|raw)):(.+)$").unwrap();
+ static ref BACKUPSPEC_REGEX: Regex = Regex::new(r"^([a-zA-Z0-9_-]+\.(?:pxar|img|conf)):(.+)$").unwrap();
}
}
fn backup_directory<P: AsRef<Path>>(
- client: &mut HttpClient,
- repo: &BackupRepository,
+ client: &BackupClient,
dir_path: P,
archive_name: &str,
- backup_id: &str,
- backup_time: DateTime<Local>,
- chunk_size: Option<u64>,
+ chunk_size: Option<usize>,
all_file_systems: bool,
verbose: bool,
) -> Result<(), Error> {
- let mut param = json!({
- "archive-name": archive_name,
- "backup-type": "host",
- "backup-id": backup_id,
- "backup-time": backup_time.timestamp(),
- });
-
- if let Some(size) = chunk_size {
- param["chunk-size"] = size.into();
- }
+ let pxar_stream = PxarBackupStream::open(dir_path.as_ref(), all_file_systems, verbose)?;
+ let chunk_stream = ChunkStream::new(pxar_stream, chunk_size);
- let query = tools::json_object_to_query(param)?;
+ let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
- let path = format!("api2/json/admin/datastore/{}/pxar?{}", repo.store(), query);
+ let stream = rx
+ .map_err(Error::from)
+ .and_then(|x| x); // flatten
- let stream = PxarBackupStream::open(dir_path.as_ref(), all_file_systems, verbose)?;
-
- let body = Body::wrap_stream(stream);
+ // spawn chunker inside a separate task so that it can run parallel
+ tokio::spawn(
+ tx.send_all(chunk_stream.then(|r| Ok(r)))
+ .map_err(|_| {}).map(|_| ())
+ );
- client.upload("application/x-proxmox-backup-pxar", body, &path).wait()?;
+ client.upload_stream(archive_name, stream, "dynamic", None).wait()?;
Ok(())
}
-/****
-fn backup_image(datastore: &DataStore, file: &std::fs::File, size: usize, target: &str, chunk_size: usize) -> Result<(), Error> {
+fn backup_image<P: AsRef<Path>>(
+ client: &BackupClient,
+ image_path: P,
+ archive_name: &str,
+ image_size: u64,
+ chunk_size: Option<usize>,
+ _verbose: bool,
+) -> Result<(), Error> {
- let mut target = PathBuf::from(target);
+ let path = image_path.as_ref().to_owned();
- if let Some(ext) = target.extension() {
- if ext != "fidx" {
- bail!("got wrong file extension - expected '.fidx'");
- }
- } else {
- target.set_extension("fidx");
- }
+ let file = tokio::fs::File::open(path).wait()?;
- let mut index = datastore.create_image_writer(&target, size, chunk_size)?;
+ let stream = tokio::codec::FramedRead::new(file, tokio::codec::BytesCodec::new())
+ .map_err(Error::from);
- tools::file_chunker(file, chunk_size, |pos, chunk| {
- index.add_chunk(pos, chunk)?;
- Ok(true)
- })?;
+ let stream = FixedChunkStream::new(stream, chunk_size.unwrap_or(4*1024*1024));
- index.close()?; // commit changes
+ client.upload_stream(archive_name, stream, "fixed", Some(image_size)).wait()?;
Ok(())
}
-*/
fn strip_chunked_file_expenstions(list: Vec<String>) -> Vec<String> {
fn list_backups(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
let path = format!("api2/json/admin/datastore/{}/backups", repo.store());
- let result = client.get(&path)?;
+ let result = client.get(&path, None)?;
record_repository(&repo);
fn list_backup_groups(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
- let mut result = client.get(&path).wait()?;
+ let mut result = client.get(&path, None).wait()?;
record_repository(&repo);
fn list_snapshots(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
let path = tools::required_string_param(¶m, "group")?;
let group = BackupGroup::parse(path)?;
- let query = tools::json_object_to_query(json!({
- "backup-type": group.backup_type(),
- "backup-id": group.backup_id(),
- }))?;
-
let client = HttpClient::new(repo.host(), repo.user())?;
- let path = format!("api2/json/admin/datastore/{}/snapshots?{}", repo.store(), query);
+ let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
- // fixme: params
- let result = client.get(&path).wait()?;
+ let result = client.get(&path, Some(json!({
+ "backup-type": group.backup_type(),
+ "backup-id": group.backup_id(),
+ }))).wait()?;
record_repository(&repo);
fn forget_snapshots(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
let path = tools::required_string_param(¶m, "snapshot")?;
let snapshot = BackupDir::parse(path)?;
- let query = tools::json_object_to_query(json!({
- "backup-type": snapshot.group().backup_type(),
- "backup-id": snapshot.group().backup_id(),
- "backup-time": snapshot.backup_time().timestamp(),
- }))?;
-
let mut client = HttpClient::new(repo.host(), repo.user())?;
- let path = format!("api2/json/admin/datastore/{}/snapshots?{}", repo.store(), query);
+ let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
- let result = client.delete(&path).wait()?;
+ let result = client.delete(&path, Some(json!({
+ "backup-type": snapshot.group().backup_type(),
+ "backup-id": snapshot.group().backup_id(),
+ "backup-time": snapshot.backup_time().timestamp(),
+ }))).wait()?;
record_repository(&repo);
fn start_garbage_collection(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
fn create_backup(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
let verbose = param["verbose"].as_bool().unwrap_or(false);
- let chunk_size_opt = param["chunk-size"].as_u64().map(|v| v*1024);
+ let chunk_size_opt = param["chunk-size"].as_u64().map(|v| (v*1024) as usize);
if let Some(size) = chunk_size_opt {
verify_chunk_size(size)?;
let mut upload_list = vec![];
+ enum BackupType { PXAR, IMAGE, CONFIG };
+
for backupspec in backupspec_list {
let (target, filename) = parse_backupspec(backupspec.as_str().unwrap())?;
- let stat = match nix::sys::stat::stat(filename) {
- Ok(s) => s,
+ use std::os::unix::fs::FileTypeExt;
+
+ let metadata = match std::fs::metadata(filename) {
+ Ok(m) => m,
Err(err) => bail!("unable to access '{}' - {}", filename, err),
};
+ let file_type = metadata.file_type();
- if (stat.st_mode & libc::S_IFDIR) != 0 {
-
- upload_list.push((filename.to_owned(), target.to_owned()));
+ let extension = Path::new(target).extension().map(|s| s.to_str().unwrap()).unwrap();
- } else if (stat.st_mode & (libc::S_IFREG|libc::S_IFBLK)) != 0 {
- if stat.st_size <= 0 { bail!("got strange file size '{}'", stat.st_size); }
- let _size = stat.st_size as usize;
+ match extension {
+ "pxar" => {
+ if !file_type.is_dir() {
+ bail!("got unexpected file type (expected directory)");
+ }
+ upload_list.push((BackupType::PXAR, filename.to_owned(), target.to_owned(), 0));
+ }
+ "img" => {
- panic!("implement me");
+ if !(file_type.is_file() || file_type.is_block_device()) {
+ bail!("got unexpected file type (expected file or block device)");
+ }
- //backup_image(&datastore, &file, size, &target, chunk_size)?;
+ let size = tools::image_size(&PathBuf::from(filename))?;
- // let idx = datastore.open_image_reader(target)?;
- // idx.print_info();
+ if size == 0 { bail!("got zero-sized file '{}'", filename); }
- } else {
- bail!("unsupported file type (expected a directory, file or block device)");
+ upload_list.push((BackupType::IMAGE, filename.to_owned(), target.to_owned(), size));
+ }
+ "conf" => {
+ if !file_type.is_file() {
+ bail!("got unexpected file type (expected regular file)");
+ }
+ upload_list.push((BackupType::CONFIG, filename.to_owned(), target.to_owned(), metadata.len()));
+ }
+ _ => {
+ bail!("got unknown archive extension '{}'", extension);
+ }
}
}
let backup_time = Local.timestamp(Local::now().timestamp(), 0);
- let mut client = HttpClient::new(repo.host(), repo.user())?;
-
+ let client = HttpClient::new(repo.host(), repo.user())?;
record_repository(&repo);
println!("Starting backup");
println!("Client name: {}", tools::nodename());
println!("Start Time: {}", backup_time.to_rfc3339());
- for (filename, target) in upload_list {
- println!("Upload '{}' to '{:?}' as {}", filename, repo, target);
- backup_directory(&mut client, &repo, &filename, &target, backup_id, backup_time,
- chunk_size_opt, all_file_systems, verbose)?;
+ let client = client.start_backup(repo.store(), "host", &backup_id, verbose).wait()?;
+
+ for (backup_type, filename, target, size) in upload_list {
+ match backup_type {
+ BackupType::CONFIG => {
+ println!("Upload config file '{}' to '{:?}' as {}", filename, repo, target);
+ client.upload_config(&filename, &target).wait()?;
+ }
+ BackupType::PXAR => {
+ println!("Upload directory '{}' to '{:?}' as {}", filename, repo, target);
+ backup_directory(&client, &filename, &target, chunk_size_opt, all_file_systems, verbose)?;
+ }
+ BackupType::IMAGE => {
+ println!("Upload image '{}' to '{:?}' as {}", filename, repo, target);
+ backup_image(&client, &filename, &target, size, chunk_size_opt, verbose)?;
+ }
+ }
}
+ client.finish().wait()?;
+
let end_time = Local.timestamp(Local::now().timestamp(), 0);
let elapsed = end_time.signed_duration_since(backup_time);
println!("Duration: {}", elapsed);
fn restore(
param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
if path.matches('/').count() == 1 {
let group = BackupGroup::parse(path)?;
- let subquery = tools::json_object_to_query(json!({
+ let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
+ let result = client.get(&path, Some(json!({
"backup-type": group.backup_type(),
"backup-id": group.backup_id(),
- }))?;
-
- let path = format!("api2/json/admin/datastore/{}/snapshots?{}", repo.store(), subquery);
- let result = client.get(&path).wait()?;
+ }))).wait()?;
let list = result["data"].as_array().unwrap();
if list.len() == 0 {
fn prune(
mut param: Value,
_info: &ApiMethod,
- _rpcenv: &mut RpcEnvironment,
+ _rpcenv: &mut dyn RpcEnvironment,
) -> Result<Value, Error> {
let repo_url = tools::required_string_param(¶m, "repository")?;
_ => return Value::Null,
};
- let mut resp = match client.get(url).wait() {
+ let mut resp = match client.get(url, None).wait() {
Ok(v) => v,
_ => return Value::Null,
};
projects / proxmox-backup.git / blobdiff