ensure event handlers are properly subscribed on init, extract duplicated progress functions, cleanup duplication

Cargo.lock

@@ -1821,6 +1821,7 @@ dependencies = [
  "harmony",
  "harmony_tui",
  "indicatif",
+ "indicatif-log-bridge",
  "inquire",
  "lazy_static",
  "log",
@@ -1840,6 +1841,7 @@ dependencies = [
  "futures-util",
  "harmony_cli",
  "indicatif",
+ "indicatif-log-bridge",
  "lazy_static",
  "log",
  "once_cell",
@@ -2451,6 +2453,16 @@ dependencies = [
  "web-time",
 ]
 
+[[package]]
+name = "indicatif-log-bridge"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "63703cf9069b85dbe6fe26e1c5230d013dee99d3559cd3d02ba39e099ef7ab02"
+dependencies = [
+ "indicatif",
+ "log",
+]
+
 [[package]]
 name = "indoc"
 version = "2.0.6"

examples/rust/src/main.rs

@@ -13,8 +13,7 @@ use harmony_cli::cli_logger;
 
 #[tokio::main]
 async fn main() {
-    env_logger::init();
-    let cli_logger_handle = tokio::spawn(cli_logger::init());
+    let cli_logger_handle = cli_logger::init();
 
     let topology = K8sAnywhereTopology::from_env();
     let mut maestro = Maestro::initialize(Inventory::autoload(), topology)

harmony/src/domain/instrumentation.rs

@@ -2,19 +2,31 @@ use log::debug;
 use once_cell::sync::Lazy;
 use tokio::sync::broadcast;
 
+use super::interpret::InterpretStatus;
+
 #[derive(Debug, Clone)]
 pub enum HarmonyEvent {
-    PrepareTopologyStarted { name: String },
+    HarmonyStarted,
+    PrepareTopologyStarted {
+        name: String,
+    },
+    TopologyPrepared {
+        name: String,
+        status: InterpretStatus,
+    },
 }
 
 static HARMONY_EVENT_BUS: Lazy<broadcast::Sender<HarmonyEvent>> = Lazy::new(|| {
     // TODO: Adjust channel capacity
-    let (tx, _rx) = broadcast::channel(18);
+    let (tx, _rx) = broadcast::channel(100);
     tx
 });
 
-pub fn instrument(event: HarmonyEvent) {
-    HARMONY_EVENT_BUS.send(event).expect("couldn't send event");
+pub fn instrument(event: HarmonyEvent) -> Result<(), &'static str> {
+    match HARMONY_EVENT_BUS.send(event) {
+        Ok(_) => Ok(()),
+        Err(_) => Err("send error: no subscribers"),
+    }
 }
 
 pub async fn subscribe<F, Fut>(name: &str, mut handler: F)

harmony/src/domain/maestro/mod.rs

@@ -44,13 +44,22 @@ impl<T: Topology> Maestro<T> {
     pub async fn prepare_topology(&self) -> Result<Outcome, InterpretError> {
         instrumentation::instrument(HarmonyEvent::PrepareTopologyStarted {
             name: self.topology.name().to_string(),
-        });
+        })
+        .unwrap();
+
+        instrumentation::instrument(HarmonyEvent::TopologyPrepared {
+            name: self.topology.name().to_string(),
+            status: InterpretStatus::SUCCESS,
+        })
+        .unwrap();
+
         let outcome = self.topology.ensure_ready().await?;
-        info!(
-            "Topology '{}' readiness check complete: {}",
-            self.topology.name(),
-            outcome.status
-        );
+
+        instrumentation::instrument(HarmonyEvent::TopologyPrepared {
+            name: self.topology.name().to_string(),
+            status: outcome.status.clone(),
+        })
+        .unwrap();
 
         self.topology_preparation_result
             .lock()

harmony/src/domain/topology/k8s_anywhere.rs

@@ -93,9 +93,8 @@ impl K8sAnywhereTopology {
             return Err("Failed to run 'helm -version'".to_string());
         }
 
-        // Print the version output
         let version_output = String::from_utf8_lossy(&version_result.stdout);
-        println!("Helm version: {}", version_output.trim());
+        debug!("Helm version: {}", version_output.trim());
 
         Ok(())
     }
@@ -126,7 +125,7 @@ impl K8sAnywhereTopology {
         // TODO this deserves some refactoring, it is becoming a bit hard to figure out
         // be careful when making modifications here
         if k8s_anywhere_config.use_local_k3d {
-            info!("Using local k3d cluster because of use_local_k3d set to true");
+            debug!("Using local k3d cluster because of use_local_k3d set to true");
         } else {
             if let Some(kubeconfig) = &k8s_anywhere_config.kubeconfig {
                 debug!("Loading kubeconfig {kubeconfig}");

harmony/src/modules/application/rust.rs

@@ -10,7 +10,7 @@ use dockerfile_builder::Dockerfile;
 use dockerfile_builder::instruction::{CMD, COPY, ENV, EXPOSE, FROM, RUN, USER, WORKDIR};
 use dockerfile_builder::instruction_builder::CopyBuilder;
 use futures_util::StreamExt;
-use log::{debug, error, info};
+use log::{debug, error, info, log_enabled};
 use serde::Serialize;
 use tar::Archive;
 
@@ -164,10 +164,12 @@ impl RustWebapp {
 
         let docker = Docker::connect_with_socket_defaults().unwrap();
 
+        let quiet = !log_enabled!(log::Level::Debug);
+
         let build_image_options = bollard::query_parameters::BuildImageOptionsBuilder::default()
             .dockerfile("Dockerfile.harmony")
             .t(image_name)
-            .q(false)
+            .q(quiet)
             .version(bollard::query_parameters::BuilderVersion::BuilderV1)
             .platform("linux/x86_64");
 

harmony_cli/Cargo.toml

@@ -17,6 +17,7 @@ console = "0.16.0"
 indicatif = "0.18.0"
 lazy_static = "1.5.0"
 log.workspace = true
+indicatif-log-bridge = "0.2.3"
 
 
 [features]

harmony_cli/src/cli_logger.rs

@@ -1,28 +1,68 @@
 use harmony::instrumentation::{self, HarmonyEvent};
-use indicatif::ProgressBar;
-use std::sync::{Arc, Mutex};
+use indicatif::{MultiProgress, ProgressBar};
+use indicatif_log_bridge::LogWrapper;
+use std::{
+    collections::{HashMap, hash_map},
+    sync::{Arc, Mutex},
+};
 
-pub async fn init() {
-    instrumentation::subscribe("CLI Logger", {
-        let current_spinner = Arc::new(Mutex::new(None::<ProgressBar>));
+pub fn init() -> tokio::task::JoinHandle<()> {
+    configure_logger();
+    let handle = tokio::spawn(handle_events());
+
+    loop {
+        if instrumentation::instrument(HarmonyEvent::HarmonyStarted).is_ok() {
+            break;
+        }
+    }
+
+    handle
+}
+
+fn configure_logger() {
+    let logger =
+        env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).build();
+    let level = logger.filter();
+    let multi = MultiProgress::new();
+    LogWrapper::new(multi.clone(), logger).try_init().unwrap();
+    log::set_max_level(level);
+}
+
+async fn handle_events() {
+    instrumentation::subscribe("Harmony CLI Logger", {
+        let progresses: Arc<Mutex<HashMap<String, MultiProgress>>> =
+            Arc::new(Mutex::new(HashMap::new()));
+        let topology_prepare_progress = Arc::new(Mutex::new(None::<ProgressBar>));
 
         move |event| {
-            let spinner_clone = Arc::clone(&current_spinner);
+            let progresses_clone = Arc::clone(&progresses);
+            let topology_prepare_progress_clone = Arc::clone(&topology_prepare_progress);
 
             async move {
-                let mut spinner_guard = spinner_clone.lock().unwrap();
+                let mut progresses = progresses_clone.lock().unwrap();
+                let mut topology_prepare_progress = topology_prepare_progress_clone.lock().unwrap();
 
                 match event {
+                    HarmonyEvent::HarmonyStarted => {}
                     HarmonyEvent::PrepareTopologyStarted { name } => {
-                        println!(
+                        let multi_progress = crate::progress::new_section(format!(
                             "{} Preparing environment: {name}...",
-                            crate::theme::EMOJI_TOPOLOGY
-                        );
+                            crate::theme::EMOJI_TOPOLOGY,
+                        ));
+                        (*progresses).insert(name, multi_progress);
                     }
+                    HarmonyEvent::TopologyPrepared { name, status } => match status {
+                        harmony::interpret::InterpretStatus::SUCCESS => todo!(),
+                        harmony::interpret::InterpretStatus::FAILURE => todo!(),
+                        harmony::interpret::InterpretStatus::RUNNING => todo!(),
+                        harmony::interpret::InterpretStatus::QUEUED => todo!(),
+                        harmony::interpret::InterpretStatus::BLOCKED => todo!(),
+                        harmony::interpret::InterpretStatus::NOOP => todo!(),
+                    },
                 }
                 true
             }
         }
     })
-    .await
+    .await;
 }

harmony_cli/src/lib.rs

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

harmony_cli/src/progress.rs

@@ -0,0 +1,40 @@
+use std::time::Duration;
+
+use indicatif::{MultiProgress, ProgressBar};
+
+pub fn new_section(title: String) -> MultiProgress {
+    let multi_progress = MultiProgress::new();
+    let _ = multi_progress.println(title);
+
+    multi_progress
+}
+
+pub fn add_spinner(multi_progress: &MultiProgress, message: String) -> ProgressBar {
+    let progress = multi_progress.add(ProgressBar::new_spinner());
+
+    progress.set_style(crate::theme::SPINNER_STYLE.clone());
+    progress.set_message(message);
+    progress.enable_steady_tick(Duration::from_millis(100));
+
+    progress
+}
+
+pub fn success(multi_progress: &MultiProgress, progress: Option<ProgressBar>, message: String) {
+    if let Some(progress) = progress {
+        multi_progress.remove(&progress)
+    }
+
+    let progress = multi_progress.add(ProgressBar::new_spinner());
+    progress.set_style(crate::theme::SUCCESS_SPINNER_STYLE.clone());
+    progress.finish_with_message(message);
+}
+
+pub fn error(multi_progress: &MultiProgress, progress: Option<ProgressBar>, message: String) {
+    if let Some(progress) = progress {
+        multi_progress.remove(&progress)
+    }
+
+    let progress = multi_progress.add(ProgressBar::new_spinner());
+    progress.set_style(crate::theme::ERROR_SPINNER_STYLE.clone());
+    progress.finish_with_message(message);
+}

harmony_composer/Cargo.toml

@@ -20,3 +20,4 @@ console = "0.16.0"
 lazy_static = "1.5.0"
 once_cell = "1.21.3"
 harmony_cli = { path = "../harmony_cli" }
+indicatif-log-bridge = "0.2.3"

harmony_composer/src/harmony_composer_logger.rs

@@ -1,61 +1,106 @@
-use indicatif::ProgressBar;
+use indicatif::{MultiProgress, ProgressBar};
+use indicatif_log_bridge::LogWrapper;
 use log::error;
 use std::{
+    collections::HashMap,
     sync::{Arc, Mutex},
-    time::Duration,
 };
 
 use crate::instrumentation::{self, HarmonyComposerEvent};
 
-pub async fn init() {
+pub fn init() -> tokio::task::JoinHandle<()> {
+    configure_logger();
+    let handle = tokio::spawn(handle_events());
+
+    loop {
+        if instrumentation::instrument(HarmonyComposerEvent::HarmonyComposerStarted).is_ok() {
+            break;
+        }
+    }
+
+    handle
+}
+
+fn configure_logger() {
+    let logger =
+        env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).build();
+    let level = logger.filter();
+    let multi = MultiProgress::new();
+    LogWrapper::new(multi.clone(), logger).try_init().unwrap();
+    log::set_max_level(level);
+}
+
+pub async fn handle_events() {
+    const PROGRESS_SETUP: &str = "project-initialization";
+    const PROGRESS_DEPLOYMENT: &str = "deployment";
+
     instrumentation::subscribe("Harmony Composer Logger", {
-        let current_spinner = Arc::new(Mutex::new(None::<ProgressBar>));
+        let progresses: Arc<Mutex<HashMap<String, MultiProgress>>> =
+            Arc::new(Mutex::new(HashMap::new()));
+        let compilation_progress = Arc::new(Mutex::new(None::<ProgressBar>));
 
         move |event| {
-            let spinner_clone = Arc::clone(&current_spinner);
+            let progresses_clone = Arc::clone(&progresses);
+            let compilation_progress_clone = Arc::clone(&compilation_progress);
 
             async move {
-                let mut spinner_guard = spinner_clone.lock().unwrap();
+                let mut progresses_guard = progresses_clone.lock().unwrap();
+                let mut compilation_progress_guard = compilation_progress_clone.lock().unwrap();
 
                 match event {
+                    HarmonyComposerEvent::HarmonyComposerStarted => {}
                     HarmonyComposerEvent::ProjectInitializationStarted => {
-                        println!(
+                        let multi_progress = harmony_cli::progress::new_section(format!(
                             "{} Initializing Harmony project...",
-                            harmony_cli::theme::EMOJI_HARMONY
-                        );
+                            harmony_cli::theme::EMOJI_HARMONY,
+                        ));
+                        (*progresses_guard).insert(PROGRESS_SETUP.to_string(), multi_progress);
                     }
                     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);
+                        let initialization_progress =
+                            (*progresses_guard).get(PROGRESS_SETUP).unwrap();
+                        let _ = initialization_progress.clear();
+
+                        let progress =
+                            harmony_cli::progress::add_spinner(initialization_progress, details);
+                        *compilation_progress_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");
-                        }
+                        let initialization_progress =
+                            (*progresses_guard).get(PROGRESS_SETUP).unwrap();
+
+                        harmony_cli::progress::success(
+                            initialization_progress,
+                            (*compilation_progress_guard).take(),
+                            "project compiled".to_string(),
+                        );
                     }
                     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}");
-                        }
+                        let initialization_progress =
+                            (*progresses_guard).get(PROGRESS_SETUP).unwrap();
+
+                        harmony_cli::progress::error(
+                            initialization_progress,
+                            (*compilation_progress_guard).take(),
+                            "failed to compile project".to_string(),
+                        );
+
+                        error!("{details}");
                     }
                     HarmonyComposerEvent::DeploymentStarted { target } => {
-                        println!(
-                            "{} Starting deployment to {target}...\n",
+                        let multi_progress = harmony_cli::progress::new_section(format!(
+                            "{} Starting deployment to {target}...\n\n",
                             harmony_cli::theme::EMOJI_DEPLOY
-                        );
+                        ));
+                        (*progresses_guard).insert(PROGRESS_DEPLOYMENT.to_string(), multi_progress);
                     }
                     HarmonyComposerEvent::DeploymentCompleted { details } => println!("\n"),
                     HarmonyComposerEvent::Shutdown => {
-                        if let Some(progress) = spinner_guard.take() {
-                            progress.abandon();
+                        for (_, progresses) in (*progresses_guard).iter() {
+                            progresses.clear().unwrap();
                         }
+
                         return false;
                     }
                 }

harmony_composer/src/instrumentation.rs

@@ -4,6 +4,7 @@ use tokio::sync::broadcast;
 
 #[derive(Debug, Clone)]
 pub enum HarmonyComposerEvent {
+    HarmonyComposerStarted,
     ProjectInitializationStarted,
     ProjectInitialized,
     ProjectCompilationStarted { details: String },
@@ -21,10 +22,11 @@ static HARMONY_COMPOSER_EVENT_BUS: Lazy<broadcast::Sender<HarmonyComposerEvent>>
         tx
     });
 
-pub fn instrument(event: HarmonyComposerEvent) {
-    HARMONY_COMPOSER_EVENT_BUS
-        .send(event)
-        .expect("couldn't send event");
+pub fn instrument(event: HarmonyComposerEvent) -> Result<(), &'static str> {
+    match HARMONY_COMPOSER_EVENT_BUS.send(event) {
+        Ok(_) => Ok(()),
+        Err(_) => Err("send error: no subscribers"),
+    }
 }
 
 pub async fn subscribe<F, Fut>(name: &str, mut handler: F)

harmony_composer/src/main.rs

@@ -70,15 +70,14 @@ struct AllArgs {
 
 #[tokio::main]
 async fn main() {
-    env_logger::init();
-    let hc_logger_handle = tokio::spawn(harmony_composer_logger::init());
+    let hc_logger_handle = 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(HarmonyComposerEvent::ProjectInitializationStarted);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectInitializationStarted).unwrap();
 
     let harmony_bin_path: PathBuf = match harmony_path {
         true => {
@@ -92,7 +91,7 @@ async fn main() {
         false => todo!("implement autodetect code"),
     };
 
-    instrumentation::instrument(HarmonyComposerEvent::ProjectInitialized);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectInitialized).unwrap();
 
     match cli_args.command {
         Some(command) => match command {
@@ -127,17 +126,20 @@ async fn main() {
                 let deploy = if args.staging {
                     instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "staging".to_string(),
-                    });
+                    })
+                    .unwrap();
                     todo!("implement staging deployment")
                 } else if args.prod {
                     instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "prod".to_string(),
-                    });
+                    })
+                    .unwrap();
                     todo!("implement prod deployment")
                 } else {
                     instrumentation::instrument(HarmonyComposerEvent::DeploymentStarted {
                         target: "dev".to_string(),
-                    });
+                    })
+                    .unwrap();
                     Command::new(harmony_bin_path).arg("-y").arg("-a").spawn()
                 }
                 .expect("failed to run harmony deploy");
@@ -145,7 +147,8 @@ async fn main() {
                 let deploy_output = deploy.wait_with_output().unwrap();
                 instrumentation::instrument(HarmonyComposerEvent::DeploymentCompleted {
                     details: String::from_utf8(deploy_output.stdout).unwrap(),
-                });
+                })
+                .unwrap();
             }
             Commands::All(_args) => todo!(
                 "take all previous match arms and turn them into separate functions, and call them all one after the other"
@@ -155,7 +158,7 @@ async fn main() {
         None => todo!("run interactively, ask for info on CLI"),
     }
 
-    instrumentation::instrument(HarmonyComposerEvent::Shutdown);
+    instrumentation::instrument(HarmonyComposerEvent::Shutdown).unwrap();
 
     let _ = tokio::try_join!(hc_logger_handle);
 }
@@ -198,18 +201,20 @@ async fn compile_harmony(
         CompileMethod::LocalCargo => {
             instrumentation::instrument(HarmonyComposerEvent::ProjectCompilationStarted {
                 details: "compiling project with cargo".to_string(),
-            });
+            })
+            .unwrap();
             compile_cargo(platform, harmony_location).await
         }
         CompileMethod::Docker => {
             instrumentation::instrument(HarmonyComposerEvent::ProjectCompilationStarted {
                 details: "compiling project with docker".to_string(),
-            });
+            })
+            .unwrap();
             compile_docker(platform, harmony_location).await
         }
     };
 
-    instrumentation::instrument(HarmonyComposerEvent::ProjectCompiled);
+    instrumentation::instrument(HarmonyComposerEvent::ProjectCompiled).unwrap();
     path
 }