mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-09 22:59:59 +00:00
weisd
492 lines
19 KiB
Rust
492 lines
19 KiB
Rust
// Copyright 2024 RustFS Team
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
#![allow(unused_variables, unused_mut, unused_must_use)]
|
|
|
|
use http::{HeaderMap, StatusCode};
|
|
//use iam::get_global_action_cred;
|
|
use matchit::Params;
|
|
use rustfs_policy::policy::action::{Action, AdminAction};
|
|
use s3s::{
|
|
Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result,
|
|
header::{CONTENT_LENGTH, CONTENT_TYPE},
|
|
s3_error,
|
|
};
|
|
use serde_urlencoded::from_bytes;
|
|
use time::OffsetDateTime;
|
|
use tracing::{debug, warn};
|
|
|
|
use crate::{
|
|
admin::{auth::validate_admin_request, router::Operation},
|
|
auth::{check_key_valid, get_session_token},
|
|
};
|
|
|
|
use rustfs_ecstore::{
|
|
config::storageclass,
|
|
global::GLOBAL_TierConfigMgr,
|
|
tier::{
|
|
tier::{ERR_TIER_BACKEND_IN_USE, ERR_TIER_BACKEND_NOT_EMPTY, ERR_TIER_MISSING_CREDENTIALS},
|
|
tier_admin::TierCreds,
|
|
tier_config::{TierConfig, TierType},
|
|
tier_handlers::{
|
|
ERR_TIER_ALREADY_EXISTS, ERR_TIER_CONNECT_ERR, ERR_TIER_INVALID_CREDENTIALS, ERR_TIER_NAME_NOT_UPPERCASE,
|
|
ERR_TIER_NOT_FOUND,
|
|
},
|
|
},
|
|
};
|
|
|
|
#[derive(Debug, Clone, serde::Deserialize, Default)]
|
|
pub struct AddTierQuery {
|
|
#[serde(rename = "accessKey")]
|
|
#[allow(dead_code)]
|
|
pub access_key: Option<String>,
|
|
#[allow(dead_code)]
|
|
pub status: Option<String>,
|
|
#[serde(rename = "secretKey")]
|
|
#[allow(dead_code)]
|
|
pub secret_key: Option<String>,
|
|
#[serde(rename = "serviceName")]
|
|
#[allow(dead_code)]
|
|
pub service_name: Option<String>,
|
|
#[serde(rename = "sessionToken")]
|
|
#[allow(dead_code)]
|
|
pub session_token: Option<String>,
|
|
pub tier: Option<String>,
|
|
#[serde(rename = "tierName")]
|
|
#[allow(dead_code)]
|
|
pub tier_name: Option<String>,
|
|
#[serde(rename = "tierType")]
|
|
#[allow(dead_code)]
|
|
pub tier_type: Option<String>,
|
|
pub force: Option<String>,
|
|
}
|
|
|
|
pub struct AddTier {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for AddTier {
|
|
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: AddTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
AddTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::SetTierAction)]).await?;
|
|
|
|
let mut input = req.input;
|
|
let body = match input.store_all_unlimited().await {
|
|
Ok(b) => b,
|
|
Err(e) => {
|
|
warn!("get body failed, e: {:?}", e);
|
|
return Err(s3_error!(InvalidRequest, "get body failed"));
|
|
}
|
|
};
|
|
|
|
let mut args: TierConfig = serde_json::from_slice(&body)
|
|
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("unmarshal body err {e}")))?;
|
|
|
|
match args.tier_type {
|
|
TierType::S3 => {
|
|
args.name = args.s3.clone().unwrap().name;
|
|
}
|
|
TierType::RustFS => {
|
|
args.name = args.rustfs.clone().unwrap().name;
|
|
}
|
|
TierType::MinIO => {
|
|
args.name = args.minio.clone().unwrap().name;
|
|
}
|
|
_ => (),
|
|
}
|
|
debug!("add tier args {:?}", args);
|
|
|
|
let mut force: bool = false;
|
|
let force_str = query.force.clone().unwrap_or_default();
|
|
if !force_str.is_empty() {
|
|
force = force_str.parse().map_err(|e| {
|
|
warn!("parse force failed, e: {:?}", e);
|
|
s3_error!(InvalidRequest, "parse force failed")
|
|
})?;
|
|
}
|
|
match args.name.as_str() {
|
|
storageclass::STANDARD | storageclass::RRS => {
|
|
warn!("tier reserved name, args.name: {}", args.name);
|
|
return Err(s3_error!(InvalidRequest, "Cannot use reserved tier name"));
|
|
}
|
|
&_ => (),
|
|
}
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
|
|
//tier_config_mgr.reload(api);
|
|
if let Err(err) = tier_config_mgr.add(args, force).await {
|
|
if err.code == ERR_TIER_ALREADY_EXISTS.code {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierNameAlreadyExist".into()),
|
|
"tier name already exists!",
|
|
));
|
|
} else if err.code == ERR_TIER_NAME_NOT_UPPERCASE.code {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierNameNotUppercase".into()),
|
|
"tier name not uppercase!",
|
|
));
|
|
} else if err.code == ERR_TIER_BACKEND_IN_USE.code {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierNameBackendInUse!".into()),
|
|
"tier name backend in use!",
|
|
));
|
|
} else if err.code == ERR_TIER_CONNECT_ERR.code {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierConnectError".into()),
|
|
"tier connect error!",
|
|
));
|
|
} else if err.code == ERR_TIER_INVALID_CREDENTIALS.code {
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom(err.code.clone().into()), err.message.clone()));
|
|
} else {
|
|
warn!("tier_config_mgr add failed, e: {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierAddFailed".into()),
|
|
format!("tier add failed. {err}"),
|
|
));
|
|
}
|
|
}
|
|
if let Err(e) = tier_config_mgr.save().await {
|
|
warn!("tier_config_mgr save failed, e: {:?}", e);
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierAddFailed".into()), "tier save failed"));
|
|
}
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
|
|
}
|
|
}
|
|
|
|
pub struct EditTier {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for EditTier {
|
|
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: AddTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
AddTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::SetTierAction)]).await?;
|
|
|
|
let mut input = req.input;
|
|
let body = match input.store_all_unlimited().await {
|
|
Ok(b) => b,
|
|
Err(e) => {
|
|
warn!("get body failed, e: {:?}", e);
|
|
return Err(s3_error!(InvalidRequest, "get body failed"));
|
|
}
|
|
};
|
|
|
|
let creds: TierCreds = serde_json::from_slice(&body)
|
|
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("unmarshal body err {e}")))?;
|
|
|
|
debug!("edit tier args {:?}", creds);
|
|
|
|
let tier_name = params.get("tiername").map(|s| s.to_string()).unwrap_or_default();
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
|
|
//tier_config_mgr.reload(api);
|
|
if let Err(err) = tier_config_mgr.edit(&tier_name, creds).await {
|
|
if err.code == ERR_TIER_NOT_FOUND.code {
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierNotFound".into()), "tier not found!"));
|
|
} else if err.code == ERR_TIER_MISSING_CREDENTIALS.code {
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierMissingCredentials".into()),
|
|
"tier missing credentials!",
|
|
));
|
|
} else {
|
|
warn!("tier_config_mgr edit failed, e: {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierEditFailed".into()),
|
|
format!("tier edit failed. {err}"),
|
|
));
|
|
}
|
|
}
|
|
if let Err(e) = tier_config_mgr.save().await {
|
|
warn!("tier_config_mgr save failed, e: {:?}", e);
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierEditFailed".into()), "tier save failed"));
|
|
}
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, serde::Deserialize, Default)]
|
|
pub struct BucketQuery {
|
|
#[serde(rename = "bucket")]
|
|
#[allow(dead_code)]
|
|
pub bucket: String,
|
|
}
|
|
pub struct ListTiers {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for ListTiers {
|
|
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: BucketQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
BucketQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::ListTierAction)]).await?;
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.read().await;
|
|
let tiers = tier_config_mgr.list_tiers();
|
|
|
|
let data = serde_json::to_vec(&tiers)
|
|
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tiers err {e}")))?;
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
|
|
}
|
|
}
|
|
|
|
pub struct RemoveTier {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for RemoveTier {
|
|
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: AddTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
AddTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::SetTierAction)]).await?;
|
|
|
|
let mut force: bool = false;
|
|
let force_str = query.force.clone().unwrap_or_default();
|
|
if !force_str.is_empty() {
|
|
force = force_str.parse().map_err(|e| {
|
|
warn!("parse force failed, e: {:?}", e);
|
|
s3_error!(InvalidRequest, "parse force failed")
|
|
})?;
|
|
}
|
|
|
|
let tier_name = params.get("tiername").map(|s| s.to_string()).unwrap_or_default();
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
|
|
//tier_config_mgr.reload(api);
|
|
if let Err(err) = tier_config_mgr.remove(&tier_name, force).await {
|
|
if err.code == ERR_TIER_NOT_FOUND.code {
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierNotFound".into()), "tier not found."));
|
|
} else if err.code == ERR_TIER_BACKEND_NOT_EMPTY.code {
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierNameBackendInUse".into()), "tier is used."));
|
|
} else {
|
|
warn!("tier_config_mgr remove failed, e: {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierRemoveFailed".into()),
|
|
format!("tier remove failed. {err}"),
|
|
));
|
|
}
|
|
}
|
|
|
|
if let Err(e) = tier_config_mgr.save().await {
|
|
warn!("tier_config_mgr save failed, e: {:?}", e);
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierRemoveFailed".into()), "tier save failed"));
|
|
}
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
|
|
}
|
|
}
|
|
|
|
#[allow(dead_code)]
|
|
pub struct VerifyTier {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for VerifyTier {
|
|
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: AddTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
AddTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::ListTierAction)]).await?;
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
|
|
tier_config_mgr.verify(&query.tier.unwrap()).await;
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
|
|
}
|
|
}
|
|
|
|
pub struct GetTierInfo {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for GetTierInfo {
|
|
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::ListTierAction)]).await?;
|
|
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: AddTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
AddTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let tier_config_mgr = GLOBAL_TierConfigMgr.read().await;
|
|
let info = tier_config_mgr.get(&query.tier.unwrap());
|
|
|
|
let data = serde_json::to_vec(&info)
|
|
.map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal tier err {e}")))?;
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, serde::Deserialize, Default)]
|
|
pub struct ClearTierQuery {
|
|
pub rand: Option<String>,
|
|
pub force: String,
|
|
}
|
|
|
|
pub struct ClearTier {}
|
|
#[async_trait::async_trait]
|
|
impl Operation for ClearTier {
|
|
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
|
let query = {
|
|
if let Some(query) = req.uri.query() {
|
|
let input: ClearTierQuery =
|
|
from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get query failed"))?;
|
|
input
|
|
} else {
|
|
ClearTierQuery::default()
|
|
}
|
|
};
|
|
|
|
let Some(input_cred) = req.credentials else {
|
|
return Err(s3_error!(InvalidRequest, "get cred failed"));
|
|
};
|
|
|
|
let (cred, owner) =
|
|
check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
|
|
|
|
validate_admin_request(&req.headers, &cred, owner, false, vec![Action::AdminAction(AdminAction::SetTierAction)]).await?;
|
|
|
|
let mut force: bool = false;
|
|
let force_str = query.force;
|
|
if !force_str.is_empty() {
|
|
force = force_str.parse().unwrap();
|
|
}
|
|
|
|
let t = OffsetDateTime::now_utc();
|
|
let mut rand = "AGD1R25GI3I1GJGUGJFD7FBS4DFAASDF".to_string();
|
|
rand.insert_str(3, &t.day().to_string());
|
|
rand.insert_str(17, &t.month().to_string());
|
|
rand.insert_str(23, &t.year().to_string());
|
|
warn!("tier_config_mgr rand: {}", rand);
|
|
if query.rand != Some(rand) {
|
|
return Err(s3_error!(InvalidRequest, "get rand failed"));
|
|
};
|
|
|
|
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
|
|
//tier_config_mgr.reload(api);
|
|
if let Err(err) = tier_config_mgr.clear_tier(force).await {
|
|
warn!("tier_config_mgr clear failed, e: {:?}", err);
|
|
return Err(S3Error::with_message(
|
|
S3ErrorCode::Custom("TierClearFailed".into()),
|
|
format!("tier clear failed. {err}"),
|
|
));
|
|
}
|
|
if let Err(e) = tier_config_mgr.save().await {
|
|
warn!("tier_config_mgr save failed, e: {:?}", e);
|
|
return Err(S3Error::with_message(S3ErrorCode::Custom("TierEditFailed".into()), "tier save failed"));
|
|
}
|
|
|
|
let mut header = HeaderMap::new();
|
|
header.insert(CONTENT_TYPE, "application/json".parse().unwrap());
|
|
header.insert(CONTENT_LENGTH, "0".parse().unwrap());
|
|
Ok(S3Response::with_headers((StatusCode::OK, Body::empty()), header))
|
|
}
|
|
}
|