Compare commits

..

No commits in common. "f3e9cb92df364a6a0f59762456a5dc5c74a47bea" and "bda476271343f0e02503bc5477b7cf9b6d51822a" have entirely different histories.

9 changed files with 23 additions and 244 deletions

View File

@ -1,6 +1,6 @@
[package] [package]
name = "runner-rs" name = "runner-rs"
version = "0.10.11" version = "0.9.25"
edition = "2021" edition = "2021"
[profile.dev] [profile.dev]
@ -9,10 +9,10 @@ debug = true
[profile.test] [profile.test]
debug = false debug = false
[dependencies] [dependencies]
anyhow = "1.0.93" anyhow = "1.0.93"
chrono = "0.4.38" chrono = "0.4.38"
clap = { version = "4.5.21", features = ["derive"] }
env_logger = "0.11.3" env_logger = "0.11.3"
inotify = "0.10.2" inotify = "0.10.2"
log = "0.4.22" log = "0.4.22"

View File

@ -4,12 +4,12 @@
"processes": [ "processes": [
{ {
"name": "temp-process", "name": "temp-process",
"path": "./temp-process", "path": "/usr/src/kii/temp-process",
"dependencies": { "dependencies": {
"files": [ "files": [
{ {
"filename": "dep-file", "filename": "dep-file",
"src": "./tests/examples/", "src": "/usr/src/kii/tests/examples/",
"triggers": { "triggers": {
"onDelete": "stop", "onDelete": "stop",
"onChange": "stay" "onChange": "stay"

View File

@ -1,27 +1,18 @@
mod options; mod options;
mod utils; mod utils;
use clap::Parser;
use log::{error, info}; use log::{error, info};
use options::{config::*, preboot}; use options::config::*;
use options::logger::setup_logger; use options::logger::setup_logger;
use options::signals::set_valid_destructor; use options::signals::set_valid_destructor;
use options::structs::Processes; use options::structs::*;
use std::sync::Arc; use std::sync::Arc;
use std::time::Duration; use std::time::Duration;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use utils::*; use utils::*;
#[allow(unused_imports)]
use options::preboot::PrebootParams;
#[tokio::main(flavor = "multi_thread")] #[tokio::main(flavor = "multi_thread")]
async fn main() { async fn main() {
let preboot = PrebootParams::parse().validate();
if let Err(_) = preboot {
return;
}
let _ = setup_logger(); let _ = setup_logger();
info!("Runner is configurating..."); info!("Runner is configurating...");

View File

@ -4,4 +4,3 @@ pub mod config;
pub mod logger; pub mod logger;
pub mod signals; pub mod signals;
pub mod structs; pub mod structs;
pub mod preboot;

View File

@ -1,196 +0,0 @@
// module to handle pre-boot params of the monitor
use anyhow::{Result, Ok, Error};
use clap::Parser;
use std::path::PathBuf;
#[derive(clap::ValueEnum, Debug, Clone)]
enum MetricsPrebootParams {
Full,
System,
Processes,
Net,
None,
}
impl std::fmt::Display for MetricsPrebootParams {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
match self {
MetricsPrebootParams::Full => write!(f, "full"),
MetricsPrebootParams::System => write!(f, "system"),
MetricsPrebootParams::Processes => write!(f, "processes"),
MetricsPrebootParams::Net => write!(f, "net"),
MetricsPrebootParams::None => write!(f, "none"),
}
}
}
#[derive(Debug, Parser)]
pub struct PrebootParams {
// actions
#[arg(
long = "no-hagent",
action,
conflicts_with="socket_path",
help="To disable work with host-agent"
)]
pub no_hostagent : bool,
#[arg(
long = "no-logs",
action,
conflicts_with="log_to",
help="To disable logs"
)]
pub no_logs: bool,
#[arg(
long = "refresh-logs",
action,
conflicts_with="no_logs",
help="To clear logs directory"
)]
pub refresh_logs : bool,
#[arg(
long = "no-remote-config",
action,
help="To disable work with remote config server",
conflicts_with="no_sub")]
pub no_remote_config : bool,
#[arg(
long = "no-sub",
action,
help="To disable subscription mechanism",
conflicts_with="no_remote_config")]
pub no_sub : bool,
// params (socket_path, log_to, remote_server_url, config)
#[arg(
long = "socket-path",
default_value="/var/run/enode/hostagent.sock",
conflicts_with="no_hostagent",
help="To set .sock file's path used in communication with host-agent"
)]
socket_path : PathBuf,
#[arg(
long = "log-to",
default_value="./",
conflicts_with="no_logs",
help="To set a path to logs directory"
)]
log_to : PathBuf,
#[arg(
long = "remote-server-url",
default_value="redis://localhost",
conflicts_with="no_remote_config",
help = "To set url of remote config server using in remote config pulling mechanism"
)]
remote_server_url : String,
#[arg(
long = "config",
short,
default_value="settings.json",
help="To set local config file path"
)]
config : PathBuf,
// value enum params (metrics)
#[arg(
long = "metrics",
short,
default_value_t=MetricsPrebootParams::Full,
help="To set metrics grubbing mode"
)]
metrics: MetricsPrebootParams,
}
impl PrebootParams {
pub fn validate(self) -> Result<Self> {
if !self.socket_path.exists() {
eprintln!("Socket-file {} doesn't exist. Cannot start", &self.socket_path.display());
return Err(Error::msg("Socket-file Not Found"));
}
// existing log dir
if !self.log_to.exists() {
eprintln!("Log directory {} doesn't exist", &self.log_to.display());
return Err(Error::msg("Log Directory Not Found. Cannot start"));
}
// existing sock file
if !self.config.exists() {
eprintln!("Local config file {} doesn't exist", &self.config.display());
return Err(Error::msg("Local Config Not Found. Cannot start"));
}
Ok(self)
}
}
// unit tests of preboot params parsing mech
#[cfg(test)]
mod preboot_unitests{
use super::*;
#[test]
fn parsing_zero_args() {
assert!(PrebootParams::try_parse_from(vec!["runner-rs"]).is_ok())
}
#[test]
fn parsing_hagent_valid_args() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--socket-path", "/path/to/socket"
]).is_ok())
}
#[test]
fn parsing_hagent_invalid_args() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--socket-path", "/path/to/socket",
"--no-hagent"
]).is_err())
}
#[test]
fn parsing_log_valid_args() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--log-to", "/path/to/log/dir"
]).is_ok())
}
#[test]
fn parsing_log_invalid_args() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--log-to /path/to/log/dir",
"--no-logs"
]).is_err())
}
#[test]
fn parsing_config_valid_args() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--no-sub",
"--remote-server-url", "redis://127.0.0.1"
]).is_ok())
}
#[test]
fn parsing_config_invalid_args_noremote_nosub() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--no-remote-config", "--no-sub"
]).is_err())
}
#[test]
fn parsing_config_invalid_args_noremote_remoteurl() {
assert!(PrebootParams::try_parse_from(vec![
"runner-rs",
"--no-remote-config",
"--remote-server-url", "redis://127.0.0.1"
]).is_err())
}
#[test]
fn parsing_metrics_args_using_value_enum() {
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "full"]).is_ok());
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "system"]).is_ok());
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "processes"]).is_ok());
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "net"]).is_ok());
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "none"]).is_ok());
assert!(PrebootParams::try_parse_from(vec!["runner-rs", "--metrics", "unusual_value"]).is_err());
}
}

