VladislavD
/
monitor
1
0
Fork 0
Code Issues 4 Pull Requests Packages Projects Releases 1 Wiki Activity

Compare commits

base: VladislavD:08ca2483e1a3ad11ae05fd313002d88f04caf85f
Branches Tags
VladislavD:rc
VladislavD:migrate
VladislavD:feature/configv2
VladislavD:master
VladislavD:0.11.10
VladislavD:v0.11.3
VladislavD:v0.10.11
VladislavD:v0.10.9
VladislavD:v0.9.25
..
compare: VladislavD:985ddc6065c61746bc7ace6ad032e7bfe0904801
Branches Tags
VladislavD:rc
VladislavD:migrate
VladislavD:feature/configv2
VladislavD:master
VladislavD:0.11.10
VladislavD:v0.11.3
VladislavD:v0.10.11
VladislavD:v0.10.9
VladislavD:v0.9.25

No commits in common. "08ca2483e1a3ad11ae05fd313002d88f04caf85f" and "985ddc6065c61746bc7ace6ad032e7bfe0904801" have entirely different histories.
08ca2483e1 ... 985ddc6065

5 changed files with 29 additions and 362 deletions
Show Stats Expand all files Collapse all files

154
src/options/config.rs
View File

@ -11,19 +11,7 @@ use tokio::time::Duration;
const CONFIG_PATH: &str = "settings.json";
/// # Fn `load_processes`
/// ## for reading and parsing *local* storing config
///
/// *input* : `&str`
///
/// *output* : `None` if local conf file doesn't exist or invalid | `Some(conf)` on finish reading and parsing
///
/// *initiator* : func `get_actual_config`
///
/// *managing* : conf file name in `&str` format
///
/// *depends on* : struct `Processes`
///
// 4ever sync
fn load_processes(json_filename: &str) -> Option<Processes> {
if let Ok(res) = fs::read_to_string(json_filename) {
if let Ok(conf) = serde_json::from_str::<Processes>(&res) {
@ -33,19 +21,6 @@ fn load_processes(json_filename: &str) -> Option<Processes> {
None
}
/// # Fn `get_actual_config`
/// ## for getting actual Monitor's config from local and remote storages
///
/// *input* : -
///
/// *output* : `None` on fatal error in mechanisms | `Some(conf)` on finish reading and parsing
///
/// *initiator* : main thread
///
/// *managing* : -
///
/// *depends on* : struct `Processes`
///
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
@ -89,19 +64,6 @@ pub async fn get_actual_config() -> Option<Processes> {
}
}
/// # Fn `get_remote_conf_watcher`
/// ## for infinitive pulling remote config
///
/// *input* : `&mut Connection`
///
/// *output* : `None` on fatal error | `Some(conf)` on succesfull pulling
///
/// *initiator* : fn `get_actual_config`
///
/// *managing* : mut ref `Connection` object
///
/// *depends on* : struct `Processes`
///
async fn get_remote_conf_watcher(conn : &mut Connection) -> Option<Processes> {
let mut conn = conn.as_pubsub();
let cont = crate::utils::get_container_id();
@ -143,22 +105,8 @@ async fn get_remote_conf_watcher(conn : &mut Connection) -> Option<Processes> {
}
None
}
/// # Fn `get_remote_conf_watcher`
/// ## for trying to pull remote config
///
/// > only for situation when local isn't None (no need to fck redis server)
///
/// *input* : `&str`
///
/// *output* : `None` on empty pubsub or error | `Some(conf)` on succesfull pulling
///
/// *initiator* : fn `get_actual_config`
///
/// *managing* : &str of Redis Server credentials
///
/// *depends on* : struct `Processes`
///
// ! 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) {
@ -217,21 +165,6 @@ fn once_get_remote_configuration(serv_info: &str) -> Option<Processes> {
// ! watchers
/// # Fn `open_watcher`
/// ## for infinitive opening Redis client
///
/// > only for situation when local isn't None (no need to fck redis server)
///
/// *input* : `Option<Processes>`
///
/// *output* : redis::Client on successful opening client
///
/// *initiator* : fn `get_actual_config`
///
/// *managing* : &str of Redis Server credentials
///
/// *depends on* : struct `redis::Client`
///
fn open_watcher(serv_info: &str) -> Client {
loop {
match Client::open(serv_info) {
@ -247,21 +180,6 @@ fn open_watcher(serv_info: &str) -> Client {
}
}
/// # Fn `get_connection_watcher`
/// ## for infinitive establishing Redis connection on existing client
///
/// > only for situation when local isn't None (no need to fck redis server)
///
/// *input* : `&Client`
///
/// *output* : `Connection`
///
/// *initiator* : fn `get_actual_config`
///
/// *managing* : &Client for opening connection
///
/// *depends on* : struct `redis::Connection`
///
fn get_connection_watcher(client: &Client) -> Connection {
loop {
match client.get_connection() {
@ -279,38 +197,11 @@ fn get_connection_watcher(client: &Client) -> Connection {
}
}
/// # Fn `restart_main_thread`
/// ## for restart monitor with new config
///
/// *input* : -
///
/// *output* : `Ok(())` on valid restart | `Err(er)` on error
///
/// *initiator* : fn `subscribe_config_stream`
///
/// *managing* : -
///
/// *depends on* : -
///
fn restart_main_thread() -> std::io::Result<()> {
let current_exe = env::current_exe()?;
Command::new(current_exe).exec();
Ok(())
}
/// # Fn `subscribe_config_stream`
/// ## for subscribe on changes, pulling to Redis pubsub to get more actual config
///
/// *input* : `Arc<Processes>`
///
/// *output* : `Ok(())` on end of work | `Err(er)` on error with subscribing mechanism
///
/// *initiator* : fn `subscribe_config_stream`
///
/// *managing* : `Arc<Processes>` to compare old config with new pulled
///
/// *depends on* : `Processes`
///
pub async fn subscribe_config_stream(actual_prcs: Arc<Processes>) -> Result<(), CustomError> {
if let Ok(client) = Client::open(format!("redis://{}/", &actual_prcs.config_server)) {
if let Ok(mut conn) = client.get_connection() {
@ -368,19 +259,6 @@ pub async fn subscribe_config_stream(actual_prcs: Arc<Processes>) -> Result<(),
Err(CustomError::Fatal)
}
/// # Fn `config_comparing`
/// ## for compare old and new configs
///
/// *input* : local: `&Processes`, remote: `&Processes`
///
/// *output* : `ConfigActuality::Local` or `ConfigActuality::Remote`
///
/// *initiator* : fn `subscribe_config_stream`, fn `get_actual_config`
///
/// *managing* : two objects `&Processes`
///
/// *depends on* : `Processes`, `ConfigActuality`
///
fn config_comparing(local: &Processes, remote: &Processes) -> ConfigActuality {
let local_date: u64 = local.date_of_creation.parse().unwrap();
let remote_date: u64 = remote.date_of_creation.parse().unwrap();
@ -399,19 +277,6 @@ fn config_comparing(local: &Processes, remote: &Processes) -> ConfigActuality {
// }
// }
/// # Fn `save_new_config`
/// ## mechanism for saving new config in local storage
///
/// *input* : `&Processes`, `&str`
///
/// *output* : `Ok(())` on succesfull saving | Err(er) on fs error
///
/// *initiator* : fn `subscribe_config_stream`, fn `get_actual_config`
///
/// *managing* : new config object: `&Processes` and config file name: `&str`
///
/// *depends on* : `Processes`
///
fn save_new_config(config: &Processes, config_file: &str) -> Result<(), CustomError> {
match serde_json::to_string_pretty(&config) {
// Ok(st) => match fs::write(config_file, st) {
@ -440,19 +305,6 @@ fn save_new_config(config: &Processes, config_file: &str) -> Result<(), CustomEr
}
}
/// # Fn `parse_extern_config`
/// ## for parsing &str to Processes
///
/// *input* : `&str`
///
/// *output* : parsed config in Some(Processes) | None on error with parsing
///
/// *initiator* : fn `subscribe_config_stream`, fn `once_get_remote_configuration`, fn `get_remote_conf`
///
/// *managing* : unparsed config `&str`
///
/// *depends on* : `Processes`
///
fn parse_extern_config(json_string: &str) -> Option<Processes> {
if let Ok(des) = serde_json::from_str::<Processes>(json_string) {
return Some(des);

13
src/options/logger.rs
View File

@ -13,19 +13,6 @@ use crate::utils::get_container_id;
// file_writer: BufWriter<File>,
// }
/// # Fn `setup_logger`
/// ## for initializing process of unstoppable grubbing metrics.
///
/// *input* : `Result<()>`
///
/// *output* : `Err` if it cant create logger | `Ok` after logger initialing
///
/// *initiator* : main thread
///
/// *managing* : -
///
/// *depends on* : -
///
pub fn setup_logger() -> Result<(), crate::options::structs::CustomError> {
// if Command::new("sh").args(["-c", "mkdir logs"]).output().is_err() {
// println!("Error: Cannot init logs directory");

42
src/options/signals.rs
View File

@ -9,19 +9,6 @@ use tokio::{
type SendersVec = Arc<Vec<Arc<mpsc::Sender<u8>>>>;
/// # Fn set_valid_destructor
/// ## for initializing process of unstoppable grubbing metrics.
///
/// *input* : `Result<()>`
///
/// *output* : `Err` if it cant create signals listeners | `Ok` on returning Monitor
///
/// *initiator* : main thread
///
/// *managing* : `Arc<Vec<Arc<mpsc::Sender<u8>>>>`
///
/// *depends on* : Sig, Signals
///
pub async fn set_valid_destructor(senders: SendersVec) -> Result<(), CustomError> {
let (mut int, mut term, mut stop) = (
Sig::new(Signals::Sigint, senders.clone()),
@ -36,30 +23,16 @@ pub async fn set_valid_destructor(senders: SendersVec) -> Result<(), CustomError
}
Ok(())
}
/// # Enum Signals
/// ## for instancing each managed system signals (such as SIGINT)
///
/// > (element needed in Sig constructor's signature)
///
/// *depends on* : -
enum Signals {
Sigint,
Sigterm,
Sigstop,
}
/// # Struct Signals
/// ## for instancing each managed system signals (such as SIGINT)
///
/// > (needed to construct system signals listener)
///
/// *depends on* : Signals
struct Sig {
signal: Signal,
sig_type: Signals,
senders: SendersVec,
}
/// ## default Sig's constructor
impl Sig {
fn new(signal_type: Signals, sends: SendersVec) -> Self {
Sig {
@ -69,9 +42,6 @@ impl Sig {
}
}
}
/// ## trait Display realization for returning String-name of signal
///
/// > (needed in logs)
impl std::fmt::Display for Signals {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
match self {
@ -81,7 +51,6 @@ impl std::fmt::Display for Signals {
}
}
}
/// ## associated func to init signals listener
impl Signals {
fn get_signal(&self) -> io::Result<Signal> {
match self {
@ -91,20 +60,9 @@ impl Signals {
}
}
}
/// # Trait SigPostProcessing
/// ## to handle post-processing jobs after getting system signal
///
/// ## > (needed in signals post-processing)
///
trait SigPostProcessing {
async fn post_processing(&mut self) -> io::Result<()>;
}
/// # Trait SigPostProcessing realization for Sig struct
/// ## to deinitialize Monitor correctly after getting signal
///
/// ## > (needed in signals post-processing)
///
impl SigPostProcessing for Sig {
async fn post_processing(&mut self) -> io::Result<()> {
// manipulations ...

153
src/options/structs.rs
View File

@ -11,20 +11,8 @@ pub enum ConfigActuality {
Remote,
}
/// # Struct for the 1st level in json conf file
/// ## for storing main config data
///
/// # struct for the 1st level in json conf file
/// > (needed in serialization and deserialization)
///
/// *depends on* : `TrackingProcess`
///
/// ```
/// {
/// -> "dateOfCreation": "1721381809104",
/// -> "configServer": "localhost",
/// -> "processes": [
/// { ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Processes {
// #[serde(rename="id")]
@ -37,24 +25,8 @@ pub struct Processes {
pub processes: Vec<TrackingProcess>,
}
/// # Struct for the 2nd level in json conf file
/// ## for each process to contain info, such as name, path and dependencies
///
/// # struct for each process to contain info, such as name, path and dependencies
/// > (needed in serialization and deserialization)
///
/// *depends on* : `Dependencies`
///
/// ```
/// ...
/// "processes": [
/// -> {
/// -> "name": "temp-process",
/// -> "path": "/home/user/monitor/runner-rs/temp-process",
/// -> "dependencies": { ... }
/// -> }, ...
/// ]
/// ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct TrackingProcess {
pub name: String,
@ -62,22 +34,8 @@ pub struct TrackingProcess {
pub dependencies: Dependencies,
}
/// # Struct for the 3d level in json conf file
/// ## for processes' dependencies including files and services
///
/// # struct for processes' dependencies including files and services
/// > (needed in serialization and deserialization)
///
/// *depends on* : `Files`, `Services`
///
/// ```
/// ...
/// "path": "/home/user/monitor/runner-rs/temp-process",
/// -> "dependencies": {
/// -> "files": [ ... ],
/// -> "services": [ ... ]
/// -> }
/// ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Dependencies {
#[serde(default)]
@ -86,49 +44,17 @@ pub struct Dependencies {
pub services: Vec<Services>,
}
/// # Struct for the 4th level in json conf file
/// ## for containing file object with its triggers to manipulate in daemons
///
/// # struct for containing file object with its triggers to manipulate in daemons
/// > (needed in serialization and deserialization)
///
/// *depends on* : `FileTriggers`
///
/// ```
/// ...
/// "files": [
/// -> {
/// -> "filename": "dep-file",
/// -> "src": "/home/user/monitor/runner-rs/tests/examples/",
/// -> "triggers": { ... }
/// -> } ,
/// ...
/// ], ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Files {
pub filename: String,
pub src: String,
pub triggers: FileTriggers,
pub triggers: FIleTriggers,
}
/// # Struct for the 4th level in json conf file
/// ## for containing service object with its triggers to manipulate in daemons
///
/// # struct for containing service object with its triggers to manipulate in daemons
/// > (needed in serialization and deserialization)
///
/// *depends on* : `ServiceTriggers`
///
/// ```
/// ...
/// "services": [
/// -> {
/// -> "hostname" : "ya.ru",
/// -> "port" : 443,
/// -> "triggers": { ... }
/// -> } ,
/// ...
/// ], ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Services {
pub hostname: String,
@ -136,23 +62,8 @@ pub struct Services {
pub triggers: ServiceTriggers,
}
/// # Struct for the 5th level in json conf file
/// ## for instancing each service's policies such as on lost or time to wait till reachable
///
/// # struct for instancing each service's policies such as on lost or time to wait till reachable
/// > (needed in serialization and deserialization)
///
/// *depends on* : -
///
/// ```
/// ...
/// "port": 443,
/// -> "triggers": {
/// -> "wait": 10,
/// -> "delay": 2,
/// -> "onLost": "hold"
/// -> }
/// ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct ServiceTriggers {
pub wait: u32,
@ -161,44 +72,24 @@ pub struct ServiceTriggers {
pub on_lost: String,
}
/// # Struct for the 5th level in json conf file
/// ## for instancing each file's policies such as on-delete or onupdate events
///
/// # struct for instancing each file's policies such as on-delete or onupdate events
/// > (needed in serialization and deserialization)
///
/// *depends on* : -
///
/// ```
/// ...
/// "src": "/home/user/monitor/runner-rs/tests/examples/",
/// -> "triggers": {
/// -> "onDelete": "stop",
/// -> "onChange": "stay"
/// -> }
/// ...
/// ```
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct FileTriggers {
pub struct FIleTriggers {
#[serde(rename = "onDelete")]
pub on_delete: String,
#[serde(rename = "onChange")]
pub on_change: String,
}
/// # Metrics struct
/// ## for gathering all system metrics (from container + each process)
///
/// > (needed in hagent communication, `?...?`)
///
/// *depends on* : `ContainerMetrics`, `ProcessMetrics`
///
///
///
#[derive(Debug, Clone, Serialize,)]
pub struct Metrics {
pub container_metrics : ContainerMetrics,
pub processes_metrics : Vec<ProcessMetrics>,
// pub net_metrics : Vec<PacketInfo>,
}
/// ## Metrics struct's constructor
impl Metrics {
pub fn new(cm: ContainerMetrics, prm: Vec<ProcessMetrics>) -> Self {
Metrics {
@ -210,12 +101,7 @@ impl Metrics {
}
/// # Container metrics struct
/// ## for gathering all container metrics
///
/// > (needed in gathering metrics)
///
/// *depends on* : -
///
#[derive(Debug, Clone, Serialize)]
pub struct ContainerMetrics {
@ -225,7 +111,7 @@ pub struct ContainerMetrics {
// pub net_activity : ???
processes : Vec<String>,
}
/// ## Container struct's constructor
impl ContainerMetrics {
pub fn new(container_id : &str, cpu: f32, ram: f32, subsystems: Vec<String>,) -> Self{
ContainerMetrics {
@ -237,12 +123,7 @@ impl ContainerMetrics {
}
}
/// # Process metrics struct
/// ## for gathering each process's all metrics
///
/// > (needed in gathering metrics)
///
/// *depends on* : -
///
#[derive(Debug, Clone, Serialize)]
pub struct ProcessMetrics {
@ -250,7 +131,7 @@ pub struct ProcessMetrics {
cpu_load : f32,
ram_load : f32,
}
/// ## Process struct's constructor
impl ProcessMetrics {
pub fn new(process_name :&str, cpu: f32, ram: f32) -> Self {
ProcessMetrics {
@ -261,13 +142,6 @@ impl ProcessMetrics {
}
}
/// # Packet info struct
/// ## for gathering info about container's net activity
///
/// > (needed in gathering metrics)
///
/// *depends on* : -
///
#[derive(Debug, Clone, Serialize)]
pub struct PacketInfo {
protocol : String,
@ -275,7 +149,6 @@ pub struct PacketInfo {
src_ip : Ipv4Addr,
size : usize,
}
/// ## PacketInfo's constructor
impl PacketInfo {
pub fn new(prt: String, dest: Ipv4Addr, src: Ipv4Addr, size_of_packet: usize) -> Self {
PacketInfo {

29
src/utils/metrics.rs
View File

@ -1,32 +1,29 @@
// submodule needed to get metrics such as
// cpu load, ram/rom load and net activity
// use std::sync::Mutex;
use std::sync::Mutex;
use std::sync::Arc;
use crate::options::structs::TrackingProcess;
use sysinfo::{Process, System};
use tokio::join;
use crate::options::structs::{ProcessMetrics, ContainerMetrics};
use crate::options::structs::{ProcessMetrics, ContainerMetrics, PacketInfo};
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>>>;
type PacketBuffer = Arc<Mutex<Vec<PacketInfo>>>;
/// # Fn init_metrics_grubber
/// ## for initializing process of unstoppable grubbing metrics.
///
/// *input* : `Result<()>`
///
/// *output* : `Err` if it cant create grubbers | `Ok` on finish
///
/// *initiator* : main thread
///
/// *managing* : object of unix-socket reader
///
/// *depends on* : network activity
///
// init_metrics_grubber - fn for initializing
// loop for unstoppable grubbing metrics.
//
// input : vec of processes
// output : Err if it cant create grubbers | Ok on finish
// initiator : main thread
// managing : object of unix-socket reader
// depends on : network activity
// !! FOR PROCESS !!
// pub async fn init_metrics_grubber(prcs: Arc<Vec<TrackingProcess>>) {
pub async fn init_metrics_grubber() {
let mut system = System::new();
// let mut buffer: Vec<PacketInfo> = vec![];