refactor(obs): migrate metrics runtime/schema and tighten migration guards (#2584)

crates/notify/src/global.rs

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationSystem};
+use crate::{
+    BucketNotificationConfig, Event, EventArgs, LifecycleError, NotificationError, NotificationMetricSnapshot,
+    NotificationSystem, NotificationTargetMetricSnapshot,
+};
 use rustfs_ecstore::config::Config;
 use rustfs_s3_common::EventName;
 use rustfs_targets::arn::TargetID;
@@ -41,6 +44,23 @@ pub fn notification_system() -> Option<Arc<NotificationSystem>> {
     NOTIFICATION_SYSTEM.get().cloned()
 }
 
+/// Returns aggregate notification delivery metrics for Prometheus collection.
+pub fn notification_metrics_snapshot() -> NotificationMetricSnapshot {
+    NOTIFICATION_SYSTEM
+        .get()
+        .map(|system| system.snapshot_metrics())
+        .unwrap_or_default()
+}
+
+/// Returns per-target notification delivery metrics for Prometheus collection.
+pub async fn notification_target_metrics() -> Vec<NotificationTargetMetricSnapshot> {
+    if let Some(system) = notification_system() {
+        system.snapshot_target_metrics().await
+    } else {
+        Vec::new()
+    }
+}
+
 /// Check if the notification system has been initialized.
 pub fn is_notification_system_initialized() -> bool {
     NOTIFICATION_SYSTEM.get().is_some()

crates/notify/src/integration.rs

@@ -106,10 +106,29 @@ pub struct NotificationMetrics {
     processed_events: AtomicUsize,
     /// Number of events that failed to handle
     failed_events: AtomicUsize,
+    /// Number of dispatch attempts skipped before delivery
+    skipped_events: AtomicUsize,
     /// System startup time
     start_time: Instant,
 }
 
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct NotificationMetricSnapshot {
+    pub current_send_in_progress: u64,
+    pub events_errors_total: u64,
+    pub events_sent_total: u64,
+    pub events_skipped_total: u64,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct NotificationTargetMetricSnapshot {
+    pub failed_messages: u64,
+    pub queue_length: u64,
+    pub target_id: String,
+    pub target_type: String,
+    pub total_messages: u64,
+}
+
 impl Default for NotificationMetrics {
     fn default() -> Self {
         Self::new()
@@ -122,6 +141,7 @@ impl NotificationMetrics {
             processing_events: AtomicUsize::new(0),
             processed_events: AtomicUsize::new(0),
             failed_events: AtomicUsize::new(0),
+            skipped_events: AtomicUsize::new(0),
             start_time: Instant::now(),
         }
     }
@@ -136,11 +156,19 @@ impl NotificationMetrics {
         self.processed_events.fetch_add(1, Ordering::Relaxed);
     }
 
+    pub fn decrement_processing(&self) {
+        self.processing_events.fetch_sub(1, Ordering::Relaxed);
+    }
+
     pub fn increment_failed(&self) {
         self.processing_events.fetch_sub(1, Ordering::Relaxed);
         self.failed_events.fetch_add(1, Ordering::Relaxed);
     }
 
+    pub fn increment_skipped(&self) {
+        self.skipped_events.fetch_add(1, Ordering::Relaxed);
+    }
+
     // Provide public methods to get count
     pub fn processing_count(&self) -> usize {
         self.processing_events.load(Ordering::Relaxed)
@@ -154,9 +182,22 @@ impl NotificationMetrics {
         self.failed_events.load(Ordering::Relaxed)
     }
 
+    pub fn skipped_count(&self) -> usize {
+        self.skipped_events.load(Ordering::Relaxed)
+    }
+
     pub fn uptime(&self) -> Duration {
         self.start_time.elapsed()
     }
+
+    pub fn snapshot(&self) -> NotificationMetricSnapshot {
+        NotificationMetricSnapshot {
+            current_send_in_progress: self.processing_count() as u64,
+            events_errors_total: self.failed_count() as u64,
+            events_sent_total: self.processed_count() as u64,
+            events_skipped_total: self.skipped_count() as u64,
+        }
+    }
 }
 
 /// The notification system that integrates all components
@@ -187,14 +228,15 @@ impl NotificationSystem {
         let concurrency_limiter =
             rustfs_utils::get_env_usize(ENV_NOTIFY_TARGET_STREAM_CONCURRENCY, DEFAULT_NOTIFY_TARGET_STREAM_CONCURRENCY);
         let (live_event_sender, _) = broadcast::channel(1024);
+        let metrics = Arc::new(NotificationMetrics::new());
         NotificationSystem {
             subscriber_view: NotificationSystemSubscriberView::new(),
-            notifier: Arc::new(EventNotifier::new()),
+            notifier: Arc::new(EventNotifier::new(metrics.clone())),
             registry: Arc::new(TargetRegistry::new()),
             config: Arc::new(RwLock::new(config)),
             stream_cancellers: Arc::new(RwLock::new(HashMap::new())),
             concurrency_limiter: Arc::new(Semaphore::new(concurrency_limiter)), // Limit the maximum number of concurrent processing events to 20
-            metrics: Arc::new(NotificationMetrics::new()),
+            metrics,
             live_event_sender,
             live_event_history: Arc::new(RwLock::new(LiveEventHistory::default())),
         }
@@ -585,10 +627,35 @@ impl NotificationSystem {
         status.insert("processing_events".to_string(), self.metrics.processing_count().to_string());
         status.insert("processed_events".to_string(), self.metrics.processed_count().to_string());
         status.insert("failed_events".to_string(), self.metrics.failed_count().to_string());
+        status.insert("skipped_events".to_string(), self.metrics.skipped_count().to_string());
 
         status
     }
 
+    pub fn snapshot_metrics(&self) -> NotificationMetricSnapshot {
+        self.metrics.snapshot()
+    }
+
+    pub async fn snapshot_target_metrics(&self) -> Vec<NotificationTargetMetricSnapshot> {
+        let targets = self.notifier.target_list().read().await.values();
+        let mut snapshots = Vec::with_capacity(targets.len());
+
+        for target in targets {
+            let delivery = target.delivery_snapshot();
+            let target_id = target.id();
+            snapshots.push(NotificationTargetMetricSnapshot {
+                failed_messages: delivery.failed_messages,
+                queue_length: delivery.queue_length,
+                target_id: target_id.to_string(),
+                target_type: target_id.name,
+                total_messages: delivery.total_messages,
+            });
+        }
+
+        snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id));
+        snapshots
+    }
+
     // Add a method to shut down the system
     pub async fn shutdown(&self) {
         info!("Turn off the notification system");

crates/notify/src/lib.rs

@@ -31,6 +31,9 @@ pub mod stream;
 
 pub use error::{LifecycleError, NotificationError};
 pub use event::{Event, EventArgs, EventArgsBuilder};
-pub use global::{initialize, is_notification_system_initialized, notification_system, notifier_global};
-pub use integration::NotificationSystem;
+pub use global::{
+    initialize, is_notification_system_initialized, notification_metrics_snapshot, notification_system,
+    notification_target_metrics, notifier_global,
+};
+pub use integration::{NotificationMetricSnapshot, NotificationSystem, NotificationTargetMetricSnapshot};
 pub use rules::BucketNotificationConfig;

crates/notify/src/notifier.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::{error::NotificationError, event::Event, rules::RulesMap};
+use crate::{error::NotificationError, event::Event, integration::NotificationMetrics, rules::RulesMap};
 use hashbrown::HashMap;
 use rustfs_config::notify::{DEFAULT_NOTIFY_SEND_CONCURRENCY, ENV_NOTIFY_SEND_CONCURRENCY};
 use rustfs_s3_common::EventName;
@@ -26,6 +26,7 @@ use tracing::{debug, error, info, instrument, warn};
 
 /// Manages event notification to targets based on rules
 pub struct EventNotifier {
+    metrics: Arc<NotificationMetrics>,
     target_list: Arc<RwLock<TargetList>>,
     bucket_rules_map: Arc<AsyncShardedHashMap<String, RulesMap, rustc_hash::FxBuildHasher>>,
     send_limiter: Arc<Semaphore>,
@@ -33,7 +34,7 @@ pub struct EventNotifier {
 
 impl Default for EventNotifier {
     fn default() -> Self {
-        Self::new()
+        Self::new(Arc::new(NotificationMetrics::new()))
     }
 }
 
@@ -42,9 +43,10 @@ impl EventNotifier {
     ///
     /// # Returns
     /// Returns a new instance of EventNotifier.
-    pub fn new() -> Self {
+    pub fn new(metrics: Arc<NotificationMetrics>) -> Self {
         let max_inflight = rustfs_utils::get_env_usize(ENV_NOTIFY_SEND_CONCURRENCY, DEFAULT_NOTIFY_SEND_CONCURRENCY);
         EventNotifier {
+            metrics,
             target_list: Arc::new(RwLock::new(TargetList::new())),
             bucket_rules_map: Arc::new(AsyncShardedHashMap::new(0)),
             send_limiter: Arc::new(Semaphore::new(max_inflight)),
@@ -182,12 +184,14 @@ impl EventNotifier {
 
         let Some(rules) = self.bucket_rules_map.get(bucket_name).await else {
             debug!("No rules found for bucket: {}", bucket_name);
+            self.metrics.increment_skipped();
             return;
         };
 
         let target_ids = rules.match_rules(event_name, object_key);
         if target_ids.is_empty() {
             debug!("No matching targets for event in bucket: {}", bucket_name);
+            self.metrics.increment_skipped();
             return;
         }
         let target_ids_len = target_ids.len();
@@ -207,7 +211,9 @@ impl EventNotifier {
                     continue;
                 }
                 let limiter = self.send_limiter.clone();
+                let metrics = self.metrics.clone();
                 let event_clone = event.clone();
+                let is_deferred = target_for_task.store().is_some();
                 let target_name_for_task = target_for_task.name(); // Get the name before generating the task
                 debug!("Preparing to send event to target: {}", target_name_for_task);
                 // Use cloned data in closures to avoid borrowing conflicts
@@ -219,22 +225,31 @@ impl EventNotifier {
                     data: event_clone.as_ref().clone(),
                 });
                 let handle = tokio::spawn(async move {
+                    metrics.increment_processing();
                     let _permit = match limiter.acquire_owned().await {
                         Ok(p) => p,
                         Err(e) => {
                             error!("Failed to acquire send permit for target {}: {}", target_name_for_task, e);
+                            metrics.increment_failed();
                             return;
                         }
                     };
                     if let Err(e) = target_for_task.save(entity_target.clone()).await {
+                        metrics.increment_failed();
                         error!("Failed to send event to target {}: {}", target_name_for_task, e);
                     } else {
+                        if is_deferred {
+                            metrics.decrement_processing();
+                        } else {
+                            metrics.increment_processed();
+                        }
                         debug!("Successfully saved event to target {}", target_name_for_task);
                     }
                 });
                 handles.push(handle);
             } else {
                 warn!("Target ID {:?} found in rules but not in target list.", target_id);
+                self.metrics.increment_skipped();
             }
         }
         // target_list is automatically released here
@@ -470,7 +485,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_send_event_skips_disabled_target() {
-        let notifier = EventNotifier::new();
+        let notifier = EventNotifier::new(Arc::new(NotificationMetrics::new()));
 
         let enabled_target = TestTarget::new("enabled-target", "webhook", true);
         let disabled_target = TestTarget::new("disabled-target", "webhook", false);

crates/notify/src/stream.rs

@@ -311,8 +311,14 @@ async fn process_batch(
                         let backoff = 1u32 << retry_count as u32;
                         tokio::time::sleep(base_delay * backoff + jitter).await;
                     }
+                    TargetError::Dropped(reason) => {
+                        warn!("Dropped queued payload for target {}: {}", target.name(), reason);
+                        metrics.increment_failed();
+                        break;
+                    }
                     _ => {
                         error!("Permanent error for target {}: {}", target.name(), e);
+                        target.record_final_failure();
                         metrics.increment_failed();
                         break;
                     }
@@ -322,6 +328,7 @@ async fn process_batch(
 
         if retry_count >= max_retries && !success {
             warn!("Max retries exceeded for event {}, target: {}, skipping", key.to_string(), target.name());
+            target.record_final_failure();
             metrics.increment_failed();
         }
     }