From 81d33a673f8a95a8597f362188760e51c71fc3c4 Mon Sep 17 00:00:00 2001 From: "Aode (lion)" Date: Mon, 6 Dec 2021 21:45:00 -0600 Subject: [PATCH] Add otlp export --- Cargo.lock | 266 ++++++++++++++++++++++++++ Cargo.toml | 5 +- config/defaults.hjson | 1 + crates/apub_lib/src/activity_queue.rs | 7 +- crates/utils/src/settings/structs.rs | 4 + crates/websocket/Cargo.toml | 2 + crates/websocket/src/handlers.rs | 49 +++-- src/lib.rs | 32 +++- src/main.rs | 4 +- 9 files changed, 348 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index cef3b7ac1..977106b01 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -352,6 +352,27 @@ dependencies = [ "event-listener", ] +[[package]] +name = "async-stream" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "171374e7e3b2504e0e5236e3b59260560f9fe94bfe9ac39ba5e4e929c5590625" +dependencies = [ + "async-stream-impl", + "futures-core", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "648ed8c8d2ce5409ccd57453d9d1b214b342a0d69376a6feda1fd6cae3299308" +dependencies = [ + "proc-macro2 1.0.33", + "quote 1.0.10", + "syn 1.0.82", +] + [[package]] name = "async-trait" version = "0.1.51" @@ -1107,6 +1128,12 @@ version = "0.4.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "31586bda1b136406162e381a3185a506cdfc1631708dd40cba2f6628d8634499" +[[package]] +name = "fixedbitset" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37ab347416e802de484e4d03c7316c48f1ecb56574dfd4a46a80f173ce1de04d" + [[package]] name = "flate2" version = "1.0.22" @@ -1530,6 +1557,18 @@ dependencies = [ "want", ] +[[package]] +name = "hyper-timeout" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1" +dependencies = [ + "hyper", + "pin-project-lite", + "tokio", + "tokio-io-timeout", +] + [[package]] name = "hyper-tls" version = "0.5.0" @@ -1982,6 +2021,8 @@ dependencies = [ "lemmy_utils", "lemmy_websocket", "openssl", + "opentelemetry", + "opentelemetry-otlp", "reqwest", "reqwest-middleware", "reqwest-tracing", @@ -1993,6 +2034,7 @@ dependencies = [ "tracing-actix-web", "tracing-error", "tracing-log", + "tracing-opentelemetry", "tracing-subscriber", "url", ] @@ -2051,6 +2093,7 @@ dependencies = [ "lemmy_db_views", "lemmy_db_views_actor", "lemmy_utils", + "opentelemetry", "rand 0.8.4", "reqwest", "reqwest-middleware", @@ -2060,6 +2103,7 @@ dependencies = [ "strum_macros", "tokio", "tracing", + "tracing-opentelemetry", ] [[package]] @@ -2336,6 +2380,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "multimap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a" + [[package]] name = "native-tls" version = "0.2.8" @@ -2506,6 +2556,42 @@ dependencies = [ "vcpkg", ] +[[package]] +name = "opentelemetry" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf9b1c4e9a6c4de793c632496fa490bdc0e1eea73f0c91394f7b6990935d22" +dependencies = [ + "async-trait", + "crossbeam-channel", + "futures", + "js-sys", + "lazy_static", + "percent-encoding", + "pin-project", + "rand 0.8.4", + "thiserror", + "tokio", + "tokio-stream", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f19d4b43842433c420c548c985d158f5628bba5b518e0be64627926d19889992" +dependencies = [ + "async-trait", + "futures", + "http", + "opentelemetry", + "prost", + "thiserror", + "tokio", + "tonic", + "tonic-build", +] + [[package]] name = "parking_lot" version = "0.11.2" @@ -2597,6 +2683,16 @@ dependencies = [ "sha-1 0.8.2", ] +[[package]] +name = "petgraph" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "467d164a6de56270bd7c4d070df81d07beace25012d5103ced4e9ff08d6afdb7" +dependencies = [ + "fixedbitset", + "indexmap", +] + [[package]] name = "phf" version = "0.7.24" @@ -2768,6 +2864,57 @@ dependencies = [ "unicode-xid 0.2.2", ] +[[package]] +name = "prost" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de5e2533f59d08fcf364fd374ebda0692a70bd6d7e66ef97f306f45c6c5d8020" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-build" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "355f634b43cdd80724ee7848f95770e7e70eefa6dcf14fea676216573b8fd603" +dependencies = [ + "bytes", + "heck", + "itertools", + "log", + "multimap", + "petgraph", + "prost", + "prost-types", + "tempfile", + "which", +] + +[[package]] +name = "prost-derive" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "600d2f334aa05acb02a755e217ef1ab6dea4d51b58b7846588b747edec04efba" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2 1.0.33", + "quote 1.0.10", + "syn 1.0.82", +] + +[[package]] +name = "prost-types" +version = "0.8.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "603bbd6394701d13f3f25aada59c7de9d35a6a5887cfc156181234a44002771b" +dependencies = [ + "bytes", + "prost", +] + [[package]] name = "quick-xml" version = "0.20.0" @@ -3164,11 +3311,13 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "89af431b8c46776b5071a9a739c2b5fadbed6be2c6158d1ac5f71c4da3d2261c" dependencies = [ "async-trait", + "opentelemetry", "reqwest", "reqwest-middleware", "task-local-extensions", "tokio", "tracing", + "tracing-opentelemetry", ] [[package]] @@ -3933,6 +4082,7 @@ dependencies = [ "libc", "memchr", "mio 0.7.14", + "num_cpus", "once_cell", "parking_lot", "pin-project-lite", @@ -3941,6 +4091,16 @@ dependencies = [ "winapi", ] +[[package]] +name = "tokio-io-timeout" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90c49f106be240de154571dd31fbe48acb10ba6c6dd6f6517ad603abffa42de9" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-macros" version = "1.6.0" @@ -3973,6 +4133,17 @@ dependencies = [ "webpki", ] +[[package]] +name = "tokio-stream" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "50145484efff8818b5ccd256697f36863f587da82cf8b409c53adf1e840798e3" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-util" version = "0.6.9" @@ -3996,6 +4167,76 @@ dependencies = [ "serde", ] +[[package]] +name = "tonic" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "796c5e1cd49905e65dd8e700d4cb1dffcbfdb4fc9d017de08c1a537afd83627c" +dependencies = [ + "async-stream", + "async-trait", + "base64 0.13.0", + "bytes", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "hyper", + "hyper-timeout", + "percent-encoding", + "pin-project", + "prost", + "prost-derive", + "tokio", + "tokio-stream", + "tokio-util", + "tower", + "tower-layer", + "tower-service", + "tracing", + "tracing-futures", +] + +[[package]] +name = "tonic-build" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12b52d07035516c2b74337d2ac7746075e7dcae7643816c1b12c5ff8a7484c08" +dependencies = [ + "proc-macro2 1.0.33", + "prost-build", + "quote 1.0.10", + "syn 1.0.82", +] + +[[package]] +name = "tower" +version = "0.4.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5651b5f6860a99bd1adb59dbfe1db8beb433e73709d9032b413a77e2fb7c066a" +dependencies = [ + "futures-core", + "futures-util", + "indexmap", + "pin-project", + "pin-project-lite", + "rand 0.8.4", + "slab", + "tokio", + "tokio-stream", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "343bc9466d3fe6b0f960ef45960509f84480bf4fd96f92901afe7ff3df9d3a62" + [[package]] name = "tower-service" version = "0.3.1" @@ -4009,6 +4250,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "375a639232caf30edfc78e8d89b2d4c375515393e7af7e16f01cd96917fb2105" dependencies = [ "cfg-if", + "log", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -4078,6 +4320,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ffbf13a0f8b054a4e59df3a173b818e9c6177c02789871f2073977fd0062076" +dependencies = [ + "opentelemetry", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", +] + [[package]] name = "tracing-subscriber" version = "0.3.3" @@ -4406,6 +4661,17 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d8b77fdfd5a253be4ab714e4ffa3c49caf146b4de743e97510c0656cf90f1e8e" +[[package]] +name = "which" +version = "4.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea187a8ef279bc014ec368c27a920da2024d2a711109bfbe3440585d5cf27ad9" +dependencies = [ + "either", + "lazy_static", + "libc", +] + [[package]] name = "winapi" version = "0.3.9" diff --git a/Cargo.toml b/Cargo.toml index 339c8265d..917da5b9a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -62,12 +62,15 @@ tokio = { version = "1.12.0", features = ["sync"] } anyhow = "1.0.44" reqwest = { version = "0.11.4", features = ["json"] } reqwest-middleware = "0.1.2" -reqwest-tracing = "0.2.0" +reqwest-tracing = { version = "0.2.0", features = ["opentelemetry_0_16"] } activitystreams = "0.7.0-alpha.11" actix-rt = { version = "2.2.0", default-features = false } serde_json = { version = "1.0.68", features = ["preserve_order"] } clokwerk = "0.3.5" doku = "0.10.1" +opentelemetry = { version = "0.16", features = ["rt-tokio"] } +opentelemetry-otlp = "0.9" +tracing-opentelemetry = "0.16" [dev-dependencies.cargo-husky] version = "1.5.0" diff --git a/config/defaults.hjson b/config/defaults.hjson index 14c1dce0b..f561810d6 100644 --- a/config/defaults.hjson +++ b/config/defaults.hjson @@ -113,4 +113,5 @@ actor_name_max_length: 20 # Maximum number of HTTP requests allowed to handle a single incoming activity (or a single object fetch through the search). http_fetch_retry_limit: 25 + opentelemetry_url: "http:#localhost:4317" } diff --git a/crates/apub_lib/src/activity_queue.rs b/crates/apub_lib/src/activity_queue.rs index c02586ce9..7357b2d69 100644 --- a/crates/apub_lib/src/activity_queue.rs +++ b/crates/apub_lib/src/activity_queue.rs @@ -80,12 +80,11 @@ async fn do_send(task: SendActivityTask, client: &ClientWithMiddleware) -> Resul match result { Ok(o) => { if !o.status().is_success() { + let status = o.status(); + let text = o.text().await?; warn!( "Send {} to {} failed with status {}: {}", - task.activity_id, - task.inbox, - o.status(), - o.text().await? + task.activity_id, task.inbox, status, text ); } } diff --git a/crates/utils/src/settings/structs.rs b/crates/utils/src/settings/structs.rs index 6a20386e7..e58db8c85 100644 --- a/crates/utils/src/settings/structs.rs +++ b/crates/utils/src/settings/structs.rs @@ -49,6 +49,10 @@ pub struct Settings { /// Maximum number of HTTP requests allowed to handle a single incoming activity (or a single object fetch through the search). #[default(25)] pub http_fetch_retry_limit: i32, + + #[default(None)] + #[doku(example = "http://localhost:4317")] + pub opentelemetry_url: Option, } #[derive(Debug, Deserialize, Serialize, Clone, SmartDefault, Document)] diff --git a/crates/websocket/Cargo.toml b/crates/websocket/Cargo.toml index 4340e4865..09962089d 100644 --- a/crates/websocket/Cargo.toml +++ b/crates/websocket/Cargo.toml @@ -34,3 +34,5 @@ strum_macros = "0.21.1" chrono = { version = "0.4.19", features = ["serde"] } actix-web = { version = "4.0.0-beta.9", default-features = false, features = ["rustls"] } actix-web-actors = { version = "4.0.0-beta.7", default-features = false } +opentelemetry = "0.16" +tracing-opentelemetry = "0.16" diff --git a/crates/websocket/src/handlers.rs b/crates/websocket/src/handlers.rs index f35e8b31b..b6e236dc7 100644 --- a/crates/websocket/src/handlers.rs +++ b/crates/websocket/src/handlers.rs @@ -62,27 +62,50 @@ impl Handler for ChatServer { } } +fn root_span() -> tracing::Span { + let span = tracing::info_span!( + parent: None, + "Websocket Request", + trace_id = tracing::field::Empty, + ); + { + use opentelemetry::trace::TraceContextExt; + use tracing_opentelemetry::OpenTelemetrySpanExt; + let trace_id = span.context().span().span_context().trace_id().to_hex(); + span.record("trace_id", &tracing::field::display(trace_id)); + } + + span +} + /// Handler for Message message. impl Handler for ChatServer { type Result = ResponseFuture>; fn handle(&mut self, msg: StandardMessage, ctx: &mut Context) -> Self::Result { let fut = self.parse_json_message(msg, ctx); - Box::pin(async move { - match fut.await { - Ok(m) => { - // info!("Message Sent: {}", m); - Ok(m) - } - Err(e) => { - error!("Error during message handling {}", e); - Ok( - e.to_json() - .unwrap_or_else(|_| String::from(r#"{"error":"failed to serialize json"}"#)), - ) + let span = root_span(); + + use tracing::Instrument; + + Box::pin( + async move { + match fut.await { + Ok(m) => { + // info!("Message Sent: {}", m); + Ok(m) + } + Err(e) => { + error!("Error during message handling {}", e); + Ok( + e.to_json() + .unwrap_or_else(|_| String::from(r#"{"error":"failed to serialize json"}"#)), + ) + } } } - }) + .instrument(span), + ) } } diff --git a/src/lib.rs b/src/lib.rs index 62d071666..29b144690 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,14 +5,21 @@ pub mod root_span_builder; pub mod scheduled_tasks; use lemmy_utils::LemmyError; +use opentelemetry::{ + sdk::{propagation::TraceContextPropagator, Resource}, + KeyValue, +}; +use opentelemetry_otlp::WithExportConfig; use tracing::subscriber::set_global_default; use tracing_error::ErrorLayer; use tracing_log::LogTracer; use tracing_subscriber::{layer::SubscriberExt, EnvFilter, Registry}; -pub fn init_tracing() -> Result<(), LemmyError> { +pub fn init_tracing(opentelemetry_url: Option<&str>) -> Result<(), LemmyError> { LogTracer::init()?; + opentelemetry::global::set_text_map_propagator(TraceContextPropagator::new()); + let env_filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("info")); let format_layer = tracing_subscriber::fmt::layer(); @@ -21,7 +28,28 @@ pub fn init_tracing() -> Result<(), LemmyError> { .with(format_layer) .with(ErrorLayer::default()); - set_global_default(subscriber)?; + if let Some(url) = opentelemetry_url { + let tracer = opentelemetry_otlp::new_pipeline() + .tracing() + .with_trace_config( + opentelemetry::sdk::trace::config() + .with_resource(Resource::new(vec![KeyValue::new("service.name", "lemmy")])), + ) + .with_exporter( + opentelemetry_otlp::new_exporter() + .tonic() + .with_endpoint(url), + ) + .install_batch(opentelemetry::runtime::Tokio)?; + + let otel_layer = tracing_opentelemetry::layer().with_tracer(tracer); + + let subscriber = subscriber.with(otel_layer); + + set_global_default(subscriber)?; + } else { + set_global_default(subscriber)?; + } Ok(()) } diff --git a/src/main.rs b/src/main.rs index cf29e7790..033643899 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,10 +49,10 @@ async fn main() -> Result<(), LemmyError> { return Ok(()); } - init_tracing()?; - let settings = Settings::init().expect("Couldn't initialize settings."); + init_tracing(settings.opentelemetry_url.as_deref())?; + // Set up the r2d2 connection pool let db_url = match get_database_url_from_env() { Ok(url) => url,