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

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

Browse Source 
# Conflicts:
#	Cargo.lock
#	Cargo.toml
#	rustfs/src/main.rs
This commit is contained in:
houseme
2025-03-06 16:54:17 +08:00
parent dab937d241 a0fef99f8a
commit 4ecb53c526
39 changed files with 687 additions and 747 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+5 -3
View File
@@ -59,7 +59,7 @@ tower.workspace = true
tracing-error.workspace = true
tracing-subscriber.workspace = true
transform-stream.workspace = true
uuid = "1.12.1"
uuid = "1.15.1"
url.workspace = true
admin = { path = "../api/admin" }
axum.workspace = true
@@ -73,7 +73,9 @@ iam = { path = "../iam" }
jsonwebtoken = "9.3.0"
tower-http = { version = "0.6.2", features = ["cors"] }
mime_guess = "2.0.5"
rust-embed = { version = "8.5.0", features = ["interpolate-folder-path"] }
rust-embed = { workspace = true, features = ["interpolate-folder-path"] }
local-ip-address = { workspace = true }
chrono = { workspace = true }
[build-dependencies]
prost-build.workspace = true
@@ -85,7 +87,7 @@ futures-util.workspace = true
# uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
ecstore = { path = "../ecstore" }
s3s.workspace = true
clap = { version = "4.5.27", features = ["derive","env"] }
clap = { version = "4.5.31", features = ["derive", "env"] }
tracing-subscriber = { version = "0.3.19", features = ["env-filter", "time"] }
hyper-util = { version = "0.1.10", features = [
"tokio",
rustfs/src/admin/handlers/service_account.rs
+11 -1
View File
@@ -53,6 +53,16 @@ impl Operation for AddServiceAccount {
.validate()
.map_err(|e| S3Error::with_message(InvalidRequest, e.to_string()))?;
let session_policy = if let Some(policy) = &create_req.policy {
let p = Policy::parse_config(policy.as_bytes()).map_err(|e| {
debug!("parse policy failed, e: {:?}", e);
s3_error!(InvalidArgument, "parse policy failed")
})?;
Some(p)
} else {
None
};
let Some(sys_cred) = get_global_action_cred() else {
return Err(s3_error!(InvalidRequest, "get sys cred failed"));
};
@@ -95,7 +105,7 @@ impl Operation for AddServiceAccount {
name: create_req.name,
description: create_req.description,
expiration: create_req.expiration,
session_policy: create_req.policy.and_then(|p| Policy::parse_config(p.as_bytes()).ok()),
session_policy,
..Default::default()
};
rustfs/src/config/mod.rs
+1 -1
View File
@@ -60,6 +60,6 @@ pub struct Opt {
#[arg(long, default_value_t = false, env = "RUSTFS_CONSOLE_ENABLE")]
pub console_enable: bool,
#[arg(long, default_value_t = format!("127.0.0.1:{}", 0), env = "RUSTFS_CONSOLE_ADDRESS")]
#[arg(long, default_value_t = format!("127.0.0.1:{}", 9002), env = "RUSTFS_CONSOLE_ADDRESS")]
pub console_address: String,
}
rustfs/src/console.rs
+53 -6
View File
@@ -9,6 +9,12 @@ use axum::{
use mime_guess::from_path;
use rust_embed::RustEmbed;
use serde::Serialize;
use shadow_rs::shadow;
use std::net::Ipv4Addr;
use std::sync::OnceLock;
use tracing::info;
shadow!(build);
#[derive(RustEmbed)]
#[folder = "$CARGO_MANIFEST_DIR/static"]
@@ -42,11 +48,12 @@ async fn static_handler(uri: axum::http::Uri) -> impl IntoResponse {
}
#[derive(Debug, Serialize)]
struct Config {
pub(crate) struct Config {
api: Api,
s3: S3,
release: Release,
license: License,
doc: String,
}
impl Config {
@@ -67,12 +74,30 @@ impl Config {
name: "Apache-2.0".to_string(),
url: "https://www.apache.org/licenses/LICENSE-2.0".to_string(),
},
doc: "https://rustfs.com/docs/".to_string(),
}
}
fn to_json(&self) -> String {
serde_json::to_string(self).unwrap_or_default()
}
pub(crate) fn version(&self) -> String {
format!(
"RELEASE.{} (rust {} {})",
self.release.date.clone(),
build::RUST_VERSION,
build::BUILD_TARGET
)
}
pub(crate) fn license(&self) -> String {
format!("{} {}", self.license.name.clone(), self.license.url.clone())
}
pub(crate) fn doc(&self) -> String {
self.doc.clone()
}
}
#[derive(Debug, Serialize)]
@@ -99,8 +124,27 @@ struct License {
url: String,
}
async fn config_handler(axum::extract::Extension(fs_addr): axum::extract::Extension<String>) -> impl IntoResponse {
let cfg = Config::new(&fs_addr, "v0.0.1", "2025-01-01").to_json();
pub(crate) static CONSOLE_CONFIG: OnceLock<Config> = OnceLock::new();
pub(crate) fn init_console_cfg(fs_addr: &str) {
CONSOLE_CONFIG.get_or_init(|| {
let ver = {
if !build::TAG.is_empty() {
build::TAG.to_string()
} else if !build::SHORT_COMMIT.is_empty() {
format!("@{}", build::SHORT_COMMIT)
} else {
build::PKG_VERSION.to_string()
}
};
Config::new(fs_addr, ver.as_str(), build::COMMIT_DATE_3339)
});
}
#[allow(clippy::const_is_empty)]
async fn config_handler() -> impl IntoResponse {
let cfg = CONSOLE_CONFIG.get().unwrap().to_json();
Response::builder()
.header("content-type", "application/json")
@@ -109,15 +153,18 @@ async fn config_handler(axum::extract::Extension(fs_addr): axum::extract::Extens
.unwrap()
}
pub async fn start_static_file_server(addrs: &str, fs_addr: &str) {
pub async fn start_static_file_server(addrs: &str, local_ip: Ipv4Addr, access_key: &str, secret_key: &str) {
// 创建路由
let app = Router::new()
.route("/config.json", get(config_handler).layer(axum::extract::Extension(fs_addr.to_owned())))
.route("/config.json", get(config_handler))
.nest_service("/", get(static_handler));
let listener = tokio::net::TcpListener::bind(addrs).await.unwrap();
let local_addr = listener.local_addr().unwrap();
println!("console running on: http://{} with s3 api {}", listener.local_addr().unwrap(), fs_addr);
info!("WebUI: http://{}:{} http://127.0.0.1:{}", local_ip, local_addr.port(), local_addr.port());
info!(" RootUser: {}", access_key);
info!(" RootPass: {}", secret_key);
axum::serve(listener, app).await.unwrap();
}
rustfs/src/main.rs
+63 -18
View File
@@ -6,13 +6,17 @@ mod grpc;
mod logging;
mod service;
mod storage;
mod utils;
use crate::auth::IAMAuth;
use crate::console::{init_console_cfg, CONSOLE_CONFIG};
use chrono::Datelike;
use clap::Parser;
use common::{
error::{Error, Result},
globals::set_global_addr,
};
use config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY};
use ecstore::heal::background_heal_ops::init_auto_heal;
use ecstore::utils::net::{self, get_available_port};
use ecstore::{
@@ -39,18 +43,20 @@ use tonic::{metadata::MetadataValue, Request, Status};
use tower_http::cors::CorsLayer;
use tracing::{debug, error, info, warn};
use tracing_error::ErrorLayer;
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
fn setup_tracing() {
use tracing_subscriber::EnvFilter;
let env_filter = EnvFilter::from_default_env();
let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info"));
let enable_color = std::io::stdout().is_terminal();
let subscriber = fmt()
let subscriber = tracing_subscriber::fmt::fmt()
.pretty()
.with_env_filter(env_filter)
.with_ansi(enable_color)
.with_file(true)
.with_line_number(true)
.finish()
.with(ErrorLayer::default());
@@ -66,6 +72,18 @@ fn check_auth(req: Request<()>) -> Result<Request<()>, Status> {
}
}
fn print_server_info() {
let cfg = CONSOLE_CONFIG.get().unwrap();
let current_year = chrono::Utc::now().year();
// 使用自定义宏打印服务器信息
info!("RustFS Object Storage Server");
info!("Copyright: 2024-{} RustFS, Inc", current_year);
info!("License: {}", cfg.license());
info!("Version: {}", cfg.version());
info!("Docs: {}", cfg.doc());
}
fn main() -> Result<()> {
//解析获得到的参数
let opt = config::Opt::parse();
@@ -102,13 +120,39 @@ async fn run(opt: config::Opt) -> Result<()> {
let listener = TcpListener::bind(server_address.clone()).await?;
//获取监听地址
let local_addr: SocketAddr = listener.local_addr()?;
let local_ip = utils::get_local_ip().ok_or(local_addr.ip()).unwrap();
// 用于 rpc
let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address.clone().as_str(), opt.volumes.clone())
.map_err(|err| Error::from_string(err.to_string()))?;
// Print RustFS-style logging for pool formatting
for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
debug!(
info!(
"Formatting {}st pool, {} set(s), {} drives per set.",
i + 1,
eps.set_count,
eps.drives_per_set
);
// Add warning for host with multiple drives in a set (similar to RustFS)
if eps.drives_per_set > 1 {
warn!("WARNING: Host local has more than 0 drives of set. A host failure will result in data becoming unavailable.");
}
}
// Detailed endpoint information (showing all API endpoints)
let api_endpoints = format!("http://{}:{}", local_ip, server_port);
let localhost_endpoint = format!("http://127.0.0.1:{}", server_port);
info!("API: {} {}", api_endpoints, localhost_endpoint);
info!(" RootUser: {}", opt.access_key.clone());
info!(" RootPass: {}", opt.secret_key.clone());
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
warn!("Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables", DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY);
}
for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
info!(
"created endpoints {}, set_count:{}, drives_per_set: {}, cmd: {:?}",
i, eps.set_count, eps.drives_per_set, eps.cmd_line
);
@@ -131,9 +175,12 @@ async fn run(opt: config::Opt) -> Result<()> {
// let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(server_address.clone(), endpoint_pools).await?);
let mut b = S3ServiceBuilder::new(store.clone());
let access_key = opt.access_key.clone();
let secret_key = opt.secret_key.clone();
//显示 info 信息
info!("authentication is enabled {}, {}", &opt.access_key, &opt.secret_key);
b.set_auth(IAMAuth::new(opt.access_key, opt.secret_key));
debug!("authentication is enabled {}, {}", &access_key, &secret_key);
b.set_auth(IAMAuth::new(access_key, secret_key));
b.set_access(store.clone());
@@ -175,11 +222,10 @@ async fn run(opt: config::Opt) -> Result<()> {
let http_server = ConnBuilder::new(TokioExecutor::new());
let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
let graceful = hyper_util::server::graceful::GracefulShutdown::new();
println!("server is running at http://{local_addr}");
loop {
let (socket, _) = tokio::select! {
res = listener.accept() => {
res = listener.accept() => {
match res {
Ok(conn) => conn,
Err(err) => {
@@ -222,7 +268,7 @@ async fn run(opt: config::Opt) -> Result<()> {
error!("ECStore init faild {:?}", &err);
Error::from_string(err.to_string())
})?;
warn!(" init store success!");
debug!("init store success!");
init_iam_sys(store.clone()).await.unwrap();
@@ -236,18 +282,17 @@ async fn run(opt: config::Opt) -> Result<()> {
// init auto heal
init_auto_heal().await;
info!("server was started");
let srv_addr = format!("http://{}:{}", local_ip, server_port);
init_console_cfg(&srv_addr);
print_server_info();
if opt.console_enable {
info!("console is enabled");
debug!("console is enabled");
let access_key = opt.access_key.clone();
let secret_key = opt.secret_key.clone();
let console_address = opt.console_address.clone();
tokio::spawn(async move {
let ep = if !opt.server_domains.is_empty() {
format!("http://{}", opt.server_domains[0].clone())
} else {
format!("http://127.0.0.1:{}", server_port)
};
console::start_static_file_server(&opt.console_address, &ep).await;
console::start_static_file_server(&console_address, local_ip, &access_key, &secret_key).await;
});
}
rustfs/src/utils.rs
+10
View File
@@ -0,0 +1,10 @@
use local_ip_address;
use std::net::IpAddr;
pub(crate) fn get_local_ip() -> Option<std::net::Ipv4Addr> {
match local_ip_address::local_ip() {
Ok(IpAddr::V4(ip)) => Some(ip),
Err(_) => None,
Ok(IpAddr::V6(_)) => todo!(),
}
}