Merge pull request 'dev' (#6) from dev into master

Reviewed-on: http://192.168.2.61/VladislavD/runner-rs/pulls/6

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> {
-    // * if no conf -> loop and +inf getting conf from redis server
-    // let mut local = load_processes(&CONFIG_PATH);
+pub async fn get_actual_config() -> Option<Processes> {
+    // * if no local conf -> loop and +inf getting conf from redis server
+    // * 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,62 +51,104 @@ 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 = invalid_config_watcher(&mut conn);
-            let _ = save_new_config(&remote_config, CONFIG_PATH);
-            Some(remote_config)
+            let remote_config = get_remote_conf_watcher(&mut conn).await;
+            if let Some(conf) = remote_config {
+                info!("Config {} was pulled from Redis-Server. Starting...", &conf.date_of_creation);
+                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") {
-                        if len == 0 {
-                            warn!("No configuration in DB yet");
-                            None
-                        } else {
-                            let conf: Res = conn.xrevrange_count("config_stream", "+", "-", 1);
-                            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!"
-                                    );
+                    let mut conn =  conn.as_pubsub();
+                    match conn.subscribe(cont) {
+                        Ok(_) => {
+                            if conn.set_read_timeout(Some(Duration::from_millis(100))).is_err() {
+                                error!("Cannot set reading pubsub timeout and pull remote config");
                                 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)
+                            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");
                                             }
-                                    None => {
-                                        error!("Invalid configuration was pulled (PARSING_ERROR). Check configs state!");
-                                        None
+                                            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;
+                                },
                             }
-                            } else {
-                                error!("Configuration pulling from Redis stream failed. Check stream state!");
-                                None
+                        },
+                        Err(_) => {
+                            error!("Redis subscription process failed. Check Redis configuration!");
+                            return None;
                         }
                     }
-                    } else {
-                        error!("Cannot find config_stream. Check Redis-stream accessibility!");
-                        None
-                    }
                 }
                 Err(_) => {
                     error!("Redis connection attempt is failed. Check Redis configuration!");
@@ -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/options/structs.rs

@@ -84,7 +84,7 @@ pub struct FIleTriggers {
 
 ///
 ///
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize,)]
 pub struct Metrics {
     pub container_metrics : ContainerMetrics,
     pub processes_metrics : Vec<ProcessMetrics>,
@@ -103,17 +103,19 @@ impl Metrics {
 
 /// 
 /// 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ContainerMetrics {
-    pub cpu_load : f32,
-    pub ram_load : f32,
+    container_id : String,
+    cpu_load : f32,
+    ram_load : f32,
     // pub net_activity : ???
-    pub processes : Vec<String>,
+    processes : Vec<String>,
 }
 
 impl ContainerMetrics {
-    pub fn new(cpu: f32, ram: f32, subsystems: Vec<String>,) -> Self{
+    pub fn new(container_id : &str, cpu: f32, ram: f32, subsystems: Vec<String>,) -> Self{
         ContainerMetrics {
+            container_id : String::from(container_id),
             cpu_load  : cpu,
             ram_load  : ram,
             processes : subsystems,
@@ -123,22 +125,24 @@ impl ContainerMetrics {
 
 /// 
 /// 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct ProcessMetrics {
-    pub cpu_load : f32,
-    pub ram_load : f32,
+    pub process_name : String,
+    cpu_load : f32,
+    ram_load : f32,
 }
 
 impl ProcessMetrics {
-    pub fn new(cpu: f32, ram: f32) -> Self {
+    pub fn new(process_name :&str, cpu: f32, ram: f32) -> Self {
         ProcessMetrics {
+            process_name : String::from(process_name),
             cpu_load : cpu,
             ram_load : ram,
         }
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize)]
 pub struct PacketInfo {
     protocol : String,
     dst_ip : Ipv4Addr,

src/utils/hagent.rs

@@ -1,13 +1,37 @@
 // module needed to check host-agent health condition and to communicate with it
-use tokio::net::UnixStream;
-//
-// code will be here
-//
+/// asdasdasds
+use tokio::{io::Interest, net::UnixStream};
+
 async fn open_unix_socket() -> Result<UnixStream, std::io::Error> {
-    let socket = UnixStream::connect("/var/run/enode/hostagent.sock=").await?;
+    let socket = UnixStream::connect("/var/run/enode/hostagent.sock").await?;
     Ok(socket)
 }
 
+async fn ha_healthcheck(socket: &UnixStream) -> Result<(), std::io::Error >{
+    socket.ready(Interest::WRITABLE).await?;
+    if socket.writable().await.is_ok() {
+        if let Err(er) = socket.try_write(b"Hello HAgent") {
+            return Err(er);
+        }
+    } else {
+        return Err(std::io::ErrorKind::WouldBlock.into());
+    }
+    Ok(())
+}
+
+
+async fn ha_send_data(socket: &UnixStream, data: &str) -> Result<(), std::io::Error > {
+    socket.ready(Interest::WRITABLE).await?;
+    if socket.writable().await.is_ok() {
+        if let Err(er) = socket.try_write(data.as_bytes()) {
+            return Err(er);
+        }
+    } else {
+        return Err(std::io::ErrorKind::WouldBlock.into());
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod hagent_unittets {
     use super::*;
@@ -15,13 +39,28 @@ mod hagent_unittets {
     // maybe bool : true -> alive, false -> dead
     // simple request on api
     async fn hagent_healthcheck() {
-        assert!(true);
+        let sock = open_unix_socket().await;
+        assert!(sock.is_ok());
+        let sock = sock.unwrap();
+        assert!(ha_healthcheck(&sock).await.is_ok());
     }
     #[tokio::test]
-    // Result<maybe Response>
+    // --Result<maybe Response>
     // one-shot func
     async fn send_metrics_to_hagent() {
-        assert!(true);
+        use crate::options::structs::{ProcessMetrics, ContainerMetrics, Metrics};
+        
+        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;
+        assert!(sock.is_ok());
+        let sock = sock.unwrap();
+        assert!(ha_healthcheck(&sock).await.is_ok());
+        assert!(ha_send_data(&sock, &metrics).await.is_ok());
+
     }
     #[tokio::test]
     async fn open_unixsocket_test() {

src/utils/metrics.rs

@@ -2,15 +2,15 @@
 // cpu load, ram/rom load and net activity
 
 use std::sync::Mutex;
-use std::{str::FromStr, sync::Arc};
+use std::sync::Arc;
 use crate::options::structs::TrackingProcess;
-use sysinfo::{Pid, Process, System};
+use sysinfo::{Process, System};
 use tokio::join;
-use crate::utils::prcs::get_pid;
 use crate::options::structs::{ProcessMetrics, ContainerMetrics, PacketInfo};
-use pcap::{Device, Capture, Active};
-use std::net::Ipv4Addr;
-use anyhow::{Result, Ok};
+use crate::utils::get_container_id;
+// use pcap::{Device, Capture, Active};
+// use std::net::Ipv4Addr;
+// use anyhow::{Result, Ok};
 
 type PacketBuffer = Arc<Mutex<Vec<PacketInfo>>>;
 
@@ -84,7 +84,12 @@ async fn get_all_container_metrics(sys: Arc<System>, prcs: Arc<Vec<TrackingProce
         get_ram_metrics_container(sys.clone()),
         get_subsystems(prcs.clone())
     );
-    ContainerMetrics::new(metrics.0, metrics.1, metrics.2)
+    ContainerMetrics::new(
+        &get_container_id().unwrap_or(String::from("unknown")),
+        metrics.0, 
+        metrics.1, 
+        metrics.2
+    )
 }
 async fn get_cpu_metrics_container(sys: Arc<System>) -> f32 {
     sys.global_cpu_usage()
@@ -107,7 +112,11 @@ async fn get_all_metrics_process(proc: Arc<Process>, sys: Arc<System>) -> Proces
         get_cpu_metrics_process(proc.clone()),
         get_ram_metrics_process(proc.clone(), sys.clone())
     );
-    ProcessMetrics::new(metrics.0, metrics.1)
+    ProcessMetrics::new(
+        proc.name().to_str().unwrap_or("unknown"), 
+        metrics.0, 
+        metrics.1
+    )
 }
 async fn get_cpu_metrics_process(proc: Arc<Process>) -> f32 {
     proc.cpu_usage()