Merge pull request #376 from code0-tech/#353-open-telementry

added open telementry

Author: raphael-goetz

Cargo.lock

@@ -7,11 +7,12 @@ 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 = ["metrics"] }
+tracing = { version = "0.1.41", features = ["log"] }
 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"

Cargo.toml

@@ -6,9 +6,17 @@ 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.
+opentelemetry:
+  enabled: false
+  service_name: aquila
+  logs_endpoint:
+  metrics_endpoint:
+  traces_endpoint:
+
 # NATS server and JetStream key-value bucket used for flow storage.
 nats:
   url: nats://localhost:4222

aquila.yml

@@ -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};
 
@@ -14,6 +15,9 @@ pub struct Config {
     pub environment: Environment,
     pub mode: Mode,
     pub log_level: String,
+    #[serde(alias = "telemetry")]
+    #[serde(default = "default_opentelemetry")]
+    pub opentelemetry: OpenTelemetry,
     pub nats: Nats,
     pub static_config: StaticConfig,
     pub dynamic_config: DynamicConfig,
@@ -78,6 +82,7 @@ impl Default for Config {
             environment: Environment::Development,
             mode: Mode::Static,
             log_level: "debug".into(),
+            opentelemetry: default_opentelemetry(),
             nats: Nats::default(),
             static_config: StaticConfig::default(),
             dynamic_config: DynamicConfig::default(),
@@ -87,6 +92,13 @@ impl Default for Config {
     }
 }
 
+fn default_opentelemetry() -> OpenTelemetry {
+    OpenTelemetry {
+        service_name: env!("CARGO_PKG_NAME").into(),
+        ..OpenTelemetry::default()
+    }
+}
+
 impl Default for Nats {
     fn default() -> Self {
         Self {
@@ -175,6 +187,28 @@ 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.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)?;
@@ -222,11 +256,21 @@ impl fmt::Display for Config {
     }
 }
 
+fn display_optional_url(url: &Option<String>) -> &str {
+    url.as_deref()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or("<disabled>")
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Mutex;
 
-    use super::Config;
+    use config::Config as ConfigLoader;
+
+    use code0_flow::flow_telemetry::OpenTelemetry;
+
+    use super::{Config, default_opentelemetry};
 
     static ENV_LOCK: Mutex<()> = Mutex::new(());
 
@@ -275,4 +319,39 @@ 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(&default_opentelemetry())
+                    .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());
+    }
+
+    #[test]
+    fn opentelemetry_default_service_name_is_aquila() {
+        assert_eq!(Config::default().opentelemetry.service_name, "aquila");
+    }
 }

src/configuration/config.rs

@@ -88,7 +88,7 @@ impl From<SerializableServiceConfiguration> 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
-            .runtimes
+        self.runtimes
             .iter()
             .find(|x| &x.token == token && x.identifier == name)
-        {
-            Some(_) => true,
-            None => false,
-        }
+            .is_some()
     }
 
     pub fn has_action(&self, token: &String, action_name: &String) -> bool {
-        match self
-            .actions
+        self.actions
             .iter()
             .find(|x| &x.token == token && &x.service_name == action_name)
-        {
-            Some(_) => true,
-            None => false,
-        }
+            .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()
             }
-        };
+        }
     }
 }
 

src/configuration/service.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,30 @@ 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.opentelemetry.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,
+        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
         .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 +59,7 @@ async fn main() {
     log::debug!("{config}");
 
     startup::run(config, app_readiness, service_config).await;
+    telemetry.shutdown();
 }
 
 fn install_panic_logging() {
@@ -52,15 +72,17 @@ fn install_panic_logging() {
             "<non-string panic payload>"
         };
 
-        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);
     }));
 }

src/main.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,
@@ -192,16 +197,22 @@ 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(_) => 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,15 +330,18 @@ 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);
                 }
             };
         }
 
         // 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"))
     }
 }