process_polling update + config structs fix

.env.example

@@ -2,7 +2,11 @@ CONFIG_SERVER_CREDS = "ws://ip.ip.ip.ip:port"
 API_GRUBBER_SOCKET = "api-grubber.sock"
 PREPROC_SOCKET = "preproc.sock"
 
+# PostgreSQL connection [DEPRECATED]
 DB_HOST = "ip.addr.postgresql.server"
 DB_USER = "db_user"
 DB_PASSWORD = "db_user_password"
-DB_DBNAME = "db_name"1
+DB_DBNAME = "db_name"1
+
+# Prometheus-Exporter info
+EXPORTER_URL = "ip.ip.ip.ip:port" 

config_api.json

@@ -1,7 +1,7 @@
 {
     "config":  [     
 		{
-			"id":"demo_vcs_vinteo_dev_api", 
+			"id":"zvks", 
 			"login" : "",
 			"pass" : "",
 			"api_key" : "6fe8b0db-62b4-4065-9c1e-441ec4228341.9acec20bd17d7178f332896f8c006452877a22b8627d089105ed39c5baef9711",

crates/api-grub/src/config.rs

@@ -10,12 +10,13 @@ use tokio::{io::AsyncReadExt, net::UnixListener};
 use tokio::time::{sleep, Duration};
 use std::result::Result::Ok as stdOk;
 use tokio::sync::mpsc::Sender;
+use integr_structs::api::v3::Config;
 
 const CONFIG_PATH: &str = "config_api.json";
 const SOCKET_PATH: &str = "api-grub.sock";
 
 // todo! rewrite to use current_exe
-pub async fn pull_local_config() -> Result<ApiConfigV2> {
+pub async fn pull_local_config() -> Result<Config> {
     // let conf_path = std::env::current_exe()?;
     let path = Path::new(CONFIG_PATH);
     // return match conf_path.parent() {
@@ -28,7 +29,7 @@ pub async fn pull_local_config() -> Result<ApiConfigV2> {
     //     None => Err(Error::msg("No local conf was found")) 
     // }
     if path.exists() && path.is_file() {
-        let config: ApiConfigV2 = from_str(
+        let config: Config = from_str(
             &fs::read_to_string(CONFIG_PATH)?
         )?;
         Ok(config)
@@ -39,7 +40,7 @@ pub async fn pull_local_config() -> Result<ApiConfigV2> {
 
 // for config pulling
 // ++++ reader to channel 
-pub async fn init_config_grub_mechanism(tx: &Sender<ApiConfigV2>) -> Result<()> {
+pub async fn init_config_grub_mechanism(tx: &Sender<Config>) -> Result<()> {
     info!("Initializing Unix-Socket listening for pulling new configs...");
     let server = init_unix_listener().await?;
     // 
@@ -51,7 +52,7 @@ pub async fn init_config_grub_mechanism(tx: &Sender<ApiConfigV2>) -> Result<()>
             if let Err(er) =  stream.read_to_string(&mut buffer).await {
                 warn!("Cannot read config from stream due to {}", er);
             } else {
-                let config: Result<ApiConfigV2, serde_json::Error> = from_str(&buffer);
+                let config: Result<Config, serde_json::Error> = from_str(&buffer);
                 if let stdOk(conf) = config {
                     info!("New config was pulled from Unix-Stream. Saving it locally and sharing with API-grub module...");
                     if let Err(er) = save_new_config(&buffer).await {
@@ -97,13 +98,13 @@ mod config_unittests {
     #[test]
     async fn check_save_new_config() {
         use std::fs;
-        use integr_structs::api::ApiConfigV2;
+        use integr_structs::api::v3::Config;
         use serde_json::to_string;
         
         let test_config_path  = "test_config_api.json";
 
         // config gen  
-        let config = to_string::<ApiConfigV2>(&ApiConfigV2::default());
+        let config = to_string::<Config>(&Config::default());
         assert!(config.is_ok());
         let config = config.unwrap();
 

crates/api-grub/src/export.rs

@@ -1,7 +1,8 @@
 use deadpool_postgres::{Config, Pool, Runtime, Client as PgClient};
+use reqwest::Client;
 use tokio_postgres::NoTls;
 use std::env;
-use anyhow::{Error, Result};
+use anyhow::{Result};
 use log::{info, error};
 
 pub struct Exporter {
@@ -35,7 +36,7 @@ impl Exporter {
     pub fn is_no_connection(&self) -> bool { self.pool.is_none() }
     pub fn init() -> Self {
         Self {
-            pool : Self::pool_construct()
+            pool : Self::pool_construct(),
         }
     }
     pub async fn get_connection_from_pool(&self) -> Option<PgClient> {
@@ -50,5 +51,16 @@ impl Exporter {
         let _ = client.query(&query, &[&metrics]).await?;
         Ok(())
     }
+    pub async fn export_metrics(metrics: &str) -> Result<()> {
+        let url = env::var("EXPORTER_URL")?;
+        // let req = Request::new(Method::PUT, 
+        //     Url::parse(metrics)?);
+        let req = Client::new()
+            .put(url)
+            .json(metrics)
+            .send().await;
+        req?;
+        Ok(())
+    }
 
 }

crates/api-grub/src/json.rs

@@ -12,7 +12,7 @@ impl JsonParser {
                 true => JsonParser::get_sum_of_metrics_in_array(target, json),
                 false => JsonParser::get_metric(target, json),
             };
-            res_vec.push(MetricOutput::new_with_slices(&target.id, &target.json_type, metric));
+            res_vec.push(MetricOutput::new_with_slices(&target.id, &target.json_type, &target.addr, metric));
         }  
         serde_json::to_value(res_vec).unwrap_or(Value::Null)
     }

crates/api-grub/src/main.rs

@@ -6,6 +6,7 @@ mod export;
 
 use anyhow::Result;
 use integr_structs::api::ApiConfigV2;
+use integr_structs::api::v3::Config;
 use logger::setup_logger;
 // use log::{info, warn};
 use config::{pull_local_config, init_config_grub_mechanism};
@@ -22,7 +23,7 @@ async fn main() -> Result<()>{
     setup_logger().await?;
     let config = get_config().await;
     // config update channel
-    let (tx, mut rx) = mpsc::channel::<ApiConfigV2>(1);
+    let (tx, mut rx) = mpsc::channel::<Config>(1);
     // futures
     // todo : rewrite with spawn
     // let config_fut = init_config_grub_mechanism(&tx);
@@ -57,7 +58,7 @@ async fn main() -> Result<()>{
     Ok(())
 }
 
-async fn get_config() -> ApiConfigV2 {
+async fn get_config() -> Config {
     return match pull_local_config().await {
         Ok(conf) => {
             info!("Local config was loaded");
@@ -65,7 +66,7 @@ async fn get_config() -> ApiConfigV2 {
         },
         Err(er) => {
             warn!("Cannot get local config due to {}", er);
-            ApiConfigV2::default()
+            Config::default()
         }
     }
 }

crates/api-grub/src/net.rs

@@ -10,23 +10,24 @@ use tokio::task::JoinHandle;
 // use tokio::sync::Mutex;
 use dotenv::dotenv;
 use crate::json::JsonParser;
-use crate::export::Exporter;
+use crate::export::{self, Exporter};
-use integr_structs::api::v3::Config;
+use integr_structs::api::v3::{Config, ConfigEndpoint, Credentials};
 
 // type BufferType = Arc<Mutex<Vec<String>>>; 
 
 // for api info pulling 
-pub async fn init_api_grub_mechanism(config: ApiConfigV2, rx: &mut Receiver<ApiConfigV2>) -> Result<()> {
+pub async fn init_api_grub_mechanism(config: Config, rx: &mut Receiver<Config>) -> Result<()> {
-    info!("Initializing API-info grubbing mechanism...");
+    info!("Initializing API-info grubbing mechanism :");
-    info!("Loading vars from .env file if exists...");
+    info!("1) Loading vars from .env file if exists...");
     let _ = dotenv().ok();
 
     let mut config = config;
     let mut poller = ApiPoll::new(&mut config).await;
+    info!("2) Api-Poller has initialized");
     let client = Exporter::init(); 
+    info!("3) Exporter has initialized");
     let shared_pool = Arc::new(client);
     loop {
-        let shared_pool = shared_pool.clone();
         if poller.is_default().await {
             sleep(Duration::from_secs(5)).await;
         } else {
@@ -36,8 +37,9 @@ pub async fn init_api_grub_mechanism(config: ApiConfigV2, rx: &mut Receiver<ApiC
                     info!("Config changed");
                 }
             }
+            let shared_pool = shared_pool.clone();
             info!("Data from API: {:?}", poller.process_polling(shared_pool).await);
-            sleep(Duration::from_secs(poller.get_delay().await as u64)).await;
+            // sleep(Duration::from_secs(poller.get_delay().await as u64)).await;
         }
     }
     // Ok(())
@@ -62,156 +64,188 @@ impl RestMethod {
     } 
 }
 struct ApiPoll<'a> {
-    config : &'a mut ApiConfigV2,
+    config : &'a mut Config,
     client : Client,
 }
 
 impl<'a> ApiPoll<'a> {
-    pub async fn new(poll_cfg : &'a mut ApiConfigV2) -> Self {
+    pub async fn new(poll_cfg : &'a mut Config) -> Self {
         Self {
             config : poll_cfg,
             client : Client::new(),
         }
     }
     // can be weak and with bug test needed
-    pub async fn change_config(&mut self, conf: ApiConfigV2) {
+    pub async fn change_config(&mut self, conf: Config) {
         *self.config = conf;
     }
     pub async fn is_default(&self) -> bool {
-        self.config.template.len() == 0
+        self.config.is_default().await
+    }
+    // pub async fn get_delay(&self) -> u32 {
+    //     self.config.timeout
+    // }
+    pub async fn process_endpoint(
+        client : Arc<Client>, 
+        config : Arc<ConfigEndpoint>,
+        creds : Credentials,
+        exporter : Arc<Exporter>
+    ) -> Result<()> {
+
+        Ok(())
     }
     pub async fn process_polling(&self, exporter: Arc<Exporter>) -> Result<()> {
         // let buffer: BufferType = Arc::new(Mutex::new(vec![]));
-        let mut join_handles: Vec<JoinHandle<Result<()>>> = vec![];
+        // let mut join_handles: Vec<JoinHandle<Result<()>>> = vec![];
         let client = Arc::new(self.client.clone());
-        let template = Arc::new(self.config.template.clone());
+        let config = Arc::new(self.config.clone());
+        let endpoints: Vec<Arc<ConfigEndpoint>> = ConfigEndpoint::from_config(config.clone());
+        let mut join_handles: Vec<JoinHandle<Result<()>>> = vec![];
 
-        if self.is_default().await { return Err(Error::msg("Default config with no endpoints")) }
+        for (idx, _) in config.config.iter().enumerate() {
-
+            // let for_creds = endpoints[idx].clone();
-        // TODO: rewrite nextly to async 
+            let creds = Credentials::from_config_endpoint(endpoints[idx].clone());
-        for point in template.iter() {
+            let endpoint = endpoints[idx].clone();
-            let point = Arc::new(point.clone());
-            // let buffer = buffer.clone();
             let client = client.clone();
             let exporter = exporter.clone();
-            let endpoint_processer = tokio::spawn(async move {
+            let join_handler = tokio::spawn(async move {
-                let point = point.clone();
+                Self::process_endpoint(
-                match client.request(RestMethod::from_str(&point.method).await, &point.url).send().await {
+                    client, 
-                    Ok(resp) => {
+                    endpoint, 
-                        if !resp.status().is_success() {
+                    creds, 
-                            error!("ErrorCode in Response from API. Check configuration");
+                    exporter.clone()
-                            return Err(Error::msg("Error during sending request"));
+                ).await
-                        }
-                        if let Ok(text) = resp.text().await {
-                            // 
-                            let metrics = ProcessedEndpoint::from_target_response(&text, &point)?;
-                            // dbg!(&metrics);
-                            println!("{}", &metrics);
-                            // 
-                            if let Some(conn) = exporter.get_connection_from_pool().await {
-                                
-                                // TEST: to exporter 
-                                let res = client.request(
-                                    RestMethod::from_str("post").await,
-                                    "http://192.168.2.34:9101/update")
-                                .json(&metrics)
-                                .send().await;
-                                if let Err(er) = res {
-                                    error!("Cannot send data to exporter due to: {}", er);
-                                } else {
-                                    println!("{:?}", res.unwrap().text().await);
-                                }
-
-                                if let Err(er) = Exporter::export_data(conn, &metrics).await {
-                                    error!("Cannot export data to DB during to: {}", er);
-                                    return Err(Error::msg("Error during exporting data to DB"));
-                                }
-                            } else {
-                                if !exporter.is_no_connection() {
-                                    return Err(Error::msg("Error during getting connection from pool"));
-                                }
-                            }
-
-                            // let mut buffer = buffer.lock().await;
-                            // buffer.push(text);
-                        } else {
-                            error!("{}: {} - Error with extracting text field from Response", &point.method.to_uppercase(), &point.url);
-                            return Err(Error::msg("Error with extracting text field from Response"));
-                        }
-                    },
-                    Err(_) => {
-                        error!("{}: {} endpoint is unreachable", &point.method.to_uppercase(), &point.url);
-                        return Err(Error::msg("Endpoint is unreachable"));
-                    },
-                }
-                Ok(())
             });
-            join_handles.push(endpoint_processer);
+            join_handles.push(join_handler);
         }
-
         for i in join_handles {
             let _ = i.await;
         }
+        // let template = Arc::new(self.config.template.clone());
 
-        // let buffer = buffer.lock().await;
+        // if self.is_default().await { return Err(Error::msg("Default config with no endpoints")) }
-        // match &buffer.len() {
+
-        //     0 => Err(Error::msg("Error due to API grubbing. Check config" )),
+        // // TODO: rewrite nextly to async 
-        //     _ => {
+        // for point in template.iter() {
+        //     let point = Arc::new(point.clone());
+        //     // let buffer = buffer.clone();
+        //     let client = client.clone();
+        //     let exporter = exporter.clone();
+        //     let endpoint_processer = tokio::spawn(async move {
+        //         let point = point.clone();
+        //         match client.request(RestMethod::from_str(&point.method).await, &point.url).send().await {
+        //             Ok(resp) => {
+        //                 if !resp.status().is_success() {
+        //                     error!("ErrorCode in Response from API. Check configuration");
+        //                     return Err(Error::msg("Error during sending request"));
+        //                 }
+        //                 if let Ok(text) = resp.text().await {
+        //                     // 
+        //                     let metrics = ProcessedEndpoint::from_target_response(&text, &point)?;
+        //                     // dbg!(&metrics);
+        //                     println!("{}", &metrics);
+        //                     // 
+        //                     if let Some(conn) = exporter.get_connection_from_pool().await {
+                                
+        //                         // TEST: to exporter 
+        //                         let res = client.request(
+        //                             RestMethod::from_str("post").await,
+        //                             "http://192.168.2.34:9101/update")
+        //                         .json(&metrics)
+        //                         .send().await;
+        //                         if let Err(er) = res {
+        //                             error!("Cannot send data to exporter due to: {}", er);
+        //                         } else {
+        //                             println!("{:?}", res.unwrap().text().await);
+        //                         }
+
+        //                         if let Err(er) = Exporter::export_data(conn, &metrics).await {
+        //                             error!("Cannot export data to DB during to: {}", er);
+        //                             return Err(Error::msg("Error during exporting data to DB"));
+        //                         }
+        //                     } else {
+        //                         if !exporter.is_no_connection() {
+        //                             return Err(Error::msg("Error during getting connection from pool"));
+        //                         }
+        //                     }
+
+        //                     // let mut buffer = buffer.lock().await;
+        //                     // buffer.push(text);
+        //                 } else {
+        //                     error!("{}: {} - Error with extracting text field from Response", &point.method.to_uppercase(), &point.url);
+        //                     return Err(Error::msg("Error with extracting text field from Response"));
+        //                 }
+        //             },
+        //             Err(_) => {
+        //                 error!("{}: {} endpoint is unreachable", &point.method.to_uppercase(), &point.url);
+        //                 return Err(Error::msg("Endpoint is unreachable"));
+        //             },
+        //         }
         //         Ok(())
-        //     },
+        //     });
+        //     join_handles.push(endpoint_processer);
         // }
+
+        // for i in join_handles {
+        //     let _ = i.await;
+        // }
+
+        // // let buffer = buffer.lock().await;
+        // // match &buffer.len() {
+        // //     0 => Err(Error::msg("Error due to API grubbing. Check config" )),
+        // //     _ => {
+        // //         Ok(())
+        // //     },
+        // // }
         Ok(())
     }
-    pub async fn get_delay(&self) -> u32 {
-        self.config.timeout
-    }
 }
 
-#[cfg(test)]
+// #[cfg(test)]
-mod net_unittests {
+// mod net_unittests {
-    use super::*;
+//     use super::*;
-    use tokio::test;
+//     use tokio::test;
     
-    #[test]
+//     #[test]
-    async fn check_str_to_rest_method() {
+//     async fn check_str_to_rest_method() {
-        assert_eq!(RestMethod::from_str("get").await, Method::GET);
+//         assert_eq!(RestMethod::from_str("get").await, Method::GET);
-        assert_eq!(RestMethod::from_str("post").await, Method::POST);
+//         assert_eq!(RestMethod::from_str("post").await, Method::POST);
-        assert_eq!(RestMethod::from_str("patch").await, Method::PATCH);
+//         assert_eq!(RestMethod::from_str("patch").await, Method::PATCH);
-        assert_eq!(RestMethod::from_str("put").await, Method::PUT);
+//         assert_eq!(RestMethod::from_str("put").await, Method::PUT);
-        assert_eq!(RestMethod::from_str("delete").await, Method::DELETE);
+//         assert_eq!(RestMethod::from_str("delete").await, Method::DELETE);
-        assert_eq!(RestMethod::from_str("invalid_method").await, Method::GET);
+//         assert_eq!(RestMethod::from_str("invalid_method").await, Method::GET);
-    } 
+//     } 
-    #[test]
+//     #[test]
-    async fn check_api_poll_change_config() {
+//     async fn check_api_poll_change_config() {
-        let mut conf1 = ApiConfigV2::default();
+//         let mut conf1 = ApiConfigV2::default();
-        let conf2 = ApiConfigV2::pattern();
+//         let conf2 = ApiConfigV2::pattern();
-        let mut poll = ApiPoll::new(&mut conf1).await;
+//         let mut poll = ApiPoll::new(&mut conf1).await;
-        poll.change_config(conf2).await;
+//         poll.change_config(conf2).await;
-        assert_eq!(poll.config.timeout, 1)
+//         assert_eq!(poll.config.timeout, 1)
-    }
+//     }
 
-    #[test]
+//     #[test]
-    async fn check_api_poll_is_default() {
+//     async fn check_api_poll_is_default() {
-        let mut conf1 = ApiConfigV2::default();
+//         let mut conf1 = ApiConfigV2::default();
-        let poll = ApiPoll::new(&mut conf1).await;
+//         let poll = ApiPoll::new(&mut conf1).await;
-        assert!(poll.is_default().await)
+//         assert!(poll.is_default().await)
-    }
+//     }
 
-    #[test]
+//     #[test]
-    async fn check_api_grubbing_mechanism_on_public_one() {
+//     async fn check_api_grubbing_mechanism_on_public_one() {
-        use log::{set_max_level, LevelFilter};
+//         use log::{set_max_level, LevelFilter};
 
-        set_max_level(LevelFilter::Off);
+//         set_max_level(LevelFilter::Off);
-        let mut conf1 = ApiConfigV2::pattern();
+//         let mut conf1 = ApiConfigV2::pattern();
-        let conf2 = ApiConfigV2::default(); 
+//         let conf2 = ApiConfigV2::default(); 
-        let exporter = Arc::new(Exporter::init());
+//         let exporter = Arc::new(Exporter::init());
 
-        let mut poll = ApiPoll::new(&mut conf1).await;
+//         let mut poll = ApiPoll::new(&mut conf1).await;
-        assert!(poll.process_polling(exporter.clone()).await.is_ok());
+//         assert!(poll.process_polling(exporter.clone()).await.is_ok());
 
-        dbg!(&poll.config);
+//         dbg!(&poll.config);
-        poll.change_config(conf2).await;
+//         poll.change_config(conf2).await;
-        dbg!(&poll.config);
+//         dbg!(&poll.config);
-        assert!(poll.process_polling(exporter.clone()).await.is_err());
+//         assert!(poll.process_polling(exporter.clone()).await.is_err());
-    }
+//     }
-}
+// }

crates/integr-structs/src/api.rs

@@ -1,7 +1,9 @@
+use core::sync;
 use std::collections::HashMap;
 use serde::{Serialize, Deserialize};
 use serde_json::{ to_string_pretty, Value };
 use anyhow::Result;
+use std::sync::Arc;
 
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -150,14 +152,14 @@ pub mod v3 {
     pub use super::*;
     
     // in config
-    #[derive(Deserialize)]  
+    #[derive(Serialize, Deserialize, Clone)]  
     pub struct Metric {
         pub id : String,
         #[serde(rename = "type")]
         pub json_type : String,
         pub addr : String,
     }
-    #[derive(Deserialize)]
+    #[derive(Serialize, Deserialize, Clone)]
     pub struct Metrics {
         pub name : String,
         pub url : String,
@@ -165,9 +167,9 @@ pub mod v3 {
         pub measure : Vec<Metric>
     }
 
-    #[derive(Deserialize)]
+    #[derive(Serialize, Deserialize, Clone)]
     pub struct ConfigEndpoint {
-        ip : String,
+        id : String,
         login : String,
         #[serde(rename = "pass")]
         password : String,
@@ -177,10 +179,27 @@ pub mod v3 {
         #[serde(default)]
         metrics : Vec<Metrics>, 
     }
+     impl ConfigEndpoint {
+        pub fn from_config(config: Arc<Config>) -> Vec<Arc<Self>> {
+            let mut result: Vec<Arc<ConfigEndpoint>> = Vec::new();
+            config.config
+                .iter()
+                .for_each(|el| {
+                    result.push(Arc::new(el.clone()))
+                });
+            result
+        }
+        pub fn get_period(&self) -> Option<u32> {
+            self.period.parse().ok()
+        }
+        pub fn get_timeout(&self) -> Option<u32> {
+            self.timeout.parse().ok()
+        }
+     }
 
-    #[derive(Deserialize)]
+    #[derive(Serialize, Deserialize, Clone)]
     pub struct Config {
-        config : Vec<ConfigEndpoint>,
+        pub config : Vec<ConfigEndpoint>,
     }
 
     impl Default for Config {
@@ -197,25 +216,13 @@ pub mod v3 {
         }
     }
 
-    pub struct Credentials<'a> {
+    pub struct Credentials {
-        ip : &'a str,
+        endpoint : Arc<ConfigEndpoint>,
-        login : &'a str,
-        password : &'a str,
-        api_key : &'a str,
-        period : &'a str,
-        timeout : &'a str,
     }
 
-    impl<'a> Credentials<'a> {
+    impl Credentials {
-        pub fn from_config_endpoint(endpoint: &'a ConfigEndpoint) -> Credentials<'a> {
+        pub fn from_config_endpoint(endpoint: Arc<ConfigEndpoint>) -> Credentials {
-            Self {
+            Self { endpoint }
-                ip : &endpoint.ip,
-                login : &endpoint.login,
-                password : &endpoint.password,
-                api_key : &endpoint.api_key,
-                period : &endpoint.period,
-                timeout : &endpoint.timeout,
-            }
         }
     }
 
@@ -225,13 +232,15 @@ pub mod v3 {
         id : String,
         #[serde(rename = "type")]
         json_type : String,
+        addr : String,
         value : Value,
     }
     impl MetricOutput {
-        pub fn new_with_slices(id : &str, json_type : &str, value : Value) -> Self {
+        pub fn new_with_slices(id : &str, json_type : &str, addr: &str,value : Value) -> Self {
             MetricOutput {
                 id : id.to_string(),
                 json_type : json_type.to_string(),
+                addr : addr.to_string(),
                 value : value,
             }
         }