gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 01:09:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(tiering): add Wasabi lifecycle target support (#5057)

Browse Source
This commit is contained in:
cxymds
2026-07-21 00:11:53 +08:00
committed by GitHub
parent 302dd42d38
commit 4f133eb95f
13 changed files with 2195 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/client/transition_api.rs
+83 -50
View File
@@ -42,7 +42,7 @@ use http::{
request::{Builder, Request},
};
use http_body::Body;
use http_body_util::BodyExt;
use http_body_util::{BodyExt, LengthLimitError, Limited};
use hyper::body::Bytes;
use hyper::body::Incoming;
use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
@@ -55,9 +55,7 @@ use rustfs_rio::HashReader;
use rustfs_utils::HashAlgorithm;
use rustfs_utils::{
net::get_endpoint_url,
retry::{
DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer, is_http_status_retryable, is_s3code_retryable,
},
retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer},
};
use rustls_pki_types::PrivateKeyDer;
use rustls_pki_types::pem::PemObject;
@@ -82,9 +80,25 @@ use url::{Url, form_urlencoded};
use uuid::Uuid;
const C_USER_AGENT: &str = "RustFS (linux; x86)";
pub(crate) const MAX_S3_ERROR_RESPONSE_SIZE: usize = 64 * 1024;
const SUCCESS_STATUS: [StatusCode; 3] = [StatusCode::OK, StatusCode::NO_CONTENT, StatusCode::PARTIAL_CONTENT];
pub(crate) async fn collect_response_body<B>(body: B, limit: usize) -> Result<Vec<u8>, std::io::Error>
where
B: Body<Data = Bytes>,
B::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
{
let body = Limited::new(body, limit).collect().await.map_err(|err| {
if err.is::<LengthLimitError>() {
std::io::Error::new(std::io::ErrorKind::InvalidData, "remote tier response body exceeds limit")
} else {
std::io::Error::other(err)
}
})?;
Ok(body.to_bytes().to_vec())
}
const C_UNKNOWN: i32 = -1;
const C_OFFLINE: i32 = 0;
const C_ONLINE: i32 = 1;
@@ -397,20 +411,17 @@ impl TransitionClient {
pub async fn doit(&self, req: Request<s3s::Body>) -> Result<Response<Incoming>, std::io::Error> {
let req_method;
let req_uri;
let req_headers;
let resp;
let http_client = self.http_client.clone();
{
req_method = req.method().clone();
req_uri = req.uri().clone();
req_headers = req.headers().clone();
debug!("endpoint_url: {}", self.endpoint_url.as_str().to_string());
resp = http_client.request(req);
}
let resp = resp.await;
debug!("http_client url: {} {}", req_method, req_uri);
debug!("http_client headers: {:?}", req_headers);
if let Err(err) = resp {
error!("http_client call error: {:?}", err);
return Err(std::io::Error::other(err));
@@ -420,28 +431,23 @@ impl TransitionClient {
Ok(r) => r,
Err(_) => return Err(std::io::Error::other("Unexpected error in response")),
};
debug!("http_resp: {:?}", resp);
debug!(status = %resp.status(), "remote tier response received");
//let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
//debug!("http_resp_body: {}", String::from_utf8(b).unwrap());
//if self.is_trace_enabled && !(self.trace_errors_only && resp.status() == StatusCode::OK) {
if !resp.status().is_success() {
//self.dump_http(&cloned_req, &resp)?;
let mut body_vec = Vec::new();
let mut body = resp.into_body();
while let Some(frame) = body.frame().await {
let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
if let Some(data) = frame.data_ref() {
body_vec.extend_from_slice(data);
}
}
let body_str = String::from_utf8_lossy(&body_vec);
warn!("err_body: {}", body_str);
Err(std::io::Error::other(format!("http_client call error: {}", body_str)))
} else {
Ok(resp)
let status = resp.status();
let request_id = resp
.headers()
.get("x-amz-request-id")
.and_then(|value| value.to_str().ok())
.unwrap_or_default()
.to_string();
warn!(status = %status, request_id, "remote tier request rejected");
}
Ok(resp)
}
pub async fn execute_method(
@@ -483,17 +489,22 @@ impl TransitionClient {
let resp_status = resp.status();
let h = resp.headers().clone();
let mut body_vec = Vec::new();
let mut body = resp.into_body();
while let Some(frame) = body.frame().await {
let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
if let Some(data) = frame.data_ref() {
body_vec.extend_from_slice(data);
}
}
let mut err_response =
http_resp_to_error_response(resp_status, &h, body_vec.clone(), &metadata.bucket_name, &metadata.object_name);
err_response.message = format!("remote tier error: {}", err_response.message);
let body_vec = collect_response_body(resp.into_body(), MAX_S3_ERROR_RESPONSE_SIZE).await?;
let parsed_error =
http_resp_to_error_response(resp_status, &h, body_vec, &metadata.bucket_name, &metadata.object_name);
let routing_region = parsed_error.region;
let code = match parsed_error.code {
S3ErrorCode::Custom(_) => S3ErrorCode::ResponseInterrupted,
code => code,
};
let err_response = ErrorResponse {
status_code: resp_status,
message: format!("remote tier request failed with status {resp_status}: {}", code.as_str()),
code,
bucket_name: metadata.bucket_name.clone(),
key: metadata.object_name.clone(),
..Default::default()
};
if self.region == "" {
return match err_response.code {
@@ -502,20 +513,20 @@ impl TransitionClient {
Err(std::io::Error::other(err_response))
}
S3ErrorCode::AccessDenied => {
if err_response.region == "" {
if routing_region.is_empty() {
return Err(std::io::Error::other(err_response));
}
if metadata.bucket_name != "" {
if let Ok(mut bucket_loc_cache) = self.bucket_loc_cache.lock() {
if let Some(location) = bucket_loc_cache.get(&metadata.bucket_name) {
if location != err_response.region {
bucket_loc_cache.set(&metadata.bucket_name, &err_response.region);
if location != routing_region {
bucket_loc_cache.set(&metadata.bucket_name, &routing_region);
//continue;
}
}
}
} else if err_response.region != metadata.bucket_location {
metadata.bucket_location = err_response.region.clone();
} else if routing_region != metadata.bucket_location {
metadata.bucket_location = routing_region;
//continue;
}
Err(std::io::Error::other(err_response))
@@ -526,18 +537,10 @@ impl TransitionClient {
};
}
if is_s3code_retryable(err_response.code.as_str()) {
continue;
}
if is_http_status_retryable(&resp_status) {
continue;
}
break;
return Err(std::io::Error::other(err_response));
}
Err(std::io::Error::other("resp err"))
Err(std::io::Error::other("remote tier request did not produce a response"))
}
async fn new_request(
@@ -1408,13 +1411,43 @@ pub struct CreateBucketConfiguration {
#[cfg(test)]
mod tests {
use super::{
SignatureType, build_tls_config, signer_error_to_io_error, to_object_info_for_provider, validate_header_values,
with_rustls_init_guard,
MAX_S3_CLIENT_RESPONSE_SIZE, MAX_S3_ERROR_RESPONSE_SIZE, SignatureType, build_tls_config, collect_response_body,
signer_error_to_io_error, to_object_info_for_provider, validate_header_values, with_rustls_init_guard,
};
use crate::client::provider_versions::ProviderVersionCapabilities;
use http::{HeaderMap, HeaderValue};
use http_body_util::Full;
use hyper::body::Bytes;
use uuid::Uuid;
#[tokio::test]
async fn response_body_limit_rejects_the_first_byte_past_the_boundary() {
let exact = collect_response_body(
Full::new(Bytes::from(vec![0_u8; MAX_S3_CLIENT_RESPONSE_SIZE])),
MAX_S3_CLIENT_RESPONSE_SIZE,
)
.await
.expect("a response exactly at the limit should be accepted");
assert_eq!(exact.len(), MAX_S3_CLIENT_RESPONSE_SIZE);
let err = collect_response_body(
Full::new(Bytes::from(vec![0_u8; MAX_S3_CLIENT_RESPONSE_SIZE + 1])),
MAX_S3_CLIENT_RESPONSE_SIZE,
)
.await
.expect_err("oversized remote response must fail closed");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
let err = collect_response_body(
Full::new(Bytes::from(vec![0_u8; MAX_S3_ERROR_RESPONSE_SIZE + 1])),
MAX_S3_ERROR_RESPONSE_SIZE,
)
.await
.expect_err("oversized S3 error response must use the smaller error limit");
assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
}
#[test]
fn rustls_guard_converts_panics_to_io_errors() {
let err = with_rustls_init_guard(|| -> Result<(), std::io::Error> { panic!("missing provider") })