Compare commits
No commits in common. "163887d42c6f1414951ab9b9d74442967c10fb86" and "064611823ab1f81fa6319d20487cc9ceeb6711b5" have entirely different histories.
163887d42c
...
064611823a
|
|
@ -4,4 +4,3 @@
|
|||
Cargo.lock
|
||||
hagent_test.sock
|
||||
release
|
||||
*.sock
|
||||
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"dateOfCreation": "1721381809112",
|
||||
"configServer": "192.168.2.37",
|
||||
"dateOfCreation": "1721381809110",
|
||||
"configServer": "localhost",
|
||||
"processes": [
|
||||
{
|
||||
"name": "temp-process",
|
||||
|
|
|
|||
|
|
@ -36,12 +36,8 @@ async fn main() -> anyhow::Result<()>{
|
|||
preboot.clone()
|
||||
).await;
|
||||
});
|
||||
handler.push(config_module);
|
||||
|
||||
let cli_module = tokio::spawn(async move {
|
||||
let _ = init_cli_pipeline().await;
|
||||
});
|
||||
handler.push(cli_module);
|
||||
handler.push(config_module);
|
||||
|
||||
for i in handler {
|
||||
let _ = i.await;
|
||||
|
|
|
|||
|
|
@ -1,11 +1,10 @@
|
|||
use log::{error, info, warn};
|
||||
use tokio::net::{TcpListener, TcpStream, UnixStream};
|
||||
use tokio::net::{TcpListener, TcpStream};
|
||||
use anyhow::{Result as DynResult, Error};
|
||||
use tokio::time::{sleep, Duration};
|
||||
use std::{borrow::BorrowMut, fs, net::{IpAddr, Ipv4Addr}};
|
||||
use std::{borrow::BorrowMut, net::{IpAddr, Ipv4Addr}};
|
||||
// use std::io::BufReader;
|
||||
use tokio::io::{BufReader, AsyncWriteExt, AsyncBufReadExt};
|
||||
use tokio::{io::AsyncReadExt, net::UnixListener};
|
||||
use noxis_cli::Cli;
|
||||
use serde_json::from_str;
|
||||
|
||||
|
|
@ -24,20 +23,21 @@ use serde_json::from_str;
|
|||
///
|
||||
pub async fn init_cli_pipeline() -> DynResult<()> {
|
||||
match init_listener().await {
|
||||
Ok(list) => {
|
||||
info!("Successfully opened UnixListener for CLI");
|
||||
Some(list) => {
|
||||
loop {
|
||||
if let Ok((socket, _)) = list.accept().await {
|
||||
if let Ok((socket, addr)) = list.accept().await {
|
||||
// isolation
|
||||
if IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) != addr.ip() {
|
||||
warn!("Declined attempt to connect TCP-socket from {}", addr);
|
||||
continue;
|
||||
}
|
||||
process_connection(socket).await;
|
||||
}
|
||||
sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
// Ok(())
|
||||
},
|
||||
Err(er) => {
|
||||
error!("Failed to open UnixListener for CLI");
|
||||
Err(er)
|
||||
},
|
||||
None => Err(Error::msg("Addr 127.0.0.1:7753 is already in use"))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -54,20 +54,17 @@ pub async fn init_cli_pipeline() -> DynResult<()> {
|
|||
///
|
||||
/// *depends on* : `tokio::net::TcpListener`
|
||||
///
|
||||
async fn init_listener() -> anyhow::Result<UnixListener> {
|
||||
// match TcpListener::bind("127.0.0.1:7753").await {
|
||||
// Ok(listener) => {
|
||||
// info!("Runner is listening localhost:7753");
|
||||
// Some(listener)
|
||||
// },
|
||||
// Err(_) => {
|
||||
// error!("Cannot create TCP listener for CLI");
|
||||
// None
|
||||
// }
|
||||
// }
|
||||
let socket_path = "noxis-rs";
|
||||
let _ = fs::remove_file(socket_path);
|
||||
Ok(UnixListener::bind(socket_path)?)
|
||||
async fn init_listener() -> Option<TcpListener> {
|
||||
match TcpListener::bind("127.0.0.1:7753").await {
|
||||
Ok(listener) => {
|
||||
info!("Runner is listening localhost:7753");
|
||||
Some(listener)
|
||||
},
|
||||
Err(_) => {
|
||||
error!("Cannot create TCP listener for CLI");
|
||||
None
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// # Fn `process_connection`
|
||||
|
|
@ -83,10 +80,11 @@ async fn init_listener() -> anyhow::Result<UnixListener> {
|
|||
///
|
||||
/// *depends on* : `tokio::net::TcpStream`
|
||||
///
|
||||
async fn process_connection(mut stream: UnixStream) {
|
||||
async fn process_connection(mut stream: TcpStream) {
|
||||
let buf_reader = BufReader::new(stream.borrow_mut());
|
||||
let mut rqst = buf_reader.lines();
|
||||
|
||||
|
||||
|
||||
while let Ok(Some(line)) = rqst.next_line().await {
|
||||
if line.is_empty() {
|
||||
break
|
||||
|
|
@ -103,6 +101,6 @@ async fn process_connection(mut stream: UnixStream) {
|
|||
println!("{}", line);
|
||||
}
|
||||
|
||||
let response = "OK";
|
||||
let response = "HTTP/1.1 200 OK\r\nContent-Length: 13\r\nContent-Type: text/plain\r\n\r\nHello, World!";
|
||||
stream.write_all(response.as_bytes()).await.unwrap();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -119,6 +119,12 @@ pub mod v2 {
|
|||
let _ = restart_main_thread();
|
||||
},
|
||||
cli_config_option = cli_future => {
|
||||
// match cli_config_option {
|
||||
// Some(config) => {},
|
||||
// None => {
|
||||
// error!("Cli pulling new config mechanism crushed, restarting ...")
|
||||
// },
|
||||
// }
|
||||
match cli_config_option {
|
||||
Err(_) => error!("Cli pulling new config mechanism crushed, restarting ..."),
|
||||
Ok(option_config) => {
|
||||
|
|
@ -138,20 +144,22 @@ pub mod v2 {
|
|||
// TODO! futures + select! [OK]
|
||||
// TODO! tests config
|
||||
}
|
||||
pub async fn get_redis_connection(params: &str) -> Option<Connection> {
|
||||
for i in 1..=3 {
|
||||
let redis_url = format!("redis://{}/", params);
|
||||
info!("Trying to connect Redis pubsub `{}`. Attempt {}", &redis_url, i);
|
||||
if let Ok(client) = Client::open(redis_url) {
|
||||
pub async fn get_redis_connection(params: Arc<PrebootParams>) -> Option<Connection> {
|
||||
if params.no_sub {
|
||||
return None;
|
||||
}
|
||||
let mut connection_delay: u64 = 1;
|
||||
loop {
|
||||
if let Ok(client) = Client::open(format!("redis://{}/", ¶ms.remote_server_url)) {
|
||||
if let Ok(conn) = client.get_connection() {
|
||||
info!("Successfully opened Redis connection");
|
||||
return Some(conn);
|
||||
}
|
||||
}
|
||||
error!("Error with subscribing Redis stream on update. Retrying in 5 secs...");
|
||||
sleep(Duration::from_secs(5)).await;
|
||||
error!("Error with subscribing Redis stream on update. Retrying in {} secs...", connection_delay);
|
||||
sleep(Duration::from_secs(connection_delay)).await;
|
||||
connection_delay *= 2;
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
// loop checking redis pubsub
|
||||
|
|
@ -163,88 +171,80 @@ pub mod v2 {
|
|||
) -> anyhow::Result<()>{
|
||||
/*...*/
|
||||
// dbg!("start of pb");
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
|
||||
let mut tx_brd_local = tx_brd_local;
|
||||
let mut local_config = Processes::default();
|
||||
|
||||
for retry in 1..=5 {
|
||||
if !tx_brd_local.is_empty() {
|
||||
match tx_brd_local.recv().await {
|
||||
Ok(lc) => local_config = lc,
|
||||
let mut _local_config = Processes::default();
|
||||
return match get_redis_connection(params.clone()).await {
|
||||
Some(mut conn) => {
|
||||
//
|
||||
let mut pub_sub = conn.as_pubsub();
|
||||
let channel_name = get_container_id().unwrap_or(String::from("default"));
|
||||
let channel_name = channel_name.trim();
|
||||
match pub_sub.subscribe(channel_name) {
|
||||
Err(er) => {
|
||||
error!("Cannot get imported local config due to {}", &er);
|
||||
return Err(anyhow::Error::msg(
|
||||
format!("Cannot get imported local config due to {}", er))
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
match get_redis_connection(&local_config.config_server).await {
|
||||
Some(mut conn) => {
|
||||
//
|
||||
let mut pub_sub = conn.as_pubsub();
|
||||
let channel_name = get_container_id().unwrap_or(String::from("default"));
|
||||
let channel_name = channel_name.trim();
|
||||
match pub_sub.subscribe(channel_name) {
|
||||
Err(er) => {
|
||||
error!("Cannot subscribe pubsub channel due to {}", &er);
|
||||
return Err(anyhow::Error::msg(format!("Cannot subscribe pubsub channel due to {}", er)))
|
||||
},
|
||||
Ok(_) => {
|
||||
info!("Successfully subscribed to {} pubsub channel", channel_name);
|
||||
loop {
|
||||
// pubsub check
|
||||
if let Ok(msg) = pub_sub.get_message() {
|
||||
// dbg!("ok on get message");
|
||||
let payload : Result<String, _> = msg.get_payload();
|
||||
match payload {
|
||||
Err(_) => error!("Cannot read new config from Redis channel. Check network or Redis configuration "),
|
||||
Ok(payload) => {
|
||||
if let Some(remote) = parse_extern_config(&payload) {
|
||||
match config_comparing(&local_config, &remote) {
|
||||
ConfigActuality::Local => {
|
||||
warn!("Pulled new config from Redis channel, it's outdated. Ignoring ...");
|
||||
},
|
||||
ConfigActuality::Remote => {
|
||||
info!("Pulled new actual config from Redis channel, version - `{}`", remote.date_of_creation);
|
||||
// to stop watching local config file mechanism
|
||||
let _ = local_conf_tx.send(true);
|
||||
let config_path = params.config.to_str().unwrap_or("settings.json");
|
||||
error!("Cannot subscribe pubsub channel due to {}", &er);
|
||||
Err(anyhow::Error::msg(format!("Cannot subscribe pubsub channel due to {}", er)))
|
||||
},
|
||||
Ok(_) => {
|
||||
info!("Successfully subscribed to {} pubsub channel", channel_name);
|
||||
loop {
|
||||
// brd check
|
||||
// if let Ok(new_lc) = tx_brd_local.recv().await {
|
||||
|
||||
if save_new_config(&remote, &config_path).is_err() {
|
||||
error!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path);
|
||||
return Err(anyhow::Error::msg(
|
||||
format!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path)
|
||||
))
|
||||
}
|
||||
return Ok(());
|
||||
},
|
||||
}
|
||||
}
|
||||
else {
|
||||
warn!("Invalid config was pulled from Redis channel")
|
||||
}
|
||||
},
|
||||
// }
|
||||
if !tx_brd_local.is_empty() {
|
||||
match tx_brd_local.recv().await {
|
||||
Ok(lc) => _local_config = lc,
|
||||
Err(er) => {
|
||||
error!("Cannot get imported local config due to {}", &er);
|
||||
return Err(anyhow::Error::msg(
|
||||
format!("Cannot get imported local config due to {}", er))
|
||||
)
|
||||
}
|
||||
}
|
||||
// delay
|
||||
// dbg!("before sleep pubsub");
|
||||
sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
},
|
||||
}
|
||||
},
|
||||
None => {
|
||||
warn!("Cannot validly connect Redis connection. Blocking task for 20 secs and restarting tries (attempt {})", retry);
|
||||
sleep(Duration::from_secs(20)).await;
|
||||
// pubsub check
|
||||
if let Ok(msg) = pub_sub.get_message() {
|
||||
// dbg!("ok on get message");
|
||||
let payload : Result<String, _> = msg.get_payload();
|
||||
match payload {
|
||||
Err(_) => error!("Cannot read new config from Redis channel. Check network or Redis configuration "),
|
||||
Ok(payload) => {
|
||||
if let Some(remote) = parse_extern_config(&payload) {
|
||||
match config_comparing(&_local_config, &remote) {
|
||||
ConfigActuality::Local => {
|
||||
warn!("Pulled new config from Redis channel, it's outdated. Ignoring ...");
|
||||
},
|
||||
ConfigActuality::Remote => {
|
||||
info!("Pulled new actual config from Redis channel, version - `{}`", remote.date_of_creation);
|
||||
// to stop watching local config file mechanism
|
||||
let _ = local_conf_tx.send(true);
|
||||
let config_path = params.config.to_str().unwrap_or("settings.json");
|
||||
|
||||
if save_new_config(&remote, &config_path).is_err() {
|
||||
error!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path);
|
||||
return Err(anyhow::Error::msg(
|
||||
format!("Error with saving new config to {}. Stopping pubsub mechanism...", config_path)
|
||||
))
|
||||
}
|
||||
return Ok(());
|
||||
},
|
||||
}
|
||||
}
|
||||
else {
|
||||
warn!("Invalid config was pulled from Redis channel")
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
// delay
|
||||
// dbg!("before sleep pubsub");
|
||||
sleep(Duration::from_millis(500)).await;
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
},
|
||||
None => Err(anyhow::Error::msg("Cannot create Redis connection"))
|
||||
}
|
||||
error!("End of retries. Stopping pubsub...");
|
||||
return Err(anyhow::Error::msg(
|
||||
format!("End of retries. Stopping pubsub...")
|
||||
))
|
||||
}
|
||||
|
||||
//
|
||||
|
|
@ -373,7 +373,7 @@ pub mod v2 {
|
|||
to_local_tx: OneShotSender<bool>
|
||||
) -> Option<Processes> {
|
||||
/* match awaits til channel*/
|
||||
// dbg!("start of cli");
|
||||
dbg!("start of cli");
|
||||
match cli_oneshot.await {
|
||||
Ok(config_from_cli) => {
|
||||
info!("New actual config `{}` from CLI was pulled. Saving and restaring ...", &config_from_cli.date_of_creation);
|
||||
|
|
|
|||
Loading…
Reference in New Issue