diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index 8741a5e5b..f96b3e0d6 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -14,6 +14,7 @@ use super::profile::{TriggerProfileCPU, TriggerProfileMemory}; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::app::admin_usecase::DefaultAdminUsecase; use crate::server::{HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH}; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; @@ -51,9 +52,9 @@ pub(crate) enum HealthProbe { } pub(crate) fn collect_dependency_readiness() -> (bool, bool) { - let storage_ready = rustfs_ecstore::new_object_layer_fn().is_some(); - let iam_ready = rustfs_iam::get().is_ok(); - (storage_ready, iam_ready) + let usecase = DefaultAdminUsecase::from_global(); + let readiness = usecase.execute_collect_dependency_readiness(); + (readiness.storage_ready, readiness.iam_ready) } pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: HealthProbe) -> HealthCheckState { diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index 0809468b5..a2a5e42dd 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/rustfs/src/admin/handlers/pools.rs @@ -14,7 +14,6 @@ use http::{HeaderMap, StatusCode}; use matchit::Params; -use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn}; use rustfs_policy::policy::action::{Action, AdminAction}; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; use serde::Deserialize; @@ -27,11 +26,13 @@ use crate::{ auth::validate_admin_request, router::{AdminOperation, Operation, S3Router}, }, + app::admin_usecase::{DefaultAdminUsecase, QueryPoolStatusRequest}, auth::{check_key_valid, get_session_token}, error::ApiError, server::{ADMIN_PREFIX, RemoteAddr}, }; use hyper::Method; +use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn}; pub fn register_pool_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -90,25 +91,8 @@ impl Operation for ListPools { ) .await?; - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - let Some(endpoints) = GLOBAL_Endpoints.get() else { - return Err(s3_error!(NotImplemented)); - }; - - if endpoints.legacy() { - return Err(s3_error!(NotImplemented)); - } - - let mut pools_status = Vec::new(); - - for (idx, _) in endpoints.as_ref().iter().enumerate() { - let state = store.status(idx).await.map_err(ApiError::from)?; - - pools_status.push(state); - } + let usecase = DefaultAdminUsecase::from_global(); + let pools_status = usecase.execute_list_pool_statuses().await.map_err(S3Error::from)?; let data = serde_json::to_vec(&pools_status) .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?; @@ -157,14 +141,6 @@ impl Operation for StatusPool { ) .await?; - let Some(endpoints) = GLOBAL_Endpoints.get() else { - return Err(s3_error!(NotImplemented)); - }; - - if endpoints.legacy() { - return Err(s3_error!(NotImplemented)); - } - let query = { if let Some(query) = req.uri.query() { let input: StatusPoolQuery = @@ -175,27 +151,14 @@ impl Operation for StatusPool { } }; - let is_byid = query.by_id.as_str() == "true"; - - let has_idx = { - if is_byid { - let a = query.pool.parse::().unwrap_or_default(); - if a < endpoints.as_ref().len() { Some(a) } else { None } - } else { - endpoints.get_pool_idx(&query.pool) - } - }; - - let Some(idx) = has_idx else { - warn!("specified pool {} not found, please specify a valid pool", &query.pool); - return Err(s3_error!(InvalidArgument)); - }; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - let pools_status = store.status(idx).await.map_err(ApiError::from)?; + let usecase = DefaultAdminUsecase::from_global(); + let pools_status = usecase + .execute_query_pool_status(QueryPoolStatusRequest { + pool: query.pool, + by_id: query.by_id.as_str() == "true", + }) + .await + .map_err(S3Error::from)?; let data = serde_json::to_vec(&pools_status) .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed"))?; diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 19ec614c0..ae40483ad 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -15,20 +15,16 @@ use super::metrics; use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::app::admin_usecase::{DefaultAdminUsecase, QueryServerInfoRequest}; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_ecstore::admin_server_info::get_server_info; -use rustfs_ecstore::data_usage::load_data_usage_from_backend; -use rustfs_ecstore::new_object_layer_fn; -use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free}; -use rustfs_ecstore::store_api::StorageAPI; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use s3s::header::CONTENT_TYPE; use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use tracing::{debug, error, info, warn}; +use tracing::warn; pub fn register_system_route(r: &mut S3Router) -> std::io::Result<()> { r.insert( @@ -110,7 +106,12 @@ impl Operation for ServerInfoHandler { ) .await?; - let info = get_server_info(true).await; + let usecase = DefaultAdminUsecase::from_global(); + let info = usecase + .execute_query_server_info(QueryServerInfoRequest { include_pools: true }) + .await + .map_err(S3Error::from)? + .info; let data = serde_json::to_vec(&info) .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse serverInfo failed"))?; @@ -158,12 +159,8 @@ impl Operation for StorageInfoHandler { ) .await?; - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - // TODO:getAggregatedBackgroundHealState - let info = store.storage_info().await; + let usecase = DefaultAdminUsecase::from_global(); + let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?; let data = serde_json::to_vec(&info) .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize storage info"))?; @@ -203,84 +200,8 @@ impl Operation for DataUsageInfoHandler { ) .await?; - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - let mut info = load_data_usage_from_backend(store.clone()).await.map_err(|e| { - error!("load_data_usage_from_backend failed {:?}", e); - s3_error!(InternalError, "load_data_usage_from_backend failed") - })?; - - let sinfo = store.storage_info().await; - - // Use the fixed capacity calculation function (built-in deduplication) - let raw_total = get_total_usable_capacity(&sinfo.disks, &sinfo); - let raw_free = get_total_usable_capacity_free(&sinfo.disks, &sinfo); - - // Add a plausibility check (extra layer of protection) - const MAX_REASONABLE_CAPACITY: u64 = 100_000 * 1024 * 1024 * 1024 * 1024; // 100 PiB - const MIN_REASONABLE_CAPACITY: u64 = 1024 * 1024 * 1024; // 1 GiB - - let total_u64 = raw_total as u64; - let free_u64 = raw_free as u64; - - // Detect outliers - if total_u64 > MAX_REASONABLE_CAPACITY { - error!( - "Abnormal total capacity detected: {} bytes ({:.2} TiB), capping to physical capacity", - total_u64, - total_u64 as f64 / (1024.0_f64.powi(4)) - ); - - let disk_count = sinfo.disks.len(); - if disk_count > 0 { - use std::collections::HashSet; - let unique_disks: HashSet = sinfo - .disks - .iter() - .map(|d| format!("{}|{}", d.endpoint, d.drive_path)) - .collect(); - - let actual_disk_count = unique_disks.len(); - - if let Some(first_disk) = sinfo.disks.first() { - info.total_capacity = first_disk.total_space * actual_disk_count as u64; - info.total_free_capacity = first_disk.available_space * actual_disk_count as u64; - - info!( - "Applied capacity correction: {} unique disks, capacity per disk: {} bytes", - actual_disk_count, first_disk.total_space - ); - } else { - info.total_capacity = 0; - info.total_free_capacity = 0; - } - } else { - info.total_capacity = 0; - info.total_free_capacity = 0; - } - } else if total_u64 < MIN_REASONABLE_CAPACITY && total_u64 > 0 { - warn!( - "Unusually small total capacity: {} bytes ({:.2} GiB)", - total_u64, - total_u64 as f64 / (1024.0_f64.powi(3)) - ); - info.total_capacity = total_u64; - info.total_free_capacity = free_u64; - } else { - info.total_capacity = total_u64; - info.total_free_capacity = free_u64; - } - - info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity); - - debug!( - "Capacity statistics: total={:.2} TiB, free={:.2} TiB, used={:.2} TiB", - info.total_capacity as f64 / (1024.0_f64.powi(4)), - info.total_free_capacity as f64 / (1024.0_f64.powi(4)), - info.total_used_capacity as f64 / (1024.0_f64.powi(4)) - ); + let usecase = DefaultAdminUsecase::from_global(); + let info = usecase.execute_query_data_usage_info().await.map_err(S3Error::from)?; let data = serde_json::to_vec(&info) .map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse DataUsageInfo failed"))?; diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 67846dcf1..de9dac41f 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -15,11 +15,18 @@ //! Admin application use-case contracts. #![allow(dead_code)] -use crate::app::context::AppContext; +use crate::app::context::{AppContext, get_global_app_context}; use crate::error::ApiError; +use rustfs_common::data_usage::DataUsageInfo; use rustfs_ecstore::admin_server_info::get_server_info; -use rustfs_madmin::InfoMessage; +use rustfs_ecstore::data_usage::load_data_usage_from_backend; +use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free}; +use rustfs_ecstore::store_api::StorageAPI; +use rustfs_ecstore::{GLOBAL_Endpoints, new_object_layer_fn}; +use rustfs_madmin::{InfoMessage, StorageInfo}; +use s3s::S3ErrorCode; use std::sync::Arc; +use tracing::{debug, error, info, warn}; pub type AdminUsecaseResult = Result; @@ -38,30 +45,297 @@ impl std::fmt::Debug for QueryServerInfoResponse { } } +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct DependencyReadiness { + pub storage_ready: bool, + pub iam_ready: bool, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct QueryPoolStatusRequest { + pub pool: String, + pub by_id: bool, +} + #[async_trait::async_trait] pub trait AdminUsecase: Send + Sync { async fn query_server_info(&self, req: QueryServerInfoRequest) -> AdminUsecaseResult; + + async fn query_storage_info(&self) -> AdminUsecaseResult; + + async fn query_data_usage_info(&self) -> AdminUsecaseResult; + + async fn list_pool_statuses(&self) -> AdminUsecaseResult>; + + async fn query_pool_status(&self, req: QueryPoolStatusRequest) -> AdminUsecaseResult; + + fn collect_dependency_readiness(&self) -> DependencyReadiness; } -#[derive(Clone)] +#[derive(Clone, Default)] pub struct DefaultAdminUsecase { - context: Arc, + context: Option>, } impl DefaultAdminUsecase { pub fn new(context: Arc) -> Self { - Self { context } + Self { context: Some(context) } } - pub fn context(&self) -> Arc { + pub fn without_context() -> Self { + Self { context: None } + } + + pub fn from_global() -> Self { + Self { + context: get_global_app_context(), + } + } + + pub fn context(&self) -> Option> { self.context.clone() } + + fn app_error(code: S3ErrorCode, message: impl Into) -> ApiError { + ApiError { + code, + message: message.into(), + source: None, + } + } + + fn app_error_default(code: S3ErrorCode) -> ApiError { + let message = ApiError::error_code_to_message(&code); + Self::app_error(code, message) + } + + pub async fn execute_query_server_info(&self, req: QueryServerInfoRequest) -> AdminUsecaseResult { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let info = get_server_info(req.include_pools).await; + Ok(QueryServerInfoResponse { info }) + } + + pub async fn execute_query_storage_info(&self) -> AdminUsecaseResult { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let Some(store) = new_object_layer_fn() else { + return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); + }; + + Ok(store.storage_info().await) + } + + pub async fn execute_query_data_usage_info(&self) -> AdminUsecaseResult { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let Some(store) = new_object_layer_fn() else { + return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); + }; + + let mut info = load_data_usage_from_backend(store.clone()).await.map_err(|e| { + error!("load_data_usage_from_backend failed {:?}", e); + Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed") + })?; + + let storage_info = store.storage_info().await; + + // Keep the same capacity correction behavior as the previous admin handler implementation. + const MAX_REASONABLE_CAPACITY: u64 = 100_000 * 1024 * 1024 * 1024 * 1024; // 100 PiB + const MIN_REASONABLE_CAPACITY: u64 = 1024 * 1024 * 1024; // 1 GiB + + let total_u64 = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64; + let free_u64 = get_total_usable_capacity_free(&storage_info.disks, &storage_info) as u64; + + if total_u64 > MAX_REASONABLE_CAPACITY { + error!( + "Abnormal total capacity detected: {} bytes ({:.2} TiB), capping to physical capacity", + total_u64, + total_u64 as f64 / (1024.0_f64.powi(4)) + ); + + let disk_count = storage_info.disks.len(); + if disk_count > 0 { + use std::collections::HashSet; + let unique_disks: HashSet = storage_info + .disks + .iter() + .map(|disk| format!("{}|{}", disk.endpoint, disk.drive_path)) + .collect(); + + let actual_disk_count = unique_disks.len(); + + if let Some(first_disk) = storage_info.disks.first() { + info.total_capacity = first_disk.total_space * actual_disk_count as u64; + info.total_free_capacity = first_disk.available_space * actual_disk_count as u64; + + info!( + "Applied capacity correction: {} unique disks, capacity per disk: {} bytes", + actual_disk_count, first_disk.total_space + ); + } else { + info.total_capacity = 0; + info.total_free_capacity = 0; + } + } else { + info.total_capacity = 0; + info.total_free_capacity = 0; + } + } else if total_u64 < MIN_REASONABLE_CAPACITY && total_u64 > 0 { + warn!( + "Unusually small total capacity: {} bytes ({:.2} GiB)", + total_u64, + total_u64 as f64 / (1024.0_f64.powi(3)) + ); + info.total_capacity = total_u64; + info.total_free_capacity = free_u64; + } else { + info.total_capacity = total_u64; + info.total_free_capacity = free_u64; + } + + info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity); + + debug!( + "Capacity statistics: total={:.2} TiB, free={:.2} TiB, used={:.2} TiB", + info.total_capacity as f64 / (1024.0_f64.powi(4)), + info.total_free_capacity as f64 / (1024.0_f64.powi(4)), + info.total_used_capacity as f64 / (1024.0_f64.powi(4)) + ); + + Ok(info) + } + + pub async fn execute_list_pool_statuses(&self) -> AdminUsecaseResult> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let Some(store) = new_object_layer_fn() else { + return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); + }; + + let Some(endpoints) = GLOBAL_Endpoints.get() else { + return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); + }; + + if endpoints.legacy() { + return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); + } + + let mut pool_statuses = Vec::new(); + for (idx, _) in endpoints.as_ref().iter().enumerate() { + let state = store.status(idx).await.map_err(ApiError::from)?; + pool_statuses.push(state); + } + + Ok(pool_statuses) + } + + pub async fn execute_query_pool_status(&self, req: QueryPoolStatusRequest) -> AdminUsecaseResult { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let Some(endpoints) = GLOBAL_Endpoints.get() else { + return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); + }; + + if endpoints.legacy() { + return Err(Self::app_error_default(S3ErrorCode::NotImplemented)); + } + + let has_idx = if req.by_id { + let idx = req.pool.parse::().unwrap_or_default(); + if idx < endpoints.as_ref().len() { Some(idx) } else { None } + } else { + endpoints.get_pool_idx(&req.pool) + }; + + let Some(idx) = has_idx else { + warn!("specified pool {} not found, please specify a valid pool", req.pool); + return Err(Self::app_error_default(S3ErrorCode::InvalidArgument)); + }; + + let Some(store) = new_object_layer_fn() else { + return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); + }; + + store.status(idx).await.map_err(ApiError::from) + } + + pub fn execute_collect_dependency_readiness(&self) -> DependencyReadiness { + if let Some(context) = &self.context { + let _ = context.object_store(); + let _ = context.iam(); + } + + DependencyReadiness { + storage_ready: new_object_layer_fn().is_some(), + iam_ready: rustfs_iam::get().is_ok(), + } + } } #[async_trait::async_trait] impl AdminUsecase for DefaultAdminUsecase { async fn query_server_info(&self, req: QueryServerInfoRequest) -> AdminUsecaseResult { - let info = get_server_info(req.include_pools).await; - Ok(QueryServerInfoResponse { info }) + self.execute_query_server_info(req).await + } + + async fn query_storage_info(&self) -> AdminUsecaseResult { + self.execute_query_storage_info().await + } + + async fn query_data_usage_info(&self) -> AdminUsecaseResult { + self.execute_query_data_usage_info().await + } + + async fn list_pool_statuses(&self) -> AdminUsecaseResult> { + self.execute_list_pool_statuses().await + } + + async fn query_pool_status(&self, req: QueryPoolStatusRequest) -> AdminUsecaseResult { + self.execute_query_pool_status(req).await + } + + fn collect_dependency_readiness(&self) -> DependencyReadiness { + self.execute_collect_dependency_readiness() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn execute_query_storage_info_returns_internal_error_when_store_uninitialized() { + let usecase = DefaultAdminUsecase::without_context(); + + let err = usecase.execute_query_storage_info().await.unwrap_err(); + assert_eq!(err.code, S3ErrorCode::InternalError); + } + + #[tokio::test] + async fn execute_query_data_usage_info_returns_internal_error_when_store_uninitialized() { + let usecase = DefaultAdminUsecase::without_context(); + + let err = usecase.execute_query_data_usage_info().await.unwrap_err(); + assert_eq!(err.code, S3ErrorCode::InternalError); + } + + #[test] + fn execute_collect_dependency_readiness_returns_state_flags() { + let usecase = DefaultAdminUsecase::without_context(); + + let readiness = usecase.execute_collect_dependency_readiness(); + let _ = readiness.storage_ready; + let _ = readiness.iam_ready; } } diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index fd61514d6..4eb51fffe 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -32,11 +32,18 @@ use crate::storage::options::{ use crate::storage::*; use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD}; use bytes::Bytes; +use datafusion::arrow::{ + csv::WriterBuilder as CsvWriterBuilder, json::WriterBuilder as JsonWriterBuilder, json::writer::JsonArray, +}; use futures::StreamExt; use http::{HeaderMap, HeaderValue, StatusCode}; use rustfs_ecstore::StorageAPI; use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::{ + lifecycle::{ + bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts}, + lifecycle::{self, TransitionOptions}, + }, metadata_sys, object_lock::objectlock_sys::{BucketObjectLockSys, check_object_lock_for_deletion}, quota::QuotaOperation, @@ -59,7 +66,11 @@ use rustfs_filemeta::{ }; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, EtagReader, HardLimitReader, HashReader, Reader, WarpReader}; -use rustfs_s3select_api::object_store::bytes_stream; +use rustfs_s3select_api::{ + object_store::bytes_stream, + query::{Context, Query}, +}; +use rustfs_s3select_query::get_global_db; use rustfs_targets::EventName; use rustfs_utils::CompressionAlgorithm; use rustfs_utils::http::{ @@ -72,7 +83,7 @@ use rustfs_utils::http::{ }; use rustfs_utils::path::{is_dir_object, path_join_buf}; use s3s::dto::*; -use s3s::header::X_AMZ_RESTORE; +use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH}; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; use std::collections::HashMap; use std::convert::Infallible; @@ -80,6 +91,8 @@ use std::ops::Add; use std::str::FromStr; use std::sync::Arc; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; use tokio_util::io::{ReaderStream, StreamReader}; use tracing::{debug, error, info, instrument, warn}; use uuid::Uuid; @@ -2149,6 +2162,267 @@ impl DefaultObjectUsecase { result } + + #[instrument(level = "debug", skip(self, req))] + pub async fn execute_restore_object(&self, req: S3Request) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + let RestoreObjectInput { + bucket, + key: object, + restore_request: rreq, + version_id, + .. + } = req.input.clone(); + + let rreq = rreq.ok_or_else(|| { + S3Error::with_message(S3ErrorCode::Custom("ErrValidRestoreObject".into()), "restore request is required") + })?; + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + let version_id_str = version_id.clone().unwrap_or_default(); + let opts = post_restore_opts(&version_id_str, &bucket, &object) + .await + .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrPostRestoreOpts".into()), "restore object failed."))?; + + let mut obj_info = store + .get_object_info(&bucket, &object, &opts) + .await + .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrInvalidObjectState".into()), "restore object failed."))?; + + // Check if object is in a transitioned state + if obj_info.transitioned_object.status != lifecycle::TRANSITION_COMPLETE { + return Err(S3Error::with_message( + S3ErrorCode::Custom("ErrInvalidTransitionedState".into()), + "restore object failed.", + )); + } + + // Validate restore request + if let Err(e) = rreq.validate(store.clone()) { + return Err(S3Error::with_message( + S3ErrorCode::Custom("ErrValidRestoreObject".into()), + format!("Restore object validation failed: {}", e), + )); + } + + // Check if restore is already in progress + if obj_info.restore_ongoing && (rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT")) { + return Err(S3Error::with_message( + S3ErrorCode::Custom("ErrObjectRestoreAlreadyInProgress".into()), + "restore object failed.", + )); + } + + let mut already_restored = false; + if let Some(restore_expires) = obj_info.restore_expires + && !obj_info.restore_ongoing + && restore_expires.unix_timestamp() != 0 + { + already_restored = true; + } + + let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), *rreq.days.as_ref().unwrap_or(&1)); + let mut metadata = obj_info.user_defined.clone(); + + let mut header = HeaderMap::new(); + + let obj_info_ = obj_info.clone(); + if rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT") { + obj_info.metadata_only = true; + metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), rreq.days.unwrap_or(1).to_string()); + let request_date = OffsetDateTime::now_utc().format(&Rfc3339).map_err(|e| { + S3Error::with_message(S3ErrorCode::InternalError, format!("format restore request date failed: {}", e)) + })?; + metadata.insert(AMZ_RESTORE_REQUEST_DATE.to_string(), request_date); + if already_restored { + metadata.insert( + X_AMZ_RESTORE.as_str().to_string(), + RestoreStatus { + is_restore_in_progress: Some(false), + restore_expiry_date: Some(Timestamp::from(restore_expiry)), + } + .to_string(), + ); + } else { + metadata.insert( + X_AMZ_RESTORE.as_str().to_string(), + RestoreStatus { + is_restore_in_progress: Some(true), + restore_expiry_date: Some(Timestamp::from(OffsetDateTime::now_utc())), + } + .to_string(), + ); + } + obj_info.user_defined = metadata; + + store + .clone() + .copy_object( + &bucket, + &object, + &bucket, + &object, + &mut obj_info, + &ObjectOptions { + version_id: obj_info_.version_id.map(|v| v.to_string()), + ..Default::default() + }, + &ObjectOptions { + version_id: obj_info_.version_id.map(|v| v.to_string()), + mod_time: obj_info_.mod_time, + ..Default::default() + }, + ) + .await + .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrCopyObject".into()), "restore object failed."))?; + + if already_restored { + let output = RestoreObjectOutput { + request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), + restore_output_path: None, + }; + return Ok(S3Response::new(output)); + } + } + + // Handle output location for SELECT requests + if let Some(output_location) = &rreq.output_location + && let Some(s3) = &output_location.s3 + && !s3.bucket_name.is_empty() + { + let restore_object = Uuid::new_v4().to_string(); + if let Ok(header_value) = format!("{}{}{}", s3.bucket_name, s3.prefix, restore_object).parse() { + header.insert(X_AMZ_RESTORE_OUTPUT_PATH, header_value); + } + } + + // Spawn restoration task in the background + let store_clone = store.clone(); + let bucket_clone = bucket.clone(); + let object_clone = object.clone(); + let rreq_clone = rreq.clone(); + let version_id_clone = version_id.clone(); + + tokio::spawn(async move { + let opts = ObjectOptions { + transition: TransitionOptions { + restore_request: rreq_clone, + restore_expiry, + ..Default::default() + }, + version_id: version_id_clone, + ..Default::default() + }; + + if let Err(err) = store_clone + .restore_transitioned_object(&bucket_clone, &object_clone, &opts) + .await + { + warn!( + "unable to restore transitioned bucket/object {}/{}: {}", + bucket_clone, + object_clone, + err.to_string() + ); + // Note: Errors from background tasks cannot be returned to client + // Consider adding to monitoring/metrics system + } else { + info!("successfully restored transitioned object: {}/{}", bucket_clone, object_clone); + } + }); + + let output = RestoreObjectOutput { + request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), + restore_output_path: None, + }; + + Ok(S3Response::with_headers(output, header)) + } + + #[instrument(level = "debug", skip(self, req))] + pub async fn execute_select_object_content( + &self, + req: S3Request, + ) -> S3Result> { + if let Some(context) = &self.context { + let _ = context.object_store(); + } + + info!("handle select_object_content"); + + let input = Arc::new(req.input); + info!("{:?}", input); + + let db = get_global_db((*input).clone(), false).await.map_err(|e| { + error!("get global db failed, {}", e.to_string()); + s3_error!(InternalError, "{}", e.to_string()) + })?; + let query = Query::new(Context { input: input.clone() }, input.request.expression.clone()); + let result = db + .execute(&query) + .await + .map_err(|e| s3_error!(InternalError, "{}", e.to_string()))?; + + let results = result + .result() + .chunk_result() + .await + .map_err(|e| s3_error!(InternalError, "{}", e.to_string()))? + .to_vec(); + + let mut buffer = Vec::new(); + if input.request.output_serialization.csv.is_some() { + let mut csv_writer = CsvWriterBuilder::new().with_header(false).build(&mut buffer); + for batch in results { + csv_writer + .write(&batch) + .map_err(|e| s3_error!(InternalError, "can't encode output to csv. e: {}", e.to_string()))?; + } + } else if input.request.output_serialization.json.is_some() { + let mut json_writer = JsonWriterBuilder::new() + .with_explicit_nulls(true) + .build::<_, JsonArray>(&mut buffer); + for batch in results { + json_writer + .write(&batch) + .map_err(|e| s3_error!(InternalError, "can't encode output to json. e: {}", e.to_string()))?; + } + json_writer + .finish() + .map_err(|e| s3_error!(InternalError, "writer output into json error, e: {}", e.to_string()))?; + } else { + return Err(s3_error!( + InvalidArgument, + "Unsupported output format. Supported formats are CSV and JSON" + )); + } + + let (tx, rx) = mpsc::channel::>(2); + let stream = ReceiverStream::new(rx); + tokio::spawn(async move { + let _ = tx + .send(Ok(SelectObjectContentEvent::Cont(ContinuationEvent::default()))) + .await; + let _ = tx + .send(Ok(SelectObjectContentEvent::Records(RecordsEvent { + payload: Some(Bytes::from(buffer)), + }))) + .await; + let _ = tx.send(Ok(SelectObjectContentEvent::End(EndEvent::default()))).await; + + drop(tx); + }); + + Ok(S3Response::new(SelectObjectContentOutput { + payload: Some(SelectObjectContentEventStream::new(stream)), + })) + } } #[async_trait::async_trait] @@ -2278,4 +2552,47 @@ mod tests { let err = usecase.execute_head_object(req).await.unwrap_err(); assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } + + #[tokio::test] + async fn execute_restore_object_rejects_missing_restore_request() { + let input = RestoreObjectInput::builder() + .bucket("test-bucket".to_string()) + .key("test-key".to_string()) + .build() + .unwrap(); + + let req = build_request(input, Method::POST); + let usecase = DefaultObjectUsecase::without_context(); + + let err = usecase.execute_restore_object(req).await.unwrap_err(); + match err.code() { + S3ErrorCode::Custom(code) => assert_eq!(code, "ErrValidRestoreObject"), + code => panic!("unexpected error code: {:?}", code), + } + } + + #[tokio::test] + async fn execute_restore_object_returns_internal_error_when_store_uninitialized() { + let restore_request = RestoreRequest { + days: Some(1), + description: None, + glacier_job_parameters: None, + output_location: None, + select_parameters: None, + tier: None, + type_: None, + }; + let input = RestoreObjectInput::builder() + .bucket("test-bucket".to_string()) + .key("test-key".to_string()) + .restore_request(Some(restore_request)) + .build() + .unwrap(); + + let req = build_request(input, Method::POST); + let usecase = DefaultObjectUsecase::without_context(); + + let err = usecase.execute_restore_object(req).await.unwrap_err(); + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } } diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index ae3a2eafc..93b0b3418 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -27,19 +27,11 @@ use crate::storage::{ decrypt_managed_encryption_key, derive_part_nonce, get_buffer_size_opt_in, get_validated_store, has_replication_rules, parse_object_lock_legal_hold, parse_object_lock_retention, validate_bucket_object_lock_enabled, }; -use bytes::Bytes; -use datafusion::arrow::{ - csv::WriterBuilder as CsvWriterBuilder, json::WriterBuilder as JsonWriterBuilder, json::writer::JsonArray, -}; use futures::StreamExt; use http::{HeaderMap, StatusCode}; use metrics::{counter, histogram}; use rustfs_ecstore::{ bucket::{ - lifecycle::{ - bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts}, - lifecycle::{self, TransitionOptions}, - }, metadata::{ BUCKET_ACL_CONFIG, BUCKET_CORS_CONFIG, BUCKET_PUBLIC_ACCESS_BLOCK_CONFIG, BUCKET_REPLICATION_CONFIG, BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG, @@ -71,40 +63,27 @@ use rustfs_ecstore::{ }, }; use rustfs_filemeta::REPLICATE_INCOMING_DELETE; -use rustfs_filemeta::RestoreStatusOps; use rustfs_filemeta::{ReplicationStatusType, VersionPurgeStatusType}; use rustfs_kms::DataKey; use rustfs_notify::{EventArgsBuilder, notifier_global}; use rustfs_policy::policy::action::{Action, S3Action}; use rustfs_rio::{CompressReader, DecryptReader, EncryptReader, HashReader, Reader, WarpReader}; -use rustfs_s3select_api::query::{Context, Query}; -use rustfs_s3select_query::get_global_db; use rustfs_targets::EventName; use rustfs_utils::{ CompressionAlgorithm, extract_params_header, extract_resp_elements, get_request_host, get_request_port, get_request_user_agent, http::{ - AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE, + AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, headers::{AMZ_DECODED_CONTENT_LENGTH, RESERVED_METADATA_PREFIX_LOWER}, }, path::is_dir_object, }; use rustfs_zip::CompressionFormat; -use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH}; use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error}; use serde::{Deserialize, Serialize}; -use std::{ - collections::HashMap, - fmt::Debug, - path::Path, - sync::{Arc, LazyLock}, -}; +use std::{collections::HashMap, fmt::Debug, path::Path, sync::LazyLock}; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tokio::{ - io::{AsyncRead, AsyncSeek}, - sync::mpsc, -}; -use tokio_stream::wrappers::ReceiverStream; +use tokio::io::{AsyncRead, AsyncSeek}; use tokio_tar::Archive; use tokio_util::io::StreamReader; use tracing::{debug, error, info, instrument, warn}; @@ -2963,247 +2942,16 @@ impl S3 for FS { } async fn restore_object(&self, req: S3Request) -> S3Result> { - let RestoreObjectInput { - bucket, - key: object, - restore_request: rreq, - version_id, - .. - } = req.input.clone(); - - let rreq = rreq.ok_or_else(|| { - S3Error::with_message(S3ErrorCode::Custom("ErrValidRestoreObject".into()), "restore request is required") - })?; - - let Some(store) = new_object_layer_fn() else { - return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); - }; - - let version_id_str = version_id.clone().unwrap_or_default(); - let opts = post_restore_opts(&version_id_str, &bucket, &object) - .await - .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrPostRestoreOpts".into()), "restore object failed."))?; - - let mut obj_info = store - .get_object_info(&bucket, &object, &opts) - .await - .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrInvalidObjectState".into()), "restore object failed."))?; - - // Check if object is in a transitioned state - if obj_info.transitioned_object.status != lifecycle::TRANSITION_COMPLETE { - return Err(S3Error::with_message( - S3ErrorCode::Custom("ErrInvalidTransitionedState".into()), - "restore object failed.", - )); - } - - // Validate restore request - if let Err(e) = rreq.validate(store.clone()) { - return Err(S3Error::with_message( - S3ErrorCode::Custom("ErrValidRestoreObject".into()), - format!("Restore object validation failed: {}", e), - )); - } - - // Check if restore is already in progress - if obj_info.restore_ongoing && (rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT")) { - return Err(S3Error::with_message( - S3ErrorCode::Custom("ErrObjectRestoreAlreadyInProgress".into()), - "restore object failed.", - )); - } - - let mut already_restored = false; - if let Some(restore_expires) = obj_info.restore_expires - && !obj_info.restore_ongoing - && restore_expires.unix_timestamp() != 0 - { - already_restored = true; - } - - let restore_expiry = lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), *rreq.days.as_ref().unwrap_or(&1)); - let mut metadata = obj_info.user_defined.clone(); - - let mut header = HeaderMap::new(); - - let obj_info_ = obj_info.clone(); - if rreq.type_.as_ref().is_none_or(|t| t.as_str() != "SELECT") { - obj_info.metadata_only = true; - metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), rreq.days.unwrap_or(1).to_string()); - metadata.insert(AMZ_RESTORE_REQUEST_DATE.to_string(), OffsetDateTime::now_utc().format(&Rfc3339).unwrap()); - if already_restored { - metadata.insert( - X_AMZ_RESTORE.as_str().to_string(), - RestoreStatus { - is_restore_in_progress: Some(false), - restore_expiry_date: Some(Timestamp::from(restore_expiry)), - } - .to_string(), - ); - } else { - metadata.insert( - X_AMZ_RESTORE.as_str().to_string(), - RestoreStatus { - is_restore_in_progress: Some(true), - restore_expiry_date: Some(Timestamp::from(OffsetDateTime::now_utc())), - } - .to_string(), - ); - } - obj_info.user_defined = metadata; - - store - .clone() - .copy_object( - &bucket, - &object, - &bucket, - &object, - &mut obj_info, - &ObjectOptions { - version_id: obj_info_.version_id.map(|v| v.to_string()), - ..Default::default() - }, - &ObjectOptions { - version_id: obj_info_.version_id.map(|v| v.to_string()), - mod_time: obj_info_.mod_time, - ..Default::default() - }, - ) - .await - .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrCopyObject".into()), "restore object failed."))?; - - if already_restored { - let output = RestoreObjectOutput { - request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), - restore_output_path: None, - }; - return Ok(S3Response::new(output)); - } - } - - // Handle output location for SELECT requests - if let Some(output_location) = &rreq.output_location - && let Some(s3) = &output_location.s3 - && !s3.bucket_name.is_empty() - { - let restore_object = Uuid::new_v4().to_string(); - header.insert( - X_AMZ_RESTORE_OUTPUT_PATH, - format!("{}{}{}", s3.bucket_name, s3.prefix, restore_object).parse().unwrap(), - ); - } - - // Spawn restoration task in the background - let store_clone = store.clone(); - let bucket_clone = bucket.clone(); - let object_clone = object.clone(); - let rreq_clone = rreq.clone(); - let version_id_clone = version_id.clone(); - - tokio::spawn(async move { - let opts = ObjectOptions { - transition: TransitionOptions { - restore_request: rreq_clone, - restore_expiry, - ..Default::default() - }, - version_id: version_id_clone, - ..Default::default() - }; - - if let Err(err) = store_clone - .restore_transitioned_object(&bucket_clone, &object_clone, &opts) - .await - { - warn!( - "unable to restore transitioned bucket/object {}/{}: {}", - bucket_clone, - object_clone, - err.to_string() - ); - // Note: Errors from background tasks cannot be returned to client - // Consider adding to monitoring/metrics system - } else { - info!("successfully restored transitioned object: {}/{}", bucket_clone, object_clone); - } - }); - - let output = RestoreObjectOutput { - request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), - restore_output_path: None, - }; - - Ok(S3Response::with_headers(output, header)) + let usecase = DefaultObjectUsecase::from_global(); + usecase.execute_restore_object(req).await } async fn select_object_content( &self, req: S3Request, ) -> S3Result> { - info!("handle select_object_content"); - - let input = Arc::new(req.input); - info!("{:?}", input); - - let db = get_global_db((*input).clone(), false).await.map_err(|e| { - error!("get global db failed, {}", e.to_string()); - s3_error!(InternalError, "{}", e.to_string()) - })?; - let query = Query::new(Context { input: input.clone() }, input.request.expression.clone()); - let result = db - .execute(&query) - .await - .map_err(|e| s3_error!(InternalError, "{}", e.to_string()))?; - - let results = result.result().chunk_result().await.unwrap().to_vec(); - - let mut buffer = Vec::new(); - if input.request.output_serialization.csv.is_some() { - let mut csv_writer = CsvWriterBuilder::new().with_header(false).build(&mut buffer); - for batch in results { - csv_writer - .write(&batch) - .map_err(|e| s3_error!(InternalError, "can't encode output to csv. e: {}", e.to_string()))?; - } - } else if input.request.output_serialization.json.is_some() { - let mut json_writer = JsonWriterBuilder::new() - .with_explicit_nulls(true) - .build::<_, JsonArray>(&mut buffer); - for batch in results { - json_writer - .write(&batch) - .map_err(|e| s3_error!(InternalError, "can't encode output to json. e: {}", e.to_string()))?; - } - json_writer - .finish() - .map_err(|e| s3_error!(InternalError, "writer output into json error, e: {}", e.to_string()))?; - } else { - return Err(s3_error!( - InvalidArgument, - "Unsupported output format. Supported formats are CSV and JSON" - )); - } - - let (tx, rx) = mpsc::channel::>(2); - let stream = ReceiverStream::new(rx); - tokio::spawn(async move { - let _ = tx - .send(Ok(SelectObjectContentEvent::Cont(ContinuationEvent::default()))) - .await; - let _ = tx - .send(Ok(SelectObjectContentEvent::Records(RecordsEvent { - payload: Some(Bytes::from(buffer)), - }))) - .await; - let _ = tx.send(Ok(SelectObjectContentEvent::End(EndEvent::default()))).await; - - drop(tx); - }); - - Ok(S3Response::new(SelectObjectContentOutput { - payload: Some(SelectObjectContentEventStream::new(stream)), - })) + let usecase = DefaultObjectUsecase::from_global(); + usecase.execute_select_object_content(req).await } #[instrument(level = "debug", skip(self, req))]