feat: Add Prometheus metrics endpoint

Cargo.lock

@@ -1037,6 +1037,7 @@ dependencies = [
  "num-traits",
  "openidconnect",
  "parking_lot",
+ "prometheus-client",
  "rand 0.10.3",
  "regex",
  "reqwest 0.13.5",
@@ -1831,6 +1832,12 @@ dependencies = [
  "litrs",
 ]
 
+[[package]]
+name = "dtoa"
+version = "1.0.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4c3cf4824e2d5f025c7b531afcb2325364084a16806f6d47fbc1f5fbd9960590"
+
 [[package]]
 name = "dtor"
 version = "1.0.6"
@@ -2009,7 +2016,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -4502,6 +4509,29 @@ dependencies = [
  "yansi",
 ]
 
+[[package]]
+name = "prometheus-client"
+version = "0.25.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1784dc11a05f5a8f57c4a62915686be140f24f70301290b997e317241fa9ffc2"
+dependencies = [
+ "dtoa",
+ "itoa",
+ "parking_lot",
+ "prometheus-client-derive-encode",
+]
+
+[[package]]
+name = "prometheus-client-derive-encode"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "01e34894696ff94f64a20c2c373a6440903e9c2789a303d68ec6e6f953f890e4"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.6",
+]
+
 [[package]]
 name = "prost"
 version = "0.14.4"
@@ -4619,7 +4649,7 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -5213,7 +5243,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -5272,7 +5302,7 @@ dependencies = [
  "security-framework",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]
