diff --git a/Cargo.lock b/Cargo.lock index 88f9733d4..e0c2f3619 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -916,8 +916,10 @@ dependencies = [ "log", "opentelemetry", "opentelemetry-otlp", + "opentelemetry-prometheus", "opentelemetry_sdk", "parking_lot", + "prometheus", "reqwest 0.13.5", "rustls", "sentry", @@ -1036,8 +1038,11 @@ dependencies = [ "nix", "num-traits", "openidconnect", + "opentelemetry", + "opentelemetry-prometheus", + "opentelemetry_sdk", "parking_lot", - "prometheus-client", + "prometheus", "rand 0.10.3", "regex", "reqwest 0.13.5", @@ -1832,12 +1837,6 @@ 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" @@ -2016,7 +2015,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -3798,7 +3797,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http", @@ -4080,6 +4079,19 @@ dependencies = [ "tonic-types", ] +[[package]] +name = "opentelemetry-prometheus" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "274c2e3a2ad55524484500d5a78dffbfee23e4a1e4c3d3fcbd881b61b033e6df" +dependencies = [ + "once_cell", + "opentelemetry", + "opentelemetry_sdk", + "prometheus", + "tracing", +] + [[package]] name = "opentelemetry-proto" version = "0.33.0" @@ -4511,26 +4523,17 @@ dependencies = [ ] [[package]] -name = "prometheus-client" -version = "0.25.1" +name = "prometheus" +version = "0.14.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1784dc11a05f5a8f57c4a62915686be140f24f70301290b997e317241fa9ffc2" +checksum = "3ca5326d8d0b950a9acd87e6a3f94745394f62e4dae1b1ee22b2bc0c394af43a" dependencies = [ - "dtoa", - "itoa", + "cfg-if", + "fnv", + "lazy_static", + "memchr", "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", + "thiserror 2.0.21", ] [[package]] @@ -4650,7 +4653,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5244,7 +5247,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5303,7 +5306,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -5613,7 +5616,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8050abb251097357e24aff63ba2c52a6309ecb7d23a5474023df960a02694d8" dependencies = [ "annotate-snippets", - "base64 0.21.7", + "base64 0.23.1", "encoding_rs_io", "granit-parser", "nohash-hasher", @@ -7112,7 +7115,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index e08d2639a..a3b63cd7d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -260,9 +260,6 @@ features = [ [workspace.dependencies.tokio-metrics] version = "0.5.0" -[workspace.dependencies.prometheus-client] -version = "0.25.1" - [workspace.dependencies.libloading] version = "0.9.0" @@ -431,6 +428,13 @@ version = "0.34.0" version = "0.33.0" features = ["rt-tokio"] +[workspace.dependencies.opentelemetry-prometheus] +version = "0.33.0" + +[workspace.dependencies.prometheus] +version = "0.14.0" +default-features = false + [workspace.dependencies.opentelemetry-otlp] version = "0.33.0" features = ["http", "grpc-tonic", "trace", "logs", "metrics"] diff --git a/changelog.d/prometheus-metrics.feature.md b/changelog.d/prometheus-metrics.feature.md index 68128f6a0..719c6f7fc 100644 --- a/changelog.d/prometheus-metrics.feature.md +++ b/changelog.d/prometheus-metrics.feature.md @@ -1 +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, along with a Grafana dashboard. Contributed by @eleboucher. +Added opt-in OpenTelemetry metrics, exported over OTLP or scraped by Prometheus from `/_continuwuity/metrics`, and a Grafana dashboard. Contributed by @eleboucher. diff --git a/conduwuit-example.toml b/conduwuit-example.toml index 77969ee04..6d83e5a2d 100644 --- a/conduwuit-example.toml +++ b/conduwuit-example.toml @@ -837,6 +837,15 @@ # #otlp_protocol = "http" +# Enable OpenTelemetry OTLP metrics export, using the same endpoint and +# `otlp_protocol` as traces. +# +# The export interval can be changed with the +# OTEL_METRIC_EXPORT_INTERVAL environment variable (milliseconds, +# defaults to 60000). +# +#allow_otlp_metrics = false + # Enable the Prometheus metrics endpoint at `/_continuwuity/metrics`. # # The endpoint is served on the same listener as the rest of the server. diff --git a/src/api/client/membership/invite.rs b/src/api/client/membership/invite.rs index ab53e2eed..fbe210224 100644 --- a/src/api/client/membership/invite.rs +++ b/src/api/client/membership/invite.rs @@ -2,6 +2,7 @@ use axum::extract::State; use conduwuit::{ Err, Result, debug_error, err, info, matrix::{event::gen_event_id_canonical_json, pdu::PartialPdu}, + metrics::KeyValue, trace, warn, }; use futures::FutureExt; @@ -113,6 +114,31 @@ pub(crate) async fn invite_helper( room_id: &RoomId, reason: Option, is_direct: bool, +) -> Result { + let result = + send_invite(services, sender_user, recipient_user, room_id, reason, is_direct).await; + + let destination = if services.globals.user_is_local(recipient_user) { + "local" + } else { + "remote" + }; + services.server.metrics.instruments.invites.add(1, &[ + KeyValue::new("direction", "outgoing"), + KeyValue::new("destination", destination), + KeyValue::new("outcome", if result.is_ok() { "success" } else { "failure" }), + ]); + + result +} + +async fn send_invite( + services: &Services, + sender_user: &UserId, + recipient_user: &UserId, + room_id: &RoomId, + reason: Option, + is_direct: bool, ) -> Result { if !services.users.is_admin(sender_user).await && services.config.block_non_admin_invites { info!( diff --git a/src/api/client/unversioned.rs b/src/api/client/unversioned.rs index d529d004d..11b482a21 100644 --- a/src/api/client/unversioned.rs +++ b/src/api/client/unversioned.rs @@ -2,6 +2,7 @@ use axum::{Json, extract::State, response::IntoResponse}; use conduwuit::{ Err, Result, matrix::versions::{unstable_features, versions}, + metrics::PROMETHEUS_CONTENT_TYPE, utils::string::constant_time_eq, }; use futures::StreamExt; @@ -80,7 +81,7 @@ pub(crate) async fn continuwuity_local_user_count( /// # `GET /_continuwuity/metrics` /// -/// Prometheus metrics in the OpenMetrics text format. Only routed when +/// Prometheus metrics in the text exposition format. Only routed when /// `allow_prometheus_metrics` is enabled. pub(crate) async fn continuwuity_metrics( State(services): State, @@ -100,14 +101,7 @@ pub(crate) async fn continuwuity_metrics( } } - services.update_metrics()?; - let body = services.server.metrics.prometheus.encode()?; + let body = services.server.metrics.render()?; - Ok(( - [( - header::CONTENT_TYPE, - "application/openmetrics-text; version=1.0.0; charset=utf-8", - )], - body, - )) + Ok(([(header::CONTENT_TYPE, PROMETHEUS_CONTENT_TYPE)], body)) } diff --git a/src/api/server/invite.rs b/src/api/server/invite.rs index f9ad2dbb6..89069e050 100644 --- a/src/api/server/invite.rs +++ b/src/api/server/invite.rs @@ -5,6 +5,7 @@ use base64::{Engine as _, engine::general_purpose}; use conduwuit::{ Err, Error, EventTypeExt, PduEvent, Result, debug, err, error, matrix::{Event, StateKey}, + metrics::KeyValue, result::FlatOk, state_res, trace, utils::hash::sha256, @@ -31,6 +32,21 @@ use crate::{Ruma, server::utils::validate_any_membership_event}; pub(crate) async fn create_invite_route( State(services): State, body: Ruma, +) -> Result { + let result = create_invite(&services, body).await; + + services.server.metrics.instruments.invites.add(1, &[ + KeyValue::new("direction", "incoming"), + KeyValue::new("destination", "local"), + KeyValue::new("outcome", if result.is_ok() { "success" } else { "failure" }), + ]); + + result +} + +async fn create_invite( + services: &crate::State, + body: Ruma, ) -> Result { if !services.server.supported_room_version(&body.room_version) { return Err(Error::BadRequest( @@ -64,7 +80,7 @@ pub(crate) async fn create_invite_route( "Validating invite room state for invite request" ); let (create_event_id, state) = validate_invite_state( - &services, + services, &body.invite_room_state, &room_version_rules, body.room_id.clone(), @@ -92,7 +108,7 @@ pub(crate) async fn create_invite_route( // And then we can validate the member event itself let (mut signed_event, sender_user, recipient_user) = validate_invite_membership_event( - &services, + services, &body.event, &room_version_rules, &body.identity, diff --git a/src/api/server/send.rs b/src/api/server/send.rs index 584962b82..415b019c3 100644 --- a/src/api/server/send.rs +++ b/src/api/server/send.rs @@ -6,6 +6,7 @@ use std::{ use axum::extract::State; use conduwuit::{ Err, Error, Result, debug, debug_error, debug_warn, err, error, + metrics::KeyValue, result::LogErr, trace, utils::{ @@ -22,7 +23,8 @@ use futures::{FutureExt, Stream, StreamExt, TryFutureExt, TryStreamExt}; use http::StatusCode; use itertools::Itertools; use ruma::{ - CanonicalJsonObject, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, ServerName, UserId, + CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, + ServerName, UserId, api::{ error::{ErrorKind, LimitExceededErrorData}, federation::transactions::{ @@ -34,7 +36,10 @@ use ruma::{ send_transaction_message, }, }, - events::receipt::{ReceiptEvent, ReceiptEventContent, ReceiptType}, + events::{ + TimelineEventType, + receipt::{ReceiptEvent, ReceiptEventContent, ReceiptType}, + }, serde::Raw, to_device::DeviceIdOrAllDevices, }; @@ -151,6 +156,12 @@ async fn process_inbound_transaction( { | Ok(pdu) => Some(pdu), | Err(error) => { + services + .server + .metrics + .instruments + .federation_incoming_pdus + .add(1, &[KeyValue::new("outcome", "unparsable")]); let queued = services .rooms .event_handler @@ -181,6 +192,17 @@ async fn process_inbound_transaction( .catch_unwind() .map_err(Error::from_panic) .await; + let outcome = if matches!(handle_result, Ok(Ok(_))) { + "success" + } else { + "failure" + }; + services + .server + .metrics + .instruments + .federation_incoming_transaction_duration + .record(txn_start_time.elapsed().as_secs_f64(), &[KeyValue::new("outcome", outcome)]); let results = match handle_result { | Ok(Ok(results)) => results, | Ok(Err(err)) => { @@ -337,6 +359,13 @@ async fn handle_room( .server .check_running() .map_err(|_| TransactionError::ShuttingDown)?; + let event_type = KeyValue::new("type", pdu_type_label(&value)); + let origin_server_ts = value + .get("origin_server_ts") + .and_then(CanonicalJsonValue::as_integer) + .and_then(|ts| u64::try_from(i64::from(ts)).ok()); + + let start = Instant::now(); let result = services .rooms .event_handler @@ -344,12 +373,44 @@ async fn handle_room( .boxed() .await .map(|_| ()); + + let outcome = KeyValue::new("outcome", match &result { + | Ok(()) => "accepted", + | Err(Error::BadRequest(ErrorKind::Forbidden, ..)) => "forbidden", + | Err(Error::BadRequest(ErrorKind::InvalidParam | ErrorKind::BadJson, ..)) => + "invalid", + | Err(_) => "error", + }); + let instruments = &services.server.metrics.instruments; + instruments + .federation_incoming_pdu_duration + .record(start.elapsed().as_secs_f64(), std::slice::from_ref(&outcome)); + if let (Ok(()), Some(ts)) = (&result, origin_server_ts) { + let lag = millis_since_unix_epoch().saturating_sub(ts); + instruments + .federation_incoming_pdu_lag + .record(Duration::from_millis(lag).as_secs_f64(), &[]); + } + instruments + .federation_incoming_pdus + .add(1, &[event_type, outcome]); + results.push((event_id, result)); } Ok(results) } async fn handle_edu(services: &Services, origin: &ServerName, edu: Edu) { + let edu_type = match &edu { + | Edu::Presence(_) => "m.presence", + | Edu::Receipt(_) => "m.receipt", + | Edu::Typing(_) => "m.typing", + | Edu::DeviceListUpdate(_) => "m.device_list_update", + | Edu::DirectToDevice(_) => "m.direct_to_device", + | Edu::SigningKeyUpdate(_) => "m.signing_key_update", + | _ => "other", + }; + let start = Instant::now(); match edu { | Edu::Presence(presence) if services.server.config.allow_incoming_presence => handle_edu_presence(services, origin, presence).await, @@ -373,6 +434,26 @@ async fn handle_edu(services: &Services, origin: &ServerName, edu: Edu) { | _ => trace!(?edu, "skipped"), } + + let edu_type = [KeyValue::new("type", edu_type)]; + let instruments = &services.server.metrics.instruments; + instruments + .federation_incoming_edu_duration + .record(start.elapsed().as_secs_f64(), &edu_type); + instruments.federation_incoming_edus.add(1, &edu_type); +} + +/// 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) + { + | Some(TimelineEventType::_Custom(_)) | None => "other".to_owned(), + | Some(event_type) => event_type.to_string(), + } } async fn handle_edu_presence( diff --git a/src/core/Cargo.toml b/src/core/Cargo.toml index b61269b8b..8e16cadf1 100644 --- a/src/core/Cargo.toml +++ b/src/core/Cargo.toml @@ -113,7 +113,8 @@ tracing-subscriber.workspace = true tracing.workspace = true url.workspace = true parking_lot.workspace = true -prometheus-client.workspace = true +opentelemetry.workspace = true +prometheus.workspace = true lock_api.workspace = true hyper-util.workspace = true resolvematrix.workspace = true @@ -128,6 +129,8 @@ hardened_malloc-rs.optional = true [dev-dependencies] maplit.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-prometheus.workspace = true [lints] workspace = true diff --git a/src/core/alloc/default.rs b/src/core/alloc/default.rs index 65354b7d9..7d6f94e34 100644 --- a/src/core/alloc/default.rs +++ b/src/core/alloc/default.rs @@ -10,3 +10,6 @@ pub fn memory_stats(_opts: &str) -> Option { None } /// Always returns None #[must_use] pub fn memory_usage() -> Option { None } + +#[must_use] +pub fn memory_usage_stats() -> Option<[(&'static str, usize); 6]> { None } diff --git a/src/core/alloc/hardened.rs b/src/core/alloc/hardened.rs index 5f8506734..69628ba7d 100644 --- a/src/core/alloc/hardened.rs +++ b/src/core/alloc/hardened.rs @@ -9,6 +9,9 @@ pub fn trim>>(_: I) -> crate::Result { Ok(()) } //TODO: get usage pub fn memory_usage() -> Option { None } +#[must_use] +pub fn memory_usage_stats() -> Option<[(&'static str, usize); 6]> { None } + #[must_use] pub fn memory_stats(_opts: &str) -> Option { Some("Extended statistics are not available from hardened_malloc.".to_owned()) diff --git a/src/core/alloc/je.rs b/src/core/alloc/je.rs index c80ab755b..584cda823 100644 --- a/src/core/alloc/je.rs +++ b/src/core/alloc/je.rs @@ -56,25 +56,15 @@ fn _static_initialization() { #[must_use] #[cfg(feature = "jemalloc_stats")] pub fn memory_usage() -> Option { - use mallctl::stats; - - let mibs = |input: Result| { - let input = input.unwrap_or_default(); + let mibs = |input: usize| { let kibs = input / 1024; let kibs = u32::try_from(kibs).unwrap_or_default(); let kibs = f64::from(kibs); kibs / 1024.0 }; - // Acquire the epoch; ensure latest stats are pulled in - acq_epoch().ok()?; - - let allocated = mibs(stats::allocated::read()); - let active = mibs(stats::active::read()); - let mapped = mibs(stats::mapped::read()); - let metadata = mibs(stats::metadata::read()); - let resident = mibs(stats::resident::read()); - let retained = mibs(stats::retained::read()); + let [allocated, active, mapped, metadata, resident, retained] = + memory_usage_stats()?.map(|(_, bytes)| mibs(bytes)); Some(format!( "allocated: {allocated:.2} MiB\nactive: {active:.2} MiB\nmapped: {mapped:.2} \ MiB\nmetadata: {metadata:.2} MiB\nresident: {resident:.2} MiB\nretained: {retained:.2} \ @@ -86,6 +76,28 @@ pub fn memory_usage() -> Option { #[cfg(not(feature = "jemalloc_stats"))] pub fn memory_usage() -> Option { None } +#[must_use] +#[cfg(feature = "jemalloc_stats")] +pub fn memory_usage_stats() -> Option<[(&'static str, usize); 6]> { + use mallctl::stats; + + // Acquire the epoch; ensure latest stats are pulled in + acq_epoch().ok()?; + + Some([ + ("allocated", stats::allocated::read().unwrap_or_default()), + ("active", stats::active::read().unwrap_or_default()), + ("mapped", stats::mapped::read().unwrap_or_default()), + ("metadata", stats::metadata::read().unwrap_or_default()), + ("resident", stats::resident::read().unwrap_or_default()), + ("retained", stats::retained::read().unwrap_or_default()), + ]) +} + +#[must_use] +#[cfg(not(feature = "jemalloc_stats"))] +pub fn memory_usage_stats() -> Option<[(&'static str, usize); 6]> { None } + pub fn memory_stats(opts: &str) -> Option { const MAX_LENGTH: usize = 1_048_576; diff --git a/src/core/alloc/mod.rs b/src/core/alloc/mod.rs index 0ed1b1a62..a34809412 100644 --- a/src/core/alloc/mod.rs +++ b/src/core/alloc/mod.rs @@ -4,7 +4,7 @@ #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] pub mod je; #[cfg(all(not(target_env = "msvc"), feature = "jemalloc"))] -pub use je::{memory_stats, memory_usage, trim}; +pub use je::{memory_stats, memory_usage, memory_usage_stats, trim}; #[cfg(all(not(target_env = "msvc"), feature = "hardened_malloc", not(feature = "jemalloc")))] pub mod hardened; @@ -13,7 +13,7 @@ pub mod hardened; feature = "hardened_malloc", not(feature = "jemalloc") ))] -pub use hardened::{memory_stats, memory_usage, trim}; +pub use hardened::{memory_stats, memory_usage, memory_usage_stats, trim}; #[cfg(any( target_env = "msvc", @@ -24,4 +24,4 @@ pub mod default; target_env = "msvc", all(not(feature = "hardened_malloc"), not(feature = "jemalloc")) ))] -pub use default::{memory_stats, memory_usage, trim}; +pub use default::{memory_stats, memory_usage, memory_usage_stats, trim}; diff --git a/src/core/config/mod.rs b/src/core/config/mod.rs index 2a239a731..0962d7a88 100644 --- a/src/core/config/mod.rs +++ b/src/core/config/mod.rs @@ -1026,6 +1026,15 @@ pub struct Config { #[serde(default = "default_otlp_protocol")] pub otlp_protocol: String, + /// Enable OpenTelemetry OTLP metrics export, using the same endpoint and + /// `otlp_protocol` as traces. + /// + /// The export interval can be changed with the + /// OTEL_METRIC_EXPORT_INTERVAL environment variable (milliseconds, + /// defaults to 60000). + #[serde(default)] + pub allow_otlp_metrics: bool, + /// Enable the Prometheus metrics endpoint at `/_continuwuity/metrics`. /// /// The endpoint is served on the same listener as the rest of the server. diff --git a/src/core/metrics/instruments.rs b/src/core/metrics/instruments.rs new file mode 100644 index 000000000..2607025a8 --- /dev/null +++ b/src/core/metrics/instruments.rs @@ -0,0 +1,296 @@ +use std::time::{Duration, SystemTime}; + +use opentelemetry::{ + KeyValue, + metrics::{Counter, Histogram, Meter, UpDownCounter}, +}; +use tokio::runtime; + +use super::process; + +pub struct Instruments { + pub http_server_request_duration: Histogram, + pub http_server_active_requests: UpDownCounter, + + pub federation_outgoing_request_duration: Histogram, + pub federation_incoming_transaction_duration: Histogram, + pub federation_incoming_pdus: Counter, + pub federation_incoming_pdu_duration: Histogram, + pub federation_incoming_pdu_lag: Histogram, + pub federation_incoming_edus: Counter, + pub federation_incoming_edu_duration: Histogram, + + pub sender_transaction_duration: Histogram, + pub sender_events: Counter, + pub sender_events_queued: Counter, + + pub media_write_size: Histogram, + pub media_read_size: Histogram, + + pub antispam_checks: Counter, + pub invites: Counter, +} + +impl Instruments { + pub(super) fn new(meter: &Meter, runtime_metrics: Option) -> Self { + register_process(meter); + register_allocator(meter); + if let Some(runtime_metrics) = runtime_metrics { + register_runtime(meter, runtime_metrics); + } + + let duration = |name: &'static str, description: &'static str| { + meter + .f64_histogram(name) + .with_description(description) + .with_unit("s") + .with_boundaries(duration_buckets()) + .build() + }; + let size = |name: &'static str, description: &'static str| { + meter + .u64_histogram(name) + .with_description(description) + .with_unit("By") + .with_boundaries(size_buckets()) + .build() + }; + + Self { + http_server_request_duration: duration( + "http.server.request.duration", + "Duration of handled HTTP requests", + ), + http_server_active_requests: meter + .i64_up_down_counter("http.server.active_requests") + .with_description("Number of HTTP requests currently being handled") + .with_unit("{request}") + .build(), + + federation_outgoing_request_duration: duration( + "continuwuity.federation.outgoing.request.duration", + "Duration of outgoing federation requests", + ), + federation_incoming_transaction_duration: duration( + "continuwuity.federation.incoming.transaction.duration", + "Time taken to process incoming federation transactions", + ), + federation_incoming_pdus: meter + .u64_counter("continuwuity.federation.incoming.pdus") + .with_description("Incoming federation PDUs by event type and outcome") + .with_unit("{pdu}") + .build(), + federation_incoming_pdu_duration: duration( + "continuwuity.federation.incoming.pdu.duration", + "Time taken to handle an incoming federation PDU", + ), + federation_incoming_pdu_lag: meter + .f64_histogram("continuwuity.federation.incoming.pdu.lag") + .with_description( + "Time between an accepted incoming PDU's origin_server_ts and its arrival", + ) + .with_unit("s") + .with_boundaries(vec![ + 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 300.0, 900.0, 3600.0, + ]) + .build(), + federation_incoming_edus: meter + .u64_counter("continuwuity.federation.incoming.edus") + .with_description("Incoming federation EDUs by type") + .with_unit("{edu}") + .build(), + federation_incoming_edu_duration: duration( + "continuwuity.federation.incoming.edu.duration", + "Time taken to handle an incoming federation EDU", + ), + + sender_transaction_duration: duration( + "continuwuity.sender.transaction.duration", + "Duration of outgoing transactions by destination type and outcome", + ), + sender_events: meter + .u64_counter("continuwuity.sender.events") + .with_description("PDUs and EDUs sent in outgoing transactions by outcome") + .with_unit("{event}") + .build(), + sender_events_queued: meter + .u64_counter("continuwuity.sender.events.queued") + .with_description("PDUs and EDUs queued for sending") + .with_unit("{event}") + .build(), + + media_write_size: size( + "continuwuity.media.write.size", + "Size of media files written, from local uploads or cached remote media", + ), + media_read_size: size("continuwuity.media.read.size", "Size of media files served"), + + antispam_checks: meter + .u64_counter("continuwuity.antispam.checks") + .with_description("Antispam service checks by check and outcome") + .with_unit("{check}") + .build(), + invites: meter + .u64_counter("continuwuity.invites") + .with_description("Invites sent by local users or received over federation") + .with_unit("{invite}") + .build(), + } + } +} + +/// Collapse non-standard methods so arbitrary client input cannot create new +/// attribute 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", + | http::Method::CONNECT => "CONNECT", + | http::Method::TRACE => "TRACE", + | _ if method.as_str() == "QUERY" => "QUERY", + | _ => "_OTHER", + } +} + +fn duration_buckets() -> Vec { (0..16).map(|i| 0.001 * f64::from(1_u32 << i)).collect() } + +fn size_buckets() -> Vec { (0..11).map(|i| 1024.0 * f64::from(4_u32.pow(i))).collect() } + +fn register_process(meter: &Meter) { + let started = SystemTime::now(); + meter + .f64_observable_gauge("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(), &[]); + }) + .build(); + + meter + .f64_observable_counter("process.cpu.time") + .with_description("Total CPU time spent by the process") + .with_unit("s") + .with_callback(|observer| { + if let Some((user, system)) = process::cpu_time() { + observer.observe(user.as_secs_f64(), &[KeyValue::new("cpu.mode", "user")]); + observer.observe(system.as_secs_f64(), &[KeyValue::new("cpu.mode", "system")]); + } + }) + .build(); + + meter + .i64_observable_up_down_counter("process.memory.usage") + .with_description("Resident memory of the process") + .with_unit("By") + .with_callback(|observer| { + if let Some(resident) = + process::memory().and_then(|(resident, _)| resident.try_into().ok()) + { + observer.observe(resident, &[]); + } + }) + .build(); + + meter + .i64_observable_up_down_counter("process.memory.virtual") + .with_description("Virtual memory of the process") + .with_unit("By") + .with_callback(|observer| { + if let Some(virt) = process::memory().and_then(|(_, virt)| virt.try_into().ok()) { + observer.observe(virt, &[]); + } + }) + .build(); + + meter + .i64_observable_up_down_counter("process.unix.file_descriptor.count") + .with_description("Number of file descriptors in use by the process") + .with_unit("{file_descriptor}") + .with_callback(|observer| { + if let Some(fds) = process::open_fds().and_then(|fds| fds.try_into().ok()) { + observer.observe(fds, &[]); + } + }) + .build(); +} + +fn register_allocator(meter: &Meter) { + meter + .u64_observable_gauge("continuwuity.jemalloc.memory") + .with_description("jemalloc memory statistics, when built with jemalloc_stats") + .with_unit("By") + .with_callback(|observer| { + for (kind, bytes) in crate::alloc::memory_usage_stats().into_iter().flatten() { + if let Ok(bytes) = bytes.try_into() { + observer.observe(bytes, &[KeyValue::new("kind", kind)]); + } + } + }) + .build(); +} + +fn register_runtime(meter: &Meter, metrics: runtime::RuntimeMetrics) { + let gauge = |name: &'static str, + description: &'static str, + read: fn(&runtime::RuntimeMetrics) -> usize| { + let metrics = metrics.clone(); + meter + .u64_observable_gauge(name) + .with_description(description) + .with_callback(move |observer| { + if let Ok(value) = read(&metrics).try_into() { + observer.observe(value, &[]); + } + }) + .build(); + }; + + gauge( + "continuwuity.tokio.workers", + "Number of tokio runtime worker threads", + runtime::RuntimeMetrics::num_workers, + ); + gauge( + "continuwuity.tokio.alive_tasks", + "Number of alive tasks in the tokio runtime", + runtime::RuntimeMetrics::num_alive_tasks, + ); + gauge( + "continuwuity.tokio.global_queue_depth", + "Number of tasks in the tokio runtime global queue", + runtime::RuntimeMetrics::global_queue_depth, + ); + + let busy_metrics = metrics.clone(); + meter + .f64_observable_counter("continuwuity.tokio.worker.busy_time") + .with_description("Total time tokio worker threads spent busy") + .with_unit("s") + .with_callback(move |observer| { + let busy: Duration = (0..busy_metrics.num_workers()) + .map(|worker| busy_metrics.worker_total_busy_duration(worker)) + .sum(); + observer.observe(busy.as_secs_f64(), &[]); + }) + .build(); + + meter + .u64_observable_counter("continuwuity.tokio.worker.parks") + .with_description("Total number of times tokio worker threads parked") + .with_unit("{park}") + .with_callback(move |observer| { + let parks = (0..metrics.num_workers()) + .map(|worker| metrics.worker_park_count(worker)) + .sum(); + observer.observe(parks, &[]); + }) + .build(); +} diff --git a/src/core/metrics/mod.rs b/src/core/metrics/mod.rs index 91cb93b86..b1b032bc4 100644 --- a/src/core/metrics/mod.rs +++ b/src/core/metrics/mod.rs @@ -1,16 +1,20 @@ -mod prometheus; +mod instruments; +mod process; +mod tests; use std::sync::atomic::AtomicU32; +use opentelemetry::metrics::Meter; +pub use opentelemetry::{KeyValue, metrics::AsyncInstrument}; +pub use prometheus::TEXT_FORMAT as PROMETHEUS_CONTENT_TYPE; +use prometheus::{Encoder, TextEncoder}; use tokio::runtime; 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 use self::instruments::{Instruments, method_label}; +use crate::{Err, Result}; pub struct Metrics { _runtime: Option, @@ -30,12 +34,18 @@ pub struct Metrics { pub requests_handle_finished: AtomicU32, pub requests_panic: AtomicU32, - pub prometheus: Prometheus, + pub meter: Meter, + pub instruments: Instruments, + registry: Option, } impl Metrics { #[must_use] - pub fn new(runtime: Option) -> Self { + pub fn new( + runtime: Option, + meter: Meter, + registry: Option, + ) -> Self { #[cfg(tokio_unstable)] let runtime_monitor = runtime.as_ref().map(RuntimeMonitor::new); @@ -47,7 +57,9 @@ impl Metrics { Self { _runtime: runtime.clone(), - prometheus: Prometheus::new(runtime_metrics.clone()), + instruments: Instruments::new(&meter, runtime_metrics.clone()), + meter, + registry, runtime_metrics, @@ -75,6 +87,19 @@ impl Metrics { .expect("next interval") } + pub fn render(&self) -> Result { + let Some(registry) = &self.registry else { + return Err!("Prometheus metrics are not enabled"); + }; + + let mut buf = Vec::new(); + TextEncoder::new() + .encode(®istry.gather(), &mut buf) + .map_err(|e| crate::err!("{e}"))?; + + Ok(String::from_utf8(buf)?) + } + #[inline] pub fn task_root(&self) -> Option<&TaskMonitor> { self.task_monitor.as_ref() } diff --git a/src/core/metrics/process.rs b/src/core/metrics/process.rs new file mode 100644 index 000000000..36192a402 --- /dev/null +++ b/src/core/metrics/process.rs @@ -0,0 +1,43 @@ +use std::time::Duration; + +/// User and system CPU time. +#[cfg(unix)] +pub(super) fn cpu_time() -> Option<(Duration, Duration)> { + use nix::sys::resource::{UsageWho, getrusage}; + + let usage = getrusage(UsageWho::RUSAGE_SELF).ok()?; + let to_duration = |tv: nix::sys::time::TimeVal| { + Duration::from_secs(tv.tv_sec().try_into().ok()?) + .checked_add(Duration::from_micros(tv.tv_usec().try_into().ok()?)) + }; + + Some((to_duration(usage.user_time())?, to_duration(usage.system_time())?)) +} + +#[cfg(not(unix))] +pub(super) fn cpu_time() -> Option<(Duration, Duration)> { None } + +/// Resident and virtual memory in bytes, read from `/proc/self/status`. +#[cfg(target_os = "linux")] +pub(super) fn memory() -> Option<(u64, u64)> { + let status = std::fs::read_to_string("/proc/self/status").ok()?; + let field = |name: &str| -> Option { + let value = status.lines().find_map(|line| line.strip_prefix(name))?; + let kib: u64 = value.trim().trim_end_matches(" kB").parse().ok()?; + kib.checked_mul(1024) + }; + + Some((field("VmRSS:")?, field("VmSize:")?)) +} + +#[cfg(not(target_os = "linux"))] +pub(super) fn memory() -> Option<(u64, u64)> { None } + +#[cfg(target_os = "linux")] +pub(super) fn open_fds() -> Option { + let count = std::fs::read_dir("/proc/self/fd").ok()?.count(); + count.try_into().ok() +} + +#[cfg(not(target_os = "linux"))] +pub(super) fn open_fds() -> Option { None } diff --git a/src/core/metrics/prometheus.rs b/src/core/metrics/prometheus.rs deleted file mode 100644 index d49ac645e..000000000 --- a/src/core/metrics/prometheus.rs +++ /dev/null @@ -1,261 +0,0 @@ -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.001, 2.0, 16)) } - -#[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/metrics/tests.rs b/src/core/metrics/tests.rs new file mode 100644 index 000000000..059c6c52f --- /dev/null +++ b/src/core/metrics/tests.rs @@ -0,0 +1,54 @@ +#![cfg(test)] + +use opentelemetry::metrics::MeterProvider; +use opentelemetry_sdk::metrics::SdkMeterProvider; + +use super::{KeyValue, Metrics, method_label}; + +fn metrics() -> (Metrics, SdkMeterProvider) { + let registry = prometheus::Registry::new(); + let exporter = opentelemetry_prometheus::exporter() + .with_registry(registry.clone()) + .scope_info_enabled(false) + .build() + .unwrap(); + let provider = SdkMeterProvider::builder().with_reader(exporter).build(); + let meter = provider.meter("continuwuity"); + let handle = tokio::runtime::Handle::current(); + + (Metrics::new(Some(handle), meter, Some(registry)), provider) +} + +#[tokio::test] +async fn renders_prometheus_names() { + let (metrics, _provider) = metrics(); + let instruments = &metrics.instruments; + instruments.http_server_request_duration.record(0.01, &[ + KeyValue::new("http.request.method", method_label(&http::Method::GET)), + KeyValue::new("http.route", "/_matrix/client/versions"), + KeyValue::new("http.response.status_code", 200_i64), + ]); + instruments + .sender_events + .add(3, &[KeyValue::new("destination", "federation"), KeyValue::new("kind", "pdu")]); + + let body = metrics.render().unwrap(); + for expected in [ + "http_server_request_duration_seconds_count{http_request_method=\"GET\",\ + http_response_status_code=\"200\",http_route=\"/_matrix/client/versions\"} 1", + "continuwuity_sender_events_total{destination=\"federation\",kind=\"pdu\"} 3", + "# TYPE process_uptime_seconds gauge", + "process_cpu_time_seconds_total{cpu_mode=\"user\"}", + "# TYPE continuwuity_tokio_worker_busy_time_seconds_total counter", + ] { + assert!(body.contains(expected), "missing {expected:?} in:\n{body}"); + } + assert!(!body.contains("otel_scope_name")); +} + +#[test] +fn method_label_is_bounded() { + let method = |name: &[u8]| http::Method::from_bytes(name).unwrap(); + assert_eq!(method_label(&method(b"QUERY")), "QUERY"); + assert_eq!(method_label(&method(b"BREW")), "_OTHER"); +} diff --git a/src/core/server.rs b/src/core/server.rs index 4b673f321..c84936478 100644 --- a/src/core/server.rs +++ b/src/core/server.rs @@ -49,7 +49,12 @@ pub struct Server { impl Server { #[must_use] - pub fn new(config: Config, runtime: Option, log: Log) -> Self { + pub fn new( + config: Config, + runtime: Option, + log: Log, + metrics: Metrics, + ) -> Self { Self { name: config.server_name.clone(), config: config::Manager::new(config), @@ -57,10 +62,10 @@ impl Server { stopping: AtomicBool::new(false), reloading: AtomicBool::new(false), restarting: AtomicBool::new(false), - runtime: runtime.clone(), + runtime, signal: broadcast::channel::<&'static str>(1).0, log, - metrics: Metrics::new(runtime), + metrics, } } diff --git a/src/main/Cargo.toml b/src/main/Cargo.toml index 611b0d5ef..6b4e0877c 100644 --- a/src/main/Cargo.toml +++ b/src/main/Cargo.toml @@ -131,9 +131,7 @@ media_thumbnail = [ "conduwuit-service/media_thumbnail", ] otlp_telemetry = [ - "dep:opentelemetry", "dep:tracing-opentelemetry", - "dep:opentelemetry_sdk", "dep:opentelemetry-otlp", ] perf_measurements = [ @@ -221,12 +219,12 @@ console-subscriber.optional = true console-subscriber.workspace = true const-str.workspace = true log.workspace = true -opentelemetry.optional = true opentelemetry.workspace = true opentelemetry-otlp.optional = true opentelemetry-otlp.workspace = true -opentelemetry_sdk.optional = true +opentelemetry-prometheus.workspace = true opentelemetry_sdk.workspace = true +prometheus.workspace = true sentry-tower.optional = true sentry-tower.workspace = true sentry-tracing.optional = true diff --git a/src/main/metrics.rs b/src/main/metrics.rs new file mode 100644 index 000000000..9e45955de --- /dev/null +++ b/src/main/metrics.rs @@ -0,0 +1,76 @@ +use conduwuit_core::{Result, config::Config, err, metrics::Metrics}; +use opentelemetry::{KeyValue, metrics::MeterProvider}; +use opentelemetry_sdk::{ + Resource, + metrics::SdkMeterProvider, + resource::{EnvResourceDetector, TelemetryResourceDetector}, +}; +use tokio::runtime; + +/// The returned provider must outlive the server so pending OTLP exports are +/// flushed on drop. +pub(crate) fn init( + config: &Config, + runtime: Option<&runtime::Handle>, +) -> Result<(Metrics, SdkMeterProvider)> { + // Defaults first so OTEL_RESOURCE_ATTRIBUTES and OTEL_SERVICE_NAME win. + let mut resource = Resource::builder_empty() + .with_detector(Box::new(TelemetryResourceDetector)) + .with_service_name(conduwuit_core::BRANDING) + .with_attributes([ + KeyValue::new("service.version", conduwuit_core::version()), + KeyValue::new("service.instance.id", config.server_name.to_string()), + ]) + .with_detector(Box::new(EnvResourceDetector::new())); + if let Some(name) = std::env::var("OTEL_SERVICE_NAME") + .ok() + .filter(|name| !name.is_empty()) + { + resource = resource.with_service_name(name); + } + let resource = resource.build(); + + let mut builder = SdkMeterProvider::builder().with_resource(resource); + + let registry = config + .allow_prometheus_metrics + .then(prometheus::Registry::new); + if let Some(registry) = ®istry { + let exporter = opentelemetry_prometheus::exporter() + .with_registry(registry.clone()) + .scope_info_enabled(false) + .build() + .map_err(|e| err!(Config("allow_prometheus_metrics", "{e}")))?; + + builder = builder.with_reader(exporter); + } + + #[cfg(feature = "otlp_telemetry")] + if config.allow_otlp_metrics { + use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; + + let exporter = match config.otlp_protocol.as_str() { + | "grpc" => MetricExporter::builder() + .with_tonic() + .with_protocol(Protocol::Grpc) + .build(), + | _ => MetricExporter::builder().with_http().build(), + } + .map_err(|e| err!(Config("allow_otlp_metrics", "{e}")))?; + + builder = builder.with_periodic_exporter(exporter); + } + + #[cfg(not(feature = "otlp_telemetry"))] + if config.allow_otlp_metrics { + conduwuit_core::warn!( + "allow_otlp_metrics is set but this build does not include the otlp_telemetry \ + feature" + ); + } + + let provider = builder.build(); + let meter = provider.meter(conduwuit_core::BRANDING); + + Ok((Metrics::new(runtime.cloned(), meter, registry), provider)) +} diff --git a/src/main/mod.rs b/src/main/mod.rs index f141b26c1..2582f15ab 100644 --- a/src/main/mod.rs +++ b/src/main/mod.rs @@ -9,6 +9,7 @@ conduwuit_macros::introspect_crate! {} mod clap; mod deadlock; mod logging; +mod metrics; mod mods; mod panic; mod restart; diff --git a/src/main/server.rs b/src/main/server.rs index 4c1abc25a..afde40910 100644 --- a/src/main/server.rs +++ b/src/main/server.rs @@ -23,6 +23,8 @@ pub(crate) struct Server { _tracing_flame_guard: TracingFlameGuard, + _meter_provider: opentelemetry_sdk::metrics::SdkMeterProvider, + #[cfg(feature = "sentry_telemetry")] _sentry_guard: Option<::sentry::ClientInitGuard>, @@ -56,6 +58,8 @@ impl Server { sys::maximize_fd_limit() .expect("Unable to increase maximum soft and hard file descriptor limit"); + let (metrics, meter_provider) = crate::metrics::init(&config, runtime)?; + let (_old_width, _new_width) = stream::set_width(config.stream_width_default); let (_old_amp, _new_amp) = stream::set_amplification(config.stream_amplification); @@ -68,15 +72,19 @@ impl Server { ); Ok(Arc::new(Self { - server: Arc::new(conduwuit_core::Server::new(config, runtime.cloned(), Log { - reload: tracing_reload_handle, - capture, - })), + server: Arc::new(conduwuit_core::Server::new( + config, + runtime.cloned(), + Log { reload: tracing_reload_handle, capture }, + metrics, + )), services: None.into(), _tracing_flame_guard: tracing_flame_guard, + _meter_provider: meter_provider, + #[cfg(feature = "sentry_telemetry")] _sentry_guard: sentry_guard, diff --git a/src/router/request.rs b/src/router/request.rs index 7c99b0668..ddbc1d542 100644 --- a/src/router/request.rs +++ b/src/router/request.rs @@ -10,7 +10,7 @@ use axum::{ }; use conduwuit::{ Result, debug, debug_error, debug_warn, err, error, - metrics::{HttpLabels, method_label}, + metrics::{KeyValue, method_label}, trace, }; use conduwuit_service::Services; @@ -37,16 +37,22 @@ 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 instruments = &services.server.metrics.instruments; + let method_attr = KeyValue::new("http.request.method", method_label(req.method())); + instruments + .http_server_active_requests + .add(1, std::slice::from_ref(&method_attr)); + conduwuit::defer! {{ + instruments + .http_server_active_requests + .add(-1, std::slice::from_ref(&method_attr)); + }}; let start = Instant::now(); let route = req .extensions() .get::() - .map_or("unmatched", truncated_matched_path) - .to_owned(); + .map(|path| KeyValue::new("http.route", truncated_matched_path(path).to_owned())); let uri = req.uri().clone(); let method = req.method().clone(); let services_ = services.clone(); @@ -73,14 +79,17 @@ pub(crate) async fn handle( 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()); + let mut attributes = vec![ + method_attr.clone(), + KeyValue::new("http.response.status_code", i64::from(status.as_u16())), + ]; + attributes.extend(route); + if status.is_server_error() { + attributes.push(KeyValue::new("error.type", status.as_str().to_owned())); + } + instruments + .http_server_request_duration + .record(start.elapsed().as_secs_f64(), &attributes); result } diff --git a/src/service/antispam/mod.rs b/src/service/antispam/mod.rs index 59b362fa1..3e235e264 100644 --- a/src/service/antispam/mod.rs +++ b/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, config: Dep, client: Dep, } @@ -24,6 +25,7 @@ impl crate::Service for Service { fn build(args: crate::Args<'_>) -> Result> { Ok(Arc::new(Self { services: Services { + server: args.server.clone(), client: args.depend::("client"), config: args.depend::("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 { diff --git a/src/service/federation/execute.rs b/src/service/federation/execute.rs index 59598c71a..78f3ee723 100644 --- a/src/service/federation/execute.rs +++ b/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::(dest, actual, &method, &url, response) diff --git a/src/service/media/mod.rs b/src/service/media/mod.rs index f7ab0ef78..7e229d9cf 100644 --- a/src/service/media/mod.rs +++ b/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, diff --git a/src/service/metrics.rs b/src/service/metrics.rs new file mode 100644 index 000000000..dec506bcf --- /dev/null +++ b/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.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) -> Arc { + let counts = Arc::new(Counts::default()); + let services = Arc::downgrade(self); + let task_counts = counts.clone(); + self.server.runtime().spawn(async move { + let mut interval = tokio::time::interval(COUNT_INTERVAL); + loop { + interval.tick().await; + let Some(services) = services.upgrade() else { + break; + }; + if !services.server.running() { + break; + } + + let local = services.users.count().await; + let remote = services.users.stream_remote_users().count().await; + let media = match services.db.get("mediaid_file") { + | Ok(map) => map.count().await, + | Err(_) => 0, + }; + + for (count, value) in [ + (&task_counts.local_users, local), + (&task_counts.remote_users, remote), + (&task_counts.media_files, media), + ] { + count.store(value.try_into().unwrap_or(u64::MAX), Ordering::Relaxed); + } + } + }); + + counts + } + + fn gauge( + self: &Arc, + name: &'static str, + description: &'static str, + unit: &'static str, + callback: F, + ) where + F: Fn(&Self, &dyn AsyncInstrument) + 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, value: usize, attributes: &[KeyValue]) { + if let Ok(value) = value.try_into() { + observer.observe(value, attributes); + } +} diff --git a/src/service/mod.rs b/src/service/mod.rs index 759e1fd8f..1a5505583 100644 --- a/src/service/mod.rs +++ b/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; diff --git a/src/service/sending/mod.rs b/src/service/sending/mod.rs index 6b913d5b9..8cb1010cf 100644 --- a/src/service/sending/mod.rs +++ b/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}")) diff --git a/src/service/sending/sender.rs b/src/service/sending/sender.rs index 9e4b3da0b..690b90767 100644 --- a/src/service/sending/sender.rs +++ b/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( diff --git a/src/service/services.rs b/src/service/services.rs index 9303b92ee..b2cd60b1b 100644 --- a/src/service/services.rs +++ b/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) -> Result> { @@ -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() {