matrix/authentication-service
1
0
Fork 0
mirror of https://github.com/matrix-org/matrix-authentication-service.git synced 2025-07-07 22:41:18 +03:00
Code Activity

OpenTelemetry upgrade

Browse Source
This commit is contained in:
Quentin Gliech
2023-08-07 14:20:37 +02:00
parent 6e8222c765
commit 699dfba55f
13 changed files with 225 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
Cargo.lock generated
View File

@ -3039,12 +3039,13 @@ dependencies = [
"mas-templates", "mas-templates",
"mas-tower", "mas-tower",
"oauth2-types", "oauth2-types",
"opentelemetry 0.20.0", "opentelemetry",
"opentelemetry-http 0.9.0", "opentelemetry-http",
"opentelemetry-jaeger", "opentelemetry-jaeger",
"opentelemetry-otlp", "opentelemetry-otlp",
"opentelemetry-prometheus", "opentelemetry-prometheus",
"opentelemetry-semantic-conventions 0.12.0", "opentelemetry-semantic-conventions",
"opentelemetry-stdout",
"opentelemetry-zipkin", "opentelemetry-zipkin",
"prometheus", "prometheus",
"rand 0.8.5", "rand 0.8.5",
@ -3192,7 +3193,8 @@ dependencies = [
"mas-templates", "mas-templates",
"mime", "mime",
"oauth2-types", "oauth2-types",
"opentelemetry 0.20.0", "opentelemetry",
"opentelemetry-semantic-conventions",
"pbkdf2", "pbkdf2",
"rand 0.8.5", "rand 0.8.5",
"rand_chacha 0.3.1", "rand_chacha 0.3.1",
@ -3231,7 +3233,7 @@ dependencies = [
"hyper-rustls", "hyper-rustls",
"mas-tower", "mas-tower",
"once_cell", "once_cell",
"opentelemetry 0.20.0", "opentelemetry",
"rustls", "rustls",
"rustls-native-certs", "rustls-native-certs",
"serde", "serde",
@ -3472,7 +3474,7 @@ dependencies = [
"mas-iana", "mas-iana",
"mas-jose", "mas-jose",
"oauth2-types", "oauth2-types",
"opentelemetry 0.20.0", "opentelemetry",
"rand_core 0.6.4", "rand_core 0.6.4",
"serde", "serde",
"serde_json", "serde_json",
@ -3527,7 +3529,7 @@ dependencies = [
"mas-storage", "mas-storage",
"mas-storage-pg", "mas-storage-pg",
"mas-tower", "mas-tower",
"opentelemetry 0.20.0", "opentelemetry",
"rand 0.8.5", "rand 0.8.5",
"rand_chacha 0.3.1", "rand_chacha 0.3.1",
"serde", "serde",
@ -3572,8 +3574,9 @@ version = "0.1.0"
dependencies = [ dependencies = [
"aws-smithy-http", "aws-smithy-http",
"http", "http",
"opentelemetry 0.20.0", "opentelemetry",
"opentelemetry-http 0.9.0", "opentelemetry-http",
"opentelemetry-semantic-conventions",
"pin-project-lite", "pin-project-lite",
"tokio", "tokio",
"tower", "tower",
@ -3876,36 +3879,14 @@ version = "0.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf"
[[package]]
name = "opentelemetry"
version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5f4b8347cc26099d3aeee044065ecc3ae11469796b4d65d065a23a584ed92a6f"
dependencies = [
"opentelemetry_api 0.19.0",
"opentelemetry_sdk 0.19.0",
]
[[package]] [[package]]
name = "opentelemetry" name = "opentelemetry"
version = "0.20.0" version = "0.20.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9591d937bc0e6d2feb6f71a559540ab300ea49955229c347a517a28d27784c54" checksum = "9591d937bc0e6d2feb6f71a559540ab300ea49955229c347a517a28d27784c54"
dependencies = [ dependencies = [
"opentelemetry_api 0.20.0", "opentelemetry_api",
"opentelemetry_sdk 0.20.0", "opentelemetry_sdk",
]
[[package]]
name = "opentelemetry-http"
version = "0.8.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a819b71d6530c4297b49b3cae2939ab3a8cc1b9f382826a1bc29dd0ca3864906"
dependencies = [
"async-trait",
"bytes 1.4.0",
"http",
"opentelemetry_api 0.19.0",
] ]
[[package]] [[package]]
@ -3918,7 +3899,7 @@ dependencies = [
"bytes 1.4.0", "bytes 1.4.0",
"http", "http",
"hyper", "hyper",
"opentelemetry_api 0.20.0", "opentelemetry_api",
"tokio", "tokio",
] ]
@ -3932,9 +3913,9 @@ dependencies = [
"futures-core", "futures-core",
"futures-util", "futures-util",
"http", "http",
"opentelemetry 0.20.0", "opentelemetry",
"opentelemetry-http 0.9.0", "opentelemetry-http",
"opentelemetry-semantic-conventions 0.12.0", "opentelemetry-semantic-conventions",
"thrift", "thrift",
"tokio", "tokio",
] ]
@ -3949,9 +3930,9 @@ dependencies = [
"futures-core", "futures-core",
"http", "http",
"opentelemetry-proto", "opentelemetry-proto",
"opentelemetry-semantic-conventions 0.12.0", "opentelemetry-semantic-conventions",
"opentelemetry_api 0.20.0", "opentelemetry_api",
"opentelemetry_sdk 0.20.0", "opentelemetry_sdk",
"prost", "prost",
"thiserror", "thiserror",
"tokio", "tokio",
@ -3965,8 +3946,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c7d81bc254e2d572120363a2b16cdb0d715d301b5789be0cfc26ad87e4e10e53" checksum = "c7d81bc254e2d572120363a2b16cdb0d715d301b5789be0cfc26ad87e4e10e53"
dependencies = [ dependencies = [
"once_cell", "once_cell",
"opentelemetry_api 0.20.0", "opentelemetry_api",
"opentelemetry_sdk 0.20.0", "opentelemetry_sdk",
"prometheus", "prometheus",
"protobuf", "protobuf",
] ]
@ -3977,64 +3958,55 @@ version = "0.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1e3f814aa9f8c905d0ee4bde026afd3b2577a97c10e1699912e3e44f0c4cbeb" checksum = "b1e3f814aa9f8c905d0ee4bde026afd3b2577a97c10e1699912e3e44f0c4cbeb"
dependencies = [ dependencies = [
"opentelemetry_api 0.20.0", "opentelemetry_api",
"opentelemetry_sdk 0.20.0", "opentelemetry_sdk",
"prost", "prost",
"tonic", "tonic",
] ]
[[package]]
name = "opentelemetry-semantic-conventions"
version = "0.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "24e33428e6bf08c6f7fcea4ddb8e358fab0fe48ab877a87c70c6ebe20f673ce5"
dependencies = [
"opentelemetry 0.19.0",
]
[[package]] [[package]]
name = "opentelemetry-semantic-conventions" name = "opentelemetry-semantic-conventions"
version = "0.12.0" version = "0.12.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73c9f9340ad135068800e7f1b24e9e09ed9e7143f5bf8518ded3d3ec69789269" checksum = "73c9f9340ad135068800e7f1b24e9e09ed9e7143f5bf8518ded3d3ec69789269"
dependencies = [ dependencies = [
"opentelemetry 0.20.0", "opentelemetry",
]
[[package]]
name = "opentelemetry-stdout"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8bd550321bc0f9d3f6dcbfe5c75262789de5b3e2776da2cbcfd2392aa05db0c6"
dependencies = [
"async-trait",
"futures-util",
"opentelemetry_api",
"opentelemetry_sdk",
"ordered-float 3.7.0",
"serde",
"serde_json",
] ]
[[package]] [[package]]
name = "opentelemetry-zipkin" name = "opentelemetry-zipkin"
version = "0.17.0" version = "0.18.0"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1fd48caee5e1db71454c95be32d1daeb6fae265321ff8f51b1efc8a50b0be80" checksum = "eb966f01235207a6933c0aec98374fe9782df1c1d2b3d1db35c458451d138143"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"futures-core", "futures-core",
"http", "http",
"once_cell", "once_cell",
"opentelemetry 0.19.0", "opentelemetry",
"opentelemetry-http 0.8.0", "opentelemetry-http",
"opentelemetry-semantic-conventions 0.11.0", "opentelemetry-semantic-conventions",
"serde", "serde",
"serde_json", "serde_json",
"thiserror", "thiserror",
"typed-builder", "typed-builder",
] ]
[[package]]
name = "opentelemetry_api"
version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed41783a5bf567688eb38372f2b7a8530f5a607a4b49d38dd7573236c23ca7e2"
dependencies = [
"futures-channel",
"futures-util",
"indexmap 1.9.3",
"once_cell",
"pin-project-lite",
"thiserror",
"urlencoding",
]
[[package]] [[package]]
name = "opentelemetry_api" name = "opentelemetry_api"
version = "0.20.0" version = "0.20.0"
@ -4051,24 +4023,6 @@ dependencies = [
"urlencoding", "urlencoding",
] ]
[[package]]
name = "opentelemetry_sdk"
version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b3a2a91fdbfdd4d212c0dcc2ab540de2c2bcbbd90be17de7a7daf8822d010c1"
dependencies = [
"async-trait",
"crossbeam-channel",
"futures-channel",
"futures-executor",
"futures-util",
"once_cell",
"opentelemetry_api 0.19.0",
"percent-encoding",
"rand 0.8.5",
"thiserror",
]
[[package]] [[package]]
name = "opentelemetry_sdk" name = "opentelemetry_sdk"
version = "0.20.0" version = "0.20.0"
@ -4081,7 +4035,7 @@ dependencies = [
"futures-executor", "futures-executor",
"futures-util", "futures-util",
"once_cell", "once_cell",
"opentelemetry_api 0.20.0", "opentelemetry_api",
"ordered-float 3.7.0", "ordered-float 3.7.0",
"percent-encoding", "percent-encoding",
"rand 0.8.5", "rand 0.8.5",
@ -6353,7 +6307,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fc09e402904a5261e42cf27aea09ccb7d5318c6717a9eec3d8e2e65c56b18f19" checksum = "fc09e402904a5261e42cf27aea09ccb7d5318c6717a9eec3d8e2e65c56b18f19"
dependencies = [ dependencies = [
"once_cell", "once_cell",
"opentelemetry 0.20.0", "opentelemetry",
"tracing", "tracing",
"tracing-core", "tracing-core",
"tracing-log", "tracing-log",

9
crates/cli/Cargo.toml
View File

@ -34,12 +34,13 @@ tracing-appender = "0.2.2"
tracing-subscriber = { version = "0.3.17", features = ["env-filter"] } tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
tracing-opentelemetry = "0.20.0" tracing-opentelemetry = "0.20.0"
opentelemetry = { version = "0.20.0", features = ["trace", "metrics", "rt-tokio"] } opentelemetry = { version = "0.20.0", features = ["trace", "metrics", "rt-tokio"] }
opentelemetry-semantic-conventions = "0.12.0" opentelemetry-http = { version = "0.9.0", features = ["tokio", "hyper"] }
opentelemetry-jaeger = { version = "0.19.0", features = ["rt-tokio", "collector_client"] } opentelemetry-jaeger = { version = "0.19.0", features = ["rt-tokio", "collector_client"] }
opentelemetry-otlp = { version = "0.13.0", features = ["trace", "metrics"] } opentelemetry-otlp = { version = "0.13.0", features = ["trace", "metrics"] }
opentelemetry-zipkin = { version = "0.17.0", features = ["opentelemetry-http"], default-features = false } opentelemetry-prometheus = "0.13.0"
opentelemetry-http = { version = "0.9.0", features = ["tokio", "hyper"] } opentelemetry-semantic-conventions = "0.12.0"
opentelemetry-prometheus = { version = "0.13.0" } opentelemetry-stdout = { version = "0.1.0", features = ["trace", "metrics"] }
opentelemetry-zipkin = { version = "0.18.0", default-features = false }
prometheus = "0.13.3" prometheus = "0.13.3"
sentry = { version = "0.31.5", default-features = false, features = ["backtrace", "contexts", "panic", "tower"] } sentry = { version = "0.31.5", default-features = false, features = ["backtrace", "contexts", "panic", "tower"] }
sentry-tracing = "0.31.5" sentry-tracing = "0.31.5"

2
crates/cli/src/main.rs
View File

@ -100,7 +100,7 @@ async fn try_main() -> anyhow::Result<()> {
}); });
// Setup OpenTelemetry tracing and metrics // Setup OpenTelemetry tracing and metrics
let (tracer, _meter) = telemetry::setup(&telemetry_config) let tracer = telemetry::setup(&telemetry_config)
.await .await
.context("failed to setup OpenTelemetry")?; .context("failed to setup OpenTelemetry")?;

98
crates/cli/src/server.rs
View File

@ -13,7 +13,6 @@
// limitations under the License. // limitations under the License.
use std::{ use std::{
borrow::Cow,
future::ready, future::ready,
net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, ToSocketAddrs}, net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr, TcpListener, ToSocketAddrs},
os::unix::net::UnixListener, os::unix::net::UnixListener,
@ -27,7 +26,7 @@ use axum::{
Extension, Router, Extension, Router,
}; };
use hyper::{ use hyper::{
header::{HeaderValue, CACHE_CONTROL}, header::{HeaderValue, CACHE_CONTROL, USER_AGENT},
Method, Request, Response, StatusCode, Version, Method, Request, Response, StatusCode, Version,
}; };
use listenfd::ListenFd; use listenfd::ListenFd;
@ -40,10 +39,11 @@ use mas_tower::{
make_span_fn, metrics_attributes_fn, DurationRecorderLayer, InFlightCounterLayer, TraceLayer, make_span_fn, metrics_attributes_fn, DurationRecorderLayer, InFlightCounterLayer, TraceLayer,
KV, KV,
}; };
use opentelemetry::{Key, KeyValue}; use opentelemetry::{trace::TraceContextExt, Key, KeyValue};
use opentelemetry_http::HeaderExtractor; use opentelemetry_http::HeaderExtractor;
use opentelemetry_semantic_conventions::trace::{ use opentelemetry_semantic_conventions::trace::{
HTTP_METHOD, HTTP_ROUTE, HTTP_SCHEME, HTTP_STATUS_CODE, HTTP_REQUEST_METHOD, HTTP_RESPONSE_STATUS_CODE, HTTP_ROUTE, NETWORK_PROTOCOL_NAME,
NETWORK_PROTOCOL_VERSION, URL_SCHEME,
}; };
use rustls::ServerConfig; use rustls::ServerConfig;
use sentry_tower::{NewSentryLayer, SentryHttpLayer}; use sentry_tower::{NewSentryLayer, SentryHttpLayer};
@ -52,35 +52,33 @@ use tower_http::{services::ServeDir, set_header::SetResponseHeaderLayer};
use tracing::{warn, Span}; use tracing::{warn, Span};
use tracing_opentelemetry::OpenTelemetrySpanExt; use tracing_opentelemetry::OpenTelemetrySpanExt;
const NET_PROTOCOL_NAME: Key = Key::from_static_str("net.protocol.name");
const NET_PROTOCOL_VERSION: Key = Key::from_static_str("net.protocol.version");
const MAS_LISTENER_NAME: Key = Key::from_static_str("mas.listener.name"); const MAS_LISTENER_NAME: Key = Key::from_static_str("mas.listener.name");
#[inline] #[inline]
fn otel_http_method<B>(request: &Request<B>) -> Cow<'static, str> { fn otel_http_method<B>(request: &Request<B>) -> &'static str {
match request.method() { match request.method() {
&Method::OPTIONS => "OPTIONS".into(), &Method::OPTIONS => "OPTIONS",
&Method::GET => "GET".into(), &Method::GET => "GET",
&Method::POST => "POST".into(), &Method::POST => "POST",
&Method::PUT => "PUT".into(), &Method::PUT => "PUT",
&Method::DELETE => "DELETE".into(), &Method::DELETE => "DELETE",
&Method::HEAD => "HEAD".into(), &Method::HEAD => "HEAD",
&Method::TRACE => "TRACE".into(), &Method::TRACE => "TRACE",
&Method::CONNECT => "CONNECT".into(), &Method::CONNECT => "CONNECT",
&Method::PATCH => "PATCH".into(), &Method::PATCH => "PATCH",
other => other.to_string().into(), _other => "_OTHER",
} }
} }
#[inline] #[inline]
fn otel_net_protocol_version<B>(request: &Request<B>) -> Cow<'static, str> { fn otel_net_protocol_version<B>(request: &Request<B>) -> &'static str {
match request.version() { match request.version() {
Version::HTTP_09 => "0.9".into(), Version::HTTP_09 => "0.9",
Version::HTTP_10 => "1.0".into(), Version::HTTP_10 => "1.0",
Version::HTTP_11 => "1.1".into(), Version::HTTP_11 => "1.1",
Version::HTTP_2 => "2.0".into(), Version::HTTP_2 => "2.0",
Version::HTTP_3 => "3.0".into(), Version::HTTP_3 => "3.0",
other => format!("{other:?}").into(), _other => "_OTHER",
} }
} }
@ -91,11 +89,7 @@ fn otel_http_route<B>(request: &Request<B>) -> Option<&str> {
.map(MatchedPath::as_str) .map(MatchedPath::as_str)
} }
fn otel_http_target<B>(request: &Request<B>) -> &str { fn otel_url_scheme<B>(request: &Request<B>) -> &'static str {
request.uri().path_and_query().map_or("", |p| p.as_str())
}
fn otel_http_scheme<B>(request: &Request<B>) -> &'static str {
// XXX: maybe we should panic if the connection info was not injected in the // XXX: maybe we should panic if the connection info was not injected in the
// request extensions // request extensions
request request
@ -117,7 +111,7 @@ fn make_http_span<B>(req: &Request<B>) -> Span {
let span_name = if let Some(route) = route.as_ref() { let span_name = if let Some(route) = route.as_ref() {
format!("{method} {route}") format!("{method} {route}")
} else { } else {
format!("{method}") method.to_owned()
}; };
let span = tracing::info_span!( let span = tracing::info_span!(
@ -125,42 +119,60 @@ fn make_http_span<B>(req: &Request<B>) -> Span {
"otel.kind" = "server", "otel.kind" = "server",
"otel.name" = span_name, "otel.name" = span_name,
"otel.status_code" = tracing::field::Empty, "otel.status_code" = tracing::field::Empty,
"net.protocol.name" = "http", "network.protocol.name" = "http",
"net.protocol.version" = otel_net_protocol_version(req).as_ref(), "network.protocol.version" = otel_net_protocol_version(req),
"http.scheme" = otel_http_scheme(req), "http.method" = method,
"http.method" = method.as_ref(),
"http.route" = tracing::field::Empty, "http.route" = tracing::field::Empty,
"http.target" = otel_http_target(req), "http.response.status_code" = tracing::field::Empty,
"http.status_code" = tracing::field::Empty, "url.path" = req.uri().path(),
"url.query" = tracing::field::Empty,
"url.scheme" = otel_url_scheme(req),
"user_agent.original" = tracing::field::Empty,
); );
if let Some(route) = route.as_ref() { if let Some(route) = route.as_ref() {
span.record("http.route", route); span.record("http.route", route);
} }
if let Some(query) = req.uri().query() {
span.record("url.query", query);
}
if let Some(user_agent) = req.headers().get(USER_AGENT) {
span.record(
"user_agent.original",
user_agent.to_str().unwrap_or("INVALID"),
);
}
// Extract the parent span context from the request headers // Extract the parent span context from the request headers
let parent_context = opentelemetry::global::get_text_map_propagator(|propagator| { let parent_context = opentelemetry::global::get_text_map_propagator(|propagator| {
let extractor = HeaderExtractor(req.headers()); let extractor = HeaderExtractor(req.headers());
let context = opentelemetry::Context::new(); let context = opentelemetry::Context::new();
propagator.extract_with_context(&context, &extractor) propagator.extract_with_context(&context, &extractor)
}); });
span.set_parent(parent_context);
if parent_context.span().span_context().is_remote() {
// For now, set_parent is broken, so in the meantime we're using add_link
// instead
span.add_link(parent_context.span().span_context().clone());
}
span span
} }
fn on_http_request_labels<B>(request: &Request<B>) -> Vec<KeyValue> { fn on_http_request_labels<B>(request: &Request<B>) -> Vec<KeyValue> {
vec![ vec![
NET_PROTOCOL_NAME.string("http"), NETWORK_PROTOCOL_NAME.string("http"),
NET_PROTOCOL_VERSION.string(otel_net_protocol_version(request)), NETWORK_PROTOCOL_VERSION.string(otel_net_protocol_version(request)),
HTTP_METHOD.string(otel_http_method(request)), HTTP_REQUEST_METHOD.string(otel_http_method(request)),
HTTP_SCHEME.string(otel_http_scheme(request).as_ref()),
HTTP_ROUTE.string(otel_http_route(request).unwrap_or("FALLBACK").to_owned()), HTTP_ROUTE.string(otel_http_route(request).unwrap_or("FALLBACK").to_owned()),
URL_SCHEME.string(otel_url_scheme(request).as_ref()),
] ]
} }
fn on_http_response_labels<B>(res: &Response<B>) -> Vec<KeyValue> { fn on_http_response_labels<B>(res: &Response<B>) -> Vec<KeyValue> {
vec![HTTP_STATUS_CODE.i64(res.status().as_u16().into())] vec![HTTP_RESPONSE_STATUS_CODE.i64(res.status().as_u16().into())]
} }
pub fn build_router<B>( pub fn build_router<B>(
@ -259,7 +271,7 @@ where
)) ))
.on_response_fn(|span: &Span, response: &Response<_>| { .on_response_fn(|span: &Span, response: &Response<_>| {
let status_code = response.status().as_u16(); let status_code = response.status().as_u16();
span.record("http.status_code", status_code); span.record("http.response.status_code", status_code);
span.record("otel.status_code", "OK"); span.record("otel.status_code", "OK");
}), }),
) )

