From 1afe7c5f98a7341c6f50b97e5fae23368ca37dbb Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 6 Jul 2026 12:50:28 +0200 Subject: [PATCH 1/4] feat: added telemetry config --- aquila.yml | 7 ++++++- src/configuration/config.rs | 21 +++++++++++++++++++++ 2 files changed, 27 insertions(+), 1 deletion(-) diff --git a/aquila.yml b/aquila.yml index 0e05d3d..9123ab1 100644 --- a/aquila.yml +++ b/aquila.yml @@ -6,9 +6,14 @@ environment: development # Valid values: static, dynamic. mode: static -# Default env_logger filter. RUST_LOG takes precedence when it is set. +# Default tracing filter. RUST_LOG takes precedence when it is set. log_level: debug +# OTLP export for logs, traces, and the deliberately small metric set. +telemetry: + enabled: false + endpoint: http://localhost:4317 + # NATS server and JetStream key-value bucket used for flow storage. nats: url: nats://localhost:4222 diff --git a/src/configuration/config.rs b/src/configuration/config.rs index ccfa4fe..ce6b654 100644 --- a/src/configuration/config.rs +++ b/src/configuration/config.rs @@ -14,6 +14,7 @@ pub struct Config { pub environment: Environment, pub mode: Mode, pub log_level: String, + pub telemetry: Telemetry, pub nats: Nats, pub static_config: StaticConfig, pub dynamic_config: DynamicConfig, @@ -28,6 +29,13 @@ pub struct Nats { pub bucket: String, } +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(default)] +pub struct Telemetry { + pub enabled: bool, + pub endpoint: String, +} + #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(default)] pub struct StaticConfig { @@ -78,6 +86,7 @@ impl Default for Config { environment: Environment::Development, mode: Mode::Static, log_level: "debug".into(), + telemetry: Telemetry::default(), nats: Nats::default(), static_config: StaticConfig::default(), dynamic_config: DynamicConfig::default(), @@ -87,6 +96,15 @@ impl Default for Config { } } +impl Default for Telemetry { + fn default() -> Self { + Self { + enabled: false, + endpoint: "http://localhost:4317".into(), + } + } +} + impl Default for Nats { fn default() -> Self { Self { @@ -175,6 +193,9 @@ impl fmt::Display for Config { writeln!(formatter, " Environment: {}", self.environment)?; writeln!(formatter, " Mode: {}", self.mode)?; writeln!(formatter, " Log level: {}", self.log_level)?; + writeln!(formatter, " OpenTelemetry")?; + writeln!(formatter, " Enabled: {}", self.telemetry.enabled)?; + writeln!(formatter, " Endpoint: {}", self.telemetry.endpoint)?; writeln!(formatter, " NATS")?; writeln!(formatter, " URL: {}", self.nats.url)?; writeln!(formatter, " Bucket: {}", self.nats.bucket)?; From 21602ec7b1bf01e3570ec0231295f0796f391067 Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 6 Jul 2026 12:50:42 +0200 Subject: [PATCH 2/4] deps: added open telementry --- Cargo.lock | 471 ++++++++++++++++++++++++++++++++++++++--------------- Cargo.toml | 8 +- 2 files changed, 348 insertions(+), 131 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 696bfca..3caaede 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,56 +11,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "anstream" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" -dependencies = [ - "anstyle", - "anstyle-parse", - "anstyle-query", - "anstyle-wincon", - "colorchoice", - "is_terminal_polyfill", - "utf8parse", -] - -[[package]] -name = "anstyle" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" - -[[package]] -name = "anstyle-parse" -version = "1.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" -dependencies = [ - "utf8parse", -] - -[[package]] -name = "anstyle-query" -version = "1.1.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" -dependencies = [ - "windows-sys 0.61.2", -] - -[[package]] -name = "anstyle-wincon" -version = "3.0.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" -dependencies = [ - "anstyle", - "once_cell_polyfill", - "windows-sys 0.61.2", -] - [[package]] name = "anyhow" version = "1.0.102" @@ -74,10 +24,13 @@ dependencies = [ "async-nats", "code0-flow", "config", - "env_logger", "futures", "futures-core", "log", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry_sdk", "prost", "serde", "serde_json", @@ -85,6 +38,9 @@ dependencies = [ "tokio-stream", "tonic", "tonic-health", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", "tucana", "uuid", ] @@ -299,12 +255,6 @@ dependencies = [ "walkdir", ] -[[package]] -name = "colorchoice" -version = "1.0.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" - [[package]] name = "config" version = "0.15.25" @@ -536,29 +486,6 @@ dependencies = [ "cfg-if", ] -[[package]] -name = "env_filter" -version = "2.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "900d271a03799a1ee8d1ca9b19893b48ca674a9284fefcfb85f05e74ed314217" -dependencies = [ - "log", - "regex", -] - -[[package]] -name = "env_logger" -version = "0.11.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de671bd27a75a797dc9ae289ba1e77276e75e2026408aab65185384e2d5cd3f6" -dependencies = [ - "anstream", - "anstyle", - "env_filter", - "jiff", - "log", -] - [[package]] name = "equivalent" version = "1.0.2" @@ -746,6 +673,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "libc", + "r-efi 5.3.0", + "wasip2", +] + [[package]] name = "getrandom" version = "0.4.2" @@ -754,7 +693,7 @@ checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", "rand_core 0.10.1", "wasip2", "wasip3", @@ -904,13 +843,16 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", "futures-channel", "futures-util", "http", "http-body", "hyper", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2", "tokio", @@ -1040,10 +982,10 @@ dependencies = [ ] [[package]] -name = "is_terminal_polyfill" -version = "1.70.2" +name = "ipnet" +version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" [[package]] name = "itertools" @@ -1060,36 +1002,14 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" -[[package]] -name = "jiff" -version = "0.2.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a3546dc96b6d42c5f24902af9e2538e82e39ad350b0c766eb3fbf2d8f3d8359" -dependencies = [ - "jiff-static", - "log", - "portable-atomic", - "portable-atomic-util", - "serde_core", -] - -[[package]] -name = "jiff-static" -version = "0.2.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a8c8b344124222efd714b73bb41f8b5120b27a7cc1c75593a6ff768d9d05aa4" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "js-sys" version = "0.3.94" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2e04e2ef80ce82e13552136fabeef8a5ed1f985a96805761cbb9a2c34e7664d9" dependencies = [ + "cfg-if", + "futures-util", "once_cell", "wasm-bindgen", ] @@ -1105,6 +1025,12 @@ dependencies = [ "serde", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + [[package]] name = "leb128fmt" version = "0.1.0" @@ -1135,6 +1061,15 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matchit" version = "0.8.4" @@ -1185,6 +1120,15 @@ dependencies = [ "signatory", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "nuid" version = "0.5.0" @@ -1215,18 +1159,101 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" -[[package]] -name = "once_cell_polyfill" -version = "1.70.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" - [[package]] name = "openssl-probe" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror", + "tracing", +] + +[[package]] +name = "opentelemetry-appender-tracing" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837" +dependencies = [ + "opentelemetry", + "tracing", + "tracing-core", + "tracing-subscriber", +] + +[[package]] +name = "opentelemetry-http" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331" +dependencies = [ + "async-trait", + "bytes", + "http", + "opentelemetry", + "reqwest", +] + +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http", + "opentelemetry", + "opentelemetry-http", + "opentelemetry-proto", + "opentelemetry_sdk", + "prost", + "reqwest", + "thiserror", + "tokio", + "tonic", + "tonic-types", +] + +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry", + "opentelemetry_sdk", + "prost", + "tonic", + "tonic-prost", +] + +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry", + "percent-encoding", + "portable-atomic", + "rand 0.9.4", + "thiserror", + "tokio", + "tokio-stream", +] + [[package]] name = "ordered-multimap" version = "0.7.3" @@ -1390,15 +1417,6 @@ version = "1.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" -[[package]] -name = "portable-atomic-util" -version = "0.2.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "091397be61a01d4be58e7841595bd4bfedb15f1cd54977d79b8271e94ed799a3" -dependencies = [ - "portable-atomic", -] - [[package]] name = "potential_utf" version = "0.1.5" @@ -1524,6 +1542,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" @@ -1537,10 +1561,20 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" dependencies = [ "libc", - "rand_chacha", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + [[package]] name = "rand" version = "0.10.1" @@ -1562,6 +1596,16 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", +] + [[package]] name = "rand_core" version = "0.6.4" @@ -1571,6 +1615,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "rand_core" version = "0.10.1" @@ -1606,6 +1659,37 @@ version = "0.8.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a" +[[package]] +name = "reqwest" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" +dependencies = [ + "base64", + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "ring" version = "0.17.14" @@ -1860,6 +1944,15 @@ dependencies = [ "digest", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -1958,6 +2051,9 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] [[package]] name = "synstructure" @@ -2003,6 +2099,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "time" version = "0.3.47" @@ -2248,6 +2353,17 @@ dependencies = [ "tonic-build", ] +[[package]] +name = "tonic-types" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73ab1b02061f83d519bba3caa167f88f261ef05720ab8ebc954ade70de3348e8" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -2267,6 +2383,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -2285,6 +2419,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -2308,6 +2443,52 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-opentelemetry" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" +dependencies = [ + "js-sys", + "opentelemetry", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -2412,12 +2593,6 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" -[[package]] -name = "utf8parse" -version = "0.2.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" - [[package]] name = "uuid" version = "1.23.4" @@ -2429,6 +2604,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "version_check" version = "0.9.5" @@ -2491,6 +2672,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.67" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03623de6905b7206edd0a75f69f747f134b7f0a2323392d664448bf2d3c5d87e" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.117" @@ -2557,6 +2748,26 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.94" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd70027e39b12f0849461e08ffc50b9cd7688d942c1c8e3c7b22273236b4dd0a" +dependencies = [ + "js-sys", + "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 = "webpki-roots" version = "0.26.11" diff --git a/Cargo.toml b/Cargo.toml index 0a54e7c..482e1ab 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,13 @@ edition = "2024" tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] } futures = "0.3.31" log = "0.4.26" -env_logger = "0.11.8" +opentelemetry = { version = "0.32.0", features = ["logs", "metrics", "trace"] } +opentelemetry-appender-tracing = "0.32.0" +opentelemetry-otlp = { version = "0.32.0", features = ["grpc-tonic", "logs", "metrics", "trace"] } +opentelemetry_sdk = { version = "0.32.1", features = ["logs", "metrics", "rt-tokio", "trace"] } +tracing = { version = "0.1.41", features = ["log"] } +tracing-opentelemetry = "0.33.0" +tracing-subscriber = { version = "0.3.20", features = ["env-filter", "fmt"] } prost = "0.14.1" tonic = "0.14.1" tucana = { version = "0.0.75", features = ["all"] } From e94f8efd9ad637c2146daa6e5253ddcc93f67d90 Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 6 Jul 2026 15:57:22 +0200 Subject: [PATCH 3/4] feat: implemented open telemetry --- src/main.rs | 37 ++- src/sagittarius/flow_service_client_impl.rs | 47 +++- src/sagittarius/module_service_client_impl.rs | 23 +- .../runtime_status_service_client_impl.rs | 18 +- .../action_transfer_service_server_impl.rs | 261 ++++++++++++++---- src/server/module_service_server_impl.rs | 5 + .../runtime_execution_service_server_impl.rs | 5 + .../runtime_status_service_server_impl.rs | 5 + src/startup/dynamic_mode.rs | 18 +- src/startup/mod.rs | 14 +- src/startup/static_mode.rs | 11 +- src/telemetry/errors.rs | 145 ++++++++++ src/telemetry/metrics.rs | 151 ++++++++++ src/telemetry/mod.rs | 134 +++++++++ 14 files changed, 770 insertions(+), 104 deletions(-) create mode 100644 src/telemetry/errors.rs create mode 100644 src/telemetry/metrics.rs create mode 100644 src/telemetry/mod.rs diff --git a/src/main.rs b/src/main.rs index d6356d9..0cfa414 100644 --- a/src/main.rs +++ b/src/main.rs @@ -9,6 +9,7 @@ pub mod flow; pub mod sagittarius; pub mod server; pub mod startup; +pub mod telemetry; const CONFIG_PATH_ENV: &str = "AQUILA_CONFIG_PATH"; const SERVICE_CONFIG_PATH_ENV: &str = "AQUILA_SERVICE_CONFIG_PATH"; @@ -26,12 +27,21 @@ async fn main() { .as_ref() .map(|config| config.log_level.as_str()) .unwrap_or("debug"); - env_logger::Builder::from_env(env_logger::Env::default().default_filter_or(log_level)).init(); + let telemetry_config = config_result + .as_ref() + .map(|config| config.telemetry.clone()) + .unwrap_or_default(); + let environment = config_result + .as_ref() + .map(|config| config.environment.to_string()) + .unwrap_or_else(|_| "unknown".into()); + let telemetry = telemetry::Telemetry::initialize(&telemetry_config, &environment, log_level) + .unwrap_or_else(|error| panic!("failed to initialize telemetry: {error}")); install_panic_logging(); let config = config_result .unwrap_or_else(|error| panic!("failed to load Aquila configuration: {error}")); - log::info!("Starting Aquila"); + log::info!("Starting Aquila runtime gateway"); let app_readiness = AppReadiness::new(); let service_config = std::env::var_os(SERVICE_CONFIG_PATH_ENV) @@ -40,6 +50,7 @@ async fn main() { log::debug!("{config}"); startup::run(config, app_readiness, service_config).await; + telemetry.shutdown(); } fn install_panic_logging() { @@ -52,15 +63,17 @@ fn install_panic_logging() { "" }; - match panic_info.location() { - Some(location) => log::error!( - "Process panic message={} file={} line={} column={}", - message, - location.file(), - location.line(), - location.column() - ), - None => log::error!("Process panic message={} location=unknown", message), - } + let location = panic_info + .location() + .map(|location| { + format!( + "{}:{}:{}", + location.file(), + location.line(), + location.column() + ) + }) + .unwrap_or_else(|| "unknown".into()); + telemetry::errors::panic(message, &location); })); } diff --git a/src/sagittarius/flow_service_client_impl.rs b/src/sagittarius/flow_service_client_impl.rs index 2ad64f7..8fbcbc3 100644 --- a/src/sagittarius/flow_service_client_impl.rs +++ b/src/sagittarius/flow_service_client_impl.rs @@ -1,4 +1,7 @@ -use crate::{authorization::authorization::get_authorization_metadata, flow::get_flow_identifier}; +use crate::{ + authorization::authorization::get_authorization_metadata, flow::get_flow_identifier, + telemetry::metrics, +}; use futures::{StreamExt, TryStreamExt}; use prost::Message; use std::{path::Path, sync::Arc}; @@ -181,8 +184,10 @@ impl SagittariusFlowClient { } if deleted_count == 0 { + metrics::flow_operation("delete", "not_found", 1); log::warn!("Flow deletion matched no stored keys flow_id={}", id); } else { + metrics::flow_operation("delete", "success", 1); log::info!( "Flow deleted successfully id={} deleted_keys={}", id, @@ -195,13 +200,19 @@ impl SagittariusFlowClient { let flow_id = flow.flow_id.clone(); let bytes = flow.encode_to_vec(); match self.store.put(key.clone(), bytes.into()).await { - Ok(_) => log::info!("Stored flow update flow_id={} key={}", flow_id, key), - Err(err) => log::error!( - "Failed to store flow update flow_id={} key={} error={:?}", - flow_id, - key, - err - ), + Ok(_) => { + metrics::flow_operation("update", "success", 1); + log::info!("Stored flow update flow_id={} key={}", flow_id, key) + } + Err(err) => { + metrics::flow_operation("update", "failure", 1); + log::error!( + "Failed to store flow update flow_id={} key={} error={:?}", + flow_id, + key, + err + ) + } }; } Data::Flows(flows) => { @@ -256,6 +267,12 @@ impl SagittariusFlowClient { purged_count, stored_count ); + metrics::flow_operation("replace", "success", stored_count as u64); + metrics::flow_operation( + "replace", + "failure", + received_count.saturating_sub(stored_count) as u64, + ); } Data::ModuleConfigurations(action_configurations) => { let (project_count, config_count) = module_config_stats(&action_configurations); @@ -290,13 +307,16 @@ impl SagittariusFlowClient { let response = match self.client.update(request).await { Ok(res) => { - log::info!("Successfully established a Stream (for Flows)"); + log::info!("Sagittarius flow synchronization stream established"); self.sagittarius_ready.store(true, Ordering::SeqCst); res } Err(status) => { self.sagittarius_ready.store(false, Ordering::SeqCst); - log::warn!("Failed to establish Flow stream: {:?}", status); + log::warn!( + "Sagittarius flow synchronization stream connection failed status={:?}", + status + ); return Err(status); } }; @@ -310,7 +330,10 @@ impl SagittariusFlowClient { } Err(status) => { self.sagittarius_ready.store(false, Ordering::SeqCst); - log::warn!("Flow stream error (will reconnect): {:?}", status); + log::warn!( + "Sagittarius flow synchronization stream failed; reconnecting status={:?}", + status + ); return Err(status); } }; @@ -318,7 +341,7 @@ impl SagittariusFlowClient { // Stream ended without an explicit error self.sagittarius_ready.store(false, Ordering::SeqCst); - log::warn!("Flow stream ended (server closed). Will reconnect."); + log::warn!("Sagittarius closed the flow synchronization stream; reconnecting"); Err(tonic::Status::unavailable("flow stream ended")) } } diff --git a/src/sagittarius/module_service_client_impl.rs b/src/sagittarius/module_service_client_impl.rs index 4419ad4..4bbbab2 100644 --- a/src/sagittarius/module_service_client_impl.rs +++ b/src/sagittarius/module_service_client_impl.rs @@ -1,4 +1,4 @@ -use crate::authorization::authorization::get_authorization_metadata; +use crate::{authorization::authorization::get_authorization_metadata, telemetry::errors}; use std::time::Duration; use tonic::transport::Channel; use tonic::{Extensions, Request}; @@ -20,6 +20,11 @@ impl SagittariusModuleServiceClient { } } + #[tracing::instrument( + name = "sagittarius.module.update", + skip_all, + fields(rpc.system = "grpc", rpc.service = "ModuleService", rpc.method = "Update") + )] pub async fn update_modules( &mut self, modules_update_request: tucana::aquila::ModuleUpdateRequest, @@ -59,12 +64,16 @@ impl SagittariusModuleServiceClient { } } Err(err) => { - log::error!( - "Sagittarius module update RPC failed module_count={} code={} message={} timeout_ms={}", - module_count, - err.code(), - err.message(), - self.unary_rpc_timeout.as_millis() + errors::record( + "dependency", + "sagittarius.module.update", + &err, + format!( + "module_count={} code={} timeout_ms={}", + module_count, + err.code(), + self.unary_rpc_timeout.as_millis() + ), ); tucana::aquila::ModuleUpdateResponse { success: false } } diff --git a/src/sagittarius/runtime_status_service_client_impl.rs b/src/sagittarius/runtime_status_service_client_impl.rs index 829d02a..3f2d5c3 100644 --- a/src/sagittarius/runtime_status_service_client_impl.rs +++ b/src/sagittarius/runtime_status_service_client_impl.rs @@ -1,4 +1,4 @@ -use crate::authorization::authorization::get_authorization_metadata; +use crate::{authorization::authorization::get_authorization_metadata, telemetry::errors}; use std::time::Duration; use tonic::{Extensions, Request, transport::Channel}; use tucana::sagittarius::runtime_status_service_client::RuntimeStatusServiceClient; @@ -35,15 +35,19 @@ impl SagittariusRuntimeStatusServiceClient { let response = match self.client.update(request).await { Ok(response) => { - log::info!("Successfully transferred RuntimeStatus.",); + log::info!("Sagittarius accepted the runtime status update"); response.into_inner() } Err(err) => { - log::error!( - "Sagittarius runtime status update RPC failed code={} message={} timeout_ms={}", - err.code(), - err.message(), - self.unary_rpc_timeout.as_millis() + errors::record( + "dependency", + "sagittarius.runtime_status.update", + &err, + format!( + "code={} timeout_ms={}", + err.code(), + self.unary_rpc_timeout.as_millis() + ), ); return tucana::aquila::RuntimeStatusUpdateResponse { success: false }; } diff --git a/src/server/action_transfer_service_server_impl.rs b/src/server/action_transfer_service_server_impl.rs index 6bac2fd..6d1d095 100644 --- a/src/server/action_transfer_service_server_impl.rs +++ b/src/server/action_transfer_service_server_impl.rs @@ -1,15 +1,17 @@ use crate::{ configuration::service::ServiceConfiguration, sagittarius::module_service_client_impl::SagittariusModuleServiceClient, + telemetry::{errors, metrics}, }; use async_nats::{Subject, Subscriber}; use futures::StreamExt; use futures_core::Stream; use prost::Message; -use std::{collections::HashMap, pin::Pin, sync::Arc}; +use std::{collections::HashMap, pin::Pin, sync::Arc, time::Instant}; use tokio::sync::Mutex; use tokio_stream::wrappers::ReceiverStream; use tonic::Status; +use tracing::Instrument; use tucana::{ aquila::{ ActionEvent, ActionExecutionRequest, ActionExecutionResponse, ActionLogon, @@ -25,6 +27,7 @@ type PendingReplies = Arc>>; struct PendingReply { reply_subject: Subject, keys: Vec, + started_at: Instant, } pub struct AquilaActionTransferServiceServer { @@ -57,8 +60,20 @@ impl AquilaActionTransferServiceServer { } #[derive(Debug)] -enum FlowIdentificationError { - KVError, +struct FlowIdentificationError { + source: Box, +} + +impl std::fmt::Display for FlowIdentificationError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("failed to identify flows") + } +} + +impl std::error::Error for FlowIdentificationError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.source.as_ref()) + } } async fn get_flows( @@ -70,8 +85,9 @@ async fn get_flows( let mut keys = match kv.keys().await { Ok(keys) => keys.boxed(), Err(err) => { - log::error!("Failed to get keys: {:?}", err); - return Err(FlowIdentificationError::KVError); + return Err(FlowIdentificationError { + source: Box::new(err), + }); } }; @@ -86,7 +102,12 @@ async fn get_flows( match decoded_flow { Ok(flow) => collector.push(flow), Err(err) => { - log::error!("Failed to decode flow {}: {:?}", key, err); + errors::record( + "flow_storage", + "flow.decode", + &err, + format!("flow.key={key}"), + ); } } } @@ -94,7 +115,12 @@ async fn get_flows( log::debug!("Flow key disappeared while reading: {}", key); } Err(err) => { - log::error!("Failed to fetch flow {}: {:?}", key, err); + errors::record( + "flow_storage", + "flow.fetch", + &err, + format!("flow.key={key}"), + ); } } } @@ -171,6 +197,18 @@ fn subject_execution_identifier(subject: &Subject) -> Option { .map(ToString::to_string) } +fn action_result_outcome(result: &ActionExecutionResponse) -> &'static str { + match result + .node_result + .as_ref() + .and_then(|node_result| node_result.result.as_ref()) + { + Some(tucana::shared::node_execution_result::Result::Success(_)) => "success", + Some(tucana::shared::node_execution_result::Result::Error(_)) => "error", + None => "missing", + } +} + fn pending_reply_keys( request_execution_id: &str, subject_execution_id: Option<&str>, @@ -198,6 +236,7 @@ fn insert_pending_reply( let pending_reply = PendingReply { reply_subject, keys: keys.clone(), + started_at: Instant::now(), }; for key in keys { @@ -272,6 +311,8 @@ async fn handle_logon( { let lock = actions.lock().await; if !lock.has_action(&token.to_string(), &identifier) { + metrics::action_connection(&identifier, "rejected"); + metrics::action_failure(&identifier, "authentication"); log::warn!( "Rejected action logon identifier={} reason=token_not_registered", identifier @@ -293,9 +334,13 @@ async fn handle_logon( .await; if !response.success { - log::error!( - "Rejected action logon identifier={} reason=sagittarius_module_update_failed", - identifier + metrics::action_connection(&identifier, "rejected"); + metrics::action_failure(&identifier, "module_update"); + errors::record_message( + "dependency", + "action.logon", + "Sagittarius rejected the action module update", + format!("action.identifier={identifier}"), ); return Err(Status::internal( "could not update action module via Sagittarius", @@ -308,11 +353,13 @@ async fn handle_logon( let sub = match client.subscribe(format!("action.{}.*", identifier)).await { Ok(s) => s, Err(err) => { - log::error!( - "Failed to subscribe to action execution subject identifier={} subject=action.{}.* error={:?}", - identifier, - identifier, - err + metrics::action_connection(&identifier, "rejected"); + metrics::action_failure(&identifier, "subscription"); + errors::record( + "messaging", + "action.subscribe", + &err, + format!("action.identifier={identifier} subject=action.{identifier}.*"), ); return Err(Status::internal( "could not register action into execution loop", @@ -321,10 +368,13 @@ async fn handle_logon( }; if let Err(err) = client.flush().await { - log::error!( - "Failed to flush action execution subscription identifier={} error={:?}", - identifier, - err + metrics::action_connection(&identifier, "rejected"); + metrics::action_failure(&identifier, "subscription_flush"); + errors::record( + "messaging", + "action.subscribe.flush", + &err, + format!("action.identifier={identifier}"), ); return Err(Status::internal( "could not register action subscription with NATS", @@ -335,8 +385,9 @@ async fn handle_logon( let tx_clone = tx.clone(); let pending_replies_clone = pending_replies.clone(); + let forwarder_identifier = identifier.clone(); tokio::spawn(async move { - forward_nats_to_action(sub, tx_clone, pending_replies_clone).await; + forward_nats_to_action(forwarder_identifier, sub, tx_clone, pending_replies_clone).await; }); if !*cfg_forwarder_started { @@ -373,9 +424,12 @@ fn spawn_cfg_forwarder( }; if tx.send(Ok(resp)).await.is_err() { + metrics::action_config_update(&action_identifier, "failed"); + metrics::action_failure(&action_identifier, "configuration_forward"); log::debug!("Config forwarder channel closed for {}", action_identifier); break; } + metrics::action_config_update(&action_identifier, "success"); } log::debug!("Config forwarder stopped for {}", action_identifier); @@ -384,6 +438,7 @@ fn spawn_cfg_forwarder( /// Looks up matching flows for an event and requests their execution. async fn handle_event( + action_identifier: &str, event: ActionEvent, kv: async_nats::jetstream::kv::Store, client: async_nats::Client, @@ -398,12 +453,14 @@ async fn handle_event( let flows = match get_flows(pattern.clone(), kv).await { Ok(f) => f, Err(err) => { - log::error!( - "Failed to find flows for action event event_type={} project_id={} pattern={} error={:?}", - event.event_type, - event.project_id, - pattern, - err + errors::record( + "flow_storage", + "action.event.find_flows", + &err, + format!( + "action.identifier={} event_type={} project_id={} pattern={}", + action_identifier, event.event_type, event.project_id, pattern + ), ); return; } @@ -432,12 +489,14 @@ async fn handle_event( ); if let Err(err) = client.request(topic.clone(), bytes.into()).await { - log::error!( - "Failed to request execution flow_id={} execution_id={} topic={} error={:?}", - flow_id, - uuid, - topic, - err + errors::record( + "messaging", + "action.event.request_execution", + &err, + format!( + "action.identifier={} flow_id={} execution_id={} topic={}", + action_identifier, flow_id, uuid, topic + ), ); } } @@ -445,11 +504,13 @@ async fn handle_event( /// Publishes execution results back to the original NATS reply subject. async fn handle_result( + action_identifier: &str, execution_result: ActionExecutionResponse, client: async_nats::Client, pending_replies: PendingReplies, ) { let execution_id = execution_result.execution_identifier.clone(); + metrics::action_result(action_identifier, action_result_outcome(&execution_result)); let pending_reply = { let mut pending = pending_replies.lock().await; @@ -457,12 +518,19 @@ async fn handle_result( }; let Some(pending_reply) = pending_reply else { - log::error!( - "No pending NATS reply subject found execution_id={}", - execution_id + metrics::action_failure(action_identifier, "result_unmatched"); + errors::record_message( + "protocol", + "action.result.match", + "No pending NATS reply subject found", + format!("action.identifier={action_identifier} execution_id={execution_id}"), ); return; }; + metrics::action_execution_duration( + action_identifier, + pending_reply.started_at.elapsed().as_secs_f64(), + ); log::debug!( "Publishing execution result execution_id={} reply_subject={}", @@ -475,21 +543,28 @@ async fn handle_result( .publish(pending_reply.reply_subject.clone(), payload.into()) .await { - log::error!( - "Failed to publish action result execution_id={} reply_subject={} error={:?}", - execution_id, - pending_reply.reply_subject, - err + metrics::action_failure(action_identifier, "result_publish"); + errors::record( + "messaging", + "action.result.publish", + &err, + format!( + "action.identifier={} execution_id={} reply_subject={}", + action_identifier, execution_id, pending_reply.reply_subject + ), ); return; } if let Err(err) = client.flush().await { - log::error!( - "Failed to flush action result execution_id={} error={:?}", - execution_id, - err + metrics::action_failure(action_identifier, "result_flush"); + errors::record( + "messaging", + "action.result.flush", + &err, + format!("action.identifier={action_identifier} execution_id={execution_id}"), ); + return; } } @@ -498,6 +573,11 @@ impl ActionTransferService for AquilaActionTransferServiceServer { type TransferStream = Pin> + Send + 'static>>; + #[tracing::instrument( + name = "aquila.action.transfer", + skip_all, + fields(rpc.system = "grpc", rpc.service = "ActionTransferService", rpc.method = "Transfer") + )] async fn transfer( &self, request: tonic::Request>, @@ -520,8 +600,14 @@ impl ActionTransferService for AquilaActionTransferServiceServer { let (tx, rx) = tokio::sync::mpsc::channel::>(32); + let stream_span = tracing::info_span!( + "aquila.action.stream", + action.identifier = tracing::field::Empty + ); tokio::spawn(async move { let mut cfg_forwarder_started = false; + let mut connected_at = None; + let mut connected_identifier = None; log::debug!("Action transfer stream started"); while let Some(next) = stream.next().await { @@ -582,6 +668,12 @@ impl ActionTransferService for AquilaActionTransferServiceServer { }; action_props = Some(accepted); + tracing::Span::current() + .record("action.identifier", identifier.as_str()); + metrics::action_connection(&identifier, "accepted"); + metrics::action_active(&identifier, 1); + connected_at = Some(std::time::Instant::now()); + connected_identifier = Some(identifier); } _ => { log::error!("Action stream protocol violation expected=logon"); @@ -628,7 +720,8 @@ impl ActionTransferService for AquilaActionTransferServiceServer { } tucana::aquila::action_transfer_request::Data::Event(event) => { log::debug!("Received event action={}", identifier); - handle_event(event, kv.clone(), client.clone()).await; + metrics::action_event(&identifier); + handle_event(&identifier, event, kv.clone(), client.clone()).await; } tucana::aquila::action_transfer_request::Data::Result(execution_result) => { log::debug!( @@ -637,14 +730,30 @@ impl ActionTransferService for AquilaActionTransferServiceServer { identifier ); - handle_result(execution_result, client.clone(), pending_replies.clone()) - .await; + handle_result( + &identifier, + execution_result, + client.clone(), + pending_replies.clone(), + ) + .await; } } } + if let Some(identifier) = connected_identifier { + metrics::action_active(&identifier, -1); + metrics::action_connection(&identifier, "closed"); + if let Some(connected_at) = connected_at { + metrics::action_connection_duration( + &identifier, + connected_at.elapsed().as_secs_f64(), + ); + } + } log::debug!("Action transfer stream ended"); - }); + } + .instrument(stream_span)); Ok(tonic::Response::new(Box::pin(ReceiverStream::new(rx)))) } @@ -652,6 +761,7 @@ impl ActionTransferService for AquilaActionTransferServiceServer { /// Forwards NATS execution requests to the connected action via gRPC and stores reply subjects. async fn forward_nats_to_action( + action_identifier: String, mut sub: Subscriber, tx: tokio::sync::mpsc::Sender>, pending_replies: PendingReplies, @@ -662,11 +772,18 @@ async fn forward_nats_to_action( let mut execution = match ActionExecutionRequest::decode(msg.payload.as_ref()) { Ok(req) => req, Err(err) => { - log::error!( - "Invalid action execution request payload subject={} payload_bytes={} error={:?}", - msg.subject, - msg.payload.len(), - err + metrics::action_execution(&action_identifier, "invalid"); + metrics::action_failure(&action_identifier, "execution_decode"); + errors::record( + "protocol", + "action.execution.decode", + &err, + format!( + "action.identifier={} subject={} payload_bytes={}", + action_identifier, + msg.subject, + msg.payload.len() + ), ); continue; } @@ -687,6 +804,8 @@ async fn forward_nats_to_action( let execution_id = execution.execution_identifier.clone(); let Some(reply_subject) = msg.reply.clone() else { + metrics::action_execution(&action_identifier, "invalid"); + metrics::action_failure(&action_identifier, "missing_reply_subject"); log::error!( "Received request without NATS reply subject execution_id={}", execution_id @@ -696,6 +815,8 @@ async fn forward_nats_to_action( let keys = pending_reply_keys(&execution_id, subject_execution_id.as_deref()); if keys.is_empty() { + metrics::action_execution(&action_identifier, "invalid"); + metrics::action_failure(&action_identifier, "missing_execution_identifier"); log::error!( "Cannot store NATS reply subject without execution identifier subject={} reply_subject={}", msg.subject, @@ -729,6 +850,8 @@ async fn forward_nats_to_action( }; if tx.send(Ok(resp)).await.is_err() { + metrics::action_execution(&action_identifier, "forward_failed"); + metrics::action_failure(&action_identifier, "execution_forward"); log::debug!("Execution forwarder channel closed"); // cleanup, since the request can no longer be delivered to the action @@ -737,6 +860,7 @@ async fn forward_nats_to_action( break; } + metrics::action_execution(&action_identifier, "forwarded"); } log::debug!("Execution forwarder stopped"); @@ -854,4 +978,35 @@ mod tests { Some("execution-id".to_string()) ); } + + #[test] + fn action_result_outcome_distinguishes_success_error_and_missing() { + use tucana::shared::{Error, NodeExecutionResult, Value, node_execution_result}; + + let response = |result| ActionExecutionResponse { + execution_identifier: "execution-id".into(), + node_result: Some(NodeExecutionResult { + result, + ..Default::default() + }), + }; + + assert_eq!( + action_result_outcome(&response(Some(node_execution_result::Result::Success( + Value::default() + )))), + "success" + ); + assert_eq!( + action_result_outcome(&response(Some(node_execution_result::Result::Error( + Error::default() + )))), + "error" + ); + assert_eq!(action_result_outcome(&response(None)), "missing"); + assert_eq!( + action_result_outcome(&ActionExecutionResponse::default()), + "missing" + ); + } } diff --git a/src/server/module_service_server_impl.rs b/src/server/module_service_server_impl.rs index 111500c..4b5c766 100644 --- a/src/server/module_service_server_impl.rs +++ b/src/server/module_service_server_impl.rs @@ -26,6 +26,11 @@ impl AquilaModuleServiceServer { #[tonic::async_trait] impl ModuleService for AquilaModuleServiceServer { + #[tracing::instrument( + name = "aquila.module.update", + skip_all, + fields(rpc.system = "grpc", rpc.service = "ModuleService", rpc.method = "Update") + )] async fn update( &self, request: tonic::Request, diff --git a/src/server/runtime_execution_service_server_impl.rs b/src/server/runtime_execution_service_server_impl.rs index 79b9bc2..05b0cc8 100644 --- a/src/server/runtime_execution_service_server_impl.rs +++ b/src/server/runtime_execution_service_server_impl.rs @@ -33,6 +33,11 @@ fn execution_result_status(execution_result: &ExecutionResult) -> &'static str { #[tonic::async_trait] impl ExecutionService for AquilaExecutionServiceServer { + #[tracing::instrument( + name = "aquila.execution.update", + skip_all, + fields(rpc.system = "grpc", rpc.service = "ExecutionService", rpc.method = "Update") + )] async fn update( &self, request: tonic::Request, diff --git a/src/server/runtime_status_service_server_impl.rs b/src/server/runtime_status_service_server_impl.rs index 8a02423..fbcf6f6 100644 --- a/src/server/runtime_status_service_server_impl.rs +++ b/src/server/runtime_status_service_server_impl.rs @@ -236,6 +236,11 @@ fn epoch_seconds_now() -> i64 { #[tonic::async_trait] impl RuntimeStatusService for AquilaRuntimeStatusServiceServer { + #[tracing::instrument( + name = "aquila.runtime_status.update", + skip_all, + fields(rpc.system = "grpc", rpc.service = "RuntimeStatusService", rpc.method = "Update") + )] async fn update( &self, request: tonic::Request, diff --git a/src/startup/dynamic_mode.rs b/src/startup/dynamic_mode.rs index 2ce1601..b936bc6 100644 --- a/src/startup/dynamic_mode.rs +++ b/src/startup/dynamic_mode.rs @@ -12,6 +12,7 @@ use crate::{ }, }, server::dynamic_server::AquilaDynamicServer, + telemetry::errors, }; use std::{sync::Arc, time::Duration}; @@ -54,7 +55,7 @@ pub async fn run( let mut server_task = tokio::spawn(async move { if let Err(err) = server.start().await { - log::error!("gRPC server error: {:?}", err); + errors::record("server", "grpc.serve", &err, "mode=dynamic"); } else { log::info!("gRPC server stopped gracefully"); } @@ -139,10 +140,15 @@ pub async fn run( match flow_client.init_flow_stream().await { Ok(_) => { - log::warn!("Flow stream ended cleanly. Reconnecting..."); + log::warn!( + "Sagittarius flow synchronization stream ended normally; reconnecting" + ); } Err(e) => { - log::warn!("Flow stream dropped: {:?}. Reconnecting...", e); + log::warn!( + "Sagittarius flow synchronization stream dropped; reconnecting error={:?}", + e + ); } } @@ -171,7 +177,7 @@ pub async fn run( match result { Ok(()) => log::warn!("gRPC server task exited unexpectedly; shutting down"), Err(err) if err.is_panic() => {} - Err(err) => log::error!("gRPC server task failed; shutting down error={:?}", err), + Err(err) => errors::record("task", "grpc.task", &err, "mode=dynamic"), } flow_task.abort(); test_execution_task.abort(); @@ -180,7 +186,7 @@ pub async fn run( match result { Ok(()) => log::warn!("Test execution stream task exited unexpectedly; shutting down"), Err(err) if err.is_panic() => {} - Err(err) => log::error!("Test execution stream task failed; shutting down error={:?}", err), + Err(err) => errors::record("task", "execution_stream.task", &err, "mode=dynamic"), } server_task.abort(); flow_task.abort(); @@ -189,7 +195,7 @@ pub async fn run( match result { Ok(()) => log::warn!("Flow stream task exited unexpectedly; shutting down"), Err(err) if err.is_panic() => {} - Err(err) => log::error!("Flow stream task failed; shutting down error={:?}", err), + Err(err) => errors::record("task", "flow_stream.task", &err, "mode=dynamic"), } server_task.abort(); test_execution_task.abort(); diff --git a/src/startup/mod.rs b/src/startup/mod.rs index c07f7b9..25ba8a2 100644 --- a/src/startup/mod.rs +++ b/src/startup/mod.rs @@ -26,7 +26,10 @@ pub async fn run( // Create connection to JetStream let client = match async_nats::connect(config.nats.url.clone()).await { Ok(client) => { - log::info!("Connected to NATS"); + log::info!( + "Aquila messaging dependency is ready dependency=nats url={}", + config.nats.url + ); client } Err(err) => { @@ -56,7 +59,10 @@ pub async fn run( let kv_store = match jet_stream.get_key_value(config.nats.bucket.clone()).await { Ok(kv) => { - log::info!("Opened NATS key-value store bucket={}", config.nats.bucket); + log::info!( + "Aquila flow store is ready backend=nats_jetstream bucket={}", + config.nats.bucket + ); Arc::new(kv) } Err(err) => { @@ -68,11 +74,11 @@ pub async fn run( }; if config.is_static() { - log::info!("Starting with static configuration"); + log::info!("Selected Aquila startup mode mode=static source=local_flow_export"); static_mode::run(config, app_readiness, service_config, client, kv_store).await; return; } - log::info!("Starting with dynamic configuration"); + log::info!("Selected Aquila startup mode mode=dynamic source=sagittarius"); dynamic_mode::run(config, app_readiness, service_config, client, kv_store).await; } diff --git a/src/startup/static_mode.rs b/src/startup/static_mode.rs index a40aae0..b862f4c 100644 --- a/src/startup/static_mode.rs +++ b/src/startup/static_mode.rs @@ -2,6 +2,7 @@ use crate::{ configuration::{config::Config, service::ServiceConfiguration, state::AppReadiness}, flow::get_flow_identifier, server::static_server::AquilaStaticServer, + telemetry::{errors, metrics}, }; use async_nats::Client; use prost::Message; @@ -47,7 +48,7 @@ pub async fn run( let mut server_task = tokio::spawn(async move { if let Err(err) = server.start().await { - log::error!("gRPC server error: {:?}", err); + errors::record("server", "grpc.serve", &err, "mode=static"); } else { log::info!("gRPC server stopped gracefully"); } @@ -69,7 +70,7 @@ pub async fn run( match result { Ok(()) => log::warn!("gRPC server task exited unexpectedly; shutting down"), Err(err) if err.is_panic() => {} - Err(err) => log::error!("gRPC server task failed; shutting down error={:?}", err), + Err(err) => errors::record("task", "grpc.task", &err, "mode=static"), } } _ = tokio::signal::ctrl_c() => { @@ -133,9 +134,13 @@ async fn init_flows_from_json( match flow_store_client.put(key.clone(), bytes.into()).await { Ok(_) => { stored_count += 1; + metrics::flow_operation("load", "success", 1); log::debug!("Stored fallback flow key={}", key); } - Err(err) => log::error!("Failed to store fallback flow key={} error={:?}", key, err), + Err(err) => { + metrics::flow_operation("load", "failure", 1); + log::error!("Failed to store fallback flow key={} error={:?}", key, err) + } }; } diff --git a/src/telemetry/errors.rs b/src/telemetry/errors.rs new file mode 100644 index 0000000..fad85b6 --- /dev/null +++ b/src/telemetry/errors.rs @@ -0,0 +1,145 @@ +use std::{ + backtrace::Backtrace, + error::Error, + sync::atomic::{AtomicBool, Ordering}, +}; + +static CAPTURE_BACKTRACES: AtomicBool = AtomicBool::new(false); + +pub const EXCEPTION_TARGET: &str = "telemetry::exception"; +pub const SUMMARY_TARGET: &str = "telemetry::error_summary"; + +pub fn enable_backtraces() { + CAPTURE_BACKTRACES.store(true, Ordering::Relaxed); +} + +pub fn record( + category: &'static str, + operation: &'static str, + error: &(dyn Error + 'static), + context: impl AsRef, +) { + let message = error.to_string(); + let chain = error_chain(error); + + tracing::error!( + target: SUMMARY_TARGET, + error_category = category, + operation, + error_message = message.as_str(), + error_context = context.as_ref(), + "operation failed" + ); + + record_exception(category, operation, &message, &chain, context.as_ref()); +} + +pub fn record_message( + category: &'static str, + operation: &'static str, + message: impl AsRef, + context: impl AsRef, +) { + let message = message.as_ref(); + tracing::error!( + target: SUMMARY_TARGET, + error_category = category, + operation, + error_message = message, + error_context = context.as_ref(), + "operation failed" + ); + + record_exception(category, operation, message, "", context.as_ref()); +} + +pub fn panic(message: &str, location: &str) { + tracing::error!( + target: SUMMARY_TARGET, + error_category = "panic", + operation = "process", + error_message = message, + error_context = location, + "process panicked" + ); + + record_exception("panic", "process", message, "", location); +} + +fn error_chain(error: &(dyn Error + 'static)) -> String { + let mut messages = Vec::new(); + let mut current = error.source(); + + while let Some(source) = current { + messages.push(source.to_string()); + current = source.source(); + } + + messages.join(": ") +} + +fn record_exception( + category: &'static str, + operation: &'static str, + message: &str, + chain: &str, + context: &str, +) { + if !CAPTURE_BACKTRACES.load(Ordering::Relaxed) { + return; + } + + tracing::error!( + target: EXCEPTION_TARGET, + error_category = category, + operation, + error_message = message, + error_chain = chain, + error_context = context, + exception_stacktrace = %Backtrace::force_capture(), + "exception" + ); +} + +#[cfg(test)] +mod tests { + use std::{error::Error, fmt}; + + use super::error_chain; + + #[derive(Debug)] + struct TestError { + message: &'static str, + source: Option>, + } + + impl fmt::Display for TestError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(self.message) + } + } + + impl Error for TestError { + fn source(&self) -> Option<&(dyn Error + 'static)> { + self.source + .as_deref() + .map(|source| source as &(dyn Error + 'static)) + } + } + + #[test] + fn formats_source_chain_from_nearest_to_root() { + let error = TestError { + message: "outer", + source: Some(Box::new(TestError { + message: "middle", + source: Some(Box::new(TestError { + message: "root", + source: None, + })), + })), + }; + + assert_eq!(error_chain(&error), "middle: root"); + } +} diff --git a/src/telemetry/metrics.rs b/src/telemetry/metrics.rs new file mode 100644 index 0000000..7dd1d39 --- /dev/null +++ b/src/telemetry/metrics.rs @@ -0,0 +1,151 @@ +use std::sync::OnceLock; + +use opentelemetry::{ + KeyValue, + metrics::{Counter, Histogram, UpDownCounter}, +}; + +static METRICS: OnceLock = OnceLock::new(); + +struct Metrics { + flow_operations: Counter, + action_connections: Counter, + active_actions: UpDownCounter, + action_connection_duration: Histogram, + action_events: Counter, + action_executions: Counter, + action_execution_duration: Histogram, + action_results: Counter, + action_config_updates: Counter, + action_failures: Counter, +} + +pub fn initialize() { + let meter = opentelemetry::global::meter(env!("CARGO_PKG_NAME")); + let _ = METRICS.set(Metrics { + flow_operations: meter.u64_counter("aquila.flow.operations").build(), + action_connections: meter.u64_counter("aquila.action.connections").build(), + active_actions: meter.i64_up_down_counter("aquila.action.active").build(), + action_connection_duration: meter + .f64_histogram("aquila.action.connection.duration") + .with_unit("s") + .build(), + action_events: meter.u64_counter("aquila.action.events").build(), + action_executions: meter.u64_counter("aquila.action.executions").build(), + action_execution_duration: meter + .f64_histogram("aquila.action.execution.duration") + .with_unit("s") + .build(), + action_results: meter.u64_counter("aquila.action.results").build(), + action_config_updates: meter + .u64_counter("aquila.action.configuration_updates") + .build(), + action_failures: meter.u64_counter("aquila.action.failures").build(), + }); +} + +pub fn flow_operation(operation: &'static str, outcome: &'static str, count: u64) { + if let Some(metrics) = METRICS.get() { + metrics.flow_operations.add( + count, + &[ + KeyValue::new("operation", operation), + KeyValue::new("outcome", outcome), + ], + ); + } +} + +fn action_attributes(identifier: &str) -> [KeyValue; 1] { + [KeyValue::new("action.identifier", identifier.to_owned())] +} + +pub fn action_connection(identifier: &str, outcome: &'static str) { + if let Some(metrics) = METRICS.get() { + metrics.action_connections.add( + 1, + &[ + KeyValue::new("action.identifier", identifier.to_owned()), + KeyValue::new("outcome", outcome), + ], + ); + } +} + +pub fn action_active(identifier: &str, delta: i64) { + if let Some(metrics) = METRICS.get() { + metrics + .active_actions + .add(delta, &action_attributes(identifier)); + } +} + +pub fn action_connection_duration(identifier: &str, seconds: f64) { + if let Some(metrics) = METRICS.get() { + metrics + .action_connection_duration + .record(seconds, &action_attributes(identifier)); + } +} + +pub fn action_event(identifier: &str) { + if let Some(metrics) = METRICS.get() { + metrics.action_events.add(1, &action_attributes(identifier)); + } +} + +pub fn action_execution(identifier: &str, outcome: &'static str) { + if let Some(metrics) = METRICS.get() { + metrics.action_executions.add( + 1, + &[ + KeyValue::new("action.identifier", identifier.to_owned()), + KeyValue::new("outcome", outcome), + ], + ); + } +} + +pub fn action_result(identifier: &str, outcome: &'static str) { + if let Some(metrics) = METRICS.get() { + metrics.action_results.add( + 1, + &[ + KeyValue::new("action.identifier", identifier.to_owned()), + KeyValue::new("outcome", outcome), + ], + ); + } +} + +pub fn action_execution_duration(identifier: &str, seconds: f64) { + if let Some(metrics) = METRICS.get() { + metrics + .action_execution_duration + .record(seconds, &action_attributes(identifier)); + } +} + +pub fn action_config_update(identifier: &str, outcome: &'static str) { + if let Some(metrics) = METRICS.get() { + metrics.action_config_updates.add( + 1, + &[ + KeyValue::new("action.identifier", identifier.to_owned()), + KeyValue::new("outcome", outcome), + ], + ); + } +} + +pub fn action_failure(identifier: &str, reason: &'static str) { + if let Some(metrics) = METRICS.get() { + metrics.action_failures.add( + 1, + &[ + KeyValue::new("action.identifier", identifier.to_owned()), + KeyValue::new("reason", reason), + ], + ); + } +} diff --git a/src/telemetry/mod.rs b/src/telemetry/mod.rs new file mode 100644 index 0000000..fbfe5ee --- /dev/null +++ b/src/telemetry/mod.rs @@ -0,0 +1,134 @@ +pub mod errors; +pub mod metrics; + +use std::error::Error; + +use opentelemetry::{KeyValue, global, trace::TracerProvider as _}; +use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge; +use opentelemetry_otlp::WithExportConfig; +use opentelemetry_sdk::{ + Resource, + logs::SdkLoggerProvider, + metrics::{PeriodicReader, SdkMeterProvider}, + propagation::TraceContextPropagator, + trace::SdkTracerProvider, +}; +use tracing_subscriber::{ + EnvFilter, Layer, filter::filter_fn, layer::SubscriberExt, util::SubscriberInitExt, +}; + +use crate::configuration::config::Telemetry as TelemetryConfig; + +pub struct Telemetry { + logger_provider: Option, + meter_provider: Option, + tracer_provider: Option, +} + +impl Telemetry { + pub fn initialize( + config: &TelemetryConfig, + environment: &str, + default_log_level: &str, + ) -> Result> { + let filter = + EnvFilter::try_from_default_env().or_else(|_| EnvFilter::try_new(default_log_level))?; + let fmt_layer = tracing_subscriber::fmt::layer() + .compact() + .with_target(false) + .with_filter(filter_fn(|metadata| { + metadata.target() != errors::EXCEPTION_TARGET + })); + + if !config.enabled { + tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .init(); + return Ok(Self { + logger_provider: None, + meter_provider: None, + tracer_provider: None, + }); + } + + errors::enable_backtraces(); + let resource = Resource::builder() + .with_service_name(env!("CARGO_PKG_NAME")) + .with_attributes([ + KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), + KeyValue::new("deployment.environment.name", environment.to_owned()), + ]) + .build(); + + let span_exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(config.endpoint.clone()) + .build()?; + let tracer_provider = SdkTracerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(span_exporter) + .build(); + let tracer = tracer_provider.tracer(env!("CARGO_PKG_NAME")); + + let log_exporter = opentelemetry_otlp::LogExporter::builder() + .with_tonic() + .with_endpoint(config.endpoint.clone()) + .build()?; + let logger_provider = SdkLoggerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(log_exporter) + .build(); + + let metric_exporter = opentelemetry_otlp::MetricExporter::builder() + .with_tonic() + .with_endpoint(config.endpoint.clone()) + .build()?; + let metric_reader = PeriodicReader::builder(metric_exporter).build(); + let meter_provider = SdkMeterProvider::builder() + .with_resource(resource) + .with_reader(metric_reader) + .build(); + + global::set_text_map_propagator(TraceContextPropagator::new()); + global::set_meter_provider(meter_provider.clone()); + metrics::initialize(); + + let trace_layer = tracing_opentelemetry::layer() + .with_tracer(tracer) + .with_error_records_to_exceptions(true) + .with_error_events_to_status(true) + .with_filter(filter_fn(|metadata| { + metadata.target() != errors::SUMMARY_TARGET + })); + let log_layer = + OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_fn(|metadata| { + metadata.target() != errors::SUMMARY_TARGET + })); + + tracing_subscriber::registry() + .with(filter) + .with(fmt_layer) + .with(trace_layer) + .with(log_layer) + .init(); + + Ok(Self { + logger_provider: Some(logger_provider), + meter_provider: Some(meter_provider), + tracer_provider: Some(tracer_provider), + }) + } + + pub fn shutdown(self) { + if let Some(provider) = self.logger_provider { + let _ = provider.shutdown(); + } + if let Some(provider) = self.meter_provider { + let _ = provider.shutdown(); + } + if let Some(provider) = self.tracer_provider { + let _ = provider.shutdown(); + } + } +} From a95831e451b7374cef8a249c18b8588637fb4471 Mon Sep 17 00:00:00 2001 From: Raphael Date: Tue, 7 Jul 2026 10:50:30 +0200 Subject: [PATCH 4/4] feat: made each endpoint configurable --- aquila.yml | 7 ++- src/configuration/config.rs | 104 ++++++++++++++++++++++++++++++--- src/main.rs | 2 +- src/telemetry/mod.rs | 112 +++++++++++++++++++++--------------- 4 files changed, 167 insertions(+), 58 deletions(-) diff --git a/aquila.yml b/aquila.yml index 9123ab1..5623649 100644 --- a/aquila.yml +++ b/aquila.yml @@ -10,9 +10,12 @@ mode: static log_level: debug # OTLP export for logs, traces, and the deliberately small metric set. -telemetry: +opentelemetry: enabled: false - endpoint: http://localhost:4317 + service_name: aquila + logs_endpoint: + metrics_endpoint: + traces_endpoint: # NATS server and JetStream key-value bucket used for flow storage. nats: diff --git a/src/configuration/config.rs b/src/configuration/config.rs index ce6b654..52b346e 100644 --- a/src/configuration/config.rs +++ b/src/configuration/config.rs @@ -14,7 +14,8 @@ pub struct Config { pub environment: Environment, pub mode: Mode, pub log_level: String, - pub telemetry: Telemetry, + #[serde(alias = "telemetry")] + pub opentelemetry: OpenTelemetry, pub nats: Nats, pub static_config: StaticConfig, pub dynamic_config: DynamicConfig, @@ -31,9 +32,12 @@ pub struct Nats { #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(default)] -pub struct Telemetry { +pub struct OpenTelemetry { pub enabled: bool, - pub endpoint: String, + pub service_name: String, + pub logs_endpoint: Option, + pub metrics_endpoint: Option, + pub traces_endpoint: Option, } #[derive(Clone, Debug, Deserialize, Serialize)] @@ -86,7 +90,7 @@ impl Default for Config { environment: Environment::Development, mode: Mode::Static, log_level: "debug".into(), - telemetry: Telemetry::default(), + opentelemetry: OpenTelemetry::default(), nats: Nats::default(), static_config: StaticConfig::default(), dynamic_config: DynamicConfig::default(), @@ -96,15 +100,42 @@ impl Default for Config { } } -impl Default for Telemetry { +impl Default for OpenTelemetry { fn default() -> Self { Self { enabled: false, - endpoint: "http://localhost:4317".into(), + service_name: env!("CARGO_PKG_NAME").into(), + logs_endpoint: None, + metrics_endpoint: None, + traces_endpoint: None, } } } +impl OpenTelemetry { + pub fn logs_endpoint(&self) -> Option<&str> { + non_empty_url(&self.logs_endpoint) + } + + pub fn metrics_endpoint(&self) -> Option<&str> { + non_empty_url(&self.metrics_endpoint) + } + + pub fn traces_endpoint(&self) -> Option<&str> { + non_empty_url(&self.traces_endpoint) + } + + pub fn has_enabled_exporter(&self) -> bool { + self.logs_endpoint().is_some() + || self.metrics_endpoint().is_some() + || self.traces_endpoint().is_some() + } +} + +fn non_empty_url(url: &Option) -> Option<&str> { + url.as_deref().filter(|value| !value.trim().is_empty()) +} + impl Default for Nats { fn default() -> Self { Self { @@ -194,8 +225,27 @@ impl fmt::Display for Config { writeln!(formatter, " Mode: {}", self.mode)?; writeln!(formatter, " Log level: {}", self.log_level)?; writeln!(formatter, " OpenTelemetry")?; - writeln!(formatter, " Enabled: {}", self.telemetry.enabled)?; - writeln!(formatter, " Endpoint: {}", self.telemetry.endpoint)?; + writeln!(formatter, " Enabled: {}", self.opentelemetry.enabled)?; + writeln!( + formatter, + " Service: {}", + self.opentelemetry.service_name + )?; + writeln!( + formatter, + " Logs: {}", + display_optional_url(&self.opentelemetry.logs_endpoint) + )?; + writeln!( + formatter, + " Metrics: {}", + display_optional_url(&self.opentelemetry.metrics_endpoint) + )?; + writeln!( + formatter, + " Traces: {}", + display_optional_url(&self.opentelemetry.traces_endpoint) + )?; writeln!(formatter, " NATS")?; writeln!(formatter, " URL: {}", self.nats.url)?; writeln!(formatter, " Bucket: {}", self.nats.bucket)?; @@ -243,11 +293,17 @@ impl fmt::Display for Config { } } +fn display_optional_url(url: &Option) -> &str { + non_empty_url(url).unwrap_or("") +} + #[cfg(test)] mod tests { use std::sync::Mutex; - use super::Config; + use config::Config as ConfigLoader; + + use super::{Config, OpenTelemetry}; static ENV_LOCK: Mutex<()> = Mutex::new(()); @@ -296,4 +352,34 @@ mod tests { assert!(!output.contains("super-secret")); assert!(!output.contains("Config {")); } + + #[test] + fn opentelemetry_endpoints_are_enabled_by_presence() { + let config: OpenTelemetry = ConfigLoader::builder() + .add_source( + ConfigLoader::try_from(&OpenTelemetry::default()) + .expect("default telemetry config should serialize"), + ) + .set_override("enabled", true) + .expect("enabled override should apply") + .set_override("service_name", "sagittarius") + .expect("service name override should apply") + .set_override("logs_endpoint", "") + .expect("logs override should apply") + .set_override("metrics_endpoint", " ") + .expect("metrics override should apply") + .set_override("traces_endpoint", "http://localhost:4317") + .expect("traces override should apply") + .build() + .expect("telemetry config should build") + .try_deserialize() + .expect("telemetry config should deserialize"); + + assert!(config.enabled); + assert_eq!(config.service_name, "sagittarius"); + assert_eq!(config.logs_endpoint(), None); + assert_eq!(config.metrics_endpoint(), None); + assert_eq!(config.traces_endpoint(), Some("http://localhost:4317")); + assert!(config.has_enabled_exporter()); + } } diff --git a/src/main.rs b/src/main.rs index 0cfa414..c64a036 100644 --- a/src/main.rs +++ b/src/main.rs @@ -29,7 +29,7 @@ async fn main() { .unwrap_or("debug"); let telemetry_config = config_result .as_ref() - .map(|config| config.telemetry.clone()) + .map(|config| config.opentelemetry.clone()) .unwrap_or_default(); let environment = config_result .as_ref() diff --git a/src/telemetry/mod.rs b/src/telemetry/mod.rs index fbfe5ee..488c6cf 100644 --- a/src/telemetry/mod.rs +++ b/src/telemetry/mod.rs @@ -17,7 +17,7 @@ use tracing_subscriber::{ EnvFilter, Layer, filter::filter_fn, layer::SubscriberExt, util::SubscriberInitExt, }; -use crate::configuration::config::Telemetry as TelemetryConfig; +use crate::configuration::config::OpenTelemetry as TelemetryConfig; pub struct Telemetry { logger_provider: Option, @@ -40,7 +40,7 @@ impl Telemetry { metadata.target() != errors::EXCEPTION_TARGET })); - if !config.enabled { + if !config.enabled || !config.has_enabled_exporter() { tracing_subscriber::registry() .with(filter) .with(fmt_layer) @@ -52,59 +52,79 @@ impl Telemetry { }); } - errors::enable_backtraces(); let resource = Resource::builder() - .with_service_name(env!("CARGO_PKG_NAME")) + .with_service_name(config.service_name.clone()) .with_attributes([ KeyValue::new("service.version", env!("CARGO_PKG_VERSION")), KeyValue::new("deployment.environment.name", environment.to_owned()), ]) .build(); - let span_exporter = opentelemetry_otlp::SpanExporter::builder() - .with_tonic() - .with_endpoint(config.endpoint.clone()) - .build()?; - let tracer_provider = SdkTracerProvider::builder() - .with_resource(resource.clone()) - .with_batch_exporter(span_exporter) - .build(); - let tracer = tracer_provider.tracer(env!("CARGO_PKG_NAME")); + let tracer_provider = if let Some(endpoint) = config.traces_endpoint() { + let exporter = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint.to_owned()) + .build()?; + let provider = SdkTracerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(exporter) + .build(); + global::set_text_map_propagator(TraceContextPropagator::new()); + Some(provider) + } else { + None + }; - let log_exporter = opentelemetry_otlp::LogExporter::builder() - .with_tonic() - .with_endpoint(config.endpoint.clone()) - .build()?; - let logger_provider = SdkLoggerProvider::builder() - .with_resource(resource.clone()) - .with_batch_exporter(log_exporter) - .build(); + let logger_provider = if let Some(endpoint) = config.logs_endpoint() { + let exporter = opentelemetry_otlp::LogExporter::builder() + .with_tonic() + .with_endpoint(endpoint.to_owned()) + .build()?; + Some( + SdkLoggerProvider::builder() + .with_resource(resource.clone()) + .with_batch_exporter(exporter) + .build(), + ) + } else { + None + }; - let metric_exporter = opentelemetry_otlp::MetricExporter::builder() - .with_tonic() - .with_endpoint(config.endpoint.clone()) - .build()?; - let metric_reader = PeriodicReader::builder(metric_exporter).build(); - let meter_provider = SdkMeterProvider::builder() - .with_resource(resource) - .with_reader(metric_reader) - .build(); + let meter_provider = if let Some(endpoint) = config.metrics_endpoint() { + let exporter = opentelemetry_otlp::MetricExporter::builder() + .with_tonic() + .with_endpoint(endpoint.to_owned()) + .build()?; + let reader = PeriodicReader::builder(exporter).build(); + let provider = SdkMeterProvider::builder() + .with_resource(resource) + .with_reader(reader) + .build(); + global::set_meter_provider(provider.clone()); + metrics::initialize(); + Some(provider) + } else { + None + }; - global::set_text_map_propagator(TraceContextPropagator::new()); - global::set_meter_provider(meter_provider.clone()); - metrics::initialize(); + if config.logs_endpoint().is_some() || config.traces_endpoint().is_some() { + errors::enable_backtraces(); + } - let trace_layer = tracing_opentelemetry::layer() - .with_tracer(tracer) - .with_error_records_to_exceptions(true) - .with_error_events_to_status(true) - .with_filter(filter_fn(|metadata| { - metadata.target() != errors::SUMMARY_TARGET - })); - let log_layer = - OpenTelemetryTracingBridge::new(&logger_provider).with_filter(filter_fn(|metadata| { + let trace_layer = tracer_provider.as_ref().map(|provider| { + tracing_opentelemetry::layer() + .with_tracer(provider.tracer(env!("CARGO_PKG_NAME"))) + .with_error_records_to_exceptions(true) + .with_error_events_to_status(true) + .with_filter(filter_fn(|metadata| { + metadata.target() != errors::SUMMARY_TARGET + })) + }); + let log_layer = logger_provider.as_ref().map(|provider| { + OpenTelemetryTracingBridge::new(provider).with_filter(filter_fn(|metadata| { metadata.target() != errors::SUMMARY_TARGET - })); + })) + }); tracing_subscriber::registry() .with(filter) @@ -114,9 +134,9 @@ impl Telemetry { .init(); Ok(Self { - logger_provider: Some(logger_provider), - meter_provider: Some(meter_provider), - tracer_provider: Some(tracer_provider), + logger_provider, + meter_provider, + tracer_provider, }) }