diff options
| -rw-r--r-- | src/api/api_server.rs | 342 | ||||
| -rw-r--r-- | src/api/lib.rs | 4 | ||||
| -rw-r--r-- | src/api/s3_get.rs | 131 | ||||
| -rw-r--r-- | src/api/s3_list.rs | 112 | ||||
| -rw-r--r-- | src/api/s3_put.rs | 190 | ||||
| -rw-r--r-- | src/core/object_table.rs | 8 |
6 files changed, 482 insertions, 305 deletions
diff --git a/src/api/api_server.rs b/src/api/api_server.rs index 97cc6b93..f7ca156d 100644 --- a/src/api/api_server.rs +++ b/src/api/api_server.rs @@ -1,30 +1,25 @@ -use std::collections::VecDeque; +use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::time::{Duration, UNIX_EPOCH}; use futures::future::Future; -use futures::stream::*; use hyper::body::{Bytes, HttpBody}; use hyper::server::conn::AddrStream; use hyper::service::{make_service_fn, service_fn}; -use hyper::{Body, Method, Request, Response, Server, StatusCode}; +use hyper::{Body, Method, Request, Response, Server}; -use garage_util::data::*; use garage_util::error::Error; -use garage_table::EmptyKey; - -use garage_core::block::INLINE_THRESHOLD; -use garage_core::block_ref_table::*; use garage_core::garage::Garage; -use garage_core::object_table::*; -use garage_core::version_table::*; use crate::http_util::*; use crate::signature::check_signature; -type BodyType = Box<dyn HttpBody<Data = Bytes, Error = Error> + Send + Unpin>; +use crate::s3_get::{handle_get, handle_head}; +use crate::s3_list::handle_list; +use crate::s3_put::{handle_delete, handle_put}; + +pub type BodyType = Box<dyn HttpBody<Data = Bytes, Error = Error> + Send + Unpin>; pub async fn run_api_server( garage: Arc<Garage>, @@ -128,298 +123,41 @@ async fn handler_inner( _ => Err(Error::BadRequest(format!("Invalid method"))), } } else { - // TODO: listing buckets - Err(Error::Forbidden("Unimplemented".into())) - } -} - -async fn handle_put( - garage: Arc<Garage>, - mime_type: &str, - bucket: &str, - key: &str, - body: Body, -) -> Result<UUID, Error> { - let version_uuid = gen_uuid(); - - let mut chunker = BodyChunker::new(body, garage.config.block_size); - let first_block = match chunker.next().await? { - Some(x) => x, - None => return Err(Error::BadRequest(format!("Empty body"))), - }; - - let mut object_version = ObjectVersion { - uuid: version_uuid, - timestamp: now_msec(), - mime_type: mime_type.to_string(), - size: first_block.len() as u64, - is_complete: false, - data: ObjectVersionData::DeleteMarker, - }; - - if first_block.len() < INLINE_THRESHOLD { - object_version.data = ObjectVersionData::Inline(first_block); - object_version.is_complete = true; - - let object = Object::new(bucket.into(), key.into(), vec![object_version]); - garage.object_table.insert(&object).await?; - return Ok(version_uuid); - } - - let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]); - - let first_block_hash = hash(&first_block[..]); - object_version.data = ObjectVersionData::FirstBlock(first_block_hash); - let object = Object::new(bucket.into(), key.into(), vec![object_version.clone()]); - garage.object_table.insert(&object).await?; - - let mut next_offset = first_block.len(); - let mut put_curr_version_block = put_block_meta(garage.clone(), &version, 0, first_block_hash); - let mut put_curr_block = garage - .block_manager - .rpc_put_block(first_block_hash, first_block); - - loop { - let (_, _, next_block) = - futures::try_join!(put_curr_block, put_curr_version_block, chunker.next())?; - if let Some(block) = next_block { - let block_hash = hash(&block[..]); - let block_len = block.len(); - put_curr_version_block = - put_block_meta(garage.clone(), &version, next_offset as u64, block_hash); - put_curr_block = garage.block_manager.rpc_put_block(block_hash, block); - next_offset += block_len; - } else { - break; - } - } - - // TODO: if at any step we have an error, we should undo everything we did - - object_version.is_complete = true; - 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) -} - -async fn put_block_meta( - garage: Arc<Garage>, - version: &Version, - offset: u64, - hash: Hash, -) -> Result<(), Error> { - // TODO: don't clone, restart from empty block list ?? - let mut version = version.clone(); - version.add_block(VersionBlock { offset, hash }).unwrap(); - - let block_ref = BlockRef { - block: hash, - version: version.uuid, - deleted: false, - }; - - futures::try_join!( - garage.version_table.insert(&version), - garage.block_ref_table.insert(&block_ref), - )?; - Ok(()) -} - -struct BodyChunker { - body: Body, - read_all: bool, - block_size: usize, - buf: VecDeque<u8>, -} - -impl BodyChunker { - fn new(body: Body, block_size: usize) -> Self { - Self { - body, - read_all: false, - block_size, - buf: VecDeque::new(), - } - } - async fn next(&mut self) -> Result<Option<Vec<u8>>, Error> { - while !self.read_all && self.buf.len() < self.block_size { - if let Some(block) = self.body.next().await { - let bytes = block?; - trace!("Body next: {} bytes", bytes.len()); - self.buf.extend(&bytes[..]); - } else { - self.read_all = true; - } - } - if self.buf.len() == 0 { - Ok(None) - } else if self.buf.len() <= self.block_size { - let block = self.buf.drain(..).collect::<Vec<u8>>(); - Ok(Some(block)) - } else { - let block = self.buf.drain(..self.block_size).collect::<Vec<u8>>(); - Ok(Some(block)) - } - } -} - -async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<UUID, Error> { - let exists = match garage - .object_table - .get(&bucket.to_string(), &key.to_string()) - .await? - { - None => false, - Some(o) => { - let mut has_active_version = false; - for v in o.versions().iter() { - if v.data != ObjectVersionData::DeleteMarker { - has_active_version = true; - break; + match req.method() { + &Method::PUT | &Method::DELETE => Err(Error::Forbidden( + "Cannot create or delete buckets using S3 api, please talk to Garage directly" + .into(), + )), + &Method::GET => { + let mut params = HashMap::new(); + if let Some(query) = req.uri().query() { + let query_pairs = url::form_urlencoded::parse(query.as_bytes()); + for (key, val) in query_pairs { + params.insert(key.to_lowercase(), val.to_string()); + } + } + if ["delimiter", "prefix"] + .iter() + .all(|x| params.contains_key(&x.to_string())) + { + let delimiter = params.get("delimiter").unwrap(); + let max_keys = params + .get("max-keys") + .map(|x| { + x.parse::<usize>().map_err(|e| { + Error::BadRequest(format!("Invalid value for max-keys: {}", e)) + }) + }) + .unwrap_or(Ok(1000))?; + let prefix = params.get("prefix").unwrap(); + Ok(handle_list(garage, bucket, delimiter, max_keys, prefix).await?) + } else { + Err(Error::BadRequest(format!( + "Not a list call, so what is it?" + ))) } } - has_active_version - } - }; - - if !exists { - // No need to delete - return Ok([0u8; 32].into()); - } - - let version_uuid = gen_uuid(); - - let object = Object::new( - bucket.into(), - key.into(), - vec![ObjectVersion { - uuid: version_uuid, - timestamp: now_msec(), - mime_type: "application/x-delete-marker".into(), - size: 0, - is_complete: true, - data: ObjectVersionData::DeleteMarker, - }], - ); - - garage.object_table.insert(&object).await?; - return Ok(version_uuid); -} - -fn object_headers(version: &ObjectVersion) -> http::response::Builder { - let date = UNIX_EPOCH + Duration::from_millis(version.timestamp); - let date_str = httpdate::fmt_http_date(date); - - Response::builder() - .header("Content-Type", version.mime_type.to_string()) - .header("Content-Length", format!("{}", version.size)) - .header("Last-Modified", date_str) -} - -async fn handle_head( - garage: Arc<Garage>, - bucket: &str, - key: &str, -) -> Result<Response<BodyType>, Error> { - let object = match garage - .object_table - .get(&bucket.to_string(), &key.to_string()) - .await? - { - None => return Err(Error::NotFound), - Some(o) => o, - }; - - let version = match object - .versions() - .iter() - .rev() - .filter(|v| v.is_complete && v.data != ObjectVersionData::DeleteMarker) - .next() - { - Some(v) => v, - None => return Err(Error::NotFound), - }; - - let body: BodyType = Box::new(BytesBody::from(vec![])); - let response = object_headers(&version) - .status(StatusCode::OK) - .body(body) - .unwrap(); - Ok(response) -} - -async fn handle_get( - garage: Arc<Garage>, - bucket: &str, - key: &str, -) -> Result<Response<BodyType>, Error> { - let object = match garage - .object_table - .get(&bucket.to_string(), &key.to_string()) - .await? - { - None => return Err(Error::NotFound), - Some(o) => o, - }; - - let last_v = match object - .versions() - .iter() - .rev() - .filter(|v| v.is_complete) - .next() - { - Some(v) => v, - None => return Err(Error::NotFound), - }; - - let resp_builder = object_headers(&last_v).status(StatusCode::OK); - - match &last_v.data { - ObjectVersionData::DeleteMarker => Err(Error::NotFound), - ObjectVersionData::Inline(bytes) => { - let body: BodyType = Box::new(BytesBody::from(bytes.to_vec())); - Ok(resp_builder.body(body)?) - } - ObjectVersionData::FirstBlock(first_block_hash) => { - let read_first_block = garage.block_manager.rpc_get_block(&first_block_hash); - let get_next_blocks = garage.version_table.get(&last_v.uuid, &EmptyKey); - - let (first_block, version) = futures::try_join!(read_first_block, get_next_blocks)?; - let version = match version { - Some(v) => v, - None => return Err(Error::NotFound), - }; - - let mut blocks = version - .blocks() - .iter() - .map(|vb| (vb.hash, None)) - .collect::<Vec<_>>(); - blocks[0].1 = Some(first_block); - - let body_stream = futures::stream::iter(blocks) - .map(move |(hash, data_opt)| { - let garage = garage.clone(); - async move { - if let Some(data) = data_opt { - Ok(Bytes::from(data)) - } else { - garage - .block_manager - .rpc_get_block(&hash) - .await - .map(Bytes::from) - } - } - }) - .buffered(2); - let body: BodyType = Box::new(StreamBody::new(Box::pin(body_stream))); - Ok(resp_builder.body(body)?) + _ => Err(Error::BadRequest(format!("Invalid method"))), } } } diff --git a/src/api/lib.rs b/src/api/lib.rs index 3bcd02f1..da537222 100644 --- a/src/api/lib.rs +++ b/src/api/lib.rs @@ -4,3 +4,7 @@ extern crate log; pub mod api_server; pub mod http_util; pub mod signature; + +pub mod s3_get; +pub mod s3_list; +pub mod s3_put; diff --git a/src/api/s3_get.rs b/src/api/s3_get.rs new file mode 100644 index 00000000..b933f549 --- /dev/null +++ b/src/api/s3_get.rs @@ -0,0 +1,131 @@ +use std::sync::Arc; +use std::time::{Duration, UNIX_EPOCH}; + +use futures::stream::*; +use hyper::body::Bytes; +use hyper::{Response, StatusCode}; + +use garage_util::error::Error; + +use garage_table::EmptyKey; + +use garage_core::garage::Garage; +use garage_core::object_table::*; + +use crate::api_server::BodyType; +use crate::http_util::*; + +fn object_headers(version: &ObjectVersion) -> http::response::Builder { + let date = UNIX_EPOCH + Duration::from_millis(version.timestamp); + let date_str = httpdate::fmt_http_date(date); + + Response::builder() + .header("Content-Type", version.mime_type.to_string()) + .header("Content-Length", format!("{}", version.size)) + .header("Last-Modified", date_str) +} + +pub async fn handle_head( + garage: Arc<Garage>, + bucket: &str, + key: &str, +) -> Result<Response<BodyType>, Error> { + let object = match garage + .object_table + .get(&bucket.to_string(), &key.to_string()) + .await? + { + None => return Err(Error::NotFound), + Some(o) => o, + }; + + let version = match object + .versions() + .iter() + .rev() + .filter(|v| v.is_complete && v.data != ObjectVersionData::DeleteMarker) + .next() + { + Some(v) => v, + None => return Err(Error::NotFound), + }; + + let body: BodyType = Box::new(BytesBody::from(vec![])); + let response = object_headers(&version) + .status(StatusCode::OK) + .body(body) + .unwrap(); + Ok(response) +} + +pub async fn handle_get( + garage: Arc<Garage>, + bucket: &str, + key: &str, +) -> Result<Response<BodyType>, Error> { + let object = match garage + .object_table + .get(&bucket.to_string(), &key.to_string()) + .await? + { + None => return Err(Error::NotFound), + Some(o) => o, + }; + + let last_v = match object + .versions() + .iter() + .rev() + .filter(|v| v.is_complete) + .next() + { + Some(v) => v, + None => return Err(Error::NotFound), + }; + + let resp_builder = object_headers(&last_v).status(StatusCode::OK); + + match &last_v.data { + ObjectVersionData::DeleteMarker => Err(Error::NotFound), + ObjectVersionData::Inline(bytes) => { + let body: BodyType = Box::new(BytesBody::from(bytes.to_vec())); + Ok(resp_builder.body(body)?) + } + ObjectVersionData::FirstBlock(first_block_hash) => { + let read_first_block = garage.block_manager.rpc_get_block(&first_block_hash); + let get_next_blocks = garage.version_table.get(&last_v.uuid, &EmptyKey); + + let (first_block, version) = futures::try_join!(read_first_block, get_next_blocks)?; + let version = match version { + Some(v) => v, + None => return Err(Error::NotFound), + }; + + let mut blocks = version + .blocks() + .iter() + .map(|vb| (vb.hash, None)) + .collect::<Vec<_>>(); + blocks[0].1 = Some(first_block); + + let body_stream = futures::stream::iter(blocks) + .map(move |(hash, data_opt)| { + let garage = garage.clone(); + async move { + if let Some(data) = data_opt { + Ok(Bytes::from(data)) + } else { + garage + .block_manager + .rpc_get_block(&hash) + .await + .map(Bytes::from) + } + } + }) + .buffered(2); + let body: BodyType = Box::new(StreamBody::new(Box::pin(body_stream))); + Ok(resp_builder.body(body)?) + } + } +} diff --git a/src/api/s3_list.rs b/src/api/s3_list.rs new file mode 100644 index 00000000..6004bff0 --- /dev/null +++ b/src/api/s3_list.rs @@ -0,0 +1,112 @@ +use std::collections::BTreeMap; +use std::fmt::Write; +use std::sync::Arc; + +use chrono::{DateTime, NaiveDateTime, Utc, SecondsFormat}; +use hyper::Response; + +use garage_util::error::Error; + +use garage_core::garage::Garage; +use garage_core::object_table::*; + +use crate::api_server::BodyType; +use crate::http_util::*; + +#[derive(Debug)] +struct ListResultInfo { + last_modified: u64, + size: u64, +} + +pub async fn handle_list( + garage: Arc<Garage>, + bucket: &str, + delimiter: &str, + max_keys: usize, + prefix: &str, +) -> Result<Response<BodyType>, Error> { + let mut result = BTreeMap::<String, ListResultInfo>::new(); + let mut truncated = true; + let mut next_chunk_start = prefix.to_string(); + + println!("List request: `{}` {} `{}`", delimiter, max_keys, prefix); + + while result.len() < max_keys && truncated { + let objects = garage + .object_table + .get_range( + &bucket.to_string(), + Some(next_chunk_start.clone()), + Some(()), + max_keys, + ) + .await?; + for object in objects.iter() { + if let Some(version) = object + .versions() + .iter() + .find(|x| x.is_complete && x.data != ObjectVersionData::DeleteMarker) + { + let relative_key = match object.key.starts_with(prefix) { + true => &object.key[prefix.len()..], + false => { + truncated = false; + break; + } + }; + let delimited_key = match relative_key.find(delimiter) { + Some(i) => relative_key.split_at(i).1, + None => &relative_key, + }; + let delimited_key = delimited_key.to_string(); + let new_info = match result.get(&delimited_key) { + None => ListResultInfo { + last_modified: version.timestamp, + size: version.size, + }, + Some(lri) => ListResultInfo { + last_modified: std::cmp::max(version.timestamp, lri.last_modified), + size: 0, + }, + }; + println!("Entry: {} {:?}", delimited_key, new_info); + result.insert(delimited_key, new_info); + } + } + if objects.len() < max_keys { + truncated = false; + } + if objects.len() > 0 { + next_chunk_start = objects[objects.len() - 1].key.clone(); + } + } + + let mut xml = String::new(); + writeln!(&mut xml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap(); + writeln!( + &mut xml, + r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">"# + ) + .unwrap(); + writeln!(&mut xml, "\t<Bucket>{}</Bucket>", bucket).unwrap(); + writeln!(&mut xml, "\t<Prefix>{}</Prefix>", prefix).unwrap(); + writeln!(&mut xml, "\t<KeyCount>{}</KeyCount>", result.len()).unwrap(); + writeln!(&mut xml, "\t<MaxKeys>{}</MaxKeys>", max_keys).unwrap(); + writeln!(&mut xml, "\t<IsTruncated>{}</IsTruncated>", truncated).unwrap(); + for (key, info) in result.iter() { + let last_modif = NaiveDateTime::from_timestamp(info.last_modified as i64 / 1000, 0); + let last_modif = DateTime::<Utc>::from_utc(last_modif, Utc); + let last_modif = last_modif.to_rfc3339_opts(SecondsFormat::Millis, true); + writeln!(&mut xml, "\t<Contents>").unwrap(); + writeln!(&mut xml, "\t\t<Key>{}</Key>", key).unwrap(); + writeln!(&mut xml, "\t\t<LastModified>{}</LastModified>", last_modif).unwrap(); + writeln!(&mut xml, "\t\t<Size>{}</Size>", info.size).unwrap(); + writeln!(&mut xml, "\t\t<StorageClass>STANDARD</StorageClass>").unwrap(); + writeln!(&mut xml, "\t</Contents>").unwrap(); + } + writeln!(&mut xml, "</ListBucketResult>").unwrap(); + println!("{}", xml); + + Ok(Response::new(Box::new(BytesBody::from(xml.into_bytes())))) +} diff --git a/src/api/s3_put.rs b/src/api/s3_put.rs new file mode 100644 index 00000000..15c2ccb7 --- /dev/null +++ b/src/api/s3_put.rs @@ -0,0 +1,190 @@ +use std::collections::VecDeque; +use std::sync::Arc; + +use futures::stream::*; +use hyper::Body; + +use garage_util::data::*; +use garage_util::error::Error; + +use garage_core::block::INLINE_THRESHOLD; +use garage_core::block_ref_table::*; +use garage_core::garage::Garage; +use garage_core::object_table::*; +use garage_core::version_table::*; + +pub async fn handle_put( + garage: Arc<Garage>, + mime_type: &str, + bucket: &str, + key: &str, + body: Body, +) -> Result<UUID, Error> { + let version_uuid = gen_uuid(); + + let mut chunker = BodyChunker::new(body, garage.config.block_size); + let first_block = match chunker.next().await? { + Some(x) => x, + None => return Err(Error::BadRequest(format!("Empty body"))), + }; + + let mut object_version = ObjectVersion { + uuid: version_uuid, + timestamp: now_msec(), + mime_type: mime_type.to_string(), + size: first_block.len() as u64, + is_complete: false, + data: ObjectVersionData::DeleteMarker, + }; + + if first_block.len() < INLINE_THRESHOLD { + object_version.data = ObjectVersionData::Inline(first_block); + object_version.is_complete = true; + + let object = Object::new(bucket.into(), key.into(), vec![object_version]); + garage.object_table.insert(&object).await?; + return Ok(version_uuid); + } + + let version = Version::new(version_uuid, bucket.into(), key.into(), false, vec![]); + + let first_block_hash = hash(&first_block[..]); + object_version.data = ObjectVersionData::FirstBlock(first_block_hash); + let object = Object::new(bucket.into(), key.into(), vec![object_version.clone()]); + garage.object_table.insert(&object).await?; + + let mut next_offset = first_block.len(); + let mut put_curr_version_block = put_block_meta(garage.clone(), &version, 0, first_block_hash); + let mut put_curr_block = garage + .block_manager + .rpc_put_block(first_block_hash, first_block); + + loop { + let (_, _, next_block) = + futures::try_join!(put_curr_block, put_curr_version_block, chunker.next())?; + if let Some(block) = next_block { + let block_hash = hash(&block[..]); + let block_len = block.len(); + put_curr_version_block = + put_block_meta(garage.clone(), &version, next_offset as u64, block_hash); + put_curr_block = garage.block_manager.rpc_put_block(block_hash, block); + next_offset += block_len; + } else { + break; + } + } + + // TODO: if at any step we have an error, we should undo everything we did + + object_version.is_complete = true; + 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) +} + +async fn put_block_meta( + garage: Arc<Garage>, + version: &Version, + offset: u64, + hash: Hash, +) -> Result<(), Error> { + // TODO: don't clone, restart from empty block list ?? + let mut version = version.clone(); + version.add_block(VersionBlock { offset, hash }).unwrap(); + + let block_ref = BlockRef { + block: hash, + version: version.uuid, + deleted: false, + }; + + futures::try_join!( + garage.version_table.insert(&version), + garage.block_ref_table.insert(&block_ref), + )?; + Ok(()) +} + +struct BodyChunker { + body: Body, + read_all: bool, + block_size: usize, + buf: VecDeque<u8>, +} + +impl BodyChunker { + fn new(body: Body, block_size: usize) -> Self { + Self { + body, + read_all: false, + block_size, + buf: VecDeque::new(), + } + } + async fn next(&mut self) -> Result<Option<Vec<u8>>, Error> { + while !self.read_all && self.buf.len() < self.block_size { + if let Some(block) = self.body.next().await { + let bytes = block?; + trace!("Body next: {} bytes", bytes.len()); + self.buf.extend(&bytes[..]); + } else { + self.read_all = true; + } + } + if self.buf.len() == 0 { + Ok(None) + } else if self.buf.len() <= self.block_size { + let block = self.buf.drain(..).collect::<Vec<u8>>(); + Ok(Some(block)) + } else { + let block = self.buf.drain(..self.block_size).collect::<Vec<u8>>(); + Ok(Some(block)) + } + } +} + +pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<UUID, Error> { + let exists = match garage + .object_table + .get(&bucket.to_string(), &key.to_string()) + .await? + { + None => false, + Some(o) => { + let mut has_active_version = false; + for v in o.versions().iter() { + if v.data != ObjectVersionData::DeleteMarker { + has_active_version = true; + break; + } + } + has_active_version + } + }; + + if !exists { + // No need to delete + return Ok([0u8; 32].into()); + } + + let version_uuid = gen_uuid(); + + let object = Object::new( + bucket.into(), + key.into(), + vec![ObjectVersion { + uuid: version_uuid, + timestamp: now_msec(), + mime_type: "application/x-delete-marker".into(), + size: 0, + is_complete: true, + data: ObjectVersionData::DeleteMarker, + }], + ); + + garage.object_table.insert(&object).await?; + return Ok(version_uuid); +} diff --git a/src/core/object_table.rs b/src/core/object_table.rs index 1fe2b3d4..e4c78524 100644 --- a/src/core/object_table.rs +++ b/src/core/object_table.rs @@ -158,8 +158,10 @@ impl TableSchema for ObjectTable { Ok(()) } - fn matches_filter(_entry: &Self::E, _filter: &Self::Filter) -> bool { - // TODO - true + fn matches_filter(entry: &Self::E, _filter: &Self::Filter) -> bool { + entry + .versions + .iter() + .any(|x| x.is_complete && x.data != ObjectVersionData::DeleteMarker) } } |