feat(metrics): Record metrics with OpenTelemetry and export over OTLP

src/service/antispam/mod.rs

@@ -1,7 +1,7 @@
 use std::{fmt::Debug, sync::Arc};
 
 use async_trait::async_trait;
-use conduwuit::{Result, config::Antispam, debug};
+use conduwuit::{Result, Server, config::Antispam, debug, metrics::KeyValue};
 use ruma::{
 	OwnedRoomId, OwnedUserId,
 	api::{auth_scheme::AppserviceToken, path_builder::VersionHistory},
@@ -11,6 +11,7 @@ use ruminuwuity::{draupnir_antispam, meowlnir_antispam};
 use crate::{client, config, sending, service::Dep};
 
 struct Services {
+	server: Arc<Server>,
 	config: Dep<config::Service>,
 	client: Dep<client::Service>,
 }
@@ -24,6 +25,7 @@ impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		Ok(Arc::new(Self {
 			services: Services {
+				server: args.server.clone(),
 				client: args.depend::<client::Service>("client"),
 				config: args.depend::<config::Service>("config"),
 			},
@@ -101,6 +103,7 @@ impl Service {
 			} else {
 				Ok(())
 			};
+			self.record("user_may_invite", &result);
 			return result;
 		}
 		Ok(())
@@ -146,6 +149,7 @@ impl Service {
 			} else {
 				Ok(())
 			};
+			self.record("user_may_join_room", &result);
 			return result;
 		}
 		Ok(())
@@ -160,24 +164,37 @@ impl Service {
 	) -> Result<()> {
 		if let Some(Antispam { meowlnir: Some(meowlnir), .. }) = &self.services.config.antispam {
 			debug!(?room_id, ?user_id, "Asking meowlnir for accept_make_join");
-			self.send_antispam_request(
-				meowlnir.base_url.as_str(),
-				&meowlnir.secret,
-				meowlnir_antispam::accept_make_join::v1::Request::new(
-					meowlnir.management_room.clone(),
-					user_id,
-					room_id,
-				),
-			)
-			.await
-			.inspect(|_| debug!("meowlnir allowed the make_join"))
-			.inspect_err(|e| debug!("meowlnir denied the make_join: {e:?}"))
-			.map(|_| ())
+			let result = self
+				.send_antispam_request(
+					meowlnir.base_url.as_str(),
+					&meowlnir.secret,
+					meowlnir_antispam::accept_make_join::v1::Request::new(
+						meowlnir.management_room.clone(),
+						user_id,
+						room_id,
+					),
+				)
+				.await
+				.inspect(|_| debug!("meowlnir allowed the make_join"))
+				.inspect_err(|e| debug!("meowlnir denied the make_join: {e:?}"))
+				.map(|_| ());
+			self.record("accept_make_join", &result);
+			result
 		} else {
 			Ok(())
 		}
 	}
 
+	fn record(&self, check: &'static str, result: &Result) {
+		let outcome = if result.is_ok() { "allowed" } else { "denied" };
+		self.services
+			.server
+			.metrics
+			.instruments
+			.antispam_checks
+			.add(1, &[KeyValue::new("check", check), KeyValue::new("outcome", outcome)]);
+	}
+
 	/// Returns whether all joins should be checked with Meowlnir.
 	/// Is always false if Meowlnir is not configured.
 	pub fn check_all_joins(&self) -> bool {

src/service/federation/execute.rs

@@ -1,7 +1,7 @@
 use std::{borrow::Cow, fmt::Debug, mem, time::Instant};
 
 use conduwuit::{
-	Err, Error, Result, debug, debug_error, debug_info, debug_warn, err, metrics::OutcomeLabels,
+	Err, Error, Result, debug, debug_error, debug_info, debug_warn, err, metrics::KeyValue,
 	trace, utils::response::LimitReadExt,
 };
 use ipaddress::IPAddress;
@@ -191,10 +191,9 @@ impl super::Service {
 		self.services
 			.server
 			.metrics
-			.prometheus
-			.federation_request_duration
-			.get_or_create(&OutcomeLabels { outcome })
-			.observe(elapsed.as_secs_f64());
+			.instruments
+			.federation_outgoing_request_duration
+			.record(elapsed.as_secs_f64(), &[KeyValue::new("outcome", outcome)]);
 		match response {
 			| Ok(response) =>
 				self.handle_response::<T>(dest, actual, &method, &url, response)

src/service/media/mod.rs

@@ -10,7 +10,9 @@ use std::{path::PathBuf, sync::Arc, time::SystemTime};
 use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose};
 use conduwuit::{
-	Err, Result, Server, debug, debug_error, debug_info, debug_warn, err, error, trace,
+	Err, Result, Server, debug, debug_error, debug_info, debug_warn, err, error,
+	metrics::KeyValue,
+	trace,
 	utils::{
 		self, MutexMap,
 		time::{self, TimeDirection},
@@ -110,9 +112,26 @@ impl Service {
 			err!(Database(error!("Failed to write media file for MXC {mxc} at key {key:?}: {e}")))
 		})?;
 
+		self.services
+			.server
+			.metrics
+			.instruments
+			.media_write_size
+			.record(file.len().try_into().unwrap_or(u64::MAX), &[self.origin(mxc)]);
+
 		Ok(())
 	}
 
+	fn origin(&self, mxc: &Mxc<'_>) -> KeyValue {
+		let origin = if self.services.globals.server_is_ours(mxc.server_name) {
+			"local"
+		} else {
+			"remote"
+		};
+
+		KeyValue::new("origin", origin)
+	}
+
 	/// Deletes a file in the database and from the media directory via an MXC
 	pub async fn delete(&self, mxc: &Mxc<'_>) -> Result<()> {
 		match self.db.search_mxc_metadata_prefix(mxc).await {
@@ -177,6 +196,13 @@ impl Service {
 					.read_to_end(&mut content)
 					.await?;
 
+				self.services
+					.server
+					.metrics
+					.instruments
+					.media_read_size
+					.record(content.len().try_into().unwrap_or(u64::MAX), &[self.origin(mxc)]);
+
 				Ok(Some(FileMeta {
 					content: Some(content),
 					content_type,

src/service/metrics.rs

@@ -0,0 +1,220 @@
+use std::{
+	sync::{
+		Arc,
+		atomic::{AtomicU64, Ordering},
+	},
+	time::Duration,
+};
+
+use conduwuit::{
+	metrics::{AsyncInstrument, KeyValue},
+	utils::millis_since_unix_epoch,
+};
+use futures::StreamExt;
+
+use crate::Services;
+
+const COUNT_INTERVAL: Duration = Duration::from_mins(1);
+
+impl Services {
+	pub(crate) fn register_metrics(self: &Arc<Self>) {
+		self.gauge(
+			"continuwuity.sender.queue.depth",
+			"Messages waiting in the sender worker channels",
+			"{message}",
+			|services, observer| {
+				observe(observer, services.sending.queue_depth(), &[]);
+			},
+		);
+
+		self.gauge(
+			"continuwuity.federation.incoming.transactions.active",
+			"Incoming federation transactions currently being processed",
+			"{transaction}",
+			|services, observer| {
+				observe(observer, services.transactions.txn_active_handle_count(), &[]);
+			},
+		);
+
+		let counts = self.spawn_counter();
+		let users = counts.clone();
+		self.gauge(
+			"continuwuity.users",
+			"Known users, local including deactivated, refreshed every minute",
+			"{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")]);
+			},
+		);
+		self.gauge(
+			"continuwuity.media.files",
+			"Stored media files including thumbnails, refreshed every minute",
+			"{file}",
+			move |_, observer| {
+				observer.observe(counts.media_files.load(Ordering::Relaxed), &[]);
+			},
+		);
+
+		self.gauge(
+			"continuwuity.federation.destinations",
+			"Remote servers in retry backoff or with a stale destination cache entry",
+			"{server}",
+			|services, observer| {
+				let now = millis_since_unix_epoch();
+				let backoff = services
+					.federation
+					.remote_health
+					.read()
+					.values()
+					.filter(|&&(_, next_retry)| now < next_retry)
+					.count();
+				let stale = services.federation.stale_destinations.read().len();
+
+				observe(observer, backoff, &[KeyValue::new("state", "backoff")]);
+				observe(observer, stale, &[KeyValue::new("state", "stale")]);
+			},
+		);
+
+		self.gauge(
+			"continuwuity.database.memory",
+			"RocksDB memory usage by component",
+			"By",
+			|services, observer| {
+				let Ok(usage) = services.db.db.memory_usage_stats() else {
+					return;
+				};
+
+				for (name, bytes) in [
+					("mem_tables", usage.mem_table_total),
+					("mem_tables_unflushed", usage.mem_table_unflushed),
+					("table_readers", usage.mem_table_readers_total),
+					("row_cache", usage.row_cache),
+				] {
+					observer.observe(bytes, &[KeyValue::new("name", name)]);
+				}
+			},
+		);
+
+		self.gauge(
+			"continuwuity.database.block_cache",
+			"RocksDB block cache usage by column family",
+			"By",
+			|services, observer| {
+				let Ok(usage) = services.db.db.memory_usage_stats() else {
+					return;
+				};
+
+				for (column, bytes) in usage.col_caches {
+					observer.observe(bytes, &[KeyValue::new("column", column)]);
+				}
+			},
+		);
+
+		self.gauge(
+			"continuwuity.database.size",
+			"Total size of RocksDB SST files",
+			"By",
+			|services, observer| {
+				let size = services
+					.db
+					.iter()
+					.filter_map(|(_, map)| {
+						map.property_integer(c"rocksdb.total-sst-files-size").ok()
+					})
+					.fold(0_u64, u64::saturating_add);
+
+				observer.observe(size, &[]);
+			},
+		);
+
+		self.gauge(
+			"continuwuity.cache.entries",
+			"Entries held in in-memory caches",
+			"{entry}",
+			|services, observer| {
+				let (auth_chain, _) = services.rooms.auth_chain.get_cache_usage();
+				let stateinfo = services.rooms.state_compressor.stateinfo_cache.lock().len();
+
+				observe(observer, auth_chain, &[KeyValue::new("name", "auth_chain")]);
+				observe(observer, stateinfo, &[KeyValue::new("name", "stateinfo")]);
+			},
+		);
+	}
+
+	/// 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,
+		description: &'static str,
+		unit: &'static str,
+		callback: F,
+	) where
+		F: Fn(&Self, &dyn AsyncInstrument<u64>) + Send + Sync + 'static,
+	{
+		let services = Arc::downgrade(self);
+		self.server
+			.metrics
+			.meter
+			.u64_observable_gauge(name)
+			.with_description(description)
+			.with_unit(unit)
+			.with_callback(move |observer| {
+				if let Some(services) = services.upgrade() {
+					callback(&services, observer);
+				}
+			})
+			.build();
+	}
+}
+
+#[derive(Default)]
+struct Counts {
+	local_users: AtomicU64,
+	remote_users: AtomicU64,
+	media_files: AtomicU64,
+}
+
+fn observe(observer: &dyn AsyncInstrument<u64>, value: usize, attributes: &[KeyValue]) {
+	if let Ok(value) = value.try_into() {
+		observer.observe(value, attributes);
+	}
+}

src/service/mod.rs

@@ -7,6 +7,7 @@ extern crate conduwuit_database as database;
 conduwuit_macros::introspect_crate! {}
 
 mod manager;
+mod metrics;
 mod migrations;
 mod service;
 pub mod services;

src/service/sending/mod.rs

@@ -14,6 +14,7 @@ use std::{
 use async_trait::async_trait;
 use conduwuit::{
 	Result, Server, debug, debug_warn, err, error,
+	metrics::KeyValue,
 	smallvec::SmallVec,
 	utils::{ReadyExt, TryReadyExt, available_parallelism, math::usize_from_u64_truncated},
 	warn,
@@ -481,6 +482,22 @@ impl Service {
 			.expect("missing sender worker channels")
 			.0;
 
+		let kind = match msg.event {
+			| SendingEvent::Pdu(_) => Some("pdu"),
+			| SendingEvent::Edu(_) => Some("edu"),
+			| SendingEvent::Flush => None,
+		};
+		if let Some(kind) = kind {
+			self.server
+				.metrics
+				.instruments
+				.sender_events_queued
+				.add(1, &[
+					KeyValue::new("destination", msg.dest.kind()),
+					KeyValue::new("kind", kind),
+				]);
+		}
+
 		debug_assert!(!sender.is_full(), "channel full");
 		debug_assert!(!sender.is_closed(), "channel closed");
 		sender.send(msg).map_err(|e| err!("{e}"))

src/service/sending/sender.rs

@@ -13,7 +13,7 @@ use conduwuit::{debug_info, debug_warn, info, trace};
 use conduwuit_core::{
 	Error, Event, Result, at, debug, err, error,
 	matrix::pdu::sticky,
-	metrics::TransactionLabels,
+	metrics::KeyValue,
 	result::LogErr,
 	utils::{
 		ReadyExt, calculate_hash,
@@ -140,17 +140,6 @@ impl Service {
 		futures: &mut SendingFutures<'a>,
 		statuses: &mut CurTransactionStatus,
 	) {
-		let (dest, outcome) = match &response {
-			| Ok(dest) => (dest, "success"),
-			| Err((dest, _)) => (dest, "failure"),
-		};
-		self.server
-			.metrics
-			.prometheus
-			.sender_transactions
-			.get_or_create(&TransactionLabels { destination: dest.kind(), outcome })
-			.inc();
-
 		match response {
 			| Ok(dest) => self.handle_response_ok(&dest, futures, statuses).await,
 			| Err((dest, e)) => {
@@ -770,13 +759,44 @@ impl Service {
 			debug_warn!(?dest, "Attempted to send empty transaction!");
 			return std::future::ready(Ok(dest)).boxed();
 		}
-		match dest {
+
+		let destination = KeyValue::new("destination", dest.kind());
+		let (pdus, edus) =
+			events
+				.iter()
+				.fold((0_u64, 0_u64), |(pdus, edus), event| match event {
+					| SendingEvent::Pdu(_) => (pdus.saturating_add(1), edus),
+					| SendingEvent::Edu(_) => (pdus, edus.saturating_add(1)),
+					| SendingEvent::Flush => (pdus, edus),
+				});
+
+		let start = Instant::now();
+		let send = match dest {
 			| Destination::Federation(server) =>
 				self.send_events_dest_federation(server, events).boxed(),
 			| Destination::Appservice(id) => self.send_events_dest_appservice(id, events).boxed(),
 			| Destination::Push(user_id, pushkey) =>
 				self.send_events_dest_push(user_id, pushkey, events).boxed(),
-		}
+		};
+
+		send.inspect(move |result| {
+			let instruments = &self.server.metrics.instruments;
+			let outcome =
+				KeyValue::new("outcome", if result.is_ok() { "success" } else { "failure" });
+			instruments
+				.sender_transaction_duration
+				.record(start.elapsed().as_secs_f64(), &[destination.clone(), outcome.clone()]);
+			for (kind, count) in [("pdu", pdus), ("edu", edus)] {
+				if count > 0 {
+					instruments.sender_events.add(count, &[
+						destination.clone(),
+						KeyValue::new("kind", kind),
+						outcome.clone(),
+					]);
+				}
+			}
+		})
+		.boxed()
 	}
 
 	#[tracing::instrument(

src/service/services.rs

@@ -1,10 +1,7 @@
 use std::{any::Any, collections::BTreeMap, sync::Arc};
 
 use conduwuit::{
-	Result, Server, SyncRwLock, debug, debug_info, error, info,
-	metrics::{ColumnLabels, NameLabels},
-	trace,
-	utils::{millis_since_unix_epoch, stream::IterStream},
+	Result, Server, SyncRwLock, debug, debug_info, error, info, trace, utils::stream::IterStream,
 };
 use database::Database;
 use futures::{Stream, StreamExt, TryStreamExt};
@@ -73,7 +70,7 @@ impl Services {
 			}};
 		}
 
-		Ok(Arc::new(Self {
+		let services = Arc::new(Self {
 			// firstrun service should be built first so other services
 			// can check first-run state
 			firstrun: build!(firstrun::Service),
@@ -131,7 +128,11 @@ impl Services {
 			service,
 			server,
 			db,
-		}))
+		});
+
+		services.register_metrics();
+
+		Ok(services)
 	}
 
 	pub async fn start(self: &Arc<Self>) -> Result<Arc<Self>> {
@@ -211,61 +212,6 @@ impl Services {
 			.await
 	}
 
-	/// Refresh the Prometheus gauges which are sampled rather than recorded
-	/// as events happen.
-	pub fn update_metrics(&self) -> Result {
-		let prometheus = &self.server.metrics.prometheus;
-
-		prometheus
-			.sender_queue_depth
-			.set(i64::try_from(self.sending.queue_depth())?);
-
-		let now = millis_since_unix_epoch();
-		let unhealthy = self
-			.federation
-			.remote_health
-			.read()
-			.values()
-			.filter(|&&(_, next_retry)| now < next_retry)
-			.count();
-		prometheus
-			.federation_unhealthy_destinations
-			.set(i64::try_from(unhealthy)?);
-		prometheus
-			.federation_stale_destinations
-			.set(i64::try_from(self.federation.stale_destinations.read().len())?);
-
-		let usage = self.db.db.memory_usage_stats()?;
-		for (name, bytes) in [
-			("mem_tables", usage.mem_table_total),
-			("mem_tables_unflushed", usage.mem_table_unflushed),
-			("table_readers", usage.mem_table_readers_total),
-			("row_cache", usage.row_cache),
-		] {
-			prometheus
-				.database_memory
-				.get_or_create(&NameLabels { name })
-				.set(i64::try_from(bytes)?);
-		}
-		for (column, bytes) in usage.col_caches {
-			prometheus
-				.database_block_cache
-				.get_or_create(&ColumnLabels { column })
-				.set(i64::try_from(bytes)?);
-		}
-
-		let (auth_chain, _) = self.rooms.auth_chain.get_cache_usage();
-		let stateinfo = self.rooms.state_compressor.stateinfo_cache.lock().len();
-		for (name, entries) in [("auth_chain", auth_chain), ("stateinfo", stateinfo)] {
-			prometheus
-				.cache_entries
-				.get_or_create(&NameLabels { name })
-				.set(i64::try_from(entries)?);
-		}
-
-		Ok(())
-	}
-
 	fn interrupt(&self) {
 		debug!("Interrupting services...");
 		for (name, (service, ..)) in self.service.read().iter() {