178
crates/cli/src/telemetry.rs
View File

@ -25,26 +25,29 @@ use opentelemetry::{
propagation::TextMapPropagator, propagation::TextMapPropagator,
sdk::{ sdk::{
self, self,
metrics::controllers::BasicController, metrics::{
reader::{DefaultAggregationSelector, DefaultTemporalitySelector},
MeterProvider, PeriodicReader,
},
propagation::{BaggagePropagator, TextMapCompositePropagator, TraceContextPropagator}, propagation::{BaggagePropagator, TextMapCompositePropagator, TraceContextPropagator},
trace::{Sampler, Tracer}, trace::{Sampler, Tracer, TracerProvider},
Resource, Resource,
}, },
Context, trace::TracerProvider as _,
}; };
use opentelemetry_jaeger::Propagator as JaegerPropagator; use opentelemetry_jaeger::Propagator as JaegerPropagator;
use opentelemetry_otlp::MetricsExporterBuilder;
use opentelemetry_prometheus::PrometheusExporter; use opentelemetry_prometheus::PrometheusExporter;
use opentelemetry_semantic_conventions as semcov; use opentelemetry_semantic_conventions as semcov;
use opentelemetry_zipkin::{B3Encoding, Propagator as ZipkinPropagator}; use opentelemetry_zipkin::{B3Encoding, Propagator as ZipkinPropagator};
use prometheus::Registry;
use tokio::sync::OnceCell; use tokio::sync::OnceCell;
use url::Url; use url::Url;
static METRICS_BASIC_CONTROLLER: OnceCell<BasicController> = OnceCell::const_new(); static METER_PROVIDER: OnceCell<MeterProvider> = OnceCell::const_new();
static PROMETHEUS_EXPORTER: OnceCell<PrometheusExporter> = OnceCell::const_new(); static PROMETHEUS_REGISTRY: OnceCell<Registry> = OnceCell::const_new();
pub async fn setup( pub async fn setup(config: &TelemetryConfig) -> anyhow::Result<Option<Tracer>> {
config: &TelemetryConfig,
) -> anyhow::Result<(Option<Tracer>, Option<BasicController>)> {
global::set_error_handler(|e| tracing::error!("{}", e))?; global::set_error_handler(|e| tracing::error!("{}", e))?;
let propagator = propagator(&config.tracing.propagators); let propagator = propagator(&config.tracing.propagators);
@ -57,20 +60,16 @@ pub async fn setup(
.await .await
.context("Failed to configure traces exporter")?; .context("Failed to configure traces exporter")?;
let meter = meter(&config.metrics.exporter).context("Failed to configure metrics exporter")?; init_meter(&config.metrics.exporter).context("Failed to configure metrics exporter")?;
if let Some(meter) = meter.as_ref() {
METRICS_BASIC_CONTROLLER.set(meter.clone())?;
}
Ok((tracer, meter)) Ok(tracer)
} }
pub fn shutdown() { pub fn shutdown() {
global::shutdown_tracer_provider(); global::shutdown_tracer_provider();
if let Some(controller) = METRICS_BASIC_CONTROLLER.get() { if let Some(meter_provider) = METER_PROVIDER.get() {
let cx = Context::new(); meter_provider.shutdown().unwrap();
controller.stop(&cx).unwrap();
} }
} }
@ -100,11 +99,11 @@ async fn http_client() -> anyhow::Result<impl opentelemetry_http::HttpClient + '
Ok(client) Ok(client)
} }
fn stdout_tracer() -> Tracer { fn stdout_tracer_provider() -> TracerProvider {
sdk::export::trace::stdout::new_pipeline() let exporter = opentelemetry_stdout::SpanExporter::default();
.with_pretty_print(true) TracerProvider::builder()
.with_trace_config(trace_config()) .with_simple_exporter(exporter)
.install_simple() .build()
} }
fn otlp_tracer(endpoint: Option<&Url>) -> anyhow::Result<Tracer> { fn otlp_tracer(endpoint: Option<&Url>) -> anyhow::Result<Tracer> {
@ -125,24 +124,24 @@ fn otlp_tracer(endpoint: Option<&Url>) -> anyhow::Result<Tracer> {
Ok(tracer) Ok(tracer)
} }
fn jaeger_agent_tracer(host: &str, port: u16) -> anyhow::Result<Tracer> { fn jaeger_agent_tracer_provider(host: &str, port: u16) -> anyhow::Result<TracerProvider> {
let pipeline = opentelemetry_jaeger::new_agent_pipeline() let pipeline = opentelemetry_jaeger::new_agent_pipeline()
.with_service_name(env!("CARGO_PKG_NAME")) .with_service_name(env!("CARGO_PKG_NAME"))
.with_trace_config(trace_config()) .with_trace_config(trace_config())
.with_endpoint((host, port)); .with_endpoint((host, port));
let tracer = pipeline let tracer_provider = pipeline
.install_batch(opentelemetry::runtime::Tokio) .build_batch(opentelemetry::runtime::Tokio)
.context("Failed to configure Jaeger agent exporter")?; .context("Failed to configure Jaeger agent exporter")?;
Ok(tracer) Ok(tracer_provider)
} }
async fn jaeger_collector_tracer( async fn jaeger_collector_tracer_provider(
endpoint: &str, endpoint: &str,
username: Option<&str>, username: Option<&str>,
password: Option<&str>, password: Option<&str>,
) -> anyhow::Result<Tracer> { ) -> anyhow::Result<TracerProvider> {
let http_client = http_client().await?; let http_client = http_client().await?;
let mut pipeline = opentelemetry_jaeger::new_collector_pipeline() let mut pipeline = opentelemetry_jaeger::new_collector_pipeline()
.with_service_name(env!("CARGO_PKG_NAME")) .with_service_name(env!("CARGO_PKG_NAME"))
@ -158,11 +157,11 @@ async fn jaeger_collector_tracer(
pipeline = pipeline.with_password(password); pipeline = pipeline.with_password(password);
} }
let tracer = pipeline let tracer_provider = pipeline
.install_batch(opentelemetry::runtime::Tokio) .build_batch(opentelemetry::runtime::Tokio)
.context("Failed to configure Jaeger collector exporter")?; .context("Failed to configure Jaeger collector exporter")?;
Ok(tracer) Ok(tracer_provider)
} }
async fn zipkin_tracer(collector_endpoint: &Option<Url>) -> anyhow::Result<Tracer> { async fn zipkin_tracer(collector_endpoint: &Option<Url>) -> anyhow::Result<Tracer> {
@ -185,28 +184,43 @@ async fn zipkin_tracer(collector_endpoint: &Option<Url>) -> anyhow::Result<Trace
} }
async fn tracer(config: &TracingExporterConfig) -> anyhow::Result<Option<Tracer>> { async fn tracer(config: &TracingExporterConfig) -> anyhow::Result<Option<Tracer>> {
let tracer = match config { let tracer_provider = match config {
TracingExporterConfig::None => return Ok(None), TracingExporterConfig::None => return Ok(None),
TracingExporterConfig::Stdout => stdout_tracer(), TracingExporterConfig::Stdout => stdout_tracer_provider(),
TracingExporterConfig::Otlp { endpoint } => otlp_tracer(endpoint.as_ref())?, TracingExporterConfig::Otlp { endpoint } => {
// The OTLP exporter already creates a tracer and installs it
return Ok(Some(otlp_tracer(endpoint.as_ref())?));
}
TracingExporterConfig::Jaeger(JaegerExporterProtocolConfig::UdpThriftCompact { TracingExporterConfig::Jaeger(JaegerExporterProtocolConfig::UdpThriftCompact {
agent_host, agent_host,
agent_port, agent_port,
}) => jaeger_agent_tracer(agent_host, *agent_port)?, }) => jaeger_agent_tracer_provider(agent_host, *agent_port)?,
TracingExporterConfig::Jaeger(JaegerExporterProtocolConfig::HttpThriftBinary { TracingExporterConfig::Jaeger(JaegerExporterProtocolConfig::HttpThriftBinary {
endpoint, endpoint,
username, username,
password, password,
}) => jaeger_collector_tracer(endpoint, username.as_deref(), password.as_deref()).await?, }) => {
jaeger_collector_tracer_provider(endpoint, username.as_deref(), password.as_deref())
.await?
}
TracingExporterConfig::Zipkin { collector_endpoint } => { TracingExporterConfig::Zipkin { collector_endpoint } => {
zipkin_tracer(collector_endpoint).await? // The Zipkin exporter already creates a tracer and installs it
return Ok(Some(zipkin_tracer(collector_endpoint).await?));
} }
}; };
let tracer = tracer_provider.versioned_tracer(
env!("CARGO_PKG_NAME"),
Some(env!("CARGO_PKG_VERSION")),
Some(semcov::SCHEMA_URL),
None,
);
global::set_tracer_provider(tracer_provider);
Ok(Some(tracer)) Ok(Some(tracer))
} }
fn otlp_meter(endpoint: Option<&url::Url>) -> anyhow::Result<BasicController> { fn otlp_metric_reader(endpoint: Option<&url::Url>) -> anyhow::Result<PeriodicReader> {
use opentelemetry_otlp::WithExportConfig; use opentelemetry_otlp::WithExportConfig;
let mut exporter = opentelemetry_otlp::new_exporter().tonic(); let mut exporter = opentelemetry_otlp::new_exporter().tonic();
@ -214,35 +228,17 @@ fn otlp_meter(endpoint: Option<&url::Url>) -> anyhow::Result<BasicController> {
exporter = exporter.with_endpoint(endpoint.to_string()); exporter = exporter.with_endpoint(endpoint.to_string());
} }
let controller = opentelemetry_otlp::new_pipeline() let exporter = MetricsExporterBuilder::from(exporter).build_metrics_exporter(
.metrics( Box::new(DefaultTemporalitySelector::new()),
sdk::metrics::selectors::simple::inexpensive(), Box::new(DefaultAggregationSelector::new()),
sdk::export::metrics::aggregation::cumulative_temporality_selector(), )?;
opentelemetry::runtime::Tokio,
)
.with_resource(resource())
.with_exporter(exporter)
.build()
.context("Failed to configure OTLP metrics exporter")?;
Ok(controller) Ok(PeriodicReader::builder(exporter, opentelemetry::runtime::Tokio).build())
} }
fn stdout_meter() -> anyhow::Result<BasicController> { fn stdout_metric_reader() -> PeriodicReader {
let exporter = sdk::export::metrics::stdout().build()?; let exporter = opentelemetry_stdout::MetricsExporter::default();
let controller = sdk::metrics::controllers::basic(sdk::metrics::processors::factory( PeriodicReader::builder(exporter, opentelemetry::runtime::Tokio).build()
sdk::metrics::selectors::simple::inexpensive(),
exporter.temporality_selector(),
))
.with_resource(resource())
.with_exporter(exporter)
.build();
let cx = Context::new();
controller.start(&cx, opentelemetry::runtime::Tokio)?;
global::set_meter_provider(controller.clone());
Ok(controller)
} }
pub fn prometheus_service<T>() -> tower::util::ServiceFn< pub fn prometheus_service<T>() -> tower::util::ServiceFn<
@ -250,15 +246,15 @@ pub fn prometheus_service<T>() -> tower::util::ServiceFn<
> { > {
use prometheus::{Encoder, TextEncoder}; use prometheus::{Encoder, TextEncoder};
if !PROMETHEUS_EXPORTER.initialized() { if !PROMETHEUS_REGISTRY.initialized() {
tracing::warn!("A Prometheus resource was mounted on a listener, but the Prometheus exporter was not setup in the config"); tracing::warn!("A Prometheus resource was mounted on a listener, but the Prometheus exporter was not setup in the config");
} }
tower::service_fn(move |_req| { tower::service_fn(move |_req| {
let response = if let Some(exporter) = PROMETHEUS_EXPORTER.get() { let response = if let Some(registry) = PROMETHEUS_REGISTRY.get() {
let mut buffer = vec![]; let mut buffer = vec![];
let encoder = TextEncoder::new(); let encoder = TextEncoder::new();
let metric_families = exporter.registry().gather(); let metric_families = registry.gather();
// That shouldn't panic, unless we're constructing invalid labels // That shouldn't panic, unless we're constructing invalid labels
encoder.encode(&metric_families, &mut buffer).unwrap(); encoder.encode(&metric_families, &mut buffer).unwrap();
@ -280,32 +276,41 @@ pub fn prometheus_service<T>() -> tower::util::ServiceFn<
}) })
} }
fn prometheus_meter() -> anyhow::Result<BasicController> { fn prometheus_metric_reader() -> anyhow::Result<PrometheusExporter> {
let controller = sdk::metrics::controllers::basic(sdk::metrics::processors::factory( let registry = Registry::new();
// All histogram metrics are in milliseconds. Each bucket is ~2x the previous one. PROMETHEUS_REGISTRY.set(registry.clone())?;
sdk::metrics::selectors::simple::histogram([
1.0, 3.0, 5.0, 10.0, 30.0, 50.0, 100.0, 300.0, 1000.0,
]),
sdk::export::metrics::aggregation::cumulative_temporality_selector(),
))
.with_resource(resource())
.build();
let exporter = opentelemetry_prometheus::exporter(controller.clone()).try_init()?; let exporter = opentelemetry_prometheus::exporter()
PROMETHEUS_EXPORTER.set(exporter)?; .with_registry(registry)
.without_scope_info()
.build()?;
Ok(controller) Ok(exporter)
} }
fn meter(config: &MetricsExporterConfig) -> anyhow::Result<Option<BasicController>> { fn init_meter(config: &MetricsExporterConfig) -> anyhow::Result<()> {
let controller = match config { let mut meter_provider_builder = MeterProvider::builder();
MetricsExporterConfig::None => None, match config {
MetricsExporterConfig::Stdout => Some(stdout_meter()?), MetricsExporterConfig::None => {}
MetricsExporterConfig::Otlp { endpoint } => Some(otlp_meter(endpoint.as_ref())?), MetricsExporterConfig::Stdout => {
MetricsExporterConfig::Prometheus => Some(prometheus_meter()?), meter_provider_builder = meter_provider_builder.with_reader(stdout_metric_reader());
}
MetricsExporterConfig::Otlp { endpoint } => {
meter_provider_builder =
meter_provider_builder.with_reader(otlp_metric_reader(endpoint.as_ref())?);
}
MetricsExporterConfig::Prometheus => {
meter_provider_builder =
meter_provider_builder.with_reader(prometheus_metric_reader()?);
}
}; };
Ok(controller) let meter_provider = meter_provider_builder.with_resource(resource()).build();
METER_PROVIDER.set(meter_provider.clone())?;
global::set_meter_provider(meter_provider.clone());
Ok(())
} }
fn trace_config() -> sdk::trace::Config { fn trace_config() -> sdk::trace::Config {
@ -326,6 +331,7 @@ fn resource() -> Resource {
Box::new(sdk::resource::EnvResourceDetector::new()), Box::new(sdk::resource::EnvResourceDetector::new()),
Box::new(sdk::resource::OsResourceDetector), Box::new(sdk::resource::OsResourceDetector),
Box::new(sdk::resource::ProcessResourceDetector), Box::new(sdk::resource::ProcessResourceDetector),
Box::new(sdk::resource::TelemetryResourceDetector),
], ],
); );

1
crates/handlers/Cargo.toml
View File

@ -13,6 +13,7 @@ futures-util = "0.3.28"
# Logging and tracing # Logging and tracing
tracing = "0.1.37" tracing = "0.1.37"
opentelemetry = "0.20.0" opentelemetry = "0.20.0"
opentelemetry-semantic-conventions = "0.12.0"
# Error management # Error management
thiserror = "1.0.44" thiserror = "1.0.44"

19
crates/handlers/src/app_state.rs
View File

@ -29,7 +29,7 @@ use mas_storage_pg::PgRepository;
use mas_templates::Templates; use mas_templates::Templates;
use opentelemetry::{ use opentelemetry::{
metrics::{Histogram, MetricsError, Unit}, metrics::{Histogram, MetricsError, Unit},
Context, KeyValue, KeyValue,
}; };
use rand::SeedableRng; use rand::SeedableRng;
use sqlx::PgPool; use sqlx::PgPool;
@ -60,7 +60,12 @@ impl AppState {
/// Returns an error if the metrics could not be initialized. /// Returns an error if the metrics could not be initialized.
pub fn init_metrics(&mut self) -> Result<(), MetricsError> { pub fn init_metrics(&mut self) -> Result<(), MetricsError> {
// XXX: do we want to put that somewhere else? // XXX: do we want to put that somewhere else?
let meter = opentelemetry::global::meter("mas-handlers"); let meter = opentelemetry::global::meter_with_version(
env!("CARGO_PKG_NAME"),
Some(env!("CARGO_PKG_VERSION")),
Some(opentelemetry_semantic_conventions::SCHEMA_URL),
None,
);
let pool = self.pool.clone(); let pool = self.pool.clone();
let usage = meter let usage = meter
.i64_observable_up_down_counter("db.connections.usage") .i64_observable_up_down_counter("db.connections.usage")
@ -75,13 +80,13 @@ impl AppState {
.init(); .init();
// Observe the number of active and idle connections in the pool // Observe the number of active and idle connections in the pool
meter.register_callback(move |cx| { meter.register_callback(&[usage.as_any(), max.as_any()], move |observer| {
let idle = u32::try_from(pool.num_idle()).unwrap_or(u32::MAX); let idle = u32::try_from(pool.num_idle()).unwrap_or(u32::MAX);
let used = pool.size() - idle; let used = pool.size() - idle;
let max_conn = pool.options().get_max_connections(); let max_conn = pool.options().get_max_connections();
usage.observe(cx, i64::from(idle), &[KeyValue::new("state", "idle")]); observer.observe_i64(&usage, i64::from(idle), &[KeyValue::new("state", "idle")]);
usage.observe(cx, i64::from(used), &[KeyValue::new("state", "used")]); observer.observe_i64(&usage, i64::from(used), &[KeyValue::new("state", "used")]);
max.observe(cx, i64::from(max_conn), &[]); observer.observe_i64(&max, i64::from(max_conn), &[]);
})?; })?;
// Track the connection acquisition time // Track the connection acquisition time
@ -212,7 +217,7 @@ impl FromRequestParts<AppState> for BoxRepository {
let duration_ms = duration.as_millis().try_into().unwrap_or(u64::MAX); let duration_ms = duration.as_millis().try_into().unwrap_or(u64::MAX);
if let Some(histogram) = &state.conn_acquisition_histogram { if let Some(histogram) = &state.conn_acquisition_histogram {
histogram.record(&Context::new(), duration_ms, &[]); histogram.record(duration_ms, &[]);
} }
Ok(repo Ok(repo

2
crates/handlers/src/views/index.rs
View File

@ -42,5 +42,7 @@ pub async fn get(
let content = templates.render_index(&ctx).await?; let content = templates.render_index(&ctx).await?;
tracing::info!("rendered index page");
Ok((cookie_jar, Html(content))) Ok((cookie_jar, Html(content)))
} }

1
crates/tower/Cargo.toml
View File

@ -14,6 +14,7 @@ tower = "0.4.13"
tokio = { version = "1.30.0", features = ["time"] } tokio = { version = "1.30.0", features = ["time"] }
opentelemetry = { version = "0.20.0", features = ["metrics"] } opentelemetry = { version = "0.20.0", features = ["metrics"] }
opentelemetry-http = "0.9.0" opentelemetry-http = "0.9.0"
opentelemetry-semantic-conventions = "0.12.0"
pin-project-lite = "0.2.12" pin-project-lite = "0.2.12"
[features] [features]

1
crates/tower/src/lib.rs
View File

@ -27,6 +27,7 @@ fn meter() -> opentelemetry::metrics::Meter {
opentelemetry::global::meter_with_version( opentelemetry::global::meter_with_version(
env!("CARGO_PKG_NAME"), env!("CARGO_PKG_NAME"),
Some(env!("CARGO_PKG_VERSION")), Some(env!("CARGO_PKG_VERSION")),
Some(opentelemetry_semantic_conventions::SCHEMA_URL),
None, None,
) )
} }

5
crates/tower/src/metrics/duration.rs
View File

@ -14,7 +14,7 @@
use std::future::Future; use std::future::Future;
use opentelemetry::{metrics::Histogram, Context, KeyValue}; use opentelemetry::{metrics::Histogram, KeyValue};
use pin_project_lite::pin_project; use pin_project_lite::pin_project;
use tokio::time::Instant; use tokio::time::Instant;
use tower::{Layer, Service}; use tower::{Layer, Service};
@ -195,8 +195,7 @@ where
} }
} }
this.histogram this.histogram.record(duration_ms, &attributes);
.record(&Context::new(), duration_ms, &attributes);
std::task::Poll::Ready(result) std::task::Poll::Ready(result)
} }
} }

