gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-12 16:16:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

Feature/improve profiling (#1038)

Browse Source 
Co-authored-by: Jitter <jitterx69@gmail.com>
Co-authored-by: weisd <im@weisd.in>
This commit is contained in:
houseme
2025-12-07 22:39:47 +08:00
committed by GitHub
parent cd6a26bc3a
commit e2d8e9e3d3
9 changed files with 585 additions and 361 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+3 -3
View File
@@ -133,11 +133,11 @@ sysctl = { workspace = true }
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
[target.'cfg(all(target_os = "linux", target_env = "musl"))'.dependencies]
[target.'cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))'.dependencies]
mimalloc = { workspace = true }
[target.'cfg(all(target_os = "linux", target_env = "gnu"))'.dependencies]
[target.'cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))'.dependencies]
tikv-jemallocator = { workspace = true }
[target.'cfg(all(not(target_env = "msvc"), not(target_os = "windows")))'.dependencies]
tikv-jemalloc-ctl = { workspace = true }
jemalloc_pprof = { workspace = true }
pprof = { workspace = true }
rustfs/src/admin/handlers.rs
+17 -10
View File
@@ -1276,15 +1276,20 @@ pub struct ProfileHandler {}
#[async_trait::async_trait]
impl Operation for ProfileHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
#[cfg(target_os = "windows")]
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
{
return Ok(S3Response::new((
StatusCode::NOT_IMPLEMENTED,
Body::from("CPU profiling is not supported on Windows platform".to_string()),
)));
let requested_url = req.uri.to_string();
let target_os = std::env::consts::OS;
let target_arch = std::env::consts::ARCH;
let target_env = option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown");
let msg = format!(
"CPU profiling is not supported on this platform. target_os={}, target_env={}, target_arch={}, requested_url={}",
target_os, target_env, target_arch, requested_url
);
return Ok(S3Response::new((StatusCode::NOT_IMPLEMENTED, Body::from(msg))));
}
#[cfg(not(target_os = "windows"))]
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
{
use rustfs_config::{DEFAULT_CPU_FREQ, ENV_CPU_FREQ};
use rustfs_utils::get_env_usize;
@@ -1369,15 +1374,17 @@ impl Operation for ProfileStatusHandler {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
use std::collections::HashMap;
#[cfg(target_os = "windows")]
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
let message = format!("CPU profiling is not supported on {} platform", std::env::consts::OS);
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
let status = HashMap::from([
("enabled", "false"),
("status", "not_supported"),
("platform", "windows"),
("message", "CPU profiling is not supported on Windows platform"),
("platform", std::env::consts::OS),
("message", message.as_str()),
]);
#[cfg(not(target_os = "windows"))]
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
let status = {
use rustfs_config::{DEFAULT_ENABLE_PROFILING, ENV_ENABLE_PROFILING};
use rustfs_utils::get_env_bool;
rustfs/src/admin/handlers/profile.rs
+13 -43
View File
@@ -24,30 +24,15 @@ pub struct TriggerProfileCPU {}
impl Operation for TriggerProfileCPU {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
info!("Triggering CPU profile dump via S3 request...");
#[cfg(target_os = "windows")]
{
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/plain".parse().unwrap());
return Ok(S3Response::with_headers(
(
StatusCode::NOT_IMPLEMENTED,
Body::from("CPU profiling is not supported on Windows".to_string()),
),
header,
));
}
#[cfg(not(target_os = "windows"))]
{
let dur = std::time::Duration::from_secs(60);
match crate::profiling::dump_cpu_pprof_for(dur).await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().unwrap());
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump CPU profile: {e}"))),
let dur = std::time::Duration::from_secs(60);
match crate::profiling::dump_cpu_pprof_for(dur).await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().unwrap());
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump CPU profile: {e}"))),
}
}
}
@@ -57,29 +42,14 @@ pub struct TriggerProfileMemory {}
impl Operation for TriggerProfileMemory {
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
info!("Triggering Memory profile dump via S3 request...");
#[cfg(target_os = "windows")]
{
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/plain".parse().unwrap());
return Ok(S3Response::with_headers(
(
StatusCode::NOT_IMPLEMENTED,
Body::from("Memory profiling is not supported on Windows".to_string()),
),
header,
));
}
#[cfg(not(target_os = "windows"))]
{
match crate::profiling::dump_memory_pprof_now().await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().unwrap());
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump Memory profile: {e}"))),
match crate::profiling::dump_memory_pprof_now().await {
Ok(path) => {
let mut header = HeaderMap::new();
header.insert(CONTENT_TYPE, "text/html".parse().unwrap());
Ok(S3Response::with_headers((StatusCode::OK, Body::from(path.display().to_string())), header))
}
Err(e) => Err(s3s::s3_error!(InternalError, "{}", format!("Failed to dump Memory profile: {e}"))),
}
}
}
rustfs/src/init.rs
+280
View File
@@ -0,0 +1,280 @@
// 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.
use crate::storage::ecfs::{process_lambda_configurations, process_queue_configurations, process_topic_configurations};
use crate::{admin, config};
use rustfs_config::{DEFAULT_UPDATE_CHECK, ENV_UPDATE_CHECK};
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_notify::notifier_global;
use rustfs_targets::arn::{ARN, TargetIDError};
use s3s::s3_error;
use std::env;
use std::io::Error;
use tracing::{debug, error, info, instrument, warn};
pub(crate) fn init_update_check() {
let update_check_enable = env::var(ENV_UPDATE_CHECK)
.unwrap_or_else(|_| DEFAULT_UPDATE_CHECK.to_string())
.parse::<bool>()
.unwrap_or(DEFAULT_UPDATE_CHECK);
if !update_check_enable {
return;
}
// Async update check with timeout
tokio::spawn(async {
use crate::update::{UpdateCheckError, check_updates};
// Add timeout to prevent hanging network calls
match tokio::time::timeout(std::time::Duration::from_secs(30), check_updates()).await {
Ok(Ok(result)) => {
if result.update_available {
if let Some(latest) = &result.latest_version {
info!(
"🚀 Version check: New version available: {} -> {} (current: {})",
result.current_version, latest.version, result.current_version
);
if let Some(notes) = &latest.release_notes {
info!("📝 Release notes: {}", notes);
}
if let Some(url) = &latest.download_url {
info!("🔗 Download URL: {}", url);
}
}
} else {
debug!("✅ Version check: Current version is up to date: {}", result.current_version);
}
}
Ok(Err(UpdateCheckError::HttpError(e))) => {
debug!("Version check: network error (this is normal): {}", e);
}
Ok(Err(e)) => {
debug!("Version check: failed (this is normal): {}", e);
}
Err(_) => {
debug!("Version check: timeout after 30 seconds (this is normal)");
}
}
});
}
#[instrument(skip_all)]
pub(crate) async fn add_bucket_notification_configuration(buckets: Vec<String>) {
let region_opt = rustfs_ecstore::global::get_global_region();
let region = match region_opt {
Some(ref r) if !r.is_empty() => r,
_ => {
warn!("Global region is not set; attempting notification configuration for all buckets with an empty region.");
""
}
};
for bucket in buckets.iter() {
let has_notification_config = metadata_sys::get_notification_config(bucket).await.unwrap_or_else(|err| {
warn!("get_notification_config err {:?}", err);
None
});
match has_notification_config {
Some(cfg) => {
info!(
target: "rustfs::main::add_bucket_notification_configuration",
bucket = %bucket,
"Bucket '{}' has existing notification configuration: {:?}", bucket, cfg);
let mut event_rules = Vec::new();
process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
if let Err(e) = notifier_global::add_event_specific_rules(bucket, region, &event_rules)
.await
.map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))
{
error!("Failed to add rules for bucket '{}': {:?}", bucket, e);
}
}
None => {
info!(
target: "rustfs::main::add_bucket_notification_configuration",
bucket = %bucket,
"Bucket '{}' has no existing notification configuration.", bucket);
}
}
}
}
/// Initialize KMS system and configure if enabled
#[instrument(skip(opt))]
pub(crate) async fn init_kms_system(opt: &config::Opt) -> std::io::Result<()> {
// Initialize global KMS service manager (starts in NotConfigured state)
let service_manager = rustfs_kms::init_global_kms_service_manager();
// If KMS is enabled in configuration, configure and start the service
if opt.kms_enable {
info!("KMS is enabled via command line, configuring and starting service...");
// Create KMS configuration from command line options
let kms_config = match opt.kms_backend.as_str() {
"local" => {
let key_dir = opt
.kms_key_dir
.as_ref()
.ok_or_else(|| Error::other("KMS key directory is required for local backend"))?;
rustfs_kms::config::KmsConfig {
backend: rustfs_kms::config::KmsBackend::Local,
backend_config: rustfs_kms::config::BackendConfig::Local(rustfs_kms::config::LocalConfig {
key_dir: std::path::PathBuf::from(key_dir),
master_key: None,
file_permissions: Some(0o600),
}),
default_key_id: opt.kms_default_key_id.clone(),
timeout: std::time::Duration::from_secs(30),
retry_attempts: 3,
enable_cache: true,
cache_config: rustfs_kms::config::CacheConfig::default(),
}
}
"vault" => {
let vault_address = opt
.kms_vault_address
.as_ref()
.ok_or_else(|| Error::other("Vault address is required for vault backend"))?;
let vault_token = opt
.kms_vault_token
.as_ref()
.ok_or_else(|| Error::other("Vault token is required for vault backend"))?;
rustfs_kms::config::KmsConfig {
backend: rustfs_kms::config::KmsBackend::Vault,
backend_config: rustfs_kms::config::BackendConfig::Vault(rustfs_kms::config::VaultConfig {
address: vault_address.clone(),
auth_method: rustfs_kms::config::VaultAuthMethod::Token {
token: vault_token.clone(),
},
namespace: None,
mount_path: "transit".to_string(),
kv_mount: "secret".to_string(),
key_path_prefix: "rustfs/kms/keys".to_string(),
tls: None,
}),
default_key_id: opt.kms_default_key_id.clone(),
timeout: std::time::Duration::from_secs(30),
retry_attempts: 3,
enable_cache: true,
cache_config: rustfs_kms::config::CacheConfig::default(),
}
}
_ => return Err(Error::other(format!("Unsupported KMS backend: {}", opt.kms_backend))),
};
// Configure the KMS service
service_manager
.configure(kms_config)
.await
.map_err(|e| Error::other(format!("Failed to configure KMS: {e}")))?;
// Start the KMS service
service_manager
.start()
.await
.map_err(|e| Error::other(format!("Failed to start KMS: {e}")))?;
info!("KMS service configured and started successfully from command line options");
} else {
// Try to load persisted KMS configuration from cluster storage
info!("Attempting to load persisted KMS configuration from cluster storage...");
if let Some(persisted_config) = admin::handlers::kms_dynamic::load_kms_config().await {
info!("Found persisted KMS configuration, attempting to configure and start service...");
// Configure the KMS service with persisted config
match service_manager.configure(persisted_config).await {
Ok(()) => {
// Start the KMS service
match service_manager.start().await {
Ok(()) => {
info!("KMS service configured and started successfully from persisted configuration");
}
Err(e) => {
warn!("Failed to start KMS with persisted configuration: {}", e);
}
}
}
Err(e) => {
warn!("Failed to configure KMS with persisted configuration: {}", e);
}
}
} else {
info!("No persisted KMS configuration found. KMS is ready for dynamic configuration via API.");
}
}
Ok(())
}
/// Initialize the adaptive buffer sizing system with workload profile configuration.
///
/// This system provides intelligent buffer size selection based on file size and workload type.
/// Workload-aware buffer sizing is enabled by default with the GeneralPurpose profile,
/// which provides the same buffer sizes as the original implementation for compatibility.
///
/// # Configuration
/// - Default: Enabled with GeneralPurpose profile
/// - Opt-out: Use `--buffer-profile-disable` flag
/// - Custom profile: Set via `--buffer-profile` or `RUSTFS_BUFFER_PROFILE` environment variable
///
/// # Arguments
/// * `opt` - The application configuration options
pub(crate) fn init_buffer_profile_system(opt: &config::Opt) {
use crate::config::workload_profiles::{
RustFSBufferConfig, WorkloadProfile, init_global_buffer_config, set_buffer_profile_enabled,
};
if opt.buffer_profile_disable {
// User explicitly disabled buffer profiling - use GeneralPurpose profile in disabled mode
info!("Buffer profiling disabled via --buffer-profile-disable, using GeneralPurpose profile");
set_buffer_profile_enabled(false);
} else {
// Enabled by default: use configured workload profile
info!("Buffer profiling enabled with profile: {}", opt.buffer_profile);
// Parse the workload profile from configuration string
let profile = WorkloadProfile::from_name(&opt.buffer_profile);
// Log the selected profile for operational visibility
info!("Active buffer profile: {:?}", profile);
// Initialize the global buffer configuration
init_global_buffer_config(RustFSBufferConfig::new(profile));
// Enable buffer profiling globally
set_buffer_profile_enabled(true);
info!("Buffer profiling system initialized successfully");
}
}
rustfs/src/main.rs
+8 -283
View File
@@ -17,8 +17,8 @@ mod auth;
mod config;
mod error;
// mod grpc;
mod init;
pub mod license;
#[cfg(not(target_os = "windows"))]
mod profiling;
mod server;
mod storage;
@@ -26,11 +26,11 @@ mod update;
mod version;
// Ensure the correct path for parse_license is imported
use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system, init_update_check};
use crate::server::{
SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, init_event_notifier, shutdown_event_notifier,
start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown,
};
use crate::storage::ecfs::{process_lambda_configurations, process_queue_configurations, process_topic_configurations};
use chrono::Datelike;
use clap::Parser;
use license::init_license;
@@ -39,9 +39,6 @@ use rustfs_ahm::{
scanner::data_scanner::ScannerConfig, shutdown_ahm_services,
};
use rustfs_common::globals::set_global_addr;
use rustfs_config::DEFAULT_UPDATE_CHECK;
use rustfs_config::ENV_UPDATE_CHECK;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys;
use rustfs_ecstore::bucket::replication::{GLOBAL_REPLICATION_POOL, init_background_replication};
use rustfs_ecstore::config as ecconfig;
@@ -58,22 +55,18 @@ use rustfs_ecstore::{
update_erasure_type,
};
use rustfs_iam::init_iam_sys;
use rustfs_notify::notifier_global;
use rustfs_obs::{init_obs, set_global_guard};
use rustfs_targets::arn::{ARN, TargetIDError};
use rustfs_utils::net::parse_and_resolve_address;
use s3s::s3_error;
use std::env;
use std::io::{Error, Result};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};
#[cfg(all(target_os = "linux", target_env = "gnu"))]
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
#[global_allocator]
static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
#[cfg(all(target_os = "linux", target_env = "musl"))]
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
#[global_allocator]
static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
@@ -131,7 +124,6 @@ async fn async_main() -> Result<()> {
info!("{}", LOGO);
// Initialize performance profiling if enabled
#[cfg(not(target_os = "windows"))]
profiling::init_from_env().await;
// Run parameters
@@ -297,8 +289,8 @@ async fn run(opt: config::Opt) -> Result<()> {
let _ = create_ahm_services_cancel_token();
// Check environment variables to determine if scanner and heal should be enabled
let enable_scanner = parse_bool_env_var("RUSTFS_ENABLE_SCANNER", true);
let enable_heal = parse_bool_env_var("RUSTFS_ENABLE_HEAL", true);
let enable_scanner = rustfs_utils::get_env_bool("RUSTFS_ENABLE_SCANNER", true);
let enable_heal = rustfs_utils::get_env_bool("RUSTFS_ENABLE_HEAL", true);
info!(
target: "rustfs::main::run",
@@ -353,17 +345,6 @@ async fn run(opt: config::Opt) -> Result<()> {
Ok(())
}
/// Parse a boolean environment variable with default value
///
/// Returns true if the environment variable is not set or set to true/1/yes/on/enabled,
/// false if set to false/0/no/off/disabled
fn parse_bool_env_var(var_name: &str, default: bool) -> bool {
env::var(var_name)
.unwrap_or_else(|_| default.to_string())
.parse::<bool>()
.unwrap_or(default)
}
/// Handles the shutdown process of the server
async fn handle_shutdown(
state_manager: &ServiceStateManager,
@@ -381,8 +362,8 @@ async fn handle_shutdown(
state_manager.update(ServiceState::Stopping);
// Check environment variables to determine what services need to be stopped
let enable_scanner = parse_bool_env_var("RUSTFS_ENABLE_SCANNER", true);
let enable_heal = parse_bool_env_var("RUSTFS_ENABLE_HEAL", true);
let enable_scanner = rustfs_utils::get_env_bool("RUSTFS_ENABLE_SCANNER", true);
let enable_heal = rustfs_utils::get_env_bool("RUSTFS_ENABLE_HEAL", true);
// Stop background services based on what was enabled
if enable_scanner || enable_heal {
@@ -443,259 +424,3 @@ async fn handle_shutdown(
);
println!("Server stopped successfully.");
}
fn init_update_check() {
let update_check_enable = env::var(ENV_UPDATE_CHECK)
.unwrap_or_else(|_| DEFAULT_UPDATE_CHECK.to_string())
.parse::<bool>()
.unwrap_or(DEFAULT_UPDATE_CHECK);
if !update_check_enable {
return;
}
// Async update check with timeout
tokio::spawn(async {
use crate::update::{UpdateCheckError, check_updates};
// Add timeout to prevent hanging network calls
match tokio::time::timeout(std::time::Duration::from_secs(30), check_updates()).await {
Ok(Ok(result)) => {
if result.update_available {
if let Some(latest) = &result.latest_version {
info!(
"🚀 Version check: New version available: {} -> {} (current: {})",
result.current_version, latest.version, result.current_version
);
if let Some(notes) = &latest.release_notes {
info!("📝 Release notes: {}", notes);
}
if let Some(url) = &latest.download_url {
info!("🔗 Download URL: {}", url);
}
}
} else {
debug!("✅ Version check: Current version is up to date: {}", result.current_version);
}
}
Ok(Err(UpdateCheckError::HttpError(e))) => {
debug!("Version check: network error (this is normal): {}", e);
}
Ok(Err(e)) => {
debug!("Version check: failed (this is normal): {}", e);
}
Err(_) => {
debug!("Version check: timeout after 30 seconds (this is normal)");
}
}
});
}
#[instrument(skip_all)]
async fn add_bucket_notification_configuration(buckets: Vec<String>) {
let region_opt = rustfs_ecstore::global::get_global_region();
let region = match region_opt {
Some(ref r) if !r.is_empty() => r,
_ => {
warn!("Global region is not set; attempting notification configuration for all buckets with an empty region.");
""
}
};
for bucket in buckets.iter() {
let has_notification_config = metadata_sys::get_notification_config(bucket).await.unwrap_or_else(|err| {
warn!("get_notification_config err {:?}", err);
None
});
match has_notification_config {
Some(cfg) => {
info!(
target: "rustfs::main::add_bucket_notification_configuration",
bucket = %bucket,
"Bucket '{}' has existing notification configuration: {:?}", bucket, cfg);
let mut event_rules = Vec::new();
process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), |arn_str| {
ARN::parse(arn_str)
.map(|arn| arn.target_id)
.map_err(|e| TargetIDError::InvalidFormat(e.to_string()))
});
if let Err(e) = notifier_global::add_event_specific_rules(bucket, region, &event_rules)
.await
.map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))
{
error!("Failed to add rules for bucket '{}': {:?}", bucket, e);
}
}
None => {
info!(
target: "rustfs::main::add_bucket_notification_configuration",
bucket = %bucket,
"Bucket '{}' has no existing notification configuration.", bucket);
}
}
}
}
/// Initialize KMS system and configure if enabled
#[instrument(skip(opt))]
async fn init_kms_system(opt: &config::Opt) -> Result<()> {
// Initialize global KMS service manager (starts in NotConfigured state)
let service_manager = rustfs_kms::init_global_kms_service_manager();
// If KMS is enabled in configuration, configure and start the service
if opt.kms_enable {
info!("KMS is enabled via command line, configuring and starting service...");
// Create KMS configuration from command line options
let kms_config = match opt.kms_backend.as_str() {
"local" => {
let key_dir = opt
.kms_key_dir
.as_ref()
.ok_or_else(|| Error::other("KMS key directory is required for local backend"))?;
rustfs_kms::config::KmsConfig {
backend: rustfs_kms::config::KmsBackend::Local,
backend_config: rustfs_kms::config::BackendConfig::Local(rustfs_kms::config::LocalConfig {
key_dir: std::path::PathBuf::from(key_dir),
master_key: None,
file_permissions: Some(0o600),
}),
default_key_id: opt.kms_default_key_id.clone(),
timeout: std::time::Duration::from_secs(30),
retry_attempts: 3,
enable_cache: true,
cache_config: rustfs_kms::config::CacheConfig::default(),
}
}
"vault" => {
let vault_address = opt
.kms_vault_address
.as_ref()
.ok_or_else(|| Error::other("Vault address is required for vault backend"))?;
let vault_token = opt
.kms_vault_token
.as_ref()
.ok_or_else(|| Error::other("Vault token is required for vault backend"))?;
rustfs_kms::config::KmsConfig {
backend: rustfs_kms::config::KmsBackend::Vault,
backend_config: rustfs_kms::config::BackendConfig::Vault(rustfs_kms::config::VaultConfig {
address: vault_address.clone(),
auth_method: rustfs_kms::config::VaultAuthMethod::Token {
token: vault_token.clone(),
},
namespace: None,
mount_path: "transit".to_string(),
kv_mount: "secret".to_string(),
key_path_prefix: "rustfs/kms/keys".to_string(),
tls: None,
}),
default_key_id: opt.kms_default_key_id.clone(),
timeout: std::time::Duration::from_secs(30),
retry_attempts: 3,
enable_cache: true,
cache_config: rustfs_kms::config::CacheConfig::default(),
}
}
_ => return Err(Error::other(format!("Unsupported KMS backend: {}", opt.kms_backend))),
};
// Configure the KMS service
service_manager
.configure(kms_config)
.await
.map_err(|e| Error::other(format!("Failed to configure KMS: {e}")))?;
// Start the KMS service
service_manager
.start()
.await
.map_err(|e| Error::other(format!("Failed to start KMS: {e}")))?;
info!("KMS service configured and started successfully from command line options");
} else {
// Try to load persisted KMS configuration from cluster storage
info!("Attempting to load persisted KMS configuration from cluster storage...");
if let Some(persisted_config) = admin::handlers::kms_dynamic::load_kms_config().await {
info!("Found persisted KMS configuration, attempting to configure and start service...");
// Configure the KMS service with persisted config
match service_manager.configure(persisted_config).await {
Ok(()) => {
// Start the KMS service
match service_manager.start().await {
Ok(()) => {
info!("KMS service configured and started successfully from persisted configuration");
}
Err(e) => {
warn!("Failed to start KMS with persisted configuration: {}", e);
}
}
}
Err(e) => {
warn!("Failed to configure KMS with persisted configuration: {}", e);
}
}
} else {
info!("No persisted KMS configuration found. KMS is ready for dynamic configuration via API.");
}
}
Ok(())
}
/// Initialize the adaptive buffer sizing system with workload profile configuration.
///
/// This system provides intelligent buffer size selection based on file size and workload type.
/// Workload-aware buffer sizing is enabled by default with the GeneralPurpose profile,
/// which provides the same buffer sizes as the original implementation for compatibility.
///
/// # Configuration
/// - Default: Enabled with GeneralPurpose profile
/// - Opt-out: Use `--buffer-profile-disable` flag
/// - Custom profile: Set via `--buffer-profile` or `RUSTFS_BUFFER_PROFILE` environment variable
///
/// # Arguments
/// * `opt` - The application configuration options
fn init_buffer_profile_system(opt: &config::Opt) {
use crate::config::workload_profiles::{
RustFSBufferConfig, WorkloadProfile, init_global_buffer_config, set_buffer_profile_enabled,
};
if opt.buffer_profile_disable {
// User explicitly disabled buffer profiling - use GeneralPurpose profile in disabled mode
info!("Buffer profiling disabled via --buffer-profile-disable, using GeneralPurpose profile");
set_buffer_profile_enabled(false);
} else {
// Enabled by default: use configured workload profile
info!("Buffer profiling enabled with profile: {}", opt.buffer_profile);
// Parse the workload profile from configuration string
let profile = WorkloadProfile::from_name(&opt.buffer_profile);
// Log the selected profile for operational visibility
info!("Active buffer profile: {:?}", profile);
// Initialize the global buffer configuration
init_global_buffer_config(RustFSBufferConfig::new(profile));
// Enable buffer profiling globally
set_buffer_profile_enabled(true);
info!("Buffer profiling system initialized successfully");
}
}
rustfs/src/profiling.rs
+37 -8
View File
@@ -12,20 +12,49 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(not(target_os = "linux"))]
pub async fn init_from_env() {}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub async fn init_from_env() {
let (target_os, target_env, target_arch) = get_platform_info();
tracing::info!(
target: "rustfs::main::run",
target_os = %target_os,
target_env = %target_env,
target_arch = %target_arch,
"profiling: disabled on this platform. target_os={}, target_env={}, target_arch={}",
target_os, target_env, target_arch
);
}
#[cfg(not(target_os = "linux"))]
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
fn get_platform_info() -> (String, String, String) {
(
std::env::consts::OS.to_string(),
option_env!("CARGO_CFG_TARGET_ENV").unwrap_or("unknown").to_string(),
std::env::consts::ARCH.to_string(),
)
}
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub async fn dump_cpu_pprof_for(_duration: std::time::Duration) -> Result<std::path::PathBuf, String> {
Err("CPU profiling is only supported on Linux".to_string())
let (target_os, target_env, target_arch) = get_platform_info();
let msg = format!(
"CPU profiling is not supported on this platform. target_os={}, target_env={}, target_arch={}",
target_os, target_env, target_arch
);
Err(msg)
}
#[cfg(not(target_os = "linux"))]
#[cfg(not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")))]
pub async fn dump_memory_pprof_now() -> Result<std::path::PathBuf, String> {
Err("Memory profiling is only supported on Linux".to_string())
let (target_os, target_env, target_arch) = get_platform_info();
let msg = format!(
"Memory profiling is not supported on this platform. target_os={}, target_env={}, target_arch={}",
target_os, target_env, target_arch
);
Err(msg)
}
#[cfg(target_os = "linux")]
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
mod linux_impl {
use chrono::Utc;
use jemalloc_pprof::PROF_CTL;
@@ -298,5 +327,5 @@ mod linux_impl {
}
}
#[cfg(target_os = "linux")]
#[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))]
pub use linux_impl::{dump_cpu_pprof_for, dump_memory_pprof_now, init_from_env};