@@ -5912,9 +5942,9 @@ dependencies = [
 
 [[package]]
 name = "smallvec"
-version = "1.16.2"
+version = "1.16.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e"
+checksum = "ba467056f1b547ed52077911161fc86985becbc60e8e1857c8a144dab0def891"
 dependencies = [
  "serde",
 ]
@@ -6095,9 +6125,9 @@ dependencies = [
 
 [[package]]
 name = "termimad"
-version = "0.35.5"
+version = "0.35.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "092b11fb38404fa873f019f8e5f300fa5787f381b7a58b923546d94b5a9065fd"
+checksum = "2374179aac8b77011a1aefe2f44d0c34eb680c9b4bd1c369a88991262c60c85a"
 dependencies = [
  "coolor",
  "crokey",
@@ -7081,7 +7111,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]

Cargo.toml

@@ -260,6 +260,9 @@ features = [
 [workspace.dependencies.tokio-metrics]
 version = "0.5.0"
 
+[workspace.dependencies.prometheus-client]
+version = "0.25.1"
+
 [workspace.dependencies.libloading]
 version = "0.9.0"
 

changelog.d/prometheus-metrics.feature.md

@@ -0,0 +1 @@
+Added an opt-in Prometheus metrics endpoint at `/_continuwuity/metrics` covering HTTP requests, federation, the sender queue, database and cache memory, and the tokio runtime. Contributed by @eleboucher.

conduwuit-example.toml

@@ -837,6 +837,19 @@
 #
 #otlp_protocol = "http"
 
+# Enable the Prometheus metrics endpoint at `/_continuwuity/metrics`.
+#
+# The endpoint is served on the same listener as the rest of the server.
+# Set `prometheus_metrics_token` or restrict access at your reverse proxy
+# to avoid exposing it publicly.
+#
+#allow_prometheus_metrics = false
+
+# Bearer token required to scrape the Prometheus metrics endpoint. When
+# set, requests must include an `Authorization: Bearer <token>` header.
+#
+#prometheus_metrics_token =
+
 # If the 'perf_measurements' compile-time feature is enabled, enables
 # collecting folded stack trace profile of tracing spans using
 # tracing_flame. The resulting profile can be visualized with inferno[1],

docs/guides/_meta.json

@@ -13,5 +13,15 @@
         "type": "file",
         "name": "performance",
         "label": "Performance tuning"
+    },
+    {
+        "type": "file",
+        "name": "metrics",
+        "label": "Prometheus metrics"
+    },
+    {
+        "type": "file",
+        "name": "oidc",
+        "label": "Delegated authentication"
     }
 ]

docs/guides/metrics.mdx

@@ -0,0 +1,49 @@
+# Prometheus metrics
+
+Continuwuity can expose metrics in the OpenMetrics text format at `/_continuwuity/metrics`. The endpoint is disabled by default.
+
+## Configuration
+
+```toml
+[global]
+allow_prometheus_metrics = true
+prometheus_metrics_token = "a-long-random-secret"
+```
+
+The endpoint is served on the same listener as the rest of the server. Either set `prometheus_metrics_token` or block `/_continuwuity/metrics` at your reverse proxy so it is not publicly readable.
+
+## Scrape config
+
+```yaml
+scrape_configs:
+  - job_name: continuwuity
+    scheme: https
+    metrics_path: /_continuwuity/metrics
+    authorization:
+      credentials: a-long-random-secret
+    static_configs:
+      - targets: ["matrix.example.com"]
+```
+
+## Exported metrics
+
+All metrics are prefixed with `continuwuity_`.
+
+| Metric | Type | Labels | Description |
+| --- | --- | --- | --- |
+| `build_info` | info | `version` | Running server version |
+| `http_request_duration_seconds` | histogram | `method`, `route`, `status` | Handled HTTP requests, by route template |
+| `http_requests_in_flight` | gauge | | Requests currently being handled |
+| `federation_request_duration_seconds` | histogram | `outcome` | Outgoing federation requests |
+| `federation_unhealthy_destinations` | gauge | | Remote servers currently in retry backoff |
+| `federation_stale_destinations` | gauge | | Remote servers with a stale destination cache entry |
+| `sender_transactions_total` | counter | `destination`, `outcome` | Outgoing federation, appservice and push transactions |
+| `sender_queue_depth` | gauge | | Messages waiting for the sender workers |
+| `database_memory_bytes` | gauge | `name` | RocksDB memtable, table reader and row cache usage |
+| `database_block_cache_bytes` | gauge | `column` | RocksDB block cache usage |
+| `cache_entries` | gauge | `name` | Entries in the auth chain and state info caches |
+| `tokio_workers` | gauge | | Tokio runtime worker threads |
+| `tokio_alive_tasks` | gauge | | Alive tokio tasks |
+| `tokio_global_queue_depth` | gauge | | Tasks in the tokio global queue |
+| `tokio_worker_busy_seconds_total` | counter | | Time worker threads spent busy |
+| `tokio_worker_parks_total` | counter | | Number of times worker threads parked |

src/api/client/unversioned.rs

@@ -1,9 +1,11 @@
 use axum::{Json, extract::State, response::IntoResponse};
 use conduwuit::{
-	Result,
+	Err, Result,
 	matrix::versions::{unstable_features, versions},
+	utils::string::constant_time_eq,
 };
 use futures::StreamExt;
+use http::{HeaderMap, header};
 use ruma::{api::client::discovery::get_supported_versions, assign};
 
 use crate::Ruma;
@@ -75,3 +77,37 @@ pub(crate) async fn continuwuity_local_user_count(
 		"count": user_count
 	})))
 }
+
+/// # `GET /_continuwuity/metrics`
+///
+/// Prometheus metrics in the OpenMetrics text format. Only routed when
+/// `allow_prometheus_metrics` is enabled.
+pub(crate) async fn continuwuity_metrics(
+	State(services): State<crate::State>,
+	headers: HeaderMap,
+) -> Result<impl IntoResponse> {
+	if let Some(token) = &services.config.prometheus_metrics_token {
+		let authorized = headers
+			.get(header::AUTHORIZATION)
+			.and_then(|value| value.to_str().ok())
+			.and_then(|value| value.split_once(' '))
+			.is_some_and(|(scheme, provided)| {
+				scheme.eq_ignore_ascii_case("bearer") && constant_time_eq(provided, token)
+			});
+
+		if !authorized {
+			return Err!(Request(Forbidden("Invalid metrics token")));
+		}
+	}
+
+	services.update_metrics()?;
+	let body = services.server.metrics.prometheus.encode()?;
+
+	Ok((
+		[(
+			header::CONTENT_TYPE,
+			"application/openmetrics-text; version=1.0.0; charset=utf-8",
+		)],
+		body,
+	))
+}

