fix: cleanup after open telemetry feature from code0 flow

Author: raphael-goetz

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<String>,
-    pub metrics_endpoint: Option<String>,
-    pub traces_endpoint: Option<String>,
-}
-
 #[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<String>) -> 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<String>) -> &str {
-    non_empty_url(url).unwrap_or("<disabled>")
+    url.as_deref()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or("<disabled>")
 }
 
 #[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");
+    }
 }

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(

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 {

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();
 

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),
         );