Merge branch 'main' into next-0.10

author: Alex Auvolat <alex@adnab.me> 2024-02-13 11:24:56 +0100
committer: Alex Auvolat <alex@adnab.me> 2024-02-13 11:36:28 +0100
commit: cf2af186fcc0c8f581a966454b6cd4720d3821f0 (patch)
tree: 37a978ba9ffb780fc828cff7b8ec93662d50884f /src/api/admin
parent: db48dd3d6c1f9e86a62e9b8edfce2c1620bcd5f3 (diff)
parent: 823078b4cdaf93e09de0847c5eaa75beb7b26b7f (diff)
download: garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.tar.gz
garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.zip
5 files changed, 85 insertions, 83 deletions
diff --git a/src/api/admin/api_server.rs b/src/api/admin/api_server.rs
index 41a5e68c..2b9be24e 100644
--- a/src/api/admin/api_server.rs
+++ b/src/api/admin/api_server.rs
@@ -3,9 +3,9 @@ use std::sync::Arc;
 
 use async_trait::async_trait;
 
-use futures::future::Future;
 use http::header::{ACCESS_CONTROL_ALLOW_METHODS, ACCESS_CONTROL_ALLOW_ORIGIN, ALLOW};
-use hyper::{Body, Request, Response, StatusCode};
+use hyper::{body::Incoming as IncomingBody, Request, Response, StatusCode};
+use tokio::sync::watch;
 
 use opentelemetry::trace::SpanRef;
 
@@ -27,7 +27,9 @@ use crate::admin::error::*;
 use crate::admin::key::*;
 use crate::admin::router_v0;
 use crate::admin::router_v1::{Authorization, Endpoint};
