src/api2/admin/datastore/catar.rs

   1 use failure::*;
   2
   3 use crate::tools;
   4 use crate::tools::wrapped_reader_stream::*;
   5 use crate::backup::datastore::*;
   6 use crate::backup::dynamic_index::*;
   7 //use crate::server::rest::*;
   8 use crate::api::schema::*;
   9 use crate::api::router::*;
  10
  11 use chrono::{Utc, TimeZone};
  12
  13 use serde_json::Value;
  14 use std::io::Write;
  15 use futures::*;
  16 use std::path::PathBuf;
  17 use std::sync::Arc;
  18
  19 use hyper::Body;
  20 use hyper::http::request::Parts;
  21
  22 pub struct UploadCaTar {
  23     stream: Body,
  24     index: DynamicIndexWriter,
  25     count: usize,
  26 }
  27
  28 impl Future for UploadCaTar {
  29     type Item = ();
  30     type Error = failure::Error;
  31
  32     fn poll(&mut self) -> Poll<(), failure::Error> {
  33         loop {
  34             match try_ready!(self.stream.poll()) {
  35                 Some(chunk) => {
  36                     self.count += chunk.len();
  37                     if let Err(err) = self.index.write(&chunk) {
  38                         bail!("writing chunk failed - {}", err);
  39                     }
  40                 }
  41                 None => {
  42                     self.index.close()?;
  43                     return Ok(Async::Ready(()))
  44                 }
  45             }
  46         }
  47     }
  48 }
  49
  50 fn upload_catar(
  51     parts: Parts,
  52     req_body: Body,
  53     param: Value,
  54     _info: &ApiAsyncMethod,
  55     _rpcenv: &mut RpcEnvironment,
  56 ) -> Result<BoxFut, Error> {
  57
  58     let store = tools::required_string_param(&param, "store")?;
  59     let archive_name = tools::required_string_param(&param, "archive_name")?;
  60
  61     let backup_type = tools::required_string_param(&param, "type")?;
  62     let backup_id = tools::required_string_param(&param, "id")?;
  63     let backup_time = tools::required_integer_param(&param, "time")?;
  64
  65     println!("Upload {}.catar to {} ({}/{}/{}/{}.didx)", archive_name, store,
  66              backup_type, backup_id, backup_time, archive_name);
  67
  68     let content_type = parts.headers.get(http::header::CONTENT_TYPE)
  69         .ok_or(format_err!("missing content-type header"))?;
  70
  71     if content_type != "application/x-proxmox-backup-catar" {
  72         bail!("got wrong content-type for catar archive upload");
  73     }
  74
  75     let chunk_size = 4*1024*1024;
  76
  77     let datastore = DataStore::lookup_datastore(store)?;
  78
  79     let mut path = datastore.create_backup_dir(backup_type, backup_id, backup_time)?;
  80
  81     let mut full_archive_name = PathBuf::from(archive_name);
  82     full_archive_name.set_extension("didx");
  83
  84     path.push(full_archive_name);
  85
  86     let index = datastore.create_dynamic_writer(path, chunk_size)?;
  87
  88     let upload = UploadCaTar { stream: req_body, index, count: 0};
  89
  90     let resp = upload.and_then(|_| {
  91
  92         let response = http::Response::builder()
  93             .status(200)
  94             .body(hyper::Body::empty())
  95             .unwrap();
  96
  97         Ok(response)
  98     });
  99
 100     Ok(Box::new(resp))
 101 }
 102
 103 pub fn api_method_upload_catar() -> ApiAsyncMethod {
 104     ApiAsyncMethod::new(
 105         upload_catar,
 106         ObjectSchema::new("Upload .catar backup file.")
 107             .required("store", StringSchema::new("Datastore name."))
 108             .required("archive_name", StringSchema::new("Backup archive name."))
 109             .required("type", StringSchema::new("Backup type.")
 110                       .format(Arc::new(ApiStringFormat::Enum(vec!["ct".into(), "host".into()]))))
 111             .required("id", StringSchema::new("Backup ID."))
 112             .required("time", IntegerSchema::new("Backup time (Unix epoch.)")
 113                       .minimum(1547797308))
 114
 115     )
 116 }
 117
 118 fn download_catar(
 119     _parts: Parts,
 120     _req_body: Body,
 121     param: Value,
 122     _info: &ApiAsyncMethod,
 123     _rpcenv: &mut RpcEnvironment,
 124 ) -> Result<BoxFut, Error> {
 125
 126     let store = tools::required_string_param(&param, "store")?;
 127     let archive_name = tools::required_string_param(&param, "archive_name")?;
 128
 129     let backup_type = tools::required_string_param(&param, "type")?;
 130     let backup_id = tools::required_string_param(&param, "id")?;
 131     let backup_time = tools::required_integer_param(&param, "time")?;
 132     let backup_time = Utc.timestamp(backup_time, 0);
 133
 134     println!("Download {}.catar from {} ({}/{}/{}/{}.didx)", archive_name, store,
 135              backup_type, backup_id, backup_time, archive_name);
 136
 137     let datastore = DataStore::lookup_datastore(store)?;
 138
 139     let mut path = datastore.get_backup_dir(backup_type, backup_id, backup_time);
 140
 141     let mut full_archive_name = PathBuf::from(archive_name);
 142     full_archive_name.set_extension("didx");
 143
 144     path.push(full_archive_name);
 145
 146     let index = datastore.open_dynamic_reader(path)?;
 147     let reader = BufferedDynamicReader::new(index);
 148     let stream = WrappedReaderStream::new(reader);
 149
 150     // fixme: set size, content type?
 151     let response = http::Response::builder()
 152         .status(200)
 153         .body(Body::wrap_stream(stream))?;
 154
 155     Ok(Box::new(future::ok(response)))
 156 }
 157
 158 pub fn api_method_download_catar() -> ApiAsyncMethod {
 159     ApiAsyncMethod::new(
 160         download_catar,
 161         ObjectSchema::new("Download .catar backup file.")
 162             .required("store", StringSchema::new("Datastore name."))
 163             .required("archive_name", StringSchema::new("Backup archive name."))
 164             .required("type", StringSchema::new("Backup type.")
 165                       .format(Arc::new(ApiStringFormat::Enum(vec!["ct".into(), "host".into()]))))
 166             .required("id", StringSchema::new("Backup ID."))
 167             .required("time", IntegerSchema::new("Backup time (Unix epoch.)")
 168                       .minimum(1547797308))
 169
 170     )
 171 }