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

fix clippy

Browse Source
This commit is contained in:
likewu
2025-06-23 12:03:46 +08:00
parent cc71f40a6d
commit 2c176dd864
70 changed files with 2779 additions and 1969 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/client/admin_handler_utils.rs
+1 -1
View File
@@ -30,4 +30,4 @@ impl AdminError {
status_code: StatusCode::INTERNAL_SERVER_ERROR,
}
}
}
}
ecstore/src/client/api_bucket_policy.rs
+53 -43
View File
@@ -1,11 +1,11 @@
#![allow(clippy::map_entry)]
use std::collections::HashMap;
use http::{HeaderMap, StatusCode};
use bytes::Bytes;
use http::{HeaderMap, StatusCode};
use std::collections::HashMap;
use crate::client::{
api_error_response::http_resp_to_error_response,
transition_api::{RequestMetadata, TransitionClient, ReaderImpl}
transition_api::{ReaderImpl, RequestMetadata, TransitionClient},
};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
@@ -23,9 +23,9 @@ impl TransitionClient {
url_values.insert("policy".to_string(), "".to_string());
let mut req_metadata = RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_body: ReaderImpl::Body(Bytes::from(policy.as_bytes().to_vec())),
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_body: ReaderImpl::Body(Bytes::from(policy.as_bytes().to_vec())),
content_length: policy.len() as i64,
object_name: "".to_string(),
custom_header: HeaderMap::new(),
@@ -43,9 +43,9 @@ impl TransitionClient {
let resp = self.execute_method(http::Method::PUT, &mut req_metadata).await?;
//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, "")));
}
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, "")));
}
//}
Ok(())
}
@@ -54,23 +54,28 @@ impl TransitionClient {
let mut url_values = HashMap::new();
url_values.insert("policy".to_string(), "".to_string());
let resp = self.execute_method(http::Method::DELETE, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
object_name: "".to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::DELETE,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
object_name: "".to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
//defer closeResponse(resp)
if resp.status() != StatusCode::NO_CONTENT {
@@ -89,23 +94,28 @@ impl TransitionClient {
let mut url_values = HashMap::new();
url_values.insert("policy".to_string(), "".to_string());
let resp = self.execute_method(http::Method::GET, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
object_name: "".to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::GET,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
object_name: "".to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
let policy = String::from_utf8_lossy(&resp.body().bytes().expect("err").to_vec()).to_string();
Ok(policy)
ecstore/src/client/api_error_response.rs
+73 -59
View File
@@ -1,26 +1,25 @@
#![allow(clippy::map_entry)]
use std::fmt::Display;
use http::StatusCode;
use serde::{Serialize, Deserialize};
use serde::{ser::Serializer, de::Deserializer};
use serde::{Deserialize, Serialize};
use serde::{de::Deserializer, ser::Serializer};
use std::fmt::Display;
use s3s::S3ErrorCode;
use s3s::Body;
use s3s::S3ErrorCode;
const REPORT_ISSUE: &str = "Please report this issue at https://github.com/rustfs/rustfs/issues.";
#[derive(Serialize, Deserialize)]
#[derive(Debug, Clone, thiserror::Error, PartialEq, Eq)]
#[derive(Serialize, Deserialize, Debug, Clone, thiserror::Error, PartialEq, Eq)]
#[serde(default, rename_all = "PascalCase")]
pub struct ErrorResponse {
#[serde(serialize_with = "serialize_code", deserialize_with = "deserialize_code")]
pub code: S3ErrorCode,
pub message: String,
pub code: S3ErrorCode,
pub message: String,
pub bucket_name: String,
pub key: String,
pub resource: String,
pub request_id: String,
pub host_id: String,
pub key: String,
pub resource: String,
pub request_id: String,
pub host_id: String,
pub region: String,
pub server: String,
#[serde(skip)]
@@ -29,14 +28,14 @@ pub struct ErrorResponse {
fn serialize_code<S>(data: &S3ErrorCode, s: S) -> Result<S::Ok, S::Error>
where
S: Serializer
S: Serializer,
{
s.serialize_str("")
}
fn deserialize_code<'de, D>(d: D) -> Result<S3ErrorCode, D::Error>
where
D: Deserializer<'de>
D: Deserializer<'de>,
{
Ok(S3ErrorCode::from_bytes(String::deserialize(d)?.as_bytes()).unwrap_or(S3ErrorCode::Custom("".into())))
}
@@ -45,14 +44,14 @@ impl Default for ErrorResponse {
fn default() -> Self {
ErrorResponse {
code: S3ErrorCode::Custom("".into()),
message: Default::default(),
message: Default::default(),
bucket_name: Default::default(),
key: Default::default(),
resource: Default::default(),
request_id: Default::default(),
host_id: Default::default(),
region: Default::default(),
server: Default::default(),
key: Default::default(),
resource: Default::default(),
request_id: Default::default(),
host_id: Default::default(),
region: Default::default(),
server: Default::default(),
status_code: Default::default(),
}
}
@@ -76,7 +75,12 @@ pub fn to_error_response(err: &std::io::Error) -> ErrorResponse {
}
}
pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucket_name: &str, object_name: &str) -> ErrorResponse {
pub fn http_resp_to_error_response(
resp: http::Response<Body>,
b: Vec<u8>,
bucket_name: &str,
object_name: &str,
) -> ErrorResponse {
let err_body = String::from_utf8(b).unwrap();
//let err_body = xml_decode_and_body(resp.body, &err_resp);
let err_resp_ = serde_xml_rs::from_str::<ErrorResponse>(&err_body);
@@ -87,18 +91,18 @@ pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucke
if object_name == "" {
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::NoSuchBucket,
message: "The specified bucket does not exist.".to_string(),
code: S3ErrorCode::NoSuchBucket,
message: "The specified bucket does not exist.".to_string(),
bucket_name: bucket_name.to_string(),
..Default::default()
};
} else {
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::NoSuchKey,
message: "The specified key does not exist.".to_string(),
code: S3ErrorCode::NoSuchKey,
message: "The specified key does not exist.".to_string(),
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
};
}
@@ -106,18 +110,18 @@ pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucke
StatusCode::FORBIDDEN => {
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::AccessDenied,
message: "Access Denied.".to_string(),
code: S3ErrorCode::AccessDenied,
message: "Access Denied.".to_string(),
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
};
}
StatusCode::CONFLICT => {
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::BucketNotEmpty,
message: "Bucket not empty.".to_string(),
code: S3ErrorCode::BucketNotEmpty,
message: "Bucket not empty.".to_string(),
bucket_name: bucket_name.to_string(),
..Default::default()
};
@@ -125,10 +129,10 @@ pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucke
StatusCode::PRECONDITION_FAILED => {
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::PreconditionFailed,
message: "Pre condition failed.".to_string(),
code: S3ErrorCode::PreconditionFailed,
message: "Pre condition failed.".to_string(),
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
};
}
@@ -137,10 +141,10 @@ pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucke
if err_body.len() > 0 {
msg = err_body;
}
err_resp = ErrorResponse{
err_resp = ErrorResponse {
status_code: resp.status(),
code: S3ErrorCode::Custom(resp.status().to_string().into()),
message: msg,
code: S3ErrorCode::Custom(resp.status().to_string().into()),
message: msg,
bucket_name: bucket_name.to_string(),
..Default::default()
};
@@ -188,45 +192,55 @@ pub fn http_resp_to_error_response(resp: http::Response<Body>, b: Vec<u8>, bucke
pub fn err_transfer_acceleration_bucket(bucket_name: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::InvalidArgument,
message: "The name of the bucket used for Transfer Acceleration must be DNS-compliant and must not contain periods ‘.’.".to_string(),
code: S3ErrorCode::InvalidArgument,
message: "The name of the bucket used for Transfer Acceleration must be DNS-compliant and must not contain periods ‘.’."
.to_string(),
bucket_name: bucket_name.to_string(),
..Default::default()
}
}
pub fn err_entity_too_large(total_size: i64, max_object_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
let msg = format!("Your proposed upload size ‘{}’ exceeds the maximum allowed object size ‘{}’ for single PUT operation.", total_size, max_object_size);
let msg = format!(
"Your proposed upload size ‘{}’ exceeds the maximum allowed object size ‘{}’ for single PUT operation.",
total_size, max_object_size
);
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::EntityTooLarge,
message: msg,
code: S3ErrorCode::EntityTooLarge,
message: msg,
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
}
}
pub fn err_entity_too_small(total_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
let msg = format!("Your proposed upload size ‘{}’ is below the minimum allowed object size ‘0B’ for single PUT operation.", total_size);
let msg = format!(
"Your proposed upload size ‘{}’ is below the minimum allowed object size ‘0B’ for single PUT operation.",
total_size
);
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::EntityTooSmall,
message: msg,
code: S3ErrorCode::EntityTooSmall,
message: msg,
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
}
}
pub fn err_unexpected_eof(total_read: i64, total_size: i64, bucket_name: &str, object_name: &str) -> ErrorResponse {
let msg = format!("Data read ‘{}’ is not equal to the size ‘{}’ of the input Reader.", total_read, total_size);
let msg = format!(
"Data read ‘{}’ is not equal to the size ‘{}’ of the input Reader.",
total_read, total_size
);
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::Custom("UnexpectedEOF".into()),
message: msg,
code: S3ErrorCode::Custom("UnexpectedEOF".into()),
message: msg,
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
key: object_name.to_string(),
..Default::default()
}
}
@@ -234,9 +248,9 @@ pub fn err_unexpected_eof(total_read: i64, total_size: i64, bucket_name: &str, o
pub fn err_invalid_argument(message: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::BAD_REQUEST,
code: S3ErrorCode::InvalidArgument,
message: message.to_string(),
request_id: "rustfs".to_string(),
code: S3ErrorCode::InvalidArgument,
message: message.to_string(),
request_id: "rustfs".to_string(),
..Default::default()
}
}
@@ -244,9 +258,9 @@ pub fn err_invalid_argument(message: &str) -> ErrorResponse {
pub fn err_api_not_supported(message: &str) -> ErrorResponse {
ErrorResponse {
status_code: StatusCode::NOT_IMPLEMENTED,
code: S3ErrorCode::Custom("APINotSupported".into()),
message: message.to_string(),
request_id: "rustfs".to_string(),
code: S3ErrorCode::Custom("APINotSupported".into()),
message: message.to_string(),
request_id: "rustfs".to_string(),
..Default::default()
}
}
}
ecstore/src/client/api_get_object.rs
+48 -40
View File
@@ -1,13 +1,13 @@
#![allow(clippy::map_entry)]
use bytes::Bytes;
use http::HeaderMap;
use tokio::io::BufReader;
use std::io::Cursor;
use tokio::io::BufReader;
use crate::client::{
transition_api::{ObjectInfo, to_object_info, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient},
api_error_response::err_invalid_argument,
api_get_options::GetObjectOptions,
transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info},
};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
@@ -16,24 +16,34 @@ impl TransitionClient {
todo!();
}
pub async fn get_object_inner(&self, bucket_name: &str, object_name: &str, opts: &GetObjectOptions) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
let resp = self.execute_method(http::Method::GET, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: opts.to_query_values(),
custom_header: opts.header(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
pub async fn get_object_inner(
&self,
bucket_name: &str,
object_name: &str,
opts: &GetObjectOptions,
) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
let resp = self
.execute_method(
http::Method::GET,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: opts.to_query_values(),
custom_header: opts.header(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
let resp = &resp;
let object_stat = to_object_info(bucket_name, object_name, resp.headers())?;
@@ -45,20 +55,20 @@ impl TransitionClient {
#[derive(Default)]
struct GetRequest {
pub buffer: Vec<u8>,
pub offset: i64,
pub did_offset_change: bool,
pub been_read: bool,
pub is_read_at: bool,
pub is_read_op: bool,
pub is_first_req: bool,
pub buffer: Vec<u8>,
pub offset: i64,
pub did_offset_change: bool,
pub been_read: bool,
pub is_read_at: bool,
pub is_read_op: bool,
pub is_first_req: bool,
pub setting_object_info: bool,
}
struct GetResponse {
pub size: i64,
pub size: i64,
//pub error: error,
pub did_read: bool,
pub did_read: bool,
pub object_info: ObjectInfo,
}
@@ -79,9 +89,7 @@ struct Object {
impl Object {
pub fn new() -> Object {
Self {
..Default::default()
}
Self { ..Default::default() }
}
fn do_get_request(&self, request: &GetRequest) -> Result<GetResponse, std::io::Error> {
@@ -121,7 +129,7 @@ impl Object {
fn stat(&self) -> Result<ObjectInfo, std::io::Error> {
if !self.is_started || !self.object_info_set {
let _ = self.do_get_request(&GetRequest {
is_first_req: !self.is_started,
is_first_req: !self.is_started,
setting_object_info: !self.object_info_set,
..Default::default()
})?;
@@ -134,12 +142,12 @@ impl Object {
self.curr_offset = offset;
let mut read_at_req = GetRequest {
is_read_op: true,
is_read_at: true,
is_read_op: true,
is_read_at: true,
did_offset_change: true,
been_read: self.been_read,
been_read: self.been_read,
offset,
buffer: b.to_vec(),
buffer: b.to_vec(),
..Default::default()
};
@@ -160,8 +168,8 @@ impl Object {
fn seek(&mut self, offset: i64, whence: i64) -> Result<i64, std::io::Error> {
if !self.is_started || !self.object_info_set {
let seek_req = GetRequest {
is_read_op: false,
offset: offset,
is_read_op: false,
offset: offset,
is_first_req: true,
..Default::default()
};
@@ -195,4 +203,4 @@ impl Object {
self.is_closed = true;
Ok(())
}
}
}
ecstore/src/client/api_get_options.rs
+12 -12
View File
@@ -1,6 +1,6 @@
#![allow(clippy::map_entry)]
use std::collections::HashMap;
use http::{HeaderMap, HeaderName, HeaderValue};
use std::collections::HashMap;
use time::OffsetDateTime;
use tracing::warn;
@@ -8,9 +8,9 @@ use crate::client::api_error_response::err_invalid_argument;
#[derive(Default)]
pub struct AdvancedGetOptions {
replication_deletemarker: bool,
replication_deletemarker: bool,
is_replication_ready_for_deletemarker: bool,
replication_proxy_request: String,
replication_proxy_request: String,
}
pub struct GetObjectOptions {
@@ -50,7 +50,7 @@ impl GetObjectOptions {
}
}
if self.checksum {
headers.insert("x-amz-checksum-mode", "ENABLED".parse().expect("err"));
headers.insert("x-amz-checksum-mode", "ENABLED".parse().expect("err"));
}
headers
}
@@ -96,15 +96,15 @@ impl GetObjectOptions {
pub fn set_range(&mut self, start: i64, end: i64) -> Result<(), std::io::Error> {
if start == 0 && end < 0 {
self.set("Range", &format!("bytes={}", end));
}
else if 0 < start && end == 0 {
} else if 0 < start && end == 0 {
self.set("Range", &format!("bytes={}-", start));
}
else if 0 <= start && start <= end {
} else if 0 <= start && start <= end {
self.set("Range", &format!("bytes={}-{}", start, end));
}
else {
return Err(std::io::Error::other(err_invalid_argument(&format!("Invalid range specified: start={} end={}", start, end))));
} else {
return Err(std::io::Error::other(err_invalid_argument(&format!(
"Invalid range specified: start={} end={}",
start, end
))));
}
Ok(())
}
@@ -124,4 +124,4 @@ impl GetObjectOptions {
url_values
}
}
}
ecstore/src/client/api_list.rs
+87 -34
View File
@@ -1,23 +1,36 @@
#![allow(clippy::map_entry)]
use std::collections::HashMap;
use bytes::Bytes;
use http::{HeaderMap, StatusCode};
use std::collections::HashMap;
use crate::client::{
api_error_response::http_resp_to_error_response,
api_s3_datatypes::{
ListBucketResult, ListBucketV2Result, ListMultipartUploadsResult, ListObjectPartsResult, ListVersionsResult, ObjectPart,
},
credentials,
api_s3_datatypes::{ListBucketV2Result, ListMultipartUploadsResult, ListBucketResult, ListObjectPartsResult, ListVersionsResult, ObjectPart},
transition_api::{ReaderImpl, TransitionClient, RequestMetadata,},
transition_api::{ReaderImpl, RequestMetadata, TransitionClient},
};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
use crate::store_api::BucketInfo;
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
impl TransitionClient {
pub fn list_buckets(&self) -> Result<Vec<BucketInfo>, std::io::Error> {
todo!();
}
pub async fn list_objects_v2_query(&self, bucket_name: &str, object_prefix: &str, continuation_token: &str, fetch_owner: bool, metadata: bool, delimiter: &str, start_after: &str, max_keys: i64, headers: HeaderMap) -> Result<ListBucketV2Result, std::io::Error> {
pub async fn list_objects_v2_query(
&self,
bucket_name: &str,
object_prefix: &str,
continuation_token: &str,
fetch_owner: bool,
metadata: bool,
delimiter: &str,
start_after: &str,
max_keys: i64,
headers: HeaderMap,
) -> Result<ListBucketV2Result, std::io::Error> {
let mut url_values = HashMap::new();
url_values.insert("list-type".to_string(), "2".to_string());
@@ -43,23 +56,28 @@ impl TransitionClient {
url_values.insert("max-keys".to_string(), max_keys.to_string());
}
let mut resp = self.execute_method(http::Method::GET, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: "".to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let mut resp = self
.execute_method(
http::Method::GET,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: "".to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, "")));
}
@@ -76,12 +94,12 @@ impl TransitionClient {
if list_bucket_result.is_truncated && list_bucket_result.next_continuation_token == "" {
return Err(std::io::Error::other(credentials::ErrorResponse {
sts_error: credentials::STSError {
r#type: "".to_string(),
code: "NotImplemented".to_string(),
message: "Truncated response should have continuation token set".to_string(),
},
request_id: "".to_string(),
sts_error: credentials::STSError {
r#type: "".to_string(),
code: "NotImplemented".to_string(),
message: "Truncated response should have continuation token set".to_string(),
},
request_id: "".to_string(),
}));
}
@@ -97,7 +115,14 @@ impl TransitionClient {
Ok(list_bucket_result)
}
pub fn list_object_versions_query(&self, bucket_name: &str, opts: &ListObjectsOptions, key_marker: &str, version_id_marker: &str, delimiter: &str) -> Result<ListVersionsResult, std::io::Error> {
pub fn list_object_versions_query(
&self,
bucket_name: &str,
opts: &ListObjectsOptions,
key_marker: &str,
version_id_marker: &str,
delimiter: &str,
) -> Result<ListVersionsResult, std::io::Error> {
/*if err := s3utils.CheckValidBucketName(bucketName); err != nil {
return ListVersionsResult{}, err
}
@@ -177,15 +202,36 @@ impl TransitionClient {
todo!();
}
pub fn list_objects_query(&self, bucket_name: &str, object_prefix: &str, object_marker: &str, delimiter: &str, max_keys: i64, headers: HeaderMap) -> Result<ListBucketResult, std::io::Error> {
pub fn list_objects_query(
&self,
bucket_name: &str,
object_prefix: &str,
object_marker: &str,
delimiter: &str,
max_keys: i64,
headers: HeaderMap,
) -> Result<ListBucketResult, std::io::Error> {
todo!();
}
pub fn list_multipart_uploads_query(&self, bucket_name: &str, key_marker: &str, upload_id_marker: &str, prefix: &str, delimiter: &str, max_uploads: i64) -> Result<ListMultipartUploadsResult, std::io::Error> {
pub fn list_multipart_uploads_query(
&self,
bucket_name: &str,
key_marker: &str,
upload_id_marker: &str,
prefix: &str,
delimiter: &str,
max_uploads: i64,
) -> Result<ListMultipartUploadsResult, std::io::Error> {
todo!();
}
pub fn list_object_parts(&self, bucket_name: &str, object_name: &str, upload_id: &str) -> Result<HashMap<i64, ObjectPart>, std::io::Error> {
pub fn list_object_parts(
&self,
bucket_name: &str,
object_name: &str,
upload_id: &str,
) -> Result<HashMap<i64, ObjectPart>, std::io::Error> {
todo!();
}
@@ -193,7 +239,14 @@ impl TransitionClient {
todo!();
}
pub async fn list_object_parts_query(&self, bucket_name: &str, object_name: &str, upload_id: &str, part_number_marker: i64, max_parts: i64) -> Result<ListObjectPartsResult, std::io::Error> {
pub async fn list_object_parts_query(
&self,
bucket_name: &str,
object_name: &str,
upload_id: &str,
part_number_marker: i64,
max_parts: i64,
) -> Result<ListObjectPartsResult, std::io::Error> {
todo!();
}
}
@@ -226,4 +279,4 @@ fn decode_s3_name(name: &str, encoding_type: &str) -> Result<String, std::io::Er
return Ok(name.to_string());
}
}
}
}
ecstore/src/client/api_put_object.rs
+98 -57
View File
@@ -1,33 +1,30 @@
#![allow(clippy::map_entry)]
use std::{collections::HashMap, sync::Arc};
use bytes::Bytes;
use time::{OffsetDateTime, macros::format_description, Duration};
use http::{HeaderMap, HeaderName, HeaderValue};
use std::{collections::HashMap, sync::Arc};
use time::{Duration, OffsetDateTime, macros::format_description};
use tracing::{error, info, warn};
use s3s::dto::{
ObjectLockRetentionMode, ObjectLockLegalHoldStatus,
ReplicationStatus,
};
use s3s::header::{
X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_OBJECT_LOCK_LEGAL_HOLD,
X_AMZ_STORAGE_CLASS, X_AMZ_WEBSITE_REDIRECT_LOCATION, X_AMZ_REPLICATION_STATUS,
};
use reader::hasher::Hasher;
//use crate::disk::{BufferReader, Reader};
use rustfs_utils::{
crypto::base64_encode,
net::{is_amz_header, is_standard_header, is_storageclass_header, is_rustfs_header, is_minio_header},
use s3s::dto::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ReplicationStatus};
use s3s::header::{
X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_REPLICATION_STATUS,
X_AMZ_STORAGE_CLASS, X_AMZ_WEBSITE_REDIRECT_LOCATION,
};
//use crate::disk::{BufferReader, Reader};
use crate::checksum::ChecksumMode;
use crate::client::{
api_s3_datatypes::{CompletePart, ObjectPart, CompleteMultipartUpload},
api_error_response::{err_entity_too_large, err_invalid_argument},
api_put_object_common::optimal_part_info,
transition_api::{TransitionClient, UploadInfo, ReaderImpl},
api_error_response::{err_invalid_argument, err_entity_too_large},
api_put_object_multipart::UploadPartParams,
api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ObjectPart},
constants::{ISO8601_DATEFORMAT, MAX_MULTIPART_PUT_OBJECT_SIZE, MIN_PART_SIZE, TOTAL_WORKERS},
credentials::SignatureType,
constants::{MAX_MULTIPART_PUT_OBJECT_SIZE, TOTAL_WORKERS, MIN_PART_SIZE, ISO8601_DATEFORMAT,},
transition_api::{ReaderImpl, TransitionClient, UploadInfo},
};
use rustfs_utils::{
crypto::base64_encode,
net::{is_amz_header, is_minio_header, is_rustfs_header, is_standard_header, is_storageclass_header},
};
#[derive(Debug, Clone)]
@@ -123,17 +120,21 @@ impl Default for PutObjectOptions {
impl PutObjectOptions {
fn set_matche_tag(&mut self, etag: &str) {
if etag == "*" {
self.custom_header.insert("If-Match", HeaderValue::from_str("*").expect("err"));
self.custom_header
.insert("If-Match", HeaderValue::from_str("*").expect("err"));
} else {
self.custom_header.insert("If-Match", HeaderValue::from_str(&format!("\"{}\"", etag)).expect("err"));
self.custom_header
.insert("If-Match", HeaderValue::from_str(&format!("\"{}\"", etag)).expect("err"));
}
}
fn set_matche_tag_except(&mut self, etag: &str) {
if etag == "*" {
self.custom_header.insert("If-None-Match", HeaderValue::from_str("*").expect("err"));
self.custom_header
.insert("If-None-Match", HeaderValue::from_str("*").expect("err"));
} else {
self.custom_header.insert("If-None-Match", HeaderValue::from_str(&format!("\"{etag}\"")).expect("err"));
self.custom_header
.insert("If-None-Match", HeaderValue::from_str(&format!("\"{etag}\"")).expect("err"));
}
}
@@ -160,7 +161,10 @@ impl PutObjectOptions {
}
if self.expires.unix_timestamp() != 0 {
header.insert("Expires", HeaderValue::from_str(&self.expires.format(ISO8601_DATEFORMAT).unwrap()).expect("err")); //rustfs invalid heade
header.insert(
"Expires",
HeaderValue::from_str(&self.expires.format(ISO8601_DATEFORMAT).unwrap()).expect("err"),
); //rustfs invalid heade
}
if self.mode.as_str() != "" {
@@ -168,7 +172,10 @@ impl PutObjectOptions {
}
if self.retain_until_date.unix_timestamp() != 0 {
header.insert(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, HeaderValue::from_str(&self.retain_until_date.format(ISO8601_DATEFORMAT).unwrap()).expect("err"));
header.insert(
X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE,
HeaderValue::from_str(&self.retain_until_date.format(ISO8601_DATEFORMAT).unwrap()).expect("err"),
);
}
if self.legalhold.as_str() != "" {
@@ -180,15 +187,22 @@ impl PutObjectOptions {
}
if self.website_redirect_location != "" {
header.insert(X_AMZ_WEBSITE_REDIRECT_LOCATION, HeaderValue::from_str(&self.website_redirect_location).expect("err"));
header.insert(
X_AMZ_WEBSITE_REDIRECT_LOCATION,
HeaderValue::from_str(&self.website_redirect_location).expect("err"),
);
}
if !self.internal.replication_status.as_str().is_empty() {
header.insert(X_AMZ_REPLICATION_STATUS, HeaderValue::from_str(self.internal.replication_status.as_str()).expect("err"));
header.insert(
X_AMZ_REPLICATION_STATUS,
HeaderValue::from_str(self.internal.replication_status.as_str()).expect("err"),
);
}
for (k, v) in &self.user_metadata {
if is_amz_header(k) || is_standard_header(k) || is_storageclass_header(k) || is_rustfs_header(k) || is_minio_header(k) {
if is_amz_header(k) || is_standard_header(k) || is_storageclass_header(k) || is_rustfs_header(k) || is_minio_header(k)
{
if let Ok(header_name) = HeaderName::from_bytes(k.as_bytes()) {
header.insert(header_name, HeaderValue::from_str(&v).unwrap());
}
@@ -208,12 +222,12 @@ impl PutObjectOptions {
fn validate(&self, c: TransitionClient) -> Result<(), std::io::Error> {
//if self.checksum.is_set() {
/*if !self.trailing_header_support {
return Err(Error::from(err_invalid_argument("Checksum requires Client with TrailingHeaders enabled")));
}*/
/*else if self.override_signer_type == SignatureType::SignatureV2 {
return Err(Error::from(err_invalid_argument("Checksum cannot be used with v2 signatures")));
}*/
/*if !self.trailing_header_support {
return Err(Error::from(err_invalid_argument("Checksum requires Client with TrailingHeaders enabled")));
}*/
/*else if self.override_signer_type == SignatureType::SignatureV2 {
return Err(Error::from(err_invalid_argument("Checksum cannot be used with v2 signatures")));
}*/
//}
Ok(())
@@ -221,19 +235,37 @@ impl PutObjectOptions {
}
impl TransitionClient {
pub async fn put_object(self: Arc<Self>, bucket_name: &str, object_name: &str, mut reader: ReaderImpl, object_size: i64,
opts: &PutObjectOptions
pub async fn put_object(
self: Arc<Self>,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
object_size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
if object_size < 0 && opts.disable_multipart {
return Err(std::io::Error::other("object size must be provided with disable multipart upload"));
}
self.put_object_common(bucket_name, object_name, reader, object_size, opts).await
self.put_object_common(bucket_name, object_name, reader, object_size, opts)
.await
}
pub async fn put_object_common(self: Arc<Self>, bucket_name: &str, object_name: &str, mut reader: ReaderImpl, size: i64, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn put_object_common(
self: Arc<Self>,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
if size > MAX_MULTIPART_PUT_OBJECT_SIZE {
return Err(std::io::Error::other(err_entity_too_large(size, MAX_MULTIPART_PUT_OBJECT_SIZE, bucket_name, object_name)));
return Err(std::io::Error::other(err_entity_too_large(
size,
MAX_MULTIPART_PUT_OBJECT_SIZE,
bucket_name,
object_name,
)));
}
let mut opts = opts.clone();
opts.auto_checksum.set_default(ChecksumMode::ChecksumCRC32C);
@@ -255,19 +287,30 @@ impl TransitionClient {
return Err(std::io::Error::other("no length provided and multipart disabled"));
}
if opts.concurrent_stream_parts && opts.num_threads > 1 {
return self.put_object_multipart_stream_parallel(bucket_name, object_name, reader, &opts).await;
return self
.put_object_multipart_stream_parallel(bucket_name, object_name, reader, &opts)
.await;
}
return self.put_object_multipart_stream_no_length(bucket_name, object_name, reader, &opts).await;
return self
.put_object_multipart_stream_no_length(bucket_name, object_name, reader, &opts)
.await;
}
if size <= part_size || opts.disable_multipart {
return self.put_object_gcs(bucket_name, object_name, reader, size, &opts).await;
}
self.put_object_multipart_stream(bucket_name, object_name, reader, size, &opts).await
self.put_object_multipart_stream(bucket_name, object_name, reader, size, &opts)
.await
}
pub async fn put_object_multipart_stream_no_length(&self, bucket_name: &str, object_name: &str, mut reader: ReaderImpl, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn put_object_multipart_stream_no_length(
&self,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut total_uploaded_size: i64 = 0;
let mut compl_multipart_upload = CompleteMultipartUpload::default();
@@ -295,12 +338,8 @@ impl TransitionClient {
while part_number <= total_parts_count {
buf = match &mut reader {
ReaderImpl::Body(content_body) => {
content_body.to_vec()
}
ReaderImpl::ObjectBody(content_body) => {
content_body.read_all().await?
}
ReaderImpl::Body(content_body) => content_body.to_vec(),
ReaderImpl::ObjectBody(content_body) => content_body.read_all().await?,
};
let length = buf.len();
@@ -354,13 +393,13 @@ impl TransitionClient {
let part = parts_info[&i].clone();
all_parts.push(part.clone());
compl_multipart_upload.parts.push(CompletePart {
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
checksum_crc64nvme: part.checksum_crc64nvme,
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
checksum_crc64nvme: part.checksum_crc64nvme,
..Default::default()
});
}
@@ -374,9 +413,11 @@ impl TransitionClient {
};
//apply_auto_checksum(&mut opts, all_parts);
let mut upload_info = self.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts).await?;
let mut upload_info = self
.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts)
.await?;
upload_info.size = total_uploaded_size;
Ok(upload_info)
}
}
}
ecstore/src/client/api_put_object_common.rs
+33 -12
View File
@@ -1,10 +1,10 @@
#![allow(clippy::map_entry)]
use crate::client::{
api_put_object::PutObjectOptions,
constants::{ABS_MIN_PART_SIZE, MAX_MULTIPART_PUT_OBJECT_SIZE, MAX_PARTS_COUNT, MAX_PART_SIZE, MIN_PART_SIZE},
transition_api::TransitionClient,
transition_api::ReaderImpl,
api_error_response::{err_entity_too_large, err_invalid_argument},
api_put_object::PutObjectOptions,
constants::{ABS_MIN_PART_SIZE, MAX_MULTIPART_PUT_OBJECT_SIZE, MAX_PART_SIZE, MAX_PARTS_COUNT, MIN_PART_SIZE},
transition_api::ReaderImpl,
transition_api::TransitionClient,
};
const NULL_VERSION_ID: &str = "null";
@@ -28,27 +28,43 @@ pub fn optimal_part_info(object_size: i64, configured_part_size: u64) -> Result<
}
if object_size > MAX_MULTIPART_PUT_OBJECT_SIZE {
return Err(std::io::Error::other(err_entity_too_large(object_size, MAX_MULTIPART_PUT_OBJECT_SIZE, "", "")));
return Err(std::io::Error::other(err_entity_too_large(
object_size,
MAX_MULTIPART_PUT_OBJECT_SIZE,
"",
"",
)));
}
let mut part_size_flt: f64;
if configured_part_size > 0 {
if configured_part_size as i64 > object_size {
return Err(std::io::Error::other(err_entity_too_large(configured_part_size as i64, object_size, "", "")));
return Err(std::io::Error::other(err_entity_too_large(
configured_part_size as i64,
object_size,
"",
"",
)));
}
if !unknown_size {
if object_size > (configured_part_size as i64 * MAX_PARTS_COUNT) {
return Err(std::io::Error::other(err_invalid_argument("Part size * max_parts(10000) is lesser than input objectSize.")));
return Err(std::io::Error::other(err_invalid_argument(
"Part size * max_parts(10000) is lesser than input objectSize.",
)));
}
}
if (configured_part_size as i64) < ABS_MIN_PART_SIZE {
return Err(std::io::Error::other(err_invalid_argument("Input part size is smaller than allowed minimum of 5MiB.")));
return Err(std::io::Error::other(err_invalid_argument(
"Input part size is smaller than allowed minimum of 5MiB.",
)));
}
if configured_part_size as i64 > MAX_PART_SIZE {
return Err(std::io::Error::other(err_invalid_argument("Input part size is bigger than allowed maximum of 5GiB.")));
return Err(std::io::Error::other(err_invalid_argument(
"Input part size is bigger than allowed maximum of 5GiB.",
)));
}
part_size_flt = configured_part_size as f64;
@@ -64,13 +80,18 @@ pub fn optimal_part_info(object_size: i64, configured_part_size: u64) -> Result<
let total_parts_count = (object_size as f64 / part_size_flt).ceil() as i64;
let part_size = part_size_flt.ceil() as i64;
let last_part_size = object_size - (total_parts_count-1) * part_size;
let last_part_size = object_size - (total_parts_count - 1) * part_size;
Ok((total_parts_count, part_size, last_part_size))
}
impl TransitionClient {
pub async fn new_upload_id(&self ,bucket_name: &str, object_name: &str, opts: &PutObjectOptions) -> Result<String, std::io::Error> {
pub async fn new_upload_id(
&self,
bucket_name: &str,
object_name: &str,
opts: &PutObjectOptions,
) -> Result<String, std::io::Error> {
let init_multipart_upload_result = self.initiate_multipart_upload(bucket_name, object_name, opts).await?;
Ok(init_multipart_upload_result.upload_id)
}
}
}
ecstore/src/client/api_put_object_multipart.rs
+157 -92
View File
@@ -1,50 +1,61 @@
#![allow(clippy::map_entry)]
use bytes::Bytes;
use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
use s3s::S3ErrorCode;
use std::io::Read;
use std::{collections::HashMap, sync::Arc};
use bytes::Bytes;
use s3s::S3ErrorCode;
use time::{format_description, OffsetDateTime};
use uuid::Uuid;
use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
use url::form_urlencoded::Serializer;
use time::{OffsetDateTime, format_description};
use tokio_util::sync::CancellationToken;
use tracing::warn;
use tracing::{error, info};
use url::form_urlencoded::Serializer;
use uuid::Uuid;
use s3s::{dto::StreamingBlob, Body};
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
use reader::hasher::Hasher;
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
use s3s::{Body, dto::StreamingBlob};
//use crate::disk::{Reader, BufferReader};
use crate::client::{
transition_api::{RequestMetadata, TransitionClient, UploadInfo, ReaderImpl,},
api_error_response::{err_entity_too_large, err_entity_too_small, err_invalid_argument, http_resp_to_error_response, to_error_response},
api_error_response::{
err_entity_too_large, err_entity_too_small, err_invalid_argument, http_resp_to_error_response, to_error_response,
},
api_put_object::PutObjectOptions,
api_put_object_common::optimal_part_info,
api_s3_datatypes::{CompleteMultipartUpload, CompleteMultipartUploadResult, CompletePart, InitiateMultipartUploadResult, ObjectPart},
constants::{ABS_MIN_PART_SIZE, MAX_PART_SIZE, MAX_SINGLE_PUT_OBJECT_SIZE, ISO8601_DATEFORMAT, },
};
use rustfs_utils::{
path::trim_etag,
crypto::base64_encode,
api_s3_datatypes::{
CompleteMultipartUpload, CompleteMultipartUploadResult, CompletePart, InitiateMultipartUploadResult, ObjectPart,
},
constants::{ABS_MIN_PART_SIZE, ISO8601_DATEFORMAT, MAX_PART_SIZE, MAX_SINGLE_PUT_OBJECT_SIZE},
transition_api::{ReaderImpl, RequestMetadata, TransitionClient, UploadInfo},
};
use crate::{
disk::DiskAPI,
store_api::{
GetObjectReader, StorageAPI,
},
checksum::ChecksumMode,
disk::DiskAPI,
store_api::{GetObjectReader, StorageAPI},
};
use rustfs_utils::{crypto::base64_encode, path::trim_etag};
impl TransitionClient {
pub async fn put_object_multipart(&self, bucket_name: &str, object_name: &str, mut reader: ReaderImpl, size: i64,
opts: &PutObjectOptions
pub async fn put_object_multipart(
&self,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let info = self.put_object_multipart_no_stream(bucket_name, object_name, &mut reader, opts).await;
let info = self
.put_object_multipart_no_stream(bucket_name, object_name, &mut reader, opts)
.await;
if let Err(err) = &info {
let err_resp = to_error_response(err);
if err_resp.code == S3ErrorCode::AccessDenied && err_resp.message.contains("Access Denied") {
if size > MAX_SINGLE_PUT_OBJECT_SIZE {
return Err(std::io::Error::other(err_entity_too_large(size, MAX_SINGLE_PUT_OBJECT_SIZE, bucket_name, object_name)));
return Err(std::io::Error::other(err_entity_too_large(
size,
MAX_SINGLE_PUT_OBJECT_SIZE,
bucket_name,
object_name,
)));
}
return self.put_object_gcs(bucket_name, object_name, reader, size, opts).await;
}
@@ -52,7 +63,13 @@ impl TransitionClient {
Ok(info?)
}
pub async fn put_object_multipart_no_stream(&self, bucket_name: &str, object_name: &str, reader: &mut ReaderImpl, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn put_object_multipart_no_stream(
&self,
bucket_name: &str,
object_name: &str,
reader: &mut ReaderImpl,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut total_uploaded_size: i64 = 0;
let mut compl_multipart_upload = CompleteMultipartUpload::default();
@@ -91,10 +108,10 @@ impl TransitionClient {
let mut sha256_hex: String;
//if hash_sums["md5"] != nil {
md5_base64 = base64_encode(&hash_sums["md5"]);
md5_base64 = base64_encode(&hash_sums["md5"]);
//}
//if hash_sums["sha256"] != nil {
sha256_hex = hex_simd::encode_to_string(hash_sums["sha256"].clone(), hex_simd::AsciiCase::Lower);
sha256_hex = hex_simd::encode_to_string(hash_sums["sha256"].clone(), hex_simd::AsciiCase::Lower);
//}
if hash_sums.len() == 0 {
let csum;
@@ -137,12 +154,12 @@ impl TransitionClient {
let part = parts_info[&i].clone();
all_parts.push(part.clone());
compl_multipart_upload.parts.push(CompletePart {
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
checksum_crc64nvme: part.checksum_crc64nvme,
..Default::default()
});
@@ -151,18 +168,25 @@ impl TransitionClient {
compl_multipart_upload.parts.sort();
let mut opts = PutObjectOptions {
//server_side_encryption: opts.server_side_encryption,
auto_checksum: opts.auto_checksum,
auto_checksum: opts.auto_checksum,
..Default::default()
};
//apply_auto_checksum(&mut opts, all_parts);
let mut upload_info = self.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts).await?;
let mut upload_info = self
.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts)
.await?;
upload_info.size = total_uploaded_size;
Ok(upload_info)
}
pub async fn initiate_multipart_upload(&self, bucket_name: &str, object_name: &str, opts: &PutObjectOptions) -> Result<InitiateMultipartUploadResult, std::io::Error> {
pub async fn initiate_multipart_upload(
&self,
bucket_name: &str,
object_name: &str,
opts: &PutObjectOptions,
) -> Result<InitiateMultipartUploadResult, std::io::Error> {
let mut url_values = HashMap::new();
url_values.insert("uploads".to_string(), "".to_string());
@@ -180,9 +204,9 @@ impl TransitionClient {
let mut req_metadata = RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: url_values,
query_values: url_values,
custom_header,
content_body: ReaderImpl::Body(Bytes::new()),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
content_sha256_hex: "".to_string(),
@@ -197,9 +221,9 @@ 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)));
}
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, object_name)));
}
//}
let initiate_multipart_upload_result = InitiateMultipartUploadResult::default();
Ok(initiate_multipart_upload_result)
@@ -207,13 +231,20 @@ impl TransitionClient {
pub async fn upload_part(&self, p: &mut UploadPartParams) -> Result<ObjectPart, std::io::Error> {
if p.size > MAX_PART_SIZE {
return Err(std::io::Error::other(err_entity_too_large(p.size, MAX_PART_SIZE, &p.bucket_name, &p.object_name)));
return Err(std::io::Error::other(err_entity_too_large(
p.size,
MAX_PART_SIZE,
&p.bucket_name,
&p.object_name,
)));
}
if p.size <= -1 {
return Err(std::io::Error::other(err_entity_too_small(p.size, &p.bucket_name, &p.object_name)));
}
if p.part_number <= 0 {
return Err(std::io::Error::other(err_invalid_argument("Part number cannot be negative or equal to zero.")));
return Err(std::io::Error::other(err_invalid_argument(
"Part number cannot be negative or equal to zero.",
)));
}
if p.upload_id == "" {
return Err(std::io::Error::other(err_invalid_argument("UploadID cannot be empty.")));
@@ -224,55 +255,85 @@ impl TransitionClient {
url_values.insert("uploadId".to_string(), p.upload_id.clone());
let buf = match &mut p.reader {
ReaderImpl::Body(content_body) => {
content_body.to_vec()
}
ReaderImpl::ObjectBody(content_body) => {
content_body.read_all().await?
}
ReaderImpl::Body(content_body) => content_body.to_vec(),
ReaderImpl::ObjectBody(content_body) => content_body.read_all().await?,
};
let mut req_metadata = RequestMetadata {
bucket_name: p.bucket_name.clone(),
object_name: p.object_name.clone(),
query_values: url_values,
custom_header: p.custom_header.clone(),
content_body: ReaderImpl::Body(Bytes::from(buf)),
content_length: p.size,
bucket_name: p.bucket_name.clone(),
object_name: p.object_name.clone(),
query_values: url_values,
custom_header: p.custom_header.clone(),
content_body: ReaderImpl::Body(Bytes::from(buf)),
content_length: p.size,
content_md5_base64: p.md5_base64.clone(),
content_sha256_hex: p.sha256_hex.clone(),
stream_sha256: p.stream_sha256,
trailer: p.trailer.clone(),
pre_sign_url: Default::default(),
stream_sha256: p.stream_sha256,
trailer: p.trailer.clone(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
};
let resp = self.execute_method(http::Method::PUT, &mut req_metadata).await?;
//defer closeResponse(resp)
//if resp.is_none() {
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], &p.bucket_name.clone(), &p.object_name)));
}
if resp.status() != StatusCode::OK {
return Err(std::io::Error::other(http_resp_to_error_response(
resp,
vec![],
&p.bucket_name.clone(),
&p.object_name,
)));
}
//}
let h = resp.headers();
let mut obj_part = ObjectPart {
checksum_crc32: if let Some(h_checksum_crc32) = h.get(ChecksumMode::ChecksumCRC32.key()) { h_checksum_crc32.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc32c: if let Some(h_checksum_crc32c) = h.get(ChecksumMode::ChecksumCRC32C.key()) { h_checksum_crc32c.to_str().expect("err").to_string() } else { "".to_string() },
checksum_sha1: if let Some(h_checksum_sha1) = h.get(ChecksumMode::ChecksumSHA1.key()) { h_checksum_sha1.to_str().expect("err").to_string() } else { "".to_string() },
checksum_sha256: if let Some(h_checksum_sha256) = h.get(ChecksumMode::ChecksumSHA256.key()) { h_checksum_sha256.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc64nvme: if let Some(h_checksum_crc64nvme) = h.get(ChecksumMode::ChecksumCRC64NVME.key()) { h_checksum_crc64nvme.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc32: if let Some(h_checksum_crc32) = h.get(ChecksumMode::ChecksumCRC32.key()) {
h_checksum_crc32.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_crc32c: if let Some(h_checksum_crc32c) = h.get(ChecksumMode::ChecksumCRC32C.key()) {
h_checksum_crc32c.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_sha1: if let Some(h_checksum_sha1) = h.get(ChecksumMode::ChecksumSHA1.key()) {
h_checksum_sha1.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_sha256: if let Some(h_checksum_sha256) = h.get(ChecksumMode::ChecksumSHA256.key()) {
h_checksum_sha256.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_crc64nvme: if let Some(h_checksum_crc64nvme) = h.get(ChecksumMode::ChecksumCRC64NVME.key()) {
h_checksum_crc64nvme.to_str().expect("err").to_string()
} else {
"".to_string()
},
..Default::default()
};
obj_part.size = p.size;
obj_part.part_num = p.part_number;
obj_part.etag = if let Some(h_etag) = h.get("ETag") { h_etag.to_str().expect("err").trim_matches('"').to_string() } else { "".to_string() };
obj_part.etag = if let Some(h_etag) = h.get("ETag") {
h_etag.to_str().expect("err").trim_matches('"').to_string()
} else {
"".to_string()
};
Ok(obj_part)
}
pub async fn complete_multipart_upload(&self, bucket_name: &str, object_name: &str, upload_id: &str,
complete: CompleteMultipartUpload, opts: &PutObjectOptions
pub async fn complete_multipart_upload(
&self,
bucket_name: &str,
object_name: &str,
upload_id: &str,
complete: CompleteMultipartUpload,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut url_values = HashMap::new();
url_values.insert("uploadId".to_string(), upload_id.to_string());
@@ -284,19 +345,19 @@ impl TransitionClient {
let mut req_metadata = RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: url_values,
content_body: ReaderImpl::Body(complete_multipart_upload_buffer),
content_length: 100,//complete_multipart_upload_bytes.len(),
content_sha256_hex: "".to_string(),//hex_simd::encode_to_string(complete_multipart_upload_bytes, hex_simd::AsciiCase::Lower),
custom_header: headers,
stream_sha256: Default::default(),
trailer: Default::default(),
query_values: url_values,
content_body: ReaderImpl::Body(complete_multipart_upload_buffer),
content_length: 100, //complete_multipart_upload_bytes.len(),
content_sha256_hex: "".to_string(), //hex_simd::encode_to_string(complete_multipart_upload_bytes, hex_simd::AsciiCase::Lower),
custom_header: headers,
stream_sha256: Default::default(),
trailer: Default::default(),
content_md5_base64: "".to_string(),
pre_sign_url: Default::default(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
};
let resp = self.execute_method(http::Method::POST, &mut req_metadata).await?;
@@ -307,7 +368,7 @@ impl TransitionClient {
let (exp_time, rule_id) = if let Some(h_x_amz_expiration) = resp.headers().get(X_AMZ_EXPIRATION) {
(
OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(),
"".to_string()
"".to_string(),
)
} else {
(OffsetDateTime::now_utc(), "".to_string())
@@ -315,17 +376,21 @@ impl TransitionClient {
let h = resp.headers();
Ok(UploadInfo {
bucket: complete_multipart_upload_result.bucket,
key: complete_multipart_upload_result.key,
etag: trim_etag(&complete_multipart_upload_result.etag),
version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) { h_x_amz_version_id.to_str().expect("err").to_string() } else { "".to_string() },
location: complete_multipart_upload_result.location,
expiration: exp_time,
bucket: complete_multipart_upload_result.bucket,
key: complete_multipart_upload_result.key,
etag: trim_etag(&complete_multipart_upload_result.etag),
version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) {
h_x_amz_version_id.to_str().expect("err").to_string()
} else {
"".to_string()
},
location: complete_multipart_upload_result.location,
expiration: exp_time,
expiration_rule_id: rule_id,
checksum_sha256: complete_multipart_upload_result.checksum_sha256,
checksum_sha1: complete_multipart_upload_result.checksum_sha1,
checksum_crc32: complete_multipart_upload_result.checksum_crc32,
checksum_crc32c: complete_multipart_upload_result.checksum_crc32c,
checksum_sha256: complete_multipart_upload_result.checksum_sha256,
checksum_sha1: complete_multipart_upload_result.checksum_sha1,
checksum_crc32: complete_multipart_upload_result.checksum_crc32,
checksum_crc32c: complete_multipart_upload_result.checksum_crc32c,
checksum_crc64nvme: complete_multipart_upload_result.checksum_crc64nvme,
..Default::default()
})
ecstore/src/client/api_put_object_streaming.rs
+150 -71
View File
@@ -1,28 +1,28 @@
#![allow(clippy::map_entry)]
use std::sync::RwLock;
use std::{collections::HashMap, sync::Arc};
use bytes::Bytes;
use futures::future::join_all;
use http::{HeaderMap, HeaderName, HeaderValue, StatusCode};
use time::{format_description, OffsetDateTime};
use std::sync::RwLock;
use std::{collections::HashMap, sync::Arc};
use time::{OffsetDateTime, format_description};
use tokio::{select, sync::mpsc};
use tokio_util::sync::CancellationToken;
use tracing::warn;
use uuid::Uuid;
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
use reader::hasher::Hasher;
use crate::checksum::{ChecksumMode, add_auto_checksum_headers, apply_auto_checksum};
use crate::client::{
constants::ISO8601_DATEFORMAT,
api_error_response::{err_invalid_argument, err_unexpected_eof, http_resp_to_error_response},
api_put_object::PutObjectOptions,
api_put_object_common::{is_object, optimal_part_info},
api_put_object_multipart::UploadPartParams,
api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ObjectPart},
transition_api::{TransitionClient, RequestMetadata, UploadInfo, ReaderImpl},
api_put_object_common::{is_object, optimal_part_info,},
api_error_response::{err_invalid_argument, http_resp_to_error_response, err_unexpected_eof},
constants::ISO8601_DATEFORMAT,
transition_api::{ReaderImpl, RequestMetadata, TransitionClient, UploadInfo},
};
use reader::hasher::Hasher;
use rustfs_utils::{crypto::base64_encode, path::trim_etag};
use crate::checksum::{add_auto_checksum_headers, apply_auto_checksum, ChecksumMode};
use s3s::header::{X_AMZ_EXPIRATION, X_AMZ_VERSION_ID};
pub struct UploadedPartRes {
pub error: std::io::Error,
@@ -37,23 +37,39 @@ pub struct UploadPartReq {
}
impl TransitionClient {
pub async fn put_object_multipart_stream(self: Arc<Self>, bucket_name: &str, object_name: &str,
mut reader: ReaderImpl, size: i64, opts: &PutObjectOptions
pub async fn put_object_multipart_stream(
self: Arc<Self>,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let info: UploadInfo;
if opts.concurrent_stream_parts && opts.num_threads > 1 {
info = self.put_object_multipart_stream_parallel(bucket_name, object_name, reader, opts).await?;
info = self
.put_object_multipart_stream_parallel(bucket_name, object_name, reader, opts)
.await?;
} else if !is_object(&reader) && !opts.send_content_md5 {
info = self.put_object_multipart_stream_from_readat(bucket_name, object_name, reader, size, opts).await?;
info = self
.put_object_multipart_stream_from_readat(bucket_name, object_name, reader, size, opts)
.await?;
} else {
info = self.put_object_multipart_stream_optional_checksum(bucket_name, object_name, reader, size, opts).await?;
info = self
.put_object_multipart_stream_optional_checksum(bucket_name, object_name, reader, size, opts)
.await?;
}
Ok(info)
}
pub async fn put_object_multipart_stream_from_readat(&self, bucket_name: &str, object_name: &str,
mut reader: ReaderImpl, size: i64, opts: &PutObjectOptions
pub async fn put_object_multipart_stream_from_readat(
&self,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let ret = optimal_part_info(size, opts.part_size)?;
let (total_parts_count, part_size, lastpart_size) = ret;
@@ -68,8 +84,13 @@ impl TransitionClient {
todo!();
}
pub async fn put_object_multipart_stream_optional_checksum(&self, bucket_name: &str, object_name: &str,
mut reader: ReaderImpl, size: i64, opts: &PutObjectOptions
pub async fn put_object_multipart_stream_optional_checksum(
&self,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut opts = opts.clone();
if opts.checksum.is_set() {
@@ -130,7 +151,7 @@ impl TransitionClient {
}
}
let hooked = ReaderImpl::Body(Bytes::from(buf));//newHook(BufferReader::new(buf), opts.progress);
let hooked = ReaderImpl::Body(Bytes::from(buf)); //newHook(BufferReader::new(buf), opts.progress);
let mut p = UploadPartParams {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
@@ -154,7 +175,12 @@ impl TransitionClient {
if size > 0 {
if total_uploaded_size != size {
return Err(std::io::Error::other(err_unexpected_eof(total_uploaded_size, size, bucket_name, object_name)));
return Err(std::io::Error::other(err_unexpected_eof(
total_uploaded_size,
size,
bucket_name,
object_name,
)));
}
}
@@ -167,12 +193,12 @@ impl TransitionClient {
all_parts.push(part.clone());
compl_multipart_upload.parts.push(CompletePart {
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
checksum_crc64nvme: part.checksum_crc64nvme,
});
}
@@ -181,18 +207,24 @@ impl TransitionClient {
let mut opts = PutObjectOptions {
//server_side_encryption: opts.server_side_encryption,
auto_checksum: opts.auto_checksum,
auto_checksum: opts.auto_checksum,
..Default::default()
};
apply_auto_checksum(&mut opts, &mut all_parts);
let mut upload_info = self.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts).await?;
let mut upload_info = self
.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts)
.await?;
upload_info.size = total_uploaded_size;
Ok(upload_info)
}
pub async fn put_object_multipart_stream_parallel(self: Arc<Self>, bucket_name: &str, object_name: &str,
mut reader: ReaderImpl/*GetObjectReader*/, opts: &PutObjectOptions
pub async fn put_object_multipart_stream_parallel(
self: Arc<Self>,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl, /*GetObjectReader*/
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut opts = opts.clone();
if opts.checksum.is_set() {
@@ -239,7 +271,11 @@ impl TransitionClient {
}
if buf.len() != part_size as usize {
return Err(std::io::Error::other(format!("read buffer < {} than expected partSize: {}", buf.len(), part_size)));
return Err(std::io::Error::other(format!(
"read buffer < {} than expected partSize: {}",
buf.len(),
part_size
)));
}
match &mut reader {
@@ -286,21 +322,21 @@ impl TransitionClient {
//defer wg.Done()
let mut p = UploadPartParams {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
upload_id: clone_upload_id,
reader: ReaderImpl::Body(Bytes::from(buf.clone())),
reader: ReaderImpl::Body(Bytes::from(buf.clone())),
part_number,
md5_base64,
size: length as i64,
size: length as i64,
//sse: opts.server_side_encryption,
stream_sha256: !opts.disable_content_sha256,
custom_header,
sha256_hex: "".to_string(),
trailer: HeaderMap::new(),
sha256_hex: "".to_string(),
trailer: HeaderMap::new(),
};
let obj_part = clone_self.upload_part(&mut p).await.expect("err");
let mut clone_parts_info = clone_parts_info.write().unwrap();
clone_parts_info.entry(part_number).or_insert(obj_part);
@@ -328,12 +364,12 @@ impl TransitionClient {
all_parts.push(part.clone());
compl_multipart_upload.parts.push(CompletePart {
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
etag: part.etag,
part_num: part.part_num,
checksum_crc32: part.checksum_crc32,
checksum_crc32c: part.checksum_crc32c,
checksum_sha1: part.checksum_sha1,
checksum_sha256: part.checksum_sha256,
checksum_crc64nvme: part.checksum_crc64nvme,
..Default::default()
});
@@ -343,41 +379,60 @@ impl TransitionClient {
let mut opts = PutObjectOptions {
//server_side_encryption: opts.server_side_encryption,
auto_checksum: opts.auto_checksum,
auto_checksum: opts.auto_checksum,
..Default::default()
};
apply_auto_checksum(&mut opts, &mut all_parts);
let mut upload_info = self.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts).await?;
let mut upload_info = self
.complete_multipart_upload(bucket_name, object_name, &upload_id, compl_multipart_upload, &opts)
.await?;
upload_info.size = total_uploaded_size;
Ok(upload_info)
}
pub async fn put_object_gcs(&self, bucket_name: &str, object_name: &str, mut reader: ReaderImpl, size: i64, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn put_object_gcs(
&self,
bucket_name: &str,
object_name: &str,
mut reader: ReaderImpl,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let mut opts = opts.clone();
if opts.checksum.is_set() {
opts.send_content_md5 = false;
}
let mut md5_base64: String = "".to_string();
let progress_reader = reader;//newHook(reader, opts.progress);
let progress_reader = reader; //newHook(reader, opts.progress);
self.put_object_do(bucket_name, object_name, progress_reader, &md5_base64, "", size, &opts).await
self.put_object_do(bucket_name, object_name, progress_reader, &md5_base64, "", size, &opts)
.await
}
pub async fn put_object_do(&self, bucket_name: &str, object_name: &str, reader: ReaderImpl, md5_base64: &str, sha256_hex: &str, size: i64, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn put_object_do(
&self,
bucket_name: &str,
object_name: &str,
reader: ReaderImpl,
md5_base64: &str,
sha256_hex: &str,
size: i64,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let custom_header = opts.header();
let mut req_metadata = RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
custom_header,
content_body: reader,
content_length: size,
content_body: reader,
content_length: size,
content_md5_base64: md5_base64.to_string(),
content_sha256_hex: sha256_hex.to_string(),
stream_sha256: !opts.disable_content_sha256,
stream_sha256: !opts.disable_content_sha256,
add_crc: Default::default(),
bucket_location: Default::default(),
pre_sign_url: Default::default(),
@@ -386,7 +441,7 @@ impl TransitionClient {
expires: Default::default(),
trailer: Default::default(),
};
let mut add_crc = false;//self.trailing_header_support && md5_base64 == "" && !s3utils.IsGoogleEndpoint(self.endpoint_url) && (opts.disable_content_sha256 || self.secure);
let mut add_crc = false; //self.trailing_header_support && md5_base64 == "" && !s3utils.IsGoogleEndpoint(self.endpoint_url) && (opts.disable_content_sha256 || self.secure);
let mut opts = opts.clone();
if opts.checksum.is_set() {
req_metadata.add_crc = opts.checksum;
@@ -422,26 +477,50 @@ impl TransitionClient {
let (exp_time, rule_id) = if let Some(h_x_amz_expiration) = resp.headers().get(X_AMZ_EXPIRATION) {
(
OffsetDateTime::parse(h_x_amz_expiration.to_str().unwrap(), ISO8601_DATEFORMAT).unwrap(),
"".to_string()
"".to_string(),
)
} else {
(OffsetDateTime::now_utc(), "".to_string())
};
let h = resp.headers();
Ok(UploadInfo {
bucket: bucket_name.to_string(),
key: object_name.to_string(),
etag: trim_etag(h.get("ETag").expect("err").to_str().expect("err")),
version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) { h_x_amz_version_id.to_str().expect("err").to_string() } else { "".to_string() },
size: size,
expiration: exp_time,
bucket: bucket_name.to_string(),
key: object_name.to_string(),
etag: trim_etag(h.get("ETag").expect("err").to_str().expect("err")),
version_id: if let Some(h_x_amz_version_id) = h.get(X_AMZ_VERSION_ID) {
h_x_amz_version_id.to_str().expect("err").to_string()
} else {
"".to_string()
},
size: size,
expiration: exp_time,
expiration_rule_id: rule_id,
checksum_crc32: if let Some(h_checksum_crc32) = h.get(ChecksumMode::ChecksumCRC32.key()) { h_checksum_crc32.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc32c: if let Some(h_checksum_crc32c) = h.get(ChecksumMode::ChecksumCRC32C.key()) { h_checksum_crc32c.to_str().expect("err").to_string() } else { "".to_string() },
checksum_sha1: if let Some(h_checksum_sha1) = h.get(ChecksumMode::ChecksumSHA1.key()) { h_checksum_sha1.to_str().expect("err").to_string() } else { "".to_string() },
checksum_sha256: if let Some(h_checksum_sha256) = h.get(ChecksumMode::ChecksumSHA256.key()) { h_checksum_sha256.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc64nvme: if let Some(h_checksum_crc64nvme) = h.get(ChecksumMode::ChecksumCRC64NVME.key()) { h_checksum_crc64nvme.to_str().expect("err").to_string() } else { "".to_string() },
checksum_crc32: if let Some(h_checksum_crc32) = h.get(ChecksumMode::ChecksumCRC32.key()) {
h_checksum_crc32.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_crc32c: if let Some(h_checksum_crc32c) = h.get(ChecksumMode::ChecksumCRC32C.key()) {
h_checksum_crc32c.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_sha1: if let Some(h_checksum_sha1) = h.get(ChecksumMode::ChecksumSHA1.key()) {
h_checksum_sha1.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_sha256: if let Some(h_checksum_sha256) = h.get(ChecksumMode::ChecksumSHA256.key()) {
h_checksum_sha256.to_str().expect("err").to_string()
} else {
"".to_string()
},
checksum_crc64nvme: if let Some(h_checksum_crc64nvme) = h.get(ChecksumMode::ChecksumCRC64NVME.key()) {
h_checksum_crc64nvme.to_str().expect("err").to_string()
} else {
"".to_string()
},
..Default::default()
})
}
}
}
ecstore/src/client/api_remove.rs
+229 -144
View File
@@ -1,26 +1,24 @@
#![allow(clippy::map_entry)]
use bytes::Bytes;
use http::{HeaderMap, HeaderValue, Method, StatusCode};
use s3s::S3ErrorCode;
use s3s::dto::ReplicationStatus;
use s3s::header::X_AMZ_BYPASS_GOVERNANCE_RETENTION;
use std::fmt::Display;
use std::{collections::HashMap, sync::Arc};
use bytes::Bytes;
use s3s::header::X_AMZ_BYPASS_GOVERNANCE_RETENTION;
use s3s::S3ErrorCode;
use time::OffsetDateTime;
use s3s::dto::ReplicationStatus;
use tokio::sync::mpsc::{self, Receiver, Sender};
use http::{HeaderMap, HeaderValue, Method, StatusCode};
use reader::hasher::{sum_sha256_hex, sum_md5_base64};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
use crate::client::{
api_error_response::{ErrorResponse, http_resp_to_error_response, to_error_response},
transition_api::{ReaderImpl, RequestMetadata, TransitionClient},
};
use crate::{
disk::DiskAPI,
store_api::{
GetObjectReader, ObjectInfo, StorageAPI,
},
};
use crate::client::{
transition_api::{TransitionClient, RequestMetadata, ReaderImpl},
api_error_response::{http_resp_to_error_response, to_error_response, ErrorResponse,},
store_api::{GetObjectReader, ObjectInfo, StorageAPI},
};
use reader::hasher::{sum_md5_base64, sum_sha256_hex};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
struct RemoveBucketOptions {
forced_elete: bool,
@@ -62,23 +60,28 @@ impl TransitionClient {
headers.insert(rustFSForceDelete, "true");
}*/
let resp = self.execute_method(Method::DELETE, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: headers,
object_name: "".to_string(),
query_values: Default::default(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
Method::DELETE,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: headers,
object_name: "".to_string(),
query_values: Default::default(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
{
let mut bucket_loc_cache = self.bucket_loc_cache.lock().unwrap();
@@ -88,23 +91,28 @@ impl TransitionClient {
}
pub async fn remove_bucket(&self, bucket_name: &str) -> Result<(), std::io::Error> {
let resp = self.execute_method(http::Method::DELETE, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: Default::default(),
object_name: "".to_string(),
query_values: Default::default(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::DELETE,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: Default::default(),
object_name: "".to_string(),
query_values: Default::default(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
{
let mut bucket_loc_cache = self.bucket_loc_cache.lock().unwrap();
@@ -119,7 +127,12 @@ impl TransitionClient {
res.err
}
pub async fn remove_object_inner(&self, bucket_name: &str, object_name: &str, opts: RemoveObjectOptions) -> Result<RemoveObjectResult, std::io::Error> {
pub async fn remove_object_inner(
&self,
bucket_name: &str,
object_name: &str,
opts: RemoveObjectOptions,
) -> Result<RemoveObjectResult, std::io::Error> {
let mut url_values = HashMap::new();
if opts.version_id != "" {
@@ -129,49 +142,72 @@ impl TransitionClient {
let mut headers = HeaderMap::new();
if opts.governance_bypass {
headers.insert(X_AMZ_BYPASS_GOVERNANCE_RETENTION, "true".parse().expect("err"));//amzBypassGovernance
headers.insert(X_AMZ_BYPASS_GOVERNANCE_RETENTION, "true".parse().expect("err")); //amzBypassGovernance
}
let resp = self.execute_method(http::Method::DELETE, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
query_values: url_values,
custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::DELETE,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
query_values: url_values,
custom_header: headers,
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
Ok(RemoveObjectResult {
object_name: object_name.to_string(),
object_version_id: opts.version_id,
delete_marker: resp.headers().get("x-amz-delete-marker").expect("err") == "true",
delete_marker_version_id: resp.headers().get("x-amz-version-id").expect("err").to_str().expect("err").to_string(),
object_version_id: opts.version_id,
delete_marker: resp.headers().get("x-amz-delete-marker").expect("err") == "true",
delete_marker_version_id: resp
.headers()
.get("x-amz-version-id")
.expect("err")
.to_str()
.expect("err")
.to_string(),
..Default::default()
})
}
pub async fn remove_objects_with_result(self: Arc<Self>, bucket_name: &str, objects_rx: Receiver<ObjectInfo>, opts: RemoveObjectsOptions) -> Receiver<RemoveObjectResult> {
pub async fn remove_objects_with_result(
self: Arc<Self>,
bucket_name: &str,
objects_rx: Receiver<ObjectInfo>,
opts: RemoveObjectsOptions,
) -> Receiver<RemoveObjectResult> {
let (result_tx, mut result_rx) = mpsc::channel(1);
let self_clone = Arc::clone(&self);
let bucket_name_owned = bucket_name.to_string();
tokio::spawn(async move {
self_clone.remove_objects_inner(&bucket_name_owned, objects_rx, &result_tx, opts).await;
self_clone
.remove_objects_inner(&bucket_name_owned, objects_rx, &result_tx, opts)
.await;
});
result_rx
}
pub async fn remove_objects(self: Arc<Self>, bucket_name: &str, objects_rx: Receiver<ObjectInfo>, opts: RemoveObjectsOptions) -> Receiver<RemoveObjectError> {
pub async fn remove_objects(
self: Arc<Self>,
bucket_name: &str,
objects_rx: Receiver<ObjectInfo>,
opts: RemoveObjectsOptions,
) -> Receiver<RemoveObjectError> {
let (error_tx, mut error_rx) = mpsc::channel(1);
let self_clone = Arc::clone(&self);
@@ -179,26 +215,36 @@ impl TransitionClient {
let (result_tx, mut result_rx) = mpsc::channel(1);
tokio::spawn(async move {
self_clone.remove_objects_inner(&bucket_name_owned, objects_rx, &result_tx, opts).await;
self_clone
.remove_objects_inner(&bucket_name_owned, objects_rx, &result_tx, opts)
.await;
});
tokio::spawn(async move {
while let Some(res) = result_rx.recv().await {
if res.err.is_none() {
continue;
}
error_tx.send(RemoveObjectError {
object_name: res.object_name,
version_id: res.object_version_id,
err: res.err,
..Default::default()
}).await;
error_tx
.send(RemoveObjectError {
object_name: res.object_name,
version_id: res.object_version_id,
err: res.err,
..Default::default()
})
.await;
}
});
error_rx
}
pub async fn remove_objects_inner(&self, bucket_name: &str, mut objects_rx: Receiver<ObjectInfo>, result_tx: &Sender<RemoveObjectResult>, opts: RemoveObjectsOptions) -> Result<(), std::io::Error> {
pub async fn remove_objects_inner(
&self,
bucket_name: &str,
mut objects_rx: Receiver<ObjectInfo>,
result_tx: &Sender<RemoveObjectResult>,
opts: RemoveObjectsOptions,
) -> Result<(), std::io::Error> {
let max_entries = 1000;
let mut finish = false;
let mut url_values = HashMap::new();
@@ -213,11 +259,17 @@ impl TransitionClient {
while let Some(object) = objects_rx.recv().await {
if has_invalid_xml_char(&object.name) {
let remove_result = self.remove_object_inner(bucket_name, &object.name, RemoveObjectOptions {
version_id: object.version_id.expect("err").to_string(),
governance_bypass: opts.governance_bypass,
..Default::default()
}).await?;
let remove_result = self
.remove_object_inner(
bucket_name,
&object.name,
RemoveObjectOptions {
version_id: object.version_id.expect("err").to_string(),
governance_bypass: opts.governance_bypass,
..Default::default()
},
)
.await?;
let remove_result_clone = remove_result.clone();
if !remove_result.err.is_none() {
match to_error_response(&remove_result.err.expect("err")).code {
@@ -241,7 +293,7 @@ impl TransitionClient {
}
if count == 0 {
break;
}
}
if count < max_entries {
finish = true;
}
@@ -252,23 +304,28 @@ impl TransitionClient {
}
let remove_bytes = generate_remove_multi_objects_request(&batch);
let resp = self.execute_method(http::Method::POST, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values.clone(),
content_body: ReaderImpl::Body(Bytes::from(remove_bytes.clone())),
content_length: remove_bytes.len() as i64,
content_md5_base64: sum_md5_base64(&remove_bytes),
content_sha256_hex: sum_sha256_hex(&remove_bytes),
custom_header: headers,
object_name: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::POST,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
query_values: url_values.clone(),
content_body: ReaderImpl::Body(Bytes::from(remove_bytes.clone())),
content_length: remove_bytes.len() as i64,
content_md5_base64: sum_md5_base64(&remove_bytes),
content_sha256_hex: sum_sha256_hex(&remove_bytes),
custom_header: headers,
object_name: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
let body_bytes: Vec<u8> = resp.body().bytes().expect("err").to_vec();
process_remove_multi_objects_response(ReaderImpl::Body(Bytes::from(body_bytes)), result_tx.clone());
@@ -285,49 +342,77 @@ impl TransitionClient {
Ok(())
}
pub async fn abort_multipart_upload(&self, bucket_name: &str, object_name: &str, upload_id: &str) -> Result<(), std::io::Error> {
pub async fn abort_multipart_upload(
&self,
bucket_name: &str,
object_name: &str,
upload_id: &str,
) -> Result<(), std::io::Error> {
let mut url_values = HashMap::new();
url_values.insert("uploadId".to_string(), upload_id.to_string());
let resp = self.execute_method(http::Method::DELETE, &mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
}).await?;
let resp = self
.execute_method(
http::Method::DELETE,
&mut RequestMetadata {
bucket_name: bucket_name.to_string(),
object_name: object_name.to_string(),
query_values: url_values,
content_sha256_hex: EMPTY_STRING_SHA256_HASH.to_string(),
custom_header: HeaderMap::new(),
content_body: ReaderImpl::Body(Bytes::new()),
content_length: 0,
content_md5_base64: "".to_string(),
stream_sha256: false,
trailer: HeaderMap::new(),
pre_sign_url: Default::default(),
add_crc: Default::default(),
extra_pre_sign_header: Default::default(),
bucket_location: Default::default(),
expires: Default::default(),
},
)
.await?;
//if resp.is_some() {
if resp.status() != StatusCode::NO_CONTENT {
let error_response: ErrorResponse;
match resp.status() {
StatusCode::NOT_FOUND => {
error_response = ErrorResponse {
code: S3ErrorCode::NoSuchUpload,
message: "The specified multipart upload does not exist.".to_string(),
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
request_id: resp.headers().get("x-amz-request-id").expect("err").to_str().expect("err").to_string(),
host_id: resp.headers().get("x-amz-id-2").expect("err").to_str().expect("err").to_string(),
region: resp.headers().get("x-amz-bucket-region").expect("err").to_str().expect("err").to_string(),
..Default::default()
};
}
_ => {
return Err(std::io::Error::other(http_resp_to_error_response(resp, vec![], bucket_name, object_name)));
}
if resp.status() != StatusCode::NO_CONTENT {
let error_response: ErrorResponse;
match resp.status() {
StatusCode::NOT_FOUND => {
error_response = ErrorResponse {
code: S3ErrorCode::NoSuchUpload,
message: "The specified multipart upload does not exist.".to_string(),
bucket_name: bucket_name.to_string(),
key: object_name.to_string(),
request_id: resp
.headers()
.get("x-amz-request-id")
.expect("err")
.to_str()
.expect("err")
.to_string(),
host_id: resp
.headers()
.get("x-amz-id-2")
.expect("err")
.to_str()
.expect("err")
.to_string(),
region: resp
.headers()
.get("x-amz-bucket-region")
.expect("err")
.to_str()
.expect("err")
.to_string(),
..Default::default()
};
}
_ => {
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));
}
return Err(std::io::Error::other(error_response));
}
//}
Ok(())
}
@@ -336,8 +421,8 @@ impl TransitionClient {
#[derive(Debug, Default)]
struct RemoveObjectError {
object_name: String,
version_id: String,
err: Option<std::io::Error>,
version_id: String,
err: Option<std::io::Error>,
}
impl Display for RemoveObjectError {
@@ -367,7 +452,7 @@ impl Clone for RemoveObjectResult {
delete_marker_version_id: self.delete_marker_version_id.clone(),
err: None, //err
}
}
}
}
pub struct RemoveObjectsOptions {
ecstore/src/client/api_s3_datatypes.rs
+59 -59
View File
@@ -1,12 +1,12 @@
#![allow(clippy::map_entry)]
use std::collections::HashMap;
use s3s::dto::Owner;
use serde::{Serialize, Deserialize};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use time::OffsetDateTime;
use rustfs_utils::crypto::base64_decode;
use crate::checksum::ChecksumMode;
use crate::client::transition_api::ObjectMultipartInfo;
use rustfs_utils::crypto::base64_decode;
use super::transition_api;
@@ -31,15 +31,15 @@ pub struct ListBucketV2Result {
pub name: String,
pub next_continuation_token: String,
pub continuation_token: String,
pub prefix: String,
pub prefix: String,
pub fetch_owner: String,
pub start_after: String,
}
pub struct Version {
etag: String,
is_latest: bool,
key: String,
etag: String,
is_latest: bool,
key: String,
last_modified: OffsetDateTime,
owner: Owner,
size: i64,
@@ -53,15 +53,15 @@ pub struct Version {
pub struct ListVersionsResult {
versions: Vec<Version>,
common_prefixes: Vec<CommonPrefix>,
name: String,
prefix: String,
delimiter: String,
max_keys: i64,
encoding_type: String,
is_truncated: bool,
key_marker: String,
version_id_marker: String,
next_key_marker: String,
name: String,
prefix: String,
delimiter: String,
max_keys: i64,
encoding_type: String,
is_truncated: bool,
key_marker: String,
version_id_marker: String,
next_key_marker: String,
next_version_id_marker: String,
}
@@ -77,35 +77,35 @@ pub struct ListBucketResult {
delimiter: String,
encoding_type: String,
is_truncated: bool,
marker: String,
max_keys: i64,
name: String,
next_marker: String,
prefix: String,
marker: String,
max_keys: i64,
name: String,
next_marker: String,
prefix: String,
}
pub struct ListMultipartUploadsResult {
bucket: String,
key_marker: String,
upload_id_marker: String,
next_key_marker: String,
bucket: String,
key_marker: String,
upload_id_marker: String,
next_key_marker: String,
next_upload_id_marker: String,
encoding_type: String,
max_uploads: i64,
is_truncated: bool,
uploads: Vec<ObjectMultipartInfo>,
prefix: String,
delimiter: String,
common_prefixes: Vec<CommonPrefix>,
encoding_type: String,
max_uploads: i64,
is_truncated: bool,
uploads: Vec<ObjectMultipartInfo>,
prefix: String,
delimiter: String,
common_prefixes: Vec<CommonPrefix>,
}
pub struct Initiator {
id: String,
id: String,
display_name: String,
}
pub struct CopyObjectResult {
pub etag: String,
pub etag: String,
pub last_modified: OffsetDateTime,
}
@@ -115,10 +115,10 @@ pub struct ObjectPart {
pub part_num: i64,
pub last_modified: OffsetDateTime,
pub size: i64,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc64nvme: String,
}
@@ -129,14 +129,13 @@ impl Default for ObjectPart {
part_num: 0,
last_modified: OffsetDateTime::now_utc(),
size: 0,
checksum_crc32: Default::default(),
checksum_crc32c: Default::default(),
checksum_sha1: Default::default(),
checksum_sha256: Default::default(),
checksum_crc32: Default::default(),
checksum_crc32c: Default::default(),
checksum_sha1: Default::default(),
checksum_sha256: Default::default(),
checksum_crc64nvme: Default::default(),
}
}
}
impl ObjectPart {
@@ -216,15 +215,15 @@ pub struct CompleteMultipartUploadResult {
pub checksum_crc64nvme: String,
}
#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
#[derive(serde::Serialize)]
pub struct CompletePart { //api has
#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, serde::Serialize)]
pub struct CompletePart {
//api has
pub etag: String,
pub part_num: i64,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc64nvme: String,
}
@@ -258,8 +257,7 @@ pub struct CopyObjectPartResult {
pub last_modified: OffsetDateTime,
}
#[derive(Debug, Default)]
#[derive(serde::Serialize)]
#[derive(Debug, Default, serde::Serialize)]
pub struct CompleteMultipartUpload {
pub parts: Vec<CompletePart>,
}
@@ -287,12 +285,14 @@ pub struct CreateBucketConfiguration {
}
#[derive(serde::Serialize)]
pub struct DeleteObject { //api has
pub struct DeleteObject {
//api has
pub key: String,
pub version_id: String,
}
pub struct DeletedObject { //s3s has
pub struct DeletedObject {
//s3s has
pub key: String,
pub version_id: String,
pub deletemarker: bool,
@@ -300,15 +300,15 @@ pub struct DeletedObject { //s3s has
}
pub struct NonDeletedObject {
pub key: String,
pub code: String,
pub message: String,
pub key: String,
pub code: String,
pub message: String,
pub version_id: String,
}
#[derive(serde::Serialize)]
pub struct DeleteMultiObjects {
pub quiet: bool,
pub quiet: bool,
pub objects: Vec<DeleteObject>,
}
@@ -331,6 +331,6 @@ impl DeleteMultiObjects {
}
pub struct DeleteMultiObjectsResult {
pub deleted_objects: Vec<DeletedObject>,
pub deleted_objects: Vec<DeletedObject>,
pub undeleted_objects: Vec<NonDeletedObject>,
}
ecstore/src/client/bucket_cache.rs
+22 -25
View File
@@ -1,20 +1,20 @@
#![allow(clippy::map_entry)]
use http::Request;
use hyper::StatusCode;
use hyper::body::Incoming;
use std::{collections::HashMap, sync::Arc};
use tracing::warn;
use tracing::{error, info, debug};
use hyper::StatusCode;
use tracing::{debug, error, info};
use reader::hasher::{Hasher, Sha256};
use s3s::S3ErrorCode;
use s3s::Body;
use crate::client::{
api_error_response::{http_resp_to_error_response, to_error_response},
transition_api::{TransitionClient, Document},
transition_api::{Document, TransitionClient},
};
use crate::signer;
use reader::hasher::{Hasher, Sha256};
use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
use s3s::Body;
use s3s::S3ErrorCode;
use super::constants::UNSIGNED_PAYLOAD;
use super::credentials::SignatureType;
@@ -25,9 +25,7 @@ pub struct BucketLocationCache {
impl BucketLocationCache {
pub fn new() -> BucketLocationCache {
BucketLocationCache{
items: HashMap::new(),
}
BucketLocationCache { items: HashMap::new() }
}
pub fn get(&self, bucket_name: &str) -> Option<String> {
@@ -50,7 +48,7 @@ impl TransitionClient {
async fn get_bucket_location_inner(&self, bucket_name: &str) -> Result<String, std::io::Error> {
if self.region != "" {
return Ok(self.region.clone())
return Ok(self.region.clone());
}
let mut location;
@@ -81,11 +79,7 @@ impl TransitionClient {
let mut target_url = self.endpoint_url.clone();
let scheme = self.endpoint_url.scheme();
let h = target_url.host().expect("host is none.");
let default_port = if scheme == "https" {
443
} else {
80
};
let default_port = if scheme == "https" { 443 } else { 80 };
let p = target_url.port().unwrap_or(default_port);
let is_virtual_style = self.is_virtual_host_style_request(&target_url, bucket_name);
@@ -128,9 +122,9 @@ impl TransitionClient {
}
let mut signer_type = value.signer_type.clone();
let mut access_key_id = value.access_key_id;
let mut secret_access_key = value.secret_access_key;
let mut session_token = value.session_token;
let mut access_key_id = value.access_key_id;
let mut secret_access_key = value.secret_access_key;
let mut session_token = value.session_token;
if self.override_signer_type != SignatureType::SignatureDefault {
signer_type = self.override_signer_type.clone();
@@ -164,7 +158,10 @@ impl TransitionClient {
content_sha256 = UNSIGNED_PAYLOAD.to_string();
}
req_builder.headers_mut().expect("err").insert("X-Amz-Content-Sha256", content_sha256.parse().unwrap());
req_builder
.headers_mut()
.expect("err")
.insert("X-Amz-Content-Sha256", content_sha256.parse().unwrap());
let req_builder = signer::sign_v4(req_builder, 0, &access_key_id, &secret_access_key, &session_token, "us-east-1");
let req = match req_builder.body(Body::empty()) {
Ok(req) => return Ok(req),
@@ -177,9 +174,9 @@ 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, "");
match err_resp.code {
if resp.status() != StatusCode::OK {
let err_resp = http_resp_to_error_response(resp, vec![], bucket_name, "");
match err_resp.code {
S3ErrorCode::NotImplemented => {
match err_resp.server.as_str() {
"AmazonSnowball" => {
@@ -205,13 +202,13 @@ async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket
return Err(std::io::Error::other(err_resp));
}
}
}
}
//}
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 mut location = location_constraint;
let mut location = location_constraint;
if location == "" {
location = "us-east-1".to_string();
}
@@ -221,4 +218,4 @@ async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket
}
Ok(location)
}
}
ecstore/src/client/constants.rs
+5 -4
View File
@@ -1,7 +1,7 @@
#![allow(clippy::map_entry)]
use std::{collections::HashMap, sync::Arc};
use time::{macros::format_description, format_description::FormatItem};
use lazy_static::lazy_static;
use std::{collections::HashMap, sync::Arc};
use time::{format_description::FormatItem, macros::format_description};
pub const ABS_MIN_PART_SIZE: i64 = 1024 * 1024 * 5;
pub const MAX_PARTS_COUNT: i64 = 10000;
@@ -16,8 +16,9 @@ pub const UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
pub const TOTAL_WORKERS: i64 = 4;
pub const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
pub const ISO8601_DATEFORMAT: &[FormatItem<'_>] = format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]Z");
pub const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
pub const ISO8601_DATEFORMAT: &[FormatItem<'_>] =
format_description!("[year]-[month]-[day]T[hour]:[minute]:[second].[subsecond]Z");
const GetObjectAttributesTags: &str = "ETag,Checksum,StorageClass,ObjectSize,ObjectParts";
const GetObjectAttributesMaxParts: i64 = 1000;
ecstore/src/client/credentials.rs
+13 -14
View File
@@ -19,8 +19,7 @@ pub struct Credentials<P: Provider + Default> {
provider: P,
}
impl<P: Provider + Default> Credentials<P>
{
impl<P: Provider + Default> Credentials<P> {
pub fn new(provider: P) -> Self {
Self {
provider: provider,
@@ -109,8 +108,8 @@ impl Provider for Static {
#[derive(Debug, Clone, Default)]
pub struct STSError {
pub r#type: String,
pub code: String,
pub r#type: String,
pub code: String,
pub message: String,
}
@@ -128,21 +127,21 @@ impl Display for ErrorResponse {
impl ErrorResponse {
fn error(&self) -> String {
if self.sts_error.message == "" {
return format!("Error response code {}.", self.sts_error.code);
}
return self.sts_error.message.clone();
if self.sts_error.message == "" {
return format!("Error response code {}.", self.sts_error.code);
}
return self.sts_error.message.clone();
}
}
struct Error {
code: String,
message: String,
code: String,
message: String,
bucket_name: String,
key: String,
resource: String,
request_id: String,
host_id: String,
key: String,
resource: String,
request_id: String,
host_id: String,
region: String,
server: String,
status_code: i64,
ecstore/src/client/mod.rs
+11 -11
View File
@@ -1,18 +1,18 @@
pub mod constants;
pub mod transition_api;
pub mod api_list;
pub mod api_error_response;
pub mod api_s3_datatypes;
pub mod admin_handler_utils;
pub mod api_bucket_policy;
pub mod api_put_object_common;
pub mod api_get_options;
pub mod api_error_response;
pub mod api_get_object;
pub mod api_get_options;
pub mod api_list;
pub mod api_put_object;
pub mod api_put_object_streaming;
pub mod api_put_object_common;
pub mod api_put_object_multipart;
pub mod api_put_object_streaming;
pub mod api_remove;
pub mod api_s3_datatypes;
pub mod bucket_cache;
pub mod constants;
pub mod credentials;
pub mod object_api_utils;
pub mod object_handlers_common;
pub mod admin_handler_utils;
pub mod credentials;
pub mod bucket_cache;
pub mod transition_api;
ecstore/src/client/object_api_utils.rs
+42 -31
View File
@@ -1,17 +1,14 @@
#![allow(clippy::map_entry)]
use std::{collections::HashMap, sync::Arc};
use http::HeaderMap;
use tokio::io::BufReader;
use std::io::Cursor;
use std::{collections::HashMap, sync::Arc};
use tokio::io::BufReader;
use s3s::S3ErrorCode;
use crate::store_api::{
GetObjectReader, HTTPRangeSpec,
ObjectInfo, ObjectOptions,
};
use crate::error::ErrorResponse;
use crate::store_api::{GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions};
use rustfs_filemeta::fileinfo::ObjectPartInfo;
use rustfs_rio::HashReader;
use crate::error::ErrorResponse;
use s3s::S3ErrorCode;
//#[derive(Clone)]
pub struct PutObjReader {
@@ -57,7 +54,11 @@ fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRa
i += 1;
}
Some(HTTPRangeSpec {start: start as usize, end: Some(end as usize), is_suffix_length: false})
Some(HTTPRangeSpec {
start: start as usize,
end: Some(end as usize),
is_suffix_length: false,
})
}
fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u64) {
@@ -71,7 +72,7 @@ fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u
for (i, part) in oi.parts.iter().enumerate() {
cumulative_actual_size += part.actual_size as i64;
if cumulative_actual_size <= offset {
compressed_offset += part.size as i64;
compressed_offset += part.size as i64;
} else {
first_part_idx = i as i64;
skip_length = cumulative_actual_size - part.actual_size as i64;
@@ -81,23 +82,32 @@ fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u
skip_length = offset - skip_length;
let parts: &[ObjectPartInfo] = &oi.parts;
if skip_length > 0 && parts.len() > first_part_idx as usize && parts[first_part_idx as usize].index.as_ref().expect("err").len() > 0 {
if skip_length > 0
&& parts.len() > first_part_idx as usize
&& parts[first_part_idx as usize].index.as_ref().expect("err").len() > 0
{
todo!();
}
(compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num)
}
pub fn new_getobjectreader(rs: HTTPRangeSpec, oi: &ObjectInfo, opts: &ObjectOptions, h: &HeaderMap) -> Result<(ObjReaderFn, i64, i64), ErrorResponse> {
pub fn new_getobjectreader(
rs: HTTPRangeSpec,
oi: &ObjectInfo,
opts: &ObjectOptions,
h: &HeaderMap,
) -> Result<(ObjReaderFn, i64, i64), ErrorResponse> {
//let (_, mut is_encrypted) = crypto.is_encrypted(oi.user_defined)?;
let mut is_encrypted = false;
let is_compressed = false;//oi.is_compressed_ok();
let is_compressed = false; //oi.is_compressed_ok();
let mut get_fn: ObjReaderFn;
let (off, length) = match rs.get_offset_length(oi.size) {
Ok(x) => x,
Err(err) => return Err(ErrorResponse {
let (off, length) = match rs.get_offset_length(oi.size) {
Ok(x) => x,
Err(err) => {
return Err(ErrorResponse {
code: S3ErrorCode::InvalidRange,
message: err.to_string(),
key: None,
@@ -105,26 +115,27 @@ pub fn new_getobjectreader(rs: HTTPRangeSpec, oi: &ObjectInfo, opts: &ObjectOpti
region: None,
request_id: None,
host_id: "".to_string(),
}),
});
}
};
get_fn = Arc::new(move |input_reader: BufReader<Cursor<Vec<u8>>>, _: HeaderMap| {
//Box::pin({
/*let r = GetObjectReader {
object_info: oi.clone(),
stream: StreamingBlob::new(HashReader::new(input_reader, 10, None, None, 10)),
};
get_fn = Arc::new(move |input_reader: BufReader<Cursor<Vec<u8>>>, _: HeaderMap| {
//Box::pin({
/*let r = GetObjectReader {
object_info: oi.clone(),
stream: StreamingBlob::new(HashReader::new(input_reader, 10, None, None, 10)),
};
r*/
todo!();
//})
});
r*/
todo!();
//})
});
Ok((get_fn, off as i64, length as i64))
}
pub fn extract_etag(metadata: &HashMap<String, String>) -> String {
if let Some(etag) = metadata.get("etag") {
if let Some(etag) = metadata.get("etag") {
etag.clone()
} else {
metadata["md5Sum"].clone()
}
}
metadata["md5Sum"].clone()
}
}
ecstore/src/client/object_handlers_common.rs
+16 -10
View File
@@ -1,15 +1,17 @@
use lock::local_locker::MAX_DELETE_LIST;
use crate::StorageAPI;
use crate::bucket::lifecycle::lifecycle;
use crate::bucket::versioning::VersioningApi;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::store::ECStore;
use crate::store_api::{ObjectOptions, ObjectToDelete,};
use crate::StorageAPI;
use crate::bucket::lifecycle::lifecycle;
use crate::store_api::{ObjectOptions, ObjectToDelete};
use lock::local_locker::MAX_DELETE_LIST;
pub async fn delete_object_versions(api: ECStore, bucket: &str, to_del: &[ObjectToDelete], lc_event: lifecycle::Event) {
let mut remaining = to_del;
loop {
if remaining.len() <= 0 {break};
if remaining.len() <= 0 {
break;
};
let mut to_del = remaining;
if to_del.len() > MAX_DELETE_LIST {
remaining = &to_del[MAX_DELETE_LIST..];
@@ -18,10 +20,14 @@ pub async fn delete_object_versions(api: ECStore, bucket: &str, to_del: &[Object
remaining = &[];
}
let vc = BucketVersioningSys::get(bucket).await.expect("err!");
let deleted_objs = api.delete_objects(bucket, to_del.to_vec(), ObjectOptions {
//prefix_enabled_fn: vc.prefix_enabled(""),
version_suspended: vc.suspended(),
..Default::default()
});
let deleted_objs = api.delete_objects(
bucket,
to_del.to_vec(),
ObjectOptions {
//prefix_enabled_fn: vc.prefix_enabled(""),
version_suspended: vc.suspended(),
..Default::default()
},
);
}
}
ecstore/src/client/transition_api.rs
+285 -148
View File
@@ -1,61 +1,63 @@
#![allow(clippy::map_entry)]
use std::pin::Pin;
use bytes::Bytes;
use futures::Future;
use http::{HeaderMap, HeaderName};
use serde::{Serialize, Deserialize};
use uuid::Uuid;
use http::{
HeaderValue, Response, StatusCode,
request::{Builder, Request},
};
use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
use hyper_util::{client::legacy::Client, client::legacy::connect::HttpConnector, rt::TokioExecutor};
use rand::Rng;
use std::{collections::HashMap, sync::{Arc, Mutex}};
use serde::{Deserialize, Serialize};
use std::io::Cursor;
use std::pin::Pin;
use std::sync::atomic::{AtomicI32, Ordering};
use std::task::{Context, Poll};
use std::{
collections::HashMap,
sync::{Arc, Mutex},
};
use time::Duration;
use time::OffsetDateTime;
use hyper_rustls::{ConfigBuilderExt, HttpsConnector};
use hyper_util::{client::legacy::Client, rt::TokioExecutor, client::legacy::connect::HttpConnector};
use http::{StatusCode, HeaderValue, request::{Request, Builder}, Response};
use tracing::{error, debug};
use url::{form_urlencoded, Url};
use tokio::io::BufReader;
use std::io::Cursor;
use tracing::{debug, error};
use url::{Url, form_urlencoded};
use uuid::Uuid;
use s3s::S3ErrorCode;
use s3s::{dto::Owner, Body};
use s3s::dto::ReplicationStatus;
use crate::client::bucket_cache::BucketLocationCache;
use reader::hasher::{Sha256, MD5,};
use crate::client::{
api_error_response::{err_invalid_argument, http_resp_to_error_response, to_error_response},
api_get_options::GetObjectOptions,
api_put_object::PutObjectOptions,
api_put_object_multipart::UploadPartParams,
api_s3_datatypes::{
CompleteMultipartUpload, CompletePart, ListBucketResult, ListBucketV2Result, ListMultipartUploadsResult,
ListObjectPartsResult, ObjectPart,
},
constants::{UNSIGNED_PAYLOAD, UNSIGNED_PAYLOAD_TRAILER},
credentials::{CredContext, Credentials, SignatureType, Static},
};
use crate::signer;
use crate::{checksum::ChecksumMode, store_api::GetObjectReader};
use reader::hasher::{MD5, Sha256};
use rustfs_rio::HashReader;
use rustfs_utils::{
net::get_endpoint_url,
retry::{new_retry_timer, MAX_RETRY},
};
use crate::{
store_api::GetObjectReader,
checksum::ChecksumMode,
};
use crate::signer;
use crate::client::{
constants::{UNSIGNED_PAYLOAD, UNSIGNED_PAYLOAD_TRAILER},
credentials::{Credentials, SignatureType, CredContext, Static,},
api_error_response::{to_error_response, http_resp_to_error_response, err_invalid_argument},
api_put_object_multipart::UploadPartParams,
api_put_object::PutObjectOptions,
api_get_options::GetObjectOptions,
api_s3_datatypes::{CompleteMultipartUpload, CompletePart, ListBucketResult, ListBucketV2Result, ListMultipartUploadsResult, ListObjectPartsResult, ObjectPart},
retry::{MAX_RETRY, new_retry_timer},
};
use s3s::S3ErrorCode;
use s3s::dto::ReplicationStatus;
use s3s::{Body, dto::Owner};
const C_USER_AGENT_PREFIX: &str = "RustFS (linux; x86)";
const C_USER_AGENT: &str = "RustFS (linux; x86)";
const C_USER_AGENT: &str = "RustFS (linux; x86)";
const SUCCESS_STATUS: [StatusCode; 3] = [
StatusCode::OK,
StatusCode::NO_CONTENT,
StatusCode::PARTIAL_CONTENT,
];
const SUCCESS_STATUS: [StatusCode; 3] = [StatusCode::OK, StatusCode::NO_CONTENT, StatusCode::PARTIAL_CONTENT];
const C_UNKNOWN: i32 = -1;
const C_OFFLINE: i32 = 0;
const C_ONLINE: i32 = 1;
const C_ONLINE: i32 = 1;
//pub type ReaderImpl = Box<dyn Reader + Send + Sync + 'static>;
pub enum ReaderImpl {
@@ -71,7 +73,7 @@ pub struct TransitionClient {
pub override_signer_type: SignatureType,
/*app_info: TODO*/
pub secure: bool,
pub http_client: Client<HttpsConnector<HttpConnector>, Body>,
pub http_client: Client<HttpsConnector<HttpConnector>, Body>,
//pub http_trace: Httptrace.ClientTrace,
pub bucket_loc_cache: Arc<Mutex<BucketLocationCache>>,
pub is_trace_enabled: Arc<Mutex<bool>>,
@@ -83,7 +85,7 @@ pub struct TransitionClient {
pub random: u64,
pub lookup: BucketLookupType,
//pub lookupFn: func(u url.URL, bucketName string) BucketLookupType,
pub md5_hasher: Arc<Mutex<Option<MD5>>>,
pub md5_hasher: Arc<Mutex<Option<MD5>>>,
pub sha256_hasher: Option<Sha256>,
pub health_status: AtomicI32,
pub trailing_header_support: bool,
@@ -92,16 +94,16 @@ pub struct TransitionClient {
#[derive(Debug, Default)]
pub struct Options {
pub creds: Credentials<Static>,
pub secure: bool,
pub creds: Credentials<Static>,
pub secure: bool,
//pub transport: http.RoundTripper,
//pub trace: *httptrace.ClientTrace,
pub region: String,
pub region: String,
pub bucket_lookup: BucketLookupType,
//pub custom_region_via_url: func(u url.URL) string,
//pub bucket_lookup_via_url: func(u url.URL, bucketName string) BucketLookupType,
pub trailing_headers: bool,
pub custom_md5: Option<MD5>,
pub custom_md5: Option<MD5>,
pub custom_sha256: Option<Sha256>,
pub max_retries: i64,
}
@@ -136,15 +138,13 @@ impl TransitionClient {
//if scheme == "https" {
// client = Client::builder(TokioExecutor::new()).build_http();
//} else {
let tls = rustls::ClientConfig::builder()
.with_native_roots()?
.with_no_client_auth();
let https = hyper_rustls::HttpsConnectorBuilder::new()
.with_tls_config(tls)
.https_or_http()
.enable_http1()
.build();
client = Client::builder(TokioExecutor::new()).build(https);
let tls = rustls::ClientConfig::builder().with_native_roots()?.with_no_client_auth();
let https = hyper_rustls::HttpsConnectorBuilder::new()
.with_tls_config(tls)
.https_or_http()
.enable_http1()
.build();
client = Client::builder(TokioExecutor::new()).build(https);
//}
let mut clnt = TransitionClient {
@@ -167,7 +167,7 @@ impl TransitionClient {
trailing_header_support: opts.trailing_headers,
max_retries: opts.max_retries,
};
{
let mut md5_hasher = clnt.md5_hasher.lock().unwrap();
if md5_hasher.is_none() {
@@ -218,7 +218,11 @@ impl TransitionClient {
todo!();
}
pub fn hash_materials(&self, is_md5_requested: bool, is_sha256_requested: bool) -> (HashMap<String, MD5>, HashMap<String, Vec<u8>>) {
pub fn hash_materials(
&self,
is_md5_requested: bool,
is_sha256_requested: bool,
) -> (HashMap<String, MD5>, HashMap<String, Vec<u8>>) {
todo!();
}
@@ -227,7 +231,8 @@ impl TransitionClient {
}
fn mark_offline(&self) {
self.health_status.compare_exchange(C_ONLINE, C_OFFLINE, Ordering::SeqCst, Ordering::SeqCst);
self.health_status
.compare_exchange(C_ONLINE, C_OFFLINE, Ordering::SeqCst, Ordering::SeqCst);
}
fn is_offline(&self) -> bool {
@@ -261,7 +266,9 @@ impl TransitionClient {
debug!("endpoint_url: {}", self.endpoint_url.as_str().to_string());
resp = http_client.request(req);
}
let resp = resp.await/*.map_err(Into::into)*/.map(|res| res.map(Body::from));
let resp = resp
.await /*.map_err(Into::into)*/
.map(|res| res.map(Body::from));
debug!("http_client url: {} {}", req_method, req_uri);
debug!("http_client headers: {:?}", req_headers);
if let Err(err) = resp {
@@ -282,7 +289,11 @@ impl TransitionClient {
Ok(resp)
}
pub async fn execute_method(&self, method: http::Method, metadata: &mut RequestMetadata) -> Result<http::Response<Body>, std::io::Error> {
pub async fn execute_method(
&self,
method: http::Method,
metadata: &mut RequestMetadata,
) -> Result<http::Response<Body>, std::io::Error> {
if self.is_offline() {
let mut s = self.endpoint_url.to_string();
s.push_str(" is offline.");
@@ -295,16 +306,18 @@ impl TransitionClient {
let mut resp: http::Response<Body>;
//if metadata.content_body != nil {
//body_seeker = BufferReader::new(metadata.content_body.read_all().await?);
retryable = true;
if !retryable {
req_retry = 1;
}
//body_seeker = BufferReader::new(metadata.content_body.read_all().await?);
retryable = true;
if !retryable {
req_retry = 1;
}
//}
//let mut retry_timer = RetryTimer::new();
//while let Some(v) = retry_timer.next().await {
for _ in [1;1]/*new_retry_timer(req_retry, DefaultRetryUnit, DefaultRetryCap, MaxJitter)*/ {
for _ in [1; 1]
/*new_retry_timer(req_retry, DefaultRetryUnit, DefaultRetryCap, MaxJitter)*/
{
let req = self.new_request(method, metadata).await?;
resp = self.doit(req).await?;
@@ -355,7 +368,11 @@ impl TransitionClient {
Err(std::io::Error::other("resp err"))
}
async fn new_request(&self, method: http::Method, metadata: &mut RequestMetadata) -> Result<http::Request<Body>, std::io::Error> {
async fn new_request(
&self,
method: http::Method,
metadata: &mut RequestMetadata,
) -> Result<http::Request<Body>, std::io::Error> {
let location = metadata.bucket_location.clone();
if location == "" {
if metadata.bucket_name != "" {
@@ -366,8 +383,13 @@ impl TransitionClient {
let is_makebucket = metadata.object_name == "" && method == http::Method::PUT && metadata.query_values.len() == 0;
let is_virtual_host = self.is_virtual_host_style_request(&self.endpoint_url, &metadata.bucket_name) && !is_makebucket;
let target_url = self.make_target_url(&metadata.bucket_name, &metadata.object_name, &location,
is_virtual_host, &metadata.query_values)?;
let target_url = self.make_target_url(
&metadata.bucket_name,
&metadata.object_name,
&location,
is_virtual_host,
&metadata.query_values,
)?;
let mut req_builder = Request::builder().method(method).uri(target_url.to_string());
@@ -377,10 +399,10 @@ impl TransitionClient {
value = creds_provider.get_with_context(Some(self.cred_context()))?;
}
let mut signer_type = value.signer_type.clone();
let access_key_id = value.access_key_id;
let mut signer_type = value.signer_type.clone();
let access_key_id = value.access_key_id;
let secret_access_key = value.secret_access_key;
let session_token = value.session_token;
let session_token = value.session_token;
if self.override_signer_type != SignatureType::SignatureDefault {
signer_type = self.override_signer_type.clone();
@@ -392,24 +414,39 @@ impl TransitionClient {
if metadata.expires != 0 && metadata.pre_sign_url {
if signer_type == SignatureType::SignatureAnonymous {
return Err(std::io::Error::other(err_invalid_argument("Presigned URLs cannot be generated with anonymous credentials.")));
return Err(std::io::Error::other(err_invalid_argument(
"Presigned URLs cannot be generated with anonymous credentials.",
)));
}
if metadata.extra_pre_sign_header.is_some() {
if signer_type == SignatureType::SignatureV2 {
return Err(std::io::Error::other(err_invalid_argument("Extra signed headers for Presign with Signature V2 is not supported.")));
return Err(std::io::Error::other(err_invalid_argument(
"Extra signed headers for Presign with Signature V2 is not supported.",
)));
}
for (k, v) in metadata.extra_pre_sign_header.as_ref().unwrap() {
req_builder = req_builder.header(k, v);
}
}
if signer_type == SignatureType::SignatureV2 {
req_builder = signer::pre_sign_v2(req_builder, &access_key_id, &secret_access_key, metadata.expires, is_virtual_host);
req_builder =
signer::pre_sign_v2(req_builder, &access_key_id, &secret_access_key, metadata.expires, is_virtual_host);
} else if signer_type == SignatureType::SignatureV4 {
req_builder = signer::pre_sign_v4(req_builder, &access_key_id, &secret_access_key, &session_token, &location, metadata.expires, OffsetDateTime::now_utc());
req_builder = signer::pre_sign_v4(
req_builder,
&access_key_id,
&secret_access_key,
&session_token,
&location,
metadata.expires,
OffsetDateTime::now_utc(),
);
}
let req = match req_builder.body(Body::empty()) {
Ok(req) => req,
Err(err) => { return Err(std::io::Error::other(err)); }
Err(err) => {
return Err(std::io::Error::other(err));
}
};
return Ok(req);
}
@@ -423,7 +460,10 @@ impl TransitionClient {
//req.content_length = metadata.content_length;
if metadata.content_length <= -1 {
let chunked_value = HeaderValue::from_str(&vec!["chunked"].join(",")).expect("err");
req_builder.headers_mut().expect("err").insert(http::header::TRANSFER_ENCODING, chunked_value);
req_builder
.headers_mut()
.expect("err")
.insert(http::header::TRANSFER_ENCODING, chunked_value);
}
if metadata.content_md5_base64.len() > 0 {
@@ -434,15 +474,17 @@ impl TransitionClient {
if signer_type == SignatureType::SignatureAnonymous {
let req = match req_builder.body(Body::empty()) {
Ok(req) => req,
Err(err) => { return Err(std::io::Error::other(err)); }
Err(err) => {
return Err(std::io::Error::other(err));
}
};
return Ok(req);
}
if signer_type == SignatureType::SignatureV2 {
req_builder = signer::sign_v2(req_builder, metadata.content_length, &access_key_id, &secret_access_key, is_virtual_host);
}
else if metadata.stream_sha256 && !self.secure {
req_builder =
signer::sign_v2(req_builder, metadata.content_length, &access_key_id, &secret_access_key, is_virtual_host);
} else if metadata.stream_sha256 && !self.secure {
if metadata.trailer.len() > 0 {
//req.Trailer = metadata.trailer;
for (_, v) in &metadata.trailer {
@@ -451,8 +493,7 @@ impl TransitionClient {
}
//req_builder = signer::streaming_sign_v4(req_builder, &access_key_id,
// &secret_access_key, &session_token, &location, metadata.content_length, OffsetDateTime::now_utc(), self.sha256_hasher());
}
else {
} else {
let mut sha_header = UNSIGNED_PAYLOAD.to_string();
if metadata.content_sha256_hex != "" {
sha_header = metadata.content_sha256_hex.clone();
@@ -462,9 +503,17 @@ impl TransitionClient {
} else if metadata.trailer.len() > 0 {
sha_header = UNSIGNED_PAYLOAD_TRAILER.to_string();
}
req_builder = req_builder.header::<HeaderName, HeaderValue>("X-Amz-Content-Sha256".parse().unwrap(), sha_header.parse().expect("err"));
req_builder = req_builder
.header::<HeaderName, HeaderValue>("X-Amz-Content-Sha256".parse().unwrap(), sha_header.parse().expect("err"));
req_builder = signer::sign_v4_trailer(req_builder, &access_key_id, &secret_access_key, &session_token, &location, metadata.trailer.clone());
req_builder = signer::sign_v4_trailer(
req_builder,
&access_key_id,
&secret_access_key,
&session_token,
&location,
metadata.trailer.clone(),
);
}
let req;
@@ -482,11 +531,9 @@ impl TransitionClient {
//req = req_builder.body(s3s::Body::from(metadata.content_body.read_all().await?));
}
match req {
match req {
Ok(req) => Ok(req),
Err(err) => {
Err(std::io::Error::other(err))
}
Err(err) => Err(std::io::Error::other(err)),
}
}
@@ -498,14 +545,17 @@ impl TransitionClient {
}*/
}
fn make_target_url(&self, bucket_name: &str, object_name: &str, bucket_location: &str, is_virtual_host_style: bool, query_values: &HashMap<String, String>) -> Result<Url, std::io::Error> {
fn make_target_url(
&self,
bucket_name: &str,
object_name: &str,
bucket_location: &str,
is_virtual_host_style: bool,
query_values: &HashMap<String, String>,
) -> Result<Url, std::io::Error> {
let scheme = self.endpoint_url.scheme();
let host = self.endpoint_url.host().unwrap();
let default_port = if scheme == "https" {
443
} else {
80
};
let default_port = if scheme == "https" { 443 } else { 80 };
let port = self.endpoint_url.port().unwrap_or(default_port);
let mut url_str = format!("{scheme}://{host}:{port}/");
@@ -562,7 +612,7 @@ impl TransitionClient {
}
struct LockedRandSource {
src: u64,//rand.Source,
src: u64, //rand.Source,
}
impl LockedRandSource {
@@ -604,76 +654,159 @@ impl TransitionCore {
Ok(Self(Arc::new(client)))
}
pub fn list_objects(&self, bucket: &str, prefix: &str, marker: &str, delimiter: &str, max_keys: i64) -> Result<ListBucketResult, std::io::Error> {
pub fn list_objects(
&self,
bucket: &str,
prefix: &str,
marker: &str,
delimiter: &str,
max_keys: i64,
) -> Result<ListBucketResult, std::io::Error> {
let client = self.0.clone();
client.list_objects_query(bucket, prefix, marker, delimiter, max_keys, HeaderMap::new())
}
pub async fn list_objects_v2(&self, bucket_name: &str, object_prefix: &str, start_after: &str, continuation_token: &str, delimiter: &str, max_keys: i64) -> Result<ListBucketV2Result, std::io::Error> {
pub async fn list_objects_v2(
&self,
bucket_name: &str,
object_prefix: &str,
start_after: &str,
continuation_token: &str,
delimiter: &str,
max_keys: i64,
) -> Result<ListBucketV2Result, std::io::Error> {
let client = self.0.clone();
client.list_objects_v2_query(bucket_name, object_prefix, continuation_token, true, false, delimiter, start_after, max_keys, HeaderMap::new()).await
client
.list_objects_v2_query(
bucket_name,
object_prefix,
continuation_token,
true,
false,
delimiter,
start_after,
max_keys,
HeaderMap::new(),
)
.await
}
/*pub fn copy_object(&self, source_bucket: &str, source_object: &str, dest_bucket: &str, dest_object: &str, metadata: HashMap<String, String>, src_opts: CopySrcOptions, dst_opts: PutObjectOptions) -> Result<ObjectInfo> {
self.0.copy_object_do(source_bucket, source_object, dest_bucket, dest_object, metadata, src_opts, dst_opts)
}*/
pub fn copy_object_part(&self, src_bucket: &str, src_object: &str, dest_bucket: &str, dest_object: &str, upload_id: &str,
part_id: i32, start_offset: i32, length: i64, metadata: HashMap<String, String>,
pub fn copy_object_part(
&self,
src_bucket: &str,
src_object: &str,
dest_bucket: &str,
dest_object: &str,
upload_id: &str,
part_id: i32,
start_offset: i32,
length: i64,
metadata: HashMap<String, String>,
) -> Result<CompletePart, std::io::Error> {
//self.0.copy_object_part_do(src_bucket, src_object, dest_bucket, dest_object, upload_id,
// part_id, start_offset, length, metadata)
todo!();
}
pub async fn put_object(&self, bucket: &str, object: &str, data: ReaderImpl, size: i64, md5_base64: &str, sha256_hex: &str, opts: &PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
let hook_reader = data;//newHook(data, opts.progress);
pub async fn put_object(
&self,
bucket: &str,
object: &str,
data: ReaderImpl,
size: i64,
md5_base64: &str,
sha256_hex: &str,
opts: &PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let hook_reader = data; //newHook(data, opts.progress);
let client = self.0.clone();
client.put_object_do(bucket, object, hook_reader, md5_base64, sha256_hex, size, opts).await
client
.put_object_do(bucket, object, hook_reader, md5_base64, sha256_hex, size, opts)
.await
}
pub async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: PutObjectOptions) -> Result<String, std::io::Error> {
pub async fn new_multipart_upload(
&self,
bucket: &str,
object: &str,
opts: PutObjectOptions,
) -> Result<String, std::io::Error> {
let client = self.0.clone();
let result = client.initiate_multipart_upload(bucket, object, &opts).await?;
Ok(result.upload_id)
}
pub fn list_multipart_uploads(&self, bucket: &str, prefix: &str, key_marker: &str, upload_id_marker: &str, delimiter: &str, max_uploads: i64) -> Result<ListMultipartUploadsResult, std::io::Error> {
pub fn list_multipart_uploads(
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
max_uploads: i64,
) -> Result<ListMultipartUploadsResult, std::io::Error> {
let client = self.0.clone();
client.list_multipart_uploads_query(bucket, key_marker, upload_id_marker, prefix, delimiter, max_uploads)
}
pub async fn put_object_part(&self, bucket: &str, object: &str, upload_id: &str, part_id: i64,
data: ReaderImpl, size: i64, opts: PutObjectPartOptions
pub async fn put_object_part(
&self,
bucket: &str,
object: &str,
upload_id: &str,
part_id: i64,
data: ReaderImpl,
size: i64,
opts: PutObjectPartOptions,
) -> Result<ObjectPart, std::io::Error> {
let mut p = UploadPartParams {
bucket_name: bucket.to_string(),
object_name: object.to_string(),
upload_id: upload_id.to_string(),
reader: data,
part_number: part_id,
md5_base64: opts.md5_base64,
sha256_hex: opts.sha256_hex,
size: size,
bucket_name: bucket.to_string(),
object_name: object.to_string(),
upload_id: upload_id.to_string(),
reader: data,
part_number: part_id,
md5_base64: opts.md5_base64,
sha256_hex: opts.sha256_hex,
size: size,
//sse: opts.sse,
stream_sha256: !opts.disable_content_sha256,
custom_header: opts.custom_header,
trailer: opts.trailer,
trailer: opts.trailer,
};
let client = self.0.clone();
client.upload_part(&mut p).await
}
pub async fn list_object_parts(&self, bucket: &str, object: &str, upload_id: &str, part_number_marker: i64, max_parts: i64) -> Result<ListObjectPartsResult, std::io::Error> {
pub async fn list_object_parts(
&self,
bucket: &str,
object: &str,
upload_id: &str,
part_number_marker: i64,
max_parts: i64,
) -> Result<ListObjectPartsResult, std::io::Error> {
let client = self.0.clone();
client.list_object_parts_query(bucket, object, upload_id, part_number_marker, max_parts).await
client
.list_object_parts_query(bucket, object, upload_id, part_number_marker, max_parts)
.await
}
pub async fn complete_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, parts: &[CompletePart], opts: PutObjectOptions) -> Result<UploadInfo, std::io::Error> {
pub async fn complete_multipart_upload(
&self,
bucket: &str,
object: &str,
upload_id: &str,
parts: &[CompletePart],
opts: PutObjectOptions,
) -> Result<UploadInfo, std::io::Error> {
let client = self.0.clone();
let res = client.complete_multipart_upload(bucket, object, upload_id, CompleteMultipartUpload {
parts: parts.to_vec(),
}, &opts).await?;
let res = client
.complete_multipart_upload(bucket, object, upload_id, CompleteMultipartUpload { parts: parts.to_vec() }, &opts)
.await?;
Ok(res)
}
@@ -692,7 +825,12 @@ impl TransitionCore {
client.put_bucket_policy(bucket_name, bucket_policy).await
}
pub async fn get_object(&self, bucket_name: &str, object_name: &str, opts: &GetObjectOptions) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
pub async fn get_object(
&self,
bucket_name: &str,
object_name: &str,
opts: &GetObjectOptions,
) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
let client = self.0.clone();
client.get_object_inner(bucket_name, object_name, opts).await
}
@@ -709,8 +847,8 @@ pub struct PutObjectPartOptions {
#[derive(Debug, Clone, Deserialize, Serialize)]
pub struct ObjectInfo {
pub etag: String,
pub name: String,
pub etag: String,
pub name: String,
pub mod_time: OffsetDateTime,
pub size: usize,
pub content_type: Option<String>,
@@ -730,16 +868,16 @@ pub struct ObjectInfo {
#[serde(skip, default = "replication_status_default")]
pub replication_status: ReplicationStatus,
pub replication_ready: bool,
pub expiration: OffsetDateTime,
pub expiration: OffsetDateTime,
pub expiration_rule_id: String,
pub num_versions: usize,
pub restore: RestoreInfo,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc64nvme: String,
pub checksum_mode: String,
}
@@ -767,22 +905,21 @@ impl Default for ObjectInfo {
version_id: Uuid::nil(),
replication_status: ReplicationStatus::from_static(ReplicationStatus::PENDING),
replication_ready: false,
expiration: OffsetDateTime::now_utc(),
expiration: OffsetDateTime::now_utc(),
expiration_rule_id: "".to_string(),
num_versions: 0,
restore: RestoreInfo::default(),
checksum_crc32: "".to_string(),
checksum_crc32c: "".to_string(),
checksum_sha1: "".to_string(),
checksum_sha256: "".to_string(),
checksum_crc32: "".to_string(),
checksum_crc32c: "".to_string(),
checksum_sha1: "".to_string(),
checksum_sha256: "".to_string(),
checksum_crc64nvme: "".to_string(),
checksum_mode: "".to_string(),
}
}
}
#[derive(Serialize, Deserialize)]
#[derive(Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct RestoreInfo {
ongoing_restore: bool,
expiry_time: OffsetDateTime,
@@ -809,19 +946,19 @@ pub struct ObjectMultipartInfo {
}
pub struct UploadInfo {
pub bucket: String,
pub key: String,
pub etag: String,
pub size: i64,
pub bucket: String,
pub key: String,
pub etag: String,
pub size: i64,
pub last_modified: OffsetDateTime,
pub location: String,
pub version_id: String,
pub expiration: OffsetDateTime,
pub location: String,
pub version_id: String,
pub expiration: OffsetDateTime,
pub expiration_rule_id: String,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc32: String,
pub checksum_crc32c: String,
pub checksum_sha1: String,
pub checksum_sha256: String,
pub checksum_crc64nvme: String,
pub checksum_mode: String,
}
@@ -885,4 +1022,4 @@ impl tower::Service<Request<Body>> for SendRequest {
}
#[derive(Serialize, Deserialize)]
pub struct Document(pub String);
pub struct Document(pub String);