diff --git a/Cargo.lock b/Cargo.lock index 9658ff6..ef728a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -132,6 +132,39 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b34d609dfbaf33d6889b2b7106d3ca345eacad44200913df5ba02bfd31d2ba9" +[[package]] +name = "async-stream" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd56dd203fef61ac097dd65721a419ddccb106b2d2b70ba60a6b529f03961a51" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.60", +] + +[[package]] +name = "async-trait" +version = "0.1.80" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6fa2087f2753a7da8cc1c0dbfcf89579dd57458e36769de5ac750b4671737ca" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.60", +] + [[package]] name = "atoi" version = "2.0.0" @@ -156,6 +189,51 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1fdabc7756949593fe60f30ec81974b613357de856987752631dea1e3394c80" +[[package]] +name = "axum" +version = "0.6.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf" +dependencies = [ + "async-trait", + "axum-core", + "bitflags 1.3.2", + "bytes", + "futures-util", + "http 0.2.12", + "http-body 0.4.6", + "hyper 0.14.28", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "rustversion", + "serde", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c" +dependencies = [ + "async-trait", + "bytes", + "futures-util", + "http 0.2.12", + "http-body 0.4.6", + "mime", + "rustversion", + "tower-layer", + "tower-service", +] + [[package]] name = "backtrace" version = "0.3.71" @@ -360,6 +438,15 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crossbeam-channel" +version = "0.5.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab3db02a9c5b5121e1e42fbdb1aeb65f5e02624cc58c43f2884c6ccac0b82f95" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-queue" version = "0.3.11" @@ -632,6 +719,37 @@ version = "0.28.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4271d37baee1b8c7e4b708028c57d816cf9d2434acb33a549475f78c181f6253" +[[package]] +name = "glob" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b" + +[[package]] +name = "h2" +version = "0.3.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "81fe527a889e1532da5c525686d96d4c2e74cdd345badf8dfef9f6b39dd5f5e8" +dependencies = [ + "bytes", + "fnv", + "futures-core", + "futures-sink", + "futures-util", + "http 0.2.12", + "indexmap 2.2.6", + "slab", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "hashbrown" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" + [[package]] name = "hashbrown" version = "0.14.3" @@ -648,7 +766,7 @@ version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e8094feaf31ff591f651a2664fb9cfd92bba7a60ce3197265e9482ebe753c8f7" dependencies = [ - "hashbrown", + "hashbrown 0.14.3", ] [[package]] @@ -705,6 +823,17 @@ dependencies = [ "windows-sys 0.52.0", ] +[[package]] +name = "http" +version = "0.2.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "601cbb57e577e2f5ef5be8e7b83f0f63994f25aa94d673e54a92d5c516d101f1" +dependencies = [ + "bytes", + "fnv", + "itoa", +] + [[package]] name = "http" version = "1.1.0" @@ -716,6 +845,17 @@ dependencies = [ "itoa", ] +[[package]] +name = "http-body" +version = "0.4.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7ceab25649e9960c0311ea418d17bee82c0dcec1bd053b5f9a66e265a693bed2" +dependencies = [ + "bytes", + "http 0.2.12", + "pin-project-lite", +] + [[package]] name = "http-body" version = "1.0.0" @@ -723,7 +863,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1cac85db508abc24a2e48553ba12a996e87244a0395ce011e62b37158745d643" dependencies = [ "bytes", - "http", + "http 1.1.0", ] [[package]] @@ -734,8 +874,8 @@ checksum = "0475f8b2ac86659c21b64320d5d653f9efe42acd2a4e560073ec61a155a34f1d" dependencies = [ "bytes", "futures-core", - "http", - "http-body", + "http 1.1.0", + "http-body 1.0.0", "pin-project-lite", ] @@ -751,6 +891,30 @@ version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" +[[package]] +name = "hyper" +version = "0.14.28" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf96e135eb83a2a8ddf766e426a841d8ddd7449d5f00d34ea02b41d2f19eef80" +dependencies = [ + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "h2", + "http 0.2.12", + "http-body 0.4.6", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", + "want", +] + [[package]] name = "hyper" version = "1.3.1" @@ -760,8 +924,8 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "http", - "http-body", + "http 1.1.0", + "http-body 1.0.0", "httparse", "httpdate", "itoa", @@ -771,6 +935,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 0.14.28", + "pin-project-lite", + "tokio", + "tokio-io-timeout", +] + [[package]] name = "hyper-util" version = "0.1.3" @@ -780,9 +956,9 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "http", - "http-body", - "hyper", + "http 1.1.0", + "http-body 1.0.0", + "hyper 1.3.1", "pin-project-lite", "socket2", "tokio", @@ -824,6 +1000,16 @@ dependencies = [ "unicode-normalization", ] +[[package]] +name = "indexmap" +version = "1.9.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bd070e393353796e801d209ad339e89596eb4c8d430d18ede6a1cced8fafbd99" +dependencies = [ + "autocfg", + "hashbrown 0.12.3", +] + [[package]] name = "indexmap" version = "2.2.6" @@ -831,7 +1017,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "168fb715dda47215e360912c096649d23d58bf392ac62f73919e831745e40f26" dependencies = [ "equivalent", - "hashbrown", + "hashbrown 0.14.3", ] [[package]] @@ -881,11 +1067,15 @@ dependencies = [ "figment", "futures-util", "http-body-util", - "hyper", + "hyper 1.3.1", "hyper-util", "lavina-core", "mgmt-api", "nonempty", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", "projection-irc", "projection-xmpp", "prometheus", @@ -895,6 +1085,7 @@ dependencies = [ "serde_json", "tokio", "tracing", + "tracing-opentelemetry", "tracing-subscriber", ] @@ -967,6 +1158,12 @@ version = "0.4.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "90ed8c1e510134f979dbc4f070f87d4313098b704861a105fe34231c70a3901c" +[[package]] +name = "matchit" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" + [[package]] name = "md-5" version = "0.10.6" @@ -1120,6 +1317,89 @@ version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92" +[[package]] +name = "opentelemetry" +version = "0.22.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "900d57987be3f2aeb70d385fff9b27fb74c5723cc9a52d904d4f9c807a0667bf" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "once_cell", + "pin-project-lite", + "thiserror", + "urlencoding", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1a016b8d9495c639af2145ac22387dcb88e44118e45320d9238fbf4e7889abcb" +dependencies = [ + "async-trait", + "futures-core", + "http 0.2.12", + "opentelemetry", + "opentelemetry-proto", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", + "prost", + "thiserror", + "tokio", + "tonic", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a8fddc9b68f5b80dae9d6f510b88e02396f006ad48cac349411fbecc80caae4" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", +] + +[[package]] +name = "opentelemetry-semantic-conventions" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9ab5bd6c42fb9349dcf28af2ba9a0667f697f9bdcca045d39f2cec5543e2910" + +[[package]] +name = "opentelemetry_sdk" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e90c7113be649e31e9a0f8b5ee24ed7a16923b322c3c5ab6367469c049d6b7e" +dependencies = [ + "async-trait", + "crossbeam-channel", + "futures-channel", + "futures-executor", + "futures-util", + "glob", + "once_cell", + "opentelemetry", + "ordered-float", + "percent-encoding", + "rand", + "thiserror", + "tokio", + "tokio-stream", +] + +[[package]] +name = "ordered-float" +version = "4.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a76df7075c7d4d01fdcb46c912dd17fba5b60c78ea480b475f2b6ab6f666584e" +dependencies = [ + "num-traits", +] + [[package]] name = "overload" version = "0.1.1" @@ -1346,6 +1626,29 @@ dependencies = [ "thiserror", ] +[[package]] +name = "prost" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0f5d036824e4761737860779c906171497f6d55681139d8312388f8fe398922" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.12.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19de2de2a00075bf566bee3bd4db014b11587e84184d3f7a791bc17f1a8e9e48" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn 2.0.60", +] + [[package]] name = "proto-irc" version = "0.0.2-dev" @@ -1468,10 +1771,10 @@ dependencies = [ "bytes", "futures-core", "futures-util", - "http", - "http-body", + "http 1.1.0", + "http-body 1.0.0", "http-body-util", - "hyper", + "hyper 1.3.1", "hyper-util", "ipnet", "js-sys", @@ -1587,6 +1890,12 @@ dependencies = [ "untrusted", ] +[[package]] +name = "rustversion" +version = "1.0.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "80af6f9131f277a45a3fba6ce8e2258037bb0477a67e610d3c1fe046ab31de47" + [[package]] name = "ryu" version = "1.0.17" @@ -1821,7 +2130,7 @@ dependencies = [ "futures-util", "hashlink", "hex", - "indexmap", + "indexmap 2.2.6", "log", "memchr", "once_cell", @@ -2108,6 +2417,16 @@ dependencies = [ "windows-sys 0.48.0", ] +[[package]] +name = "tokio-io-timeout" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30b74022ada614a1b4834de765f9bb43877f910cc8ce4be40e89042c9223a8bf" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "tokio-macros" version = "2.2.0" @@ -2129,6 +2448,31 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "267ac89e0bec6e691e5813911606935d77c476ff49024f98abcea3e7b15e37af" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5419f34732d9eb6ee4c3578b7989078579b7f039cbbb9ca2c4da015749371e15" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", + "tracing", +] + [[package]] name = "toml" version = "0.8.12" @@ -2156,13 +2500,40 @@ version = "0.22.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fb686a972ccef8537b39eead3968b0e8616cb5040dbb9bba93007c8e07c9215f" dependencies = [ - "indexmap", + "indexmap 2.2.6", "serde", "serde_spanned", "toml_datetime", "winnow", ] +[[package]] +name = "tonic" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76c4eb7a4e9ef9d4763600161f12f5070b92a578e1b634db88a6887844c91a13" +dependencies = [ + "async-stream", + "async-trait", + "axum", + "base64 0.21.7", + "bytes", + "h2", + "http 0.2.12", + "http-body 0.4.6", + "hyper 0.14.28", + "hyper-timeout", + "percent-encoding", + "pin-project", + "prost", + "tokio", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "tower" version = "0.4.13" @@ -2171,9 +2542,13 @@ checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" dependencies = [ "futures-core", "futures-util", + "indexmap 1.9.3", "pin-project", "pin-project-lite", + "rand", + "slab", "tokio", + "tokio-util", "tower-layer", "tower-service", "tracing", @@ -2235,6 +2610,24 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9be14ba1bbe4ab79e9229f7f89fab8d120b865859f10527f31c033e599d2284" +dependencies = [ + "js-sys", + "once_cell", + "opentelemetry", + "opentelemetry_sdk", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-subscriber" version = "0.3.18" @@ -2456,6 +2849,16 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "web-time" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "whoami" version = "1.5.1" diff --git a/Cargo.toml b/Cargo.toml index 0158e42..fa04096 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -59,6 +59,11 @@ projection-irc = { path = "crates/projection-irc" } projection-xmpp = { path = "crates/projection-xmpp" } mgmt-api = { path = "crates/mgmt-api" } clap.workspace = true +opentelemetry = "0.22.0" +opentelemetry-semantic-conventions = "0.14.0" +opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio"] } +opentelemetry-otlp = "0.15.0" +tracing-opentelemetry = "0.23.0" [dev-dependencies] assert_matches.workspace = true diff --git a/crates/lavina-core/src/auth.rs b/crates/lavina-core/src/auth.rs index ccf962d..6e6f4ba 100644 --- a/crates/lavina-core/src/auth.rs +++ b/crates/lavina-core/src/auth.rs @@ -25,6 +25,7 @@ impl<'a> Authenticator<'a> { Self { storage } } + #[tracing::instrument(skip(self, provided_password), name = "Authenticator::authenticate")] pub async fn authenticate(&self, login: &str, provided_password: &str) -> Result { let Some(stored_user) = self.storage.retrieve_user_by_name(login).await? else { return Ok(Verdict::UserNotFound); @@ -46,6 +47,7 @@ impl<'a> Authenticator<'a> { Ok(Verdict::InvalidPassword) } + #[tracing::instrument(skip(self, provided_password), name = "Authenticator::set_password")] pub async fn set_password(&self, login: &str, provided_password: &str) -> Result { let Some(u) = self.storage.retrieve_user_by_name(login).await? else { return Ok(UpdatePasswordResult::UserNotFound); diff --git a/crates/lavina-core/src/player.rs b/crates/lavina-core/src/player.rs index 2f5edb4..6dc65de 100644 --- a/crates/lavina-core/src/player.rs +++ b/crates/lavina-core/src/player.rs @@ -15,6 +15,7 @@ use prometheus::{IntGauge, Registry as MetricsRegistry}; use serde::Serialize; use tokio::sync::mpsc::{channel, Receiver, Sender}; use tokio::sync::RwLock; +use tracing::{Instrument, Span}; use crate::dialog::DialogRegistry; use crate::prelude::*; @@ -59,6 +60,7 @@ pub struct PlayerConnection { } impl PlayerConnection { /// Handled in [Player::send_message]. + #[tracing::instrument(skip(self, body), name = "PlayerConnection::send_message")] pub async fn send_message(&mut self, room_id: RoomId, body: Str) -> Result { let (promise, deferred) = oneshot(); let cmd = ClientCommand::SendMessage { room_id, body, promise }; @@ -67,6 +69,7 @@ impl PlayerConnection { } /// Handled in [Player::join_room]. + #[tracing::instrument(skip(self), name = "PlayerConnection::join_room")] pub async fn join_room(&mut self, room_id: RoomId) -> Result { let (promise, deferred) = oneshot(); let cmd = ClientCommand::JoinRoom { room_id, promise }; @@ -75,6 +78,7 @@ impl PlayerConnection { } /// Handled in [Player::change_topic]. + #[tracing::instrument(skip(self, new_topic), name = "PlayerConnection::change_topic")] pub async fn change_topic(&mut self, room_id: RoomId, new_topic: Str) -> Result<()> { let (promise, deferred) = oneshot(); let cmd = ClientCommand::ChangeTopic { @@ -87,6 +91,7 @@ impl PlayerConnection { } /// Handled in [Player::leave_room]. + #[tracing::instrument(skip(self), name = "PlayerConnection::leave_room")] pub async fn leave_room(&mut self, room_id: RoomId) -> Result<()> { let (promise, deferred) = oneshot(); let cmd = ClientCommand::LeaveRoom { room_id, promise }; @@ -99,6 +104,7 @@ impl PlayerConnection { } /// Handled in [Player::get_rooms]. + #[tracing::instrument(skip(self), name = "PlayerConnection::get_rooms")] pub async fn get_rooms(&self) -> Result> { let (promise, deferred) = oneshot(); let cmd = ClientCommand::GetRooms { promise }; @@ -107,6 +113,7 @@ impl PlayerConnection { } /// Handler in [Player::send_dialog_message]. + #[tracing::instrument(skip(self, body), name = "PlayerConnection::send_dialog_message")] pub async fn send_dialog_message(&self, recipient: PlayerId, body: Str) -> Result<()> { let (promise, deferred) = oneshot(); let cmd = ClientCommand::SendDialogMessage { @@ -122,14 +129,14 @@ impl PlayerConnection { /// Handle to a player actor. #[derive(Clone)] pub struct PlayerHandle { - tx: Sender, + tx: Sender<(ActorCommand, Span)>, } impl PlayerHandle { pub async fn subscribe(&self) -> PlayerConnection { let (sender, receiver) = channel(32); let (promise, deferred) = oneshot(); let cmd = ActorCommand::AddConnection { sender, promise }; - let _ = self.tx.send(cmd).await; + self.send(cmd).await; let connection_id = deferred.await.unwrap(); PlayerConnection { connection_id, @@ -139,8 +146,9 @@ impl PlayerHandle { } async fn send(&self, command: ActorCommand) { + let span = tracing::span!(tracing::Level::INFO, "PlayerHandle::send"); // TODO either handle the error or doc why it is safe to ignore - let _ = self.tx.send(command).await; + let _ = self.tx.send((command, span)).await; } pub async fn update(&self, update: Updates) { @@ -332,7 +340,7 @@ struct Player { connections: AnonTable>, my_rooms: HashMap, banned_from: HashSet, - rx: Receiver, + rx: Receiver<(ActorCommand, Span)>, handle: PlayerHandle, rooms: RoomRegistry, dialogs: DialogRegistry, @@ -379,20 +387,36 @@ impl Player { } } while let Some(cmd) = self.rx.recv().await { - match cmd { - ActorCommand::AddConnection { sender, promise } => { - let connection_id = self.connections.insert(sender); - if let Err(connection_id) = promise.send(ConnectionId(connection_id)) { - log::warn!("Connection {connection_id:?} terminated before finalization"); - self.terminate_connection(connection_id); + let (cmd, span) = cmd; + let should_stop = async { + match cmd { + ActorCommand::AddConnection { sender, promise } => { + let connection_id = self.connections.insert(sender); + if let Err(connection_id) = promise.send(ConnectionId(connection_id)) { + log::warn!("Connection {connection_id:?} terminated before finalization"); + self.terminate_connection(connection_id); + } + false } + ActorCommand::TerminateConnection(connection_id) => { + self.terminate_connection(connection_id); + false + } + ActorCommand::Update(update) => { + self.handle_update(update).await; + false + } + ActorCommand::ClientCommand(cmd, connection_id) => { + self.handle_cmd(cmd, connection_id).await; + false + } + ActorCommand::Stop => true, } - ActorCommand::TerminateConnection(connection_id) => { - self.terminate_connection(connection_id); - } - ActorCommand::Update(update) => self.handle_update(update).await, - ActorCommand::ClientCommand(cmd, connection_id) => self.handle_cmd(cmd, connection_id).await, - ActorCommand::Stop => break, + } + .instrument(span) + .await; + if should_stop { + break; } } log::debug!("Shutting down player actor #{:?}", self.player_id); @@ -400,6 +424,7 @@ impl Player { } /// Handle an incoming update by changing the internal state and broadcasting it to all connections if necessary. + #[tracing::instrument(skip(self, update), name = "Player::handle_update")] async fn handle_update(&mut self, update: Updates) { log::debug!( "Player received an update, broadcasting to {} connections", @@ -461,6 +486,7 @@ impl Player { } } + #[tracing::instrument(skip(self), name = "Player::join_room")] async fn join_room(&mut self, connection_id: ConnectionId, room_id: RoomId) -> JoinResult { if self.banned_from.contains(&room_id) { return JoinResult::Banned; @@ -488,6 +514,7 @@ impl Player { JoinResult::Success(room_info) } + #[tracing::instrument(skip(self), name = "Player::leave_room")] async fn leave_room(&mut self, connection_id: ConnectionId, room_id: RoomId) { let room = self.my_rooms.remove(&room_id); if let Some(room) = room { @@ -501,6 +528,7 @@ impl Player { self.broadcast_update(update, connection_id).await; } + #[tracing::instrument(skip(self, body), name = "Player::send_message")] async fn send_message(&mut self, connection_id: ConnectionId, room_id: RoomId, body: Str) -> SendMessageResult { let Some(room) = self.my_rooms.get(&room_id) else { tracing::info!("no room found"); @@ -518,6 +546,7 @@ impl Player { SendMessageResult::Success(created_at) } + #[tracing::instrument(skip(self, new_topic), name = "Player::change_topic")] async fn change_topic(&mut self, connection_id: ConnectionId, room_id: RoomId, new_topic: Str) { let Some(room) = self.my_rooms.get(&room_id) else { tracing::info!("no room found"); @@ -528,6 +557,7 @@ impl Player { self.broadcast_update(update, connection_id).await; } + #[tracing::instrument(skip(self), name = "Player::get_rooms")] async fn get_rooms(&self) -> Vec { let mut response = vec![]; for (_, handle) in &self.my_rooms { @@ -536,6 +566,7 @@ impl Player { response } + #[tracing::instrument(skip(self, body), name = "Player::send_dialog_message")] async fn send_dialog_message(&self, connection_id: ConnectionId, recipient: PlayerId, body: Str) { let created_at = chrono::Utc::now(); self.dialogs.send_message(self.player_id.clone(), recipient.clone(), body.clone(), &created_at).await.unwrap(); @@ -552,6 +583,7 @@ impl Player { /// /// This is called after handling a client command. /// Sending the update to the connection which sent the command is handled by the connection itself. + #[tracing::instrument(skip(self, update), name = "Player::broadcast_update")] async fn broadcast_update(&self, update: Updates, except: ConnectionId) { for (a, b) in &self.connections { if ConnectionId(a) == except { diff --git a/crates/lavina-core/src/repo/auth.rs b/crates/lavina-core/src/repo/auth.rs index f7c0d69..ae67df5 100644 --- a/crates/lavina-core/src/repo/auth.rs +++ b/crates/lavina-core/src/repo/auth.rs @@ -3,6 +3,7 @@ use anyhow::Result; use crate::repo::Storage; impl Storage { + #[tracing::instrument(skip(self), name = "Storage::set_argon2_challenge")] pub async fn set_argon2_challenge(&self, user_id: u32, hash: &str) -> Result<()> { let mut executor = self.conn.lock().await; sqlx::query( diff --git a/crates/lavina-core/src/repo/mod.rs b/crates/lavina-core/src/repo/mod.rs index 9c7aff6..f0d210c 100644 --- a/crates/lavina-core/src/repo/mod.rs +++ b/crates/lavina-core/src/repo/mod.rs @@ -40,6 +40,7 @@ impl Storage { Ok(Storage { conn }) } + #[tracing::instrument(skip(self), name = "Storage::retrieve_user_by_name")] pub async fn retrieve_user_by_name(&self, name: &str) -> Result> { let mut executor = self.conn.lock().await; let res = sqlx::query_as( @@ -55,6 +56,7 @@ impl Storage { Ok(res) } + #[tracing::instrument(skip(self), name = "Storage::retrieve_room_by_name")] pub async fn retrieve_room_by_name(&self, name: &str) -> Result> { let mut executor = self.conn.lock().await; let res = sqlx::query_as( @@ -69,6 +71,7 @@ impl Storage { Ok(res) } + #[tracing::instrument(skip(self, topic), name = "Storage::create_new_room")] pub async fn create_new_room(&mut self, name: &str, topic: &str) -> Result { let mut executor = self.conn.lock().await; let (id,): (u32,) = sqlx::query_as( @@ -84,6 +87,7 @@ impl Storage { Ok(id) } + #[tracing::instrument(skip(self, content, created_at), name = "Storage::insert_message")] pub async fn insert_message( &mut self, room_id: u32, @@ -127,6 +131,7 @@ impl Storage { Ok(()) } + #[tracing::instrument(skip(self), name = "Storage::create_user")] pub async fn create_user(&mut self, name: &str) -> Result<()> { let query = sqlx::query( "insert into users(name) @@ -139,6 +144,7 @@ impl Storage { Ok(()) } + #[tracing::instrument(skip(self, pwd), name = "Storage::set_password")] pub async fn set_password<'a>(&'a self, name: &'a str, pwd: &'a str) -> Result> { async fn inner(txn: &mut Transaction<'_, Sqlite>, name: &str, pwd: &str) -> Result> { let id: Option<(u32,)> = sqlx::query_as("select * from users where name = ? limit 1;") diff --git a/crates/lavina-core/src/repo/room.rs b/crates/lavina-core/src/repo/room.rs index 96b89f2..38de47d 100644 --- a/crates/lavina-core/src/repo/room.rs +++ b/crates/lavina-core/src/repo/room.rs @@ -3,6 +3,7 @@ use anyhow::Result; use crate::repo::Storage; impl Storage { + #[tracing::instrument(skip(self), name = "Storage::add_room_member")] pub async fn add_room_member(&self, room_id: u32, player_id: u32) -> Result<()> { let mut executor = self.conn.lock().await; sqlx::query( @@ -17,6 +18,7 @@ impl Storage { Ok(()) } + #[tracing::instrument(skip(self), name = "Storage::remove_room_member")] pub async fn remove_room_member(&self, room_id: u32, player_id: u32) -> Result<()> { let mut executor = self.conn.lock().await; sqlx::query( @@ -31,6 +33,7 @@ impl Storage { Ok(()) } + #[tracing::instrument(skip(self, topic), name = "Storage::set_room_topic")] pub async fn set_room_topic(&mut self, id: u32, topic: &str) -> Result<()> { let mut executor = self.conn.lock().await; sqlx::query( diff --git a/crates/lavina-core/src/room.rs b/crates/lavina-core/src/room.rs index d50e169..17a463b 100644 --- a/crates/lavina-core/src/room.rs +++ b/crates/lavina-core/src/room.rs @@ -59,6 +59,7 @@ impl RoomRegistry { Ok(()) } + #[tracing::instrument(skip(self), name = "RoomRegistry::get_or_create_room")] pub async fn get_or_create_room(&mut self, room_id: RoomId) -> Result { let mut inner = self.0.write().await; if let Some(room_handle) = inner.get_or_load_room(&room_id).await? { @@ -83,11 +84,13 @@ impl RoomRegistry { } } + #[tracing::instrument(skip(self), name = "RoomRegistry::get_room")] pub async fn get_room(&self, room_id: &RoomId) -> Option { let mut inner = self.0.write().await; inner.get_or_load_room(room_id).await.unwrap() } + #[tracing::instrument(skip(self), name = "RoomRegistry::get_all_rooms")] pub async fn get_all_rooms(&self) -> Vec { let handles = { let inner = self.0.read().await; @@ -109,6 +112,7 @@ struct RoomRegistryInner { } impl RoomRegistryInner { + #[tracing::instrument(skip(self), name = "RoomRegistryInner::get_or_load_room")] async fn get_or_load_room(&mut self, room_id: &RoomId) -> Result> { if let Some(room_handle) = self.rooms.get(room_id) { log::debug!("Room {} was loaded already", &room_id.0); @@ -138,12 +142,14 @@ impl RoomRegistryInner { #[derive(Clone)] pub struct RoomHandle(Arc>); impl RoomHandle { + #[tracing::instrument(skip(self, player_handle), name = "RoomHandle::subscribe")] pub async fn subscribe(&self, player_id: &PlayerId, player_handle: PlayerHandle) { let mut lock = self.0.write().await; tracing::info!("Adding a subscriber to a room"); lock.subscriptions.insert(player_id.clone(), player_handle); } + #[tracing::instrument(skip(self), name = "RoomHandle::add_member")] pub async fn add_member(&self, player_id: &PlayerId, player_storage_id: u32) { let mut lock = self.0.write().await; tracing::info!("Adding a new member to a room"); @@ -157,11 +163,13 @@ impl RoomHandle { lock.broadcast_update(update, player_id).await; } + #[tracing::instrument(skip(self), name = "RoomHandle::unsubscribe")] pub async fn unsubscribe(&self, player_id: &PlayerId) { let mut lock = self.0.write().await; lock.subscriptions.remove(player_id); } + #[tracing::instrument(skip(self), name = "RoomHandle::remove_member")] pub async fn remove_member(&self, player_id: &PlayerId, player_storage_id: u32) { let mut lock = self.0.write().await; tracing::info!("Removing a member from a room"); @@ -175,6 +183,7 @@ impl RoomHandle { lock.broadcast_update(update, player_id).await; } + #[tracing::instrument(skip(self, body, created_at), name = "RoomHandle::send_message")] pub async fn send_message(&self, player_id: &PlayerId, body: Str, created_at: DateTime) { let mut lock = self.0.write().await; let res = lock.send_message(player_id, body, created_at).await; @@ -183,6 +192,7 @@ impl RoomHandle { } } + #[tracing::instrument(skip(self), name = "RoomHandle::get_room_info")] pub async fn get_room_info(&self) -> RoomInfo { let lock = self.0.read().await; RoomInfo { @@ -192,6 +202,7 @@ impl RoomHandle { } } + #[tracing::instrument(skip(self, new_topic), name = "RoomHandle::set_topic")] pub async fn set_topic(&self, changer_id: &PlayerId, new_topic: Str) { let mut lock = self.0.write().await; let storage_id = lock.storage_id; @@ -220,6 +231,7 @@ struct Room { storage: Storage, } impl Room { + #[tracing::instrument(skip(self, body, created_at), name = "Room::send_message")] async fn send_message(&mut self, author_id: &PlayerId, body: Str, created_at: DateTime) -> Result<()> { tracing::info!("Adding a message to room"); self.storage @@ -246,6 +258,7 @@ impl Room { /// /// This is called after handling a client command. /// Sending the update to the player who sent the command is handled by the player actor. + #[tracing::instrument(skip(self, update), name = "Room::broadcast_update")] async fn broadcast_update(&self, update: Updates, except: &PlayerId) { tracing::debug!("Broadcasting an update to {} subs", self.subscriptions.len()); for (player_id, sub) in &self.subscriptions { diff --git a/crates/projection-irc/src/lib.rs b/crates/projection-irc/src/lib.rs index 342682a..a320c5d 100644 --- a/crates/projection-irc/src/lib.rs +++ b/crates/projection-irc/src/lib.rs @@ -678,6 +678,7 @@ enum HandleResult { Leave, } +#[tracing::instrument(skip_all, name = "handle_incoming_message")] async fn handle_incoming_message( buffer: &str, config: &ServerConfig, diff --git a/docs/running.md b/docs/running.md index ad422a1..74b4c60 100644 --- a/docs/running.md +++ b/docs/running.md @@ -23,6 +23,11 @@ hostname = "localhost" [storage] db_path = "db.sqlite" + +[tracing] +# otlp grpc endpoint +endpoint = "http://jaeger:4317" +service_name = "lavina" ``` ## With Docker Compose @@ -41,6 +46,15 @@ services: - '5222:5222' # xmpp - '6667:6667' # irc non-tls - '127.0.0.1:1380:8080' # management http (private) + # if you want to observe traces + jaeger: + image: "jaegertracing/all-in-one:1.56" + ports: + - "16686:16686" # web ui + - "4317:4317" # grpc ingest endpoint + environment: + - COLLECTOR_OTLP_ENABLED=true + - SPAN_STORAGE_TYPE=memory ``` ## With Cargo diff --git a/src/main.rs b/src/main.rs index 98c45f8..d72055b 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,8 +6,17 @@ use std::path::Path; use clap::Parser; use figment::providers::Format; use figment::{providers::Toml, Figment}; +use opentelemetry::KeyValue; +use opentelemetry_otlp::WithExportConfig; +use opentelemetry_sdk::trace::{BatchConfig, RandomIdGenerator, Sampler}; +use opentelemetry_sdk::{runtime, Resource}; +use opentelemetry_semantic_conventions::resource::SERVICE_NAME; +use opentelemetry_semantic_conventions::SCHEMA_URL; use prometheus::Registry as MetricsRegistry; use serde::Deserialize; +use tracing_opentelemetry::OpenTelemetryLayer; +use tracing_subscriber::fmt::Subscriber; +use tracing_subscriber::prelude::*; use lavina_core::prelude::*; use lavina_core::repo::Storage; @@ -19,6 +28,13 @@ struct ServerConfig { irc: projection_irc::ServerConfig, xmpp: projection_xmpp::ServerConfig, storage: lavina_core::repo::StorageConfig, + tracing: Option, +} + +#[derive(Deserialize, Debug)] +struct TracingConfig { + endpoint: String, + service_name: String, } #[derive(Parser)] @@ -36,9 +52,9 @@ fn load_config() -> Result { #[tokio::main] async fn main() -> Result<()> { - set_up_logging()?; let sleep = ctrl_c()?; let config = load_config()?; + set_up_logging(&config.tracing)?; tracing::info!("Booting up"); tracing::info!("Loaded config: {config:?}"); @@ -47,6 +63,7 @@ async fn main() -> Result<()> { irc: irc_config, xmpp: xmpp_config, storage: storage_config, + tracing: _, } = config; let metrics = MetricsRegistry::new(); let storage = Storage::open(storage_config).await?; @@ -87,7 +104,44 @@ fn ctrl_c() -> Result> { Ok(recv(chan)) } -fn set_up_logging() -> Result<()> { - tracing_subscriber::fmt::init(); +fn set_up_logging(tracing_config: &Option) -> Result<()> { + let subscriber = tracing_subscriber::registry().with(tracing_subscriber::fmt::layer()); + + let targets = { + use std::{env, str::FromStr}; + use tracing_subscriber::{filter::Targets, layer::SubscriberExt}; + match env::var("RUST_LOG") { + Ok(var) => Targets::from_str(&var) + .map_err(|e| { + eprintln!("Ignoring `RUST_LOG={:?}`: {}", var, e); + }) + .unwrap_or_default(), + Err(env::VarError::NotPresent) => Targets::new().with_default(Subscriber::DEFAULT_MAX_LEVEL), + Err(e) => { + eprintln!("Ignoring `RUST_LOG`: {}", e); + Targets::new().with_default(Subscriber::DEFAULT_MAX_LEVEL) + } + } + }; + if let Some(config) = tracing_config { + let trace_config = opentelemetry_sdk::trace::Config::default() + .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(1.0)))) + .with_id_generator(RandomIdGenerator::default()) + .with_resource(Resource::from_schema_url( + [KeyValue::new(SERVICE_NAME, config.service_name.to_string())], + SCHEMA_URL, + )); + let trace_exporter = opentelemetry_otlp::new_exporter().tonic().with_endpoint(&config.endpoint); + let tracer = opentelemetry_otlp::new_pipeline() + .tracing() + .with_trace_config(trace_config) + .with_batch_config(BatchConfig::default()) + .with_exporter(trace_exporter) + .install_batch(runtime::Tokio)?; + let subscriber = subscriber.with(OpenTelemetryLayer::new(tracer)); + targets.with_subscriber(subscriber).try_init()?; + } else { + targets.with_subscriber(subscriber).try_init()?; + } Ok(()) }