crates/api-grub/src/export.rs

@@ -1,54 +0,0 @@
-use deadpool_postgres::{Config, Pool, Runtime, Client as PgClient};
-use tokio_postgres::NoTls;
-use std::env;
-use anyhow::{Error, Result};
-use log::{info, error};
-
-pub struct Exporter {
-    pool : Option<Pool>,
-}
-
-impl Exporter {
-    fn config_construct() -> Result<Config> {
-        let mut cfg = Config::new();
-        cfg.host = Some(env::var("DB_HOST")?);
-        cfg.dbname = Some(env::var("DB_DBNAME")?);
-        cfg.user = Some(env::var("DB_USER")?);
-        cfg.password = Some(env::var("DB_PASSWORD")?);
-        Ok(cfg)
-    }
-    fn pool_construct() -> Option<Pool> {
-        return match Self::config_construct() {
-            Ok(config) => {
-                if let Ok(pool) = config.create_pool(Some(Runtime::Tokio1), NoTls) {
-                    info!("Connected to PostgreSQL");
-                    return Some(pool);
-                }
-                None
-            },
-            Err(_) => {
-                error!("Bad DB credentials or it's unreachable");
-                None
-            },
-        } 
-    } 
-    pub fn is_no_connection(&self) -> bool { self.pool.is_none() }
-    pub fn init() -> Self {
-        Self {
-            pool : Self::pool_construct()
-        }
-    }
-    pub async fn get_connection_from_pool(&self) -> Option<PgClient> {
-        if let Some(pool) = &self.pool {
-            return Some(pool.get().await.ok()?);
-        }
-        None
-    } 
-    pub async fn export_data(client: PgClient, metrics: &str) -> Result<()> {
-        // client.
-        let query = client.prepare_cached("INSERT INTO metrics (body) VALUES ($1);").await?; 
-        let _ = client.query(&query, &[&metrics]).await?;
-        Ok(())
-    }
-
-}

crates/api-grub/src/json.rs

