fix(metrics): Address review feedback

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
 

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")));
 		}
 	}
 

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<String> {
+	#[derive(serde::Deserialize)]
+	struct Typed {
+		#[serde(rename = "type")]
+		kind: Option<String>,
+	}
+
+	serde_json::from_str::<Typed>(pdu.get()).ok()?.kind
+}
+
 async fn handle_edu_presence(
 	services: &Services,
 	origin: &ServerName,

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<f64> { (0..16).map(|i| 0.001 * f64::from(1_u32
 fn size_buckets() -> Vec<f64> { (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();
 

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<u64> {
-	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()
 }
 

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;
 

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},

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()

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<Self>) {
@@ -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<Self>) -> Arc<Counts> {
-		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<F>(
 		self: &Arc<Self>,
 		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<Option<(Instant, Arc<MemoryUsage>)>>);
+
+impl MemoryUsageCache {
+	fn get(&self, services: &Services) -> Option<Arc<MemoryUsage>> {
+		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<u64> {
+	services
+		.db
+		.get(map)
+		.ok()?
+		.property_integer(c"rocksdb.estimate-num-keys")
+		.ok()
 }
 
 fn observe(observer: &dyn AsyncInstrument<u64>, value: usize, attributes: &[KeyValue]) {