delivery_metrics: DRY and encapsulate better

Make it more difficult to misuse and less boilerplate to use
This commit is contained in:
Wez Furlong
2023-06-24 07:17:07 -07:00
parent 7d4cc52d8d
commit 6514eaa6df
4 changed files with 44 additions and 28 deletions
+35 -10
View File
@@ -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.
+3 -6
View File
@@ -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();
}
}
+3 -6
View File
@@ -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()),
+3 -6
View File
@@ -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();
}
};