@@ -1,153 +0,0 @@
-// use serde::{de::value, Serialize};
-use serde_json::{json, Value};
-use integr_structs::api::v3::{Metric, MetricOutput};
-
-pub struct JsonParser;
-
-impl JsonParser {
-    pub fn parse(targets: &Vec<Metric>, json: &str) -> Value {
-        let mut res_vec: Vec<MetricOutput> = Vec::new(); 
-        for target in targets {
-            let metric = match target.addr.contains("[") {
-                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));
-        }  
-        serde_json::to_value(res_vec).unwrap_or(Value::Null)
-    }
-    fn get_sum_of_metrics_in_array(target: &Metric, json: &str) -> Value {
-        if target.addr.is_empty() {
-            return Value::Null;
-        }
-
-        let mut vec_value: Vec<Value> = Vec::new();
-        let mut array_key = String::new();
-        let mut value_json: Value = serde_json::from_str(json).unwrap_or(Value::Null);
-
-        let target_attr_vec = target.addr
-            .split_terminator('.')
-            .collect::<Vec<&str>>();
-        // for keys in [] brackets
-        let mut key_tag = String::new();
-
-        for (global_idx, &key) in target_attr_vec.iter().enumerate() {
-            // if array 
-            let key_checked = if key.contains('[') {
-                let key_idx = key.find("[").unwrap();
-                key_tag = key.chars()
-                    .enumerate()
-                    .filter(|(idx, chr)| *idx > key_idx && *chr != ']')
-                    .map(|(_, chr)| chr)
-                    .collect::<String>();
-                // dbg!(&key_tag);
-                key.chars()
-                    .enumerate()
-                    .filter(|(idx, _)| *idx < key_idx)
-                    .map(|(_, chr)| chr)
-                    .collect::<String>()
-                // dbg!(value_json.get(&array_key).unwrap_or(&Value::Null));
-                // value_json = value_json.get(array_key).unwrap_or(&Value::Null).clone();
-                // continue;
-                // new_key.as_str()
-                // dbg!(key);       APPROVED 
-            // TODO: need to check key in [] type of [KEY]
-            } else {key.to_owned()};
-            
-            // if already array 
-            match value_json.get(key_checked) {
-                Some(val) => {
-                    match val {
-                        Value::Array(array) => {
-                            // form new target array
-                            let new_array_target = target_attr_vec
-                                .iter()
-                                .enumerate()
-                                .filter(|(idx, _)| *idx > global_idx)
-                                .map(|(_, &chr)| chr.to_owned())
-                                .collect::<Vec<String>>();
-                            // get_values_in_array
-                            // get_tags_in_array
-                            // // slice_tags_with_values
-                            // dbg!(&array);
-                            // dbg!(&new_array_target);
-                            let res_arr = Self::get_values_in_array(array, &new_array_target);
-                            if &key_tag == "" {
-                                return res_arr.into();
-                            }
-                            return Self::slice_with_tags_in_array(array, &res_arr, &key_tag).into()
-                            
-                        },
-                        _ => value_json = val.clone(),
-                    }
-                },
-                None => return Value::Null,
-            }
-        }
-        value_json
-    }
-
-    fn slice_with_tags_in_array(array: &Vec<Value>, metrics: &Vec<Value>, tag_name: &str) -> Vec<Value> {
-        if tag_name.is_empty() {
-            return metrics.clone();
-        }
-        // array[0].as_object().unwrap_or(json!(Value::Null))
-        let mut values: Vec<Value> = Vec::new();
-        array.iter()
-            .enumerate()
-            .map(|(idx, val)| {
-                let val = val.get(tag_name).unwrap_or(&Value::Null).clone();
-                (serde_json::from_value::<serde_json::Map<String, Value>>(json!({tag_name: val})), 
-                serde_json::from_value::<serde_json::Map<String, Value>>(metrics[idx].clone()))
-            })
-            .for_each(|(key, val)| {
-                // dbg!(&key);
-                if key.is_ok() && val.is_ok() {
-                    let mut key = key.unwrap();
-                    let mut val = val.unwrap();
-                    key.append(&mut val);
-                    values.push(json!(key));
-                }
-            });
-        if values.len() == 0 {
-            return metrics.clone();
-        }
-        values
-    }
-    fn get_values_in_array(array: &Vec<Value>, fields: &Vec<String>) -> Vec<Value> {
-        let mut values: Vec<Value> = Vec::new();
-        for obj in array {
-            // dbg!(obj);
-            let mut obrez = obj.clone();
-            for field in fields {
-                obrez = obrez.get(field).unwrap_or(&Value::Null).clone();
-                match obrez {
-                    Value::Object(_) => {continue},
-                    _ => {
-                        values.push(json!({field: obrez.clone()}));
-                    },
-                    // None => {values.push(Value::Null)},
-                }
-            }
-        }
-        values
-    }
-
-    fn get_metric(target: &Metric, json: &str) -> Value {
-        if target.addr.is_empty() {
-            return Value::Null;
-        }
-        let mut value_json: Value = serde_json::from_str(json).unwrap_or(Value::Null);
-
-        let target_attr_vec = target.addr
-            .split_terminator('.')
-            .collect::<Vec<&str>>();
-        for key in target_attr_vec {
-            match value_json.get(key) {
-                Some(val) => value_json = val.clone(),
-                None => return Value::Null,
-            }
-        }
-        value_json    
-    }
-}

crates/api-grub/src/main.rs

@@ -1,17 +1,14 @@
 mod config;
 mod net;
 mod logger;
-mod json;
-mod export;
 
 use anyhow::Result;
 use integr_structs::api::ApiConfigV2;
 use logger::setup_logger;
-// use log::{info, warn};
+use log::{info, warn};
 use config::{pull_local_config, init_config_grub_mechanism};
 use net::init_api_grub_mechanism;
 use tokio::sync::mpsc;
