NationTech/harmony
6
0
Fork 1
Code Issues 32 Pull Requests 7 Actions 7 Packages Projects Releases 1 Wiki Activity

make instrumentation sync instead of async to avoid concurrency issues

Browse Source
This commit is contained in:
Ian Letourneau 2025-08-29 06:03:59 -04:00
parent 78b80c2169
commit 7bb3602ab8
8 changed files with 228 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
harmony/src/domain/instrumentation.rs
View File

@ -1,6 +1,5 @@
use log::debug;
use once_cell::sync::Lazy;
use tokio::sync::broadcast;
use std::{collections::HashMap, sync::Mutex};
use crate::modules::application::ApplicationFeatureStatus;
@ -40,43 +39,43 @@ pub enum HarmonyEvent {
},
}
static HARMONY_EVENT_BUS: Lazy<broadcast::Sender<HarmonyEvent>> = Lazy::new(|| {
// TODO: Adjust channel capacity
let (tx, _rx) = broadcast::channel(100);
tx
});
type Subscriber = Box<dyn Fn(&HarmonyEvent) + Send + Sync>;
pub fn instrument(event: HarmonyEvent) -> Result<(), &'static str> {
if cfg!(any(test, feature = "testing")) {
let _ = event; // Suppress the "unused variable" warning for `event`
Ok(())
} else {
match HARMONY_EVENT_BUS.send(event) {
Ok(_) => Ok(()),
Err(_) => Err("send error: no subscribers"),
}
}
}
static SUBSCRIBERS: Lazy<Mutex<HashMap<String, Subscriber>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
pub async fn subscribe<F, Fut>(name: &str, mut handler: F)
/// Subscribes a listener to all instrumentation events.
///
/// Simply provide a unique name and a closure to run when an event happens.
///
/// # Example
/// ```
/// instrumentation::subscribe("my_logger", |event| {
/// println!("Event occurred: {:?}", event);
/// });
/// ```
pub fn subscribe<F>(name: &str, callback: F)
where
F: FnMut(HarmonyEvent) -> Fut + Send + 'static,
Fut: Future<Output = bool> + Send,
F: Fn(&HarmonyEvent) + Send + Sync + 'static,
{
let mut rx = HARMONY_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,
}
}
let mut subs = SUBSCRIBERS.lock().unwrap();
subs.insert(name.to_string(), Box::new(callback));
}
/// Instruments an event, notifying all subscribers.
///
/// This will call every closure that was registered with `subscribe`.
///
/// # Example
/// ```
/// instrumentation::instrument(HarmonyEvent::HarmonyStarted);
/// ```
pub fn instrument(event: HarmonyEvent) -> Result<(), &'static str> {
let subs = SUBSCRIBERS.lock().unwrap();
for callback in subs.values() {
callback(&event);
}
Ok(())
}

1
harmony/src/domain/maestro/mod.rs
View File

