filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Live and historical metrics

Browse Source
This commit is contained in:
mdecimus
2024-08-27 17:54:34 +02:00
parent 18a24f7220
commit 62f55ad62b
15 changed files with 346 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/jmap/src/api/http.rs
View File

@@ -302,10 +302,10 @@ impl JMAP {
.path()
.strip_prefix("/api/telemetry/")
.and_then(|p| {
p.strip_suffix("traces/live/")
p.strip_prefix("traces/live/")
.map(|t| ("traces", t))
.or_else(|| {
p.strip_suffix("metrics/live/")
p.strip_prefix("metrics/live/")
.map(|t| ("metrics", t))
})
})

30
crates/jmap/src/api/management/enterprise/telemetry.rs
View File

@@ -263,7 +263,7 @@ impl JMAP {
if elapsed >= throttle {
last_message = Instant::now();
yield Ok(Frame::data(Bytes::from(format!(
"event: state\ndata: {}\n\n",
"event: trace\ndata: {}\n\n",
serde_json::to_string(
&JsonEventSerializer::new(std::mem::take(&mut events))
.with_description()
@@ -339,7 +339,7 @@ impl JMAP {
let before = params
.parse::<Timestamp>("before")
.map(|t| t.into_inner())
.unwrap_or(0);
.unwrap_or(u64::MAX);
let after = params
.parse::<Timestamp>("after")
.map(|t| t.into_inner())
@@ -377,6 +377,15 @@ impl JMAP {
count,
sum,
},
Metric::Gauge {
id,
timestamp,
value,
} => Metric::Gauge {
id: id.name(),
timestamp: DateTime::from_timestamp(timestamp as i64).to_rfc3339(),
value,
},
});
}
@@ -405,6 +414,23 @@ impl JMAP {
}
}
// Refresh expensive metrics
for metric_type in [
MetricType::QueueCount,
MetricType::UserCount,
MetricType::DomainCount,
] {
if metric_types.contains(&metric_type) {
let value = match metric_type {
MetricType::QueueCount => self.core.total_queued_messages().await?,
MetricType::UserCount => self.core.total_accounts().await?,
MetricType::DomainCount => self.core.total_domains().await?,
_ => unreachable!(),
};
Collector::update_gauge(metric_type, value);
}
}
Ok(HttpResponse {
status: StatusCode::OK,
content_type: "text/event-stream".into(),

21
crates/jmap/src/api/management/mod.rs
View File

@@ -115,16 +115,29 @@ pub fn decode_path_element(item: &str) -> Cow<'_, str> {
.unwrap_or_else(|| item.into())
}
pub(super) struct FutureTimestamp(u64);
pub(super) struct Timestamp(u64);
impl FromStr for Timestamp {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
if let Some(dt) = DateTime::parse_rfc3339(s) {
Ok(Timestamp(dt.to_timestamp() as u64))
} else {
Err(())
}
}
}
impl FromStr for FutureTimestamp {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
if let Some(dt) = DateTime::parse_rfc3339(s) {
let instant = dt.to_timestamp() as u64;
if instant >= now() {
return Ok(Timestamp(instant));
return Ok(FutureTimestamp(instant));
}
}
@@ -132,6 +145,12 @@ impl FromStr for Timestamp {
}
}
impl FutureTimestamp {
pub fn into_inner(self) -> u64 {
self.0
}
}
impl Timestamp {
pub fn into_inner(self) -> u64 {
self.0

12
crates/jmap/src/api/management/queue.rs
View File

@@ -26,7 +26,7 @@ use crate::{
JMAP,
};
use super::{decode_path_element, Timestamp};
use super::{decode_path_element, FutureTimestamp};
#[derive(Debug, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
pub struct Message {
@@ -117,8 +117,12 @@ impl JMAP {
let text = params.get("text");
let from = params.get("from");
let to = params.get("to");
let before = params.parse::<Timestamp>("before").map(|t| t.into_inner());
let after = params.parse::<Timestamp>("after").map(|t| t.into_inner());
let before = params
.parse::<FutureTimestamp>("before")
.map(|t| t.into_inner());
let after = params
.parse::<FutureTimestamp>("after")
.map(|t| t.into_inner());
let page = params.parse::<usize>("page").unwrap_or_default();
let limit = params.parse::<usize>("limit").unwrap_or_default();
let values = params.has_key("values");
@@ -228,7 +232,7 @@ impl JMAP {
}
("messages", Some(queue_id), &Method::PATCH) => {
let time = params
.parse::<Timestamp>("at")
.parse::<FutureTimestamp>("at")
.map(|t| t.into_inner())
.unwrap_or_else(now);
let item = params.get("filter");

46
crates/jmap/src/services/housekeeper.rs
View File

@@ -148,6 +148,7 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver<Event>) {
// Metrics history
#[cfg(feature = "enterprise")]
let metrics_history = SharedMetricHistory::default();
let mut next_metric_update = Instant::now();
loop {
match tokio::time::timeout(queue.wake_up_time(), rx.recv()).await {
@@ -425,12 +426,20 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver<Event>) {
ActionClass::OtelMetrics,
);
let update_other_metrics = if Instant::now() >= next_metric_update {
next_metric_update =
Instant::now() + Duration::from_secs(86400);
true
} else {
false
};
let core = core_.clone();
tokio::spawn(async move {
#[cfg(feature = "enterprise")]
if core.is_enterprise_edition() {
// Obtain queue size
match core.message_queue_size().await {
match core.total_queued_messages().await {
Ok(total) => {
Collector::update_gauge(
MetricType::QueueCount,
@@ -445,6 +454,36 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver<Event>) {
}
}
if update_other_metrics {
match core.total_accounts().await {
Ok(total) => {
Collector::update_gauge(
MetricType::UserCount,
total,
);
}
Err(err) => {
trc::error!(
err.details("Failed to obtain account count")
);
}
}
match core.total_domains().await {
Ok(total) => {
Collector::update_gauge(
MetricType::DomainCount,
total,
);
}
Err(err) => {
trc::error!(
err.details("Failed to obtain domain count")
);
}
}
}
match tokio::task::spawn_blocking(memory_stats::memory_stats)
.await
{
@@ -486,8 +525,9 @@ pub fn spawn_housekeeper(core: JmapInstance, mut rx: mpsc::Receiver<Event>) {
let metrics_history = metrics_history.clone();
let core = core_.clone();
tokio::spawn(async move {
if let Err(err) =
metrics_store.write_metrics(core, metrics_history).await
if let Err(err) = metrics_store
.write_metrics(core, now(), metrics_history)
.await
{
trc::error!(err.details("Failed to write metrics"));
}