From f14c401f39c2345803ac02f446922f169a3d2e67 Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Thu, 24 Sep 2026 10:44:15 +0200 Subject: [PATCH] feat: Add Prometheus metrics endpoint --- Cargo.lock | 48 +++- Cargo.toml | 3 + changelog.d/prometheus-metrics.feature.md | 1 + conduwuit-example.toml | 13 ++ docs/guides/_meta.json | 10 + docs/guides/metrics.mdx | 49 ++++ src/api/client/unversioned.rs | 38 +++- src/api/router.rs | 4 + src/core/Cargo.toml | 1 + src/core/config/check.rs | 8 + src/core/config/mod.rs | 14 ++ src/core/metrics/mod.rs | 15 +- src/core/metrics/prometheus.rs | 261 ++++++++++++++++++++++ src/core/utils/string.rs | 11 + src/core/utils/string/tests.rs | 7 + src/database/engine/memory_usage.rs | 40 +++- src/router/layers.rs | 2 +- src/router/request.rs | 41 +++- src/service/federation/execute.rs | 22 +- src/service/pusher/webpush/helpers.rs | 17 +- src/service/pusher/webpush/mod.rs | 8 +- src/service/sending/dest.rs | 9 + src/service/sending/mod.rs | 4 + src/service/sending/sender.rs | 12 + src/service/services.rs | 60 ++++- 25 files changed, 652 insertions(+), 46 deletions(-) create mode 100644 changelog.d/prometheus-metrics.feature.md create mode 100644 docs/guides/metrics.mdx create mode 100644 src/core/metrics/prometheus.rs diff --git a/Cargo.lock b/Cargo.lock index a1ff5302c..be75970ec 100644 --- a/Cargo.lock +++ b/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]] diff --git a/Cargo.toml b/Cargo.toml index 8f4ffde59..a05f253d0 100644 --- a/Cargo.toml +++ b/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" diff --git a/changelog.d/prometheus-metrics.feature.md b/changelog.d/prometheus-metrics.feature.md new file mode 100644 index 000000000..c85e61979 --- /dev/null +++ b/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. diff --git a/conduwuit-example.toml b/conduwuit-example.toml index d0bf4e51f..77969ee04 100644 --- a/conduwuit-example.toml +++ b/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 ` 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], diff --git a/docs/guides/_meta.json b/docs/guides/_meta.json index 39f11fdcd..604e54fdb 100644 --- a/docs/guides/_meta.json +++ b/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" } ] diff --git a/docs/guides/metrics.mdx b/docs/guides/metrics.mdx new file mode 100644 index 000000000..b3873ac4b --- /dev/null +++ b/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 | diff --git a/src/api/client/unversioned.rs b/src/api/client/unversioned.rs index 8f5233aaf..d529d004d 100644 --- a/src/api/client/unversioned.rs +++ b/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, + headers: HeaderMap, +) -> Result { + 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, + )) +} diff --git a/src/api/router.rs b/src/api/router.rs index de2736e2b..5fbe21134 100644 --- a/src/api/router.rs +++ b/src/api/router.rs @@ -199,6 +199,10 @@ pub fn build(router: Router, state: State) -> Router { .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) diff --git a/src/core/Cargo.toml b/src/core/Cargo.toml index 674355770..b61269b8b 100644 --- a/src/core/Cargo.toml +++ b/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 diff --git a/src/core/config/check.rs b/src/core/config/check.rs index 6990abcff..1e3efdb5e 100644 --- a/src/core/config/check.rs +++ b/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", diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index e59077d78..2a239a731 100644 --- a/src/core/config/mod.rs +++ b/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 ` header. + /// + /// display: sensitive + pub prometheus_metrics_token: Option, + /// 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], diff --git a/src/core/metrics/mod.rs b/src/core/metrics/mod.rs index 8f7a55718..91cb93b86 100644 --- a/src/core/metrics/mod.rs +++ b/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, @@ -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()), diff --git a/src/core/metrics/prometheus.rs b/src/core/metrics/prometheus.rs new file mode 100644 index 000000000..56f57d8eb --- /dev/null +++ b/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, + + pub federation_request_duration: Family, + pub federation_unhealthy_destinations: Gauge, + pub federation_stale_destinations: Gauge, + + pub sender_transactions: Family, + pub sender_queue_depth: Gauge, + + pub database_memory: Family, + pub database_block_cache: Family, + pub cache_entries: Family, +} + +impl Prometheus { + pub(super) fn new(runtime_metrics: Option) -> 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 { + 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"); + } +} diff --git a/src/core/utils/string.rs b/src/core/utils/string.rs index 7d81903db..216ed35cb 100644 --- a/src/core/utils/string.rs +++ b/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 { diff --git a/src/core/utils/string/tests.rs b/src/core/utils/string/tests.rs index e8c17de6d..80302d977 100644 --- a/src/core/utils/string/tests.rs +++ b/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")); +} diff --git a/src/database/engine/memory_usage.rs b/src/database/engine/memory_usage.rs index d5e5945d8..b3fe44c54 100644 --- a/src/database/engine/memory_usage.rs +++ b/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 { 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 { + 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::>()?; + + 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, + }) + } } diff --git a/src/router/layers.rs b/src/router/layers.rs index aeec5e84c..d2ec6c0d8 100644 --- a/src/router/layers.rs +++ b/src/router/layers.rs @@ -241,7 +241,7 @@ fn request_path_str(request: &http::Request) -> &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)) diff --git a/src/router/request.rs b/src/router/request.rs index 2f7151b1c..7c99b0668 100644 --- a/src/router/request.rs +++ b/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>, @@ -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::() + .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( diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index 52af916e5..59598c71a 100644 --- a/src/service/federation/execute.rs +++ b/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::(dest, actual, &method, &url, response) diff --git a/src/service/pusher/webpush/helpers.rs b/src/service/pusher/webpush/helpers.rs index 7ac455e3f..ae2a28f11 100644 --- a/src/service/pusher/webpush/helpers.rs +++ b/src/service/pusher/webpush/helpers.rs @@ -37,14 +37,6 @@ pub(crate) fn decode_auth(auth: &str) -> Result { 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> { 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")); - } } diff --git a/src/service/pusher/webpush/mod.rs b/src/service/pusher/webpush/mod.rs index f09f65890..6f66d6fdb 100644 --- a/src/service/pusher/webpush/mod.rs +++ b/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); diff --git a/src/service/sending/dest.rs b/src/service/sending/dest.rs index 878b0fc41..1c4987e4d 100644 --- a/src/service/sending/dest.rs +++ b/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 { diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index 9cdd0242a..6b913d5b9 100644 --- a/src/service/sending/mod.rs +++ b/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; diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index cac87bc78..9e4b3da0b 100644 --- a/src/service/sending/sender.rs +++ b/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)) => { diff --git a/src/service/services.rs b/src/service/services.rs index 0d6c76676..9303b92ee 100644 --- a/src/service/services.rs +++ b/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() {