mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-30 08:49:26 +00:00
feat(event-notifier): improve environment variable handling
- Fix deserialization error when parsing config from environment variables - Add proper array format support for adapters configuration - Update environment variable examples with correct format - Improve documentation for configuration loading - Implement helper functions for environment variable validation This change fixes the "invalid type: map, expected a sequence" error by ensuring proper formatting of array-type fields in environment variables.
This commit is contained in:
houseme
@@ -80,16 +80,20 @@ pub async fn event_bus(
|
||||
} else {
|
||||
tracing::info!("no unhandled events need to be saved");
|
||||
}
|
||||
tracing::info!("shutdown_complete is Some: {}", shutdown_complete.is_some());
|
||||
// send a completion signal
|
||||
tracing::debug!("shutdown_complete is Some: {}", shutdown_complete.is_some());
|
||||
|
||||
if let Some(complete_sender) = shutdown_complete {
|
||||
let _ = complete_sender.send(());
|
||||
// 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;
|
||||
}
|
||||
// else => break,
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
use crate::Error;
|
||||
use figment::providers::Format;
|
||||
use config::{Config, Environment, File, FileFormat};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::HashMap;
|
||||
use std::env;
|
||||
|
||||
/// Configuration for the notification system.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
@@ -65,7 +65,7 @@ pub enum AdapterConfig {
|
||||
|
||||
/// Configuration for the notification system.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct NotificationConfig {
|
||||
pub struct NotifierConfig {
|
||||
#[serde(default = "default_store_path")]
|
||||
pub store_path: String,
|
||||
#[serde(default = "default_channel_capacity")]
|
||||
@@ -73,7 +73,7 @@ pub struct NotificationConfig {
|
||||
pub adapters: Vec<AdapterConfig>,
|
||||
}
|
||||
|
||||
impl Default for NotificationConfig {
|
||||
impl Default for NotifierConfig {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
store_path: default_store_path(),
|
||||
@@ -83,46 +83,88 @@ impl Default for NotificationConfig {
|
||||
}
|
||||
}
|
||||
|
||||
impl NotificationConfig {
|
||||
impl NotifierConfig {
|
||||
/// create a new configuration with default values
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// create a configuration from a configuration file
|
||||
pub fn from_file(path: &str) -> Result<Self, Error> {
|
||||
let config = figment::Figment::new()
|
||||
.merge(figment::providers::Toml::file(path))
|
||||
.extract()?;
|
||||
/// 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_notifier::NotifierConfig;
|
||||
///
|
||||
/// let config = NotifierConfig::load_config(None);
|
||||
/// ```
|
||||
pub fn 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 = std::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()
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
Ok(config)
|
||||
}
|
||||
// Log using proper logging instead of println when possible
|
||||
println!("Using config file base: {}", config_dir);
|
||||
|
||||
/// Read configuration from multiple sources (support TOML, YAML, .env)
|
||||
pub fn load() -> Result<Self, Error> {
|
||||
let figment = figment::Figment::new()
|
||||
// First try to read the config.toml of the current directory
|
||||
.merge(figment::providers::Toml::file("event.toml"))
|
||||
// Then try to read the config.yaml of the current directory
|
||||
.merge(figment::providers::Yaml::file("event.yaml"))
|
||||
// Finally read the environment variable and overwrite the previous value
|
||||
.merge(figment::providers::Env::prefixed("EVENT_NOTIF_"));
|
||||
|
||||
Ok(figment.extract()?)
|
||||
}
|
||||
|
||||
/// loading configuration from env file
|
||||
pub fn from_env_file(path: &str) -> Result<Self, Error> {
|
||||
// loading env files
|
||||
dotenvy::from_path(path).map_err(|e| Error::ConfigError(format!("unable to load env file: {}", e)))?;
|
||||
|
||||
// Extract configuration from environment variables using figurement
|
||||
let figment = figment::Figment::new().merge(figment::providers::Env::prefixed("EVENT_NOTIF_"));
|
||||
|
||||
Ok(figment.extract()?)
|
||||
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))
|
||||
.add_source(
|
||||
Environment::default()
|
||||
.prefix("NOTIFIER")
|
||||
.prefix_separator("__")
|
||||
.separator("__")
|
||||
.list_separator("_")
|
||||
.with_list_parse_key("adapters")
|
||||
.try_parsing(true),
|
||||
)
|
||||
.build()
|
||||
.unwrap_or_default();
|
||||
println!("Loaded config: {:?}", app_config);
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const DEFAULT_CONFIG_FILE: &str = "obs";
|
||||
|
||||
/// Provide temporary directories as default storage paths
|
||||
fn default_store_path() -> String {
|
||||
std::env::temp_dir().join("event-notification").to_string_lossy().to_string()
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use config::ConfigError;
|
||||
use thiserror::Error;
|
||||
use tokio::sync::mpsc::error;
|
||||
use tokio::task::JoinError;
|
||||
@@ -35,7 +36,7 @@ pub enum Error {
|
||||
#[error("Configuration error: {0}")]
|
||||
ConfigError(String),
|
||||
#[error("Configuration loading error: {0}")]
|
||||
Figment(#[from] figment::Error),
|
||||
Config(#[from] ConfigError),
|
||||
}
|
||||
|
||||
impl Error {
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
use crate::{create_adapters, Error, Event, NotificationConfig, NotificationSystem};
|
||||
use crate::{create_adapters, Error, Event, NotifierConfig, NotifierSystem};
|
||||
use std::sync::{atomic, Arc};
|
||||
use tokio::sync::{Mutex, OnceCell};
|
||||
|
||||
static GLOBAL_SYSTEM: OnceCell<Arc<Mutex<NotificationSystem>>> = OnceCell::const_new();
|
||||
static GLOBAL_SYSTEM: OnceCell<Arc<Mutex<NotifierSystem>>> = OnceCell::const_new();
|
||||
static INITIALIZED: atomic::AtomicBool = atomic::AtomicBool::new(false);
|
||||
static READY: atomic::AtomicBool = atomic::AtomicBool::new(false);
|
||||
static INIT_LOCK: Mutex<()> = Mutex::const_new(());
|
||||
@@ -24,7 +24,7 @@ static INIT_LOCK: Mutex<()> = Mutex::const_new(());
|
||||
/// - Creating adapters fails.
|
||||
/// - Starting the notification system fails.
|
||||
/// - Setting the global system instance fails.
|
||||
pub async fn initialize(config: NotificationConfig) -> Result<(), Error> {
|
||||
pub async fn initialize(config: NotifierConfig) -> Result<(), Error> {
|
||||
let _lock = INIT_LOCK.lock().await;
|
||||
|
||||
// Check if the system is already initialized.
|
||||
@@ -52,7 +52,7 @@ pub async fn initialize(config: NotificationConfig) -> Result<(), Error> {
|
||||
|
||||
// Attempt to initialize, and reset the INITIALIZED flag if it fails.
|
||||
let result: Result<(), Error> = async {
|
||||
let system = NotificationSystem::new(config.clone()).await.map_err(|e| {
|
||||
let system = NotifierSystem::new(config.clone()).await.map_err(|e| {
|
||||
tracing::error!("Failed to create NotificationSystem: {:?}", e);
|
||||
e
|
||||
})?;
|
||||
@@ -127,24 +127,29 @@ pub async fn send_event(event: Event) -> Result<(), Error> {
|
||||
pub async fn shutdown() -> Result<(), Error> {
|
||||
if let Some(system) = GLOBAL_SYSTEM.get() {
|
||||
tracing::info!("Shutting down notification system start");
|
||||
let (complete_tx, complete_rx) = tokio::sync::oneshot::channel();
|
||||
|
||||
{
|
||||
let result = {
|
||||
let mut system_guard = system.lock().await;
|
||||
// set the complete channel and trigger cancellation
|
||||
system_guard.set_shutdown_complete_channel(complete_tx);
|
||||
system_guard.shutdown();
|
||||
tracing::info!("Notification system shutdown triggered");
|
||||
system_guard.shutdown().await
|
||||
};
|
||||
if let Err(e) = &result {
|
||||
tracing::error!("Notification system shutdown failed: {}", e);
|
||||
} else {
|
||||
tracing::info!("Event bus shutdown completed");
|
||||
}
|
||||
|
||||
// wait for the cleaning to be completed
|
||||
let _ = complete_rx.await;
|
||||
tracing::info!("Event bus shutdown completed");
|
||||
|
||||
tracing::info!(
|
||||
"Shutdown method called set static value start, READY: {}, INITIALIZED: {}",
|
||||
READY.load(atomic::Ordering::SeqCst),
|
||||
INITIALIZED.load(atomic::Ordering::SeqCst)
|
||||
);
|
||||
READY.store(false, atomic::Ordering::SeqCst);
|
||||
INITIALIZED.store(false, atomic::Ordering::SeqCst);
|
||||
tracing::info!("Notification system is ready to process events");
|
||||
Ok(())
|
||||
tracing::info!(
|
||||
"Shutdown method called set static value end, READY: {}, INITIALIZED: {}",
|
||||
READY.load(atomic::Ordering::SeqCst),
|
||||
INITIALIZED.load(atomic::Ordering::SeqCst)
|
||||
);
|
||||
result
|
||||
} else {
|
||||
Err(Error::custom("Notification system not initialized"))
|
||||
}
|
||||
@@ -155,7 +160,7 @@ pub async fn shutdown() -> Result<(), Error> {
|
||||
/// # Errors
|
||||
///
|
||||
/// Returns an error if the system is not initialized.
|
||||
async fn get_system() -> Result<Arc<Mutex<NotificationSystem>>, Error> {
|
||||
async fn get_system() -> Result<Arc<Mutex<NotifierSystem>>, Error> {
|
||||
GLOBAL_SYSTEM
|
||||
.get()
|
||||
.cloned()
|
||||
@@ -165,13 +170,13 @@ async fn get_system() -> Result<Arc<Mutex<NotificationSystem>>, Error> {
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::{AdapterConfig, NotificationConfig, WebhookConfig};
|
||||
use crate::{AdapterConfig, NotifierConfig, WebhookConfig};
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_initialize_success() {
|
||||
tracing_subscriber::fmt::init();
|
||||
let config = NotificationConfig::default(); // assume there is a default configuration
|
||||
let config = NotifierConfig::default(); // assume there is a default configuration
|
||||
let result = initialize(config).await;
|
||||
assert!(!result.is_ok(), "Initialization should succeed");
|
||||
assert!(!is_initialized(), "System should be marked as initialized");
|
||||
@@ -181,7 +186,7 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn test_initialize_twice() {
|
||||
tracing_subscriber::fmt::init();
|
||||
let config = NotificationConfig::default();
|
||||
let config = NotifierConfig::default();
|
||||
let _ = initialize(config.clone()).await; // first initialization
|
||||
let result = initialize(config).await; // second initialization
|
||||
assert!(!result.is_ok(), "Initialization should succeed");
|
||||
@@ -192,7 +197,7 @@ mod tests {
|
||||
async fn test_initialize_failure_resets_state() {
|
||||
tracing_subscriber::fmt::init();
|
||||
// simulate wrong configuration
|
||||
let config = NotificationConfig {
|
||||
let config = NotifierConfig {
|
||||
adapters: vec![
|
||||
// assuming that the empty adapter will cause failure
|
||||
AdapterConfig::Webhook(WebhookConfig {
|
||||
@@ -217,7 +222,7 @@ mod tests {
|
||||
assert!(!is_initialized(), "System should not be initialized initially");
|
||||
assert!(!is_ready(), "System should not be ready initially");
|
||||
|
||||
let config = NotificationConfig::default();
|
||||
let config = NotifierConfig::default();
|
||||
let _ = initialize(config).await;
|
||||
assert!(!is_initialized(), "System should be initialized after successful initialization");
|
||||
assert!(!is_ready(), "System should be ready after successful initialization");
|
||||
|
||||
@@ -22,10 +22,10 @@ pub use config::KafkaConfig;
|
||||
pub use config::MqttConfig;
|
||||
#[cfg(feature = "webhook")]
|
||||
pub use config::WebhookConfig;
|
||||
pub use config::{AdapterConfig, NotificationConfig};
|
||||
pub use config::{AdapterConfig, NotifierConfig};
|
||||
pub use error::Error;
|
||||
|
||||
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 notifier::NotificationSystem;
|
||||
pub use notifier::NotifierSystem;
|
||||
pub use store::EventStore;
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotificationConfig};
|
||||
use crate::{event_bus, ChannelAdapter, Error, Event, EventStore, NotifierConfig};
|
||||
use std::sync::Arc;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio_util::sync::CancellationToken;
|
||||
@@ -7,17 +7,18 @@ use tokio_util::sync::CancellationToken;
|
||||
/// It manages the event bus and the adapters.
|
||||
/// It is responsible for sending and receiving events.
|
||||
/// It also handles the shutdown process.
|
||||
pub struct NotificationSystem {
|
||||
pub struct NotifierSystem {
|
||||
tx: mpsc::Sender<Event>,
|
||||
rx: Option<mpsc::Receiver<Event>>,
|
||||
store: Arc<EventStore>,
|
||||
shutdown: CancellationToken,
|
||||
shutdown_complete: Option<tokio::sync::oneshot::Sender<()>>,
|
||||
shutdown_receiver: Option<tokio::sync::oneshot::Receiver<()>>,
|
||||
}
|
||||
|
||||
impl NotificationSystem {
|
||||
impl NotifierSystem {
|
||||
/// Creates a new `NotificationSystem` instance.
|
||||
pub async fn new(config: NotificationConfig) -> Result<Self, Error> {
|
||||
pub async fn new(config: NotifierConfig) -> Result<Self, Error> {
|
||||
let (tx, rx) = mpsc::channel::<Event>(config.channel_capacity);
|
||||
let store = Arc::new(EventStore::new(&config.store_path).await?);
|
||||
let shutdown = CancellationToken::new();
|
||||
@@ -29,13 +30,15 @@ impl NotificationSystem {
|
||||
tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
|
||||
}
|
||||
}
|
||||
|
||||
// Initialize shutdown_complete to Some(tx)
|
||||
let (complete_tx, complete_rx) = tokio::sync::oneshot::channel();
|
||||
Ok(Self {
|
||||
tx,
|
||||
rx: Some(rx),
|
||||
store,
|
||||
shutdown,
|
||||
shutdown_complete: None,
|
||||
shutdown_complete: Some(complete_tx),
|
||||
shutdown_receiver: Some(complete_rx),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -43,37 +46,65 @@ impl NotificationSystem {
|
||||
/// It initializes the event bus and the producer.
|
||||
pub async fn start(&mut self, adapters: Vec<Arc<dyn ChannelAdapter>>) -> Result<(), Error> {
|
||||
if self.shutdown.is_cancelled() {
|
||||
return Err(Error::custom("System is shutting down"));
|
||||
let error = Error::custom("System is shutting down");
|
||||
self.handle_error("start", &error);
|
||||
return Err(error);
|
||||
}
|
||||
|
||||
self.log(tracing::Level::INFO, "start", "Starting the notification system");
|
||||
let rx = self.rx.take().ok_or_else(|| Error::EventBusStarted)?;
|
||||
let shutdown_clone = self.shutdown.clone();
|
||||
let store_clone = self.store.clone();
|
||||
let shutdown_complete = self.shutdown_complete.take();
|
||||
|
||||
tokio::spawn(async move {
|
||||
if let Err(e) = event_bus(rx, adapters, store_clone, shutdown_clone, shutdown_complete).await {
|
||||
tracing::error!("Event bus failed: {}", e);
|
||||
}
|
||||
});
|
||||
|
||||
self.log(tracing::Level::INFO, "start", "Notification system started successfully");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sends an event to the notification system.
|
||||
/// This method is used to send events to the event bus.
|
||||
pub async fn send_event(&self, event: Event) -> Result<(), Error> {
|
||||
self.log(tracing::Level::DEBUG, "send_event", &format!("Sending event: {:?}", event));
|
||||
if self.shutdown.is_cancelled() {
|
||||
return Err(Error::custom("System is shutting down"));
|
||||
let error = Error::custom("System is shutting down");
|
||||
self.handle_error("send_event", &error);
|
||||
return Err(error);
|
||||
}
|
||||
self.tx.send(event).await.map_err(|e| Error::ChannelSend(Box::new(e)))?;
|
||||
if let Err(e) = self.tx.send(event).await {
|
||||
let error = Error::ChannelSend(Box::new(e));
|
||||
self.handle_error("send_event", &error);
|
||||
return Err(error);
|
||||
}
|
||||
self.log(tracing::Level::INFO, "send_event", "Event sent successfully");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Shuts down the notification system.
|
||||
/// This method is used to cancel the event bus and producer tasks.
|
||||
pub fn shutdown(&self) {
|
||||
pub async fn shutdown(&mut self) -> Result<(), Error> {
|
||||
tracing::info!("Shutting down the notification system");
|
||||
self.shutdown.cancel();
|
||||
// wait for the event bus to be completely closed
|
||||
if let Some(receiver) = self.shutdown_receiver.take() {
|
||||
match receiver.await {
|
||||
Ok(_) => {
|
||||
tracing::info!("Event bus shutdown completed successfully");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => {
|
||||
let error = Error::custom(format!("Failed to receive shutdown completion: {}", e).as_str());
|
||||
self.handle_error("shutdown", &error);
|
||||
Err(error)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
tracing::warn!("Shutdown receiver not available, the event bus might still be running");
|
||||
Err(Error::custom("Shutdown receiver not available"))
|
||||
}
|
||||
}
|
||||
|
||||
/// shutdown state
|
||||
@@ -81,9 +112,17 @@ impl NotificationSystem {
|
||||
self.shutdown.is_cancelled()
|
||||
}
|
||||
|
||||
pub fn set_shutdown_complete_channel(&mut self, tx: tokio::sync::oneshot::Sender<()>) {
|
||||
// storage completion channel for use by event bus
|
||||
tracing::info!("Shutting down the notification system set shutdown complete channel");
|
||||
self.shutdown_complete = Some(tx);
|
||||
fn handle_error(&self, context: &str, error: &Error) {
|
||||
self.log(tracing::Level::ERROR, context, &format!("{:?}", error));
|
||||
// TODO Can be extended to record to files or send to monitoring systems
|
||||
}
|
||||
fn log(&self, level: tracing::Level, context: &str, message: &str) {
|
||||
match level {
|
||||
tracing::Level::ERROR => tracing::error!("[{}] {}", context, message),
|
||||
tracing::Level::WARN => tracing::warn!("[{}] {}", context, message),
|
||||
tracing::Level::INFO => tracing::info!("[{}] {}", context, message),
|
||||
tracing::Level::DEBUG => tracing::debug!("[{}] {}", context, message),
|
||||
tracing::Level::TRACE => tracing::trace!("[{}] {}", context, message),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user