Merge branch 'main' of github.com:rustfs/s3-rustfs into feature/observability-metrics

# Conflicts:
#	Cargo.lock
#	Cargo.toml

rustfs/Cargo.toml

@@ -22,6 +22,7 @@ api = { workspace = true }
 appauth = { workspace = true }
 atoi = { workspace = true }
 atomic_enum = { workspace = true }
+aws-sdk-s3 = { workspace = true }
 axum.workspace = true
 axum-extra = { workspace = true }
 axum-server = { workspace = true }
@@ -43,14 +44,18 @@ hyper-util.workspace = true
 http.workspace = true
 http-body.workspace = true
 iam = { workspace = true }
+include_dir = { workspace = true }
+jsonwebtoken = { workspace = true }
 lock.workspace = true
 matchit = { workspace = true }
 mime.workspace = true
 mime_guess = { workspace = true }
 opentelemetry = { workspace = true }
+percent-encoding = { workspace = true }
 pin-project-lite.workspace = true
 protos.workspace = true
 query = { workspace = true }
+regex = { workspace = true }
 rmp-serde.workspace = true
 rustfs-config = { workspace = true }
 rustfs-notify = { workspace = true }
@@ -64,6 +69,7 @@ serde_json.workspace = true
 serde_urlencoded = { workspace = true }
 shadow-rs = { workspace = true, features = ["build", "metadata"] }
 socket2 = { workspace = true }
+thiserror = { workspace = true }
 tracing.workspace = true
 time = { workspace = true, features = ["parsing", "formatting", "serde"] }
 tokio-util.workspace = true
@@ -85,6 +91,7 @@ tower-http = { workspace = true, features = [
     "compression-gzip",
     "cors",
 ] }
+urlencoding = { workspace = true }
 uuid = { workspace = true }
 
 [target.'cfg(target_os = "linux")'.dependencies]

rustfs/src/admin/handlers.rs

@@ -2,10 +2,15 @@ use super::router::Operation;
 use crate::auth::check_key_valid;
 use crate::auth::get_condition_values;
 use crate::auth::get_session_token;
+//use ecstore::error::Error as ec_Error;
+use crate::storage::error::to_s3_error;
 use bytes::Bytes;
 use common::error::Error as ec_Error;
 use ecstore::admin_server_info::get_server_info;
+use ecstore::bucket::metadata_sys::{self, get_replication_config};
+use ecstore::bucket::target::BucketTarget;
 use ecstore::bucket::versioning_sys::BucketVersioningSys;
+use ecstore::cmd::bucket_targets::{self, GLOBAL_Bucket_Target_Sys};
 use ecstore::global::GLOBAL_ALlHealState;
 use ecstore::heal::data_usage::load_data_usage_from_backend;
 use ecstore::heal::heal_commands::HealOpts;
@@ -23,9 +28,11 @@ use http::{HeaderMap, Uri};
 use hyper::StatusCode;
 use iam::get_global_action_cred;
 use iam::store::MappedPolicy;
+// use lazy_static::lazy_static;
 use madmin::metrics::RealtimeMetrics;
 use madmin::utils::parse_duration;
 use matchit::Params;
+use percent_encoding::{percent_encode, AsciiSet, CONTROLS};
 use policy::policy::action::Action;
 use policy::policy::action::S3Action;
 use policy::policy::default::DEFAULT_POLICIES;
@@ -36,6 +43,7 @@ use s3s::stream::{ByteStream, DynByteStream};
 use s3s::{s3_error, Body, S3Error, S3Request, S3Response, S3Result};
 use s3s::{S3ErrorCode, StdError};
 use serde::{Deserialize, Serialize};
+// use serde_json::to_vec;
 use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::pin::Pin;
@@ -47,6 +55,7 @@ use tokio::time::interval;
 use tokio::{select, spawn};
 use tokio_stream::wrappers::ReceiverStream;
 use tracing::{error, info, warn};
+// use url::UrlQuery;
 
 pub mod event;
 pub mod group;
@@ -57,6 +66,7 @@ pub mod service_account;
 pub mod sts;
 pub mod trace;
 pub mod user;
+use urlencoding::decode;
 
 #[derive(Debug, Serialize, Default)]
 #[serde(rename_all = "PascalCase", default)]
