]> git.proxmox.com Git - proxmox-backup.git/blobdiff - src/bin/proxmox-backup-client.rs
src/bin/proxmox-backup-client.rs: implement image/block-device upload
[proxmox-backup.git] / src / bin / proxmox-backup-client.rs
index 6ae03bc662f8ff3e4f16f315ac1b0f61dccac629..545c208d7e2e13ae1000de20cca7d7ec420b7744 100644 (file)
@@ -14,7 +14,7 @@ use proxmox_backup::client::*;
 use proxmox_backup::backup::*;
 //use proxmox_backup::backup::image_index::*;
 //use proxmox_backup::config::datastore;
-//use proxmox_backup::catar::encoder::*;
+//use proxmox_backup::pxar::encoder::*;
 //use proxmox_backup::backup::datastore::*;
 
 use serde_json::{json, Value};
@@ -24,9 +24,11 @@ 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_-]+\.(?:catar|raw)):(.+)$").unwrap();
+    static ref BACKUPSPEC_REGEX: Regex = Regex::new(r"^([a-zA-Z0-9_-]+\.(?:pxar|img)):(.+)$").unwrap();
 }
 
 
@@ -43,7 +45,7 @@ fn record_repository(repo: &BackupRepository) {
         _ => return,
     };
 
-    let mut data = tools::file_get_json(&path).unwrap_or(json!({}));
+    let mut data = tools::file_get_json(&path, None).unwrap_or(json!({}));
 
     let repo = repo.to_string();
 
@@ -78,7 +80,7 @@ fn record_repository(repo: &BackupRepository) {
     let _ = tools::file_set_contents(path, new_data.to_string().as_bytes(), None);
 }
 
