Compare commits

..

5 Commits

Author SHA1 Message Date
prplV c50c444f21 pub commands (need later) 2025-04-10 08:54:12 -04:00
prplV 71acb4a32e services v2 controller without impl 2025-04-10 08:54:00 -04:00
prplV f504632c4d files v2 controller without impl 2025-04-10 08:53:52 -04:00
prplV 2b82fb7aac prcs v2 controller with impl 2025-04-10 08:53:38 -04:00
prplV 584404c050 preparing for prcsv2 2025-04-10 08:52:58 -04:00
8 changed files with 230 additions and 18 deletions

View File

@ -12,7 +12,7 @@ pub struct Cli {
subcommand,
help = "to manage Noxis work",
)]
command : Commands,
pub command : Commands,
}
#[derive(Debug, Subcommand, serde::Serialize, serde::Deserialize)]
@ -56,13 +56,13 @@ pub struct StartAction {
num_args = 1..,
value_delimiter = ' '
)]
flags : Vec<String>,
pub flags : Vec<String>,
}
#[derive(Debug, Parser, serde::Serialize, serde::Deserialize)]
pub struct ConfigCommand {
#[command(subcommand)]
action : ConfigAction,
pub action : ConfigAction,
}
#[derive(Debug, Subcommand, serde::Serialize, serde::Deserialize)]
@ -89,12 +89,12 @@ pub struct LocalConfig {
action,
help = "to read following input as JSON",
)]
is_json : bool,
pub is_json : bool,
// value
#[arg(
help = "path to config file or config String (with --json flag)",
)]
config : String,
pub config : String,
}
#[derive(Debug, Parser, serde::Serialize, serde::Deserialize)]
@ -102,16 +102,16 @@ pub struct ProcessCommand {
#[arg(
help = "name of needed process",
)]
process : String,
pub process : String,
#[command(
subcommand,
help = "To get current process's status",
)]
action : ProcessAction,
pub action : ProcessAction,
}
#[derive(Debug, Subcommand, serde::Serialize, serde::Deserialize)]
enum ProcessAction {
pub enum ProcessAction {
#[command(
about = "To get info about current process status",
)]

View File

@ -12,7 +12,7 @@
"src": "./tests/examples/",
"triggers": {
"onDelete": "stop",
"onChange": "stay"
"onChange": "restart"
}
}
],
@ -23,7 +23,7 @@
"triggers": {
"wait": 10,
"delay": 2,
"onLost": "hold"
"onLost": "restart"
}
}
]

View File

