mirror of
https://forgejo.ellis.link/continuwuation/continuwuity.git
synced 2026-11-14 12:46:49 +00:00
refactor(metrics): Use Tower HTTP instrumentation
This commit is contained in:
1 parent
4996471878
commit
9d454fd883
17 files changed
+264
-150
No files matched your search
Generated
+38
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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` |
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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`. \
|
||||
|
||||
@@ -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<f64>,
|
||||
pub http_server_active_requests: UpDownCounter<i64>,
|
||||
|
||||
pub federation_outgoing_request_duration: Histogram<f64>,
|
||||
pub federation_incoming_transaction_duration: Histogram<f64>,
|
||||
pub federation_incoming_pdus: Counter<u64>,
|
||||
@@ -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<f64> { (0..16).map(|i| 0.001 * f64::from(1_u32 << i)).collect() }
|
||||
pub fn duration_buckets() -> Vec<f64> { (0..16).map(|i| 0.001 * f64::from(1_u32 << i)).collect() }
|
||||
|
||||
fn size_buckets() -> Vec<f64> { (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| {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+3
-13
@@ -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);
|
||||
|
||||
|
||||
+53
-23
@@ -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()
|
||||
}
|
||||
@@ -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)]
|
||||
|
||||
+9
-3
@@ -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}");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+51
-5
@@ -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<Services>) -> 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<B>(mut request: http::Request<B>) -> http::Request<B> {
|
||||
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<T>(request: &http::Request<T>) -> tracing::Span {
|
||||
let path = request
|
||||
.extensions()
|
||||
.get::<MatchedPath>()
|
||||
.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<T>(request: &http::Request<T>) -> &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");
|
||||
}
|
||||
}
|
||||
+5
-45
@@ -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<Arc<Services>>,
|
||||
@@ -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::<MatchedPath>()
|
||||
.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(
|
||||
|
||||
Reference in new issue
Block a user