gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-18 10:43:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feature up/ilm (#305)

Browse Source 
* fix

* fix

* fix

* fix delete-marker expiration. add api_restore.

* fix

* time retry object upload

* lock file

* make fmt

* fix

* restore object

* fix

* fix

* serde-rs-xml -> quick-xml

* fix

* checksum

* fix

* fix

* fix

* fix

* fix

* fix

* fix
This commit is contained in:
likewu
2025-07-29 14:21:19 +08:00
committed by GitHub
parent 56fd8132e9
commit d56cee26db
35 changed files with 1281 additions and 374 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/client/api_bucket_policy.rs
+2 -2
View File
@@ -63,7 +63,7 @@ impl TransitionClient {
//defer closeResponse(resp)
//if resp != nil {
if resp.status() != StatusCode::NO_CONTENT && resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, "")));
return Err(std::io::Error::other(http_resp_to_error_response(&resp, vec![], bucket_name, "")));
}
//}
Ok(())
@@ -98,7 +98,7 @@ impl TransitionClient {
//defer closeResponse(resp)
if resp.status() != StatusCode::NO_CONTENT {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, "")));
return Err(std::io::Error::other(http_resp_to_error_response(&resp, vec![], bucket_name, "")));
}
Ok(())
crates/ecstore/src/client/api_error_response.rs
+2 -2
View File
@@ -95,13 +95,13 @@ pub fn to_error_response(err: &std::io::Error) -> ErrorResponse {
}
pub fn http_resp_to_error_response(
resp: http::Response<Body>,
resp: &http::Response<Body>,
b: Vec<u8>,
bucket_name: &str,
object_name: &str,
) -> ErrorResponse {
let err_body = String::from_utf8(b).unwrap();
let err_resp_ = serde_xml_rs::from_str::<ErrorResponse>(&err_body);
let err_resp_ = quick_xml::de::from_str::<ErrorResponse>(&err_body);
let mut err_resp = ErrorResponse::default();
if err_resp_.is_err() {
match resp.status() {
crates/ecstore/src/client/api_get_object_acl.rs
+2 -2
View File
@@ -87,11 +87,11 @@ impl TransitionClient {
if resp.status() != http::StatusCode::OK {
let b = resp.body().bytes().expect("err").to_vec();
return Err(std::io::Error::other(http_resp_to_error_response(resp, b, bucket_name, object_name)));
return Err(std::io::Error::other(http_resp_to_error_response(&resp, b, bucket_name, object_name)));
}
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let mut res = match serde_xml_rs::from_str::<AccessControlPolicy>(&String::from_utf8(b).unwrap()) {
let mut res = match quick_xml::de::from_str::<AccessControlPolicy>(&String::from_utf8(b).unwrap()) {
Ok(result) => result,
Err(err) => {
return Err(std::io::Error::other(err.to_string()));
crates/ecstore/src/client/api_get_object_attributes.rs
+2 -2
View File
@@ -144,7 +144,7 @@ impl ObjectAttributes {
self.version_id = h.get(X_AMZ_VERSION_ID).unwrap().to_str().unwrap().to_string();
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let mut response = match serde_xml_rs::from_str::<ObjectAttributesResponse>(&String::from_utf8(b).unwrap()) {
let mut response = match quick_xml::de::from_str::<ObjectAttributesResponse>(&String::from_utf8(b).unwrap()) {
Ok(result) => result,
Err(err) => {
return Err(std::io::Error::other(err.to_string()));
@@ -226,7 +226,7 @@ impl TransitionClient {
if resp.status() != http::StatusCode::OK {
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let err_body = String::from_utf8(b).unwrap();
let mut er = match serde_xml_rs::from_str::<AccessControlPolicy>(&err_body) {
let mut er = match quick_xml::de::from_str::<AccessControlPolicy>(&err_body) {
Ok(result) => result,
Err(err) => {
return Err(std::io::Error::other(err.to_string()));
crates/ecstore/src/client/api_list.rs
+2 -2
View File
@@ -98,12 +98,12 @@ impl TransitionClient {
)
.await?;
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, "")));
return Err(std::io::Error::other(http_resp_to_error_response(&resp, vec![], bucket_name, "")));
}
//let mut list_bucket_result = ListBucketV2Result::default();
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let mut list_bucket_result = match serde_xml_rs::from_str::<ListBucketV2Result>(&String::from_utf8(b).unwrap()) {
let mut list_bucket_result = match quick_xml::de::from_str::<ListBucketV2Result>(&String::from_utf8(b).unwrap()) {
Ok(result) => result,
Err(err) => {
return Err(std::io::Error::other(err.to_string()));
crates/ecstore/src/client/api_put_object.rs
+5 -4
View File
@@ -85,7 +85,7 @@ pub struct PutObjectOptions {
pub expires: OffsetDateTime,
pub mode: ObjectLockRetentionMode,
pub retain_until_date: OffsetDateTime,
//pub server_side_encryption: encrypt.ServerSide,
//pub server_side_encryption: encrypt::ServerSide,
pub num_threads: u64,
pub storage_class: String,
pub website_redirect_location: String,
@@ -135,7 +135,7 @@ impl Default for PutObjectOptions {
#[allow(dead_code)]
impl PutObjectOptions {
fn set_match_tag(&mut self, etag: &str) {
fn set_match_etag(&mut self, etag: &str) {
if etag == "*" {
self.custom_header
.insert("If-Match", HeaderValue::from_str("*").expect("err"));
@@ -145,7 +145,7 @@ impl PutObjectOptions {
}
}
fn set_match_tag_except(&mut self, etag: &str) {
fn set_match_etag_except(&mut self, etag: &str) {
if etag == "*" {
self.custom_header
.insert("If-None-Match", HeaderValue::from_str("*").expect("err"));
@@ -366,7 +366,8 @@ impl TransitionClient {
md5_base64 = base64_encode(hash.as_ref());
} else {
let mut crc = opts.auto_checksum.hasher()?;
let csum = crc.hash_encode(&buf[..length]);
crc.update(&buf[..length]);
let csum = crc.finalize();
if let Ok(header_name) = HeaderName::from_bytes(opts.auto_checksum.key().as_bytes()) {
custom_header.insert(header_name, base64_encode(csum.as_ref()).parse().expect("err"));
crates/ecstore/src/client/api_put_object_common.rs
-1
View File
@@ -12,7 +12,6 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(unused_imports)]
#![allow(unused_variables)]
#![allow(unused_mut)]
#![allow(unused_assignments)]
crates/ecstore/src/client/api_put_object_multipart.rs
+15 -20
View File
@@ -11,7 +11,6 @@
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(unused_imports)]
#![allow(unused_variables)]
#![allow(unused_mut)]
#![allow(unused_assignments)]
@@ -19,20 +18,14 @@
#![allow(clippy::all)]
use bytes::Bytes;
use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
use http::{HeaderMap, HeaderName, StatusCode};
use s3s::S3ErrorCode;
use std::io::Read;
use std::{collections::HashMap, sync::Arc};
use time::{OffsetDateTime, format_description};
use tokio_util::sync::CancellationToken;
use std::collections::HashMap;
use time::OffsetDateTime;
use tracing::warn;
use tracing::{error, info};
use url::form_urlencoded::Serializer;
use uuid::Uuid;
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
use s3s::{Body, dto::StreamingBlob};
//use crate::disk::{Reader, BufferReader};
use crate::checksum::ChecksumMode;
use crate::client::{
api_error_response::{
err_entity_too_large, err_entity_too_small, err_invalid_argument, http_resp_to_error_response, to_error_response,
@@ -42,15 +35,11 @@ use crate::client::{
api_s3_datatypes::{
CompleteMultipartUpload, CompleteMultipartUploadResult, CompletePart, InitiateMultipartUploadResult, ObjectPart,
},
constants::{ABS_MIN_PART_SIZE, ISO8601_DATEFORMAT, MAX_PART_SIZE, MAX_SINGLE_PUT_OBJECT_SIZE},
constants::{ISO8601_DATEFORMAT, MAX_PART_SIZE, MAX_SINGLE_PUT_OBJECT_SIZE},
transition_api::{ReaderImpl, RequestMetadata, TransitionClient, UploadInfo},
};
use crate::{
checksum::ChecksumMode,
disk::DiskAPI,
store_api::{GetObjectReader, StorageAPI},
};
use rustfs_utils::{crypto::base64_encode, path::trim_etag};
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
impl TransitionClient {
pub async fn put_object_multipart(
@@ -133,7 +122,8 @@ impl TransitionClient {
//}
if hash_sums.len() == 0 {
let mut crc = opts.auto_checksum.hasher()?;
let csum = crc.hash_encode(&buf[..length]);
crc.update(&buf[..length]);
let csum = crc.finalize();
if let Ok(header_name) = HeaderName::from_bytes(opts.auto_checksum.key().as_bytes()) {
custom_header.insert(header_name, base64_encode(csum.as_ref()).parse().expect("err"));
@@ -236,7 +226,12 @@ impl TransitionClient {
let resp = self.execute_method(http::Method::POST, &mut req_metadata).await?;
//if resp.is_none() {
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, object_name)));
return Err(std::io::Error::other(http_resp_to_error_response(
&resp,
vec![],
bucket_name,
object_name,
)));
}
//}
let initiate_multipart_upload_result = InitiateMultipartUploadResult::default();
@@ -293,7 +288,7 @@ impl TransitionClient {
let resp = self.execute_method(http::Method::PUT, &mut req_metadata).await?;
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(
resp,
&resp,
vec![],
&p.bucket_name.clone(),
&p.object_name,
crates/ecstore/src/client/api_put_object_streaming.rs
+10 -3
View File
@@ -156,7 +156,8 @@ impl TransitionClient {
md5_base64 = base64_encode(hash.as_ref());
} else {
let mut crc = opts.auto_checksum.hasher()?;
let csum = crc.hash_encode(&buf[..length]);
crc.update(&buf[..length]);
let csum = crc.finalize();
if let Ok(header_name) = HeaderName::from_bytes(opts.auto_checksum.key().as_bytes()) {
custom_header.insert(header_name, base64_encode(csum.as_ref()).parse().expect("err"));
@@ -303,7 +304,8 @@ impl TransitionClient {
let mut custom_header = HeaderMap::new();
if !opts.send_content_md5 {
let mut crc = opts.auto_checksum.hasher()?;
let csum = crc.hash_encode(&buf[..length]);
crc.update(&buf[..length]);
let csum = crc.finalize();
if let Ok(header_name) = HeaderName::from_bytes(opts.auto_checksum.key().as_bytes()) {
custom_header.insert(header_name, base64_encode(csum.as_ref()).parse().expect("err"));
@@ -477,7 +479,12 @@ impl TransitionClient {
let resp = self.execute_method(http::Method::PUT, &mut req_metadata).await?;
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, object_name)));
return Err(std::io::Error::other(http_resp_to_error_response(
&resp,
vec![],
bucket_name,
object_name,
)));
}
let (exp_time, rule_id) = if let Some(h_x_amz_expiration) = resp.headers().get(X_AMZ_EXPIRATION) {
crates/ecstore/src/client/api_remove.rs
+6 -1
View File
@@ -425,7 +425,12 @@ impl TransitionClient {
};
}
_ => {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, object_name)));
return Err(std::io::Error::other(http_resp_to_error_response(
&resp,
vec![],
bucket_name,
object_name,
)));
}
}
return Err(std::io::Error::other(error_response));
crates/ecstore/src/client/api_restore.rs
+2 -2
View File
@@ -125,7 +125,7 @@ impl TransitionClient {
version_id: &str,
restore_req: &RestoreRequest,
) -> Result<(), std::io::Error> {
let restore_request = match serde_xml_rs::to_string(restore_req) {
let restore_request = match quick_xml::se::to_string(restore_req) {
Ok(buf) => buf,
Err(e) => {
return Err(std::io::Error::other(e));
@@ -165,7 +165,7 @@ impl TransitionClient {
let b = resp.body().bytes().expect("err").to_vec();
if resp.status() != http::StatusCode::ACCEPTED && resp.status() != http::StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, b, bucket_name, "")));
return Err(std::io::Error::other(http_resp_to_error_response(&resp, b, bucket_name, "")));
}
Ok(())
}
crates/ecstore/src/client/api_s3_datatypes.rs
+2 -2
View File
@@ -279,7 +279,7 @@ pub struct CompleteMultipartUpload {
impl CompleteMultipartUpload {
pub fn marshal_msg(&self) -> Result<String, std::io::Error> {
//let buf = serde_json::to_string(self)?;
let buf = match serde_xml_rs::to_string(self) {
let buf = match quick_xml::se::to_string(self) {
Ok(buf) => buf,
Err(e) => {
return Err(std::io::Error::other(e));
@@ -329,7 +329,7 @@ pub struct DeleteMultiObjects {
impl DeleteMultiObjects {
pub fn marshal_msg(&self) -> Result<String, std::io::Error> {
//let buf = serde_json::to_string(self)?;
let buf = match serde_xml_rs::to_string(self) {
let buf = match quick_xml::se::to_string(self) {
Ok(buf) => buf,
Err(e) => {
return Err(std::io::Error::other(e));
crates/ecstore/src/client/api_stat.rs
+1 -1
View File
@@ -59,7 +59,7 @@ impl TransitionClient {
if let Ok(resp) = resp {
let b = resp.body().bytes().expect("err").to_vec();
let resperr = http_resp_to_error_response(resp, b, bucket_name, "");
let resperr = http_resp_to_error_response(&resp, b, bucket_name, "");
/*if to_error_response(resperr).code == "NoSuchBucket" {
return Ok(false);
}
crates/ecstore/src/client/bucket_cache.rs
+2 -2
View File
@@ -177,7 +177,7 @@ impl TransitionClient {
async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket_name: &str) -> Result<String, std::io::Error> {
//if resp != nil {
if resp.status() != StatusCode::OK {
let err_resp = http_resp_to_error_response(resp, vec![], bucket_name, "");
let err_resp = http_resp_to_error_response(&resp, vec![], bucket_name, "");
match err_resp.code {
S3ErrorCode::NotImplemented => {
match err_resp.server.as_str() {
@@ -208,7 +208,7 @@ async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket
//}
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let Document(location_constraint) = serde_xml_rs::from_str::<Document>(&String::from_utf8(b).unwrap()).unwrap();
let Document(location_constraint) = quick_xml::de::from_str::<Document>(&String::from_utf8(b).unwrap()).unwrap();
let mut location = location_constraint;
if location == "" {
crates/ecstore/src/client/transition_api.rs
+18 -10
View File
@@ -19,7 +19,7 @@
#![allow(clippy::all)]
use bytes::Bytes;
use futures::Future;
use futures::{Future, StreamExt};
use http::{HeaderMap, HeaderName};
use http::{
HeaderValue, Response, StatusCode,
@@ -65,7 +65,9 @@ use crate::{checksum::ChecksumMode, store_api::GetObjectReader};
use rustfs_rio::HashReader;
use rustfs_utils::{
net::get_endpoint_url,
retry::{MAX_RETRY, new_retry_timer},
retry::{
DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer, is_http_status_retryable, is_s3code_retryable,
},
};
use s3s::S3ErrorCode;
use s3s::dto::ReplicationStatus;
@@ -186,6 +188,7 @@ impl TransitionClient {
clnt.trailing_header_support = opts.trailing_headers && clnt.override_signer_type == SignatureType::SignatureV4;
clnt.max_retries = MAX_RETRY;
if opts.max_retries > 0 {
clnt.max_retries = opts.max_retries;
}
@@ -313,12 +316,9 @@ impl TransitionClient {
}
//}
//let mut retry_timer = RetryTimer::new();
//while let Some(v) = retry_timer.next().await {
for _ in [1; 1]
/*new_retry_timer(req_retry, default_retry_unit, default_retry_cap, max_jitter)*/
{
let req = self.new_request(method, metadata).await?;
let mut retry_timer = RetryTimer::new(req_retry, DEFAULT_RETRY_UNIT, DEFAULT_RETRY_CAP, MAX_JITTER, self.random);
while let Some(v) = retry_timer.next().await {
let req = self.new_request(&method, metadata).await?;
resp = self.doit(req).await?;
@@ -329,7 +329,7 @@ impl TransitionClient {
}
let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
let err_response = http_resp_to_error_response(resp, b.clone(), &metadata.bucket_name, &metadata.object_name);
let err_response = http_resp_to_error_response(&resp, b.clone(), &metadata.bucket_name, &metadata.object_name);
if self.region == "" {
match err_response.code {
@@ -360,6 +360,14 @@ impl TransitionClient {
}
}
if is_s3code_retryable(err_response.code.as_str()) {
continue;
}
if is_http_status_retryable(&resp.status()) {
continue;
}
break;
}
@@ -368,7 +376,7 @@ impl TransitionClient {
async fn new_request(
&self,
method: http::Method,
method: &http::Method,
metadata: &mut RequestMetadata,
) -> Result<http::Request<Body>, std::io::Error> {
let location = metadata.bucket_location.clone();