From 4cf2dd5e283f72b8059c1be34e74580448d3ea36 Mon Sep 17 00:00:00 2001 From: Wez Furlong Date: Thu, 4 May 2023 08:59:56 -0700 Subject: [PATCH] refactor: move DeliveryMetrics to its own file --- crates/kumod/src/delivery_metrics.rs | 102 +++++++++++++++++++++++++++ crates/kumod/src/egress_path.rs | 83 +--------------------- crates/kumod/src/lua_deliver.rs | 3 +- crates/kumod/src/main.rs | 1 + crates/kumod/src/queue.rs | 4 +- crates/kumod/src/ready_queue.rs | 33 +++------ crates/kumod/src/smtp_dispatcher.rs | 5 +- 7 files changed, 119 insertions(+), 112 deletions(-) create mode 100644 crates/kumod/src/delivery_metrics.rs diff --git a/crates/kumod/src/delivery_metrics.rs b/crates/kumod/src/delivery_metrics.rs new file mode 100644 index 00000000..3885dc77 --- /dev/null +++ b/crates/kumod/src/delivery_metrics.rs @@ -0,0 +1,102 @@ +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, + + pub ready_count: IntGauge, + + pub msgs_delivered: IntCounter, + pub global_msgs_delivered: IntCounter, + + pub msgs_transfail: IntCounter, + pub global_msgs_transfail: IntCounter, + + pub msgs_fail: IntCounter, + pub global_msgs_fail: IntCounter, +} + +impl DeliveryMetrics { + pub fn wrap_connection(&self, client: T) -> MetricsWrappedConnection { + self.connection_gauge.inc(); + self.global_connection_gauge.inc(); + self.connection_total.inc(); + self.global_connection_total.inc(); + MetricsWrappedConnection { + client, + metrics: self.clone(), + armed: true, + } + } + + pub fn new(service: &str, service_type: &str) -> Self { + DeliveryMetrics { + connection_gauge: crate::metrics_helper::connection_gauge_for_service(&service), + global_connection_gauge: crate::metrics_helper::connection_gauge_for_service( + service_type, + ), + connection_total: crate::metrics_helper::connection_total_for_service(&service), + global_connection_total: crate::metrics_helper::connection_total_for_service( + service_type, + ), + ready_count: crate::metrics_helper::ready_count_gauge_for_service(&service), + msgs_delivered: crate::metrics_helper::total_msgs_delivered_for_service(&service), + global_msgs_delivered: crate::metrics_helper::total_msgs_delivered_for_service( + service_type, + ), + msgs_transfail: crate::metrics_helper::total_msgs_transfail_for_service(&service), + global_msgs_transfail: crate::metrics_helper::total_msgs_transfail_for_service( + service_type, + ), + msgs_fail: crate::metrics_helper::total_msgs_fail_for_service(&service), + global_msgs_fail: crate::metrics_helper::total_msgs_fail_for_service(service_type), + } + } +} + +/// A helper struct to manage the number of connections. +/// It increments counters when created by DeliveryMetrics::wrap_connection +/// and decrements them when dropped +#[derive(Debug)] +pub struct MetricsWrappedConnection { + client: T, + metrics: DeliveryMetrics, + armed: bool, +} + +impl MetricsWrappedConnection { + /// Propagate the count from one type of connection to another + pub fn map_connection(mut self, client: O) -> MetricsWrappedConnection { + self.armed = false; + MetricsWrappedConnection { + client, + metrics: self.metrics.clone(), + armed: true, + } + } +} + +impl Drop for MetricsWrappedConnection { + fn drop(&mut self) { + if self.armed { + self.metrics.connection_gauge.dec(); + self.metrics.global_connection_gauge.dec(); + } + } +} + +impl std::ops::Deref for MetricsWrappedConnection { + type Target = T; + fn deref(&self) -> &T { + &self.client + } +} + +impl std::ops::DerefMut for MetricsWrappedConnection { + fn deref_mut(&mut self) -> &mut T { + &mut self.client + } +} diff --git a/crates/kumod/src/egress_path.rs b/crates/kumod/src/egress_path.rs index 9ec64259..7b61e62d 100644 --- a/crates/kumod/src/egress_path.rs +++ b/crates/kumod/src/egress_path.rs @@ -1,9 +1,6 @@ use cidr_map::{AnyIpCidr, CidrSet}; use mlua::prelude::*; -use prometheus::{IntCounter, IntGauge}; -use rfc5321::{ - SmtpClientTimeouts, -}; +use rfc5321::SmtpClientTimeouts; use serde::{Deserialize, Serialize}; use std::str::FromStr; use throttle::ThrottleSpec; @@ -130,84 +127,6 @@ impl EgressPathConfig { } } -#[derive(Clone, Debug)] -pub struct DeliveryMetrics { - pub connection_gauge: IntGauge, - pub global_connection_gauge: IntGauge, - pub connection_total: IntCounter, - pub global_connection_total: IntCounter, - - pub ready_count: IntGauge, - - pub msgs_delivered: IntCounter, - pub global_msgs_delivered: IntCounter, - - pub msgs_transfail: IntCounter, - pub global_msgs_transfail: IntCounter, - - pub msgs_fail: IntCounter, - pub global_msgs_fail: IntCounter, -} - -impl DeliveryMetrics { - pub fn wrap_connection(&self, client: T) -> MetricsWrappedConnection { - self.connection_gauge.inc(); - self.global_connection_gauge.inc(); - self.connection_total.inc(); - self.global_connection_total.inc(); - MetricsWrappedConnection { - client, - metrics: self.clone(), - armed: true, - } - } -} - -/// A helper struct to manage the number of connections. -/// It increments counters when created by DeliveryMetrics::wrap_connection -/// and decrements them when dropped -#[derive(Debug)] -pub struct MetricsWrappedConnection { - client: T, - metrics: DeliveryMetrics, - armed: bool, -} - -impl MetricsWrappedConnection { - /// Propagate the count from one type of connection to another - pub fn map_connection(mut self, client: O) -> MetricsWrappedConnection { - self.armed = false; - MetricsWrappedConnection { - client, - metrics: self.metrics.clone(), - armed: true, - } - } - -} - -impl Drop for MetricsWrappedConnection { - fn drop(&mut self) { - if self.armed { - self.metrics.connection_gauge.dec(); - self.metrics.global_connection_gauge.dec(); - } - } -} - -impl std::ops::Deref for MetricsWrappedConnection { - type Target = T; - fn deref(&self) -> &T { - &self.client - } -} - -impl std::ops::DerefMut for MetricsWrappedConnection { - fn deref_mut(&mut self) -> &mut T { - &mut self.client - } -} - /// Use an exponential decay curve in the increasing form, asymptotic up to connection_limit, /// passes through 0.0, increasing but bounded to connection_limit. /// diff --git a/crates/kumod/src/lua_deliver.rs b/crates/kumod/src/lua_deliver.rs index ee41bd7c..2aba8ba9 100644 --- a/crates/kumod/src/lua_deliver.rs +++ b/crates/kumod/src/lua_deliver.rs @@ -1,7 +1,6 @@ use crate::ready_queue::{Dispatcher, QueueDispatcher}; use async_trait::async_trait; -use config::load_config; -use config::LuaConfig; +use config::{load_config, LuaConfig}; use message::Message; use serde::{Deserialize, Serialize}; use serde_json::Value; diff --git a/crates/kumod/src/main.rs b/crates/kumod/src/main.rs index b230aec1..90372c60 100644 --- a/crates/kumod/src/main.rs +++ b/crates/kumod/src/main.rs @@ -36,6 +36,7 @@ static TRACING_FILTER_RELOAD_HANDLE: OnceCell< #[global_allocator] static GLOBAL: Jemalloc = Jemalloc; +mod delivery_metrics; mod egress_path; mod egress_source; mod http_server; diff --git a/crates/kumod/src/queue.rs b/crates/kumod/src/queue.rs index 222f28fc..f980541f 100644 --- a/crates/kumod/src/queue.rs +++ b/crates/kumod/src/queue.rs @@ -406,7 +406,9 @@ impl Queue { let sources = self.rr.all_sources(); for source in &sources { - if let Some(site) = ReadyQueueManager::get_opt(&self.name, &self.queue_config, source).await { + if let Some(site) = + ReadyQueueManager::get_opt(&self.name, &self.queue_config, source).await + { site.lock().await.bounce_all(bounce).await; } } diff --git a/crates/kumod/src/ready_queue.rs b/crates/kumod/src/ready_queue.rs index f729b940..2da1544b 100644 --- a/crates/kumod/src/ready_queue.rs +++ b/crates/kumod/src/ready_queue.rs @@ -1,4 +1,5 @@ -use crate::egress_path::{ideal_connection_count, DeliveryMetrics, EgressPathConfig}; +use crate::delivery_metrics::DeliveryMetrics; +use crate::egress_path::{ideal_connection_count, EgressPathConfig}; use crate::egress_source::EgressSource; use crate::http_server::admin_bounce_v1::AdminBounceEntry; use crate::lifecycle::{Activity, ShutdownSubcription}; @@ -45,9 +46,11 @@ impl ReadyQueueManager { MANAGER.lock().await } - pub async fn get_opt(queue_name: &str, - queue_config: &QueueConfig, - egress_source: &str) -> Option { + pub async fn get_opt( + queue_name: &str, + queue_config: &QueueConfig, + egress_source: &str, + ) -> Option { let components = QueueNameComponents::parse(queue_name); let needs_mx = matches!(&queue_config.protocol, DeliveryProto::Smtp); @@ -111,27 +114,7 @@ impl ReadyQueueManager { }) .expect("failed to spawn maintainer"); let service = format!("smtp_client:{name}"); - let metrics = DeliveryMetrics { - connection_gauge: crate::metrics_helper::connection_gauge_for_service(&service), - global_connection_gauge: crate::metrics_helper::connection_gauge_for_service( - "smtp_client", - ), - connection_total: crate::metrics_helper::connection_total_for_service(&service), - global_connection_total: crate::metrics_helper::connection_total_for_service( - "smtp_client", - ), - ready_count: crate::metrics_helper::ready_count_gauge_for_service(&service), - msgs_delivered: crate::metrics_helper::total_msgs_delivered_for_service(&service), - global_msgs_delivered: crate::metrics_helper::total_msgs_delivered_for_service( - "smtp_client", - ), - msgs_transfail: crate::metrics_helper::total_msgs_transfail_for_service(&service), - global_msgs_transfail: crate::metrics_helper::total_msgs_transfail_for_service( - "smtp_client", - ), - msgs_fail: crate::metrics_helper::total_msgs_fail_for_service(&service), - global_msgs_fail: crate::metrics_helper::total_msgs_fail_for_service("smtp_client"), - }; + let metrics = DeliveryMetrics::new(&service, "smtp_client"); let ready = Arc::new(StdMutex::new(VecDeque::new())); let notify = Arc::new(Notify::new()); ReadyQueueHandle(Arc::new(Mutex::new(ReadyQueue { diff --git a/crates/kumod/src/smtp_dispatcher.rs b/crates/kumod/src/smtp_dispatcher.rs index 910a6103..5644d3be 100644 --- a/crates/kumod/src/smtp_dispatcher.rs +++ b/crates/kumod/src/smtp_dispatcher.rs @@ -1,5 +1,6 @@ -use crate::egress_path::{MetricsWrappedConnection, Tls}; -use crate::lifecycle::{ShutdownSubcription}; +use crate::delivery_metrics::MetricsWrappedConnection; +use crate::egress_path::Tls; +use crate::lifecycle::ShutdownSubcription; use crate::logging::{log_disposition, LogDisposition, RecordType}; use crate::ready_queue::{Dispatcher, QueueDispatcher}; use crate::runtime::{rt_spawn, spawn};