mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-03 10:48:13 +00:00
improve code
This commit is contained in:
@@ -424,21 +424,21 @@ impl EventNotifierConfig {
|
|||||||
let mut adapters = Vec::new();
|
let mut adapters = Vec::new();
|
||||||
|
|
||||||
// Add all enabled webhook configurations
|
// Add all enabled webhook configurations
|
||||||
for (_, webhook) in &self.webhook {
|
for webhook in self.webhook.values() {
|
||||||
if webhook.common.enable {
|
if webhook.common.enable {
|
||||||
adapters.push(AdapterConfig::Webhook(webhook.clone()));
|
adapters.push(AdapterConfig::Webhook(webhook.clone()));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Add all enabled Kafka configurations
|
// Add all enabled Kafka configurations
|
||||||
for (_, kafka) in &self.kafka {
|
for kafka in self.kafka.values() {
|
||||||
if kafka.common.enable {
|
if kafka.common.enable {
|
||||||
adapters.push(AdapterConfig::Kafka(kafka.clone()));
|
adapters.push(AdapterConfig::Kafka(kafka.clone()));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Add all enabled MQTT configurations
|
// Add all enabled MQTT configurations
|
||||||
for (_, mqtt) in &self.mqtt {
|
for mqtt in self.mqtt.values() {
|
||||||
if mqtt.common.enable {
|
if mqtt.common.enable {
|
||||||
adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
|
adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -6,7 +6,6 @@ mod event;
|
|||||||
mod global;
|
mod global;
|
||||||
mod notifier;
|
mod notifier;
|
||||||
mod store;
|
mod store;
|
||||||
mod target;
|
|
||||||
|
|
||||||
pub use adapter::create_adapters;
|
pub use adapter::create_adapters;
|
||||||
#[cfg(all(feature = "kafka", target_os = "linux"))]
|
#[cfg(all(feature = "kafka", target_os = "linux"))]
|
||||||
|
|||||||
@@ -25,18 +25,17 @@ pub static GLOBAL_EVENT_CONFIG: Lazy<Mutex<Option<EventNotifierConfig>>> = Lazy:
|
|||||||
|
|
||||||
/// EventManager Responsible for managing all operations of the event system
|
/// EventManager Responsible for managing all operations of the event system
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
pub struct EventManager<S: StorageAPI> {
|
pub struct EventManager {
|
||||||
api: Arc<S>,
|
api: Arc<ECStore>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl<S: StorageAPI> EventManager {
|
impl EventManager {
|
||||||
/// Create a new Event Manager
|
/// Create a new Event Manager
|
||||||
pub async fn new(api: Arc<S>) -> Self {
|
pub async fn new(api: Arc<ECStore>) -> Self {
|
||||||
// Update the global access point at the same time
|
// Set the global storage API
|
||||||
if let Ok(mut global_api) = GLOBAL_STORE_API.lock() {
|
{
|
||||||
if let Some(store) = api.as_any().downcast_ref::<ECStore>() {
|
let mut global_api = GLOBAL_STORE_API.lock().await;
|
||||||
*global_api = Some(Arc::new(store.clone()));
|
*global_api = Some(api.clone());
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
Self { api }
|
Self { api }
|
||||||
@@ -82,10 +81,7 @@ impl<S: StorageAPI> EventManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
save_event_config(self.api.clone(), cfg).await?;
|
save_event_config(self.api.clone(), cfg).await?;
|
||||||
*GLOBAL_EVENT_CONFIG
|
*GLOBAL_EVENT_CONFIG.lock().await = Some(cfg.clone());
|
||||||
.lock()
|
|
||||||
.await
|
|
||||||
.map_err(|e| Error::msg(format!("Failed to acquire global config lock: {}", e)))? = Some(cfg.clone());
|
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -199,7 +195,7 @@ async fn save_event_config<S: StorageAPI>(api: Arc<S>, config: &EventNotifierCon
|
|||||||
let config_file = get_event_config_file();
|
let config_file = get_event_config_file();
|
||||||
let data = config.marshal()?;
|
let data = config.marshal()?;
|
||||||
|
|
||||||
save_config(api, &config_file, data).await;
|
save_config(api, &config_file, data).await
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the event profile path
|
/// Get the event profile path
|
||||||
|
|||||||
@@ -15,7 +15,7 @@ use uuid::Uuid;
|
|||||||
pub struct Key {
|
pub struct Key {
|
||||||
/// Key name
|
/// Key name
|
||||||
pub name: String,
|
pub name: String,
|
||||||
/// Whether or not to compress
|
/// Whether to compress
|
||||||
pub compress: bool,
|
pub compress: bool,
|
||||||
/// filename extension
|
/// filename extension
|
||||||
pub extension: String,
|
pub extension: String,
|
||||||
@@ -35,6 +35,7 @@ impl Key {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Convert to string form
|
/// Convert to string form
|
||||||
|
#[allow(clippy::inherent_to_string)]
|
||||||
pub fn to_string(&self) -> String {
|
pub fn to_string(&self) -> String {
|
||||||
let mut key_str = self.name.clone();
|
let mut key_str = self.name.clone();
|
||||||
if self.item_count > 1 {
|
if self.item_count > 1 {
|
||||||
@@ -47,6 +48,7 @@ impl Key {
|
|||||||
}
|
}
|
||||||
|
|
||||||
/// Parse key from file name
|
/// Parse key from file name
|
||||||
|
#[allow(clippy::redundant_closure)]
|
||||||
pub fn parse_key(filename: &str) -> Key {
|
pub fn parse_key(filename: &str) -> Key {
|
||||||
let compress = filename.ends_with(".snappy");
|
let compress = filename.ends_with(".snappy");
|
||||||
let filename = if compress {
|
let filename = if compress {
|
||||||
|
|||||||
+1
-1
@@ -10,7 +10,7 @@ pub(crate) async fn init_event_notifier(notifier_config: Option<String>) {
|
|||||||
NotifierConfig::event_load_config(notifier_config)
|
NotifierConfig::event_load_config(notifier_config)
|
||||||
} else {
|
} else {
|
||||||
info!("event_config is empty");
|
info!("event_config is empty");
|
||||||
rustfs_event::get_event_notifier_config().clone()
|
// rustfs_event::get_event_notifier_config().clone()
|
||||||
};
|
};
|
||||||
|
|
||||||
info!("using event_config: {:?}", config);
|
info!("using event_config: {:?}", config);
|
||||||
|
|||||||
+1
-2
@@ -48,7 +48,6 @@ use iam::init_iam_sys;
|
|||||||
use license::init_license;
|
use license::init_license;
|
||||||
use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
|
use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
|
||||||
use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
|
use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
|
||||||
use rustfs_event::GLOBAL_EventSys;
|
|
||||||
use rustfs_obs::{init_obs, set_global_guard, SystemObserver};
|
use rustfs_obs::{init_obs, set_global_guard, SystemObserver};
|
||||||
use rustls::ServerConfig;
|
use rustls::ServerConfig;
|
||||||
use s3s::{host::MultiDomain, service::S3ServiceBuilder};
|
use s3s::{host::MultiDomain, service::S3ServiceBuilder};
|
||||||
@@ -511,7 +510,7 @@ async fn run(opt: config::Opt) -> Result<()> {
|
|||||||
GLOBAL_ConfigSys.init(store.clone()).await?;
|
GLOBAL_ConfigSys.init(store.clone()).await?;
|
||||||
|
|
||||||
// event system configuration
|
// event system configuration
|
||||||
GLOBAL_EventSys.init(store.clone()).await?;
|
// GLOBAL_EventSys.init(store.clone()).await?;
|
||||||
|
|
||||||
// Initialize event notifier
|
// Initialize event notifier
|
||||||
event::init_event_notifier(opt.event_config).await;
|
event::init_event_notifier(opt.event_config).await;
|
||||||
|
|||||||
@@ -26,9 +26,6 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
|
|||||||
|
|
||||||
if let Some(cred) = &req_info.cred {
|
if let Some(cred) = &req_info.cred {
|
||||||
let Ok(iam_store) = iam::get() else {
|
let Ok(iam_store) = iam::get() else {
|
||||||
let _api_rejected_auth_total_key = rustfs_obs::API_REJECTED_AUTH_TOTAL_MD.get_full_metric_name();
|
|
||||||
let desc = rustfs_obs::API_REJECTED_AUTH_TOTAL_MD.clone().help;
|
|
||||||
tracing::info!(api_rejected_auth_total_key = 1_u64, desc);
|
|
||||||
return Err(S3Error::with_message(
|
return Err(S3Error::with_message(
|
||||||
S3ErrorCode::InternalError,
|
S3ErrorCode::InternalError,
|
||||||
format!("authorize_request {:?}", IamError::IamSysNotInitialized),
|
format!("authorize_request {:?}", IamError::IamSysNotInitialized),
|
||||||
|
|||||||
Reference in New Issue
Block a user