From ea3ec965ed75b0059530449adbe3b5670084310c Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Tue, 6 Oct 2026 21:59:04 +0200 Subject: [PATCH] fix(metrics): Address review feedback --- docs/guides/metrics.mdx | 2 +- src/api/client/unversioned.rs | 2 +- src/api/server/send.rs | 28 +++++--- src/core/metrics/instruments.rs | 7 +- src/core/metrics/process.rs | 6 +- src/database/engine.rs | 1 + src/database/mod.rs | 1 + src/main/logging.rs | 14 +++- src/service/metrics.rs | 118 ++++++++++++++------------------ 9 files changed, 98 insertions(+), 81 deletions(-) diff --git a/docs/guides/metrics.mdx b/docs/guides/metrics.mdx index 2216a2dee..f807fd80a 100644 --- a/docs/guides/metrics.mdx +++ b/docs/guides/metrics.mdx @@ -93,7 +93,7 @@ Names below are as they appear in Prometheus. OTLP uses the dotted OpenTelemetry | `continuwuity_tokio_worker_busy_time_seconds_total` | counter | | Time worker threads spent busy | | `continuwuity_tokio_worker_parks_total` | counter | | Number of times worker threads parked | -User and media file counts are refreshed every minute. +User and media file counts are estimates taken from RocksDB (`rocksdb.estimate-num-keys`), so they may lag slightly or drift after deletions. ## Grafana dashboard diff --git a/src/api/client/unversioned.rs b/src/api/client/unversioned.rs index 11b482a21..814ee0d9e 100644 --- a/src/api/client/unversioned.rs +++ b/src/api/client/unversioned.rs @@ -97,7 +97,7 @@ pub(crate) async fn continuwuity_metrics( }); if !authorized { - return Err!(Request(Forbidden("Invalid metrics token"))); + return Err!(Request(Unauthorized("Invalid metrics token"))); } } diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 415b019c3..0a384717f 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -161,7 +161,10 @@ async fn process_inbound_transaction( .metrics .instruments .federation_incoming_pdus - .add(1, &[KeyValue::new("outcome", "unparsable")]); + .add(1, &[ + KeyValue::new("type", pdu_type_label(raw_pdu_type(pdu).as_deref())), + KeyValue::new("outcome", "unparsable"), + ]); let queued = services .rooms .event_handler @@ -359,7 +362,10 @@ async fn handle_room( .server .check_running() .map_err(|_| TransactionError::ShuttingDown)?; - let event_type = KeyValue::new("type", pdu_type_label(&value)); + let event_type = KeyValue::new( + "type", + pdu_type_label(value.get("type").and_then(CanonicalJsonValue::as_str)), + ); let origin_server_ts = value .get("origin_server_ts") .and_then(CanonicalJsonValue::as_integer) @@ -445,17 +451,23 @@ async fn handle_edu(services: &Services, origin: &ServerName, edu: Edu) { /// Known event types, or `other`, so remote servers cannot create new /// attribute values. -fn pdu_type_label(value: &CanonicalJsonObject) -> String { - match value - .get("type") - .and_then(CanonicalJsonValue::as_str) - .map(TimelineEventType::from) - { +fn pdu_type_label(event_type: Option<&str>) -> String { + match event_type.map(TimelineEventType::from) { | Some(TimelineEventType::_Custom(_)) | None => "other".to_owned(), | Some(event_type) => event_type.to_string(), } } +fn raw_pdu_type(pdu: &serde_json::value::RawValue) -> Option { + #[derive(serde::Deserialize)] + struct Typed { + #[serde(rename = "type")] + kind: Option, + } + + serde_json::from_str::(pdu.get()).ok()?.kind +} + async fn handle_edu_presence( services: &Services, origin: &ServerName, diff --git a/src/core/metrics/instruments.rs b/src/core/metrics/instruments.rs index e59978ccd..b9ba973b2 100644 --- a/src/core/metrics/instruments.rs +++ b/src/core/metrics/instruments.rs @@ -1,4 +1,4 @@ -use std::time::{Duration, SystemTime}; +use std::time::{Duration, Instant}; use opentelemetry::{ KeyValue, @@ -133,14 +133,13 @@ pub fn duration_buckets() -> Vec { (0..16).map(|i| 0.001 * f64::from(1_u32 fn size_buckets() -> Vec { (0..11).map(|i| 1024.0 * f64::from(4_u32.pow(i))).collect() } fn register_process(meter: &Meter) { - let started = SystemTime::now(); + let started = Instant::now(); meter .f64_observable_gauge(metric::PROCESS_UPTIME) .with_description("Time since the server started") .with_unit("s") .with_callback(move |observer| { - let uptime = started.elapsed().unwrap_or_default(); - observer.observe(uptime.as_secs_f64(), &[]); + observer.observe(started.elapsed().as_secs_f64(), &[]); }) .build(); diff --git a/src/core/metrics/process.rs b/src/core/metrics/process.rs index 36192a402..69ce48861 100644 --- a/src/core/metrics/process.rs +++ b/src/core/metrics/process.rs @@ -35,7 +35,11 @@ pub(super) fn memory() -> Option<(u64, u64)> { None } #[cfg(target_os = "linux")] pub(super) fn open_fds() -> Option { - let count = std::fs::read_dir("/proc/self/fd").ok()?.count(); + // The directory handle used for the listing is itself an open descriptor. + let count = std::fs::read_dir("/proc/self/fd") + .ok()? + .count() + .checked_sub(1)?; count.try_into().ok() } diff --git a/src/database/engine.rs b/src/database/engine.rs index a0a99ad60..7fd48a7d9 100644 --- a/src/database/engine.rs +++ b/src/database/engine.rs @@ -6,6 +6,7 @@ pub(crate) mod descriptor; mod files; mod logger; mod memory_usage; +pub use memory_usage::MemoryUsage; mod open; mod repair; diff --git a/src/database/mod.rs b/src/database/mod.rs index fcc4f4df7..e6585319c 100644 --- a/src/database/mod.rs +++ b/src/database/mod.rs @@ -33,6 +33,7 @@ use conduwuit::{Result, Server, err}; pub use self::{ de::{Ignore, IgnoreAll}, deserialized::Deserialized, + engine::MemoryUsage, handle::Handle, keyval::{KeyVal, Slice, serialize_key, serialize_val}, map::{Get, Map, Qry, compact}, diff --git a/src/main/logging.rs b/src/main/logging.rs index 1cffed4f1..79ff3c51e 100644 --- a/src/main/logging.rs +++ b/src/main/logging.rs @@ -6,6 +6,7 @@ use conduwuit_core::{ debug_warn, err, log::{ConsoleFormat, ConsoleWriter, LogLevelReloadHandles, capture, fmt_span}, result::UnwrapOrErr, + warn, }; #[cfg(feature = "otlp_telemetry")] use opentelemetry::trace::TracerProvider; @@ -88,10 +89,21 @@ pub(crate) fn init( .with_protocol(opentelemetry_otlp::Protocol::Grpc) // TODO: build from env when 0.32 is released .build() .expect("Failed to create OTLP gRPC exporter"), - | _ => opentelemetry_otlp::SpanExporter::builder() + | "http" => opentelemetry_otlp::SpanExporter::builder() .with_http() .build() .expect("Failed to create OTLP HTTP exporter"), + | protocol => { + warn!( + "Invalid OTLP protocol '{}', falling back to HTTP. Valid options are \ + 'http' or 'grpc'.", + protocol + ); + opentelemetry_otlp::SpanExporter::builder() + .with_http() + .build() + .expect("Failed to create OTLP HTTP exporter") + }, }; let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() diff --git a/src/service/metrics.rs b/src/service/metrics.rs index dec506bcf..4a25c4d23 100644 --- a/src/service/metrics.rs +++ b/src/service/metrics.rs @@ -1,20 +1,19 @@ use std::{ - sync::{ - Arc, - atomic::{AtomicU64, Ordering}, - }, - time::Duration, + sync::{Arc, Mutex}, + time::{Duration, Instant}, }; use conduwuit::{ metrics::{AsyncInstrument, KeyValue}, utils::millis_since_unix_epoch, }; -use futures::StreamExt; +use database::MemoryUsage; use crate::Services; -const COUNT_INTERVAL: Duration = Duration::from_mins(1); +/// Both database memory gauges are collected in the same scrape, so the +/// statistics are shared between them for this long. +const MEMORY_USAGE_TTL: Duration = Duration::from_secs(1); impl Services { pub(crate) fn register_metrics(self: &Arc) { @@ -36,25 +35,28 @@ impl Services { }, ); - let counts = self.spawn_counter(); - let users = counts.clone(); self.gauge( "continuwuity.users", - "Known users, local including deactivated, refreshed every minute", + "Estimated known users, local including deactivated", "{user}", - move |_, observer| { - let local = users.local_users.load(Ordering::Relaxed); - let remote = users.remote_users.load(Ordering::Relaxed); - observer.observe(local, &[KeyValue::new("origin", "local")]); - observer.observe(remote, &[KeyValue::new("origin", "remote")]); + |services, observer| { + for (map, origin) in + [("userid_password", "local"), ("remoteuserid_remoteuser", "remote")] + { + if let Some(keys) = estimate_keys(services, map) { + observer.observe(keys, &[KeyValue::new("origin", origin)]); + } + } }, ); self.gauge( "continuwuity.media.files", - "Stored media files including thumbnails, refreshed every minute", + "Estimated stored media files including thumbnails", "{file}", - move |_, observer| { - observer.observe(counts.media_files.load(Ordering::Relaxed), &[]); + |services, observer| { + if let Some(keys) = estimate_keys(services, "mediaid_file") { + observer.observe(keys, &[]); + } }, ); @@ -78,12 +80,14 @@ impl Services { }, ); + let memory = Arc::new(MemoryUsageCache::default()); + let block_cache_memory = memory.clone(); self.gauge( "continuwuity.database.memory", "RocksDB memory usage by component", "By", - |services, observer| { - let Ok(usage) = services.db.db.memory_usage_stats() else { + move |services, observer| { + let Some(usage) = memory.get(services) else { return; }; @@ -102,13 +106,13 @@ impl Services { "continuwuity.database.block_cache", "RocksDB block cache usage by column family", "By", - |services, observer| { - let Ok(usage) = services.db.db.memory_usage_stats() else { + move |services, observer| { + let Some(usage) = block_cache_memory.get(services) else { return; }; - for (column, bytes) in usage.col_caches { - observer.observe(bytes, &[KeyValue::new("column", column)]); + for (column, bytes) in &usage.col_caches { + observer.observe(*bytes, &[KeyValue::new("column", column.clone())]); } }, ); @@ -144,43 +148,6 @@ impl Services { ); } - /// Counting requires iterating the database, so it runs in the background - /// instead of in the collection callback. - fn spawn_counter(self: &Arc) -> Arc { - let counts = Arc::new(Counts::default()); - let services = Arc::downgrade(self); - let task_counts = counts.clone(); - self.server.runtime().spawn(async move { - let mut interval = tokio::time::interval(COUNT_INTERVAL); - loop { - interval.tick().await; - let Some(services) = services.upgrade() else { - break; - }; - if !services.server.running() { - break; - } - - let local = services.users.count().await; - let remote = services.users.stream_remote_users().count().await; - let media = match services.db.get("mediaid_file") { - | Ok(map) => map.count().await, - | Err(_) => 0, - }; - - for (count, value) in [ - (&task_counts.local_users, local), - (&task_counts.remote_users, remote), - (&task_counts.media_files, media), - ] { - count.store(value.try_into().unwrap_or(u64::MAX), Ordering::Relaxed); - } - } - }); - - counts - } - fn gauge( self: &Arc, name: &'static str, @@ -207,10 +174,31 @@ impl Services { } #[derive(Default)] -struct Counts { - local_users: AtomicU64, - remote_users: AtomicU64, - media_files: AtomicU64, +struct MemoryUsageCache(Mutex)>>); + +impl MemoryUsageCache { + fn get(&self, services: &Services) -> Option> { + let mut cached = self.0.lock().expect("locked"); + if let Some((at, usage)) = &*cached { + if at.elapsed() < MEMORY_USAGE_TTL { + return Some(usage.clone()); + } + } + + let usage = Arc::new(services.db.db.memory_usage_stats().ok()?); + *cached = Some((Instant::now(), usage.clone())); + Some(usage) + } +} + +/// Cheap estimate from RocksDB properties, unlike iterating the column. +fn estimate_keys(services: &Services, map: &'static str) -> Option { + services + .db + .get(map) + .ok()? + .property_integer(c"rocksdb.estimate-num-keys") + .ok() } fn observe(observer: &dyn AsyncInstrument, value: usize, attributes: &[KeyValue]) {