src/api/router.rs

@@ -199,6 +199,10 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 		.ruma_route(&admin::rooms::ban::ban_room)
 		.ruma_route(&admin::rooms::list::list_rooms);
 
+	if config.allow_prometheus_metrics {
+		router = router.route("/_continuwuity/metrics", get(client::continuwuity_metrics));
+	}
+
 	if config.allow_federation {
 		router = router
 			.ruma_route(&server::get_server_version_route)

src/core/Cargo.toml

@@ -113,6 +113,7 @@ tracing-subscriber.workspace = true
 tracing.workspace = true
 url.workspace = true
 parking_lot.workspace = true
+prometheus-client.workspace = true
 lock_api.workspace = true
 hyper-util.workspace = true
 resolvematrix.workspace = true

src/core/config/check.rs

@@ -194,6 +194,14 @@ pub fn check(config: &Config) -> Result {
 		);
 	}
 
+	if config.allow_prometheus_metrics && config.prometheus_metrics_token.is_none() {
+		warn!(
+			"The Prometheus metrics endpoint is enabled without `prometheus_metrics_token`. \
+			 Anyone who can reach /_continuwuity/metrics can read server metrics unless access \
+			 is restricted at your reverse proxy."
+		);
+	}
+
 	if config.allow_outgoing_presence && !config.allow_local_presence {
 		return Err!(Config(
 			"allow_local_presence",

src/core/config/mod.rs

@@ -1026,6 +1026,20 @@ pub struct Config {
 	#[serde(default = "default_otlp_protocol")]
 	pub otlp_protocol: String,
 
+	/// Enable the Prometheus metrics endpoint at `/_continuwuity/metrics`.
+	///
+	/// The endpoint is served on the same listener as the rest of the server.
+	/// Set `prometheus_metrics_token` or restrict access at your reverse proxy
+	/// to avoid exposing it publicly.
+	#[serde(default)]
+	pub allow_prometheus_metrics: bool,
+
+	/// Bearer token required to scrape the Prometheus metrics endpoint. When
+	/// set, requests must include an `Authorization: Bearer <token>` header.
+	///
+	/// display: sensitive
+	pub prometheus_metrics_token: Option<String>,
+
 	/// If the 'perf_measurements' compile-time feature is enabled, enables
 	/// collecting folded stack trace profile of tracing spans using
 	/// tracing_flame. The resulting profile can be visualized with inferno[1],

src/core/metrics/mod.rs

@@ -1,3 +1,5 @@
+mod prometheus;
+
 use std::sync::atomic::AtomicU32;
 
 use tokio::runtime;
@@ -5,6 +7,11 @@ use tokio_metrics::TaskMonitor;
 #[cfg(tokio_unstable)]
 use tokio_metrics::{RuntimeIntervals, RuntimeMonitor};
 
+pub use self::prometheus::{
+	ColumnLabels, HttpLabels, NameLabels, OutcomeLabels, Prometheus, TransactionLabels,
+	method_label,
+};
+
 pub struct Metrics {
 	_runtime: Option<runtime::Handle>,
 
@@ -22,6 +29,8 @@ pub struct Metrics {
 	pub requests_handle_active: AtomicU32,
 	pub requests_handle_finished: AtomicU32,
 	pub requests_panic: AtomicU32,
+
+	pub prometheus: Prometheus,
 }
 
 impl Metrics {
@@ -33,10 +42,14 @@ impl Metrics {
 		#[cfg(tokio_unstable)]
 		let runtime_intervals = runtime_monitor.as_ref().map(RuntimeMonitor::intervals);
 
+		let runtime_metrics = runtime.as_ref().map(runtime::Handle::metrics);
+
 		Self {
 			_runtime: runtime.clone(),
 
-			runtime_metrics: runtime.as_ref().map(runtime::Handle::metrics),
+			prometheus: Prometheus::new(runtime_metrics.clone()),
+
+			runtime_metrics,
 
 			task_monitor: runtime.map(|_| TaskMonitor::new()),
 

src/core/metrics/prometheus.rs

@@ -0,0 +1,261 @@
+use std::time::Duration;
+
+use prometheus_client::{
+	collector::Collector,
+	encoding::{DescriptorEncoder, EncodeLabelSet, EncodeMetric, text},
+	metrics::{
+		counter::{ConstCounter, Counter},
+		family::Family,
+		gauge::{ConstGauge, Gauge},
+		histogram::{Histogram, exponential_buckets},
+		info::Info,
+	},
+	registry::{Registry, Unit},
+};
+use tokio::runtime;
+
+use crate::Result;
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct HttpLabels {
+	pub method: &'static str,
+	pub route: String,
+	pub status: u16,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct OutcomeLabels {
+	pub outcome: &'static str,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct TransactionLabels {
+	pub destination: &'static str,
+	pub outcome: &'static str,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct NameLabels {
+	pub name: &'static str,
+}
+
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)]
+pub struct ColumnLabels {
+	pub column: String,
+}
+
+pub struct Prometheus {
+	registry: Registry,
+
+	pub http_requests_in_flight: Gauge,
+	pub http_request_duration: Family<HttpLabels, Histogram>,
+
+	pub federation_request_duration: Family<OutcomeLabels, Histogram>,
+	pub federation_unhealthy_destinations: Gauge,
+	pub federation_stale_destinations: Gauge,
+
+	pub sender_transactions: Family<TransactionLabels, Counter>,
+	pub sender_queue_depth: Gauge,
+
+	pub database_memory: Family<NameLabels, Gauge>,
+	pub database_block_cache: Family<ColumnLabels, Gauge>,
+	pub cache_entries: Family<NameLabels, Gauge>,
+}
+
+impl Prometheus {
+	pub(super) fn new(runtime_metrics: Option<runtime::RuntimeMetrics>) -> Self {
+		let mut this = Self {
+			registry: Registry::with_prefix("continuwuity"),
+			http_requests_in_flight: Gauge::default(),
+			http_request_duration: Family::new_with_constructor(duration_histogram),
+			federation_request_duration: Family::new_with_constructor(duration_histogram),
+			federation_unhealthy_destinations: Gauge::default(),
+			federation_stale_destinations: Gauge::default(),
+			sender_transactions: Family::default(),
+			sender_queue_depth: Gauge::default(),
+			database_memory: Family::default(),
+			database_block_cache: Family::default(),
+			cache_entries: Family::default(),
+		};
+
+		this.registry.register(
+			"http_requests_in_flight",
+			"Number of client and federation requests currently being handled",
+			this.http_requests_in_flight.clone(),
+		);
+		this.registry.register_with_unit(
+			"http_request_duration",
+			"Duration of handled HTTP requests",
+			Unit::Seconds,
+			this.http_request_duration.clone(),
+		);
+		this.registry.register_with_unit(
+			"federation_request_duration",
+			"Duration of outgoing federation requests",
+			Unit::Seconds,
+			this.federation_request_duration.clone(),
+		);
+		this.registry.register(
+			"federation_unhealthy_destinations",
+			"Number of remote servers currently in retry backoff",
+			this.federation_unhealthy_destinations.clone(),
+		);
+		this.registry.register(
+			"federation_stale_destinations",
+			"Number of remote servers with a stale destination cache entry",
+			this.federation_stale_destinations.clone(),
+		);
+		this.registry.register(
+			"sender_transactions",
+			"Outgoing transactions completed by the sender, by destination type and outcome",
+			this.sender_transactions.clone(),
+		);
+		this.registry.register(
+			"sender_queue_depth",
+			"Number of messages waiting in the sender worker channels",
+			this.sender_queue_depth.clone(),
+		);
+		this.registry.register_with_unit(
+			"database_memory",
+			"RocksDB memory usage by component",
+			Unit::Bytes,
+			this.database_memory.clone(),
+		);
+		this.registry.register_with_unit(
+			"database_block_cache",
+			"RocksDB block cache usage by column family",
+			Unit::Bytes,
+			this.database_block_cache.clone(),
+		);
+		this.registry.register(
+			"cache_entries",
+			"Number of entries held in in-memory caches",
+			this.cache_entries.clone(),
+		);
+
+		this.registry.register(
+			"build",
+			"Build information",
+			Info::new([("version", crate::version())]),
+		);
+
+		if let Some(runtime_metrics) = runtime_metrics {
+			this.registry
+				.register_collector(Box::new(RuntimeCollector(runtime_metrics)));
+		}
+
+		this
+	}
+
+	pub fn encode(&self) -> Result<String> {
+		let mut buf = String::new();
+		text::encode(&mut buf, &self.registry)?;
+		Ok(buf)
+	}
+}
+
+/// Collapse non-standard methods so arbitrary client input cannot create new
+/// label values.
+#[must_use]
+pub fn method_label(method: &http::Method) -> &'static str {
+	match *method {
+		| http::Method::GET => "GET",
+		| http::Method::HEAD => "HEAD",
+		| http::Method::POST => "POST",
+		| http::Method::PUT => "PUT",
+		| http::Method::DELETE => "DELETE",
+		| http::Method::OPTIONS => "OPTIONS",
+		| http::Method::PATCH => "PATCH",
+		| _ => "OTHER",
+	}
+}
+
+fn duration_histogram() -> Histogram { Histogram::new(exponential_buckets(0.005, 2.0, 14)) }
+
+#[derive(Debug)]
+struct RuntimeCollector(runtime::RuntimeMetrics);
+
+impl Collector for RuntimeCollector {
+	fn encode(&self, mut encoder: DescriptorEncoder<'_>) -> Result<(), std::fmt::Error> {
+		let metrics = &self.0;
+		let workers = metrics.num_workers();
+		let busy: Duration = (0..workers)
+			.map(|worker| metrics.worker_total_busy_duration(worker))
+			.sum();
+		let parks: u64 = (0..workers)
+			.map(|worker| metrics.worker_park_count(worker))
+			.sum();
+
+		let gauges = [
+			("tokio_workers", "Number of tokio runtime worker threads.", workers),
+			(
+				"tokio_alive_tasks",
+				"Number of alive tasks in the tokio runtime.",
+				metrics.num_alive_tasks(),
+			),
+			(
+				"tokio_global_queue_depth",
+				"Number of tasks in the tokio runtime global queue.",
+				metrics.global_queue_depth(),
+			),
+		];
+		for (name, help, value) in gauges {
+			let gauge = ConstGauge::new(value);
+			gauge.encode(encoder.encode_descriptor(name, help, None, gauge.metric_type())?)?;
+		}
+
+		let busy = ConstCounter::new(busy.as_secs_f64());
+		busy.encode(encoder.encode_descriptor(
+			"tokio_worker_busy",
+			"Total time tokio worker threads spent busy.",
+			Some(&Unit::Seconds),
+			busy.metric_type(),
+		)?)?;
+
+		let parks = ConstCounter::new(parks);
+		parks.encode(encoder.encode_descriptor(
+			"tokio_worker_parks",
+			"Total number of times tokio worker threads parked.",
+			None,
+			parks.metric_type(),
+		)?)?;
+
+		Ok(())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::{HttpLabels, Prometheus, method_label};
+
+	#[tokio::test]
+	async fn encodes_openmetrics() {
+		let prometheus = Prometheus::new(Some(tokio::runtime::Handle::current().metrics()));
+		prometheus
+			.http_request_duration
+			.get_or_create(&HttpLabels {
+				method: method_label(&http::Method::GET),
+				route: "/_matrix/client/versions".to_owned(),
+				status: 200,
+			})
+			.observe(0.01);
+
+		let body = prometheus.encode().unwrap();
+		for name in [
+			"continuwuity_http_request_duration_seconds_count{method=\"GET\",route=\"/_matrix/\
+			 client/versions\",status=\"200\"} 1",
+			"continuwuity_http_requests_in_flight 0",
+			"continuwuity_build_info{version=",
+			"# TYPE continuwuity_tokio_worker_busy_seconds counter",
+		] {
+			assert!(body.contains(name), "missing {name:?} in:\n{body}");
+		}
+		assert!(body.ends_with("# EOF\n"));
+	}
+
+	#[test]
+	fn method_label_is_bounded() {
+		let custom = http::Method::from_bytes(b"BREW").unwrap();
+		assert_eq!(method_label(&custom), "OTHER");
+	}
+}

src/core/utils/string.rs

@@ -49,6 +49,17 @@ where
 	Ok(out)
 }
 
+/// Compare two strings without short-circuiting on the first mismatch, for
+/// checking secrets.
+#[must_use]
+pub fn constant_time_eq(a: &str, b: &str) -> bool {
+	a.len() == b.len()
+		&& a.bytes()
+			.zip(b.bytes())
+			.fold(0_u8, |acc, (x, y)| acc | (x ^ y))
+			== 0
+}
+
 #[inline]
 #[must_use]
 pub fn camel_to_snake_string(s: &str) -> String {

src/core/utils/string/tests.rs

@@ -68,3 +68,10 @@ fn between_infallible() {
 	assert_eq!("\"foo".between_infallible(("\"", "\"")), "\"foo");
 	assert_eq!("foo".between_infallible(("\"", "\"")), "foo");
 }
+
+#[test]
+fn constant_time_eq() {
+	assert!(super::constant_time_eq("abc", "abc"));
+	assert!(!super::constant_time_eq("abc", "abz"));
+	assert!(!super::constant_time_eq("abc", "abcd"));
+}

src/database/engine/memory_usage.rs

@@ -5,12 +5,18 @@ use rocksdb::perf::get_memory_usage_stats;
 
 use crate::or_else;
 
+pub struct MemoryUsage {
+	pub mem_table_total: u64,
+	pub mem_table_unflushed: u64,
+	pub mem_table_readers_total: u64,
+	pub row_cache: u64,
+	pub col_caches: Vec<(String, u64)>,
+}
+
 impl super::Engine {
 	pub fn memory_usage(&self) -> Result<String> {
 		let mut res = String::new();
-		let stats =
-			get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
-				.or_else(or_else)?;
+		let stats = self.memory_usage_stats()?;
 		let mibs = |input| f64::from(u32::try_from(input / 1024).unwrap_or(0)) / 1024.0;
 		writeln!(
 			res,
@@ -19,13 +25,35 @@ impl super::Engine {
 			mibs(stats.mem_table_total),
 			mibs(stats.mem_table_unflushed),
 			mibs(stats.mem_table_readers_total),
-			mibs(u64::try_from(self.ctx.row_cache.lock().get_usage())?),
+			mibs(stats.row_cache),
 		)?;
 
-		for (name, cache) in &*self.ctx.col_cache.lock() {
-			writeln!(res, "{name} cache: {:.2} MiB", mibs(u64::try_from(cache.get_usage())?))?;
+		for (name, usage) in stats.col_caches {
+			writeln!(res, "{name} cache: {:.2} MiB", mibs(usage))?;
 		}
 
 		Ok(res)
 	}
+
+	pub fn memory_usage_stats(&self) -> Result<MemoryUsage> {
+		let stats =
+			get_memory_usage_stats(Some(&[&self.db]), Some(&[&*self.ctx.row_cache.lock()]))
+				.or_else(or_else)?;
+
+		let col_caches = self
+			.ctx
+			.col_cache
+			.lock()
+			.iter()
+			.map(|(name, cache)| Ok((name.clone(), u64::try_from(cache.get_usage())?)))
+			.collect::<Result<_>>()?;
+
+		Ok(MemoryUsage {
+			mem_table_total: stats.mem_table_total,
+			mem_table_unflushed: stats.mem_table_unflushed,
+			mem_table_readers_total: stats.mem_table_readers_total,
+			row_cache: u64::try_from(self.ctx.row_cache.lock().get_usage())?,
+			col_caches,
+		})
+	}
 }

src/router/layers.rs

@@ -241,7 +241,7 @@ fn request_path_str<T>(request: &http::Request<T>) -> &str {
 		.as_str()
 }
 
-fn truncated_matched_path(path: &MatchedPath) -> &str {
+pub(crate) fn truncated_matched_path(path: &MatchedPath) -> &str {
 	path.as_str()
 		.rsplit_once(':')
 		.map_or(path.as_str(), |path| path.0.strip_suffix('/').unwrap_or(path.0))

src/router/request.rs

@@ -1,20 +1,26 @@
 use std::{
 	fmt::Debug,
 	sync::{Arc, atomic::Ordering},
-	time::Duration,
+	time::{Duration, Instant},
 };
 
 use axum::{
-	extract::State,
+	extract::{MatchedPath, State},
 	response::{IntoResponse, Response},
 };
-use conduwuit::{Result, debug, debug_error, debug_warn, err, error, trace};
+use conduwuit::{
+	Result, debug, debug_error, debug_warn, err, error,
+	metrics::{HttpLabels, method_label},
+	trace,
+};
 use conduwuit_service::Services;
 use futures::FutureExt;
 use http::{Method, StatusCode, Uri};
 use tokio::time::sleep;
 use tracing::Span;
 
+use crate::layers::truncated_matched_path;
+
 #[tracing::instrument(name = "request", level = "debug", skip_all)]
 pub(crate) async fn handle(
 	State(services): State<Arc<Services>>,
@@ -31,6 +37,16 @@ pub(crate) async fn handle(
 		return Err(StatusCode::SERVICE_UNAVAILABLE);
 	}
 
+	let prometheus = &services.server.metrics.prometheus;
+	prometheus.http_requests_in_flight.inc();
+	conduwuit::defer! {{ prometheus.http_requests_in_flight.dec(); }};
+
+	let start = Instant::now();
+	let route = req
+		.extensions()
+		.get::<MatchedPath>()
+		.map_or("unmatched", truncated_matched_path)
+		.to_owned();
 	let uri = req.uri().clone();
 	let method = req.method().clone();
 	let services_ = services.clone();
@@ -49,9 +65,24 @@ pub(crate) async fn handle(
 		}
 	});
 
-	task.await
+	let result = task
+		.await
 		.map_err(unhandled)
-		.and_then(move |result| handle_result(&method, &uri, result))
+		.and_then(|result| handle_result(&method, &uri, result));
+
+	let status = result
+		.as_ref()
+		.map_or_else(|status| *status, Response::status);
+	prometheus
+		.http_request_duration
+		.get_or_create(&HttpLabels {
+			method: method_label(&method),
+			route,
+			status: status.as_u16(),
+		})
+		.observe(start.elapsed().as_secs_f64());
+
+	result
 }
 
 #[tracing::instrument(

src/service/federation/execute.rs

@@ -1,8 +1,8 @@
 use std::{borrow::Cow, fmt::Debug, mem, time::Instant};
 
 use conduwuit::{
-	Err, Error, Result, debug, debug_error, debug_info, debug_warn, err, trace,
-	utils::response::LimitReadExt,
+	Err, Error, Result, debug, debug_error, debug_info, debug_warn, err, metrics::OutcomeLabels,
+	trace, utils::response::LimitReadExt,
 };
 use ipaddress::IPAddress;
 use reqwest::{Client, Method, Request, Response, Url};
@@ -175,12 +175,26 @@ impl super::Service {
 		);
 		let start = Instant::now();
 		let response = client.execute(request).await;
+		let elapsed = start.elapsed();
 		debug_info!(
-			"Request finished in {:?}: {} matrix-federation://{dest}{}",
-			start.elapsed(),
+			"Request finished in {elapsed:?}: {} matrix-federation://{dest}{}",
 			method.as_str(),
 			url.path(),
 		);
+		let outcome = match &response {
+			| Ok(response) if response.status().is_success() => "success",
+			| Ok(response) if response.status().is_client_error() => "client_error",
+			| Ok(response) if response.status().is_server_error() => "server_error",
+			| Ok(_) => "other",
+			| Err(_) => "network_error",
+		};
+		self.services
+			.server
+			.metrics
+			.prometheus
+			.federation_request_duration
+			.get_or_create(&OutcomeLabels { outcome })
+			.observe(elapsed.as_secs_f64());
 		match response {
 			| Ok(response) =>
 				self.handle_response::<T>(dest, actual, &method, &url, response)

src/service/pusher/webpush/helpers.rs

@@ -37,14 +37,6 @@ pub(crate) fn decode_auth(auth: &str) -> Result<Auth> {
 	Ok(Auth::from(bytes))
 }
 
-pub(super) fn constant_time_eq(a: &str, b: &str) -> bool {
-	a.len() == b.len()
-		&& a.bytes()
-			.zip(b.bytes())
-			.fold(0_u8, |acc, (x, y)| acc | (x ^ y))
-			== 0
-}
-
 pub(super) fn build_payload(notify: &Notification, data: &JsonObject) -> Result<Vec<u8>> {
 	let JsonValue::Object(mut payload) = serde_json::to_value(notify)? else {
 		return Err!("Notification did not serialize to an object");
@@ -117,7 +109,7 @@ mod tests {
 	};
 	use serde_json::{Value as JsonValue, json};
 
-	use super::{MAX_BODY_LENGTH, build_payload, constant_time_eq};
+	use super::{MAX_BODY_LENGTH, build_payload};
 
 	fn payload_of(notify: &Notification, data: &JsonObject) -> JsonObject {
 		let encoded = build_payload(notify, data).unwrap();
@@ -198,11 +190,4 @@ mod tests {
 
 		build_payload(&Notification::new(vec![]), &data).unwrap_err();
 	}
-
-	#[test]
-	fn token_comparison() {
-		assert!(constant_time_eq("abc", "abc"));
-		assert!(!constant_time_eq("abc", "abz"));
-		assert!(!constant_time_eq("abc", "abcd"));
-	}
 }

src/service/pusher/webpush/mod.rs

@@ -3,7 +3,11 @@ mod vapid;
 
 use std::time::{Duration, SystemTime};
 
-use conduwuit_core::{Err, Result, debug, debug_warn, err, utils, warn};
+use conduwuit_core::{
+	Err, Result, debug, debug_warn, err,
+	utils::{self, string::constant_time_eq},
+	warn,
+};
 use conduwuit_database::{Deserialized, Json};
 use futures::StreamExt;
 pub(super) use helpers::{decode_auth, decode_pushkey};
@@ -20,7 +24,7 @@ use tokio::time::sleep;
 pub(super) use vapid::Vapid;
 use web_push_native::WebPushBuilder;
 
-use self::helpers::{build_payload, constant_time_eq};
+use self::helpers::build_payload;
 use super::{Service, WEBPUSH_SCHEMES};
 
 const TTL: Duration = Duration::from_hours(48);

src/service/sending/dest.rs

@@ -10,6 +10,15 @@ pub enum Destination {
 }
 
 impl Destination {
+	#[must_use]
+	pub(super) fn kind(&self) -> &'static str {
+		match self {
+			| Self::Appservice(_) => "appservice",
+			| Self::Push(..) => "push",
+			| Self::Federation(_) => "federation",
+		}
+	}
+
 	/// Gets the prefix for this destination.
 	#[must_use]
 	pub(super) fn get_prefix(&self) -> Vec<u8> {

src/service/sending/mod.rs

@@ -486,6 +486,10 @@ impl Service {
 		sender.send(msg).map_err(|e| err!("{e}"))
 	}
 
+	pub fn queue_depth(&self) -> usize {
+		self.channels.iter().map(|(sender, _)| sender.len()).sum()
+	}
+
 	pub(super) fn shard_id(&self, dest: &Destination) -> usize {
 		if self.channels.len() <= 1 {
 			return 0;

src/service/sending/sender.rs

@@ -13,6 +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,
 	result::LogErr,
 	utils::{
 		ReadyExt, calculate_hash,
@@ -139,6 +140,17 @@ 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)) => {

src/service/services.rs

@@ -1,7 +1,10 @@
 use std::{any::Any, collections::BTreeMap, sync::Arc};
 
 use conduwuit::{
-	Result, Server, SyncRwLock, debug, debug_info, error, info, trace, utils::stream::IterStream,
+	Result, Server, SyncRwLock, debug, debug_info, error, info,
+	metrics::{ColumnLabels, NameLabels},
+	trace,
+	utils::{millis_since_unix_epoch, stream::IterStream},
 };
 use database::Database;
 use futures::{Stream, StreamExt, TryStreamExt};
@@ -208,6 +211,61 @@ 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() {