aboutsummaryrefslogtreecommitdiff
path: root/src/api/s3_put.rs
diff options
context:
space:
mode:
authorAlex Auvolat <alex@adnab.me>2020-04-26 20:39:32 +0000
committerAlex Auvolat <alex@adnab.me>2020-04-26 20:39:32 +0000
commit81ecc4999e16c58ce6d0e97501f7b6b1497f6cf6 (patch)
treed22638738b6105ebc8c7eb8fe1d633bf3627b034 /src/api/s3_put.rs
parent1999c0ae5152917625558c588e71b27f63950f41 (diff)
downloadgarage-81ecc4999e16c58ce6d0e97501f7b6b1497f6cf6.tar.gz
garage-81ecc4999e16c58ce6d0e97501f7b6b1497f6cf6.zip
Implement multipart uploads
Diffstat (limited to 'src/api/s3_put.rs')
-rw-r--r--src/api/s3_put.rs234
1 files changed, 216 insertions, 18 deletions
diff --git a/src/api/s3_put.rs b/src/api/s3_put.rs
index 084af916..f1755c17 100644
--- a/src/api/s3_put.rs
+++ b/src/api/s3_put.rs
@@ -1,11 +1,13 @@
use std::collections::VecDeque;
+use std::fmt::Write;
use std::sync::Arc;
use futures::stream::*;
-use hyper::Body;
+use hyper::{Body, Request, Response};
use garage_util::data::*;
use garage_util::error::Error;
+use garage_table::*;
use garage_core::block::INLINE_THRESHOLD;
use garage_core::block_ref_table::*;
@@ -13,14 +15,17 @@ use garage_core::garage::Garage;
use garage_core::object_table::*;
use garage_core::version_table::*;
+use crate::http_util::*;
+
pub async fn handle_put(
garage: Arc<Garage>,
- mime_type: &str,
+ req: Request<Body>,
bucket: &str,
key: &str,
- body: Body,
-) -> Result<UUID, Error> {
+) -> Result<Response<BodyType>, Error> {
let version_uuid = gen_uuid();
+ let mime_type = get_mime_type(&req)?;
+ let body = req.into_body();
let mut chunker = BodyChunker::new(body, garage.config.block_size);
let first_block = match chunker.next().await? {
@@ -31,10 +36,10 @@ pub async fn handle_put(
let mut object_version = ObjectVersion {
uuid: version_uuid,
timestamp: now_msec(),
- mime_type: mime_type.to_string(),
+ mime_type,
size: first_block.len() as u64,
state: ObjectVersionState::Uploading,
- data: ObjectVersionData::DeleteMarker,
+ data: ObjectVersionData::Uploading,
};
if first_block.len() < INLINE_THRESHOLD {
@@ -43,7 +48,7 @@ pub async fn handle_put(
let object = Object::new(bucket.into(), key.into(), vec![object_version]);
garage.object_table.insert(&object).await?;
- return Ok(version_uuid);
+ return Ok(put_response(version_uuid));
}
let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]);
@@ -53,9 +58,30 @@ pub async fn handle_put(
let object = Object::new(bucket.into(), key.into(), vec![object_version.clone()]);
garage.object_table.insert(&object).await?;
+ let total_size = read_and_put_blocks(&garage, version, 1, first_block, first_block_hash, &mut chunker).await?;
+
+ // TODO: if at any step we have an error, we should undo everything we did
+
+ object_version.state = ObjectVersionState::Complete;
+ object_version.size = total_size;
+
+ let object = Object::new(bucket.into(), key.into(), vec![object_version]);
+ garage.object_table.insert(&object).await?;
+
+ Ok(put_response(version_uuid))
+}
+
+async fn read_and_put_blocks(
+ garage: &Arc<Garage>,
+ version: Version,
+ part_number: u64,
+ first_block: Vec<u8>,
+ first_block_hash: Hash,
+ chunker: &mut BodyChunker,
+) -> Result<u64, Error> {
let mut next_offset = first_block.len();
let mut put_curr_version_block =
- put_block_meta(garage.clone(), &version, 0, 0, first_block_hash);
+ put_block_meta(garage.clone(), &version, part_number, 0, first_block_hash, first_block.len() as u64);
let mut put_curr_block = garage
.block_manager
.rpc_put_block(first_block_hash, first_block);
@@ -67,7 +93,7 @@ pub async fn handle_put(
let block_hash = hash(&block[..]);
let block_len = block.len();
put_curr_version_block =
- put_block_meta(garage.clone(), &version, 0, next_offset as u64, block_hash);
+ put_block_meta(garage.clone(), &version, part_number, next_offset as u64, block_hash, block_len as u64);
put_curr_block = garage.block_manager.rpc_put_block(block_hash, block);
next_offset += block_len;
} else {
@@ -75,15 +101,7 @@ pub async fn handle_put(
}
}
- // TODO: if at any step we have an error, we should undo everything we did
-
- object_version.state = ObjectVersionState::Complete;
- object_version.size = next_offset as u64;
-
- let object = Object::new(bucket.into(), key.into(), vec![object_version]);
- garage.object_table.insert(&object).await?;
-
- Ok(version_uuid)
+ Ok(next_offset as u64)
}
async fn put_block_meta(
@@ -92,6 +110,7 @@ async fn put_block_meta(
part_number: u64,
offset: u64,
hash: Hash,
+ size: u64,
) -> Result<(), Error> {
// TODO: don't clone, restart from empty block list ??
let mut version = version.clone();
@@ -100,6 +119,7 @@ async fn put_block_meta(
part_number,
offset,
hash,
+ size,
})
.unwrap();
@@ -154,6 +174,184 @@ impl BodyChunker {
}
}
+fn put_response(version_uuid: UUID) -> Response<BodyType> {
+ let resp_bytes = format!("{}\n", hex::encode(version_uuid));
+ Response::new(Box::new(BytesBody::from(resp_bytes)))
+}
+
+pub async fn handle_create_multipart_upload(
+ garage: Arc<Garage>,
+ req: &Request<Body>,
+ bucket: &str,
+ key: &str,
+) -> Result<Response<BodyType>, Error> {
+ let version_uuid = gen_uuid();
+ let mime_type = get_mime_type(req)?;
+
+ let object_version = ObjectVersion {
+ uuid: version_uuid,
+ timestamp: now_msec(),
+ mime_type,
+ size: 0,
+ state: ObjectVersionState::Uploading,
+ data: ObjectVersionData::Uploading,
+ };
+ let object = Object::new(bucket.to_string(), key.to_string(), vec![object_version]);
+ garage.object_table.insert(&object).await?;
+
+ let mut xml = String::new();
+ writeln!(&mut xml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
+ writeln!(
+ &mut xml,
+ r#"<InitiateMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">"#
+ )
+ .unwrap();
+ writeln!(&mut xml, "\t<Bucket>{}</Bucket>", bucket).unwrap();
+ writeln!(&mut xml, "\t<Key>{}</Key>", xml_escape(key)).unwrap();
+ writeln!(
+ &mut xml,
+ "\t<UploadId>{}</UploadId>",
+ hex::encode(version_uuid)
+ )
+ .unwrap();
+ writeln!(&mut xml, "</InitiateMultipartUploadResult>").unwrap();
+
+ Ok(Response::new(Box::new(BytesBody::from(xml.into_bytes()))))
+}
+
+pub async fn handle_put_part(
+ garage: Arc<Garage>,
+ req: Request<Body>,
+ bucket: &str,
+ key: &str,
+ part_number_str: &str,
+ upload_id: &str,
+) -> Result<Response<BodyType>, Error> {
+ // Check parameters
+ let part_number = part_number_str
+ .parse::<u64>()
+ .map_err(|e| Error::BadRequest(format!("Invalid part number: {}", e)))?;
+
+ let version_uuid = uuid_from_str(upload_id).map_err(|_| Error::BadRequest(format!("Invalid upload ID")))?;
+
+ // Read first chuck, and at the same time try to get object to see if it exists
+ let mut chunker = BodyChunker::new(req.into_body(), garage.config.block_size);
+
+ let bucket = bucket.to_string();
+ let key = key.to_string();
+ let get_object_fut = garage.object_table.get(&bucket, &key);
+ let get_first_block_fut = chunker.next();
+ let (object, first_block) = futures::try_join!(get_object_fut, get_first_block_fut)?;
+
+ // Check object is valid and multipart block can be accepted
+ let first_block = match first_block {
+ None => return Err(Error::BadRequest(format!("Empty body"))),
+ Some(x) => x,
+ };
+ let object = match object {
+ None => return Err(Error::BadRequest(format!("Object not found"))),
+ Some(x) => x,
+ };
+ if !object.versions().iter().any(|v| {
+ v.uuid == version_uuid
+ && v.state == ObjectVersionState::Uploading
+ && v.data == ObjectVersionData::Uploading
+ }) {
+ return Err(Error::BadRequest(format!(
+ "Multipart upload does not exist or is otherwise invalid"
+ )));
+ }
+
+ // Copy block to store
+ let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]);
+ let first_block_hash = hash(&first_block[..]);
+ read_and_put_blocks(&garage, version, part_number, first_block, first_block_hash, &mut chunker).await?;
+
+ Ok(Response::new(Box::new(BytesBody::from(vec![]))))
+}
+
+pub async fn handle_complete_multipart_upload(
+ garage: Arc<Garage>,
+ _req: Request<Body>,
+ bucket: &str,
+ key: &str,
+ upload_id: &str,
+) -> Result<Response<BodyType>, Error> {
+ let version_uuid = uuid_from_str(upload_id).map_err(|_| Error::BadRequest(format!("Invalid upload ID")))?;
+
+ let bucket = bucket.to_string();
+ let key = key.to_string();
+ let (object, version) = futures::try_join!(
+ garage.object_table.get(&bucket, &key),
+ garage.version_table.get(&version_uuid, &EmptyKey),
+ )?;
+ let object = match object {
+ None => return Err(Error::BadRequest(format!("Object not found"))),
+ Some(x) => x,
+ };
+ let object_version = object.versions().iter().find(|v| {
+ v.uuid == version_uuid
+ && v.state == ObjectVersionState::Uploading
+ && v.data == ObjectVersionData::Uploading
+ });
+ let mut object_version = match object_version {
+ None => return Err(Error::BadRequest(format!(
+ "Multipart upload does not exist or has already been completed"
+ ))),
+ Some(x) => x.clone(),
+ };
+ let version = match version {
+ None => return Err(Error::BadRequest(format!("Version not found"))),
+ Some(x) => x,
+ };
+ if version.blocks().len() == 0 {
+ return Err(Error::BadRequest(format!("No data was uploaded")));
+ }
+
+ // TODO: check that all the parts that they pretend they gave us are indeed there
+ // TODO: check MD5 sum of all uploaded parts? but that would mean we have to store them somewhere...
+
+ let total_size = version.blocks().iter().map(|x| x.size).fold(0, |x, y| x+y);
+ object_version.size = total_size;
+ object_version.state = ObjectVersionState::Complete;
+ object_version.data = ObjectVersionData::FirstBlock(version.blocks()[0].hash);
+ let final_object = Object::new(bucket.clone(), key.clone(), vec![object_version]);
+ garage.object_table.insert(&final_object).await?;
+
+ let mut xml = String::new();
+ writeln!(&mut xml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
+ writeln!(
+ &mut xml,
+ r#"<CompleteMultipartUploadResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">"#
+ )
+ .unwrap();
+ writeln!(&mut xml, "\t<Location>{}</Location>", garage.config.s3_api.s3_region).unwrap();
+ writeln!(&mut xml, "\t<Bucket>{}</Bucket>", bucket).unwrap();
+ writeln!(&mut xml, "\t<Key>{}</Key>", xml_escape(&key)).unwrap();
+ writeln!(&mut xml, "</CompleteMultipartUploadResult>").unwrap();
+
+ Ok(Response::new(Box::new(BytesBody::from(xml.into_bytes()))))
+}
+
+fn get_mime_type(req: &Request<Body>) -> Result<String, Error> {
+ Ok(req
+ .headers()
+ .get(hyper::header::CONTENT_TYPE)
+ .map(|x| x.to_str())
+ .unwrap_or(Ok("blob"))?
+ .to_string())
+}
+
+fn uuid_from_str(id: &str) -> Result<UUID, ()> {
+ let id_bin = hex::decode(id).map_err(|_| ())?;
+ if id_bin.len() != 32 {
+ return Err(());
+ }
+ let mut uuid = [0u8; 32];
+ uuid.copy_from_slice(&id_bin[..]);
+ Ok(UUID::from(uuid))
+}
+
pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<UUID, Error> {
let object = match garage
.object_table