@ -15,6 +15,7 @@ use utils::*;
use options::preboot::PrebootParams;
use tokio::sync::{broadcast, oneshot};
use options::config::v2::init_config_mechanism;
use utils::v2::init_monitoring;
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() -> anyhow::Result<()>{
@ -23,11 +24,13 @@ async fn main() -> anyhow::Result<()>{
let _ = setup_logger();
info!("Noxis is configurating...");
let (tx_brd, mut _rx_brd) = broadcast::channel::<Processes>(1);
let (_tx_oneshot, rx_oneshot) = oneshot::channel::<Processes>();
//
let (tx_brd, mut rx_brd) = broadcast::channel::<Processes>(1);
// cli <-> config
let (tx_oneshot, rx_oneshot) = oneshot::channel::<Processes>();
let mut handler: Vec<tokio::task::JoinHandle<()>> = vec![];
// initilaizing task for config manipulations
let config_module = tokio::spawn(async move {
let _ = init_config_mechanism(
rx_oneshot,
@ -37,6 +40,7 @@ async fn main() -> anyhow::Result<()>{
});
handler.push(config_module);
// initilaizing task for cli manipulation
let cli_module = tokio::spawn(async move {
if let Err(er) = init_cli_pipeline().await {
error!("CLI pipeline failed due to {}", er)
@ -44,14 +48,22 @@ async fn main() -> anyhow::Result<()>{
});
handler.push(cli_module);
// initilaizing task for deinitializing `Noxis`
let ctrlc = tokio::spawn(async move {
if let Err(er) = set_valid_destructor(vec![].into()).await {
error!("CTRLC mod failed!");
error!("Destructor mod failed due to {}", er);
}
std::process::exit(0);
});
handler.push(ctrlc);
let monitoring = tokio::spawn(async move {
if let Err(er) = init_monitoring(&mut rx_brd).await {
error!("Monitoring mod failed due to {}", er);
}
});
handler.push(monitoring);
for i in handler {
let _ = i.await;
}

View File

@ -3,6 +3,39 @@
use std::net::Ipv4Addr;
use serde::{Deserialize, Serialize};
pub enum DependencyType {
File,
Service,
}
pub enum Triggers<'a> {
File{ on_change: &'a str, on_delete: &'a str },
Service(&'a str),
}
impl std::fmt::Display for DependencyType {
fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
return match self {
DependencyType::File => write!(f, "File"),
DependencyType::Service => write!(f, "Service"),
}
}
}
pub enum ProcessState {
Pending,
Holding,
Stopped,
StoppedByCli,
}
pub enum Events<'a> {
Positive(&'a str),
Negative(NegativeOutcomes<'a>)
}
pub enum NegativeOutcomes<'a> {
FileWasChanged(&'a str, DependencyType, &'a str),
FileWasMovedOrDeleted(&'a str, DependencyType, &'a str),
ServiceIsUnreachable(&'a str, DependencyType, &'a str),
}
/// # an Error enum (next will be deleted and replaced)
pub enum CustomError {
Fatal,

View File

@ -6,12 +6,11 @@ pub mod services;
// TODO : saving current flags state
use crate::options::structs::CustomError;
use crate::options::structs::TrackingProcess;
use crate::options::structs::{CustomError, TrackingProcess, Processes};
use files::create_watcher;
use files::file_handler;
use inotify::Inotify;
use log::{error, warn};
use log::{error, warn, info};
use prcs::{
freeze_process, is_active, is_frozen, restart_process, start_process, terminate_process,
unfreeze_process,
@ -22,9 +21,50 @@ use std::sync::Arc;
use tokio::join;
use tokio::sync::mpsc;
use tokio::time::Duration;
use tokio::sync::broadcast::Receiver;
use tokio::sync::mpsc::{Receiver as MpscReciever, Sender as MpscSender};
const GET_ID_CMD: &str = "hostname";
pub mod v2 {
use super::*;
// spawn tasks
// spawn prc
// spawn files
// spawn services
// ## for ... i.await in loop
pub async fn init_monitoring(
local_config: &mut Receiver<Processes>,
) -> anyhow::Result<()> {
let config = if !local_config.is_empty() {
local_config.recv().await?
} else {
let mut tick = tokio::time::interval(Duration::from_millis(500));
loop {
tick.tick().await;
break match local_config.try_recv() {
Ok(conf) => conf,
Err(_) => continue,
}
}
};
info!("Processing {} processes ...", config.processes.len());
// LinkedList <ProcessController>
// LinkedList <FileController>
Ok(())
}
// spawn prc check with semaphore check
async fn prcs_monitoriing() -> anyhow::Result<()> { Ok(()) }
// spawn file check with semaphore check
async fn files_monitoriing() -> anyhow::Result<()> { Ok(()) }
// spawn service check with semaphore check
async fn services_monitoriing() -> anyhow::Result<()> { Ok(()) }
}
/// # Fn `run_daemons`
/// ## async func to run 3 main daemons: process, service and file monitors and manage process state according to given messages into channel
///
@ -133,7 +173,6 @@ async fn process_protocol_symbol(proc: Arc<TrackingProcess>, val: u8) -> Result<
},
// // 9 - File-dependency change -> staying (after check)
9 => {
// no need to trash logs
warn!("File-dependency warning (file changed). Ignoring event on {} process...", &proc.name);
tokio::time::sleep(Duration::from_millis(100)).await;
},

View File

@ -5,7 +5,25 @@ use std::borrow::BorrowMut;
use std::path::Path;
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::sync::mpsc::Sender as MpscSender;
use tokio::time::Duration;
use crate::options::structs::Events;
pub mod v2 {
use std::collections::HashMap;
use crate::options::structs::Triggers;
use super::*;
struct FilesController<'a> {
name: &'a str,
watcher: Inotify,
// obj: Arc<Files>,
triggers: HashMap<&'a str, Triggers<'a>>,
event_registrator: Vec<MpscSender<Events<'a>>>,
}
}
/// # Fn `create_watcher`
/// ## for creating watcher on file's delete | update events

View File

@ -3,6 +3,100 @@ use log::{error, warn};
use std::process::{Command, Output};
use std::sync::Arc;
use tokio::time::Duration;
use crate::options::structs::{TrackingProcess, ProcessState, Events, NegativeOutcomes};
use std::collections::HashSet;
use tokio::sync::mpsc::Receiver as MpscReciever;
pub mod v2 {
use log::info;
use crate::options::structs::DependencyType;
use super::*;
pub struct ProcessController<'a> {
name: &'a str,
obj: Arc<TrackingProcess>,
state: ProcessState,
event_reader: MpscReciever<Events<'a>>,
negative_events: HashSet<&'a str>,
}
impl<'a> ProcessController<'a> {
pub async fn process(&mut self) {
if let Ok(event) = self.event_reader.try_recv() {
match event {
Events::Positive(target) => {
if self.negative_events.contains(target) {
self.negative_events.remove(target);
}
},
Events::Negative(event) => {
match event {
NegativeOutcomes::FileWasChanged(target, dep_type, trigger) |
NegativeOutcomes::FileWasMovedOrDeleted(target, dep_type, trigger) |
NegativeOutcomes::ServiceIsUnreachable(target, dep_type, trigger) => {
if !self.negative_events.contains(target) {
self.negative_events.insert(target);
self.trigger_on(
target,
trigger,
dep_type
).await;
}
},
}
},
}
}
match self.state {
ProcessState::Holding => {
if self.negative_events.len() == 0 {
info!("No negative dependecies events on {} process. Unfreezing ...", self.name);
unfreeze_process(self.name).await;
}
},
ProcessState::Stopped => {
if self.negative_events.len() == 0 {
info!("No negative dependecies events on {} process. Starting ...", self.name);
if let Err(_) = start_process(self.name, &self.obj.path).await {
error!("Cannot start process {} due to {}", self.name, "system unrecognized error");
}
}
},
_ => {},
}
}
async fn trigger_on(&mut self, dep_name: &str, trigger: &str, dep_type: DependencyType) {
match trigger {
"stay" => {
info!("Event on {} `{}` for {}. Ignoring ...", dep_type, dep_name, self.name);
},
"stop" => {
if is_active(self.name).await {
info!("Event on {} `{}` for {}. Stopping ...", dep_type, dep_name, self.name);
terminate_process(self.name).await;
self.state = ProcessState::Stopped;
}
},
"hold" => {
if !is_frozen(self.name).await {
info!("Event on {} `{}` for {}. Freezing ...", dep_type, dep_name, self.name);
freeze_process(self.name).await;
self.state = ProcessState::Holding;
}
},
"restart" => {
info!("Event on {} `{}` for {}. Restarting ...", dep_type, dep_name, self.name);
let _ = restart_process(self.name, &self.obj.path).await;
},
_ => error!("Impermissible trigger in file-trigger for {}. Ignoring event ...", self.name),
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
async fn trigger_on_servcie(&mut self, file_name: &str, trigger: &str) {}
}
}
/// # Fn `get_pid`
/// ## for initializing process of unstoppable grubbing metrics.

View File

@ -5,6 +5,22 @@ use std::net::{TcpStream, ToSocketAddrs};
use std::sync::Arc;
use tokio::sync::mpsc;
use tokio::time::{Duration, Instant};
use tokio::sync::mpsc::Sender as MpscSender;
use crate::options::structs::Events;
pub mod v2 {
use crate::options::structs::Triggers;
use super::*;
use std::collections::HashMap;
struct ServicesController<'a> {
name: &'a str,
obj: Arc<Services>,
triggers: HashMap<&'a str, Triggers<'a>>,
event_registrator: Vec<MpscSender<Events<'a>>>,
}
}
/// # Fn `service_handler`
/// ## function to realize mechanism of current process' dep services monitoring