split instrumentation in 2 different places: harmony domain (for domain observability) & harmoy composer (for build/commands observability)

Cargo.lock

@@ -1816,10 +1816,14 @@ version = "0.1.0"
 dependencies = [
  "assert_cmd",
  "clap",
+ "console",
  "env_logger",
  "harmony",
  "harmony_tui",
+ "indicatif",
  "inquire",
+ "lazy_static",
+ "log",
  "tokio",
 ]
 
@@ -1834,10 +1838,11 @@ dependencies = [
  "current_platform",
  "env_logger",
  "futures-util",
- "harmony",
+ "harmony_cli",
  "indicatif",
  "lazy_static",
  "log",
+ "once_cell",
  "serde_json",
  "tokio",
 ]

examples/rust/src/main.rs

@@ -9,10 +9,12 @@ use harmony::{
     },
     topology::{K8sAnywhereTopology, Url},
 };
+use harmony_cli::cli_logger;
 
 #[tokio::main]
 async fn main() {
     env_logger::init();
+    let cli_logger_handle = tokio::spawn(cli_logger::init());
 
     let topology = K8sAnywhereTopology::from_env();
     let mut maestro = Maestro::initialize(Inventory::autoload(), topology)
@@ -40,4 +42,6 @@ async fn main() {
 
     maestro.register_all(vec![Box::new(app)]);
     harmony_cli::init(maestro, None).await.unwrap();
+
+    let _ = tokio::try_join!(cli_logger_handle);
 }

harmony/src/domain/instrumentation.rs

@@ -2,29 +2,19 @@ use log::debug;
 use once_cell::sync::Lazy;
 use tokio::sync::broadcast;
 
-// FIXME: Ce module d'instrumentation ne peut pas fonctionner à la fois pour Harmony Composer et
-// Harmony CLI. Elle n'a donc pas entièrement sa place ici et devra être séparée en deux.
 #[derive(Debug, Clone)]
 pub enum HarmonyEvent {
-    ProjectInitializationStarted,
-    ProjectInitialized,
-    ProjectCompilationStarted { details: String },
-    ProjectCompiled,
-    ProjectCompilationFailed { details: String },
-    DeploymentStarted { target: String },
-    DeploymentCompleted { details: String },
     PrepareTopologyStarted { name: String },
-    Shutdown,
 }
 
-static EVENT_BUS: Lazy<broadcast::Sender<HarmonyEvent>> = Lazy::new(|| {
+static HARMONY_EVENT_BUS: Lazy<broadcast::Sender<HarmonyEvent>> = Lazy::new(|| {
     // TODO: Adjust channel capacity
-    let (tx, _rx) = broadcast::channel(16);
+    let (tx, _rx) = broadcast::channel(18);
     tx
 });
 
 pub fn instrument(event: HarmonyEvent) {
-    EVENT_BUS.send(event).expect("couldn't send event");
+    HARMONY_EVENT_BUS.send(event).expect("couldn't send event");
 }
 
 pub async fn subscribe<F, Fut>(name: &str, mut handler: F)
@@ -32,7 +22,7 @@ where
     F: FnMut(HarmonyEvent) -> Fut + Send + 'static,
     Fut: Future<Output = bool> + Send,
 {
-    let mut rx = EVENT_BUS.subscribe();
+    let mut rx = HARMONY_EVENT_BUS.subscribe();
     debug!("[{name}] Service started. Listening for events...");
     loop {
         match rx.recv().await {

harmony/src/domain/maestro/mod.rs

@@ -42,12 +42,9 @@ impl<T: Topology> Maestro<T> {
     /// Ensures the associated Topology is ready for operations.
     /// Delegates the readiness check and potential setup actions to the Topology.
     pub async fn prepare_topology(&self) -> Result<Outcome, InterpretError> {
-        // FIXME: Cette instrumentation ne peut pas communiquer avec Harmony Composer puisqu'il
-        // s'agit d'un process différent.
-        //
-        // instrumentation::instrument(HarmonyEvent::PrepareTopologyStarted {
-        //     name: self.topology.name().to_string(),
-        // });
+        instrumentation::instrument(HarmonyEvent::PrepareTopologyStarted {
+            name: self.topology.name().to_string(),
+        });
         let outcome = self.topology.ensure_ready().await?;
         info!(
             "Topology '{}' readiness check complete: {}",

harmony_cli/Cargo.toml

@@ -13,6 +13,10 @@ harmony_tui = { path = "../harmony_tui", optional = true }
 inquire.workspace = true
 tokio.workspace = true
 env_logger.workspace = true
+console = "0.16.0"
+indicatif = "0.18.0"
+lazy_static = "1.5.0"
+log.workspace = true
 
 
 [features]

harmony_cli/src/cli_logger.rs

@@ -0,0 +1,28 @@
+use harmony::instrumentation::{self, HarmonyEvent};
+use indicatif::ProgressBar;
+use std::sync::{Arc, Mutex};
+
+pub async fn init() {
+    instrumentation::subscribe("CLI Logger", {
+        let current_spinner = Arc::new(Mutex::new(None::<ProgressBar>));
+
+        move |event| {
+            let spinner_clone = Arc::clone(&current_spinner);
+
+            async move {
+                let mut spinner_guard = spinner_clone.lock().unwrap();
+
+                match event {
+                    HarmonyEvent::PrepareTopologyStarted { name } => {
+                        println!(
+                            "{} Preparing environment: {name}...",
+                            crate::theme::EMOJI_TOPOLOGY
+                        );
+                    }
+                }
+                true
+            }
+        }
+    })
+    .await
+}

harmony_cli/src/lib.rs

@@ -4,6 +4,9 @@ use harmony;
 use harmony::{score::Score, topology::Topology};
 use inquire::Confirm;
 
+pub mod cli_logger; // FIXME: Don't make me pub
+pub mod theme;
+
 #[cfg(feature = "tui")]
 use harmony_tui;
 

harmony_cli/src/theme.rs

@@ -0,0 +1,22 @@
+use console::Emoji;
+use indicatif::ProgressStyle;
+use lazy_static::lazy_static;
+
+pub static EMOJI_HARMONY: Emoji<'_, '_> = Emoji("🎼", "");
+pub static EMOJI_SUCCESS: Emoji<'_, '_> = Emoji("✅", "");
+pub static EMOJI_ERROR: Emoji<'_, '_> = Emoji("⚠️", "");
+pub static EMOJI_DEPLOY: Emoji<'_, '_> = Emoji("🚀", "");
+pub static EMOJI_TOPOLOGY: Emoji<'_, '_> = Emoji("📦", "");
+
+lazy_static! {
+    pub static ref SPINNER_STYLE: ProgressStyle = ProgressStyle::default_spinner()
+        .template("    {spinner:.green} {msg}")
+        .unwrap()
+        .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]);
+    pub static ref SUCCESS_SPINNER_STYLE: ProgressStyle = SPINNER_STYLE
+        .clone()
+        .tick_strings(&[format!("{}", EMOJI_SUCCESS).as_str()]);
+    pub static ref ERROR_SPINNER_STYLE: ProgressStyle = SPINNER_STYLE
+        .clone()
+        .tick_strings(&[format!("{}", EMOJI_ERROR).as_str()]);
+}

harmony_composer/Cargo.toml

@@ -15,7 +15,8 @@ current_platform = "0.2.0"
 futures-util = "0.3.31"
 serde_json = "1.0.140"
 cargo_metadata = "0.20.0"
-harmony = { path = "../harmony" }
 indicatif = "0.18.0"
 console = "0.16.0"
 lazy_static = "1.5.0"
+once_cell = "1.21.3"
+harmony_cli = { path = "../harmony_cli" }

harmony_composer/src/cli_logger.rs

@@ -1,84 +0,0 @@
-use console::Emoji;
-use harmony::instrumentation::{self, HarmonyEvent};
-use indicatif::{ProgressBar, ProgressStyle};
-use lazy_static::lazy_static;
-use log::error;
-use std::{
-    sync::{Arc, Mutex},
-    time::Duration,
-};
-
-static EMOJI_HARMONY: Emoji<'_, '_> = Emoji("🎼", "");
-static EMOJI_SUCCESS: Emoji<'_, '_> = Emoji("✅", "");
-static EMOJI_ERROR: Emoji<'_, '_> = Emoji("⚠️", "");
-static EMOJI_DEPLOY: Emoji<'_, '_> = Emoji("🚀", "");
-static EMOJI_TOPOLOGY: Emoji<'_, '_> = Emoji("📦", "");
-
-lazy_static! {
-    pub static ref SPINNER_STYLE: ProgressStyle = ProgressStyle::default_spinner()
-        .template("    {spinner:.green} {msg}")
-        .unwrap()
-        .tick_strings(&["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]);
-    pub static ref SUCCESS_SPINNER_STYLE: ProgressStyle = SPINNER_STYLE
-        .clone()
-        .tick_strings(&[format!("{}", EMOJI_SUCCESS).as_str()]);
-    pub static ref ERROR_SPINNER_STYLE: ProgressStyle = SPINNER_STYLE
-        .clone()
-        .tick_strings(&[format!("{}", EMOJI_ERROR).as_str()]);
-}
-
-pub async fn init() {
-    instrumentation::subscribe("CLI Logger", {
-        let current_spinner = Arc::new(Mutex::new(None::<ProgressBar>));
-
-        move |event| {
-            let spinner_clone = Arc::clone(&current_spinner);
-
-            async move {
-                let mut spinner_guard = spinner_clone.lock().unwrap();
-
-                match event {
-                    HarmonyEvent::ProjectInitializationStarted => {
-                        println!("{} Initializing Harmony project...", EMOJI_HARMONY);
-                    }
-                    HarmonyEvent::ProjectInitialized => println!("\n"),
-                    HarmonyEvent::ProjectCompilationStarted { details } => {
-                        let progress = ProgressBar::new_spinner();
-                        progress.set_style(SPINNER_STYLE.clone());
-                        progress.set_message(details);
-                        progress.enable_steady_tick(Duration::from_millis(100));
-                        *spinner_guard = Some(progress);
-                    }
-                    HarmonyEvent::ProjectCompiled => {
-                        if let Some(progress) = spinner_guard.take() {
-                            progress.set_style(SUCCESS_SPINNER_STYLE.clone());
-                            progress.finish_with_message("project compiled");
-                        }
-                    }
-                    HarmonyEvent::ProjectCompilationFailed { details } => {
-                        if let Some(progress) = spinner_guard.take() {
-                            progress.set_style(ERROR_SPINNER_STYLE.clone());
-                            progress.finish_with_message("failed to compile project");
-                            error!("{details}");
-                        }
-                    }
-                    HarmonyEvent::DeploymentStarted { target } => {
-                        println!("{} Starting deployment to {target}...", EMOJI_DEPLOY);
-                    }
-                    HarmonyEvent::DeploymentCompleted { details } => println!("\n"),
-                    HarmonyEvent::PrepareTopologyStarted { name } => {
-                        println!("{} Preparing environment: {name}...", EMOJI_TOPOLOGY);
-                    }
-                    HarmonyEvent::Shutdown => {
-                        if let Some(progress) = spinner_guard.take() {
-                            progress.abandon();
-                        }
-                        return false;
-                    }
-                }
-                true
-            }
-        }
-    })
-    .await
-}

harmony_composer/src/harmony_composer_logger.rs

@@ -0,0 +1,67 @@
+use indicatif::ProgressBar;
+use log::error;
+use std::{
+    sync::{Arc, Mutex},
+    time::Duration,
+};
+
+use crate::instrumentation::{self, HarmonyComposerEvent};
+
+pub async fn init() {
+    instrumentation::subscribe("Harmony Composer Logger", {
+        let current_spinner = Arc::new(Mutex::new(None::<ProgressBar>));
+
+        move |event| {
+            let spinner_clone = Arc::clone(&current_spinner);
+
+            async move {
+                let mut spinner_guard = spinner_clone.lock().unwrap();
+
+                match event {
+                    HarmonyComposerEvent::ProjectInitializationStarted => {
+                        println!(
+                            "{} Initializing Harmony project...",
+                            harmony_cli::theme::EMOJI_HARMONY
+                        );
+                    }
+                    HarmonyComposerEvent::ProjectInitialized => println!("\n"),
+                    HarmonyComposerEvent::ProjectCompilationStarted { details } => {
+                        let progress = ProgressBar::new_spinner();
+                        progress.set_style(harmony_cli::theme::SPINNER_STYLE.clone());
+                        progress.set_message(details);
+                        progress.enable_steady_tick(Duration::from_millis(100));
+                        *spinner_guard = Some(progress);
+                    }
+                    HarmonyComposerEvent::ProjectCompiled => {
+                        if let Some(progress) = spinner_guard.take() {
+                            progress.set_style(harmony_cli::theme::SUCCESS_SPINNER_STYLE.clone());
+                            progress.finish_with_message("project compiled");
+                        }
+                    }
+                    HarmonyComposerEvent::ProjectCompilationFailed { details } => {
+                        if let Some(progress) = spinner_guard.take() {
+                            progress.set_style(harmony_cli::theme::ERROR_SPINNER_STYLE.clone());
+                            progress.finish_with_message("failed to compile project");
+                            error!("{details}");
+                        }
+                    }
+                    HarmonyComposerEvent::DeploymentStarted { target } => {
+                        println!(
+                            "{} Starting deployment to {target}...\n",
+                            harmony_cli::theme::EMOJI_DEPLOY
+                        );
+                    }
+                    HarmonyComposerEvent::DeploymentCompleted { details } => println!("\n"),
+                    HarmonyComposerEvent::Shutdown => {
+                        if let Some(progress) = spinner_guard.take() {
+                            progress.abandon();
+                        }
+                        return false;
+                    }
+                }
+                true
+            }
+        }
+    })
+    .await
+}

harmony_composer/src/instrumentation.rs

@@ -0,0 +1,51 @@
+use log::debug;
+use once_cell::sync::Lazy;
+use tokio::sync::broadcast;
+
+#[derive(Debug, Clone)]
+pub enum HarmonyComposerEvent {
+    ProjectInitializationStarted,
+    ProjectInitialized,
+    ProjectCompilationStarted { details: String },
+    ProjectCompiled,
+    ProjectCompilationFailed { details: String },
+    DeploymentStarted { target: String },
+    DeploymentCompleted { details: String },
+    Shutdown,
+}
+
+static HARMONY_COMPOSER_EVENT_BUS: Lazy<broadcast::Sender<HarmonyComposerEvent>> =
+    Lazy::new(|| {
+        // TODO: Adjust channel capacity
+        let (tx, _rx) = broadcast::channel(16);
+        tx
+    });
+
+pub fn instrument(event: HarmonyComposerEvent) {
+    HARMONY_COMPOSER_EVENT_BUS
+        .send(event)
+        .expect("couldn't send event");
+}
+
+pub async fn subscribe<F, Fut>(name: &str, mut handler: F)
+where
+    F: FnMut(HarmonyComposerEvent) -> Fut + Send + 'static,
+    Fut: Future<Output = bool> + Send,
+{
+    let mut rx = HARMONY_COMPOSER_EVENT_BUS.subscribe();
+    debug!("[{name}] Service started. Listening for events...");
+    loop {
+        match rx.recv().await {
+            Ok(event) => {
+                if !handler(event).await {
+                    debug!("[{name}] Handler requested exit.");
+                    break;
+                }
+            }
+            Err(broadcast::error::RecvError::Lagged(n)) => {
+                debug!("[{name}] Lagged behind by {n} messages.");
+            }
+            Err(_) => break,
+        }
+    }
+}

harmony_composer/src/main.rs

@@ -7,14 +7,15 @@ use bollard::secret::HostConfig;
 use cargo_metadata::{Artifact, Message, MetadataCommand};
 use clap::{Args, Parser, Subcommand};
 use futures_util::StreamExt;
-use harmony::instrumentation::{self, HarmonyEvent};
+use instrumentation::HarmonyComposerEvent;
 use log::{debug, info, log_enabled};
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::process::{Command, Stdio};
 use tokio::fs;
 
-mod cli_logger;
+mod harmony_composer_logger;
+mod instrumentation;
 
 #[derive(Parser)]
 #[command(version, about, long_about = None, flatten_help = true, propagate_version = true)]
@@ -70,14 +71,14 @@ struct AllArgs {
 #[tokio::main]
 async fn main() {
     env_logger::init();
-    let cli_logger_handle = tokio::spawn(cli_logger::init());
+    let hc_logger_handle = tokio::spawn(harmony_composer_logger::init());
     let cli_args = GlobalArgs::parse();
 
     let harmony_path = Path::new(&cli_args.harmony_path)
         .try_exists()
         .expect("couldn't check if path exists");
 
-    instrumentation::instrument(HarmonyEvent::ProjectInitializationStarted);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectInitializationStarted);
 
     let harmony_bin_path: PathBuf = match harmony_path {
         true => {
@@ -91,7 +92,7 @@ async fn main() {
         false => todo!("implement autodetect code"),
     };
 
-    instrumentation::instrument(HarmonyEvent::ProjectInitialized);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectInitialized);
 
     match cli_args.command {
         Some(command) => match command {
@@ -124,17 +125,17 @@ async fn main() {
             }
             Commands::Deploy(args) => {
                 let deploy = if args.staging {
-                    instrumentation::instrument(HarmonyEvent::DeploymentStarted {
+                    instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "staging".to_string(),
                     });
                     todo!("implement staging deployment")
                 } else if args.prod {
-                    instrumentation::instrument(HarmonyEvent::DeploymentStarted {
+                    instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "prod".to_string(),
                     });
                     todo!("implement prod deployment")
                 } else {
-                    instrumentation::instrument(HarmonyEvent::DeploymentStarted {
+                    instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "dev".to_string(),
                     });
                     Command::new(harmony_bin_path).arg("-y").arg("-a").spawn()
@@ -142,7 +143,7 @@ async fn main() {
                 .expect("failed to run harmony deploy");
 
                 let deploy_output = deploy.wait_with_output().unwrap();
-                instrumentation::instrument(HarmonyEvent::DeploymentCompleted {
+                instrumentation::instrument(HarmonyComposerEvent::DeploymentCompleted {
                     details: String::from_utf8(deploy_output.stdout).unwrap(),
                 });
             }
@@ -154,9 +155,9 @@ async fn main() {
         None => todo!("run interactively, ask for info on CLI"),
     }
 
-    instrumentation::instrument(HarmonyEvent::Shutdown);
+    instrumentation::instrument(HarmonyComposerEvent::Shutdown);
 
-    let _ = tokio::try_join!(cli_logger_handle);
+    let _ = tokio::try_join!(hc_logger_handle);
 }
 
 #[derive(Clone, Debug, clap::ValueEnum)]
@@ -195,20 +196,20 @@ async fn compile_harmony(
 
     let path = match method {
         CompileMethod::LocalCargo => {
-            instrumentation::instrument(HarmonyEvent::ProjectCompilationStarted {
+            instrumentation::instrument(HarmonyComposerEvent::ProjectCompilationStarted {
                 details: "compiling project with cargo".to_string(),
             });
             compile_cargo(platform, harmony_location).await
         }
         CompileMethod::Docker => {
-            instrumentation::instrument(HarmonyEvent::ProjectCompilationStarted {
+            instrumentation::instrument(HarmonyComposerEvent::ProjectCompilationStarted {
                 details: "compiling project with docker".to_string(),
             });
             compile_docker(platform, harmony_location).await
         }
     };
 
-    instrumentation::instrument(HarmonyEvent::ProjectCompiled);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectCompiled);
     path
 }