-fn complete_repository(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+fn complete_repository(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
 
     let mut result = vec![];
 
@@ -93,10 +95,10 @@ fn complete_repository(arg: &str, param: &HashMap<String, String>) -> Vec<String
         _ => return result,
     };
 
-    let data = tools::file_get_json(&path).unwrap_or(json!({}));
+    let data = tools::file_get_json(&path, None).unwrap_or(json!({}));
 
     if let Some(map) = data.as_object() {
-        for (repo, count) in map {
+        for (repo, _count) in map {
             result.push(repo.to_owned());
         }
     }
@@ -105,37 +107,61 @@ fn complete_repository(arg: &str, param: &HashMap<String, String>) -> Vec<String
 }
 
 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>,
     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 {
+        unimplemented!();
+    }
+
+    let pxar_stream = PxarBackupStream::open(dir_path.as_ref(), all_file_systems, verbose)?;
+    let chunk_stream = ChunkStream::new(pxar_stream);
+
+    let (tx, rx) = mpsc::channel(10); // allow to buffer 10 chunks
+
+    let stream = rx
+        .map_err(Error::from)
+        .and_then(|x| x); // flatten
+
+    // 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(|e| {}).map(|_| ())
+    );
+
+    client.upload_stream(archive_name, stream, "dynamic", None).wait()?;
+
+    Ok(())
+}
+
+fn backup_image<P: AsRef<Path>>(
+    client: &BackupClient,
+    image_path: P,
+    archive_name: &str,
+    image_size: u64,
+    chunk_size: Option<u64>,
+    verbose: bool,
+) -> Result<(), Error> {
 
-    if let Some(size) = chunk_size {
-        param["chunk-size"] = size.into();
+    if let Some(_size) = chunk_size {
+        unimplemented!();
     }
 
-    let query = tools::json_object_to_query(param)?;
+    let path = image_path.as_ref().to_owned();
 
-    let path = format!("api2/json/admin/datastore/{}/catar?{}", repo.store(), query);
+    let file = tokio::fs::File::open(path).wait()?;
 
-    let stream = CaTarBackupStream::open(dir_path.as_ref(), all_file_systems, verbose)?;
+    let stream = tokio::codec::FramedRead::new(file, tokio::codec::BytesCodec::new())
+        .map_err(Error::from);
 
-    let body = Body::wrap_stream(stream);
+    let stream = FixedChunkStream::new(stream, 4*1024*1024);
 
-    client.upload("application/x-proxmox-backup-catar", body, &path)?;
+    client.upload_stream(archive_name, stream, "fixed", Some(image_size)).wait()?;
 
     Ok(())
 }
@@ -183,6 +209,7 @@ fn strip_chunked_file_expenstions(list: Vec<String>) -> Vec<String> {
     result
 }
 
+/* not used:
 fn list_backups(
     param: Value,
     _info: &ApiMethod,
@@ -192,11 +219,11 @@ fn list_backups(
     let repo_url = tools::required_string_param(&param, "repository")?;
     let repo: BackupRepository = repo_url.parse()?;
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
+    let mut client = HttpClient::new(repo.host(), repo.user())?;
 
     let path = format!("api2/json/admin/datastore/{}/backups", repo.store());
 
-    let result = client.get(&path)?;
+    let result = client.get(&path, None)?;
 
     record_repository(&repo);
 
@@ -223,6 +250,7 @@ fn list_backups(
     //Ok(result)
     Ok(Value::Null)
 }
+ */
 
 fn list_backup_groups(
     param: Value,
@@ -233,11 +261,11 @@ fn list_backup_groups(
     let repo_url = tools::required_string_param(&param, "repository")?;
     let repo: BackupRepository = repo_url.parse()?;
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
+    let client = HttpClient::new(repo.host(), repo.user())?;
 
     let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
 
-    let mut result = client.get(&path)?;
+    let mut result = client.get(&path, None).wait()?;
 
     record_repository(&repo);
 
@@ -293,17 +321,14 @@ fn list_snapshots(
     let path = tools::required_string_param(&param, "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 mut client = HttpClient::new(repo.host(), repo.user());
+    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)?;
+    let result = client.get(&path, Some(json!({
+        "backup-type": group.backup_type(),
+        "backup-id": group.backup_id(),
+    }))).wait()?;
 
     record_repository(&repo);
 
@@ -341,17 +366,15 @@ fn forget_snapshots(
     let path = tools::required_string_param(&param, "snapshot")?;
     let snapshot = BackupDir::parse(path)?;
 
-    let query = tools::json_object_to_query(json!({
+    let mut client = HttpClient::new(repo.host(), repo.user())?;
+
+    let path = format!("api2/json/admin/datastore/{}/snapshots", repo.store());
+
+    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(),
-    }))?;
-
-    let mut client = HttpClient::new(repo.host(), repo.user());
-
-    let path = format!("api2/json/admin/datastore/{}/snapshots?{}", repo.store(), query);
-
-    let result = client.delete(&path)?;
+    }))).wait()?;
 
     record_repository(&repo);
 
@@ -367,11 +390,11 @@ fn start_garbage_collection(
     let repo_url = tools::required_string_param(&param, "repository")?;
     let repo: BackupRepository = repo_url.parse()?;
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
+    let mut client = HttpClient::new(repo.host(), repo.user())?;
 
     let path = format!("api2/json/admin/datastore/{}/gc", repo.store());
 
-    let result = client.post(&path)?;
+    let result = client.post(&path, None).wait()?;
 
     record_repository(&repo);
 
@@ -412,28 +435,30 @@ fn create_backup(
 
     let mut upload_list = vec![];
 
+    enum BackupType { PXAR, IMAGE };
+
     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 {
+        if file_type.is_dir() {
 
-            upload_list.push((filename.to_owned(), target.to_owned()));
+            upload_list.push((BackupType::PXAR, filename.to_owned(), target.to_owned(), 0));
 
-        } 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;
+        } else if file_type.is_file() || file_type.is_block_device() {
 
-            panic!("implement me");
+            let size = tools::image_size(&PathBuf::from(filename))?;
 
-            //backup_image(&datastore, &file, size, &target, chunk_size)?;
+            if size == 0 { bail!("got zero-sized file '{}'", filename); }
 
-            // let idx = datastore.open_image_reader(target)?;
-            // idx.print_info();
+            upload_list.push((BackupType::IMAGE, filename.to_owned(), target.to_owned(), size));
 
         } else {
             bail!("unsupported file type (expected a directory, file or block device)");
@@ -442,22 +467,30 @@ fn create_backup(
 
     let backup_time = Local.timestamp(Local::now().timestamp(), 0);
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
-
-    client.login()?; // login before starting backup
-
+    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::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);
@@ -474,8 +507,8 @@ fn complete_backup_source(arg: &str, param: &HashMap<String, String>) -> Vec<Str
     let data: Vec<&str> = arg.splitn(2, ':').collect();
 
     if data.len() != 2 {
-        result.push(String::from("root.catar:/"));
-        result.push(String::from("etc.catar:/etc"));
+        result.push(String::from("root.pxar:/"));
+        result.push(String::from("etc.pxar:/etc"));
         return result;
     }
 
@@ -499,9 +532,7 @@ fn restore(
 
     let archive_name = tools::required_string_param(&param, "archive-name")?;
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
-
-    client.login()?; // login before starting
+    let mut client = HttpClient::new(repo.host(), repo.user())?;
 
     record_repository(&repo);
 
@@ -512,13 +543,11 @@ fn restore(
     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()?;
 
         let list = result["data"].as_array().unwrap();
         if list.len() == 0 {
@@ -544,14 +573,14 @@ fn restore(
 
     let target = tools::required_string_param(&param, "target")?;
 
-    if archive_name.ends_with(".catar") {
-        let path = format!("api2/json/admin/datastore/{}/catar?{}", repo.store(), query);
+    if archive_name.ends_with(".pxar") {
+        let path = format!("api2/json/admin/datastore/{}/pxar?{}", repo.store(), query);
 
         println!("DOWNLOAD FILE {} to {}", path, target);
 
         let target = PathBuf::from(target);
-        let writer = CaTarBackupWriter::new(&target, true)?;
-        client.download(&path, Box::new(writer))?;
+        let writer = PxarDecodeWriter::new(&target, true)?;
+        client.download(&path, Box::new(writer)).wait()?;
     } else {
         bail!("unknown file extensions - unable to download '{}'", archive_name);
     }
@@ -568,43 +597,69 @@ fn prune(
     let repo_url = tools::required_string_param(&param, "repository")?;
     let repo: BackupRepository = repo_url.parse()?;
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
+    let mut client = HttpClient::new(repo.host(), repo.user())?;
 
     let path = format!("api2/json/admin/datastore/{}/prune", repo.store());
 
     param.as_object_mut().unwrap().remove("repository");
 
-    let result = client.post_json(&path, param)?;
+    let result = client.post(&path, Some(param)).wait()?;
 
     record_repository(&repo);
 
     Ok(result)
 }
 
-fn complete_backup_group(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+// like get, but simply ignore errors and return Null instead
+fn try_get(repo: &BackupRepository, url: &str) -> Value {
 
-    let mut result = vec![];
+    let client = match HttpClient::new(repo.host(), repo.user()) {
+        Ok(v) => v,
+        _ => return Value::Null,
+    };
+
+    let mut resp = match client.get(url, None).wait() {
+        Ok(v) => v,
+        _ => return Value::Null,
+    };
+
+    if let Some(map) = resp.as_object_mut() {
+        if let Some(data) = map.remove("data") {
+            return data;
+        }
+    }
+    Value::Null
+}
+
+fn extract_repo(param: &HashMap<String, String>) -> Option<BackupRepository> {
 
     let repo_url = match param.get("repository") {
         Some(v) => v,
-        _ => return result,
+        _ => return None,
     };
 
     let repo: BackupRepository = match repo_url.parse() {
         Ok(v) => v,
-        _ => return result,
+        _ => return None,
     };
 
-    let mut client = HttpClient::new(repo.host(), repo.user());
+    Some(repo)
+}
+
+fn complete_backup_group(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
 
-    let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
+    let mut result = vec![];
 
-    let resp = match client.try_get(&path) {
-        Ok(v) => v,
+    let repo = match extract_repo(param) {
+        Some(v) => v,
         _ => return result,
     };
 
-    if let Some(list) = resp["data"].as_array() {
+    let path = format!("api2/json/admin/datastore/{}/groups", repo.store());
+
+    let data = try_get(&repo, &path);
+
+    if let Some(list) = data.as_array() {
         for item in list {
             if let (Some(backup_id), Some(backup_type)) =
                 (item["backup-id"].as_str(), item["backup-type"].as_str())
@@ -617,6 +672,102 @@ fn complete_backup_group(arg: &str, param: &HashMap<String, String>) -> Vec<Stri
     result
 }
 
+fn complete_group_or_snapshot(arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+
+    let mut result = vec![];
+
+     let repo = match extract_repo(param) {
+        Some(v) => v,
+        _ => return result,
+    };
+
+    if arg.matches('/').count() < 2 {
+        let groups = complete_backup_group(arg, param);
+        for group in groups {
+            result.push(group.to_string());
+            result.push(format!("{}/", group));
+        }
+        return result;
+    }
+
+    let mut parts = arg.split('/');
+    let query = tools::json_object_to_query(json!({
+        "backup-type": parts.next().unwrap(),
+        "backup-id": parts.next().unwrap(),
+    })).unwrap();
+
+    let path = format!("api2/json/admin/datastore/{}/snapshots?{}", repo.store(), query);
+
+    let data = try_get(&repo, &path);
+
+    if let Some(list) = data.as_array() {
+        for item in list {
+            if let (Some(backup_id), Some(backup_type), Some(backup_time)) =
+                (item["backup-id"].as_str(), item["backup-type"].as_str(), item["backup-time"].as_i64())
+            {
+                let snapshot = BackupDir::new(backup_type, backup_id, backup_time);
+                result.push(snapshot.relative_path().to_str().unwrap().to_owned());
+            }
+        }
+    }
+
+    result
+}
+
+fn complete_archive_name(_arg: &str, param: &HashMap<String, String>) -> Vec<String> {
+
+    let mut result = vec![];
+
+    let repo = match extract_repo(param) {
+        Some(v) => v,
+        _ => return result,
+    };
+
+    let snapshot = match param.get("snapshot") {
+        Some(path) => {
+            match BackupDir::parse(path) {
+                Ok(v) => v,
+                _ => return result,
+            }
+        }
+        _ => return result,
+    };
+
+    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(),
+    })).unwrap();
+
+    let path = format!("api2/json/admin/datastore/{}/files?{}", repo.store(), query);
+
+    let data = try_get(&repo, &path);
+
+    if let Some(list) = data.as_array() {
+        for item in list {
+            if let Some(filename) = item.as_str() {
+                result.push(filename.to_owned());
+            }
+        }
+    }
+
+    strip_chunked_file_expenstions(result)
+}
+
+fn complete_chunk_size(_arg: &str, _param: &HashMap<String, String>) -> Vec<String> {
+
+    let mut result = vec![];
+
+    let mut size = 64;
+    loop {
+        result.push(size.to_string());
+        size = size * 2;
+        if size > 4096 { break; }
+    }
+
+    result
+}
+
 fn main() {
 
     let repo_url_schema: Arc<Schema> = Arc::new(
@@ -660,7 +811,8 @@ fn main() {
         ))
         .arg_param(vec!["repository", "backupspec"])
         .completion_cb("repository", complete_repository)
-        .completion_cb("backupspec", complete_backup_source);
+        .completion_cb("backupspec", complete_backup_source)
+        .completion_cb("chunk-size", complete_chunk_size);
 
     let list_cmd_def = CliCommand::new(
         ApiMethod::new(
@@ -690,7 +842,8 @@ fn main() {
                 .required("snapshot", StringSchema::new("Snapshot path."))
         ))
         .arg_param(vec!["repository", "snapshot"])
-        .completion_cb("repository", complete_repository);
+        .completion_cb("repository", complete_repository)
+        .completion_cb("snapshot", complete_group_or_snapshot);
 
     let garbage_collect_cmd_def = CliCommand::new(
         ApiMethod::new(
@@ -711,7 +864,10 @@ fn main() {
                 .required("target", StringSchema::new("Target directory path."))
         ))
         .arg_param(vec!["repository", "snapshot", "archive-name", "target"])
-        .completion_cb("repository", complete_repository);
+        .completion_cb("repository", complete_repository)
+        .completion_cb("snapshot", complete_group_or_snapshot)
+        .completion_cb("archive-name", complete_archive_name)
+        .completion_cb("target", tools::complete_file_name);
 
     let prune_cmd_def = CliCommand::new(
         ApiMethod::new(
@@ -733,6 +889,9 @@ fn main() {
         .insert("restore".to_owned(), restore_cmd_def.into())
         .insert("snapshots".to_owned(), snapshots_cmd_def.into());
 
-    run_cli_command(cmd_def.into());
+    hyper::rt::run(futures::future::lazy(move || {
+        run_cli_command(cmd_def.into());
+        Ok(())
+    }));
 
 }