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

fix

Browse Source
This commit is contained in:
houseme
2025-06-04 23:46:14 +08:00
parent 703c465bd0
commit 8555d48ca2
23 changed files with 717 additions and 664 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/event/src/adapter/kafka.rs
+3 -4
View File
@@ -1,8 +1,7 @@
use crate::config::{default_queue_limit, STORE_PREFIX}; use crate::config::kafka::KafkaConfig;
use crate::KafkaConfig; use crate::config::{default_queue_limit, DEFAULT_RETRY_INTERVAL, STORE_PREFIX};
use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{ChannelAdapter, ChannelAdapterType};
use crate::{Error, QueueStore}; use crate::{Error, Event, QueueStore};
use crate::{Event, DEFAULT_RETRY_INTERVAL};
use async_trait::async_trait; use async_trait::async_trait;
use rdkafka::error::KafkaError; use rdkafka::error::KafkaError;
use rdkafka::producer::{FutureProducer, FutureRecord}; use rdkafka::producer::{FutureProducer, FutureRecord};
crates/event/src/adapter/mod.rs
+23 -4
View File
@@ -1,6 +1,5 @@
use crate::AdapterConfig; use crate::config::adapter::AdapterConfig;
use crate::Error; use crate::{Error, Event};
use crate::Event;
use async_trait::async_trait; use async_trait::async_trait;
use std::sync::Arc; use std::sync::Arc;
@@ -11,6 +10,26 @@ pub(crate) mod mqtt;
#[cfg(feature = "webhook")] #[cfg(feature = "webhook")]
pub(crate) mod webhook; pub(crate) mod webhook;
#[allow(dead_code)]
const NOTIFY_KAFKA_SUB_SYS: &str = "notify_kafka";
#[allow(dead_code)]
const NOTIFY_MQTT_SUB_SYS: &str = "notify_mqtt";
#[allow(dead_code)]
const NOTIFY_MY_SQL_SUB_SYS: &str = "notify_mysql";
#[allow(dead_code)]
const NOTIFY_NATS_SUB_SYS: &str = "notify_nats";
#[allow(dead_code)]
const NOTIFY_NSQ_SUB_SYS: &str = "notify_nsq";
#[allow(dead_code)]
const NOTIFY_ES_SUB_SYS: &str = "notify_elasticsearch";
#[allow(dead_code)]
const NOTIFY_AMQP_SUB_SYS: &str = "notify_amqp";
#[allow(dead_code)]
const NOTIFY_POSTGRES_SUB_SYS: &str = "notify_postgres";
#[allow(dead_code)]
const NOTIFY_REDIS_SUB_SYS: &str = "notify_redis";
const NOTIFY_WEBHOOK_SUB_SYS: &str = "notify_webhook";
/// The `ChannelAdapterType` enum represents the different types of channel adapters. /// The `ChannelAdapterType` enum represents the different types of channel adapters.
/// ///
/// It is used to identify the type of adapter being used in the system. /// It is used to identify the type of adapter being used in the system.
@@ -68,7 +87,7 @@ pub trait ChannelAdapter: Send + Sync + 'static {
} }
/// Creates channel adapters based on the provided configuration. /// Creates channel adapters based on the provided configuration.
pub fn create_adapters(configs: &[AdapterConfig]) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> { pub fn create_adapters(configs: Vec<AdapterConfig>) -> Result<Vec<Arc<dyn ChannelAdapter>>, Error> {
let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new(); let mut adapters: Vec<Arc<dyn ChannelAdapter>> = Vec::new();
for config in configs { for config in configs {
crates/event/src/adapter/mqtt.rs
+2 -3
View File
@@ -1,7 +1,6 @@
use crate::Error; use crate::config::mqtt::MqttConfig;
use crate::Event;
use crate::MqttConfig;
use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{ChannelAdapter, ChannelAdapterType};
use crate::{Error, Event};
use async_trait::async_trait; use async_trait::async_trait;
use rumqttc::{AsyncClient, MqttOptions, QoS}; use rumqttc::{AsyncClient, MqttOptions, QoS};
use std::time::Duration; use std::time::Duration;
crates/event/src/adapter/webhook.rs
+17 -1
View File
@@ -1,6 +1,6 @@
use crate::config::webhook::WebhookConfig;
use crate::config::STORE_PREFIX; use crate::config::STORE_PREFIX;
use crate::store::queue::Store; use crate::store::queue::Store;
use crate::WebhookConfig;
use crate::{ChannelAdapter, ChannelAdapterType}; use crate::{ChannelAdapter, ChannelAdapterType};
use crate::{Error, QueueStore}; use crate::{Error, QueueStore};
use crate::{Event, DEFAULT_RETRY_INTERVAL}; use crate::{Event, DEFAULT_RETRY_INTERVAL};
@@ -14,6 +14,22 @@ use std::time::Duration;
use tokio::time::sleep; use tokio::time::sleep;
use ChannelAdapterType::Webhook; use ChannelAdapterType::Webhook;
// Webhook constants
pub const WEBHOOK_ENDPOINT: &str = "endpoint";
pub const WEBHOOK_AUTH_TOKEN: &str = "auth_token";
pub const WEBHOOK_QUEUE_DIR: &str = "queue_dir";
pub const WEBHOOK_QUEUE_LIMIT: &str = "queue_limit";
pub const WEBHOOK_CLIENT_CERT: &str = "client_cert";
pub const WEBHOOK_CLIENT_KEY: &str = "client_key";
pub const ENV_WEBHOOK_ENABLE: &str = "RUSTFS_NOTIFY_WEBHOOK_ENABLE";
pub const ENV_WEBHOOK_ENDPOINT: &str = "RUSTFS_NOTIFY_WEBHOOK_ENDPOINT";
pub const ENV_WEBHOOK_AUTH_TOKEN: &str = "RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN";
pub const ENV_WEBHOOK_QUEUE_DIR: &str = "RUSTFS_NOTIFY_WEBHOOK_QUEUE_DIR";
pub const ENV_WEBHOOK_QUEUE_LIMIT: &str = "RUSTFS_NOTIFY_WEBHOOK_QUEUE_LIMIT";
pub const ENV_WEBHOOK_CLIENT_CERT: &str = "RUSTFS_NOTIFY_WEBHOOK_CLIENT_CERT";
pub const ENV_WEBHOOK_CLIENT_KEY: &str = "RUSTFS_NOTIFY_WEBHOOK_CLIENT_KEY";
/// Webhook adapter for sending events to a webhook endpoint. /// Webhook adapter for sending events to a webhook endpoint.
pub struct WebhookAdapter { pub struct WebhookAdapter {
/// Configuration information /// Configuration information
crates/event/src/bus.rs
-92
View File
@@ -1,92 +0,0 @@
use crate::ChannelAdapter;
use crate::Error;
use crate::EventStore;
use crate::{Event, Log};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::mpsc;
use tokio::time::Duration;
use tokio_util::sync::CancellationToken;
use tracing::instrument;
/// Handles incoming events from the producer.
///
/// This function is responsible for receiving events from the producer and sending them to the appropriate adapters.
/// It also handles the shutdown process and saves any pending logs to the event store.
#[instrument(skip_all)]
pub async fn event_bus(
mut rx: mpsc::Receiver<Event>,
adapters: Vec<Arc<dyn ChannelAdapter>>,
store: Arc<EventStore>,
shutdown: CancellationToken,
shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
) -> Result<(), Error> {
let mut unprocessed_events = Vec::new();
loop {
tokio::select! {
Some(event) = rx.recv() => {
let mut send_tasks = Vec::new();
for adapter in &adapters {
if event.channels.contains(&adapter.name()) {
let adapter = adapter.clone();
let event = event.clone();
send_tasks.push(tokio::spawn(async move {
if let Err(e) = adapter.send(&event).await {
tracing::error!("Failed to send event to {}: {}", adapter.name(), e);
Err(e)
} else {
Ok(())
}
}));
}
}
for task in send_tasks {
if task.await?.is_err() {
// If sending fails, add the event to the unprocessed list
let failed_event = event.clone();
unprocessed_events.push(failed_event);
}
}
}
_ = shutdown.cancelled() => {
tracing::info!("Shutting down event bus, saving pending logs...");
// Check if there are still unprocessed messages in the channel
while let Ok(Some(event)) = tokio::time::timeout(
Duration::from_millis(100),
rx.recv()
).await {
unprocessed_events.push(event);
}
// save only if there are unprocessed events
if !unprocessed_events.is_empty() {
tracing::info!("Save {} unhandled events", unprocessed_events.len());
// create and save logging
let shutdown_log = Log {
event_name: crate::event::Name::Everything,
key: format!("shutdown_{}", SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()),
records: unprocessed_events,
};
store.save_logs(&[shutdown_log]).await?;
} else {
tracing::info!("no unhandled events need to be saved");
}
tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some());
if let Some(complete_sender) = shutdown_complete {
// send a completion signal
let result = complete_sender.send(());
match result {
Ok(_) => tracing::info!("Event bus shutdown signal sent"),
Err(e) => tracing::error!("Failed to send event bus shutdown signal: {:?}", e),
}
tracing::info!("Shutting down event bus");
}
tracing::info!("Event bus shutdown complete");
break;
}
}
}
Ok(())
}
crates/event/src/config.rs
-449
View File
@@ -1,449 +0,0 @@
use config::{Config, File, FileFormat};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::env;
use std::path::Path;
use tracing::info;
/// The default configuration file name
const DEFAULT_CONFIG_FILE: &str = "event";
/// The prefix for the configuration file
pub const STORE_PREFIX: &str = "rustfs";
/// The default retry interval for the webhook adapter
pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
/// The default maximum retry count for the webhook adapter
pub const DEFAULT_MAX_RETRIES: u32 = 3;
/// Add a common field for the adapter configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AdapterCommon {
/// Adapter identifier for unique identification
pub identifier: String,
/// Adapter description information
pub comment: String,
/// Whether to enable this adapter
#[serde(default)]
pub enable: bool,
#[serde(default = "default_queue_dir")]
pub queue_dir: String,
#[serde(default = "default_queue_limit")]
pub queue_limit: u64,
}
impl Default for AdapterCommon {
fn default() -> Self {
Self {
identifier: String::new(),
comment: String::new(),
enable: false,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
}
}
}
/// Configuration for the webhook adapter.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct WebhookConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub endpoint: String,
pub auth_token: Option<String>,
pub custom_headers: Option<HashMap<String, String>>,
pub max_retries: u32,
pub retry_interval: Option<u64>,
pub timeout: Option<u64>,
#[serde(default)]
pub client_cert: Option<String>,
#[serde(default)]
pub client_key: Option<String>,
}
impl WebhookConfig {
/// validate the configuration for the webhook adapter
///
/// # Returns
///
/// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid.
///
/// # Example
///
/// ```
/// use rustfs_event::WebhookConfig;
/// use rustfs_event::AdapterCommon;
/// use rustfs_event::DEFAULT_RETRY_INTERVAL;
///
/// let config = WebhookConfig {
/// common: AdapterCommon::default(),
/// endpoint: "https://example.com/webhook".to_string(),
/// auth_token: Some("my_token".to_string()),
/// custom_headers: None,
/// max_retries: 3,
/// retry_interval: Some(DEFAULT_RETRY_INTERVAL),
/// timeout: Some(5000),
/// client_cert: None,
/// client_key: None,
/// };
///
/// assert!(config.validate().is_ok());
pub fn validate(&self) -> Result<(), String> {
// If not enabled, the other fields are not validated
if !self.common.enable {
return Ok(());
}
// verify that endpoint cannot be empty
if self.endpoint.trim().is_empty() {
return Err("Webhook endpoint cannot be empty".to_string());
}
// verification timeout must be reasonable
if self.timeout.is_some() {
match self.timeout {
Some(timeout) if timeout > 0 => {
info!("Webhook timeout is set to {}", timeout);
}
_ => return Err("Webhook timeout must be greater than 0".to_string()),
}
}
// Verify that the maximum number of retry is reasonable
if self.max_retries > 10 {
return Err("Maximum retry count cannot exceed 10".to_string());
}
// Verify the queue directory path
if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() {
return Err("Queue directory path should be absolute".to_string());
}
// The authentication certificate and key must appear in pairs
if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some())
{
return Err("Certificate and key must be specified as a pair".to_string());
}
Ok(())
}
/// Create a new webhook configuration
pub fn new(identifier: impl Into<String>, endpoint: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
endpoint: endpoint.into(),
..Default::default()
}
}
}
/// Configuration for the Kafka adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub brokers: String,
pub topic: String,
pub max_retries: u32,
pub timeout: u64,
}
impl Default for KafkaConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
brokers: String::new(),
topic: String::new(),
max_retries: 3,
timeout: 5000,
}
}
}
impl KafkaConfig {
/// Create a new Kafka configuration
pub fn new(identifier: impl Into<String>, brokers: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
brokers: brokers.into(),
topic: topic.into(),
..Default::default()
}
}
}
/// Configuration for the MQTT adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MqttConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub broker: String,
pub port: u16,
pub client_id: String,
pub topic: String,
pub max_retries: u32,
}
impl Default for MqttConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
broker: String::new(),
port: 1883,
client_id: String::new(),
topic: String::new(),
max_retries: 3,
}
}
}
impl MqttConfig {
/// Create a new MQTT configuration
pub fn new(identifier: impl Into<String>, broker: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
broker: broker.into(),
topic: topic.into(),
..Default::default()
}
}
}
/// Configuration for the adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum AdapterConfig {
Webhook(WebhookConfig),
Kafka(KafkaConfig),
Mqtt(MqttConfig),
}
/// Configuration for the notification system.
///
/// This struct contains the configuration for the notification system, including
/// the storage path, channel capacity, and a list of adapters.
///
/// # Fields
///
/// - `store_path`: The path to the storage directory.
/// - `channel_capacity`: The capacity of the notification channel.
/// - `adapters`: A list of adapters to be used for notifications.
///
/// # Example
///
/// ```
/// use rustfs_event::NotifierConfig;
///
/// let config = NotifierConfig::new();
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NotifierConfig {
#[serde(default = "default_queue_dir")]
pub store_path: String,
#[serde(default = "default_queue_limit")]
pub channel_capacity: u64,
pub adapters: Vec<AdapterConfig>,
}
impl Default for NotifierConfig {
fn default() -> Self {
Self {
store_path: default_queue_dir(),
channel_capacity: default_queue_limit(),
adapters: Vec::new(),
}
}
}
impl NotifierConfig {
/// create a new configuration with default values
pub fn new() -> Self {
Self::default()
}
/// Loading the configuration file
/// Supports TOML, YAML and .env formats, read in order by priority
///
/// # Parameters
/// - `config_dir`: Configuration file path
///
/// # Returns
/// Configuration information
///
/// # Example
/// ```
/// use rustfs_event::NotifierConfig;
///
/// let config = NotifierConfig::event_load_config(None);
/// ```
pub fn event_load_config(config_dir: Option<String>) -> NotifierConfig {
let config_dir = if let Some(path) = config_dir {
// If a path is provided, check if it's empty
if path.is_empty() {
// If empty, use the default config file name
DEFAULT_CONFIG_FILE.to_string()
} else {
// Use the provided path
let path = Path::new(&path);
if path.extension().is_some() {
// If path has extension, use it as is (extension will be added by Config::builder)
path.with_extension("").to_string_lossy().into_owned()
} else {
// If path is a directory, append the default config file name
path.to_string_lossy().into_owned()
}
}
} else {
// If no path provided, use current directory + default config file
match env::current_dir() {
Ok(dir) => dir.join(DEFAULT_CONFIG_FILE).to_string_lossy().into_owned(),
Err(_) => {
eprintln!("Warning: Failed to get current directory, using default config file");
DEFAULT_CONFIG_FILE.to_string()
}
}
};
// Log using proper logging instead of println when possible
println!("Using config file base: {}", config_dir);
let app_config = Config::builder()
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Toml).required(false))
.add_source(File::with_name(config_dir.as_str()).format(FileFormat::Yaml).required(false))
.build()
.unwrap_or_default();
match app_config.try_deserialize::<NotifierConfig>() {
Ok(app_config) => {
println!("Parsed AppConfig: {:?} \n", app_config);
app_config
}
Err(e) => {
println!("Failed to deserialize config: {}", e);
NotifierConfig::default()
}
}
}
/// unmarshal the configuration from a byte array
pub fn unmarshal(data: &[u8]) -> common::error::Result<NotifierConfig> {
let m: NotifierConfig = serde_json::from_slice(data)?;
Ok(m)
}
/// marshal the configuration to a byte array
pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
let data = serde_json::to_vec(&self)?;
Ok(data)
}
/// merge the configuration with default values
pub fn merge(&self) -> NotifierConfig {
self.clone()
}
}
/// Provide temporary directories as default storage paths
fn default_queue_dir() -> String {
env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
})
}
/// Provides the recommended default channel capacity for high concurrency systems
pub(crate) fn default_queue_limit() -> u64 {
env::var("EVENT_CHANNEL_CAPACITY")
.unwrap_or_else(|_| "10000".to_string())
.parse()
.unwrap_or(10000) // Default to 10000 if parsing fails
}
/// Event Notifier Configuration
/// This struct contains the configuration for the event notifier system,
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct EventNotifierConfig {
/// A collection of webhook configurations, with the key being a unique identifier
#[serde(default)]
pub webhook: HashMap<String, WebhookConfig>,
/// A collection of Kafka configurations, with the key being a unique identifier
#[serde(default)]
pub kafka: HashMap<String, KafkaConfig>,
///MQTT configuration collection, with the key being a unique identifier
#[serde(default)]
pub mqtt: HashMap<String, MqttConfig>,
}
impl EventNotifierConfig {
/// Create a new default configuration
pub fn new() -> Self {
Self::default()
}
/// Load the configuration from the file
pub fn event_load_config(_config_dir: Option<String>) -> EventNotifierConfig {
// The existing implementation remains the same, but returns EventNotifierConfig
// ...
Self::default()
}
/// Deserialization configuration
pub fn unmarshal(data: &[u8]) -> common::error::Result<EventNotifierConfig> {
let m: EventNotifierConfig = serde_json::from_slice(data)?;
Ok(m)
}
/// Serialization configuration
pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
let data = serde_json::to_vec(&self)?;
Ok(data)
}
/// Convert this configuration to a list of adapter configurations
pub fn to_adapter_configs(&self) -> Vec<AdapterConfig> {
let mut adapters = Vec::new();
// Add all enabled webhook configurations
for webhook in self.webhook.values() {
if webhook.common.enable {
adapters.push(AdapterConfig::Webhook(webhook.clone()));
}
}
// Add all enabled Kafka configurations
for kafka in self.kafka.values() {
if kafka.common.enable {
adapters.push(AdapterConfig::Kafka(kafka.clone()));
}
}
// Add all enabled MQTT configurations
for mqtt in self.mqtt.values() {
if mqtt.common.enable {
adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
}
}
adapters
}
}
crates/event/src/config/adapter.rs
+42
View File
@@ -0,0 +1,42 @@
use crate::config::kafka::KafkaConfig;
use crate::config::mqtt::MqttConfig;
use crate::config::webhook::WebhookConfig;
use crate::config::{default_queue_dir, default_queue_limit};
use serde::{Deserialize, Serialize};
/// Add a common field for the adapter configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AdapterCommon {
/// Adapter identifier for unique identification
pub identifier: String,
/// Adapter description information
pub comment: String,
/// Whether to enable this adapter
#[serde(default)]
pub enable: bool,
#[serde(default = "default_queue_dir")]
pub queue_dir: String,
#[serde(default = "default_queue_limit")]
pub queue_limit: u64,
}
impl Default for AdapterCommon {
fn default() -> Self {
Self {
identifier: String::new(),
comment: String::new(),
enable: false,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
}
}
}
/// Configuration for the adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum AdapterConfig {
Webhook(WebhookConfig),
Kafka(KafkaConfig),
Mqtt(MqttConfig),
}
crates/event/src/config/kafka.rs
+44
View File
@@ -0,0 +1,44 @@
use crate::config::adapter::AdapterCommon;
use crate::config::{default_queue_dir, default_queue_limit};
use serde::{Deserialize, Serialize};
/// Configuration for the Kafka adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct KafkaConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub brokers: String,
pub topic: String,
pub max_retries: u32,
pub timeout: u64,
}
impl Default for KafkaConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
brokers: String::new(),
topic: String::new(),
max_retries: 3,
timeout: 5000,
}
}
}
impl KafkaConfig {
/// Create a new Kafka configuration
pub fn new(identifier: impl Into<String>, brokers: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
brokers: brokers.into(),
topic: topic.into(),
..Default::default()
}
}
}
crates/event/src/config/mod.rs
+38
View File
@@ -0,0 +1,38 @@
use std::env;
pub mod adapter;
pub mod kafka;
pub mod mqtt;
pub mod notifier;
pub mod webhook;
/// The default configuration file name
const DEFAULT_CONFIG_FILE: &str = "event";
/// The prefix for the configuration file
pub const STORE_PREFIX: &str = "rustfs";
/// The default retry interval for the webhook adapter
pub const DEFAULT_RETRY_INTERVAL: u64 = 3;
/// The default maximum retry count for the webhook adapter
pub const DEFAULT_MAX_RETRIES: u32 = 3;
/// The default notification queue limit
pub const DEFAULT_NOTIFY_QUEUE_LIMIT: u64 = 10000;
/// Provide temporary directories as default storage paths
pub(crate) fn default_queue_dir() -> String {
env::var("EVENT_QUEUE_DIR").unwrap_or_else(|e| {
tracing::info!("Failed to get `EVENT_QUEUE_DIR` failed err: {}", e.to_string());
env::temp_dir().join(DEFAULT_CONFIG_FILE).to_string_lossy().to_string()
})
}
/// Provides the recommended default channel capacity for high concurrency systems
pub(crate) fn default_queue_limit() -> u64 {
env::var("EVENT_CHANNEL_CAPACITY")
.unwrap_or_else(|_| DEFAULT_NOTIFY_QUEUE_LIMIT.to_string())
.parse()
.unwrap_or(DEFAULT_NOTIFY_QUEUE_LIMIT) // Default to 10000 if parsing fails
}
crates/event/src/config/mqtt.rs
+46
View File
@@ -0,0 +1,46 @@
use crate::config::adapter::AdapterCommon;
use crate::config::{default_queue_dir, default_queue_limit};
use serde::{Deserialize, Serialize};
/// Configuration for the MQTT adapter.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MqttConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub broker: String,
pub port: u16,
pub client_id: String,
pub topic: String,
pub max_retries: u32,
}
impl Default for MqttConfig {
fn default() -> Self {
Self {
common: AdapterCommon::default(),
broker: String::new(),
port: 1883,
client_id: String::new(),
topic: String::new(),
max_retries: 3,
}
}
}
impl MqttConfig {
/// Create a new MQTT configuration
pub fn new(identifier: impl Into<String>, broker: impl Into<String>, topic: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
broker: broker.into(),
topic: topic.into(),
..Default::default()
}
}
}
crates/event/src/config/notifier.rs
+73
View File
@@ -0,0 +1,73 @@
use crate::config::{adapter::AdapterConfig, kafka::KafkaConfig, mqtt::MqttConfig, webhook::WebhookConfig};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
/// Event Notifier Configuration
/// This struct contains the configuration for the event notifier system,
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct EventNotifierConfig {
/// A collection of webhook configurations, with the key being a unique identifier
#[serde(default)]
pub webhook: HashMap<String, WebhookConfig>,
/// A collection of Kafka configurations, with the key being a unique identifier
#[serde(default)]
pub kafka: HashMap<String, KafkaConfig>,
///MQTT configuration collection, with the key being a unique identifier
#[serde(default)]
pub mqtt: HashMap<String, MqttConfig>,
}
impl EventNotifierConfig {
/// Create a new default configuration
pub fn new() -> Self {
Self::default()
}
/// Load the configuration from the file
pub fn event_load_config(_config_dir: Option<String>) -> EventNotifierConfig {
// The existing implementation remains the same, but returns EventNotifierConfig
// ...
Self::default()
}
/// Deserialization configuration
pub fn unmarshal(data: &[u8]) -> common::error::Result<EventNotifierConfig> {
let m: EventNotifierConfig = serde_json::from_slice(data)?;
Ok(m)
}
/// Serialization configuration
pub fn marshal(&self) -> common::error::Result<Vec<u8>> {
let data = serde_json::to_vec(&self)?;
Ok(data)
}
/// Convert this configuration to a list of adapter configurations
pub fn to_adapter_configs(&self) -> Vec<AdapterConfig> {
let mut adapters = Vec::new();
// Add all enabled webhook configurations
for webhook in self.webhook.values() {
if webhook.common.enable {
adapters.push(AdapterConfig::Webhook(webhook.clone()));
}
}
// Add all enabled Kafka configurations
for kafka in self.kafka.values() {
if kafka.common.enable {
adapters.push(AdapterConfig::Kafka(kafka.clone()));
}
}
// Add all enabled MQTT configurations
for mqtt in self.mqtt.values() {
if mqtt.common.enable {
adapters.push(AdapterConfig::Mqtt(mqtt.clone()));
}
}
adapters
}
}
crates/event/src/config/webhook.rs
+88
View File
@@ -0,0 +1,88 @@
use crate::config::adapter::AdapterCommon;
use crate::config::{default_queue_dir, default_queue_limit};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;
use tracing::info;
/// Configuration for the webhook adapter.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct WebhookConfig {
#[serde(flatten)]
pub common: AdapterCommon,
pub endpoint: String,
pub auth_token: Option<String>,
pub custom_headers: Option<HashMap<String, String>>,
pub max_retries: u32,
pub retry_interval: Option<u64>,
pub timeout: Option<u64>,
#[serde(default)]
pub client_cert: Option<String>,
#[serde(default)]
pub client_key: Option<String>,
}
impl WebhookConfig {
/// validate the configuration for the webhook adapter
///
/// # Returns
///
/// - `Result<(), String>`: Ok if the configuration is valid, Err with a message if invalid.
///
/// # Errors
/// - Returns an error if the configuration is invalid, such as empty endpoint, unreasonable timeout, or mismatched certificate and key.
pub fn validate(&self) -> Result<(), String> {
// If not enabled, the other fields are not validated
if !self.common.enable {
return Ok(());
}
// verify that endpoint cannot be empty
if self.endpoint.trim().is_empty() {
return Err("Webhook endpoint cannot be empty".to_string());
}
// verification timeout must be reasonable
if self.timeout.is_some() {
match self.timeout {
Some(timeout) if timeout > 0 => {
info!("Webhook timeout is set to {}", timeout);
}
_ => return Err("Webhook timeout must be greater than 0".to_string()),
}
}
// Verify that the maximum number of retry is reasonable
if self.max_retries > 10 {
return Err("Maximum retry count cannot exceed 10".to_string());
}
// Verify the queue directory path
if !self.common.queue_dir.is_empty() && !Path::new(&self.common.queue_dir).is_absolute() {
return Err("Queue directory path should be absolute".to_string());
}
// The authentication certificate and key must appear in pairs
if (self.client_cert.is_some() && self.client_key.is_none()) || (self.client_cert.is_none() && self.client_key.is_some())
{
return Err("Certificate and key must be specified as a pair".to_string());
}
Ok(())
}
/// Create a new webhook configuration
pub fn new(identifier: impl Into<String>, endpoint: impl Into<String>) -> Self {
Self {
common: AdapterCommon {
identifier: identifier.into(),
comment: String::new(),
enable: true,
queue_dir: default_queue_dir(),
queue_limit: default_queue_limit(),
},
endpoint: endpoint.into(),
..Default::default()
}
}
}
crates/event/src/event_notifier.rs
+142
View File
@@ -0,0 +1,142 @@
use crate::config::notifier::EventNotifierConfig;
use crate::Event;
use common::error::{Error, Result};
use ecstore::store::ECStore;
use std::sync::Arc;
use tokio::sync::{broadcast, mpsc};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, instrument, warn};
use tracing_subscriber::util::SubscriberInitExt;
/// 事件通知器
pub struct EventNotifier {
/// 事件发送通道
sender: mpsc::Sender<Event>,
/// 接收器任务句柄
task_handle: Option<tokio::task::JoinHandle<()>>,
/// 配置信息
config: EventNotifierConfig,
/// 关闭标记
shutdown: CancellationToken,
/// 关闭通知通道
shutdown_complete_tx: Option<broadcast::Sender<()>>,
}
impl EventNotifier {
/// 创建新的事件通知器
#[instrument(skip_all)]
pub async fn new(store: Arc<ECStore>) -> Result<Self> {
let manager = crate::store::manager::EventManager::new(store);
// 初始化配置
let config = manager.init().await?;
// 创建适配器
let adapters = manager.create_adapters().await?;
info!("创建了 {} 个适配器", adapters.len());
// 创建关闭标记
let shutdown = CancellationToken::new();
let (shutdown_complete_tx, _) = broadcast::channel(1);
// 创建事件通道 - 使用默认容量，因为每个适配器都有自己的队列
// 这里使用较小的通道容量，因为事件会被快速分发到适配器
let (sender, mut receiver) = mpsc::channel(100);
let shutdown_clone = shutdown.clone();
let shutdown_complete_tx_clone = shutdown_complete_tx.clone();
let adapters_clone = adapters.clone();
// 启动事件处理任务
let task_handle = tokio::spawn(async move {
debug!("事件处理任务启动");
loop {
tokio::select! {
Some(event) = receiver.recv() => {
debug!("收到事件：{}", event.id);
// 分发到所有适配器
for adapter in &adapters_clone {
let adapter_name = adapter.name();
match adapter.send(&event).await {
Ok(_) => {
debug!("事件 {} 成功发送到适配器 {}", event.id, adapter_name);
}
Err(e) => {
error!("事件 {} 发送到适配器 {} 失败：{}", event.id, adapter_name, e);
}
}
}
}
_ = shutdown_clone.cancelled() => {
info!("接收到关闭信号，事件处理任务停止");
let _ = shutdown_complete_tx_clone.send(());
break;
}
}
}
debug!("事件处理任务已停止");
});
Ok(Self {
sender,
task_handle: Some(task_handle),
config,
shutdown,
shutdown_complete_tx: Some(shutdown_complete_tx),
})
}
/// 关闭事件通知器
pub async fn shutdown(&mut self) -> Result<()> {
info!("关闭事件通知器");
self.shutdown.cancel();
if let Some(shutdown_tx) = self.shutdown_complete_tx.take() {
let mut rx = shutdown_tx.subscribe();
// 等待关闭完成信号或超时
tokio::select! {
_ = rx.recv() => {
debug!("收到关闭完成信号");
}
_ = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
warn!("关闭超时，强制终止");
}
}
}
if let Some(handle) = self.task_handle.take() {
handle.abort();
match handle.await {
Ok(_) => debug!("事件处理任务已正常终止"),
Err(e) => {
if e.is_cancelled() {
debug!("事件处理任务已取消");
} else {
error!("等待事件处理任务终止时出错：{}", e);
}
}
}
}
info!("事件通知器已完全关闭");
Ok(())
}
/// 发送事件
pub async fn send(&self, event: Event) -> Result<()> {
self.sender
.send(event)
.await
.map_err(|e| Error::msg(format!("发送事件到通道失败：{}", e)))
}
/// 获取当前配置
pub fn config(&self) -> &EventNotifierConfig {
&self.config
}
}
crates/event/src/event_system.rs
+81
View File
@@ -0,0 +1,81 @@
use crate::config::notifier::EventNotifierConfig;
use crate::event_notifier::EventNotifier;
use common::error::Result;
use ecstore::store::ECStore;
use once_cell::sync::OnceCell;
use std::sync::{Arc, Mutex};
use tracing::{debug, error, info};
/// 全局事件系统
pub struct EventSystem {
/// 事件通知器
notifier: Mutex<Option<EventNotifier>>,
}
impl EventSystem {
/// 创建一个新的事件系统
pub fn new() -> Self {
Self {
notifier: Mutex::new(None),
}
}
/// 初始化事件系统
pub async fn init(&self, store: Arc<ECStore>) -> Result<EventNotifierConfig> {
info!("初始化事件系统");
let notifier = EventNotifier::new(store).await?;
let config = notifier.config().clone();
let mut guard = self
.notifier
.lock()
.map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?;
*guard = Some(notifier);
debug!("事件系统初始化完成");
Ok(config)
}
/// 发送事件
pub async fn send_event(&self, event: crate::Event) -> Result<()> {
let guard = self
.notifier
.lock()
.map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?;
if let Some(notifier) = &*guard {
notifier.send(event).await
} else {
error!("事件系统未初始化");
Err(common::error::Error::msg("事件系统未初始化"))
}
}
/// 关闭事件系统
pub async fn shutdown(&self) -> Result<()> {
info!("关闭事件系统");
let mut guard = self
.notifier
.lock()
.map_err(|e| common::error::Error::msg(format!("获取锁失败：{}", e)))?;
if let Some(ref mut notifier) = *guard {
notifier.shutdown().await?;
*guard = None;
info!("事件系统已关闭");
Ok(())
} else {
debug!("事件系统已经关闭");
Ok(())
}
}
}
/// 全局事件系统实例
pub static GLOBAL_EVENT_SYS: OnceCell<EventSystem> = OnceCell::new();
/// 初始化全局事件系统
pub fn init_global_event_system() -> &'static EventSystem {
GLOBAL_EVENT_SYS.get_or_init(EventSystem::new)
}
crates/event/src/lib.rs
+9 -9
View File
@@ -1,8 +1,9 @@
mod adapter; mod adapter;
mod bus;
mod config; mod config;
mod error; mod error;
mod event; mod event;
mod event_notifier;
mod event_system;
mod global; mod global;
mod notifier; mod notifier;
mod store; mod store;
@@ -16,20 +17,19 @@ pub use adapter::mqtt::MqttAdapter;
pub use adapter::webhook::WebhookAdapter; pub use adapter::webhook::WebhookAdapter;
pub use adapter::ChannelAdapter; pub use adapter::ChannelAdapter;
pub use adapter::ChannelAdapterType; pub use adapter::ChannelAdapterType;
pub use bus::event_bus; pub use config::adapter::AdapterCommon;
pub use config::AdapterCommon; pub use config::adapter::AdapterConfig;
pub use config::EventNotifierConfig;
#[cfg(all(feature = "kafka", target_os = "linux"))] #[cfg(all(feature = "kafka", target_os = "linux"))]
pub use config::KafkaConfig; pub use config::kafka::KafkaConfig;
#[cfg(feature = "mqtt")] #[cfg(feature = "mqtt")]
pub use config::MqttConfig; pub use config::mqtt::MqttConfig;
pub use config::notifier::EventNotifierConfig;
#[cfg(feature = "webhook")] #[cfg(feature = "webhook")]
pub use config::WebhookConfig; pub use config::webhook::WebhookConfig;
pub use config::{AdapterConfig, NotifierConfig, DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL}; pub use config::{DEFAULT_MAX_RETRIES, DEFAULT_RETRY_INTERVAL};
pub use error::Error; pub use error::Error;
pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source}; pub use event::{Bucket, Event, EventBuilder, Identity, Log, Metadata, Name, Object, Source};
pub use global::{initialize, is_initialized, is_ready, send_event, shutdown}; pub use global::{initialize, is_initialized, is_ready, send_event, shutdown};
pub use notifier::NotifierSystem; pub use notifier::NotifierSystem;
pub use store::event::EventStore;
pub use store::queue::QueueStore; pub use store::queue::QueueStore;
crates/event/src/store/event.rs
-60
View File
@@ -1,60 +0,0 @@
use crate::Error;
use crate::Log;
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::fs::{create_dir_all, File, OpenOptions};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, BufWriter};
use tokio::sync::RwLock;
use tracing::instrument;
/// `EventStore` is a struct that manages the storage of event logs.
pub struct EventStore {
path: String,
lock: Arc<RwLock<()>>,
}
impl EventStore {
pub async fn new(path: &str) -> Result<Self, Error> {
create_dir_all(path).await?;
Ok(Self {
path: path.to_string(),
lock: Arc::new(RwLock::new(())),
})
}
#[instrument(skip(self))]
pub async fn save_logs(&self, logs: &[Log]) -> Result<(), Error> {
let _guard = self.lock.write().await;
let file_path = format!(
"{}/events_{}.jsonl",
self.path,
SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs()
);
let file = OpenOptions::new().create(true).append(true).open(&file_path).await?;
let mut writer = BufWriter::new(file);
for log in logs {
let line = serde_json::to_string(log)?;
writer.write_all(line.as_bytes()).await?;
writer.write_all(b"\n").await?;
}
writer.flush().await?;
tracing::info!("Saved logs to {} end", file_path);
Ok(())
}
pub async fn load_logs(&self) -> Result<Vec<Log>, Error> {
let _guard = self.lock.read().await;
let mut logs = Vec::new();
let mut entries = tokio::fs::read_dir(&self.path).await?;
while let Some(entry) = entries.next_entry().await? {
let file = File::open(entry.path()).await?;
let reader = BufReader::new(file);
let mut lines = reader.lines();
while let Some(line) = lines.next_line().await? {
let log: Log = serde_json::from_str(&line)?;
logs.push(log);
}
}
Ok(logs)
}
}
crates/event/src/store/manager.rs
+1 -1
View File
@@ -162,7 +162,7 @@ impl EventManager {
}; };
let adapter_configs = config.to_adapter_configs(); let adapter_configs = config.to_adapter_configs();
match adapter::create_adapters(&adapter_configs) { match adapter::create_adapters(adapter_configs) {
Ok(adapters) => Ok(adapters), Ok(adapters) => Ok(adapters),
Err(err) => { Err(err) => {
tracing::error!("Failed to create adapters: {:?}", err); tracing::error!("Failed to create adapters: {:?}", err);
crates/event/src/store/mod.rs
-1
View File
@@ -1,3 +1,2 @@
pub(crate) mod event;
pub(crate) mod manager; pub(crate) mod manager;
pub(crate) mod queue; pub(crate) mod queue;
crates/event/src/store/queue.rs
+71 -8
View File
@@ -1,17 +1,18 @@
use common::error::{Error, Result}; use common::error::{Error, Result};
use ecstore::utils::path::dir;
use serde::{de::DeserializeOwned, Serialize}; use serde::{de::DeserializeOwned, Serialize};
use snap::raw::{Decoder, Encoder}; use snap::raw::{Decoder, Encoder};
use std::collections::HashMap; use std::collections::HashMap;
use std::io::Read; use std::io::Read;
use std::marker::PhantomData; use std::marker::PhantomData;
use std::path::PathBuf; use std::path::{Path, PathBuf};
use std::sync::{Arc, RwLock}; use std::sync::{Arc, RwLock};
use std::time::{SystemTime, UNIX_EPOCH}; use std::time::{SystemTime, UNIX_EPOCH};
use std::{fs, io}; use std::{fs, io};
use uuid::Uuid; use uuid::Uuid;
/// Keys in storage /// Keys in storage
#[derive(Debug, Clone, PartialEq, Eq)] #[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct Key { pub struct Key {
/// Key name /// Key name
pub name: String, pub name: String,
@@ -42,7 +43,7 @@ impl Key {
key_str = format!("{}:{}", self.item_count, self.name); key_str = format!("{}:{}", self.item_count, self.name);
} }
let compress_ext = if self.compress { ".snappy" } else { "" }; let compress_ext = if self.compress { COMPRESS_EXT } else { "" };
format!("{}{}{}", key_str, self.extension, compress_ext) format!("{}{}{}", key_str, self.extension, compress_ext)
} }
} }
@@ -50,7 +51,7 @@ impl Key {
/// Parse key from file name /// Parse key from file name
#[allow(clippy::redundant_closure)] #[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(COMPRESS_EXT);
let filename = if compress { let filename = if compress {
&filename[..filename.len() - 7] // 移除 ".snappy" &filename[..filename.len() - 7] // 移除 ".snappy"
} else { } else {
@@ -85,7 +86,7 @@ pub fn parse_key(filename: &str) -> Key {
/// Store the characteristics of the project /// Store the characteristics of the project
pub trait Store<T>: Send + Sync pub trait Store<T>: Send + Sync
where where
T: Serialize + DeserializeOwned + Send + Sync, T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
{ {
/// Store a single item /// Store a single item
fn put(&self, item: T) -> Result<Key>; fn put(&self, item: T) -> Result<Key>;
@@ -142,26 +143,88 @@ pub struct QueueStore<T> {
entries: Arc<RwLock<HashMap<String, i64>>>, entries: Arc<RwLock<HashMap<String, i64>>>,
/// Type tags /// Type tags
_phantom: PhantomData<T>, _phantom: PhantomData<T>,
/// Whether to compress
compress: bool,
/// Store name
name: String,
} }
impl<T> QueueStore<T> impl<T> QueueStore<T>
where where
T: Serialize + DeserializeOwned + Send + Sync, T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
{ {
/// Create a new queue store /// Create a new queue store
pub fn new(directory: impl Into<PathBuf>, limit: u64, ext: Option<String>) -> Self { pub fn new<P: AsRef<Path>>(directory: P, name: String, limit: u64, ext: Option<String>) -> Self {
let limit = if limit == 0 { DEFAULT_LIMIT } else { limit }; let limit = if limit == 0 { DEFAULT_LIMIT } else { limit };
let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string()); let ext = ext.unwrap_or_else(|| DEFAULT_EXT.to_string());
let mut path = PathBuf::from(directory.as_ref());
path.push(&name);
// Create a directory (if it doesn't exist)
if !path.exists() {
if let Err(e) = fs::create_dir_all(&path) {
tracing::error!("创建存储目录失败 {}: {}", path.display(), e);
}
}
Self { Self {
directory: directory.into(), directory: directory.into(),
name,
entry_limit: limit, entry_limit: limit,
file_ext: ext, file_ext: ext,
compress: true, // Default to compressing
entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))), entries: Arc::new(RwLock::new(HashMap::with_capacity(limit as usize))),
_phantom: PhantomData, _phantom: PhantomData,
} }
} }
/// Set the file extension
pub fn with_file_ext(mut self, file_ext: &str) -> Self {
self.file_ext = file_ext.to_string();
self
}
/// Set whether to compress or not
pub fn with_compression(mut self, compress: bool) -> Self {
self.compress = compress;
self
}
/// Get the file path
fn get_file_path(&self, key: &Key) -> PathBuf {
let mut filename = key.to_string();
filename.push_str(if self.compress { COMPRESS_EXT } else { &self.file_ext });
self.directory.join(filename)
}
/// Serialize the project
fn serialize_item(&self, item: &T) -> Result<Vec<u8>> {
let data = serde_json::to_vec(item).map_err(|e| Error::msg(format!("Serialization failed: {}", e)))?;
if self.compress {
let mut encoder = Encoder::new();
Ok(encoder
.compress_vec(&data)
.map_err(|e| Error::msg(format!("Compression failed: {}", e)))?)
} else {
Ok(data)
}
}
/// Deserialize the project
fn deserialize_item(&self, data: &[u8], is_compressed: bool) -> Result<T> {
let data = if is_compressed {
let mut decoder = Decoder::new();
decoder
.decompress_vec(data)
.map_err(|e| Error::msg(format!("Unzipping failed: {}", e)))?
} else {
data.to_vec()
};
serde_json::from_slice(&data).map_err(|e| Error::msg(format!("Deserialization failed: {}", e)))
}
/// Lists all files in the directory, sorted by modification time (oldest takes precedence.)） /// Lists all files in the directory, sorted by modification time (oldest takes precedence.)）
fn list_files(&self) -> Result<Vec<fs::DirEntry>> { fn list_files(&self) -> Result<Vec<fs::DirEntry>> {
let mut files = Vec::new(); let mut files = Vec::new();
@@ -264,7 +327,7 @@ where
impl<T> Store<T> for QueueStore<T> impl<T> Store<T> for QueueStore<T>
where where
T: Serialize + DeserializeOwned + Send + Sync + Clone, T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static,
{ {
fn open(&self) -> Result<()> { fn open(&self) -> Result<()> {
// Create a directory (if it doesn't exist) // Create a directory (if it doesn't exist)
ecstore/src/config/com.rs
+27 -31
View File
@@ -114,22 +114,12 @@ async fn new_and_save_server_config<S: StorageAPI>(api: Arc<S>) -> Result<Config
Ok(cfg) Ok(cfg)
} }
fn get_config_file() -> String {
format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE)
}
pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<Config> { pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<Config> {
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); let data = handle_read_config(api.clone()).await?;
let data = match read_config(api.clone(), config_file.as_str()).await {
Ok(res) => res,
Err(err) => {
return if is_err_config_not_found(&err) {
warn!("config not found, start to init");
let cfg = new_and_save_server_config(api).await?;
warn!("config init done");
Ok(cfg)
} else {
error!("read config err {:?}", &err);
Err(err)
}
}
};
read_server_config(api, data.as_slice()).await read_server_config(api, data.as_slice()).await
} }
@@ -137,21 +127,8 @@ pub async fn read_config_without_migrate<S: StorageAPI>(api: Arc<S>) -> Result<C
async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<Config> { async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<Config> {
let cfg = { let cfg = {
if data.is_empty() { if data.is_empty() {
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); let cfg_data = handle_read_config(api.clone()).await?;
let cfg_data = match read_config(api.clone(), config_file.as_str()).await {
Ok(res) => res,
Err(err) => {
return if is_err_config_not_found(&err) {
warn!("config not found init start");
let cfg = new_and_save_server_config(api).await?;
warn!("config not found init done");
Ok(cfg)
} else {
error!("read config err {:?}", &err);
Err(err)
}
}
};
// TODO: decrypt // TODO: decrypt
Config::unmarshal(cfg_data.as_slice())? Config::unmarshal(cfg_data.as_slice())?
@@ -163,10 +140,29 @@ async fn read_server_config<S: StorageAPI>(api: Arc<S>, data: &[u8]) -> Result<C
Ok(cfg.merge()) Ok(cfg.merge())
} }
async fn handle_read_config<S: StorageAPI>(api: Arc<S>) -> Result<Vec<u8>> {
let config_file = get_config_file();
match read_config(api.clone(), config_file.as_str()).await {
Ok(res) => Ok(res),
Err(err) => {
if is_err_config_not_found(&err) {
warn!("config not found, start to init");
let cfg = new_and_save_server_config(api).await?;
warn!("config init done");
// This returns the serialized data, keeping the interface consistent
cfg.marshal()
} else {
error!("read config err {:?}", &err);
Err(err)
}
}
}
}
async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Result<()> { async fn save_server_config<S: StorageAPI>(api: Arc<S>, cfg: &Config) -> Result<()> {
let data = cfg.marshal()?; let data = cfg.marshal()?;
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); let config_file = get_config_file();
save_config(api, &config_file, data).await save_config(api, &config_file, data).await
} }
ecstore/src/config/mod.rs
+8
View File
@@ -19,6 +19,14 @@ lazy_static! {
pub static ref GLOBAL_ConfigSys: ConfigSys = ConfigSys::new(); pub static ref GLOBAL_ConfigSys: ConfigSys = ConfigSys::new();
} }
/// Standard config keys and values.
pub const ENABLE_KEY: &str = "enable";
pub const COMMENT_KEY: &str = "comment";
/// Enable values
pub const ENABLE_ON: &str = "on";
pub const ENABLE_OFF: &str = "off";
pub const ENV_ACCESS_KEY: &str = "RUSTFS_ACCESS_KEY"; pub const ENV_ACCESS_KEY: &str = "RUSTFS_ACCESS_KEY";
pub const ENV_SECRET_KEY: &str = "RUSTFS_SECRET_KEY"; pub const ENV_SECRET_KEY: &str = "RUSTFS_SECRET_KEY";
pub const ENV_ROOT_USER: &str = "RUSTFS_ROOT_USER"; pub const ENV_ROOT_USER: &str = "RUSTFS_ROOT_USER";
rustfs/src/event.rs
+1
View File
@@ -11,6 +11,7 @@ pub(crate) async fn init_event_notifier(notifier_config: Option<String>) {
} 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()
NotifierConfig::default()
}; };
info!("using event_config: {:?}", config); info!("using event_config: {:?}", config);
rustfs/src/main.rs
+1 -1
View File
@@ -510,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_EVENT_SYS.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;