.env.example

@@ -2,11 +2,7 @@ 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
-
-# Prometheus-Exporter info
-EXPORTER_URL = "ip.ip.ip.ip:port" 

config_api.json

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

crates/api-grub/src/config.rs

@@ -10,13 +10,12 @@ 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<Config> {
+pub async fn pull_local_config() -> Result<ApiConfigV2> {
     // let conf_path = std::env::current_exe()?;
     let path = Path::new(CONFIG_PATH);
     // return match conf_path.parent() {
@@ -29,7 +28,7 @@ pub async fn pull_local_config() -> Result<Config> {
     //     None => Err(Error::msg("No local conf was found")) 
     // }
     if path.exists() && path.is_file() {
-        let config: Config = from_str(
+        let config: ApiConfigV2 = from_str(
             &fs::read_to_string(CONFIG_PATH)?
         )?;
         Ok(config)
@@ -40,7 +39,7 @@ pub async fn pull_local_config() -> Result<Config> {
 
 // for config pulling
 // ++++ reader to channel 
-pub async fn init_config_grub_mechanism(tx: &Sender<Config>) -> Result<()> {
+pub async fn init_config_grub_mechanism(tx: &Sender<ApiConfigV2>) -> Result<()> {
     info!("Initializing Unix-Socket listening for pulling new configs...");
     let server = init_unix_listener().await?;
     // 
@@ -52,7 +51,7 @@ pub async fn init_config_grub_mechanism(tx: &Sender<Config>) -> 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<Config, serde_json::Error> = from_str(&buffer);
+                let config: Result<ApiConfigV2, 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 {
@@ -98,13 +97,13 @@ mod config_unittests {
     #[test]
     async fn check_save_new_config() {
         use std::fs;
-        use integr_structs::api::v3::Config;
+        use integr_structs::api::ApiConfigV2;
         use serde_json::to_string;
         
         let test_config_path  = "test_config_api.json";
 
         // config gen  
-        let config = to_string::<Config>(&Config::default());
+        let config = to_string::<ApiConfigV2>(&ApiConfigV2::default());
         assert!(config.is_ok());
         let config = config.unwrap();
 

crates/api-grub/src/export.rs

@@ -1,8 +1,7 @@
 use deadpool_postgres::{Config, Pool, Runtime, Client as PgClient};
-use reqwest::Client;
 use tokio_postgres::NoTls;
 use std::env;
-use anyhow::{Result};
+use anyhow::{Error, Result};
 use log::{info, error};
 
 pub struct Exporter {
@@ -36,7 +35,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> {
@@ -51,16 +50,5 @@ 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, &target.addr, metric));
+            res_vec.push(MetricOutput::new_with_slices(&target.id, &target.json_type, metric));
         }  
         serde_json::to_value(res_vec).unwrap_or(Value::Null)
     }

crates/api-grub/src/main.rs

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

crates/api-grub/src/net.rs

@@ -10,24 +10,23 @@ use tokio::task::JoinHandle;
 // use tokio::sync::Mutex;
 use dotenv::dotenv;
 use crate::json::JsonParser;
-use crate::export::{self, Exporter};
+use crate::export::Exporter;
-use integr_structs::api::v3::{Config, ConfigEndpoint, Credentials};
+use integr_structs::api::v3::Config;
 
 // type BufferType = Arc<Mutex<Vec<String>>>; 
 
 // for api info pulling 
-pub async fn init_api_grub_mechanism(config: Config, rx: &mut Receiver<Config>) -> Result<()> {
+pub async fn init_api_grub_mechanism(config: ApiConfigV2, rx: &mut Receiver<ApiConfigV2>) -> Result<()> {
-    info!("Initializing API-info grubbing mechanism :");
+    info!("Initializing API-info grubbing mechanism...");
-    info!("1) Loading vars from .env file if exists...");
+    info!("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 {
@@ -37,9 +36,8 @@ pub async fn init_api_grub_mechanism(config: Config, rx: &mut Receiver<Config>)
                     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(())
@@ -64,188 +62,156 @@ impl RestMethod {
     } 
 }
 struct ApiPoll<'a> {
-    config : &'a mut Config,
+    config : &'a mut ApiConfigV2,
     client : Client,
 }
 
 impl<'a> ApiPoll<'a> {
-    pub async fn new(poll_cfg : &'a mut Config) -> Self {
+    pub async fn new(poll_cfg : &'a mut ApiConfigV2) -> Self {
         Self {
             config : poll_cfg,
             client : Client::new(),
         }
     }
     // can be weak and with bug test needed
-    pub async fn change_config(&mut self, conf: Config) {
+    pub async fn change_config(&mut self, conf: ApiConfigV2) {
         *self.config = conf;
     }
     pub async fn is_default(&self) -> bool {
-        self.config.is_default().await
+        self.config.template.len() == 0
-    }
-    // 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 client = Arc::new(self.client.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![];
+        let client = Arc::new(self.client.clone());
+        let template = Arc::new(self.config.template.clone());
 
-        for (idx, _) in config.config.iter().enumerate() {
+        if self.is_default().await { return Err(Error::msg("Default config with no endpoints")) }
-            // let for_creds = endpoints[idx].clone();
+
-            let creds = Credentials::from_config_endpoint(endpoints[idx].clone());
+        // TODO: rewrite nextly to async 
-            let endpoint = endpoints[idx].clone();
+        for point in template.iter() {
+            let point = Arc::new(point.clone());
+            // let buffer = buffer.clone();
             let client = client.clone();
             let exporter = exporter.clone();
-            let join_handler = tokio::spawn(async move {
+            let endpoint_processer = tokio::spawn(async move {
-                Self::process_endpoint(
+                let point = point.clone();
-                    client, 
+                match client.request(RestMethod::from_str(&point.method).await, &point.url).send().await {
-                    endpoint, 
+                    Ok(resp) => {
-                    creds, 
+                        if !resp.status().is_success() {
-                    exporter.clone()
+                            error!("ErrorCode in Response from API. Check configuration");
-                ).await
+                            return Err(Error::msg("Error during sending request"));
-            });
-            join_handles.push(join_handler);
                         }
+                        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 template = Arc::new(self.config.template.clone());
 
-        // if self.is_default().await { return Err(Error::msg("Default config with no endpoints")) }
+        // let buffer = buffer.lock().await;
-
+        // match &buffer.len() {
-        // // TODO: rewrite nextly to async 
+        //     0 => Err(Error::msg("Error due to API grubbing. Check config" )),
-        // 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,9 +1,7 @@
-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)]
@@ -152,14 +150,14 @@ pub mod v3 {
     pub use super::*;
     
     // in config
-    #[derive(Serialize, Deserialize, Clone)]  
+    #[derive(Deserialize)]  
     pub struct Metric {
         pub id : String,
         #[serde(rename = "type")]
         pub json_type : String,
         pub addr : String,
     }
-    #[derive(Serialize, Deserialize, Clone)]
+    #[derive(Deserialize)]
     pub struct Metrics {
         pub name : String,
         pub url : String,
@@ -167,9 +165,9 @@ pub mod v3 {
         pub measure : Vec<Metric>
     }
 
-    #[derive(Serialize, Deserialize, Clone)]
+    #[derive(Deserialize)]
     pub struct ConfigEndpoint {
-        id : String,
+        ip : String,
         login : String,
         #[serde(rename = "pass")]
         password : String,
@@ -179,27 +177,10 @@ 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(Serialize, Deserialize, Clone)]
+    #[derive(Deserialize)]
     pub struct Config {
-        pub config : Vec<ConfigEndpoint>,
+        config : Vec<ConfigEndpoint>,
     }
 
     impl Default for Config {
@@ -216,13 +197,25 @@ pub mod v3 {
         }
     }
 
-    pub struct Credentials {
+    pub struct Credentials<'a> {
-        endpoint : Arc<ConfigEndpoint>,
+        ip : &'a str,
+        login : &'a str,
+        password : &'a str,
+        api_key : &'a str,
+        period : &'a str,
+        timeout : &'a str,
     }
 
-    impl Credentials {
+    impl<'a> Credentials<'a> {
-        pub fn from_config_endpoint(endpoint: Arc<ConfigEndpoint>) -> Credentials {
+        pub fn from_config_endpoint(endpoint: &'a ConfigEndpoint) -> Credentials<'a> {
-            Self { endpoint }
+            Self {
+                ip : &endpoint.ip,
+                login : &endpoint.login,
+                password : &endpoint.password,
+                api_key : &endpoint.api_key,
+                period : &endpoint.period,
+                timeout : &endpoint.timeout,
+            }
         }
     }
 
@@ -232,15 +225,13 @@ 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, addr: &str,value : Value) -> Self {
+        pub fn new_with_slices(id : &str, json_type : &str, value : Value) -> Self {
             MetricOutput {
                 id : id.to_string(),
                 json_type : json_type.to_string(),
-                addr : addr.to_string(),
                 value : value,
             }
         }