Merge pull request #40 from tsirysndr/fix/log-engine

fix(log_engine): wrap LogEngine under Mutex
tsirysndr · May 13, 2023 · c9fcda5 · c9fcda5
2 parents 5758c0b + a6c2972
commit c9fcda5
Show file tree

Hide file tree

Showing 8 changed files with 32 additions and 22 deletions.
diff --git a/src/server/logging.rs b/src/server/logging.rs
@@ -23,7 +23,7 @@ pub struct Logging {
     processes: Arc<Mutex<Vec<(Process, String)>>>,
     config_map: Arc<Mutex<HashMap<String, ConfigurationData>>>,
     project_map: Arc<Mutex<HashMap<String, String>>>,
-    log_engine: LogEngine,
+    log_engine: Arc<Mutex<LogEngine>>,
 }
 
 impl Logging {
@@ -32,7 +32,7 @@ impl Logging {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         config_map: Arc<Mutex<HashMap<String, ConfigurationData>>>,
         project_map: Arc<Mutex<HashMap<String, String>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         Self {
             superviseur,
@@ -221,7 +221,8 @@ impl LoggingService for Logging {
         let config = config_map.get(&project_id).unwrap();
 
         let query = format!("{} AND {} AND {}", config.project, service, term);
-        let result = self.log_engine.search_in_service(&query).map_err(|e| {
+        let log_engine = self.log_engine.lock().unwrap();
+        let result = log_engine.search_in_service(&query).map_err(|e| {
             tonic::Status::internal(format!("Error searching in service: {}", e.to_string()))
         })?;
         let response = SearchResponse {

diff --git a/src/server/mod.rs b/src/server/mod.rs
@@ -45,7 +45,7 @@ pub async fn exec(port: u16, serve: bool) -> Result<(), Error> {
     let cmd_rx = Arc::new(Mutex::new(cmd_rx));
     let service_graph = Arc::new(Mutex::new(vec![] as Vec<(DependencyGraph, String)>));
     let service_map = Arc::new(Mutex::new(vec![] as Vec<(HashMap<usize, Service>, String)>));
-    let log_engine = LogEngine::new();
+    let log_engine = Arc::new(Mutex::new(LogEngine::new()));
 
     let superviseur = Superviseur::new(
         cmd_rx,

diff --git a/src/superviseur/core.rs b/src/superviseur/core.rs
@@ -48,7 +48,7 @@ impl Superviseur {
         config_map: Arc<Mutex<HashMap<String, ConfigurationData>>>,
         service_graph: Arc<Mutex<Vec<(DependencyGraph, String)>>>,
         service_map: Arc<Mutex<Vec<(HashMap<usize, Service>, String)>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         let childs = Arc::new(Mutex::new(HashMap::new()));
         thread::spawn(move || {
@@ -111,7 +111,7 @@ struct SuperviseurInternal {
     config_map: Arc<Mutex<Vec<(ConfigurationData, String)>>>,
     service_graph: Arc<Mutex<Vec<(DependencyGraph, String)>>>,
     service_map: Arc<Mutex<Vec<(HashMap<usize, Service>, String)>>>,
-    log_engine: LogEngine,
+    log_engine: Arc<Mutex<LogEngine>>,
 }
 
 impl SuperviseurInternal {
@@ -125,7 +125,7 @@ impl SuperviseurInternal {
         config_map: Arc<Mutex<HashMap<String, ConfigurationData>>>,
         service_graph: Arc<Mutex<Vec<(DependencyGraph, String)>>>,
         service_map: Arc<Mutex<Vec<(HashMap<usize, Service>, String)>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         let config_map = Arc::new(Mutex::new(
             config_map

diff --git a/src/superviseur/dependencies.rs b/src/superviseur/dependencies.rs
@@ -103,7 +103,7 @@ pub enum GraphCommand {
         Arc<Mutex<Vec<(Process, String)>>>,
         Arc<Mutex<HashMap<String, i32>>>,
         mpsc::UnboundedSender<ProcessEvent>,
-        LogEngine,
+        Arc<Mutex<LogEngine>>,
     ),
     AddEdge(usize, usize),
     StartService(Service, bool),
@@ -200,7 +200,7 @@ impl DependencyGraph {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         childs: Arc<Mutex<HashMap<String, i32>>>,
         event_tx: mpsc::UnboundedSender<ProcessEvent>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> usize {
         let mut vertex = Vertex::from(service);
 

diff --git a/src/superviseur/drivers/docker/driver.rs b/src/superviseur/drivers/docker/driver.rs
@@ -44,7 +44,7 @@ pub struct Driver {
     processes: Arc<Mutex<Vec<(Process, String)>>>,
     childs: Arc<Mutex<HashMap<String, i32>>>,
     event_tx: mpsc::UnboundedSender<ProcessEvent>,
-    log_engine: logs::LogEngine,
+    log_engine: Arc<Mutex<logs::LogEngine>>,
     config: Option<DriverConfig>,
 }
 
@@ -59,7 +59,7 @@ impl Default for Driver {
             processes: Arc::new(Mutex::new(Vec::new())),
             childs: Arc::new(Mutex::new(HashMap::new())),
             event_tx,
-            log_engine: logs::LogEngine::new(),
+            log_engine: Arc::new(Mutex::new(logs::LogEngine::new())),
             config: None,
         }
     }
@@ -73,7 +73,7 @@ impl Driver {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         event_tx: mpsc::UnboundedSender<ProcessEvent>,
         childs: Arc<Mutex<HashMap<String, i32>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         let config = service
             .r#use
@@ -465,7 +465,7 @@ impl DriverPlugin for Driver {
 
 pub async fn write_logs(
     service: Service,
-    log_engine: LogEngine,
+    log_engine: Arc<Mutex<LogEngine>>,
     project: String,
     mut stream: impl Stream<Item = Result<tty::TtyChunk, shiplift::Error>> + Unpin,
 ) {
@@ -499,6 +499,7 @@ pub async fn write_logs(
                                 .timestamp(),
                         ),
                     };
+                    let log_engine = log_engine.lock().unwrap();
                     match log_engine.insert(&log) {
                         Ok(_) => {}
                         Err(e) => {
@@ -538,6 +539,7 @@ pub async fn write_logs(
                                 .timestamp(),
                         ),
                     };
+                    let log_engine = log_engine.lock().unwrap();
                     match log_engine.insert(&log) {
                         Ok(_) => {}
                         Err(e) => {

diff --git a/src/superviseur/drivers/exec/driver.rs b/src/superviseur/drivers/exec/driver.rs
@@ -36,7 +36,7 @@ pub struct Driver {
     processes: Arc<Mutex<Vec<(Process, String)>>>,
     childs: Arc<Mutex<HashMap<String, i32>>>,
     event_tx: mpsc::UnboundedSender<ProcessEvent>,
-    log_engine: logs::LogEngine,
+    log_engine: Arc<Mutex<logs::LogEngine>>,
 }
 
 impl Default for Driver {
@@ -48,7 +48,7 @@ impl Default for Driver {
             processes: Arc::new(Mutex::new(Vec::new())),
             childs: Arc::new(Mutex::new(HashMap::new())),
             event_tx,
-            log_engine: logs::LogEngine::new(),
+            log_engine: Arc::new(Mutex::new(logs::LogEngine::new())),
         }
     }
 }
@@ -60,7 +60,7 @@ impl Driver {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         event_tx: mpsc::UnboundedSender<ProcessEvent>,
         childs: Arc<Mutex<HashMap<String, i32>>>,
-        log_engine: logs::LogEngine,
+        log_engine: Arc<Mutex<logs::LogEngine>>,
     ) -> Self {
         Self {
             project,
@@ -95,6 +95,7 @@ impl Driver {
                     output: String::from("stdout"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {
@@ -128,6 +129,7 @@ impl Driver {
                     output: String::from("stderr"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {

diff --git a/src/superviseur/drivers/flox/driver.rs b/src/superviseur/drivers/flox/driver.rs
@@ -40,7 +40,7 @@ pub struct Driver {
     processes: Arc<Mutex<Vec<(Process, String)>>>,
     childs: Arc<Mutex<HashMap<String, i32>>>,
     event_tx: mpsc::UnboundedSender<ProcessEvent>,
-    log_engine: LogEngine,
+    log_engine: Arc<Mutex<LogEngine>>,
 }
 
 impl Default for Driver {
@@ -52,7 +52,7 @@ impl Default for Driver {
             processes: Arc::new(Mutex::new(Vec::new())),
             childs: Arc::new(Mutex::new(HashMap::new())),
             event_tx,
-            log_engine: LogEngine::new(),
+            log_engine: Arc::new(Mutex::new(LogEngine::new())),
         }
     }
 }
@@ -64,7 +64,7 @@ impl Driver {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         event_tx: mpsc::UnboundedSender<ProcessEvent>,
         childs: Arc<Mutex<HashMap<String, i32>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         Self {
             project,
@@ -123,6 +123,8 @@ impl Driver {
                     output: String::from("stdout"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {
@@ -156,6 +158,7 @@ impl Driver {
                     output: String::from("stderr"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {

diff --git a/src/superviseur/drivers/nix/driver.rs b/src/superviseur/drivers/nix/driver.rs
@@ -36,7 +36,7 @@ pub struct Driver {
     processes: Arc<Mutex<Vec<(Process, String)>>>,
     childs: Arc<Mutex<HashMap<String, i32>>>,
     event_tx: mpsc::UnboundedSender<ProcessEvent>,
-    log_engine: LogEngine,
+    log_engine: Arc<Mutex<LogEngine>>,
 }
 
 impl Default for Driver {
@@ -48,7 +48,7 @@ impl Default for Driver {
             processes: Arc::new(Mutex::new(Vec::new())),
             childs: Arc::new(Mutex::new(HashMap::new())),
             event_tx,
-            log_engine: LogEngine::new(),
+            log_engine: Arc::new(Mutex::new(LogEngine::new())),
         }
     }
 }
@@ -60,7 +60,7 @@ impl Driver {
         processes: Arc<Mutex<Vec<(Process, String)>>>,
         event_tx: mpsc::UnboundedSender<ProcessEvent>,
         childs: Arc<Mutex<HashMap<String, i32>>>,
-        log_engine: LogEngine,
+        log_engine: Arc<Mutex<LogEngine>>,
     ) -> Self {
         Self {
             project,
@@ -95,6 +95,7 @@ impl Driver {
                     output: String::from("stdout"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {
@@ -128,6 +129,7 @@ impl Driver {
                     output: String::from("stderr"),
                     date: tantivy::DateTime::from_timestamp_secs(chrono::Local::now().timestamp()),
                 };
+                let log_engine = log_engine.lock().unwrap();
                 match log_engine.insert(&log) {
                     Ok(_) => {}
                     Err(e) => {