View File

@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize};
pub enum CustomError { pub enum CustomError {
Fatal, Fatal,
} }
#[derive(Debug, PartialEq)] #[derive(Debug, PartialEq)]
pub enum ConfigActuality { pub enum ConfigActuality {
Local, Local,

View File

@ -1,10 +1,6 @@
//
// module needed to check host-agent health condition and to communicate with it // module needed to check host-agent health condition and to communicate with it
//
use tokio::{io::Interest, net::UnixStream}; use tokio::{io::Interest, net::UnixStream};
use anyhow::{Ok, Result, Error}; use anyhow::{Ok, Result, Error};
// to kill lint bug
#[allow(unused_imports)]
use tokio::net::UnixListener; use tokio::net::UnixListener;
/// # Fn `open_unix_socket` /// # Fn `open_unix_socket`
@ -20,7 +16,6 @@ use tokio::net::UnixListener;
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn open_unix_socket(sock_path: &str) -> Result<UnixStream, std::io::Error> { async fn open_unix_socket(sock_path: &str) -> Result<UnixStream, std::io::Error> {
// "/var/run/enode/hostagent.sock" // "/var/run/enode/hostagent.sock"
UnixStream::connect(sock_path).await UnixStream::connect(sock_path).await
@ -39,7 +34,6 @@ async fn open_unix_socket(sock_path: &str) -> Result<UnixStream, std::io::Error>
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn ha_healthcheck(socket: &UnixStream) -> Result<(), Error> { async fn ha_healthcheck(socket: &UnixStream) -> Result<(), Error> {
socket.ready(Interest::WRITABLE).await?; socket.ready(Interest::WRITABLE).await?;
socket.writable().await?; socket.writable().await?;
@ -60,7 +54,6 @@ async fn ha_healthcheck(socket: &UnixStream) -> Result<(), Error> {
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn ha_send_data(socket: &UnixStream, data: &str) -> Result<(), Error > { async fn ha_send_data(socket: &UnixStream, data: &str) -> Result<(), Error > {
socket.ready(Interest::WRITABLE).await?; socket.ready(Interest::WRITABLE).await?;
socket.writable().await?; socket.writable().await?;
@ -77,20 +70,20 @@ mod hagent_unittets {
let _ = std::fs::remove_file(TEST_SOCKET); let _ = std::fs::remove_file(TEST_SOCKET);
UnixListener::bind(TEST_SOCKET).unwrap() UnixListener::bind(TEST_SOCKET).unwrap()
} }
// #[tokio::test] #[tokio::test]
// // maybe bool : true -> alive, false -> dead // maybe bool : true -> alive, false -> dead
// // simple request on api // simple request on api
// async fn hagent_healthcheck() { async fn hagent_healthcheck() {
// let _ = init_listener().await; let mut list = init_listener().await;
// let sock = open_unix_socket(TEST_SOCKET).await; let sock = open_unix_socket(TEST_SOCKET).await;
// assert!(sock.is_ok()); assert!(sock.is_ok());
// let sock = sock.unwrap(); let mut sock = sock.unwrap();
// assert!(ha_healthcheck(&sock).await.is_ok()); assert!(ha_healthcheck(&sock).await.is_ok());
// } }
#[tokio::test] #[tokio::test]
// --Result<maybe Response> // --Result<maybe Response>
// one-shot func // one-shot func
async fn hagent_communication_test() { async fn send_metrics_to_hagent() {
use crate::options::structs::{ProcessMetrics, ContainerMetrics, Metrics}; use crate::options::structs::{ProcessMetrics, ContainerMetrics, Metrics};
let procm = ProcessMetrics::new("test-prc", 15.0, 5.0); let procm = ProcessMetrics::new("test-prc", 15.0, 5.0);
@ -98,17 +91,17 @@ mod hagent_unittets {
let metrics = Metrics::new(contm, vec![procm]); let metrics = Metrics::new(contm, vec![procm]);
let metrics = &serde_json::to_string_pretty(&metrics).unwrap(); let metrics = &serde_json::to_string_pretty(&metrics).unwrap();
#[allow(unused_mut)] let mut list = init_listener().await;
let mut _list = init_listener().await;
let sock = open_unix_socket(TEST_SOCKET).await; let sock = open_unix_socket(TEST_SOCKET).await;
assert!(sock.is_ok()); assert!(sock.is_ok());
let sock = sock.unwrap(); let mut sock = sock.unwrap();
assert!(ha_healthcheck(&sock).await.is_ok()); assert!(ha_healthcheck(&sock).await.is_ok());
assert!(ha_send_data(&sock, &metrics).await.is_ok()); assert!(ha_send_data(&sock, &metrics).await.is_ok());
} }
#[tokio::test] #[tokio::test]
async fn open_unixsocket_test() { async fn open_unixsocket_test() {
assert!(open_unix_socket("non/valid/socket/file.sock").await.is_err()); let mut list = init_listener().await;
assert!(open_unix_socket(TEST_SOCKET).await.is_ok());
} }
} }

View File

@ -27,7 +27,6 @@ use crate::utils::get_container_id;
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
pub async fn init_metrics_grubber() { pub async fn init_metrics_grubber() {
let mut system = System::new(); let mut system = System::new();
// let mut buffer: Vec<PacketInfo> = vec![]; // let mut buffer: Vec<PacketInfo> = vec![];
@ -40,8 +39,6 @@ pub async fn init_metrics_grubber() {
// let _ = capture_packets(shared_buf.clone()).await; // let _ = capture_packets(shared_buf.clone()).await;
} }
#[allow(dead_code)]
#[allow(unused_variables)]
async fn gather_metrics(proc: Arc<Process>) { async fn gather_metrics(proc: Arc<Process>) {
} }
@ -95,7 +92,6 @@ async fn gather_metrics(proc: Arc<Process>) {
/// ///
/// *depends on* : `TrackingProcess` /// *depends on* : `TrackingProcess`
/// ///
#[allow(dead_code)]
async fn get_all_container_metrics(sys: Arc<System>, prcs: Arc<Vec<TrackingProcess>>) -> ContainerMetrics { async fn get_all_container_metrics(sys: Arc<System>, prcs: Arc<Vec<TrackingProcess>>) -> ContainerMetrics {
let metrics = join!( let metrics = join!(
get_cpu_metrics_container(sys.clone()), get_cpu_metrics_container(sys.clone()),
@ -123,7 +119,6 @@ async fn get_all_container_metrics(sys: Arc<System>, prcs: Arc<Vec<TrackingProce
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn get_cpu_metrics_container(sys: Arc<System>) -> f32 { async fn get_cpu_metrics_container(sys: Arc<System>) -> f32 {
sys.global_cpu_usage() sys.global_cpu_usage()
} }
@ -141,7 +136,6 @@ async fn get_cpu_metrics_container(sys: Arc<System>) -> f32 {
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn get_ram_metrics_container(sys: Arc<System>) -> f32 { async fn get_ram_metrics_container(sys: Arc<System>) -> f32 {
(sys.used_memory() / sys.total_memory()) as f32 * 100.0 (sys.used_memory() / sys.total_memory()) as f32 * 100.0
} }
@ -162,7 +156,6 @@ async fn get_ram_metrics_container(sys: Arc<System>) -> f32 {
/// ///
/// *depends on* : `TrackingProcess` /// *depends on* : `TrackingProcess`
/// ///
#[allow(dead_code)]
async fn get_subsystems(prcs: Arc<Vec<TrackingProcess>>) -> Vec<String> { async fn get_subsystems(prcs: Arc<Vec<TrackingProcess>>) -> Vec<String> {
prcs.iter().map(|process| process.name.clone()).collect() prcs.iter().map(|process| process.name.clone()).collect()
} }
@ -180,7 +173,6 @@ async fn get_subsystems(prcs: Arc<Vec<TrackingProcess>>) -> Vec<String> {
/// ///
/// *depends on* : - /// *depends on* : -
/// ///
#[allow(dead_code)]
async fn get_all_metrics_process(proc: Arc<Process>, sys: Arc<System>) -> ProcessMetrics { async fn get_all_metrics_process(proc: Arc<Process>, sys: Arc<System>) -> ProcessMetrics {
let metrics = join!( let metrics = join!(
get_cpu_metrics_process(proc.clone()), get_cpu_metrics_process(proc.clone()),

View File

@ -223,6 +223,7 @@ pub async fn start_process(name: &str, path: &str) -> Result<(), CustomError> {
#[cfg(test)] #[cfg(test)]
mod process_unittests { mod process_unittests {
use std::io::Write;
use super::*; use super::*;
// 1 full cycle - start -> restart -> stop // 1 full cycle - start -> restart -> stop
// 2 full cycle - start -> freeze -> unfreze -> stop // 2 full cycle - start -> freeze -> unfreze -> stop