-use crate::helpers::host_to_bucket;
+use crate::helpers::*;
+
+pub type ResBody = BoxBody<Error>;
 
 pub struct AdminApiServer {
 	garage: Arc<Garage>,
@@ -63,24 +65,27 @@ impl AdminApiServer {
 	pub async fn run(
 		self,
 		bind_addr: UnixOrTCPSocketAddress,
-		shutdown_signal: impl Future<Output = ()>,
+		must_exit: watch::Receiver<bool>,
 	) -> Result<(), GarageError> {
 		let region = self.garage.config.s3_api.s3_region.clone();
 		ApiServer::new(region, self)
-			.run_server(bind_addr, Some(0o220), shutdown_signal)
+			.run_server(bind_addr, Some(0o220), must_exit)
 			.await
 	}
 
-	fn handle_options(&self, _req: &Request<Body>) -> Result<Response<Body>, Error> {
+	fn handle_options(&self, _req: &Request<IncomingBody>) -> Result<Response<ResBody>, Error> {
 		Ok(Response::builder()
 			.status(StatusCode::NO_CONTENT)
 			.header(ALLOW, "OPTIONS, GET, POST")
 			.header(ACCESS_CONTROL_ALLOW_METHODS, "OPTIONS, GET, POST")
 			.header(ACCESS_CONTROL_ALLOW_ORIGIN, "*")
-			.body(Body::empty())?)
+			.body(empty_body())?)
 	}
 
-	async fn handle_check_domain(&self, req: Request<Body>) -> Result<Response<Body>, Error> {
+	async fn handle_check_domain(
+		&self,
+		req: Request<IncomingBody>,
+	) -> Result<Response<ResBody>, Error> {
 		let query_params: HashMap<String, String> = req
 			.uri()
 			.query()
@@ -104,7 +109,7 @@ impl AdminApiServer {
 		if self.check_domain(domain).await? {
 			Ok(Response::builder()
 				.status(StatusCode::OK)
-				.body(Body::from(format!(
+				.body(string_body(format!(
 					"Domain '{domain}' is managed by Garage"
 				)))?)
 		} else {
@@ -167,7 +172,7 @@ impl AdminApiServer {
 		}
 	}
 
-	fn handle_health(&self) -> Result<Response<Body>, Error> {
+	fn handle_health(&self) -> Result<Response<ResBody>, Error> {
 		let health = self.garage.system.health();
 
 		let (status, status_str) = match health.status {
@@ -189,10 +194,10 @@ impl AdminApiServer {
 		Ok(Response::builder()
 			.status(status)
 			.header(http::header::CONTENT_TYPE, "text/plain")
-			.body(Body::from(status_str))?)
+			.body(string_body(status_str))?)
 	}
 
-	fn handle_metrics(&self) -> Result<Response<Body>, Error> {
+	fn handle_metrics(&self) -> Result<Response<ResBody>, Error> {
 		#[cfg(feature = "metrics")]
 		{
 			use opentelemetry::trace::Tracer;
@@ -212,7 +217,7 @@ impl AdminApiServer {
 			Ok(Response::builder()
 				.status(StatusCode::OK)
 				.header(http::header::CONTENT_TYPE, encoder.format_type())
-				.body(Body::from(buffer))?)
+				.body(bytes_body(buffer.into()))?)
 		}
 		#[cfg(not(feature = "metrics"))]
 		Err(Error::bad_request(
@@ -229,7 +234,7 @@ impl ApiHandler for AdminApiServer {
 	type Endpoint = Endpoint;
 	type Error = Error;
 
-	fn parse_endpoint(&self, req: &Request<Body>) -> Result<Endpoint, Error> {
+	fn parse_endpoint(&self, req: &Request<IncomingBody>) -> Result<Endpoint, Error> {
 		if req.uri().path().starts_with("/v0/") {
 			let endpoint_v0 = router_v0::Endpoint::from_request(req)?;
 			Endpoint::from_v0(endpoint_v0)
@@ -240,9 +245,9 @@ impl ApiHandler for AdminApiServer {
 
 	async fn handle(
 		&self,
-		req: Request<Body>,
+		req: Request<IncomingBody>,
 		endpoint: Endpoint,
-	) -> Result<Response<Body>, Error> {
+	) -> Result<Response<ResBody>, Error> {
 		let expected_auth_header =
 			match endpoint.authorization_type() {
 				Authorization::None => None,
diff --git a/src/api/admin/bucket.rs b/src/api/admin/bucket.rs
index 65929d61..a8718a9f 100644
--- a/src/api/admin/bucket.rs
+++ b/src/api/admin/bucket.rs
@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use hyper::{Body, Request, Response, StatusCode};
+use hyper::{body::Incoming as IncomingBody, Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 
 use garage_util::crdt::*;
@@ -17,12 +17,13 @@ use garage_model::permission::*;
 use garage_model::s3::mpu_table;
 use garage_model::s3::object_table::*;
 
+use crate::admin::api_server::ResBody;
 use crate::admin::error::*;
 use crate::admin::key::ApiBucketKeyPerm;
 use crate::common_error::CommonError;
-use crate::helpers::{json_ok_response, parse_json_body};
+use crate::helpers::*;
 
-pub async fn handle_list_buckets(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_list_buckets(garage: &Arc<Garage>) -> Result<Response<ResBody>, Error> {
 	let buckets = garage
 		.bucket_table
 		.get_range(
@@ -90,7 +91,7 @@ pub async fn handle_get_bucket_info(
 	garage: &Arc<Garage>,
 	id: Option<String>,
 	global_alias: Option<String>,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket_id = match (id, global_alias) {
 		(Some(id), None) => parse_bucket_id(&id)?,
 		(None, Some(ga)) => garage
@@ -111,7 +112,7 @@ pub async fn handle_get_bucket_info(
 async fn bucket_info_results(
 	garage: &Arc<Garage>,
 	bucket_id: Uuid,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket = garage
 		.bucket_helper()
 		.get_existing_bucket(bucket_id)
@@ -268,9 +269,9 @@ struct GetBucketInfoKey {
 
 pub async fn handle_create_bucket(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<CreateBucketRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<CreateBucketRequest, _, Error>(req).await?;
 
 	if let Some(ga) = &req.global_alias {
 		if !is_valid_bucket_name(ga) {
@@ -360,7 +361,7 @@ struct CreateBucketLocalAlias {
 pub async fn handle_delete_bucket(
 	garage: &Arc<Garage>,
 	id: String,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let helper = garage.bucket_helper();
 
 	let bucket_id = parse_bucket_id(&id)?;
@@ -403,15 +404,15 @@ pub async fn handle_delete_bucket(
 
 	Ok(Response::builder()
 		.status(StatusCode::NO_CONTENT)
-		.body(Body::empty())?)
+		.body(empty_body())?)
 }
 
 pub async fn handle_update_bucket(
 	garage: &Arc<Garage>,
 	id: String,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<UpdateBucketRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<UpdateBucketRequest, _, Error>(req).await?;
 	let bucket_id = parse_bucket_id(&id)?;
 
 	let mut bucket = garage
@@ -470,10 +471,10 @@ struct UpdateBucketWebsiteAccess {
 
 pub async fn handle_bucket_change_key_perm(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
+	req: Request<IncomingBody>,
 	new_perm_flag: bool,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<BucketKeyPermChangeRequest>(req).await?;
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<BucketKeyPermChangeRequest, _, Error>(req).await?;
 
 	let bucket_id = parse_bucket_id(&req.bucket_id)?;
 
@@ -526,7 +527,7 @@ pub async fn handle_global_alias_bucket(
 	garage: &Arc<Garage>,
 	bucket_id: String,
 	alias: String,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
 	garage
@@ -541,7 +542,7 @@ pub async fn handle_global_unalias_bucket(
 	garage: &Arc<Garage>,
 	bucket_id: String,
 	alias: String,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
 	garage
@@ -557,7 +558,7 @@ pub async fn handle_local_alias_bucket(
 	bucket_id: String,
 	access_key_id: String,
 	alias: String,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
 	garage
@@ -573,7 +574,7 @@ pub async fn handle_local_unalias_bucket(
 	bucket_id: String,
 	access_key_id: String,
 	alias: String,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let bucket_id = parse_bucket_id(&bucket_id)?;
 
 	garage
diff --git a/src/api/admin/cluster.rs b/src/api/admin/cluster.rs
index 8677257d..8ce6c5ed 100644
--- a/src/api/admin/cluster.rs
+++ b/src/api/admin/cluster.rs
@@ -2,7 +2,7 @@ use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use hyper::{Body, Request, Response};
+use hyper::{body::Incoming as IncomingBody, Request, Response};
 use serde::{Deserialize, Serialize};
 
 use garage_util::crdt::*;
@@ -12,10 +12,11 @@ use garage_rpc::layout;
 
 use garage_model::garage::Garage;
 
+use crate::admin::api_server::ResBody;
 use crate::admin::error::*;
 use crate::helpers::{json_ok_response, parse_json_body};
 
-pub async fn handle_get_cluster_status(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_get_cluster_status(garage: &Arc<Garage>) -> Result<Response<ResBody>, Error> {
 	let layout = garage.system.cluster_layout();
 	let mut nodes = garage
 		.system
@@ -110,7 +111,7 @@ pub async fn handle_get_cluster_status(garage: &Arc<Garage>) -> Result<Response<
 	Ok(json_ok_response(&res)?)
 }
 
-pub async fn handle_get_cluster_health(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_get_cluster_health(garage: &Arc<Garage>) -> Result<Response<ResBody>, Error> {
 	use garage_rpc::system::ClusterHealthStatus;
 	let health = garage.system.health();
 	let health = ClusterHealth {
@@ -132,9 +133,9 @@ pub async fn handle_get_cluster_health(garage: &Arc<Garage>) -> Result<Response<
 
 pub async fn handle_connect_cluster_nodes(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<Vec<String>>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<Vec<String>, _, Error>(req).await?;
 
 	let res = futures::future::join_all(req.iter().map(|node| garage.system.connect(node)))
 		.await
@@ -154,7 +155,7 @@ pub async fn handle_connect_cluster_nodes(
 	Ok(json_ok_response(&res)?)
 }
 
-pub async fn handle_get_cluster_layout(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_get_cluster_layout(garage: &Arc<Garage>) -> Result<Response<ResBody>, Error> {
 	let res = format_cluster_layout(&garage.system.cluster_layout());
 
 	Ok(json_ok_response(&res)?)
@@ -290,9 +291,9 @@ struct NodeResp {
 
 pub async fn handle_update_cluster_layout(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let updates = parse_json_body::<UpdateClusterLayoutRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let updates = parse_json_body::<UpdateClusterLayoutRequest, _, Error>(req).await?;
 
 	let mut layout = garage.system.cluster_layout().clone();
 
@@ -336,9 +337,9 @@ pub async fn handle_update_cluster_layout(
 
 pub async fn handle_apply_cluster_layout(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let param = parse_json_body::<ApplyLayoutRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let param = parse_json_body::<ApplyLayoutRequest, _, Error>(req).await?;
 
 	let layout = garage.system.cluster_layout().clone();
 	let (layout, msg) = layout.apply_staged_changes(Some(param.version))?;
@@ -356,7 +357,9 @@ pub async fn handle_apply_cluster_layout(
 	Ok(json_ok_response(&res)?)
 }
 
-pub async fn handle_revert_cluster_layout(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_revert_cluster_layout(
+	garage: &Arc<Garage>,
+) -> Result<Response<ResBody>, Error> {
 	let layout = garage.system.cluster_layout().clone();
 	let layout = layout.revert_staged_changes()?;
 	garage
diff --git a/src/api/admin/error.rs b/src/api/admin/error.rs
index ed1a07bd..2668b42d 100644
--- a/src/api/admin/error.rs
+++ b/src/api/admin/error.rs
@@ -1,13 +1,13 @@
 use err_derive::Error;
 use hyper::header::HeaderValue;
-use hyper::{Body, HeaderMap, StatusCode};
+use hyper::{HeaderMap, StatusCode};
 
 pub use garage_model::helper::error::Error as HelperError;
 
 use crate::common_error::CommonError;
 pub use crate::common_error::{CommonErrorDerivative, OkOrBadRequest, OkOrInternalError};
 use crate::generic_server::ApiError;
-use crate::helpers::CustomApiErrorBody;
+use crate::helpers::*;
 
 /// Errors of this crate
 #[derive(Debug, Error)]
@@ -40,18 +40,6 @@ where
 
 impl CommonErrorDerivative for Error {}
 
-impl From<HelperError> for Error {
-	fn from(err: HelperError) -> Self {
-		match err {
-			HelperError::Internal(i) => Self::Common(CommonError::InternalError(i)),
-			HelperError::BadRequest(b) => Self::Common(CommonError::BadRequest(b)),
-			HelperError::InvalidBucketName(n) => Self::Common(CommonError::InvalidBucketName(n)),
-			HelperError::NoSuchBucket(n) => Self::Common(CommonError::NoSuchBucket(n)),
-			HelperError::NoSuchAccessKey(n) => Self::NoSuchAccessKey(n),
-		}
-	}
-}
-
 impl Error {
 	fn code(&self) -> &'static str {
 		match self {
@@ -77,14 +65,14 @@ impl ApiError for Error {
 		header_map.append(header::CONTENT_TYPE, "application/json".parse().unwrap());
 	}
 
-	fn http_body(&self, garage_region: &str, path: &str) -> Body {
+	fn http_body(&self, garage_region: &str, path: &str) -> ErrorBody {
 		let error = CustomApiErrorBody {
 			code: self.code().to_string(),
 			message: format!("{}", self),
 			path: path.to_string(),
 			region: garage_region.to_string(),
 		};
-		Body::from(serde_json::to_string_pretty(&error).unwrap_or_else(|_| {
+		let error_str = serde_json::to_string_pretty(&error).unwrap_or_else(|_| {
 			r#"
 {
 	"code": "InternalError",
@@ -92,6 +80,7 @@ impl ApiError for Error {
 }
 			"#
 			.into()
-		}))
+		});
+		error_body(error_str)
 	}
 }
diff --git a/src/api/admin/key.rs b/src/api/admin/key.rs
index 8d1c6890..1efaca16 100644
--- a/src/api/admin/key.rs
+++ b/src/api/admin/key.rs
@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use hyper::{Body, Request, Response, StatusCode};
+use hyper::{body::Incoming as IncomingBody, Request, Response, StatusCode};
 use serde::{Deserialize, Serialize};
 
 use garage_table::*;
@@ -9,10 +9,11 @@ use garage_table::*;
 use garage_model::garage::Garage;
 use garage_model::key_table::*;
 
+use crate::admin::api_server::ResBody;
 use crate::admin::error::*;
-use crate::helpers::{is_default, json_ok_response, parse_json_body};
+use crate::helpers::*;
 
-pub async fn handle_list_keys(garage: &Arc<Garage>) -> Result<Response<Body>, Error> {
+pub async fn handle_list_keys(garage: &Arc<Garage>) -> Result<Response<ResBody>, Error> {
 	let res = garage
 		.key_table
 		.get_range(
@@ -45,7 +46,7 @@ pub async fn handle_get_key_info(
 	id: Option<String>,
 	search: Option<String>,
 	show_secret_key: bool,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let key = if let Some(id) = id {
 		garage.key_helper().get_existing_key(&id).await?
 	} else if let Some(search) = search {
@@ -62,9 +63,9 @@ pub async fn handle_get_key_info(
 
 pub async fn handle_create_key(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<CreateKeyRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<CreateKeyRequest, _, Error>(req).await?;
 
 	let key = Key::new(req.name.as_deref().unwrap_or("Unnamed key"));
 	garage.key_table.insert(&key).await?;
@@ -80,9 +81,9 @@ struct CreateKeyRequest {
 
 pub async fn handle_import_key(
 	garage: &Arc<Garage>,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<ImportKeyRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<ImportKeyRequest, _, Error>(req).await?;
 
 	let prev_key = garage.key_table.get(&EmptyKey, &req.access_key_id).await?;
 	if prev_key.is_some() {
@@ -111,9 +112,9 @@ struct ImportKeyRequest {
 pub async fn handle_update_key(
 	garage: &Arc<Garage>,
 	id: String,
-	req: Request<Body>,
-) -> Result<Response<Body>, Error> {
-	let req = parse_json_body::<UpdateKeyRequest>(req).await?;
+	req: Request<IncomingBody>,
+) -> Result<Response<ResBody>, Error> {
+	let req = parse_json_body::<UpdateKeyRequest, _, Error>(req).await?;
 
 	let mut key = garage.key_helper().get_existing_key(&id).await?;
 
@@ -146,7 +147,10 @@ struct UpdateKeyRequest {
 	deny: Option<KeyPerm>,
 }
 
-pub async fn handle_delete_key(garage: &Arc<Garage>, id: String) -> Result<Response<Body>, Error> {
+pub async fn handle_delete_key(
+	garage: &Arc<Garage>,
+	id: String,
+) -> Result<Response<ResBody>, Error> {
 	let mut key = garage.key_helper().get_existing_key(&id).await?;
 
 	key.state.as_option().unwrap();
@@ -155,14 +159,14 @@ pub async fn handle_delete_key(garage: &Arc<Garage>, id: String) -> Result<Respo
 
 	Ok(Response::builder()
 		.status(StatusCode::NO_CONTENT)
-		.body(Body::empty())?)
+		.body(empty_body())?)
 }
 
 async fn key_info_results(
 	garage: &Arc<Garage>,
 	key: Key,
 	show_secret: bool,
-) -> Result<Response<Body>, Error> {
+) -> Result<Response<ResBody>, Error> {
 	let mut relevant_buckets = HashMap::new();
 
 	let key_state = key.state.as_option().unwrap();
author	Alex Auvolat <alex@adnab.me>	2024-02-13 11:24:56 +0100
committer	Alex Auvolat <alex@adnab.me>	2024-02-13 11:36:28 +0100
commit	cf2af186fcc0c8f581a966454b6cd4720d3821f0 (patch)
tree	37a978ba9ffb780fc828cff7b8ec93662d50884f /src/api/admin
parent	db48dd3d6c1f9e86a62e9b8edfce2c1620bcd5f3 (diff)
parent	823078b4cdaf93e09de0847c5eaa75beb7b26b7f (diff)
download	garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.tar.gz garage-cf2af186fcc0c8f581a966454b6cd4720d3821f0.zip