@@ -745,6 +755,278 @@ impl Operation for BackgroundHealStatusHandler {
     }
 }
 
+fn extract_query_params(uri: &Uri) -> HashMap<String, String> {
+    let mut params = HashMap::new();
+
+    if let Some(query) = uri.query() {
+        query.split('&').for_each(|pair| {
+            if let Some((key, value)) = pair.split_once('=') {
+                params.insert(key.to_string(), value.to_string());
+            }
+        });
+    }
+
+    params
+}
+
+//disable encrypto from client because rustfs use len 8 Nonce but rustfs use 12 len Nonce
+
+#[allow(dead_code)]
+fn is_local_host(_host: String) -> bool {
+    false
+}
+
+//awscurl --service s3 --region us-east-1 --access_key rustfsadmin --secret_key rustfsadmin "http://:9000/rustfs/admin/v3/replicationmetrics?bucket=1"
+pub struct GetReplicationMetricsHandler {}
+#[async_trait::async_trait]
+impl Operation for GetReplicationMetricsHandler {
+    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        error!("GetReplicationMetricsHandler");
+        let querys = extract_query_params(&_req.uri);
+        if let Some(bucket) = querys.get("bucket") {
+            error!("get bucket:{} metris", bucket);
+        }
+        //return Err(s3_error!(InvalidArgument, "Invalid bucket name"));
+        //Ok(S3Response::with_headers((StatusCode::OK, Body::from()), header))
+        return Ok(S3Response::new((StatusCode::OK, Body::from("Ok".to_string()))));
+    }
+}
+
+pub struct SetRemoteTargetHandler {}
+#[async_trait::async_trait]
+impl Operation for SetRemoteTargetHandler {
+    async fn call(&self, mut _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        //return Ok(S3Response::new((StatusCode::OK, Body::from("OK".to_string()))));
+        // println!("handle MetricsHandler, params: {:?}", _req.input);
+        info!("handle MetricsHandler, params: {:?}", _req.credentials);
+        let querys = extract_query_params(&_req.uri);
+        let Some(_cred) = _req.credentials else {
+            error!("credentials null");
+            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        };
+        let _is_owner = true; // 先按 true 处理，后期根据请求决定
+        let body = _req.input.store_all_unlimited().await.unwrap();
+        //println!("body: {}", std::str::from_utf8(&body.clone()).unwrap());
+
+        //println!("bucket is:{}", bucket.clone());
+        if let Some(bucket) = querys.get("bucket") {
+            if bucket.is_empty() {
+                println!("have bucket: {}", bucket);
+                return Ok(S3Response::new((StatusCode::OK, Body::from("fuck".to_string()))));
+            }
+            let Some(store) = new_object_layer_fn() else {
+                return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+            };
+
+            // let binfo:BucketInfo = store
+            // .get_bucket_info(bucket, &ecstore::store_api::BucketOptions::default()).await;
+            match store
+                .get_bucket_info(bucket, &ecstore::store_api::BucketOptions::default())
+                .await
+            {
+                Ok(info) => {
+                    println!("Bucket Info: {:?}", info);
+                    if !info.versionning {
+                        return Ok(S3Response::new((StatusCode::FORBIDDEN, Body::from("bucket need versioned".to_string()))));
+                    }
+                }
+                Err(err) => {
+                    eprintln!("Error: {:?}", err);
+                    return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from("empty bucket".to_string()))));
+                }
+            }
+
+            let mut remote_target: BucketTarget = serde_json::from_slice(&body).map_err(|arg0| to_s3_error(arg0.into()))?; // 错误会被传播
+            remote_target.source_bucket = bucket.clone();
+
+            info!("remote target {} And arn is:", remote_target.source_bucket.clone());
+
+            if let Some(val) = remote_target.arn.clone() {
+                info!("arn is {}", val);
+            }
+
+            if let Some(sys) = GLOBAL_Bucket_Target_Sys.get() {
+                let (arn, exist) = sys.get_remote_arn(bucket, Some(&remote_target), "").await;
+                info!("exist: {} {}", exist, arn.clone().unwrap_or_default());
+                if exist && arn.is_some() {
+                    let jsonarn = serde_json::to_string(&arn).expect("failed to serialize");
+                    //Ok(S3Response::new)
+                    return Ok(S3Response::new((StatusCode::OK, Body::from(jsonarn))));
+                } else {
+                    remote_target.arn = arn;
+                    match sys.set_target(bucket, &remote_target, false, false).await {
+                        Ok(_) => {
+                            {
+                                //todo 各种持久化的工作
+                                let targets = sys.list_targets(Some(bucket), None).await;
+                                info!("targets is {}", targets.len());
+                                match serde_json::to_vec(&targets) {
+                                    Ok(json) => {
+                                        //println!("json is:{:?}", json.clone().to_ascii_lowercase());
+                                        //metadata_sys::GLOBAL_BucketMetadataSys::
+                                        //BUCKET_TARGETS_FILE: &str = "bucket-targets.json"
+                                        let _ = metadata_sys::update(bucket, "bucket-targets.json", json).await;
+                                        // if let Err(err) = metadata_sys::GLOBAL_BucketMetadataSys.get().
+                                        //     .update(ctx, bucket, "bucketTargetsFile", tgt_bytes)
+                                        //     .await
+                                        // {
+                                        //     write_error_response(ctx, &err)?;
+                                        //     return Err(err);
+                                        // }
+                                    }
+                                    Err(e) => {
+                                        error!("序列化失败{}", e);
+                                    }
+                                }
+                            }
+
+                            let jsonarn = serde_json::to_string(&remote_target.arn.clone()).expect("failed to serialize");
+                            return Ok(S3Response::new((StatusCode::OK, Body::from(jsonarn))));
+                        }
+                        Err(e) => {
+                            error!("set target error {}", e);
+                            return Ok(S3Response::new((
+                                StatusCode::BAD_REQUEST,
+                                Body::from("remote target not ready".to_string()),
+                            )));
+                        }
+                    }
+                }
+            } else {
+                error!("GLOBAL_BUCKET _TARGET_SYS is not initialized");
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InternalError,
+                    "GLOBAL_BUCKET_TARGET_SYS is not initialized".to_string(),
+                ));
+            }
+        }
+        // return Err(s3_error!(InvalidArgument));
+        return Ok(S3Response::new((StatusCode::OK, Body::from("Ok".to_string()))));
+    }
+}
+
+pub struct ListRemoteTargetHandler {}
+#[async_trait::async_trait]
+impl Operation for ListRemoteTargetHandler {
+    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        warn!("list GetRemoteTargetHandler, params: {:?}", _req.credentials);
+
+        let querys = extract_query_params(&_req.uri);
+        let Some(_cred) = _req.credentials else {
+            error!("credentials null");
+            return Err(s3_error!(InvalidRequest, "get cred failed"));
+        };
+
+        if let Some(bucket) = querys.get("bucket") {
+            if bucket.is_empty() {
+                error!("bucket parameter is empty");
+                return Ok(S3Response::new((
+                    StatusCode::BAD_REQUEST,
+                    Body::from("Bucket parameter is required".to_string()),
+                )));
+            }
+
+            let Some(store) = new_object_layer_fn() else {
+                return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not initialized".to_string()));
+            };
+
+            match store
+                .get_bucket_info(bucket, &ecstore::store_api::BucketOptions::default())
+                .await
+            {
+                Ok(info) => {
+                    println!("Bucket Info: {:?}", info);
+                    if !info.versionning {
+                        return Ok(S3Response::new((
+                            StatusCode::FORBIDDEN,
+                            Body::from("Bucket needs versioning".to_string()),
+                        )));
+                    }
+                }
+                Err(err) => {
+                    eprintln!("Error fetching bucket info: {:?}", err);
+                    return Ok(S3Response::new((StatusCode::BAD_REQUEST, Body::from("Invalid bucket".to_string()))));
+                }
+            }
+
+            if let Some(sys) = GLOBAL_Bucket_Target_Sys.get() {
+                let targets = sys.list_targets(Some(bucket), None).await;
+                error!("target sys len {}", targets.len());
+                if targets.is_empty() {
+                    return Ok(S3Response::new((
+                        StatusCode::NOT_FOUND,
+                        Body::from("No remote targets found".to_string()),
+                    )));
+                }
+
+                let json_targets = serde_json::to_string(&targets).map_err(|e| {
+                    error!("Serialization error: {}", e);
+                    S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string())
+                })?;
+
+                return Ok(S3Response::new((StatusCode::OK, Body::from(json_targets))));
+            } else {
+                println!("GLOBAL_BUCKET_TARGET_SYS is not initialized");
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InternalError,
+                    "GLOBAL_BUCKET_TARGET_SYS is not initialized".to_string(),
+                ));
+            }
+        }
+
+        println!("Bucket parameter missing in request");
+        Ok(S3Response::new((
+            StatusCode::BAD_REQUEST,
+            Body::from("Bucket parameter is required".to_string()),
+        )))
+        //return Err(s3_error!(NotImplemented));
+    }
+}
+const COLON: AsciiSet = CONTROLS.add(b':');
+pub struct RemoveRemoteTargetHandler {}
+#[async_trait::async_trait]
+impl Operation for RemoveRemoteTargetHandler {
+    async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        error!("remove remote target called");
+        let querys = extract_query_params(&_req.uri);
+
+        if let Some(arnstr) = querys.get("arn") {
+            if let Some(bucket) = querys.get("bucket") {
+                if bucket.is_empty() {
+                    error!("bucket parameter is empty");
+                    return Ok(S3Response::new((StatusCode::NOT_FOUND, Body::from("bucket not found".to_string()))));
+                }
+                let _arn = bucket_targets::ARN::parse(arnstr);
+
+                match get_replication_config(bucket).await {
+                    Ok((conf, _ts)) => {
+                        for ru in conf.rules {
+                            let encoded = percent_encode(ru.destination.bucket.as_bytes(), &COLON);
+                            let encoded_str = encoded.to_string();
+                            if *arnstr == encoded_str {
+                                error!("target in use");
+                                return Ok(S3Response::new((StatusCode::FORBIDDEN, Body::from("Ok".to_string()))));
+                            }
+                            info!("bucket: {} and arn str is {} ", encoded_str, arnstr);
+                        }
+                    }
+                    Err(err) => {
+                        error!("get replication config err: {}", err);
+                        return Ok(S3Response::new((StatusCode::NOT_FOUND, Body::from(err.to_string()))));
+                    }
+                }
+                //percent_decode_str(&arnstr);
+                let decoded_str = decode(arnstr).unwrap();
+                error!("need delete target is {}", decoded_str);
+                bucket_targets::remove_bucket_target(bucket, arnstr).await;
+            }
+        }
+        //return Err(s3_error!(InvalidArgument, "Invalid bucket name"));
+        //Ok(S3Response::with_headers((StatusCode::OK, Body::from()), header))
+        return Ok(S3Response::new((StatusCode::OK, Body::from("Ok".to_string()))));
+    }
+}
+
 #[cfg(test)]
 mod test {
     use ecstore::heal::heal_commands::HealOpts;

rustfs/src/admin/mod.rs

@@ -11,12 +11,14 @@ use handlers::{
     sts, user,
 };
 
+use handlers::{GetReplicationMetricsHandler, ListRemoteTargetHandler, RemoveRemoteTargetHandler, SetRemoteTargetHandler};
 use hyper::Method;
 use router::{AdminOperation, S3Router};
 use rpc::register_rpc_route;
 use s3s::route::S3Route;
 
 const ADMIN_PREFIX: &str = "/rustfs/admin";
+const RUSTFS_ADMIN_PREFIX: &str = "/rustfs/admin";
 
 pub fn make_admin_route() -> Result<impl S3Route> {
     let mut r: S3Router<AdminOperation> = S3Router::new();
@@ -228,6 +230,53 @@ fn register_user_route(r: &mut S3Router<AdminOperation>) -> Result<()> {
         AdminOperation(&AddServiceAccount {}),
     )?;
 
+    r.insert(
+        Method::GET,
+        format!("{}{}", RUSTFS_ADMIN_PREFIX, "/v3/list-remote-targets").as_str(),
+        AdminOperation(&ListRemoteTargetHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/list-remote-targets").as_str(),
+        AdminOperation(&ListRemoteTargetHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", RUSTFS_ADMIN_PREFIX, "/v3/replicationmetrics").as_str(),
+        AdminOperation(&GetReplicationMetricsHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/replicationmetrics").as_str(),
+        AdminOperation(&GetReplicationMetricsHandler {}),
+    )?;
+
+    r.insert(
+        Method::PUT,
+        format!("{}{}", RUSTFS_ADMIN_PREFIX, "/v3/set-remote-target").as_str(),
+        AdminOperation(&SetRemoteTargetHandler {}),
+    )?;
+    r.insert(
+        Method::PUT,
+        format!("{}{}", ADMIN_PREFIX, "/v3/set-remote-target").as_str(),
+        AdminOperation(&SetRemoteTargetHandler {}),
+    )?;
+
+    r.insert(
+        Method::DELETE,
+        format!("{}{}", RUSTFS_ADMIN_PREFIX, "/v3/remove-remote-target").as_str(),
+        AdminOperation(&RemoveRemoteTargetHandler {}),
+    )?;
+
+    r.insert(
+        Method::DELETE,
+        format!("{}{}", ADMIN_PREFIX, "/v3/remove-remote-target").as_str(),
+        AdminOperation(&RemoveRemoteTargetHandler {}),
+    )?;
+
     // list-canned-policies?bucket=xxx
     r.insert(
         Method::GET,

rustfs/src/admin/router.rs

@@ -16,6 +16,7 @@ use s3s::S3Result;
 
 use super::rpc::RPC_PREFIX;
 use super::ADMIN_PREFIX;
+use super::RUSTFS_ADMIN_PREFIX;
 
 pub struct S3Router<T> {
     router: Router<T>,
@@ -64,7 +65,7 @@ where
             }
         }
 
-        uri.path().starts_with(ADMIN_PREFIX) || uri.path().starts_with(RPC_PREFIX)
+        uri.path().starts_with(ADMIN_PREFIX) || uri.path().starts_with(RPC_PREFIX) || uri.path().starts_with(RUSTFS_ADMIN_PREFIX)
     }
 
     async fn call(&self, req: S3Request<Body>) -> S3Result<S3Response<Body>> {

rustfs/src/grpc.rs

@@ -240,6 +240,8 @@ impl Node for NodeService {
                         }));
                     }
                 };
+
+                println!("bucket info {}", bucket_info.clone());
                 Ok(tonic::Response::new(GetBucketInfoResponse {
                     success: true,
                     bucket_info,
@@ -247,6 +249,7 @@ impl Node for NodeService {
                 }))
             }
 
+            // println!("vuc")
             Err(err) => Ok(tonic::Response::new(GetBucketInfoResponse {
                 success: false,
                 bucket_info: String::new(),

rustfs/src/main.rs

@@ -22,6 +22,7 @@ use common::{
     globals::set_global_addr,
 };
 use ecstore::bucket::metadata_sys::init_bucket_metadata_sys;
+use ecstore::cmd::bucket_replication::init_bucket_replication_pool;
 use ecstore::config as ecconfig;
 use ecstore::config::GLOBAL_ConfigSys;
 use ecstore::heal::background_heal_ops::init_auto_heal;
@@ -541,6 +542,11 @@ async fn run(opt: config::Opt) -> Result<()> {
 
     init_console_cfg(local_ip, server_port);
 
+    print_server_info();
+    init_bucket_replication_pool().await;
+
+    init_console_cfg(local_ip, server_port);
+
     print_server_info();
 
     if opt.console_enable {

rustfs/src/storage/ecfs.rs

@@ -4,14 +4,12 @@ use super::options::extract_metadata;
 use super::options::put_opts;
 use crate::auth::get_condition_values;
 use crate::storage::access::ReqInfo;
-use crate::storage::error::to_s3_error;
-use crate::storage::options::copy_dst_opts;
-use crate::storage::options::copy_src_opts;
-use crate::storage::options::{extract_metadata_from_mime, get_opts};
 use api::query::Context;
 use api::query::Query;
 use api::server::dbms::DatabaseManagerSystem;
 use bytes::Bytes;
+use chrono::DateTime;
+use chrono::Utc;
 use common::error::Result;
 use datafusion::arrow::csv::WriterBuilder as CsvWriterBuilder;
 use datafusion::arrow::json::writer::JsonArray;
@@ -30,6 +28,9 @@ use ecstore::bucket::policy_sys::PolicySys;
 use ecstore::bucket::tagging::decode_tags;
 use ecstore::bucket::tagging::encode_tags;
 use ecstore::bucket::versioning_sys::BucketVersioningSys;
+use ecstore::cmd::bucket_replication::get_must_replicate_options;
+use ecstore::cmd::bucket_replication::must_replicate;
+use ecstore::cmd::bucket_replication::schedule_replication;
 use ecstore::io::READ_BUFFER_SIZE;
 use ecstore::new_object_layer_fn;
 use ecstore::store_api::BucketOptions;
@@ -43,6 +44,7 @@ use ecstore::store_api::ObjectOptions;
 use ecstore::store_api::ObjectToDelete;
 use ecstore::store_api::PutObjReader;
 use ecstore::store_api::StorageAPI;
+// use ecstore::store_api::RESERVED_METADATA_PREFIX;
 use ecstore::store_api::RESERVED_METADATA_PREFIX_LOWER;
 use ecstore::utils::path::path_join_buf;
 use ecstore::utils::xml;
@@ -85,6 +87,13 @@ use tracing::warn;
 use transform_stream::AsyncTryStream;
 use uuid::Uuid;
 
+use crate::storage::error::to_s3_error;
+use crate::storage::options::copy_dst_opts;
+use crate::storage::options::copy_src_opts;
+use crate::storage::options::{extract_metadata_from_mime, get_opts};
+use ecstore::cmd::bucket_replication::ReplicationStatusType;
+use ecstore::cmd::bucket_replication::ReplicationType;
+
 macro_rules! try_ {
     ($result:expr) => {
         match $result {
@@ -924,6 +933,7 @@ impl S3 for FS {
             return self.put_object_extract(req).await;
         }
 
+        info!("put object");
         let input = req.input;
 
         if let Some(ref storage_class) = input.storage_class {
@@ -976,10 +986,27 @@ impl S3 for FS {
             metadata.insert(xhttp::AMZ_OBJECT_TAGGING.to_owned(), tags);
         }
 
-        let opts: ObjectOptions = put_opts(&bucket, &key, version_id, &req.headers, Some(metadata))
+        let mt = metadata.clone();
+        let mt2 = metadata.clone();
+
+        let opts: ObjectOptions = put_opts(&bucket, &key, version_id, &req.headers, Some(mt))
             .await
             .map_err(to_s3_error)?;
 
+        let repoptions =
+            get_must_replicate_options(&mt2, "", ReplicationStatusType::Unknown, ReplicationType::ObjectReplicationType, &opts);
+
+        let dsc = must_replicate(&bucket, &key, &repoptions).await;
+        warn!("dsc {}", &dsc.replicate_any().clone());
+        if dsc.replicate_any() {
+            let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-timestamp");
+            let now: DateTime<Utc> = Utc::now();
+            let formatted_time = now.to_rfc3339();
+            metadata.insert(k, formatted_time);
+            let k = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "replication-status");
+            metadata.insert(k, dsc.pending_status());
+        }
+
         debug!("put_object opts {:?}", &opts);
 
         let obj_info = store
@@ -987,9 +1014,17 @@ impl S3 for FS {
             .await
             .map_err(to_s3_error)?;
 
-        let e_tag = obj_info.etag;
+        let e_tag = obj_info.etag.clone();
 
-        // store.put_object(bucket, object, data, opts);
+        let repoptions =
+            get_must_replicate_options(&mt2, "", ReplicationStatusType::Unknown, ReplicationType::ObjectReplicationType, &opts);
+
+        let dsc = must_replicate(&bucket, &key, &repoptions).await;
+
+        if dsc.replicate_any() {
+            let objectlayer = new_object_layer_fn();
+            schedule_replication(obj_info, objectlayer.unwrap(), dsc, 1).await;
+        }
 
         let output = PutObjectOutput {
             e_tag,
@@ -1152,17 +1187,30 @@ impl S3 for FS {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
-        let oi = store
+        let obj_info = store
             .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, opts)
             .await
             .map_err(to_s3_error)?;
 
         let output = CompleteMultipartUploadOutput {
-            bucket: Some(bucket),
-            key: Some(key),
-            e_tag: oi.etag,
+            bucket: Some(bucket.clone()),
+            key: Some(key.clone()),
+            e_tag: obj_info.etag.clone(),
+            location: Some("us-east-1".to_string()),
             ..Default::default()
         };
+
+        let mt2 = HashMap::new();
+        let repoptions =
+            get_must_replicate_options(&mt2, "", ReplicationStatusType::Unknown, ReplicationType::ObjectReplicationType, opts);
+
+        let dsc = must_replicate(&bucket, &key, &repoptions).await;
+
+        if dsc.replicate_any() {
+            warn!("need multipart replication");
+            let objectlayer = new_object_layer_fn();
+            schedule_replication(obj_info, objectlayer.unwrap(), dsc, 1).await;
+        }
         Ok(S3Response::new(output))
     }
 
@@ -1725,17 +1773,35 @@ impl S3 for FS {
             .await
             .map_err(to_s3_error)?;
 
-        let replication_configuration = match metadata_sys::get_replication_config(&bucket).await {
+        let rcfg = match metadata_sys::get_replication_config(&bucket).await {
             Ok((cfg, _created)) => Some(cfg),
             Err(err) => {
-                warn!("get_object_lock_config err {:?}", err);
-                None
+                error!("get_replication_config err {:?}", err);
+                return Err(to_s3_error(err));
             }
         };
 
-        Ok(S3Response::new(GetBucketReplicationOutput {
-            replication_configuration,
-        }))
+        if rcfg.is_none() {
+            return Err(S3Error::with_message(S3ErrorCode::NoSuchBucket, "replication not found".to_string()));
+        }
+
+        // Ok(S3Response::new(GetBucketReplicationOutput {
+        //     replication_configuration: rcfg,
+        // }))
+
+        if rcfg.is_some() {
+            Ok(S3Response::new(GetBucketReplicationOutput {
+                replication_configuration: rcfg,
+            }))
+        } else {
+            let rep = ReplicationConfiguration {
+                role: "".to_string(),
+                rules: vec![],
+            };
+            Ok(S3Response::new(GetBucketReplicationOutput {
+                replication_configuration: Some(rep),
+            }))
+        }
     }
 
     async fn put_bucket_replication(
@@ -1747,6 +1813,7 @@ impl S3 for FS {
             replication_configuration,
             ..
         } = req.input;
+        warn!("put bucket replication");
 
         let Some(store) = new_object_layer_fn() else {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
@@ -1786,6 +1853,7 @@ impl S3 for FS {
             .map_err(to_s3_error)?;
 
         // TODO: remove targets
+        error!("delete bucket");
 
         Ok(S3Response::new(DeleteBucketReplicationOutput::default()))
     }
@@ -2105,13 +2173,15 @@ impl S3 for FS {
             None
         };
 
-        if legal_hold.is_none() {
-            return Err(s3_error!(InvalidRequest, "Object does not have legal hold"));
-        }
+        let status = if let Some(v) = legal_hold {
+            v
+        } else {
+            ObjectLockLegalHoldStatus::OFF.to_string()
+        };
 
         Ok(S3Response::new(GetObjectLegalHoldOutput {
             legal_hold: Some(ObjectLockLegalHold {
-                status: Some(ObjectLockLegalHoldStatus::from(legal_hold.unwrap_or_default())),
+                status: Some(ObjectLockLegalHoldStatus::from(status)),
             }),
         }))
     }
@@ -2173,6 +2243,103 @@ impl S3 for FS {
             request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)),
         }))
     }
+
+    async fn get_object_retention(
+        &self,
+        req: S3Request<GetObjectRetentionInput>,
+    ) -> S3Result<S3Response<GetObjectRetentionOutput>> {
+        let GetObjectRetentionInput {
+            bucket, key, version_id, ..
+        } = req.input;
+
+        let Some(store) = new_object_layer_fn() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        // check object lock
+        let _ = metadata_sys::get_object_lock_config(&bucket).await.map_err(to_s3_error)?;
+
+        let opts: ObjectOptions = get_opts(&bucket, &key, version_id, None, &req.headers)
+            .await
+            .map_err(to_s3_error)?;
+
+        let object_info = store.get_object_info(&bucket, &key, &opts).await.map_err(|e| {
+            error!("get_object_info failed, {}", e.to_string());
+            s3_error!(InternalError, "{}", e.to_string())
+        })?;
+
+        let mode = if let Some(ref ud) = object_info.user_defined {
+            ud.get("x-amz-object-lock-mode")
+                .map(|v| ObjectLockRetentionMode::from(v.as_str().to_string()))
+        } else {
+            None
+        };
+
+        let retain_until_date = if let Some(ref ud) = object_info.user_defined {
+            ud.get("x-amz-object-lock-retain-until-date")
+                .and_then(|v| OffsetDateTime::parse(v.as_str(), &Rfc3339).ok())
+                .map(Timestamp::from)
+        } else {
+            None
+        };
+
+        Ok(S3Response::new(GetObjectRetentionOutput {
+            retention: Some(ObjectLockRetention { mode, retain_until_date }),
+        }))
+    }
+
+    async fn put_object_retention(
+        &self,
+        req: S3Request<PutObjectRetentionInput>,
+    ) -> S3Result<S3Response<PutObjectRetentionOutput>> {
+        let PutObjectRetentionInput {
+            bucket,
+            key,
+            retention,
+            version_id,
+            ..
+        } = req.input;
+
+        let Some(store) = new_object_layer_fn() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        // check object lock
+        let _ = metadata_sys::get_object_lock_config(&bucket).await.map_err(to_s3_error)?;
+
+        // TODO: check allow
+
+        let mut eval_metadata = HashMap::new();
+
+        if let Some(v) = retention {
+            let mode = v.mode.map(|v| v.as_str().to_string()).unwrap_or_default();
+            let retain_until_date = v
+                .retain_until_date
+                .map(|v| OffsetDateTime::from(v).format(&Rfc3339).unwrap())
+                .unwrap_or_default();
+            let now = OffsetDateTime::now_utc();
+            eval_metadata.insert("x-amz-object-lock-mode".to_string(), mode);
+            eval_metadata.insert("x-amz-object-lock-retain-until-date".to_string(), retain_until_date);
+            eval_metadata.insert(
+                format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "objectlock-retention-timestamp"),
+                format!("{}.{:09}Z", now.format(&Rfc3339).unwrap(), now.nanosecond()),
+            );
+        }
+
+        let mut opts: ObjectOptions = get_opts(&bucket, &key, version_id, None, &req.headers)
+            .await
+            .map_err(to_s3_error)?;
+        opts.eval_metadata = Some(eval_metadata);
+
+        store.put_object_metadata(&bucket, &key, &opts).await.map_err(|e| {
+            error!("put_object_metadata failed, {}", e.to_string());
+            s3_error!(InternalError, "{}", e.to_string())
+        })?;
+
+        Ok(S3Response::new(PutObjectRetentionOutput {
+            request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)),
+        }))
+    }
 }
 
 #[allow(dead_code)]

rustfs/src/storage/error.rs

@@ -1,5 +1,5 @@
 use common::error::Error;
-use ecstore::{disk::error::is_err_file_not_found, store_err::StorageError};
+use ecstore::{bucket::error::BucketMetadataError, disk::error::is_err_file_not_found, store_err::StorageError};
 use s3s::{s3_error, S3Error, S3ErrorCode};
 pub fn to_s3_error(err: Error) -> S3Error {
     if let Some(storage_err) = err.downcast_ref::<StorageError>() {
@@ -80,6 +80,15 @@ pub fn to_s3_error(err: Error) -> S3Error {
             StorageError::DoneForNow => s3_error!(InternalError, "DoneForNow"),
         };
     }
+    //需要添加 not found bucket replication config
+    if let Some(meta_err) = err.downcast_ref::<BucketMetadataError>() {
+        return match meta_err {
+            BucketMetadataError::BucketReplicationConfigNotFound => {
+                S3Error::with_message(S3ErrorCode::ReplicationConfigurationNotFoundError, format!("{}", err))
+            }
+            _ => S3Error::with_message(S3ErrorCode::InternalError, format!("{}", err)), // 处理其他情况
+        };
+    }
 
     if is_err_file_not_found(&err) {
         return S3Error::with_message(S3ErrorCode::NoSuchKey, format!(" ec err {}", err));