added open telementry

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"] }

aquila.yml

@@ -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

src/configuration/config.rs

@@ -14,6 +14,8 @@ pub struct Config {
     pub environment: Environment,
     pub mode: Mode,
     pub log_level: String,
+    #[serde(alias = "telemetry")]
+    pub opentelemetry: OpenTelemetry,
     pub nats: Nats,
     pub static_config: StaticConfig,
     pub dynamic_config: DynamicConfig,
@@ -28,6 +30,16 @@ 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<String>,
+    pub metrics_endpoint: Option<String>,
+    pub traces_endpoint: Option<String>,
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(default)]
 pub struct StaticConfig {
@@ -78,6 +90,7 @@ impl Default for Config {
             environment: Environment::Development,
             mode: Mode::Static,
             log_level: "debug".into(),
+            opentelemetry: OpenTelemetry::default(),
             nats: Nats::default(),
             static_config: StaticConfig::default(),
             dynamic_config: DynamicConfig::default(),
@@ -87,6 +100,42 @@ 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 non_empty_url(url: &Option<String>) -> Option<&str> {
+    url.as_deref().filter(|value| !value.trim().is_empty())
+}
+
 impl Default for Nats {
     fn default() -> Self {
         Self {
@@ -175,6 +224,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 +293,17 @@ impl fmt::Display for Config {
     }
 }
 
+fn display_optional_url(url: &Option<String>) -> &str {
+    non_empty_url(url).unwrap_or("<disabled>")
+}
+
 #[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(());
 
@@ -275,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());
+    }
 }

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.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, &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() {
             "<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/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,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"))
     }
 }