From 1afe7c5f98a7341c6f50b97e5fae23368ca37dbb Mon Sep 17 00:00:00 2001 From: Raphael Date: Mon, 6 Jul 2026 12:50:28 +0200 Subject: [PATCH 1/8] 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/8] 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/8] 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/8] 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, }) } From e9ee4caa410b27e2191324624ddad14b2bfbe666 Mon Sep 17 00:00:00 2001 From: Raphael Date: Thu, 9 Jul 2026 16:04:55 +0200 Subject: [PATCH 5/8] deps: updated code0-flow & dropped most open telemetry crates --- Cargo.lock | 50 +++++++++----------------------------------------- Cargo.toml | 9 ++------- 2 files changed, 11 insertions(+), 48 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3caaede..e0ce9fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -28,9 +28,6 @@ dependencies = [ "futures-core", "log", "opentelemetry", - "opentelemetry-appender-tracing", - "opentelemetry-otlp", - "opentelemetry_sdk", "prost", "serde", "serde_json", @@ -39,8 +36,6 @@ dependencies = [ "tonic", "tonic-health", "tracing", - "tracing-opentelemetry", - "tracing-subscriber", "tucana", "uuid", ] @@ -236,23 +231,24 @@ dependencies = [ [[package]] name = "code0-flow" -version = "0.0.38" +version = "0.0.40" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "708985c5041a712c44a26f15b5658946469ff79dc092e5ffb37e8861029f6bd9" +checksum = "f6bac81931da426429ac4ca93b80ff246900d64e8190ec2c4e0382ff35157857" dependencies = [ "async-nats", - "async-trait", "dotenv", "futures-core", "log", - "regex", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry_sdk", "serde", - "serde_json", - "tokio", "tonic", "tonic-health", - "tucana", - "walkdir", + "tracing", + "tracing-opentelemetry", + "tracing-subscriber", ] [[package]] @@ -1802,15 +1798,6 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" -[[package]] -name = "same-file" -version = "1.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" -dependencies = [ - "winapi-util", -] - [[package]] name = "schannel" version = "0.1.29" @@ -2616,16 +2603,6 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" -[[package]] -name = "walkdir" -version = "2.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" -dependencies = [ - "same-file", - "winapi-util", -] - [[package]] name = "want" version = "0.3.1" @@ -2786,15 +2763,6 @@ dependencies = [ "rustls-pki-types", ] -[[package]] -name = "winapi-util" -version = "0.1.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" -dependencies = [ - "windows-sys 0.61.2", -] - [[package]] name = "windows-link" version = "0.2.1" diff --git a/Cargo.toml b/Cargo.toml index 482e1ab..f5b4f36 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,17 +7,12 @@ edition = "2024" tokio = { version = "1.44.1", features = ["rt-multi-thread", "signal"] } futures = "0.3.31" log = "0.4.26" -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"] } +opentelemetry = { version = "0.32.0", features = ["metrics"] } 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"] } -code0-flow = { version = "0.0.38", features = ["flow_health"] } +code0-flow = { version = "0.0.40", features = ["flow_config", "flow_health", "flow_telemetry"] } serde_json = "1.0.140" async-nats = "0.49.0" tonic-health = "0.14.1" From 39fab21b985ceab6a52d08c99f30f0eef5e8e4b5 Mon Sep 17 00:00:00 2001 From: Raphael Date: Thu, 9 Jul 2026 16:05:10 +0200 Subject: [PATCH 6/8] feat: switched to code0-flow telemetry feature --- src/main.rs | 20 +++++- src/telemetry/errors.rs | 145 ------------------------------------- src/telemetry/mod.rs | 153 +--------------------------------------- 3 files changed, 19 insertions(+), 299 deletions(-) delete mode 100644 src/telemetry/errors.rs diff --git a/src/main.rs b/src/main.rs index c64a036..ad7e47e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -35,8 +35,24 @@ async fn main() { .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}")); + let telemetry_config = telemetry::OpenTelemetry { + enabled: telemetry_config.enabled, + service_name: telemetry_config.service_name, + logs_endpoint: telemetry_config.logs_endpoint, + metrics_endpoint: telemetry_config.metrics_endpoint, + traces_endpoint: telemetry_config.traces_endpoint, + }; + let telemetry = telemetry::Telemetry::initialize( + &telemetry_config, + telemetry::TelemetrySettings { + environment: &environment, + default_log_level: log_level, + service_version: env!("CARGO_PKG_VERSION"), + instrumentation_name: env!("CARGO_PKG_NAME"), + initialize_metrics: Some(telemetry::metrics::initialize), + }, + ) + .unwrap_or_else(|error| panic!("failed to initialize telemetry: {error}")); install_panic_logging(); let config = config_result diff --git a/src/telemetry/errors.rs b/src/telemetry/errors.rs deleted file mode 100644 index fad85b6..0000000 --- a/src/telemetry/errors.rs +++ /dev/null @@ -1,145 +0,0 @@ -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/mod.rs b/src/telemetry/mod.rs index 488c6cf..98f9dc3 100644 --- a/src/telemetry/mod.rs +++ b/src/telemetry/mod.rs @@ -1,154 +1,3 @@ -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::OpenTelemetry 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 || !config.has_enabled_exporter() { - tracing_subscriber::registry() - .with(filter) - .with(fmt_layer) - .init(); - return Ok(Self { - logger_provider: None, - meter_provider: None, - tracer_provider: None, - }); - } - - let resource = Resource::builder() - .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 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 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 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 - }; - - if config.logs_endpoint().is_some() || config.traces_endpoint().is_some() { - errors::enable_backtraces(); - } - - 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) - .with(fmt_layer) - .with(trace_layer) - .with(log_layer) - .init(); - - Ok(Self { - logger_provider, - meter_provider, - 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(); - } - } -} +pub use code0_flow::flow_telemetry::{OpenTelemetry, Telemetry, TelemetrySettings, errors}; From fe4900387b0893122f517f78cee52b3d0fcf1468 Mon Sep 17 00:00:00 2001 From: Raphael Date: Thu, 9 Jul 2026 16:06:09 +0200 Subject: [PATCH 7/8] ref: cargo clippy --- src/configuration/service.rs | 22 ++++++------------- src/sagittarius/flow_service_client_impl.rs | 2 +- .../action_transfer_service_server_impl.rs | 11 ++++------ src/server/dynamic_server.rs | 7 +++--- 4 files changed, 15 insertions(+), 27 deletions(-) diff --git a/src/configuration/service.rs b/src/configuration/service.rs index 4e31ab3..c2997c5 100644 --- a/src/configuration/service.rs +++ b/src/configuration/service.rs @@ -88,7 +88,7 @@ impl From for ServiceConfiguration { fn from(value: SerializableServiceConfiguration) -> Self { Self { actions: value.actions.into_iter().map(Into::into).collect(), - runtimes: value.runtimes.into_iter().map(Into::into).collect(), + runtimes: value.runtimes.into_iter().collect(), } } } @@ -116,25 +116,17 @@ impl ServiceConfiguration { None => return false, }; - match self + self .runtimes .iter() - .find(|x| &x.token == token && x.identifier == name) - { - Some(_) => true, - None => false, - } + .find(|x| &x.token == token && x.identifier == name).is_some() } pub fn has_action(&self, token: &String, action_name: &String) -> bool { - match self + self .actions .iter() - .find(|x| &x.token == token && &x.service_name == action_name) - { - Some(_) => true, - None => false, - } + .find(|x| &x.token == token && &x.service_name == action_name).is_some() } pub fn get_action_configuration( @@ -185,9 +177,9 @@ impl ServiceConfiguration { "Couldn't parse service configuration file, Reason: {}. Starting with empty service configuration", error ); - return ServiceConfiguration::default(); + ServiceConfiguration::default() } - }; + } } } diff --git a/src/sagittarius/flow_service_client_impl.rs b/src/sagittarius/flow_service_client_impl.rs index 8fbcbc3..a788656 100644 --- a/src/sagittarius/flow_service_client_impl.rs +++ b/src/sagittarius/flow_service_client_impl.rs @@ -197,7 +197,7 @@ impl SagittariusFlowClient { } Data::UpdatedFlow(flow) => { let key = get_flow_identifier(&flow); - let flow_id = flow.flow_id.clone(); + let flow_id = flow.flow_id; let bytes = flow.encode_to_vec(); match self.store.put(key.clone(), bytes.into()).await { Ok(_) => { diff --git a/src/server/action_transfer_service_server_impl.rs b/src/server/action_transfer_service_server_impl.rs index 6d1d095..41599b5 100644 --- a/src/server/action_transfer_service_server_impl.rs +++ b/src/server/action_transfer_service_server_impl.rs @@ -219,11 +219,10 @@ fn pending_reply_keys( keys.push(request_execution_id.to_string()); } - if let Some(subject_execution_id) = subject_execution_id { - if !subject_execution_id.is_empty() && !keys.iter().any(|key| key == subject_execution_id) { + if let Some(subject_execution_id) = subject_execution_id + && !subject_execution_id.is_empty() && !keys.iter().any(|key| key == subject_execution_id) { keys.push(subject_execution_id.to_string()); } - } keys } @@ -564,7 +563,6 @@ async fn handle_result( &err, format!("action.identifier={action_identifier} execution_id={execution_id}"), ); - return; } } @@ -790,8 +788,8 @@ async fn forward_nats_to_action( }; let subject_execution_id = subject_execution_identifier(&msg.subject); - if execution.execution_identifier.is_empty() { - if let Some(subject_execution_id) = subject_execution_id.as_ref() { + if execution.execution_identifier.is_empty() + && let Some(subject_execution_id) = subject_execution_id.as_ref() { log::warn!( "Filled missing action execution identifier from NATS subject subject={} execution_id={}", msg.subject, @@ -799,7 +797,6 @@ async fn forward_nats_to_action( ); execution.execution_identifier = subject_execution_id.clone(); } - } let execution_id = execution.execution_identifier.clone(); diff --git a/src/server/dynamic_server.rs b/src/server/dynamic_server.rs index f424ad4..ae1fbf1 100644 --- a/src/server/dynamic_server.rs +++ b/src/server/dynamic_server.rs @@ -117,12 +117,11 @@ impl AquilaDynamicServer { let runtime_status_server = AquilaRuntimeStatusServiceServer::new( runtime_status_service.clone(), self.service_configuration.clone(), - Duration::from_secs(self.runtime_status_not_responding_after_secs.clone()), + Duration::from_secs(self.runtime_status_not_responding_after_secs), Duration::from_secs( - self.runtime_status_stopped_after_not_responding_secs - .clone(), + self.runtime_status_stopped_after_not_responding_secs, ), - Duration::from_secs(self.runtime_status_monitor_interval_secs.clone()), + Duration::from_secs(self.runtime_status_monitor_interval_secs), ); let action_transfer_server = AquilaActionTransferServiceServer::new( From 2481fb172d088d80b1997c550f01895ed52d10d8 Mon Sep 17 00:00:00 2001 From: Raphael Date: Thu, 9 Jul 2026 16:15:20 +0200 Subject: [PATCH 8/8] fix: cleanup after open telemetry feature from code0 flow --- src/configuration/config.rs | 66 ++++++------------- src/configuration/service.rs | 12 ++-- src/main.rs | 7 -- .../action_transfer_service_server_impl.rs | 25 +++---- src/server/dynamic_server.rs | 4 +- 5 files changed, 40 insertions(+), 74 deletions(-) diff --git a/src/configuration/config.rs b/src/configuration/config.rs index 52b346e..55b0a14 100644 --- a/src/configuration/config.rs +++ b/src/configuration/config.rs @@ -1,5 +1,6 @@ use std::{fmt, path::Path}; +use code0_flow::flow_telemetry::OpenTelemetry; use config::{Config as ConfigLoader, ConfigError, File}; use serde::{Deserialize, Serialize}; @@ -15,6 +16,7 @@ pub struct Config { pub mode: Mode, pub log_level: String, #[serde(alias = "telemetry")] + #[serde(default = "default_opentelemetry")] pub opentelemetry: OpenTelemetry, pub nats: Nats, pub static_config: StaticConfig, @@ -30,16 +32,6 @@ pub struct Nats { pub bucket: String, } -#[derive(Clone, Debug, Deserialize, Serialize)] -#[serde(default)] -pub struct OpenTelemetry { - pub enabled: bool, - pub service_name: String, - pub logs_endpoint: Option, - pub metrics_endpoint: Option, - pub traces_endpoint: Option, -} - #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(default)] pub struct StaticConfig { @@ -90,7 +82,7 @@ impl Default for Config { environment: Environment::Development, mode: Mode::Static, log_level: "debug".into(), - opentelemetry: OpenTelemetry::default(), + opentelemetry: default_opentelemetry(), nats: Nats::default(), static_config: StaticConfig::default(), dynamic_config: DynamicConfig::default(), @@ -100,42 +92,13 @@ impl Default for Config { } } -impl Default for OpenTelemetry { - fn default() -> Self { - Self { - enabled: false, - 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 default_opentelemetry() -> OpenTelemetry { + OpenTelemetry { + service_name: env!("CARGO_PKG_NAME").into(), + ..OpenTelemetry::default() } } -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 { @@ -294,7 +257,9 @@ impl fmt::Display for Config { } fn display_optional_url(url: &Option) -> &str { - non_empty_url(url).unwrap_or("") + url.as_deref() + .filter(|value| !value.trim().is_empty()) + .unwrap_or("") } #[cfg(test)] @@ -303,7 +268,9 @@ mod tests { use config::Config as ConfigLoader; - use super::{Config, OpenTelemetry}; + use code0_flow::flow_telemetry::OpenTelemetry; + + use super::{Config, default_opentelemetry}; static ENV_LOCK: Mutex<()> = Mutex::new(()); @@ -357,7 +324,7 @@ mod tests { fn opentelemetry_endpoints_are_enabled_by_presence() { let config: OpenTelemetry = ConfigLoader::builder() .add_source( - ConfigLoader::try_from(&OpenTelemetry::default()) + ConfigLoader::try_from(&default_opentelemetry()) .expect("default telemetry config should serialize"), ) .set_override("enabled", true) @@ -382,4 +349,9 @@ mod tests { assert_eq!(config.traces_endpoint(), Some("http://localhost:4317")); assert!(config.has_enabled_exporter()); } + + #[test] + fn opentelemetry_default_service_name_is_aquila() { + assert_eq!(Config::default().opentelemetry.service_name, "aquila"); + } } diff --git a/src/configuration/service.rs b/src/configuration/service.rs index c2997c5..cac2f39 100644 --- a/src/configuration/service.rs +++ b/src/configuration/service.rs @@ -116,17 +116,17 @@ impl ServiceConfiguration { None => return false, }; - self - .runtimes + self.runtimes .iter() - .find(|x| &x.token == token && x.identifier == name).is_some() + .find(|x| &x.token == token && x.identifier == name) + .is_some() } pub fn has_action(&self, token: &String, action_name: &String) -> bool { - self - .actions + self.actions .iter() - .find(|x| &x.token == token && &x.service_name == action_name).is_some() + .find(|x| &x.token == token && &x.service_name == action_name) + .is_some() } pub fn get_action_configuration( diff --git a/src/main.rs b/src/main.rs index ad7e47e..76e5be3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -35,13 +35,6 @@ async fn main() { .as_ref() .map(|config| config.environment.to_string()) .unwrap_or_else(|_| "unknown".into()); - let telemetry_config = telemetry::OpenTelemetry { - enabled: telemetry_config.enabled, - service_name: telemetry_config.service_name, - logs_endpoint: telemetry_config.logs_endpoint, - metrics_endpoint: telemetry_config.metrics_endpoint, - traces_endpoint: telemetry_config.traces_endpoint, - }; let telemetry = telemetry::Telemetry::initialize( &telemetry_config, telemetry::TelemetrySettings { diff --git a/src/server/action_transfer_service_server_impl.rs b/src/server/action_transfer_service_server_impl.rs index 41599b5..9b0e74b 100644 --- a/src/server/action_transfer_service_server_impl.rs +++ b/src/server/action_transfer_service_server_impl.rs @@ -220,9 +220,11 @@ fn pending_reply_keys( } if let Some(subject_execution_id) = subject_execution_id - && !subject_execution_id.is_empty() && !keys.iter().any(|key| key == subject_execution_id) { - keys.push(subject_execution_id.to_string()); - } + && !subject_execution_id.is_empty() + && !keys.iter().any(|key| key == subject_execution_id) + { + keys.push(subject_execution_id.to_string()); + } keys } @@ -789,14 +791,15 @@ async fn forward_nats_to_action( let subject_execution_id = subject_execution_identifier(&msg.subject); if execution.execution_identifier.is_empty() - && let Some(subject_execution_id) = subject_execution_id.as_ref() { - log::warn!( - "Filled missing action execution identifier from NATS subject subject={} execution_id={}", - msg.subject, - subject_execution_id - ); - execution.execution_identifier = subject_execution_id.clone(); - } + && let Some(subject_execution_id) = subject_execution_id.as_ref() + { + log::warn!( + "Filled missing action execution identifier from NATS subject subject={} execution_id={}", + msg.subject, + subject_execution_id + ); + execution.execution_identifier = subject_execution_id.clone(); + } let execution_id = execution.execution_identifier.clone(); diff --git a/src/server/dynamic_server.rs b/src/server/dynamic_server.rs index ae1fbf1..02ddc42 100644 --- a/src/server/dynamic_server.rs +++ b/src/server/dynamic_server.rs @@ -118,9 +118,7 @@ impl AquilaDynamicServer { runtime_status_service.clone(), self.service_configuration.clone(), Duration::from_secs(self.runtime_status_not_responding_after_secs), - Duration::from_secs( - self.runtime_status_stopped_after_not_responding_secs, - ), + Duration::from_secs(self.runtime_status_stopped_after_not_responding_secs), Duration::from_secs(self.runtime_status_monitor_interval_secs), );