feat: add spawn named thread util

Author: danielorihuela

agent-control/src/agent_control/http_server/async_bridge.rs

@@ -1,7 +1,7 @@
 use crate::event::channel::EventConsumer;
 use crate::event::{AgentControlEvent, SubAgentEvent};
+use crate::utils::threads::spawn_named_thread;
 use crossbeam::select;
-use std::thread;
 use std::thread::JoinHandle;
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{debug, error};
@@ -14,7 +14,7 @@ pub fn run_async_sync_bridge(
     agent_control_consumer: EventConsumer<AgentControlEvent>,
     sub_agent_consumer: EventConsumer<SubAgentEvent>,
 ) -> JoinHandle<()> {
-    thread::Builder::new().name("AC sync events to/from Status Http server async events bridge".to_string()).spawn(move || loop {
+    spawn_named_thread("Async-Sync bridge", move || loop {
         select! {
             recv(&agent_control_consumer.as_ref()) -> sa_event_res => {
                 match sa_event_res {
@@ -55,5 +55,5 @@ pub fn run_async_sync_bridge(
                     }
                 }
         }
-    }).expect("thread config should be valid")
+    })
 }

agent-control/src/agent_control/http_server/runner.rs

@@ -3,9 +3,9 @@ use crate::agent_control::http_server::async_bridge::run_async_sync_bridge;
 use crate::agent_control::http_server::config::ServerConfig;
 use crate::event::channel::EventConsumer;
 use crate::event::{AgentControlEvent, SubAgentEvent};
+use crate::utils::threads::spawn_named_thread;
 use crossbeam::select;
 use std::sync::Arc;
-use std::thread;
 use std::thread::JoinHandle;
 use tokio::runtime::Runtime;
 use tokio::sync::mpsc;
@@ -55,79 +55,73 @@ impl Runner {
         sub_agent_consumer: EventConsumer<SubAgentEvent>,
         maybe_opamp_client_config: Option<OpAMPClientConfig>,
     ) -> JoinHandle<()> {
-        thread::Builder::new()
-            .name("Http server".to_string())
-            .spawn(move || {
-                // Create 2 unbounded channel to send the Agent Control and Sub Agent Sync events
-                // to the Async Status Server
-                let (async_agent_control_event_publisher, async_agent_control_event_consumer) =
-                    mpsc::unbounded_channel::<AgentControlEvent>();
-                let (async_sub_agent_event_publisher, async_sub_agent_event_consumer) =
-                    mpsc::unbounded_channel::<SubAgentEvent>();
-                // Run an OS Thread that listens to sync channel and forwards the events
-                // to an async channel
-                let bridge_join_handle = run_async_sync_bridge(
-                    async_agent_control_event_publisher,
-                    async_sub_agent_event_publisher,
-                    agent_control_consumer,
-                    sub_agent_consumer,
-                );
+        spawn_named_thread("Http server", move || {
+            // Create 2 unbounded channel to send the Agent Control and Sub Agent Sync events
+            // to the Async Status Server
+            let (async_agent_control_event_publisher, async_agent_control_event_consumer) =
+                mpsc::unbounded_channel::<AgentControlEvent>();
+            let (async_sub_agent_event_publisher, async_sub_agent_event_consumer) =
+                mpsc::unbounded_channel::<SubAgentEvent>();
+            // Run an OS Thread that listens to sync channel and forwards the events
+            // to an async channel
+            let bridge_join_handle = run_async_sync_bridge(
+                async_agent_control_event_publisher,
+                async_sub_agent_event_publisher,
+                agent_control_consumer,
+                sub_agent_consumer,
+            );
 
-                // Run the async status server
-                let _ = runtime
-                    .block_on(
-                        crate::agent_control::http_server::server::run_status_server(
-                            config.clone(),
-                            async_agent_control_event_consumer,
-                            async_sub_agent_event_consumer,
-                            maybe_opamp_client_config,
-                        ),
-                    )
-                    .inspect_err(|err| {
-                        error!(error_msg = %err, "error running status server");
-                    });
+            // Run the async status server
+            let _ = runtime
+                .block_on(
+                    crate::agent_control::http_server::server::run_status_server(
+                        config.clone(),
+                        async_agent_control_event_consumer,
+                        async_sub_agent_event_consumer,
+                        maybe_opamp_client_config,
+                    ),
+                )
+                .inspect_err(|err| {
+                    error!(error_msg = %err, "error running status server");
+                });
 
-                // Wait until the bridge is closed
-                bridge_join_handle.join().unwrap();
-            })
-            .expect("thread config should be valid")
+            // Wait until the bridge is closed
+            bridge_join_handle.join().unwrap();
+        })
     }
 
     fn spawn_noop_consumer(
         agent_control_consumer: EventConsumer<AgentControlEvent>,
         sub_agent_consumer: EventConsumer<SubAgentEvent>,
     ) -> JoinHandle<()> {
-        thread::Builder::new()
-            .name("No-action consumer".to_string())
-            .spawn(move || loop {
-                select! {
-                    recv(agent_control_consumer.as_ref()) -> agent_control_consumer_res => {
-                        match agent_control_consumer_res {
-                            Ok(_) => {}
-                            Err(err) => {
-                                debug!(
-                                    error_msg = %err,
-                                    "http server event drain processor closed"
-                                );
-                                break;
-                            }
+        spawn_named_thread("No-action consumer", move || loop {
+            select! {
+                recv(agent_control_consumer.as_ref()) -> agent_control_consumer_res => {
+                    match agent_control_consumer_res {
+                        Ok(_) => {}
+                        Err(err) => {
+                            debug!(
+                                error_msg = %err,
+                                "http server event drain processor closed"
+                            );
+                            break;
                         }
-                    },
-                    recv(sub_agent_consumer.as_ref()) -> sub_agent_consumer_res => {
-                        match sub_agent_consumer_res {
-                            Ok(_) => {}
-                            Err(err) => {
-                                debug!(
-                                    error_msg = %err,
-                                    "http server event drain processor closed"
-                                );
-                                break;
-                            }
+                    }
+                },
+                recv(sub_agent_consumer.as_ref()) -> sub_agent_consumer_res => {
+                    match sub_agent_consumer_res {
+                        Ok(_) => {}
+                        Err(err) => {
+                            debug!(
+                                error_msg = %err,
+                                "http server event drain processor closed"
+                            );
+                            break;
                         }
                     }
                 }
-            })
-            .expect("thread config should be valid")
+            }
+        })
     }
 }
 

agent-control/src/k8s/garbage_collector.rs

@@ -12,6 +12,7 @@ use crate::k8s::error::GarbageCollectorK8sError::{
 use crate::k8s::error::{GarbageCollectorK8sError, K8sError};
 use crate::k8s::Error::MissingName;
 use crate::k8s::{annotations, labels};
+use crate::utils::threads::spawn_named_thread;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
 use kube::api::TypeMeta;
 use std::{sync::Arc, thread, time::Duration};
@@ -85,20 +86,17 @@ where
         let (stop_tx, stop_rx) = pub_sub();
         let interval = self.interval;
 
-        let handle = thread::Builder::new()
-            .name("Garbage collector".to_string())
-            .spawn(move || {
-                loop {
-                    let _ = self
-                        .collect()
-                        .inspect_err(|err| warn!("executing garbage collection: {err}"));
-                    if stop_rx.is_cancelled(interval) {
-                        break;
-                    }
+        let handle = spawn_named_thread("Garbage collector", move || {
+            loop {
+                let _ = self
+                    .collect()
+                    .inspect_err(|err| warn!("executing garbage collection: {err}"));
+                if stop_rx.is_cancelled(interval) {
+                    break;
                 }
-                info!("k8s garbage collector stopped");
-            })
-            .expect("thread config should be valid");
+            }
+            info!("k8s garbage collector stopped");
+        });
 
         K8sGarbageCollectorStarted { stop_tx, handle }
     }

agent-control/src/sub_agent/health/health_checker.rs

@@ -8,7 +8,7 @@ use crate::event::SubAgentInternalEvent;
 use crate::k8s;
 use crate::sub_agent::health::with_start_time::HealthWithStartTime;
 use crate::sub_agent::supervisor::starter::SupervisorStarterError;
-use std::thread;
+use crate::utils::threads::spawn_named_thread;
 use std::time::{SystemTime, SystemTimeError};
 use tracing::{debug, error};
 
@@ -220,27 +220,24 @@ pub(crate) fn spawn_health_checker<H>(
 ) where
     H: HealthChecker + Send + 'static,
 {
-    thread::Builder::new()
-        .name("Health checker".to_string())
-        .spawn(move || loop {
-            debug!(%agent_id, "starting to check health with the configured checker");
-
-            let health = health_checker.check_health().unwrap_or_else(|err| {
-                debug!(%agent_id, last_error = %err, "the configured health check failed");
-                HealthWithStartTime::from_unhealthy(Unhealthy::from(err), sub_agent_start_time)
-            });
+    spawn_named_thread("Health checker", move || loop {
+        debug!(%agent_id, "starting to check health with the configured checker");
 
-            publish_health_event(
-                &sub_agent_internal_publisher,
-                SubAgentInternalEvent::AgentHealthInfo(health),
-            );
+        let health = health_checker.check_health().unwrap_or_else(|err| {
+            debug!(%agent_id, last_error = %err, "the configured health check failed");
+            HealthWithStartTime::from_unhealthy(Unhealthy::from(err), sub_agent_start_time)
+        });
 
-            // Check the cancellation signal
-            if cancel_signal.is_cancelled(interval.into()) {
-                break;
-            }
-        })
-        .expect("thread config should be valid");
+        publish_health_event(
+            &sub_agent_internal_publisher,
+            SubAgentInternalEvent::AgentHealthInfo(health),
+        );
+
+        // Check the cancellation signal
+        if cancel_signal.is_cancelled(interval.into()) {
+            break;
+        }
+    });
 }
 
 pub(crate) fn publish_health_event(

agent-control/src/sub_agent/k8s/supervisor.rs

@@ -15,11 +15,12 @@ use crate::sub_agent::supervisor::starter::{SupervisorStarter, SupervisorStarter
 use crate::sub_agent::supervisor::stopper::SupervisorStopper;
 use crate::sub_agent::version::k8s::checkers::K8sAgentVersionChecker;
 use crate::sub_agent::version::version_checker::spawn_version_checker;
+use crate::utils::threads::spawn_named_thread;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
 use k8s_openapi::serde_json;
 use kube::{api::DynamicObject, core::TypeMeta};
 use std::sync::Arc;
-use std::thread::{self, JoinHandle};
+use std::thread::JoinHandle;
 use std::time::Duration;
 use tracing::{debug, error, info, warn};
 
@@ -131,22 +132,18 @@ impl NotStartedSupervisorK8s {
         let k8s_client = self.k8s_client.clone();
 
         info!(%agent_id, "k8s objects supervisor started");
-        let join_handle = thread::Builder::new()
-            .name("K8s objects supervisor".to_string())
-            .spawn(move || loop {
-                // Check and apply k8s objects
-                if let Err(err) =
-                    Self::apply_resources(&agent_id, resources.iter(), k8s_client.clone())
-                {
-                    error!(%agent_id, %err, "k8s resources apply failed");
-                }
-                // Check the cancellation signal
-                if stop_consumer.is_cancelled(interval) {
-                    info!(%agent_id, "k8s objects supervisor stopped");
-                    break;
-                }
-            })
-            .expect("thread config should be valid");
+        let join_handle = spawn_named_thread("K8s objects supervisor", move || loop {
+            // Check and apply k8s objects
+            if let Err(err) = Self::apply_resources(&agent_id, resources.iter(), k8s_client.clone())
+            {
+                error!(%agent_id, %err, "k8s resources apply failed");
+            }
+            // Check the cancellation signal
+            if stop_consumer.is_cancelled(interval) {
+                info!(%agent_id, "k8s objects supervisor stopped");
+                break;
+            }
+        });
 
         (stop_publisher, join_handle)
     }

agent-control/src/sub_agent/on_host/command/logging/logger.rs

@@ -1,5 +1,6 @@
 use super::file_logger::FileLogger;
 use crate::agent_control::config::AgentID;
+use crate::utils::threads::spawn_named_thread;
 use std::{sync::mpsc::Receiver, thread::JoinHandle};
 use tracing::{debug, info};
 
@@ -14,27 +15,24 @@ impl Logger {
     where
         S: ToString + Send + 'static,
     {
-        std::thread::Builder::new()
-            .name("OnHost logger".to_string())
-            .spawn(move || {
-                match self {
-                    Self::File(file_logger, agent_id) => {
-                        // If the logger is a FileLogger, set this file logging as the default.
-                        // `_guard` needs to exist in scope to keep persisting the logs in the file
-                        let _guard = file_logger.set_file_logging();
-                        rx.iter()
-                            .for_each(|line| info!(%agent_id, "{}", line.to_string()));
-                    }
-                    Self::Stderr(agent_id) => {
-                        rx.iter()
-                            .for_each(|line| debug!(%agent_id, "{}", line.to_string()));
-                    }
-                    Self::Stdout(agent_id) => {
-                        rx.iter()
-                            .for_each(|line| debug!(%agent_id, "{}", line.to_string()));
-                    }
+        spawn_named_thread("OnHost logger", move || {
+            match self {
+                Self::File(file_logger, agent_id) => {
+                    // If the logger is a FileLogger, set this file logging as the default.
+                    // `_guard` needs to exist in scope to keep persisting the logs in the file
+                    let _guard = file_logger.set_file_logging();
+                    rx.iter()
+                        .for_each(|line| info!(%agent_id, "{}", line.to_string()));
                 }
-            })
-            .expect("thread config should be valid")
+                Self::Stderr(agent_id) => {
+                    rx.iter()
+                        .for_each(|line| debug!(%agent_id, "{}", line.to_string()));
+                }
+                Self::Stdout(agent_id) => {
+                    rx.iter()
+                        .for_each(|line| debug!(%agent_id, "{}", line.to_string()));
+                }
+            }
+        })
     }
 }