Implement HTTP ranges in get

author: Alex Auvolat <alex@adnab.me> 2020-05-04 13:09:23 +0000
committer: Alex Auvolat <alex@adnab.me> 2020-05-04 13:09:23 +0000
commit: b46a7788d160f0cec285d77664e09760d25e2144 (patch)
tree: 938c28c24390834218fee2320e6d58e8096eb7a0 /src
parent: 16fbb32fd3ac00f76937c2799d01e7607449fa94 (diff)
download: garage-b46a7788d160f0cec285d77664e09760d25e2144.tar.gz
garage-b46a7788d160f0cec285d77664e09760d25e2144.zip
6 files changed, 226 insertions, 103 deletions
diff --git a/src/api/Cargo.toml b/src/api/Cargo.toml
index b8674fba..56e0e2a8 100644
--- a/src/api/Cargo.toml
+++ b/src/api/Cargo.toml
@@ -32,3 +32,4 @@ url = "2.1"
 httpdate = "0.3"
 percent-encoding = "2.1.0"
 roxmltree = "0.11"
+http-range = "0.1"
diff --git a/src/api/api_server.rs b/src/api/api_server.rs
index 913ff0af..32506ccd 100644
--- a/src/api/api_server.rs
+++ b/src/api/api_server.rs
@@ -73,8 +73,7 @@ async fn handler_inner(
 	req: Request<Body>,
 ) -> Result<Response<BodyType>, Error> {
 	let path = req.uri().path().to_string();
-	let path = percent_encoding::percent_decode_str(&path)
-		.decode_utf8()?;
+	let path = percent_encoding::percent_decode_str(&path).decode_utf8()?;
 
 	let (bucket, key) = parse_bucket_key(&path)?;
 	if bucket.len() == 0 {
@@ -110,7 +109,7 @@ async fn handler_inner(
 			}
 			&Method::GET => {
 				// GetObject query
-				Ok(handle_get(garage, &bucket, &key).await?)
+				Ok(handle_get(garage, &req, &bucket, &key).await?)
 			}
 			&Method::PUT => {
 				if params.contains_key(&"partnumber".to_string())
@@ -123,8 +122,8 @@ async fn handler_inner(
 				} else if req.headers().contains_key("x-amz-copy-source") {
 					// CopyObject query
 					let copy_source = req.headers().get("x-amz-copy-source").unwrap().to_str()?;
-					let copy_source = percent_encoding::percent_decode_str(&copy_source)
-						.decode_utf8()?;
+					let copy_source =
+						percent_encoding::percent_decode_str(&copy_source).decode_utf8()?;
 					let (source_bucket, source_key) = parse_bucket_key(&copy_source)?;
 					if !api_key.allow_read(&source_bucket) {
 						return Err(Error::Forbidden(format!(
@@ -228,22 +227,20 @@ async fn handler_inner(
 				)
 				.await?)
 			}
-            &Method::POST => {
-                if params.contains_key(&"delete".to_string()) {
-                    // DeleteObjects
-                    Ok(handle_delete_objects(garage, bucket, req).await?)
-                } else {
-                    println!(
-                        "Body: {}",
-                        std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
-                            .unwrap_or("<invalid utf8>")
-                    );
-                    Err(Error::BadRequest(format!("Unsupported call")))
-                }
-            }
-			_ => {
-                Err(Error::BadRequest(format!("Invalid method")))
-            }
+			&Method::POST => {
+				if params.contains_key(&"delete".to_string()) {
+					// DeleteObjects
+					Ok(handle_delete_objects(garage, bucket, req).await?)
+				} else {
+					println!(
+						"Body: {}",
+						std::str::from_utf8(&hyper::body::to_bytes(req.into_body()).await?)
+							.unwrap_or("<invalid utf8>")
+					);
+					Err(Error::BadRequest(format!("Unsupported call")))
+				}
+			}
+			_ => Err(Error::BadRequest(format!("Invalid method"))),
 		}
 	}
 }
diff --git a/src/api/s3_delete.rs b/src/api/s3_delete.rs
index 001eb162..e77ab314 100644
--- a/src/api/s3_delete.rs
+++ b/src/api/s3_delete.rs
@@ -1,5 +1,5 @@
-use std::sync::Arc;
 use std::fmt::Write;
+use std::sync::Arc;
 
 use hyper::{Body, Request, Response};
 
@@ -9,10 +9,14 @@ use garage_util::error::Error;
 use garage_core::garage::Garage;
 use garage_core::object_table::*;
 
-use crate::http_util::*;
 use crate::encoding::*;
+use crate::http_util::*;
 
-async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Result<(UUID, UUID), Error> {
+async fn handle_delete_internal(
+	garage: &Garage,
+	bucket: &str,
+	key: &str,
+) -> Result<(UUID, UUID), Error> {
 	let object = match garage
 		.object_table
 		.get(&bucket.to_string(), &key.to_string())
@@ -32,16 +36,16 @@ async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Res
 	let mut must_delete = None;
 	let mut timestamp = now_msec();
 	for v in interesting_versions {
-        if v.timestamp + 1 > timestamp || must_delete.is_none() {
-		    must_delete = Some(v.uuid);
-        }
+		if v.timestamp + 1 > timestamp || must_delete.is_none() {
+			must_delete = Some(v.uuid);
+		}
 		timestamp = std::cmp::max(timestamp, v.timestamp + 1);
 	}
 
-    let deleted_version = match must_delete {
-        None => return Err(Error::NotFound),
-        Some(v) => v,
-    };
+	let deleted_version = match must_delete {
+		None => return Err(Error::NotFound),
+		Some(v) => v,
+	};
 
 	let version_uuid = gen_uuid();
 
@@ -62,8 +66,13 @@ async fn handle_delete_internal(garage: &Garage, bucket: &str, key: &str) -> Res
 	return Ok((deleted_version, version_uuid));
 }
 
-pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Result<Response<BodyType>, Error> {
-    let (_deleted_version, delete_marker_version) = handle_delete_internal(&garage, bucket, key).await?;
+pub async fn handle_delete(
+	garage: Arc<Garage>,
+	bucket: &str,
+	key: &str,
+) -> Result<Response<BodyType>, Error> {
+	let (_deleted_version, delete_marker_version) =
+		handle_delete_internal(&garage, bucket, key).await?;
 
 	Ok(Response::builder()
 		.header("x-amz-version-id", hex::encode(delete_marker_version))
@@ -71,76 +80,98 @@ pub async fn handle_delete(garage: Arc<Garage>, bucket: &str, key: &str) -> Resu
 		.unwrap())
 }
 
-pub async fn handle_delete_objects(garage: Arc<Garage>, bucket: &str, req: Request<Body>) -> Result<Response<BodyType>, Error> {
-    let body = hyper::body::to_bytes(req.into_body()).await?;
-    let cmd_xml = roxmltree::Document::parse(&std::str::from_utf8(&body)?)?;
-    let cmd = parse_delete_objects_xml(&cmd_xml)
-        .map_err(|e| Error::BadRequest(format!("Invald delete XML query: {}", e)))?;
-
-    let mut retxml = String::new();
+pub async fn handle_delete_objects(
+	garage: Arc<Garage>,
+	bucket: &str,
+	req: Request<Body>,
+) -> Result<Response<BodyType>, Error> {
+	let body = hyper::body::to_bytes(req.into_body()).await?;
+	let cmd_xml = roxmltree::Document::parse(&std::str::from_utf8(&body)?)?;
+	let cmd = parse_delete_objects_xml(&cmd_xml)
+		.map_err(|e| Error::BadRequest(format!("Invald delete XML query: {}", e)))?;
+
+	let mut retxml = String::new();
 	writeln!(&mut retxml, r#"<?xml version="1.0" encoding="UTF-8"?>"#).unwrap();
 	writeln!(&mut retxml, "<DeleteObjectsOutput>").unwrap();
 
-    for obj in cmd.objects.iter() {
-        match handle_delete_internal(&garage, bucket, &obj.key).await {
-            Ok((deleted_version, delete_marker_version)) => {
-                writeln!(&mut retxml, "\t<Deleted>").unwrap();
-                writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
-                writeln!(&mut retxml, "\t\t<VersionId>{}</VersionId>", hex::encode(deleted_version)).unwrap();
-                writeln!(&mut retxml, "\t\t<DeleteMarkerVersionId>{}</DeleteMarkerVersionId>", hex::encode(delete_marker_version)).unwrap();
-                writeln!(&mut retxml, "\t</Deleted>").unwrap();
-            }
-            Err(e) => {
-                writeln!(&mut retxml, "\t<Error>").unwrap();
-                writeln!(&mut retxml, "\t\t<Code>{}</Code>", e.http_status_code()).unwrap();
-                writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
-                writeln!(&mut retxml, "\t\t<Message>{}</Message>", xml_escape(&format!("{}", e))).unwrap();
-                writeln!(&mut retxml, "\t</Error>").unwrap();
-            }
-        }
-    }
+	for obj in cmd.objects.iter() {
+		match handle_delete_internal(&garage, bucket, &obj.key).await {
+			Ok((deleted_version, delete_marker_version)) => {
+				writeln!(&mut retxml, "\t<Deleted>").unwrap();
+				writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+				writeln!(
+					&mut retxml,
+					"\t\t<VersionId>{}</VersionId>",
+					hex::encode(deleted_version)
+				)
+				.unwrap();
+				writeln!(
+					&mut retxml,
+					"\t\t<DeleteMarkerVersionId>{}</DeleteMarkerVersionId>",
+					hex::encode(delete_marker_version)
+				)
+				.unwrap();
+				writeln!(&mut retxml, "\t</Deleted>").unwrap();
+			}
+			Err(e) => {
+				writeln!(&mut retxml, "\t<Error>").unwrap();
+				writeln!(&mut retxml, "\t\t<Code>{}</Code>", e.http_status_code()).unwrap();
+				writeln!(&mut retxml, "\t\t<Key>{}</Key>", obj.key).unwrap();
+				writeln!(
+					&mut retxml,
+					"\t\t<Message>{}</Message>",
+					xml_escape(&format!("{}", e))
+				)
+				.unwrap();
+				writeln!(&mut retxml, "\t</Error>").unwrap();
+			}
+		}
+	}
 
 	writeln!(&mut retxml, "</DeleteObjectsOutput>").unwrap();
 
-	Ok(Response::new(Box::new(BytesBody::from(retxml.into_bytes()))))
+	Ok(Response::new(Box::new(BytesBody::from(
+		retxml.into_bytes(),
+	))))
 }
 
 struct DeleteRequest {
-    objects: Vec<DeleteObject>,
+	objects: Vec<DeleteObject>,
 }
 
 struct DeleteObject {
-    key: String,
+	key: String,
 }
 
 fn parse_delete_objects_xml(xml: &roxmltree::Document) -> Result<DeleteRequest, String> {
-    let mut ret = DeleteRequest{objects: vec![]};
-
-    let root = xml.root();
-    let delete = match root.first_child() {
-        Some(del) => del,
-        None => return Err(format!("Delete tag not found")),
-    };
-    if !delete.has_tag_name("Delete") {
-        return Err(format!("Invalid root tag: {:?}", root));
-    }
-
-    for item in delete.children() {
-        if item.has_tag_name("Object") {
-            if let Some(key) = item.children().find(|e| e.has_tag_name("Key")) {
-                if let Some(key_str) = key.text() {
-                    ret.objects.push(DeleteObject{key: key_str.to_string()});
-                } else {
-                    return Err(format!("No text for key: {:?}", key));
-                }
-            } else {
-                return Err(format!("No delete key for item: {:?}", item));
-            }
-        } else {
-            return Err(format!("Invalid delete item: {:?}", item));
-        }
-    }
-
-    Ok(ret)
-}
+	let mut ret = DeleteRequest { objects: vec![] };
+
+	let root = xml.root();
+	let delete = match root.first_child() {
+		Some(del) => del,
+		None => return Err(format!("Delete tag not found")),
+	};
+	if !delete.has_tag_name("Delete") {
+		return Err(format!("Invalid root tag: {:?}", root));
+	}
 
+	for item in delete.children() {
+		if item.has_tag_name("Object") {
+			if let Some(key) = item.children().find(|e| e.has_tag_name("Key")) {
+				if let Some(key_str) = key.text() {
+					ret.objects.push(DeleteObject {
+						key: key_str.to_string(),
+					});
+				} else {
+					return Err(format!("No text for key: {:?}", key));
+				}
+			} else {
+				return Err(format!("No delete key for item: {:?}", item));
+			}
+		} else {
+			return Err(format!("Invalid delete item: {:?}", item));
+		}
+	}
+
+	Ok(ret)
+}
diff --git a/src/api/s3_get.rs b/src/api/s3_get.rs
index 75478e46..3ed0f914 100644
--- a/src/api/s3_get.rs
+++ b/src/api/s3_get.rs
@@ -3,7 +3,7 @@ use std::time::{Duration, UNIX_EPOCH};
 
 use futures::stream::*;
 use hyper::body::Bytes;
-use hyper::{Response, StatusCode};
+use hyper::{Body, Request, Response, StatusCode};
 
 use garage_util::error::Error;
 
@@ -22,6 +22,7 @@ fn object_headers(version: &ObjectVersion) -> http::response::Builder {
 		.header("Content-Type", version.mime_type.to_string())
 		.header("Content-Length", format!("{}", version.size))
 		.header("Last-Modified", date_str)
+		.header("Accept-Ranges", format!("bytes"))
 }
 
 pub async fn handle_head(
@@ -59,6 +60,7 @@ pub async fn handle_head(
 
 pub async fn handle_get(
 	garage: Arc<Garage>,
+	req: &Request<Body>,
 	bucket: &str,
 	key: &str,
 ) -> Result<Response<BodyType>, Error> {
@@ -82,6 +84,25 @@ pub async fn handle_get(
 		None => return Err(Error::NotFound),
 	};
 
+	let range = match req.headers().get("range") {
+		Some(range) => {
+			let range_str = range
+				.to_str()
+				.map_err(|e| Error::BadRequest(format!("Invalid range header: {}", e)))?;
+			let mut ranges = http_range::HttpRange::parse(range_str, last_v.size)
+				.map_err(|_e| Error::BadRequest(format!("Invalid range")))?;
+			if ranges.len() > 1 {
+				return Err(Error::BadRequest(format!("Multiple ranges not supported")));
+			} else {
+				ranges.pop()
+			}
+		}
+		None => None,
+	};
+	if let Some(range) = range {
+		return handle_get_range(garage, last_v, range.start, range.start + range.length).await;
+	}
+
 	let resp_builder = object_headers(&last_v).status(StatusCode::OK);
 
 	match &last_v.data {
@@ -131,3 +152,76 @@ pub async fn handle_get(
 		}
 	}
 }
+
+pub async fn handle_get_range(
+	garage: Arc<Garage>,
+	version: &ObjectVersion,
+	begin: u64,
+	end: u64,
+) -> Result<Response<BodyType>, Error> {
+	if end > version.size {
+		return Err(Error::BadRequest(format!("Range not included in file")));
+	}
+
+	let resp_builder = object_headers(&version)
+		.header(
+			"Content-Range",
+			format!("bytes {}-{}/{}", begin, end, version.size),
+		)
+		.status(StatusCode::PARTIAL_CONTENT);
+
+	match &version.data {
+		ObjectVersionData::Uploading => Err(Error::Message(format!(
+			"Version is_complete() but data is stil Uploading (internal error)"
+		))),
+		ObjectVersionData::DeleteMarker => Err(Error::NotFound),
+		ObjectVersionData::Inline(bytes) => {
+			if end as usize <= bytes.len() {
+				let body: BodyType = Box::new(BytesBody::from(
+					bytes[begin as usize..end as usize].to_vec(),
+				));
+				Ok(resp_builder.body(body)?)
+			} else {
+				Err(Error::Message(format!("Internal error: requested range not present in inline bytes when it should have been")))
+			}
+		}
+		ObjectVersionData::FirstBlock(_first_block_hash) => {
+			let version = garage.version_table.get(&version.uuid, &EmptyKey).await?;
+			let version = match version {
+				Some(v) => v,
+				None => return Err(Error::NotFound),
+			};
+
+			let blocks = version
+				.blocks()
+				.iter()
+				.cloned()
+				.filter(|block| block.offset + block.size > begin && block.offset < end)
+				.collect::<Vec<_>>();
+
+			let body_stream = futures::stream::iter(blocks)
+				.map(move |block| {
+					let garage = garage.clone();
+					async move {
+						let data = garage.block_manager.rpc_get_block(&block.hash).await?;
+						let start_in_block = if block.offset > begin {
+							0
+						} else {
+							begin - block.offset
+						};
+						let end_in_block = if block.offset + block.size < end {
+							block.size
+						} else {
+							end - block.offset
+						};
+						Ok(Bytes::from(
+							data[start_in_block as usize..end_in_block as usize].to_vec(),
+						))
+					}
+				})
+				.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
index d4d8161e..b8babbbf 100644
--- a/src/api/s3_list.rs
+++ b/src/api/s3_list.rs
@@ -58,10 +58,10 @@ pub async fn handle_list(
 				break 'query_loop;
 			}
 			if let Some(version) = object.versions().iter().find(|x| x.is_data()) {
-                if result_keys.len() + result_common_prefixes.len() >= max_keys {
-                    truncated = true;
-                    break 'query_loop;
-                }
+				if result_keys.len() + result_common_prefixes.len() >= max_keys {
+					truncated = true;
+					break 'query_loop;
+				}
 				let common_prefix = if delimiter.len() > 0 {
 					let relative_key = &object.key[prefix.len()..];
 					match relative_key.find(delimiter) {
@@ -88,8 +88,8 @@ pub async fn handle_list(
 			}
 		}
 		if objects.len() < max_keys + 1 {
-            truncated = false;
-            break 'query_loop;
+			truncated = false;
+			break 'query_loop;
 		}
 		if objects.len() > 0 {
 			next_chunk_start = objects[objects.len() - 1].key.clone();
diff --git a/src/util/error.rs b/src/util/error.rs
index d2ed1ccf..0ca1afe7 100644
--- a/src/util/error.rs
+++ b/src/util/error.rs
@@ -116,13 +116,13 @@ impl<T> From<tokio::sync::mpsc::error::SendError<T>> for Error {
 }
 
 impl From<std::str::Utf8Error> for Error {
-    fn from(e: std::str::Utf8Error) -> Error {
-        Error::BadRequest(format!("Invalid UTF-8: {}", e))
-    }
+	fn from(e: std::str::Utf8Error) -> Error {
+		Error::BadRequest(format!("Invalid UTF-8: {}", e))
+	}
 }
 
 impl From<roxmltree::Error> for Error {
-    fn from(e: roxmltree::Error) -> Error {
-        Error::BadRequest(format!("Invalid XML: {}", e))
-    }
+	fn from(e: roxmltree::Error) -> Error {
+		Error::BadRequest(format!("Invalid XML: {}", e))
+	}
 }
author	Alex Auvolat <alex@adnab.me>	2020-05-04 13:09:23 +0000
committer	Alex Auvolat <alex@adnab.me>	2020-05-04 13:09:23 +0000
commit	b46a7788d160f0cec285d77664e09760d25e2144 (patch)
tree	938c28c24390834218fee2320e6d58e8096eb7a0 /src
parent	16fbb32fd3ac00f76937c2799d01e7607449fa94 (diff)
download	garage-b46a7788d160f0cec285d77664e09760d25e2144.tar.gz garage-b46a7788d160f0cec285d77664e09760d25e2144.zip