-use log::{error, info, warn};
 
 #[tokio::main(flavor = "multi_thread")]
 async fn main() -> Result<()>{
@@ -25,34 +22,10 @@ async fn main() -> Result<()>{
     let (tx, mut rx) = mpsc::channel::<ApiConfigV2>(1);
     // futures
     // todo : rewrite with spawn
-    // let config_fut = init_config_grub_mechanism(&tx);
+    let config_fut = init_config_grub_mechanism(&tx);
-    // let grub_fut = init_api_grub_mechanism(config, &mut rx);
+    let grub_fut = init_api_grub_mechanism(config, &mut rx);
 
-    let event_config = tokio::spawn(async move {
+    let _ = tokio::join!(config_fut, grub_fut);
-        match init_config_grub_mechanism(&tx).await {
-            Ok(_) => {
-                info!("Config task deinitialized");
-            },
-            Err(er) => {
-                error!("Config task returned an error : {}", er);
-            },
-        }
-    });
-    let event_grub = tokio::spawn(async move {
-        match init_api_grub_mechanism(config, &mut rx).await {
-            Ok(_) => {
-                info!("Grabing task deinitialized");
-            },
-            Err(er) => {
-                error!("Grabing task returned an error : {}", er);
-            },
-        }
-    });
-    let events_handler = vec![event_config, event_grub];
-    for event in events_handler {
-        let _ = event.await;
-    }
-    // let _ = tokio::join!(config_fut, grub_fut);
 
     Ok(())
 }

crates/api-grub/src/net.rs

@@ -8,41 +8,61 @@ use reqwest::{Client, Method};
 use std::sync::Arc;
 use tokio::task::JoinHandle;
 // use tokio::sync::Mutex;
+use deadpool_postgres::{Config, Pool, Runtime, Client as PgClient};
+use tokio_postgres::NoTls;
 use dotenv::dotenv;
-use crate::json::JsonParser;
+use std::env;
-use crate::export::Exporter;
-use integr_structs::api::v3::Config;
 
 // type BufferType = Arc<Mutex<Vec<String>>>; 
 
-// for api info pulling 
+struct Exporter {
-pub async fn init_api_grub_mechanism(config: ApiConfigV2, rx: &mut Receiver<ApiConfigV2>) -> Result<()> {
+    pool : Option<Pool>,
-    info!("Initializing API-info grubbing mechanism...");
-    info!("Loading vars from .env file if exists...");
-    let _ = dotenv().ok();
-
-    let mut config = config;
-    let mut poller = ApiPoll::new(&mut config).await;
-    let client = Exporter::init(); 
-    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 {
-            if rx.len() > 0 {
-                if let Some(conf) = rx.recv().await {
-                    poller.change_config(conf).await;
-                    info!("Config changed");
-                }
-            }
-            info!("Data from API: {:?}", poller.process_polling(shared_pool).await);
-            sleep(Duration::from_secs(poller.get_delay().await as u64)).await;
-        }
-    }
-    // Ok(())
 }
 
+impl Exporter {
+    fn config_construct() -> Result<Config> {
+        let mut cfg = Config::new();
+        cfg.host = Some(env::var("DB_HOST")?);
+        cfg.dbname = Some(env::var("DB_DBNAME")?);
+        cfg.user = Some(env::var("DB_USER")?);
+        cfg.password = Some(env::var("DB_PASSWORD")?);
+        Ok(cfg)
+    }
+    fn pool_construct() -> Option<Pool> {
+        return match Self::config_construct() {
+            Ok(config) => {
+                if let Ok(pool) = config.create_pool(Some(Runtime::Tokio1), NoTls) {
+                    info!("Connected to PostgreSQL");
+                    return Some(pool);
+                }
+                None
+            },
+            Err(_) => {
+                error!("Bad DB credentials or it's unreachable");
+                None
+            },
+        } 
+    } 
+    pub fn is_no_connection(&self) -> bool { self.pool.is_none() }
+    pub fn init() -> Self {
+        Self {
+            pool : Self::pool_construct()
+        }
+    }
+    pub async fn get_connection_from_pool(&self) -> Option<PgClient> {
+        if let Some(pool) = &self.pool {
+            return Some(pool.get().await.ok()?);
+        }
+        None
+    } 
+    pub async fn export_data(client: PgClient, metrics: &str) -> Result<()> {
+        // client.
+        let query = client.prepare_cached("INSERT INTO metrics (body) VALUES ($1);").await?; 
+        let _ = client.query(&query, &[&metrics]).await?;
+        Ok(())
+    }
+
+}
 
 struct RestMethod;
 
@@ -167,6 +187,34 @@ impl<'a> ApiPoll<'a> {
     }
 }
 
+// for api info pulling 
+pub async fn init_api_grub_mechanism(config: ApiConfigV2, rx: &mut Receiver<ApiConfigV2>) -> Result<()> {
+    info!("Initializing API-info grubbing mechanism...");
+    info!("Loading vars from .env file if exists...");
+    let _ = dotenv().ok();
+
+    let mut config = config;
+    let mut poller = ApiPoll::new(&mut config).await;
+    let client = Exporter::init(); 
+    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 {
+            if rx.len() > 0 {
+                if let Some(conf) = rx.recv().await {
+                    poller.change_config(conf).await;
+                    info!("Config changed");
+                }
+            }
+            info!("Data from API: {:?}", poller.process_polling(shared_pool).await);
+            sleep(Duration::from_secs(poller.get_delay().await as u64)).await;
+        }
+    }
+    // Ok(())
+}
+
 #[cfg(test)]
 mod net_unittests {
     use super::*;

crates/integr-structs/src/api.rs

@@ -145,95 +145,3 @@ impl ProcessedEndpoint {
         Ok(to_string_pretty(&val)?)
     }
 }
