diff --git a/crates/kumod/src/delivery_metrics.rs b/crates/kumod/src/delivery_metrics.rs index 301360ba..1a9f256f 100644 --- a/crates/kumod/src/delivery_metrics.rs +++ b/crates/kumod/src/delivery_metrics.rs @@ -2,21 +2,21 @@ use prometheus::{IntCounter, IntGauge}; #[derive(Clone, Debug)] pub struct DeliveryMetrics { - pub connection_gauge: IntGauge, - pub global_connection_gauge: IntGauge, - pub connection_total: IntCounter, - pub global_connection_total: IntCounter, + connection_gauge: IntGauge, + global_connection_gauge: IntGauge, + connection_total: IntCounter, + global_connection_total: IntCounter, pub ready_count: IntGauge, - pub msgs_delivered: IntCounter, - pub global_msgs_delivered: IntCounter, + msgs_delivered: IntCounter, + global_msgs_delivered: IntCounter, - pub msgs_transfail: IntCounter, - pub global_msgs_transfail: IntCounter, + msgs_transfail: IntCounter, + global_msgs_transfail: IntCounter, - pub msgs_fail: IntCounter, - pub global_msgs_fail: IntCounter, + msgs_fail: IntCounter, + global_msgs_fail: IntCounter, } impl DeliveryMetrics { @@ -55,6 +55,31 @@ impl DeliveryMetrics { global_msgs_fail: crate::metrics_helper::total_msgs_fail_for_service(service_type), } } + + pub fn inc_transfail(&self) { + self.msgs_transfail.inc(); + self.global_msgs_transfail.inc(); + } + + pub fn inc_transfail_by(&self, amount: usize) { + self.msgs_transfail.inc_by(amount as u64); + self.global_msgs_transfail.inc_by(amount as u64); + } + + pub fn inc_fail(&self) { + self.msgs_fail.inc(); + self.global_msgs_fail.inc(); + } + + pub fn inc_fail_by(&self, amount: usize) { + self.msgs_fail.inc_by(amount as u64); + self.global_msgs_fail.inc_by(amount as u64); + } + + pub fn inc_delivered(&self) { + self.msgs_delivered.inc(); + self.global_msgs_delivered.inc(); + } } /// A helper struct to manage the number of connections. diff --git a/crates/kumod/src/lua_deliver.rs b/crates/kumod/src/lua_deliver.rs index 278fea8e..2978742d 100644 --- a/crates/kumod/src/lua_deliver.rs +++ b/crates/kumod/src/lua_deliver.rs @@ -136,11 +136,9 @@ impl QueueDispatcher for LuaQueueDispatcher { }) .await?; } - dispatcher.metrics.msgs_transfail.inc(); - dispatcher.metrics.global_msgs_transfail.inc(); + dispatcher.metrics.inc_transfail(); } else { - dispatcher.metrics.msgs_fail.inc(); - dispatcher.metrics.global_msgs_fail.inc(); + dispatcher.metrics.inc_fail(); tracing::debug!( "failed to send message to {}: {response:?}", dispatcher.name, @@ -195,8 +193,7 @@ impl QueueDispatcher for LuaQueueDispatcher { SpoolManager::remove_from_spool(*msg.id()).await })?; } - dispatcher.metrics.msgs_delivered.inc(); - dispatcher.metrics.global_msgs_delivered.inc(); + dispatcher.metrics.inc_delivered(); } } diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index 7815ea9c..35e10351 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -516,8 +516,7 @@ impl Dispatcher { }) .await; Dispatcher::requeue_message(msg, true, None).await?; - dispatcher.metrics.msgs_transfail.inc(); - dispatcher.metrics.global_msgs_transfail.inc(); + dispatcher.metrics.inc_transfail(); } if consecutive_connection_failures.fetch_add(1, Ordering::SeqCst) @@ -725,11 +724,9 @@ impl Dispatcher { let egress_pool = self.egress_pool.clone(); let egress_source = self.egress_source.name.clone(); if response.is_transient() { - self.metrics.msgs_transfail.inc_by(msgs.len() as u64); - self.metrics.global_msgs_transfail.inc_by(msgs.len() as u64); + self.metrics.inc_transfail_by(msgs.len()); } else { - self.metrics.msgs_fail.inc_by(msgs.len() as u64); - self.metrics.global_msgs_fail.inc_by(msgs.len() as u64); + self.metrics.inc_fail_by(msgs.len()); } rt_spawn( format!("bulk queue op for {} msgs {name} {response:?}", msgs.len()), diff --git a/crates/kumod/src/smtp_dispatcher.rs b/crates/kumod/src/smtp_dispatcher.rs index 3934111c..d35a10f2 100644 --- a/crates/kumod/src/smtp_dispatcher.rs +++ b/crates/kumod/src/smtp_dispatcher.rs @@ -377,12 +377,10 @@ impl QueueDispatcher for SmtpDispatcher { }) .await?; } - dispatcher.metrics.msgs_transfail.inc(); - dispatcher.metrics.global_msgs_transfail.inc(); + dispatcher.metrics.inc_transfail(); } Err(ClientError::Rejected(response)) => { - dispatcher.metrics.msgs_fail.inc(); - dispatcher.metrics.global_msgs_fail.inc(); + dispatcher.metrics.inc_fail(); tracing::debug!( "failed to send message to {} {:?}: {response:?}", dispatcher.name, @@ -434,8 +432,7 @@ impl QueueDispatcher for SmtpDispatcher { SpoolManager::remove_from_spool(*msg.id()).await })?; } - dispatcher.metrics.msgs_delivered.inc(); - dispatcher.metrics.global_msgs_delivered.inc(); + dispatcher.metrics.inc_delivered(); } };