Cargo.lock

@@ -618,7 +618,7 @@ checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b"
 
 [[package]]
 name = "runner-rs"
-version = "0.5.5"
+version = "0.9.25"
 dependencies = [
  "anyhow",
  "chrono",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "runner-rs"
-version = "0.5.5"
+version = "0.9.25"
 edition = "2021"
 
 [profile.dev]

src/main.rs

@@ -19,7 +19,7 @@ async fn main() {
 
     // setting up redis connection \
     // then conf checks to choose the most actual \
-    let processes: Processes = get_actual_config().unwrap_or_else(|| {
+    let processes: Processes = get_actual_config().await.unwrap_or_else(|| {
         error!("No actual configuration for runner. Stopping...");
         std::process::exit(101);
     });

src/options/config.rs

@@ -1,6 +1,6 @@
 use crate::options::structs::*;
 use log::{error, info, warn};
-use redis::{Client, Commands, Connection, RedisResult};
+use redis::{Client, Connection};
 use std::fs::OpenOptions;
 use std::io::Write;
 use std::os::unix::process::CommandExt;
@@ -11,7 +11,6 @@ use tokio::time::Duration;
 
 const CONFIG_PATH: &str = "settings.json";
 
-type Res = RedisResult<Vec<Vec<(String, Vec<(String, String)>)>>>;
 // 4ever sync
 fn load_processes(json_filename: &str) -> Option<Processes> {
     if let Ok(res) = fs::read_to_string(json_filename) {
@@ -22,9 +21,9 @@ fn load_processes(json_filename: &str) -> Option<Processes> {
     None
 }
 
-pub fn get_actual_config() -> Option<Processes> {
+pub async fn get_actual_config() -> Option<Processes> {
-    // * if no conf -> loop and +inf getting conf from redis server
+    // * if no local conf -> loop and +inf getting conf from redis server
-    // let mut local = load_processes(&CONFIG_PATH);
+    // * if local conf -> once getting conf from redis server
     match load_processes(CONFIG_PATH) {
         Some(local_conf) => {
             info!(
@@ -32,6 +31,7 @@ pub fn get_actual_config() -> Option<Processes> {
                 &local_conf.date_of_creation
             );
             if let Some(remote_conf) =
+                // TODO : rework with pubsub mech 
                 once_get_remote_configuration(&format!("redis://{}/", local_conf.config_server))
             {
                 return match config_comparing(&local_conf, &remote_conf) {
@@ -51,61 +51,103 @@ pub fn get_actual_config() -> Option<Processes> {
             Some(local_conf)
         }
         None => {
-            // ?                          ? OUTSTANDING CONSTRUCTION ?
+            warn!("No local valid conf was found. Trying to pull remote one...");
             let mut conn = get_connection_watcher(&open_watcher("redis://localhost/"));
-            get_stream_info_watcher(&mut conn);
+            let remote_config = get_remote_conf_watcher(&mut conn).await;
-            let remote_config = invalid_config_watcher(&mut conn);
+            if let Some(conf) = remote_config {
-            let _ = save_new_config(&remote_config, CONFIG_PATH);
+                info!("Config {} was pulled from Redis-Server. Starting...", &conf.date_of_creation);
-            Some(remote_config)
+                let _ = save_new_config(&conf, CONFIG_PATH);
+                return Some(conf);
+            }
+            None
         }
     }
 }
+
+async fn get_remote_conf_watcher(conn : &mut Connection) -> Option<Processes> {
+    let mut conn = conn.as_pubsub();
+    let cont = crate::utils::get_container_id();
+    loop { 
+        match cont {
+            Some(ref cont) => {
+                let cont = cont.trim();
+                if conn.subscribe(cont).is_err() {
+                    // todo : delay
+                    continue;
+                }
+                match conn.get_message() {
+                    Ok(msg) => {
+                        let msg: Result<String, redis::RedisError> = msg.get_payload();
+                        if let Ok(payload) = msg {
+                            if let Some(remote) = parse_extern_config(&payload) {
+                                return Some(remote)
+                            }
+                            else {
+                                error!("Pulled invalid config, cannot start. Waiting for remote conf...");
+                            }
+                        } else {
+                            error!("Cannot get Redis message payload. Waiting for remote conf...");
+                        }
+                        // todo : delay
+                        continue;
+                    },
+                    Err(_) => {
+                        // todo : delay
+                        continue;
+                    },
+                }
+            },
+            None => {
+                error!("Cannot get container id. Returning");
+                break
+            },
+        }
+    }
+    None
+} 
 // ! once iter exec
 // ! only for situation when local isn't None (no need to fck redis server)
 fn once_get_remote_configuration(serv_info: &str) -> Option<Processes> {
+    let cont = crate::utils::get_container_id();
     match Client::open(serv_info) {
         Ok(client) => {
             match client.get_connection() {
                 Ok(mut conn) => {
-                    if let Ok(len) = conn.xlen::<&str, usize>("config_stream") {
+                    let mut conn =  conn.as_pubsub();
-                        if len == 0 {
+                    match conn.subscribe(cont) {
-                            warn!("No configuration in DB yet");
+                        Ok(_) => {
-                            None
+                            if conn.set_read_timeout(Some(Duration::from_millis(100))).is_err() {
-                        } else {
+                                error!("Cannot set reading pubsub timeout and pull remote config");
-                            let conf: Res = conn.xrevrange_count("config_stream", "+", "-", 1);
+                                return None;
-                            let config: &Vec<(String, Vec<(String, String)>)>;
-
-                            if conf.is_ok() {
-                                // guaranteed and safe unwrapping
-                                let conf = conf.unwrap();
-                                config = &conf[0];
-                                if config.is_empty() {
-                                    error!(
-                                        "Empty config was pulled. Check stream and configs state!"
-                                    );
-                                    return None;
-                                }
-                                match parse_extern_config(&config[0].1[0].1) {
-                                    Some(prcs) => {
-                                        info!(
-                                            "Config {} was pulled from Redis-Server",
-                                            &prcs.date_of_creation
-                                        );
-                                        Some(prcs)
-                                    }
-                                    None => {
-                                        error!("Invalid configuration was pulled (PARSING_ERROR). Check configs state!");
-                                        None
-                                    }
-                                }
-                            } else {
-                                error!("Configuration pulling from Redis stream failed. Check stream state!");
-                                None
                             }
+                            match conn.get_message() {
+                                Ok(msg) => {
+                                    info!("Pulled config from Redis Server");
+                                    let get_payload: Result<String, redis::RedisError> = msg.get_payload();
+                                    match get_payload {
+                                        Ok(payload) => {
+                                            let remote = parse_extern_config(&payload);
+                                            if remote.is_none() {
+                                                error!("Pulled config is invalid. Check it in Redis Server");
+                                            }
+                                            return remote;
+                                        },
+                                        Err(_) => {
+                                            error!("Cannot extract payload from new message. Check Redis Server state");
+                                            return None;
+                                        },
+                                    }
+                                },
+                                Err(_) => {
+                                    warn!("Cannot get config from Redis Server. Empty channel");
+                                    return None;
+                                },
+                            }
+                        },
+                        Err(_) => {
+                            error!("Redis subscription process failed. Check Redis configuration!");
+                            return None;
                         }
-                    } else {
-                        error!("Cannot find config_stream. Check Redis-stream accessibility!");
-                        None
                     }
                 }
                 Err(_) => {
@@ -154,47 +196,6 @@ fn get_connection_watcher(client: &Client) -> Connection {
         }
     }
 }
-fn get_stream_info_watcher(conn: &mut Connection) {
-    loop {
-        if let Ok(val) = conn.xlen::<&str, usize>("config_stream") {
-            if val != 0 {
-                info!("Redis stream is able and not empty now");
-                return;
-            }
-        }
-        error!("Configuration pulling from Redis stream failed. Check stream state! Retrying...");
-        std::thread::sleep(Duration::from_secs(4));
-    }
-}
-fn invalid_config_watcher(conn: &mut Connection) -> Processes {
-    loop {
-        if let Some(prcs) = get_remote_config(conn) {
-            info!(
-                "Got new config from Redis-Server, version - {}",
-                &prcs.date_of_creation
-            );
-            return prcs;
-        }
-        error!("Got INVALID configuration. Update config! Retrying...");
-        std::thread::sleep(Duration::from_secs(4));
-    }
-}
-
-// ! end of watchers
-
-fn get_remote_config(conn: &mut Connection) -> Option<Processes> {
-    let res: Res = conn.xrevrange_count("config_stream", "+", "-", 1);
-    if let Ok(result) = res {
-        if result.is_empty() {
-            return None;
-        }
-        let config = &result[0];
-        if !config.is_empty() {
-            return parse_extern_config(&config[0].1[0].1);
-        }
-    }
-    None
-}
 
 fn restart_main_thread() -> std::io::Result<()> {
     let current_exe = env::current_exe()?;

src/utils/hagent.rs

@@ -1,8 +1,7 @@
 // module needed to check host-agent health condition and to communicate with it
+/// asdasdasds
 use tokio::{io::Interest, net::UnixStream};
-//
+
-// code will be here
-//
 async fn open_unix_socket() -> Result<UnixStream, std::io::Error> {
     let socket = UnixStream::connect("/var/run/enode/hostagent.sock").await?;
     Ok(socket)
@@ -35,10 +34,6 @@ async fn ha_send_data(socket: &UnixStream, data: &str) -> Result<(), std::io::Er
 
 #[cfg(test)]
 mod hagent_unittets {
-    use log::info;
-
-    use crate::utils::metrics;
-
     use super::*;
     #[tokio::test]
     // maybe bool : true -> alive, false -> dead
@@ -53,9 +48,11 @@ mod hagent_unittets {
     // --Result<maybe Response>
     // one-shot func
     async fn send_metrics_to_hagent() {
-        let procm = crate::options::structs::ProcessMetrics::new("test-prc", 15.0, 5.0);
+        use crate::options::structs::{ProcessMetrics, ContainerMetrics, Metrics};
-        let contm = crate::options::structs::ContainerMetrics::new("test", 32.0, 12.0, vec![procm.process_name.clone()]);
+        
-        let metrics = crate::options::structs::Metrics::new(contm, vec![procm]);
+        let procm = ProcessMetrics::new("test-prc", 15.0, 5.0);
+        let contm = ContainerMetrics::new("test", 32.0, 12.0, vec![procm.process_name.clone()]);
+        let metrics = Metrics::new(contm, vec![procm]);
         let metrics = &serde_json::to_string_pretty(&metrics).unwrap();
 
         let sock = open_unix_socket().await;