-
-pub mod v3 {
-    pub use super::*;
-    
-    // in config
-    #[derive(Deserialize)]  
-    pub struct Metric {
-        pub id : String,
-        #[serde(rename = "type")]
-        pub json_type : String,
-        pub addr : String,
-    }
-    #[derive(Deserialize)]
-    pub struct Metrics {
-        pub name : String,
-        pub url : String,
-        #[serde(default)]
-        pub measure : Vec<Metric>
-    }
-
-    #[derive(Deserialize)]
-    pub struct ConfigEndpoint {
-        ip : String,
-        login : String,
-        #[serde(rename = "pass")]
-        password : String,
-        api_key : String,
-        period : String,
-        timeout : String,
-        #[serde(default)]
-        metrics : Vec<Metrics>, 
-    }
-
-    #[derive(Deserialize)]
-    pub struct Config {
-        config : Vec<ConfigEndpoint>,
-    }
-
-    impl Default for Config {
-        fn default() -> Self {
-            Self {
-                config : Vec::new()
-            }
-        }
-    }
-
-    impl Config {
-        pub async fn is_default(&self) -> bool {
-            self.config.is_empty()
-        }
-    }
-
-    pub struct Credentials<'a> {
-        ip : &'a str,
-        login : &'a str,
-        password : &'a str,
-        api_key : &'a str,
-        period : &'a str,
-        timeout : &'a str,
-    }
-
-    impl<'a> Credentials<'a> {
-        pub fn from_config_endpoint(endpoint: &'a ConfigEndpoint) -> Credentials<'a> {
-            Self {
-                ip : &endpoint.ip,
-                login : &endpoint.login,
-                password : &endpoint.password,
-                api_key : &endpoint.api_key,
-                period : &endpoint.period,
-                timeout : &endpoint.timeout,
-            }
-        }
-    }
-
-    // to prometheus and nmns
-    #[derive(Serialize, Deserialize)]
-    pub struct MetricOutput {
-        id : String,
-        #[serde(rename = "type")]
-        json_type : String,
-        value : Value,
-    }
-    impl MetricOutput {
-        pub fn new_with_slices(id : &str, json_type : &str, value : Value) -> Self {
-            MetricOutput {
-                id : id.to_string(),
-                json_type : json_type.to_string(),
-                value : value,
-            }
-        } 
-    }
-}

template_global_config.json

@@ -1,32 +1,31 @@
 { 
-    "config":  [     
+	"id" : 1 ,
-		{
+	"template" : 	
-			"id":"demo_vcs_vinteo_dev_api", 
+		[{
-			"login" : "",
+			"id" :"mock_api_1", 
-			"pass" : "",
+			"name" : "Mock / ",
-			"api_key" : "6fe8b0db-62b4-4065-9c1e-441ec4228341.9acec20bd17d7178f332896f8c006452877a22b8627d089105ed39c5baef9711",
+			"url" : "http://127.0.0.1:8081/",
-			"period" : "",
+			"method" : "GET",
-			"timeout" : "3",
+			"measure" : 
-			"metrics" : [
+				[
-				{
+					"operation", "response", "empty_field"
-					"name": "conferences", 
-					"url": "https://demo.vcs.vinteo.dev/api/v1/conferences",
-					"measure": 	[
-						{ "id":"number",  "type": "text",  "addr": "data.conferences[].number" },
-						{ "id":"total",  "type": "integer",  "addr": "data.total" },
-						{ "id":"participants_total",  "type": "integer",  "addr": "data.conferences[].participants.total" },
-						{ "id":"parts_total_in_each",  "type": "integer",  "addr": "data.conferences[description].participants.total" },
-						{ "id":"participants_online",  "type": "integer",  "addr": "data.conferences[].participants.online" }
 				]
 		},
 		{
-					"name": "abonents", 
+			"id" :"mock_api_2", 
-					"url": "https://demo.vcs.vinteo.dev/api/v1/accounts",
+			"name" : "Mock /ping ",
-					"measure": 	[
+			"url" : "http://127.0.0.1:8081/ping",
-						{ "id":"total",	"type": "integer", "addr": "data.total" }
+			"method" : "GET",
+			"measure" : 
+				[
+					"operation", "response", "empty_field"
 				]
 		}
-			]
+		],
-        }
+	"ip_address" : 	"127.0.0.1:8081",
-    ]
+	"login" : "",
+	"pass" : "" ,
+	"api_key" : 	"908c709827bd40n98r7209837x98273",
+	"period" : 		10,
+	"timeout" : 2
 }