6
crates/tower/src/metrics/in_flight.rs
View File

@ -16,7 +16,7 @@ use std::future::Future;
use opentelemetry::{ use opentelemetry::{
metrics::{Unit, UpDownCounter}, metrics::{Unit, UpDownCounter},
Context, KeyValue, KeyValue,
}; };
use pin_project_lite::pin_project; use pin_project_lite::pin_project;
use tower::{Layer, Service}; use tower::{Layer, Service};
@ -98,7 +98,7 @@ struct InFlightGuard {
impl InFlightGuard { impl InFlightGuard {
fn new(counter: UpDownCounter<i64>, attributes: Vec<KeyValue>) -> Self { fn new(counter: UpDownCounter<i64>, attributes: Vec<KeyValue>) -> Self {
counter.add(&Context::new(), 1, &attributes); counter.add(1, &attributes);
Self { Self {
counter, counter,
@ -109,7 +109,7 @@ impl InFlightGuard {
impl Drop for InFlightGuard { impl Drop for InFlightGuard {
fn drop(&mut self) { fn drop(&mut self) {
self.counter.add(&Context::new(), -1, &self.attributes); self.counter.add(-1, &self.attributes);
} }
} }

3
crates/tower/src/tracing/future.rs
View File

@ -54,7 +54,8 @@ where
// Poll the inner future, with the span entered. This is effectively what // Poll the inner future, with the span entered. This is effectively what
// [`tracing::Instrumented`] does. // [`tracing::Instrumented`] does.
let result = ready!(this.span.in_scope(|| this.inner.poll(cx))); let _guard = this.span.enter();
let result = ready!(this.inner.poll(cx));
match &result { match &result {
Ok(response) => { Ok(response) => {