From 9fb57ba3a7718b69f4e601e00be6ef318065dc9c Mon Sep 17 00:00:00 2001 From: prplV Date: Tue, 12 Nov 2024 15:24:19 +0300 Subject: [PATCH 1/2] fixed no local config mech --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/main.rs | 2 +- src/options/config.rs | 103 ++++++++++++++++++++++-------------------- src/utils/hagent.rs | 12 ++--- 5 files changed, 62 insertions(+), 59 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 426fa8a..bc70f37 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -618,7 +618,7 @@ checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b" [[package]] name = "runner-rs" -version = "0.5.5" +version = "0.9.24" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 5432f63..1216761 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "runner-rs" -version = "0.5.5" +version = "0.9.24" edition = "2021" [profile.dev] diff --git a/src/main.rs b/src/main.rs index 33fcc5a..6f161d0 100644 --- a/src/main.rs +++ b/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); }); diff --git a/src/options/config.rs b/src/options/config.rs index c678655..af233c5 100644 --- a/src/options/config.rs +++ b/src/options/config.rs @@ -22,9 +22,9 @@ fn load_processes(json_filename: &str) -> Option { None } -pub fn get_actual_config() -> Option { - // * 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 { + // * 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 +32,7 @@ pub fn get_actual_config() -> Option { &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,15 +52,60 @@ pub fn get_actual_config() -> Option { 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 { + 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 = 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 { @@ -154,47 +200,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 { - 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()?; diff --git a/src/utils/hagent.rs b/src/utils/hagent.rs index c74cb9e..3f02098 100644 --- a/src/utils/hagent.rs +++ b/src/utils/hagent.rs @@ -35,10 +35,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 +49,11 @@ mod hagent_unittets { // --Result // one-shot func async fn send_metrics_to_hagent() { - let procm = crate::options::structs::ProcessMetrics::new("test-prc", 15.0, 5.0); - 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]); + 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; From 535398e58a006c23bf8b330be719c185ba38d7d0 Mon Sep 17 00:00:00 2001 From: prplV Date: Tue, 12 Nov 2024 16:14:11 +0300 Subject: [PATCH 2/2] fixed all config error and questions --- Cargo.lock | 2 +- Cargo.toml | 2 +- src/options/config.rs | 74 ++++++++++++++++++++----------------------- src/utils/hagent.rs | 5 ++- 4 files changed, 39 insertions(+), 44 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index bc70f37..a9efa2c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -618,7 +618,7 @@ checksum = "7a66a03ae7c801facd77a29370b4faec201768915ac14a721ba36f20bc9c209b" [[package]] name = "runner-rs" -version = "0.9.24" +version = "0.9.25" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index 1216761..67d68e7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "runner-rs" -version = "0.9.24" +version = "0.9.25" edition = "2021" [profile.dev] diff --git a/src/options/config.rs b/src/options/config.rs index af233c5..7643fdd 100644 --- a/src/options/config.rs +++ b/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)>>>; // 4ever sync fn load_processes(json_filename: &str) -> Option { if let Ok(res) = fs::read_to_string(json_filename) { @@ -109,49 +108,46 @@ async fn get_remote_conf_watcher(conn : &mut Connection) -> Option { // ! 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 { + 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!" - ); - 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 + 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 conn.get_message() { + Ok(msg) => { + info!("Pulled config from Redis Server"); + let get_payload: Result = 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(_) => { diff --git a/src/utils/hagent.rs b/src/utils/hagent.rs index 3f02098..6e5910f 100644 --- a/src/utils/hagent.rs +++ b/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 { let socket = UnixStream::connect("/var/run/enode/hostagent.sock").await?; Ok(socket)