mirror/kumomta
1
0
Fork 0
mirror of https://github.com/mailscope/kumomta.git synced 2026-10-28 09:16:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: move DeliveryMetrics to its own file

Browse Source
This commit is contained in:
Wez Furlong
2023-05-04 13:51:37 -07:00
parent 28c3383169
commit 4cf2dd5e28
7 changed files with 119 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/kumod/src/delivery_metrics.rs
+102
View File
@@ -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<T>(&self, client: T) -> MetricsWrappedConnection<T> {
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<T> {
client: T,
metrics: DeliveryMetrics,
armed: bool,
}
impl<T> MetricsWrappedConnection<T> {
/// Propagate the count from one type of connection to another
pub fn map_connection<O>(mut self, client: O) -> MetricsWrappedConnection<O> {
self.armed = false;
MetricsWrappedConnection {
client,
metrics: self.metrics.clone(),
armed: true,
}
}
}
impl<T> Drop for MetricsWrappedConnection<T> {
fn drop(&mut self) {
if self.armed {
self.metrics.connection_gauge.dec();
self.metrics.global_connection_gauge.dec();
}
}
}
impl<T> std::ops::Deref for MetricsWrappedConnection<T> {
type Target = T;
fn deref(&self) -> &T {
&self.client
}
}
impl<T> std::ops::DerefMut for MetricsWrappedConnection<T> {
fn deref_mut(&mut self) -> &mut T {
&mut self.client
}
}
crates/kumod/src/egress_path.rs
+1 -82
View File
@@ -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<T>(&self, client: T) -> MetricsWrappedConnection<T> {
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<T> {
client: T,
metrics: DeliveryMetrics,
armed: bool,
}
impl<T> MetricsWrappedConnection<T> {
/// Propagate the count from one type of connection to another
pub fn map_connection<O>(mut self, client: O) -> MetricsWrappedConnection<O> {
self.armed = false;
MetricsWrappedConnection {
client,
metrics: self.metrics.clone(),
armed: true,
}
}
}
impl<T> Drop for MetricsWrappedConnection<T> {
fn drop(&mut self) {
if self.armed {
self.metrics.connection_gauge.dec();
self.metrics.global_connection_gauge.dec();
}
}
}
impl<T> std::ops::Deref for MetricsWrappedConnection<T> {
type Target = T;
fn deref(&self) -> &T {
&self.client
}
}
impl<T> std::ops::DerefMut for MetricsWrappedConnection<T> {
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.
///
crates/kumod/src/lua_deliver.rs
+1 -2
View File
@@ -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;
crates/kumod/src/main.rs
+1
View File
@@ -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;
crates/kumod/src/queue.rs
+3 -1
View File
@@ -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;
}
}
crates/kumod/src/ready_queue.rs
+8 -25
View File
@@ -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<ReadyQueueHandle> {
pub async fn get_opt(
queue_name: &str,
queue_config: &QueueConfig,
egress_source: &str,
) -> Option<ReadyQueueHandle> {
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 {
crates/kumod/src/smtp_dispatcher.rs
+3 -2
View File
@@ -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};