From 9d454fd8835a6e5ebd1e9218f6554647735082a9 Mon Sep 17 00:00:00 2001 From: Erwan Leboucher Date: Fri, 25 Sep 2026 17:22:38 +0200 Subject: [PATCH] refactor(metrics): Use Tower HTTP instrumentation --- Cargo.lock | 38 ++++++++++++ Cargo.toml | 9 +++ docs/guides/metrics.mdx | 6 +- docs/public/guides/grafana-dashboard.json | 63 ++++++++++++++++++- src/core/Cargo.toml | 1 + src/core/config/check.rs | 8 +++ src/core/metrics/instruments.rs | 54 +++++----------- src/core/metrics/mod.rs | 2 +- src/core/metrics/tests.rs | 20 +----- src/main/Cargo.toml | 1 + src/main/logging.rs | 16 +---- src/main/metrics.rs | 76 ++++++++++++++++------- src/main/mod.rs | 1 + src/main/server.rs | 12 +++- src/router/Cargo.toml | 1 + src/router/layers.rs | 56 +++++++++++++++-- src/router/request.rs | 50 ++------------- 17 files changed, 264 insertions(+), 150 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index e0c2f3619..a718f824f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -324,6 +324,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ "axum-core", + "axum-macros", "bytes", "form_urlencoded", "futures-util", @@ -392,6 +393,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-macros" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7aa268c23bfbbd2c4363b9cd302a4f504fb2a9dfe7e3451d66f35dd392e20aca" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "axum-server" version = "0.8.0" @@ -917,6 +929,7 @@ dependencies = [ "opentelemetry", "opentelemetry-otlp", "opentelemetry-prometheus", + "opentelemetry-semantic-conventions", "opentelemetry_sdk", "parking_lot", "prometheus", @@ -1040,6 +1053,7 @@ dependencies = [ "openidconnect", "opentelemetry", "opentelemetry-prometheus", + "opentelemetry-semantic-conventions", "opentelemetry_sdk", "parking_lot", "prometheus", @@ -1123,6 +1137,7 @@ dependencies = [ "hyper", "hyper-util", "log", + "opentelemetry-instrumentation-tower", "ruma", "rustls", "sd-notify", @@ -4059,6 +4074,23 @@ dependencies = [ "reqwest 0.13.5", ] +[[package]] +name = "opentelemetry-instrumentation-tower" +version = "0.19.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e7006f3b1853d315462a7237c81eec3d7adfa5b70975a02b19e92d045bc22d7" +dependencies = [ + "axum", + "http", + "http-body", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-semantic-conventions", + "pin-project-lite", + "tower-layer", + "tower-service", +] + [[package]] name = "opentelemetry-otlp" version = "0.33.0" @@ -4105,6 +4137,12 @@ dependencies = [ "tonic-prost", ] +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d288b1e1bc3590052fbb8f0f635ef248dd9022cf1fbaeb6582b7ad6081a17b87" + [[package]] name = "opentelemetry_sdk" version = "0.33.0" diff --git a/Cargo.toml b/Cargo.toml index a3b63cd7d..fc1b2ef88 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -418,6 +418,15 @@ version = "4.0.1" [workspace.dependencies.opentelemetry] version = "0.33.0" +[workspace.dependencies.opentelemetry-instrumentation-tower] +version = "0.19.0" +default-features = false +features = ["axum"] + +[workspace.dependencies.opentelemetry-semantic-conventions] +version = "0.33.0" +features = ["semconv_experimental"] + [workspace.dependencies.tracing-flame] version = "0.2.0" diff --git a/docs/guides/metrics.mdx b/docs/guides/metrics.mdx index c55bbfed5..2216a2dee 100644 --- a/docs/guides/metrics.mdx +++ b/docs/guides/metrics.mdx @@ -54,8 +54,10 @@ Names below are as they appear in Prometheus. OTLP uses the dotted OpenTelemetry | Metric | Type | Labels | Description | | --- | --- | --- | --- | | `target_info` | gauge | `service_version`, `service_instance_id` | Running server version and name | -| `http_server_request_duration_seconds` | histogram | `http_request_method`, `http_route`, `http_response_status_code`, `error_type` | Handled HTTP requests, by route template | -| `http_server_active_requests` | gauge | `http_request_method` | Requests currently being handled | +| `http_server_request_duration_seconds` | histogram | `http_request_method`, `http_route`, `http_response_status_code`, `network_protocol_name`, `network_protocol_version`, `url_scheme` | Handled HTTP requests, by route template | +| `http_server_active_requests` | gauge | `http_request_method`, `url_scheme` | Requests currently being handled | +| `http_server_request_body_size_bytes` | histogram | HTTP request duration labels | Request body sizes, when `Content-Length` is supplied | +| `http_server_response_body_size_bytes` | histogram | HTTP request duration labels | Response body sizes, when known | | `continuwuity_federation_incoming_transaction_duration_seconds` | histogram | `outcome` | Processing time of incoming transactions | | `continuwuity_federation_incoming_transactions_active` | gauge | | Incoming transactions currently being processed | | `continuwuity_federation_incoming_pdus_total` | counter | `type`, `outcome` | Incoming PDUs by event type and outcome: `accepted`, `forbidden`, `invalid`, `error` or `unparsable` | diff --git a/docs/public/guides/grafana-dashboard.json b/docs/public/guides/grafana-dashboard.json index 6e26957b1..e7922d37b 100644 --- a/docs/public/guides/grafana-dashboard.json +++ b/docs/public/guides/grafana-dashboard.json @@ -756,7 +756,7 @@ }, "gridPos": { "h": 8, - "w": 12, + "w": 6, "x": 12, "y": 24 }, @@ -806,6 +806,67 @@ } } }, + { + "type": "timeseries", + "title": "Request body traffic", + "description": "Request bodies with a Content-Length", + "id": 54, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "gridPos": { + "h": 8, + "w": 6, + "x": 18, + "y": 24 + }, + "targets": [ + { + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "expr": "sum(rate(http_server_request_body_size_bytes_sum{instance=~\"$instance\"}[$__rate_interval]))", + "legendFormat": "request", + "refId": "A", + "range": true + } + ], + "fieldConfig": { + "defaults": { + "unit": "Bps", + "min": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 10, + "showPoints": "never", + "spanNulls": false, + "stacking": { + "mode": "none", + "group": "A" + } + } + }, + "overrides": [] + }, + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + } + }, { "type": "row", "title": "Incoming federation", diff --git a/src/core/Cargo.toml b/src/core/Cargo.toml index 8e16cadf1..109535949 100644 --- a/src/core/Cargo.toml +++ b/src/core/Cargo.toml @@ -114,6 +114,7 @@ tracing.workspace = true url.workspace = true parking_lot.workspace = true opentelemetry.workspace = true +opentelemetry-semantic-conventions.workspace = true prometheus.workspace = true lock_api.workspace = true hyper-util.workspace = true diff --git a/src/core/config/check.rs b/src/core/config/check.rs index 1e3efdb5e..8aad111a2 100644 --- a/src/core/config/check.rs +++ b/src/core/config/check.rs @@ -194,6 +194,14 @@ pub fn check(config: &Config) -> Result { ); } + if !matches!(config.otlp_protocol.as_str(), "http" | "grpc") { + warn!( + "Invalid OTLP protocol '{}', falling back to HTTP. Valid options are 'http' or \ + 'grpc'.", + config.otlp_protocol + ); + } + if config.allow_prometheus_metrics && config.prometheus_metrics_token.is_none() { warn!( "The Prometheus metrics endpoint is enabled without `prometheus_metrics_token`. \ diff --git a/src/core/metrics/instruments.rs b/src/core/metrics/instruments.rs index 2607025a8..e59978ccd 100644 --- a/src/core/metrics/instruments.rs +++ b/src/core/metrics/instruments.rs @@ -2,16 +2,14 @@ use std::time::{Duration, SystemTime}; use opentelemetry::{ KeyValue, - metrics::{Counter, Histogram, Meter, UpDownCounter}, + metrics::{Counter, Histogram, Meter}, }; +use opentelemetry_semantic_conventions::{attribute, metric}; 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, @@ -57,16 +55,6 @@ impl Instruments { }; 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", @@ -139,33 +127,15 @@ impl Instruments { } } -/// 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() } +pub 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") + .f64_observable_gauge(metric::PROCESS_UPTIME) .with_description("Time since the server started") .with_unit("s") .with_callback(move |observer| { @@ -175,19 +145,23 @@ fn register_process(meter: &Meter) { .build(); meter - .f64_observable_counter("process.cpu.time") + .f64_observable_counter(metric::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")]); + observer + .observe(user.as_secs_f64(), &[KeyValue::new(attribute::CPU_MODE, "user")]); + observer.observe(system.as_secs_f64(), &[KeyValue::new( + attribute::CPU_MODE, + "system", + )]); } }) .build(); meter - .i64_observable_up_down_counter("process.memory.usage") + .i64_observable_up_down_counter(metric::PROCESS_MEMORY_USAGE) .with_description("Resident memory of the process") .with_unit("By") .with_callback(|observer| { @@ -200,7 +174,7 @@ fn register_process(meter: &Meter) { .build(); meter - .i64_observable_up_down_counter("process.memory.virtual") + .i64_observable_up_down_counter(metric::PROCESS_MEMORY_VIRTUAL) .with_description("Virtual memory of the process") .with_unit("By") .with_callback(|observer| { @@ -211,7 +185,7 @@ fn register_process(meter: &Meter) { .build(); meter - .i64_observable_up_down_counter("process.unix.file_descriptor.count") + .i64_observable_up_down_counter(metric::PROCESS_UNIX_FILE_DESCRIPTOR_COUNT) .with_description("Number of file descriptors in use by the process") .with_unit("{file_descriptor}") .with_callback(|observer| { diff --git a/src/core/metrics/mod.rs b/src/core/metrics/mod.rs index b1b032bc4..d3f1a1da2 100644 --- a/src/core/metrics/mod.rs +++ b/src/core/metrics/mod.rs @@ -13,7 +13,7 @@ use tokio_metrics::TaskMonitor; #[cfg(tokio_unstable)] use tokio_metrics::{RuntimeIntervals, RuntimeMonitor}; -pub use self::instruments::{Instruments, method_label}; +pub use self::instruments::{Instruments, duration_buckets}; use crate::{Err, Result}; pub struct Metrics { diff --git a/src/core/metrics/tests.rs b/src/core/metrics/tests.rs index 059c6c52f..fdbe52c9d 100644 --- a/src/core/metrics/tests.rs +++ b/src/core/metrics/tests.rs @@ -3,7 +3,7 @@ use opentelemetry::metrics::MeterProvider; use opentelemetry_sdk::metrics::SdkMeterProvider; -use super::{KeyValue, Metrics, method_label}; +use super::{KeyValue, Metrics}; fn metrics() -> (Metrics, SdkMeterProvider) { let registry = prometheus::Registry::new(); @@ -22,20 +22,13 @@ fn metrics() -> (Metrics, SdkMeterProvider) { #[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 + metrics + .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\"}", @@ -45,10 +38,3 @@ async fn renders_prometheus_names() { } 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/main/Cargo.toml b/src/main/Cargo.toml index 6b4e0877c..3fbcbe3b4 100644 --- a/src/main/Cargo.toml +++ b/src/main/Cargo.toml @@ -223,6 +223,7 @@ opentelemetry.workspace = true opentelemetry-otlp.optional = true opentelemetry-otlp.workspace = true opentelemetry-prometheus.workspace = true +opentelemetry-semantic-conventions.workspace = true opentelemetry_sdk.workspace = true prometheus.workspace = true sentry-tower.optional = true diff --git a/src/main/logging.rs b/src/main/logging.rs index 58c8ccfc6..1cffed4f1 100644 --- a/src/main/logging.rs +++ b/src/main/logging.rs @@ -6,7 +6,6 @@ use conduwuit_core::{ debug_warn, err, log::{ConsoleFormat, ConsoleWriter, LogLevelReloadHandles, capture, fmt_span}, result::UnwrapOrErr, - warn, }; #[cfg(feature = "otlp_telemetry")] use opentelemetry::trace::TracerProvider; @@ -89,26 +88,17 @@ pub(crate) fn init( .with_protocol(opentelemetry_otlp::Protocol::Grpc) // TODO: build from env when 0.32 is released .build() .expect("Failed to create OTLP gRPC exporter"), - | "http" => opentelemetry_otlp::SpanExporter::builder() + | _ => opentelemetry_otlp::SpanExporter::builder() .with_http() .build() .expect("Failed to create OTLP HTTP exporter"), - | protocol => { - warn!( - "Invalid OTLP protocol '{}', falling back to HTTP. Valid options are \ - 'http' or 'grpc'.", - protocol - ); - opentelemetry_otlp::SpanExporter::builder() - .with_http() - .build() - .expect("Failed to create OTLP HTTP exporter") - }, }; let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_resource(crate::metrics::resource(config)) .with_batch_exporter(exporter) .build(); + opentelemetry::global::set_tracer_provider(provider.clone()); let tracer = provider.tracer(conduwuit_core::BRANDING); diff --git a/src/main/metrics.rs b/src/main/metrics.rs index 9e45955de..deffd310b 100644 --- a/src/main/metrics.rs +++ b/src/main/metrics.rs @@ -1,36 +1,41 @@ -use conduwuit_core::{Result, config::Config, err, metrics::Metrics}; -use opentelemetry::{KeyValue, metrics::MeterProvider}; +use conduwuit_core::{ + Result, + config::Config, + err, + metrics::{Metrics, duration_buckets}, +}; +use opentelemetry::{InstrumentationScope, KeyValue, global, metrics::MeterProvider}; use opentelemetry_sdk::{ Resource, - metrics::SdkMeterProvider, + metrics::{Aggregation, Instrument, SdkMeterProvider, Stream}, resource::{EnvResourceDetector, TelemetryResourceDetector}, }; +use opentelemetry_semantic_conventions::{SCHEMA_URL, attribute, metric}; use tokio::runtime; -/// The returned provider must outlive the server so pending OTLP exports are -/// flushed on drop. +/// The returned provider must be shut down before the runtime stops so pending +/// OTLP exports are flushed. 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 resource = resource(config); - let mut builder = SdkMeterProvider::builder().with_resource(resource); + let mut builder = SdkMeterProvider::builder() + .with_resource(resource) + .with_view(|instrument: &Instrument| { + (instrument.name() == metric::HTTP_SERVER_REQUEST_DURATION) + .then(|| { + Stream::builder() + .with_aggregation(Aggregation::ExplicitBucketHistogram { + boundaries: duration_buckets(), + record_min_max: true, + }) + .build() + .ok() + }) + .flatten() + }); let registry = config .allow_prometheus_metrics @@ -70,7 +75,32 @@ pub(crate) fn init( } let provider = builder.build(); - let meter = provider.meter(conduwuit_core::BRANDING); + global::set_meter_provider(provider.clone()); + let meter = provider.meter_with_scope( + InstrumentationScope::builder(conduwuit_core::BRANDING) + .with_version(conduwuit_core::version()) + .with_schema_url(SCHEMA_URL) + .build(), + ); Ok((Metrics::new(runtime.cloned(), meter, registry), provider)) } + +pub(crate) fn resource(config: &Config) -> Resource { + // 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(attribute::SERVICE_VERSION, conduwuit_core::version()), + KeyValue::new(attribute::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); + } + resource.build() +} diff --git a/src/main/mod.rs b/src/main/mod.rs index 2582f15ab..cdc8e27ce 100644 --- a/src/main/mod.rs +++ b/src/main/mod.rs @@ -51,6 +51,7 @@ pub fn run_with_args(args: &Args) -> Result<()> { runtime.spawn(signal::signal(server.clone())); runtime.block_on(async_main(&server))?; + server.shutdown_metrics(); runtime::shutdown(&server, runtime); #[cfg(unix)] diff --git a/src/main/server.rs b/src/main/server.rs index afde40910..28b9039ef 100644 --- a/src/main/server.rs +++ b/src/main/server.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use conduwuit_core::{ Error, Result, config::Config, - info, + error, info, log::Log, utils::{stream, sys}, }; @@ -23,7 +23,7 @@ pub(crate) struct Server { _tracing_flame_guard: TracingFlameGuard, - _meter_provider: opentelemetry_sdk::metrics::SdkMeterProvider, + meter_provider: opentelemetry_sdk::metrics::SdkMeterProvider, #[cfg(feature = "sentry_telemetry")] _sentry_guard: Option<::sentry::ClientInitGuard>, @@ -83,7 +83,7 @@ impl Server { _tracing_flame_guard: tracing_flame_guard, - _meter_provider: meter_provider, + meter_provider, #[cfg(feature = "sentry_telemetry")] _sentry_guard: sentry_guard, @@ -92,4 +92,10 @@ impl Server { mods: tokio::sync::RwLock::new(Vec::new()), })) } + + pub(crate) fn shutdown_metrics(&self) { + if let Err(e) = self.meter_provider.shutdown() { + error!("Failed to shut down metrics: {e}"); + } + } } diff --git a/src/router/Cargo.toml b/src/router/Cargo.toml index 473da24d5..e1e66373a 100644 --- a/src/router/Cargo.toml +++ b/src/router/Cargo.toml @@ -111,6 +111,7 @@ http-body-util.workspace = true hyper.workspace = true hyper-util.workspace = true log.workspace = true +opentelemetry-instrumentation-tower.workspace = true ruma.workspace = true rustls.workspace = true rustls.optional = true diff --git a/src/router/layers.rs b/src/router/layers.rs index d2ec6c0d8..2e69dda38 100644 --- a/src/router/layers.rs +++ b/src/router/layers.rs @@ -11,6 +11,7 @@ use http::{ HeaderValue, Method, StatusCode, header::{self, HeaderName}, }; +use opentelemetry_instrumentation_tower::http::server::LayerBuilder as OtelHttpLayerBuilder; use tower::ServiceBuilder; use tower_http::{ catch_panic::CatchPanicLayer, @@ -49,6 +50,13 @@ pub(crate) fn build(services: &Arc) -> Result<(Router, Guard)> { let services_ = services.clone(); let layers = layers + .map_request(bound_method) + .layer( + OtelHttpLayerBuilder::builder() + .with_tracing(false) + .build() + .expect("valid HTTP metrics layer"), + ) .layer(SetSensitiveHeadersLayer::new([header::AUTHORIZATION])) .layer( TraceLayer::new_for_http() @@ -165,6 +173,30 @@ fn cors_layer(_server: &Server) -> CorsLayer { .max_age(Duration::from_hours(24)) } +/// Methods outside the standard set can never match a route; record them as +/// `_OTHER`, as the semantic conventions ask, so clients cannot create new +/// metric attribute values. +fn bound_method(mut request: http::Request) -> http::Request { + const KNOWN: [Method; 9] = [ + Method::GET, + Method::HEAD, + Method::POST, + Method::PUT, + Method::DELETE, + Method::CONNECT, + Method::OPTIONS, + Method::TRACE, + Method::PATCH, + ]; + + if !KNOWN.contains(request.method()) { + *request.method_mut() = + Method::from_bytes(b"_OTHER").expect("_OTHER is a valid method token"); + } + + request +} + fn body_limit_layer(server: &Server) -> DefaultBodyLimit { DefaultBodyLimit::max(server.config.max_request_size) } @@ -207,7 +239,7 @@ fn tracing_span(request: &http::Request) -> tracing::Span { let path = request .extensions() .get::() - .map_or_else(|| request_path_str(request), truncated_matched_path); + .map_or_else(|| request_path_str(request), MatchedPath::as_str); tracing::span! { parent: None, @@ -241,8 +273,22 @@ fn request_path_str(request: &http::Request) -> &str { .as_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)) +#[cfg(test)] +mod tests { + use http::{Method, Request}; + + use super::bound_method; + + #[test] + fn bound_method_collapses_unknown_methods() { + let request = |method: &[u8]| { + Request::builder() + .method(Method::from_bytes(method).unwrap()) + .body(()) + .unwrap() + }; + + assert_eq!(bound_method(request(b"GET")).method(), Method::GET); + assert_eq!(bound_method(request(b"BREW")).method().as_str(), "_OTHER"); + } } diff --git a/src/router/request.rs b/src/router/request.rs index ddbc1d542..d6a962727 100644 --- a/src/router/request.rs +++ b/src/router/request.rs @@ -1,26 +1,20 @@ use std::{ fmt::Debug, sync::{Arc, atomic::Ordering}, - time::{Duration, Instant}, + time::Duration, }; use axum::{ - extract::{MatchedPath, State}, + extract::State, response::{IntoResponse, Response}, }; -use conduwuit::{ - Result, debug, debug_error, debug_warn, err, error, - metrics::{KeyValue, method_label}, - trace, -}; +use conduwuit::{Result, debug, debug_error, debug_warn, err, error, 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>, @@ -37,22 +31,6 @@ pub(crate) async fn handle( return Err(StatusCode::SERVICE_UNAVAILABLE); } - 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(|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(); @@ -71,27 +49,9 @@ pub(crate) async fn handle( } }); - let result = task - .await + task.await .map_err(unhandled) - .and_then(|result| handle_result(&method, &uri, result)); - - let status = result - .as_ref() - .map_or_else(|status| *status, Response::status); - 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 + .and_then(|result| handle_result(&method, &uri, result)) } #[tracing::instrument(