@ -74,6 +74,7 @@ impl<T: Topology> Maestro<T> {
fn is_topology_initialized(&self) -> bool {
self.topology_state.status == TopologyStatus::Success
|| self.topology_state.status == TopologyStatus::Noop
}
pub async fn interpret(&self, score: Box<dyn Score<T>>) -> Result<Outcome, InterpretError> {

45
harmony_cli/src/cli_logger.rs
View File

@ -7,19 +7,11 @@ use harmony::{
};
use log::{error, info, log_enabled};
use std::io::Write;
use std::sync::{Arc, Mutex};
use std::sync::Mutex;
pub fn init() -> tokio::task::JoinHandle<()> {
pub fn init() {
configure_logger();
let handle = tokio::spawn(handle_events());
loop {
if instrumentation::instrument(HarmonyEvent::HarmonyStarted).is_ok() {
break;
}
}
handle
handle_events();
}
fn configure_logger() {
@ -86,16 +78,12 @@ fn configure_logger() {
.init();
}
async fn handle_events() {
let preparing_topology = Arc::new(Mutex::new(false));
let current_score: Arc<Mutex<Option<String>>> = Arc::new(Mutex::new(None));
fn handle_events() {
let preparing_topology = Mutex::new(false);
let current_score: Mutex<Option<String>> = Mutex::new(None);
instrumentation::subscribe("Harmony CLI Logger", {
move |event| {
let preparing_topology = Arc::clone(&preparing_topology);
let current_score = Arc::clone(&current_score);
async move {
let mut preparing_topology = preparing_topology.lock().unwrap();
let mut current_score = current_score.lock().unwrap();
@ -104,7 +92,6 @@ async fn handle_events() {
HarmonyEvent::HarmonyFinished => {
let emoji = crate::theme::EMOJI_HARMONY.to_string();
info!(emoji = emoji.as_str(); "Harmony completed");
return false;
}
HarmonyEvent::TopologyStateChanged {
topology,
@ -113,7 +100,10 @@ async fn handle_events() {
} => match status {
TopologyStatus::Queued => {}
TopologyStatus::Preparing => {
let emoji = format!("{}", style(crate::theme::EMOJI_TOPOLOGY.to_string()).yellow());
let emoji = format!(
"{}",
style(crate::theme::EMOJI_TOPOLOGY.to_string()).yellow()
);
info!(emoji = emoji.as_str(); "Preparing environment: {topology}...");
(*preparing_topology) = true;
}
@ -158,7 +148,7 @@ async fn handle_events() {
score,
outcome,
} => {
if current_score.is_some() && current_score.clone().unwrap() == score {
if current_score.is_some() && &current_score.clone().unwrap() == score {
(*current_score) = None;
}
@ -175,7 +165,7 @@ async fn handle_events() {
}
},
Err(err) => {
error!(status = "failed"; "{}", err);
error!(status = "failed"; "{err}");
}
}
}
@ -186,19 +176,16 @@ async fn handle_events() {
status,
} => match status {
ApplicationFeatureStatus::Installing => {
info!("Installing feature '{}' for '{}'...", feature, application);
info!("Installing feature '{feature}' for '{application}'...");
}
ApplicationFeatureStatus::Installed => {
info!(status = "finished"; "Feature '{}' installed", feature);
info!(status = "finished"; "Feature '{feature}' installed");
}
ApplicationFeatureStatus::Failed { details } => {
error!(status = "failed"; "Feature '{}' installation failed: {}", feature, details);
error!(status = "failed"; "Feature '{feature}' installation failed: {details}");
}
},
}
true
}
}
})
.await;
});
}

3
harmony_cli/src/lib.rs
View File

@ -115,7 +115,7 @@ pub async fn run_cli<T: Topology + Send + Sync + 'static>(
scores: Vec<Box<dyn Score<T>>>,
args: Args,
) -> Result<(), Box<dyn std::error::Error>> {
let cli_logger_handle = cli_logger::init();
cli_logger::init();
let mut maestro = Maestro::initialize(inventory, topology).await.unwrap();
maestro.register_all(scores);
@ -123,7 +123,6 @@ pub async fn run_cli<T: Topology + Send + Sync + 'static>(
let result = init(maestro, args).await;
instrumentation::instrument(instrumentation::HarmonyEvent::HarmonyFinished).unwrap();
let _ = tokio::try_join!(cli_logger_handle);
result
}

40
harmony_composer/src/harmony_composer_logger.rs
View File

@ -1,39 +1,26 @@
use harmony_cli::progress::{IndicatifProgressTracker, ProgressTracker};
use indicatif::MultiProgress;
use std::sync::Arc;
use crate::instrumentation::{self, HarmonyComposerEvent};
pub fn init() -> tokio::task::JoinHandle<()> {
pub fn init() {
configure_logger();
let handle = tokio::spawn(handle_events());
loop {
if instrumentation::instrument(HarmonyComposerEvent::HarmonyComposerStarted).is_ok() {
break;
}
}
handle
handle_events();
}
fn configure_logger() {
env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")).build();
}
pub async fn handle_events() {
let progress_tracker = Arc::new(IndicatifProgressTracker::new(MultiProgress::new()));
pub fn handle_events() {
let progress_tracker = IndicatifProgressTracker::new(MultiProgress::new());
const SETUP_SECTION: &str = "project-initialization";
const COMPILTATION_TASK: &str = "compilation";
const PROGRESS_DEPLOYMENT: &str = "deployment";
instrumentation::subscribe("Harmony Composer Logger", {
move |event| {
let progress_tracker = Arc::clone(&progress_tracker);
async move {
match event {
move |event| match event {
HarmonyComposerEvent::HarmonyComposerStarted => {}
HarmonyComposerEvent::ProjectInitializationStarted => {
progress_tracker.add_section(
@ -46,13 +33,16 @@ pub async fn handle_events() {
}
HarmonyComposerEvent::ProjectInitialized => {}
HarmonyComposerEvent::ProjectCompilationStarted { details } => {
progress_tracker.add_task(SETUP_SECTION, COMPILTATION_TASK, &details);
progress_tracker.add_task(SETUP_SECTION, COMPILTATION_TASK, details);
}
HarmonyComposerEvent::ProjectCompiled => {
progress_tracker.finish_task(COMPILTATION_TASK, "project compiled");
}
HarmonyComposerEvent::ProjectCompilationFailed { details } => {
progress_tracker.fail_task(COMPILTATION_TASK, &format!("failed to compile project:\n{details}"));
progress_tracker.fail_task(
COMPILTATION_TASK,
&format!("failed to compile project:\n{details}"),
);
}
HarmonyComposerEvent::DeploymentStarted { target, profile } => {
progress_tracker.add_section(
@ -68,15 +58,9 @@ pub async fn handle_events() {
}
HarmonyComposerEvent::DeploymentFailed { details } => {
progress_tracker.add_task(PROGRESS_DEPLOYMENT, "deployment-failed", "");
progress_tracker.fail_task("deployment-failed", &details);
},
HarmonyComposerEvent::Shutdown => {
return false;
}
}
true
progress_tracker.fail_task("deployment-failed", details);
}
HarmonyComposerEvent::Shutdown => {}
}
})
.await
}

78
harmony_composer/src/instrumentation.rs
View File

@ -1,6 +1,5 @@
use log::debug;
use once_cell::sync::Lazy;
use tokio::sync::broadcast;
use std::{collections::HashMap, sync::Mutex};
use crate::{HarmonyProfile, HarmonyTarget};
@ -27,48 +26,43 @@ pub enum HarmonyComposerEvent {
Shutdown,
}
static HARMONY_COMPOSER_EVENT_BUS: Lazy<broadcast::Sender<HarmonyComposerEvent>> =
Lazy::new(|| {
// TODO: Adjust channel capacity
let (tx, _rx) = broadcast::channel(16);
tx
});
type Subscriber = Box<dyn Fn(&HarmonyComposerEvent) + Send + Sync>;
pub fn instrument(event: HarmonyComposerEvent) -> Result<(), &'static str> {
#[cfg(not(test))]
{
match HARMONY_COMPOSER_EVENT_BUS.send(event) {
Ok(_) => Ok(()),
Err(_) => Err("send error: no subscribers"),
}
}
static SUBSCRIBERS: Lazy<Mutex<HashMap<String, Subscriber>>> =
Lazy::new(|| Mutex::new(HashMap::new()));
#[cfg(test)]
{
let _ = event; // Suppress the "unused variable" warning for `event`
Ok(())
}
}
pub async fn subscribe<F, Fut>(name: &str, mut handler: F)
/// Subscribes a listener to all instrumentation events.
///
/// Simply provide a unique name and a closure to run when an event happens.
///
/// # Example
/// ```
/// instrumentation::subscribe("my_logger", |event| {
/// println!("Event occurred: {:?}", event);
/// });
/// ```
pub fn subscribe<F>(name: &str, callback: F)
where
F: FnMut(HarmonyComposerEvent) -> Fut + Send + 'static,
Fut: Future<Output = bool> + Send,
F: Fn(&HarmonyComposerEvent) + Send + Sync + 'static,
{
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,
}
}
let mut subs = SUBSCRIBERS.lock().unwrap();
subs.insert(name.to_string(), Box::new(callback));
}
/// Instruments an event, notifying all subscribers.
///
/// This will call every closure that was registered with `subscribe`.
///
/// # Example
/// ```
/// instrumentation::instrument(HarmonyEvent::HarmonyStarted);
/// ```
pub fn instrument(event: HarmonyComposerEvent) -> Result<(), &'static str> {
let subs = SUBSCRIBERS.lock().unwrap();
for callback in subs.values() {
callback(&event);
}
Ok(())
}

4
harmony_composer/src/main.rs
View File

@ -99,7 +99,7 @@ impl std::fmt::Display for HarmonyProfile {
#[tokio::main]
async fn main() {
let hc_logger_handle = harmony_composer_logger::init();
harmony_composer_logger::init();
let cli_args = GlobalArgs::parse();
let harmony_path = Path::new(&cli_args.harmony_path)
@ -199,8 +199,6 @@ async fn main() {
}
instrumentation::instrument(HarmonyComposerEvent::Shutdown).unwrap();
let _ = tokio::try_join!(hc_logger_handle);
}
#[derive(Clone, Debug, clap::ValueEnum)]

10
harmony_tui/src/lib.rs
View File

@ -94,13 +94,9 @@ async fn init_instrumentation() -> tokio::task::JoinHandle<()> {
}
async fn handle_harmony_events() {
instrumentation::subscribe("Harmony TUI Logger", async |event| {
if let HarmonyEvent::HarmonyFinished = event {
return false;
};
true
})
.await;
instrumentation::subscribe("Harmony TUI Logger", |_| {
// TODO: Display events in the TUI
});
}
pub struct HarmonyTUI<T: Topology> {