From 9f7035588f696fed7eb4ad1d49bcdf7015c5ac4f Mon Sep 17 00:00:00 2001 From: John Coffey Date: Sat, 19 Sep 2026 08:21:45 -0700 Subject: [PATCH] Monitoring: metric history, one edition of metrics, and x:Metric over the stored samples (MON-4 to MON-7, MON-9, MON-17, MON-39) Every node writes a sample per metric on metricsCollectionInterval: counters as the increase since its last sample, gauges always, histograms as totals when changed. Samples are x:Metric in the registry's encoding under the telemetry key class, ids time-ordered. x:Metric/get and /query read them with metric and timestamp filters and full paging, hide what's past holdMetricsFor, and the data purge deletes it. The is_enterprise split is gone, so every gauge and histogram is collected and exported, and queue.count is set from the queue itself. The shared metrics suite runs. --- crates/common/src/cache/reload.rs | 2 +- crates/common/src/manager/boot.rs | 6 +- crates/common/src/telemetry/metrics/mod.rs | 1 + crates/common/src/telemetry/metrics/otel.rs | 8 +- .../src/telemetry/metrics/prometheus.rs | 7 +- crates/common/src/telemetry/metrics/store.rs | 267 ++++++++++++++++++ crates/common/src/telemetry/mod.rs | 9 +- crates/jmap/src/inbuxa/mod.rs | 1 + crates/jmap/src/inbuxa/telemetry.rs | 214 ++++++++++++++ crates/jmap/src/registry/get.rs | 4 + crates/jmap/src/registry/mod.rs | 22 +- crates/jmap/src/registry/query.rs | 9 + .../services/src/task_manager/maintenance.rs | 14 + crates/services/src/task_manager/scheduler.rs | 47 ++- crates/trc/src/ipc/metrics.rs | 29 +- crates/utils/src/snowflake.rs | 6 + tests/src/telemetry/mod.rs | 2 - tests/src/utils/server.rs | 2 +- 18 files changed, 586 insertions(+), 64 deletions(-) create mode 100644 crates/common/src/telemetry/metrics/store.rs create mode 100644 crates/jmap/src/inbuxa/telemetry.rs diff --git a/crates/common/src/cache/reload.rs b/crates/common/src/cache/reload.rs index 7478877..2ec444b 100644 --- a/crates/common/src/cache/reload.rs +++ b/crates/common/src/cache/reload.rs @@ -131,7 +131,7 @@ impl Server { // Update tracers #[cfg(not(feature = "enterprise"))] - tracers.update(false); + tracers.update(); // Reload queue settings self.inner diff --git a/crates/common/src/manager/boot.rs b/crates/common/src/manager/boot.rs index 5f1f5de..ae516b6 100644 --- a/crates/common/src/manager/boot.rs +++ b/crates/common/src/manager/boot.rs @@ -158,7 +158,7 @@ impl BootManager { #[cfg(not(feature = "enterprise"))] - telemetry.enable(false); + telemetry.enable(); if bootstrap.registry.is_bootstrap_mode() { trc::event!( @@ -241,7 +241,7 @@ impl BootManager { } StoreOp::Export(path) => { // Enable telemetry - telemetry.enable(false); + telemetry.enable(); // Parse settings and backup Box::pin(Core::parse(&mut bootstrap, storage)) @@ -252,7 +252,7 @@ impl BootManager { } StoreOp::Import(path) => { // Enable telemetry - telemetry.enable(false); + telemetry.enable(); // Parse settings and restore Box::pin(Core::parse(&mut bootstrap, storage)) diff --git a/crates/common/src/telemetry/metrics/mod.rs b/crates/common/src/telemetry/metrics/mod.rs index 831aa42..dfe9880 100644 --- a/crates/common/src/telemetry/metrics/mod.rs +++ b/crates/common/src/telemetry/metrics/mod.rs @@ -6,5 +6,6 @@ pub mod otel; pub mod prometheus; +pub mod store; // inbuxa: monitoring history (MON-4 to MON-9) diff --git a/crates/common/src/telemetry/metrics/otel.rs b/crates/common/src/telemetry/metrics/otel.rs index e7dd9ef..07558c3 100644 --- a/crates/common/src/telemetry/metrics/otel.rs +++ b/crates/common/src/telemetry/metrics/otel.rs @@ -17,12 +17,12 @@ use std::time::SystemTime; use trc::{Collector, TelemetryEvent}; impl OtelMetrics { - pub async fn push_metrics(&self, is_enterprise: bool, start_time: SystemTime) { + pub async fn push_metrics(&self, start_time: SystemTime) { let mut metrics = Vec::with_capacity(256); let time = SystemTime::now(); // Add counters - for counter in Collector::collect_counters(is_enterprise) { + for counter in Collector::collect_counters() { metrics.push(Metric::new( counter.id().as_str(), counter.id().description(), @@ -38,7 +38,7 @@ impl OtelMetrics { } // Add gauges - for gauge in Collector::collect_gauges(is_enterprise) { + for gauge in Collector::collect_gauges() { metrics.push(Metric::new( gauge.id().as_str(), gauge.id().description(), @@ -52,7 +52,7 @@ impl OtelMetrics { } // Add histograms - for histogram in Collector::collect_histograms(is_enterprise) { + for histogram in Collector::collect_histograms() { metrics.push(Metric::new( histogram.id().as_str(), histogram.id().description(), diff --git a/crates/common/src/telemetry/metrics/prometheus.rs b/crates/common/src/telemetry/metrics/prometheus.rs index cea5bfd..1e859be 100644 --- a/crates/common/src/telemetry/metrics/prometheus.rs +++ b/crates/common/src/telemetry/metrics/prometheus.rs @@ -18,10 +18,9 @@ impl Server { #[cfg(not(feature = "enterprise"))] - let is_enterprise = false; // Add counters - for counter in Collector::collect_counters(is_enterprise) { + for counter in Collector::collect_counters() { let mut metric = MetricFamily::default(); metric.set_name(metric_name(counter.id().as_str())); metric.set_help(counter.id().description().into()); @@ -31,7 +30,7 @@ impl Server { } // Add gauges - for gauge in Collector::collect_gauges(is_enterprise) { + for gauge in Collector::collect_gauges() { let mut metric = MetricFamily::default(); metric.set_name(metric_name(gauge.id().as_str())); metric.set_help(gauge.id().description().into()); @@ -41,7 +40,7 @@ impl Server { } // Add histograms - for histogram in Collector::collect_histograms(is_enterprise) { + for histogram in Collector::collect_histograms() { let mut metric = MetricFamily::default(); metric.set_name(metric_name(histogram.id().as_str())); metric.set_help(histogram.id().description().into()); diff --git a/crates/common/src/telemetry/metrics/store.rs b/crates/common/src/telemetry/metrics/store.rs new file mode 100644 index 0000000..d3249c2 --- /dev/null +++ b/crates/common/src/telemetry/metrics/store.rs @@ -0,0 +1,267 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! Metric history (monitoring spec MON-4 to MON-9, MON-17). Each sample is +//! stored under `TelemetryClass::Metric(id)`, as an `x:Metric` in the +//! registry's own encoding. The id is a snowflake of the tick's time, so key +//! order is time order and the timestamp is read from the id. + +use crate::Server; +use ahash::AHashMap; +use registry::{ + pickle::PickledStream, + schema::{ + prelude::{ObjectInner, ObjectType}, + structs::{DataRetention, Metric, MetricCount, MetricSum}, + }, +}; +use std::{future::Future, sync::Mutex, time::Duration}; +use store::{ + IterateParams, Store, ValueKey, + write::{BatchBuilder, TelemetryClass, ValueClass, key::DeserializeBigEndian, now}, +}; +use trc::{AddContext, Collector, MetricType, TelemetryEvent}; +use types::id::Id; +use utils::snowflake::SnowflakeIdGenerator; + +pub trait MetricsStore: Sync + Send { + /// Writes one tick's samples, all at `timestamp`. + fn write_metrics( + &self, + samples: Vec, + timestamp: u64, + ) -> impl Future> + Send; + + /// Deletes samples older than `keep` (MON-17). + fn purge_metrics(&self, keep: Duration) -> impl Future> + Send; +} + +impl MetricsStore for Store { + async fn write_metrics(&self, samples: Vec, timestamp: u64) -> trc::Result<()> { + let mut batch = BatchBuilder::new(); + for sample in samples { + let Some(id) = SnowflakeIdGenerator::global_id_from_timestamp(timestamp) else { + continue; + }; + batch.set( + ValueClass::Telemetry(TelemetryClass::Metric(id)), + ObjectInner::Metric(sample).to_pickled_vec(), + ); + if batch.is_large_batch() { + self.write(batch.build_all()).await?; + batch = BatchBuilder::new(); + } + } + if !batch.is_empty() { + self.write(batch.build_all()).await?; + } + Ok(()) + } + + async fn purge_metrics(&self, keep: Duration) -> trc::Result<()> { + let Some(until) = SnowflakeIdGenerator::from_duration(keep) else { + return Ok(()); + }; + self.delete_range( + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(0))), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(until))), + ) + .await + .caused_by(trc::location!()) + } +} + +/// Decodes a stored sample. Records in any other encoding (INBUXA's history +/// from before the fork) read as `None` and are skipped. +pub fn decode_metric(bytes: &[u8]) -> Option { + PickledStream::new(bytes) + .and_then(|mut stream| ObjectInner::unpickle(ObjectType::Metric, &mut stream)) + .and_then(|inner| match inner { + ObjectInner::Metric(metric) => Some(metric), + _ => None, + }) +} + +/// A stored sample as read by key: `None` when it can't be decoded. +pub struct MaybeMetric(pub Option); + +impl store::Deserialize for MaybeMetric { + fn deserialize(bytes: &[u8]) -> trc::Result { + Ok(MaybeMetric(decode_metric(bytes))) + } +} + +/// A stored sample with its id. +pub struct StoredMetric { + pub id: u64, + pub metric: Metric, +} + +impl StoredMetric { + pub fn timestamp(&self) -> u64 { + SnowflakeIdGenerator::to_timestamp(self.id) + } +} + +/// What the node wrote last, so counters and histograms are written as +/// changes (MON-4). Per process: a restart counts from the start. +static LAST: Mutex>> = Mutex::new(None); + +/// One tick's samples (MON-4 to MON-6). +pub fn sample() -> Vec { + let mut last_guard = LAST.lock().unwrap(); + let last = last_guard.get_or_insert_with(AHashMap::new); + let mut samples = Vec::new(); + + // Counters: the increase since the previous sample; none if unchanged + for counter in Collector::collect_counters() { + let Some(metric) = MetricType::parse(counter.id().as_str()) else { + continue; + }; + let total = counter.value(); + let previous = last.insert(metric, (total, 0)).map_or(0, |(count, _)| count); + let increase = total.saturating_sub(previous); + if increase > 0 { + samples.push(Metric::Counter(MetricCount { + count: increase, + metric, + })); + } + } + + // Gauges: the reading, always (MON-5) + for gauge in Collector::collect_gauges() { + samples.push(Metric::Gauge(MetricCount { + count: gauge.get(), + metric: gauge.id(), + })); + } + + // Histograms: totals, when changed (MON-4 Decision) + for histogram in Collector::collect_histograms() { + let metric = histogram.id(); + let current = (histogram.count(), histogram.sum()); + if last.insert(metric, current) != Some(current) { + samples.push(Metric::Histogram(MetricSum { + count: current.0, + sum: current.1, + metric, + })); + } + } + + samples +} + +/// The retention settings in force now (MON-3 Decision: no reload needed). +pub async fn retention(server: &Server) -> DataRetention { + server + .registry() + .object::(Id::singleton()) + .await + .ok() + .flatten() + .unwrap_or_default() +} + +impl Server { + /// Writes one tick of metric history, if it's on (MON-4, MON-9). Never + /// fails loudly: history is lost, mail isn't (MON-35). + pub async fn store_metrics(&self) { + let store = self.metrics_store(); + if store.is_none() { + return; + } + let samples = sample(); + let count = samples.len(); + let started = std::time::Instant::now(); + match store.write_metrics(samples, now()).await { + Ok(()) => trc::event!( + Telemetry(TelemetryEvent::MetricsStored), + Total = count, + Elapsed = started.elapsed(), + ), + Err(err) => { + trc::error!(err.details("Failed to store metric history")); + } + } + } + + /// The stored samples between two ids, in key order, skipping any that + /// can't be decoded or are past `holdMetricsFor` (MON-17). + pub async fn read_metrics( + &self, + from_id: u64, + to_id: u64, + ascending: bool, + mut accept: impl FnMut(&StoredMetric) -> bool + Send + Sync, + ) -> trc::Result> { + let store = self.metrics_store(); + let mut out = Vec::new(); + if store.is_none() { + return Ok(out); + } + let floor = match retention(self).await.hold_metrics_for { + Some(keep) => SnowflakeIdGenerator::from_duration(keep.into_inner()).unwrap_or(0), + None => 0, + }; + let from_id = from_id.max(floor); + if from_id > to_id { + return Ok(out); + } + let params = IterateParams::new( + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(from_id))), + ValueKey::from(ValueClass::Telemetry(TelemetryClass::Metric(to_id))), + ); + let params = if ascending { + params.ascending() + } else { + params.descending() + }; + store + .iterate(params, |key, value| { + let id = key.deserialize_be_u64(0)?; + if let Some(metric) = decode_metric(value) { + let sample = StoredMetric { id, metric }; + if accept(&sample) { + out.push(sample); + } + } + Ok(true) + }) + .await + .caused_by(trc::location!())?; + Ok(out) + } + + /// Test data for the shared metrics suite: 90 days of hourly ticks, a + /// counter, a gauge and a histogram each. + #[cfg(feature = "test_mode")] + pub async fn insert_test_metrics(&self) { + let now = now(); + for hour in (0..90 * 24u64).rev() { + let samples = vec![ + Metric::Counter(MetricCount { + count: 1 + hour % 7, + metric: MetricType::AuthSuccess, + }), + Metric::Gauge(MetricCount { + count: 20 + hour % 11, + metric: MetricType::QueueCount, + }), + Metric::Histogram(MetricSum { + count: 100 + hour, + sum: 1000 + hour * 10, + metric: MetricType::DeliveryTotalTime, + }), + ]; + self.metrics_store() + .write_metrics(samples, now - hour * 3600) + .await + .unwrap(); + } + } +} diff --git a/crates/common/src/telemetry/mod.rs b/crates/common/src/telemetry/mod.rs index e8b9fde..8052163 100644 --- a/crates/common/src/telemetry/mod.rs +++ b/crates/common/src/telemetry/mod.rs @@ -17,14 +17,13 @@ use webhooks::spawn_webhook_tracer; use crate::config::telemetry::{Telemetry, TelemetrySubscriberType}; impl Telemetry { - pub fn enable(self, is_enterprise: bool) { + pub fn enable(self) { // Spawn tracers for tracer in self.tracers.subscribers { tracer.typ.spawn( SubscriberBuilder::new(tracer.id) .with_interests(tracer.interests) .with_lossy(tracer.lossy), - is_enterprise, ); } @@ -35,7 +34,7 @@ impl Telemetry { Collector::reload(); } - pub fn update(self, is_enterprise: bool) { + pub fn update(self) { // Remove tracers that are no longer active let active_subscribers = Collector::get_subscribers(); for subscribed_id in &active_subscribers { @@ -58,7 +57,6 @@ impl Telemetry { SubscriberBuilder::new(tracer.id) .with_interests(tracer.interests) .with_lossy(tracer.lossy), - is_enterprise, ); } } @@ -96,8 +94,7 @@ impl Telemetry { } impl TelemetrySubscriberType { - // inbuxa: `_is_enterprise` is unused until monitoring history is rebuilt, and goes when it is: there is one edition - pub fn spawn(self, builder: SubscriberBuilder, _is_enterprise: bool) { + pub fn spawn(self, builder: SubscriberBuilder) { match self { TelemetrySubscriberType::ConsoleTracer(settings) => { spawn_console_tracer(builder, settings) diff --git a/crates/jmap/src/inbuxa/mod.rs b/crates/jmap/src/inbuxa/mod.rs index 9b83424..77dc4f2 100644 --- a/crates/jmap/src/inbuxa/mod.rs +++ b/crates/jmap/src/inbuxa/mod.rs @@ -12,4 +12,5 @@ pub mod ai_limits; pub mod deleted_account; pub mod fastmail; pub mod masked_email; +pub mod telemetry; pub mod undelete; diff --git a/crates/jmap/src/inbuxa/telemetry.rs b/crates/jmap/src/inbuxa/telemetry.rs new file mode 100644 index 0000000..8e7097b --- /dev/null +++ b/crates/jmap/src/inbuxa/telemetry.rs @@ -0,0 +1,214 @@ +/* + * SPDX-FileCopyrightText: 2026 Coffey Labs + * + * SPDX-License-Identifier: AGPL-3.0-only + */ + +//! `x:Metric/get` and `/query` over the stored history (monitoring spec, +//! "Interfaces"). Samples are server-level (MON-31) and read-only (MON-32). +//! A sample's `timestamp` comes from its id. + +use crate::{ + api::query::QueryResponseBuilder, + registry::{ + mapping::{RegistryGetResponse, RegistryQueryResponse}, + query::RegistryQueryFilters, + }, +}; +use common::telemetry::metrics::store::{MaybeMetric, StoredMetric}; +use jmap_proto::types::state::State; +use registry::{ + jmap::{IntoValue, JmapValue}, + schema::{ + prelude::Property, + structs::Metric, + }, + types::datetime::UTCDateTime, +}; +use store::{ + ValueKey, + registry::RegistryFilterOp, + write::{TelemetryClass, ValueClass}, +}; +use trc::MetricType; +use types::id::Id; +use utils::snowflake::SnowflakeIdGenerator; + +/// Telemetry is server-level: nobody inside a tenant reads it (MON-31). +pub fn assert_server_level(access_token: &common::auth::AccessToken) -> trc::Result<()> { + if access_token.tenant_id().is_some() { + Err(trc::JmapEvent::Forbidden + .into_err() + .details("Telemetry is server-level.")) + } else { + Ok(()) + } +} + +fn metric_type(metric: &Metric) -> MetricType { + match metric { + Metric::Counter(m) | Metric::Gauge(m) => m.metric, + Metric::Histogram(m) => m.metric, + } +} + +fn to_value(sample: StoredMetric) -> JmapValue<'static> { + let timestamp = sample.timestamp(); + let mut value = sample.metric.into_value(); + if let JmapValue::Object(obj) = &mut value { + obj.insert_unchecked( + Property::Timestamp, + JmapValue::Str(UTCDateTime::from_timestamp(timestamp as i64).to_string().into()), + ); + } + value +} + +/// `x:Metric/get`: by id, or every stored sample (up to the get limit). +pub(crate) async fn metric_get( + mut get: RegistryGetResponse<'_>, +) -> trc::Result> { + assert_server_level(get.access_token)?; + let server = get.server; + match get.ids.take() { + Some(ids) => { + for id in ids { + let sample = server + .metrics_store() + .get_value::(ValueKey::from(ValueClass::Telemetry( + TelemetryClass::Metric(id.id()), + ))) + .await + .ok() + .flatten() + .and_then(|MaybeMetric(metric)| metric) + .map(|metric| StoredMetric { id: id.id(), metric }); + match sample { + Some(sample) if !expired(server, sample.id).await => { + get.insert(id, to_value(sample)) + } + _ => get.not_found(id), + } + } + } + None => { + let max = server.core.jmap.get_max_objects; + let mut n = 0; + for sample in server + .read_metrics(0, u64::MAX, true, |_| { + n += 1; + n <= max + }) + .await? + { + get.insert(Id::from(sample.id), to_value(sample)); + } + } + } + Ok(get) +} + +async fn expired(server: &common::Server, id: u64) -> bool { + match common::telemetry::metrics::store::retention(server) + .await + .hold_metrics_for + { + Some(keep) => { + SnowflakeIdGenerator::from_duration(keep.into_inner()).is_some_and(|floor| id < floor) + } + None => false, + } +} + +fn timestamp_of(value: &serde_json::Value) -> Option { + value + .as_str() + .and_then(|s| s.parse::().ok()) + .map(|d| d.timestamp().max(0) as u64) +} + +/// `x:Metric/query`: filters `metric` (a name or a list) and the timestamp +/// comparisons; a bare `timestamp` is `unsupportedFilter`. Sorted by +/// timestamp (the id), either way. +pub(crate) async fn metric_query( + mut req: RegistryQueryResponse<'_>, +) -> trc::Result { + assert_server_level(req.access_token)?; + let (mut from_id, mut to_id) = (0u64, u64::MAX); + let mut metrics: Option> = None; + let mut bad = false; + req.request.extract_filters(|property, op, value| match property { + Property::Timestamp => { + let Some(ts) = timestamp_of(&value) else { + return false; + }; + // The ids of second `ts` run from `first_id_at(ts)` up to, not + // including, `first_id_at(ts + 1)` + let at = SnowflakeIdGenerator::first_id_at; + match op { + RegistryFilterOp::GreaterThan => from_id = from_id.max(at(ts + 1)), + RegistryFilterOp::GreaterEqualThan => from_id = from_id.max(at(ts)), + RegistryFilterOp::LowerThan => to_id = to_id.min(at(ts).saturating_sub(1)), + RegistryFilterOp::LowerEqualThan => { + to_id = to_id.min(at(ts + 1).saturating_sub(1)) + } + _ => return false, + } + true + } + Property::Metric => { + let names = match &value { + serde_json::Value::String(name) => vec![name.as_str()], + serde_json::Value::Array(list) => { + list.iter().filter_map(|v| v.as_str()).collect() + } + _ => return false, + }; + let mut parsed = Vec::with_capacity(names.len()); + for name in names { + match MetricType::parse(name) { + Some(metric) => parsed.push(metric), + None => bad = true, + } + } + metrics.get_or_insert_with(Vec::new).extend(parsed); + true + } + _ => false, + })?; + if bad { + return Err(trc::JmapEvent::UnsupportedFilter + .into_err() + .details("Unknown metric name")); + } + + let params = req + .request + .extract_parameters(req.server.core.jmap.query_max_results, None)?; + if !matches!(params.sort_by, Property::Timestamp | Property::Id) { + return Err(trc::JmapEvent::UnsupportedSort + .into_err() + .details("Metrics sort by timestamp only")); + } + let samples = req + .server + .read_metrics(from_id, to_id, params.sort_ascending, |sample| { + metrics + .as_ref() + .is_none_or(|m| m.contains(&metric_type(&sample.metric))) + }) + .await?; + + let mut response = QueryResponseBuilder::new( + samples.len(), + req.server.core.jmap.query_max_results, + State::Initial, + &req.request, + ); + for sample in samples { + if !response.add_id(Id::from(sample.id)) { + break; + } + } + Ok(response) +} diff --git a/crates/jmap/src/registry/get.rs b/crates/jmap/src/registry/get.rs index b5de3c9..a8f2692 100644 --- a/crates/jmap/src/registry/get.rs +++ b/crates/jmap/src/registry/get.rs @@ -380,6 +380,10 @@ impl RegistryGet for Server { spam_sample_get(get).await.map(|get| get.into_response()) } ObjectType::Log => log_get(get).await.map(|get| get.into_response()), + // inbuxa: monitoring history (MON-17, MON-31) + ObjectType::Metric => crate::inbuxa::telemetry::metric_get(get) + .await + .map(|get| get.into_response()), ObjectType::Bootstrap => bootstrap_get(get).await.map(|get| get.into_response()), ObjectType::AccountSettings | ObjectType::ApiKey diff --git a/crates/jmap/src/registry/mod.rs b/crates/jmap/src/registry/mod.rs index 3e20dae..10b675f 100644 --- a/crates/jmap/src/registry/mod.rs +++ b/crates/jmap/src/registry/mod.rs @@ -17,23 +17,9 @@ pub trait EnterpriseRegistry { } impl EnterpriseRegistry for Server { - fn assert_enterprise_object(&self, object_type: ObjectType) -> trc::Result<()> { - if !matches!( - object_type, - ObjectType::Metric - | ObjectType::Trace - ) { - return Ok(()); - } - - - // These are the Enterprise features INBUXA hasn't rebuilt yet - // (docs/spec/SPEC.md §4). Each type leaves this list when its rebuild - // lands. There's no edition to upgrade to, so the message says so. - Err(trc::JmapEvent::Forbidden.into_err().details(concat!( - "This feature isn't available in ", - types::brand!(), - " yet." - ))) + // inbuxa: every Enterprise feature is rebuilt (docs/spec/SPEC.md §4), the + // last being monitoring (MON-39), so nothing is refused here any more + fn assert_enterprise_object(&self, _: ObjectType) -> trc::Result<()> { + Ok(()) } } diff --git a/crates/jmap/src/registry/query.rs b/crates/jmap/src/registry/query.rs index 0d87228..c0f9288 100644 --- a/crates/jmap/src/registry/query.rs +++ b/crates/jmap/src/registry/query.rs @@ -122,6 +122,15 @@ impl RegistryQuery for Server { .await .and_then(|response| response.build()), + // inbuxa: monitoring history (MON-17, MON-31) + ObjectType::Metric => crate::inbuxa::telemetry::metric_query(RegistryQueryResponse { + server: self, + access_token, + object_type, + request, + }) + .await + .and_then(|response| response.build()), ObjectType::Log => log_query(RegistryQueryResponse { server: self, access_token, diff --git a/crates/services/src/task_manager/maintenance.rs b/crates/services/src/task_manager/maintenance.rs index 1ca8d2f..743b036 100644 --- a/crates/services/src/task_manager/maintenance.rs +++ b/crates/services/src/task_manager/maintenance.rs @@ -235,6 +235,20 @@ async fn store_maintenance( .await .caused_by(trc::location!())?; + use common::telemetry::metrics::store::MetricsStore; + // inbuxa: MON-17, MON-38: history past its retention goes; a + // failure leaves it for the next run + let retention = common::telemetry::metrics::store::retention(server).await; + if let Some(keep) = retention.hold_metrics_for + && !server.metrics_store().is_none() + && let Err(err) = server + .metrics_store() + .purge_metrics(keep.into_inner()) + .await + { + trc::error!(err.details("Failed to purge metric history")); + } + trc::event!( Store(StoreEvent::DataStorePurged), Elapsed = started.elapsed() diff --git a/crates/services/src/task_manager/scheduler.rs b/crates/services/src/task_manager/scheduler.rs index 2f4495c..dd2637a 100644 --- a/crates/services/src/task_manager/scheduler.rs +++ b/crates/services/src/task_manager/scheduler.rs @@ -40,6 +40,19 @@ enum Event { CalculateMetrics, TrainSpamClassifier, RenewNodeIdLease, + // inbuxa: MON-4: metric history + StoreMetrics, +} + +/// When the next metric-history tick is due (MON-4), read from the registry +/// so a change needs no reload (MON-3). +async fn metrics_collection_delay(server: &common::Server) -> Duration { + utils::cron::SimpleCron::from( + common::telemetry::metrics::store::retention(server) + .await + .metrics_collection_interval, + ) + .time_to_next() } #[derive(Default)] @@ -113,6 +126,12 @@ pub fn spawn_task_scheduler(inner: Arc) { // Calculate expensive metrics queue.schedule(Instant::now(), Event::CalculateMetrics); + // inbuxa: MON-4: metric history on its own schedule + queue.schedule( + Instant::now() + metrics_collection_delay(&server).await, + Event::StoreMetrics, + ); + } @@ -210,13 +229,9 @@ pub fn spawn_task_scheduler(inner: Arc) { if roles.metrics_push { let otel = otel.clone(); - - #[cfg(not(feature = "enterprise"))] - let is_enterprise = false; - tokio::spawn(async move { let elapsed = Instant::now(); - otel.push_metrics(is_enterprise, start_time).await; + otel.push_metrics(start_time).await; trc::event!( Telemetry(TelemetryEvent::MetricsPushed), @@ -244,6 +259,16 @@ pub fn spawn_task_scheduler(inner: Arc) { tokio::spawn(async move { let elapsed = Instant::now(); if server.core.network.roles.metrics_calculate { + // inbuxa: MON-7: the queue gauge from the queue itself, + // so it's right after a restart + match server.total_queued_messages().await { + Ok(total) => { + Collector::update_gauge(MetricType::QueueCount, total); + } + Err(err) => { + trc::error!(err.details("Failed to count queued messages")); + } + } if update_other_metrics { match server.total_accounts().await { @@ -300,6 +325,17 @@ pub fn spawn_task_scheduler(inner: Arc) { ); }); } + // inbuxa: MON-4: every node writes its own samples + Event::StoreMetrics => { + queue.schedule( + Instant::now() + metrics_collection_delay(&server).await, + Event::StoreMetrics, + ); + let server = server.clone(); + tokio::spawn(async move { + server.store_metrics().await; + }); + } Event::TrainSpamClassifier => { if let Some(train_frequency) = server .core @@ -394,6 +430,7 @@ impl Event { Event::CalculateMetrics => "calculateMetrics", Event::TrainSpamClassifier => "trainSpamClassifier", Event::RenewNodeIdLease => "renewNodeIdLease", + Event::StoreMetrics => "storeMetrics", } } } diff --git a/crates/trc/src/ipc/metrics.rs b/crates/trc/src/ipc/metrics.rs index bef5302..7ce82d0 100644 --- a/crates/trc/src/ipc/metrics.rs +++ b/crates/trc/src/ipc/metrics.rs @@ -207,7 +207,7 @@ impl Collector { METRIC_INTERESTS.update(interests); } - pub fn collect_counters(_is_enterprise: bool) -> impl Iterator { + pub fn collect_counters() -> impl Iterator { EVENT_COUNTERS .inner() .iter() @@ -225,21 +225,20 @@ impl Collector { }) } - pub fn collect_gauges(is_enterprise: bool) -> impl Iterator { - static E_GAUGES: &[&AtomicGauge] = + // inbuxa: MON-6: one edition, every gauge + pub fn collect_gauges() -> impl Iterator { + static GAUGES: &[&AtomicGauge] = &[&SERVER_MEMORY, &QUEUE_COUNT, &USER_COUNT, &DOMAIN_COUNT]; - static C_GAUGES: &[&AtomicGauge] = &[&SERVER_MEMORY, &USER_COUNT, &DOMAIN_COUNT]; - if is_enterprise { E_GAUGES } else { C_GAUGES } + GAUGES .iter() .copied() .chain(CONNECTION_METRICS.iter().map(|m| &m.active_connections)) } - pub fn collect_histograms( - is_enterprise: bool, - ) -> impl Iterator> { - static E_HISTOGRAMS: &[&AtomicHistogram<12>] = &[ + // inbuxa: MON-6: one edition, every histogram + pub fn collect_histograms() -> impl Iterator> { + static HISTOGRAMS: &[&AtomicHistogram<12>] = &[ &MESSAGE_INGESTION_TIME, &MESSAGE_INDEX_TIME, &MESSAGE_DELIVERY_TIME, @@ -252,17 +251,7 @@ impl Collector { &STORE_BLOB_WRITE_TIME, &DNS_LOOKUP_TIME, ]; - static C_HISTOGRAMS: &[&AtomicHistogram<12>] = &[ - &MESSAGE_DELIVERY_TIME, - &MESSAGE_INCOMING_SIZE, - &MESSAGE_SUBMISSION_SIZE, - ]; - - if is_enterprise { - E_HISTOGRAMS - } else { - C_HISTOGRAMS - } + HISTOGRAMS .iter() .copied() .chain(CONNECTION_METRICS.iter().map(|m| &m.elapsed)) diff --git a/crates/utils/src/snowflake.rs b/crates/utils/src/snowflake.rs index b5e7538..c406aab 100644 --- a/crates/utils/src/snowflake.rs +++ b/crates/utils/src/snowflake.rs @@ -96,6 +96,12 @@ impl SnowflakeIdGenerator { }) } + // inbuxa: the first id of a UNIX second, so ids can be searched by time + // (monitoring history) + pub fn first_id_at(timestamp: u64) -> u64 { + (timestamp.saturating_sub(DEFAULT_EPOCH) * 1000) << (SEQUENCE_LEN + NODE_ID_LEN) + } + pub fn to_timestamp(id: u64) -> u64 { (id >> (SEQUENCE_LEN + NODE_ID_LEN)) / 1000 + DEFAULT_EPOCH } diff --git a/tests/src/telemetry/mod.rs b/tests/src/telemetry/mod.rs index f75a680..b403067 100644 --- a/tests/src/telemetry/mod.rs +++ b/tests/src/telemetry/mod.rs @@ -6,7 +6,6 @@ #[cfg(feature = "pending-rebuild")] // inbuxa: pending-rebuild, see docs/spec/features/ pub mod alerts; -#[cfg(feature = "pending-rebuild")] // inbuxa: pending-rebuild, see docs/spec/features/ pub mod metrics; #[cfg(feature = "pending-rebuild")] // inbuxa: pending-rebuild, see docs/spec/features/ pub mod tracing; @@ -61,7 +60,6 @@ pub async fn telemetry_tests() { #[cfg(feature = "pending-rebuild")] alerts::test(&test).await; - #[cfg(feature = "pending-rebuild")] metrics::test(&test).await; #[cfg(feature = "pending-rebuild")] tracing::test(&test).await; diff --git a/tests/src/utils/server.rs b/tests/src/utils/server.rs index 8d3aaeb..ec843d5 100644 --- a/tests/src/utils/server.rs +++ b/tests/src/utils/server.rs @@ -362,7 +362,7 @@ impl TestServerBuilder { let cache = Caches::parse(&mut self.bootstrap).await; // Enable telemetry - telemetry.enable(true); + telemetry.enable(); // Build inner let (ipc, mut ipc_rxs) = build_ipc(!core.storage.coordinator.is_none());