Compare commits

..

No commits in common. "2847a5a1e9f2637e97bd00a3ef0aa1605e589ecd" and "9d2e8100dfd006795b7b9316ebcf0b93a07deab0" have entirely different histories.

9 changed files with 49 additions and 113 deletions

View File

@ -13,7 +13,3 @@ EXPORTER_URL = "http(s)://ip.ip.ip.ip:port"
ENODE_MONITORING_IP = "ip.ip.ip.ip"
ENODE_MONITORING_LOGIN = "admin_user_enode_monitoring" # admin user is required
ENODE_MONITORING_PASSWORD = "admin_password_enode_monitoring" # # admin password is required
# IM configuration for max level of logging info
# for example DEBUG, INFO, WARN, ERROR, TRACE
IM_LOG_INFO = "INFO"

25
Jenkinsfile vendored
View File

@ -1,24 +1,3 @@
def notify(
String context,
String giteaUser,
String giteaPass,
String repositoryUrl,
String repositoryName,
String commitHash,
String buildStatus
) {
def status = buildStatus == 'success' ? 'success' : 'failure'
def description = buildStatus == 'success' ? 'Build succeeded' : 'Build failed'
sh """
curl -X POST \
-u "${giteaUser}:${giteaPass}" \
-H "Content-Type: application/json" \
-d '{"context":"${context}","state": "${status}", "description": "${description}"}' \
${repositoryUrl}deployer3000/${repositoryName}/statuses/${commitHash}
"""
}
pipeline {
agent any
environment {
@ -87,10 +66,6 @@ pipeline {
http://git.entcor/api/v1/repos/deployer3000/integration-module/pulls/${prId}/merge
"""
echo "PR ${prId} merged successfully into master!"
def context = "test-org/integration-module/pipeline/pr-${env.CHANGE_TARGET}"
def commitHash = sh(script: "git rev-parse HEAD~1", returnStdout: true).trim()
notify(context, GITEA_USER, GITEA_PASS, env.GITEA_REPOSITORY_URL, "integration-module", commitHash, "success")
}
}
}

View File

@ -8,6 +8,8 @@ serde = { version = "1.0.217", features = ["derive"] }
serde_json = "1.0.135"
tokio = { version = "1.43.0", features = ["full"] }
integr-structs = {path = "../integr-structs"}
env_logger = "0.11.6"
log = "0.4.25"
anyhow = "1.0.95"
chrono = "0.4.39"
reqwest = { version = "0.12.12", features = ["rustls-tls", "json"] }
@ -18,5 +20,3 @@ md5 = "0.7.0"
rand = "0.9.0"
sysinfo = "0.33.1"
openssl = { version = "0.10", features = ["vendored"] }
tracing-subscriber = "0.3.19"
tracing = "0.1.41"

View File

@ -2,7 +2,7 @@
// 1) check changes in unix-socket
// 2) save changes in local config file
use anyhow::{Error, Ok, Result};
use tracing::{info, warn, error};
use log::{info, warn, error};
use std::{fs, path::Path};
use serde_json::from_str;
use tokio::{io::AsyncReadExt, net::UnixListener};

View File

@ -4,7 +4,7 @@ use reqwest::Client;
use tokio_postgres::NoTls;
use std::env;
use anyhow::Result;
use tracing::{debug, error, info};
use log::{debug, error, info};
use std::ops::Drop;
/// An entity which handles DB connections.
@ -89,7 +89,6 @@ impl Exporter {
}
/// Exports data in `&str` jsonb format to DB using connection from the pool
#[allow(unused)]
#[tracing::instrument(name = "PostgreSQL export")]
pub async fn export_data(client: PgClient, metrics: &str) -> Result<()> {
let query = client.prepare_cached("INSERT INTO metrics (body) VALUES ($1);").await?;
let _ = client.query(&query, &[&metrics]).await?;
@ -97,7 +96,6 @@ impl Exporter {
}
/// Exports metrics in `PrometheusMetrics` format to Exporter defined
/// as env var $EXORPTER_URL
#[tracing::instrument(name = "Prometheus export")]
pub async fn export_metrics(metrics: PrometheusMetrics) -> Result<usize> {
let url = env::var("EXPORTER_URL")?;

View File

@ -1,8 +1,9 @@
use std::str::FromStr;
use chrono::Local;
use env_logger::Builder;
use log::LevelFilter;
use std::io::Write;
use anyhow::Result;
use tracing::info;
use log::info;
/// # Fn `setup_logger`
///
@ -21,34 +22,20 @@ use tracing::info;
/// *depends on* : -
///
pub async fn setup_logger() -> Result<()> {
// Builder::new()
// .format(move |buf, record| {
// writeln!(
// buf,
// "|{}| {} [{}] - {}",
// "api-grubber",
// Local::now().format("%d-%m-%Y %H:%M:%S"),
// record.level(),
// record.args(),
// )
// })
// .filter(None, LevelFilter::Info)
// .target(env_logger::Target::Stdout)
// .init();
let log_level = std::env::var("IM_LOG_INFO").unwrap_or_else(|_| String::from("INFO"));
tracing_subscriber::fmt()
.with_max_level(
tracing::Level::from_str(&log_level)
.unwrap_or_else(|_| tracing::Level::INFO))
.with_writer(std::io::stdout)
.with_span_events(tracing_subscriber::fmt::format::FmtSpan::NEW)
// .with_timer(Local::now().format("%d-%m-%Y %H:%M:%S"))
.with_line_number(false)
.with_target(false)
.with_file(false)
.compact()
.init();
Builder::new()
.format(move |buf, record| {
writeln!(
buf,
"|{}| {} [{}] - {}",
"api-grubber",
Local::now().format("%d-%m-%Y %H:%M:%S"),
record.level(),
record.args(),
)
})
.filter(None, LevelFilter::Info)
.target(env_logger::Target::Stdout)
.init();
info!("Logger configured");
Ok(())
@ -58,17 +45,22 @@ pub async fn setup_logger() -> Result<()> {
#[cfg(test)]
mod logger_unittests {
use tokio::test;
use super::*;
#[test]
async fn check_logger_builder() {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_test_writer()
.with_span_events(tracing_subscriber::fmt::format::FmtSpan::NEW)
// .with_timer(Local::now().format("%d-%m-%Y %H:%M:%S"))
.with_line_number(false)
.with_target(false)
.with_file(false)
.compact()
Builder::new()
.format(move |buf, record| {
writeln!(
buf,
"|{}| {} [{}] - {}",
"api-grubber",
Local::now().format("%d-%m-%Y %H:%M:%S"),
record.level(),
record.args(),
)
})
.filter(None, LevelFilter::Info)
.target(env_logger::Target::Stdout)
.init();
}
}

View File

@ -11,7 +11,7 @@ use logger::setup_logger;
use config::{pull_local_config, init_config_grub_mechanism};
use net::init_api_grub_mechanism;
use tokio::sync::mpsc;
use tracing::{error, info, warn};
use log::{error, info, warn};
use monitoring::get_metrics_from_monitoring;
#[tokio::main(flavor = "multi_thread")]

View File

@ -10,7 +10,7 @@ use tokio::task::JoinHandle;
use std::pin::Pin;
use std::future::Future;
use integr_structs::api::v3::{MetricOutputExtended, PrometheusMetricsExtended};
use tracing::{error, info, warn};
use log::{error, info, warn};
use std::collections::HashMap;
/// # Fn `get_metrics_from_monitoring`
@ -38,18 +38,15 @@ use std::collections::HashMap;
/// assert_eq!(get_metrics_from_monitoring(0, 5).await, Ok(()));
/// ```
///
#[tracing::instrument(name = "CM mechanism", skip_all)]
pub async fn get_metrics_from_monitoring(duration: usize, delay: usize) -> anyhow::Result<()> {
let timer = tokio::time::Instant::now();
let mut a = MonitoringImporter::new().await;
'outer: loop {
// let mut a = MonitoringImporter::new().await;
a.start_session().await?;
tracing::debug!("CM creds struct - {:#?}", a);
info!("Started a new CM session");
let vec = Arc::new(a.get_metrics_list().await.unwrap_or_else(|_| vec![]));
tracing::debug!("Measures Vec - {:#?}", vec);
'inner: loop {
if duration != 0 && timer.elapsed() >= tokio::time::Duration::from_secs(duration as u64) {
@ -66,7 +63,6 @@ pub async fn get_metrics_from_monitoring(duration: usize, delay: usize) -> anyho
tokio::time::sleep(tokio::time::Duration::from_secs(delay as u64)).await
}
}
Ok(())
}
@ -140,7 +136,6 @@ impl MonitoringImporter {
///
/// *Also* it saves ts and access-key in it's runtime environment,
/// there's no way to get access-key of session
#[tracing::instrument(name = "CM-session mechanism", skip_all)]
pub async fn start_session(&mut self) -> anyhow::Result<()> {
if !self.is_valid().await {
return Err(Error::msg("Invalid eNODE-Monitoring configuration"));
@ -148,32 +143,18 @@ impl MonitoringImporter {
let client = Client::new();
let url = format!("http://{}/e-data-front/auth/login", self.ip);
let fortoken = ForTokenCredentials::new(&self.login, &self.password);
let mut delay = 1;
loop {
let client = client
.post(&url)
let client = client
.post(url)
.header("Content-Type", "application/json")
.json(&fortoken);
// let resp = client.send().await?;
if let Ok(resp) = client.send().await {
// let auth = resp.json::<AuthResponse>().await?;
let resp = client.send().await?;
let auth = resp.json::<AuthResponse>().await?;
self.set_ts(&fortoken.ts).await;
self.access_token = auth.access_token.to_owned();
match resp.json::<AuthResponse>().await {
Ok(auth) => {
self.set_ts(&fortoken.ts).await;
self.access_token = auth.access_token.to_owned();
tracing::trace!("Access key was changed");
break;
},
Err(er) => error!("Error with extracting access-key from CM response due to {}", er),
}
}
error!("Error while trying to create a new session, waiting {} secs and retrying ...", delay);
tokio::time::sleep(tokio::time::Duration::from_secs(delay)).await;
delay = delay * 2;
}
info!("Started a new CM session");
Ok(())
}
@ -183,9 +164,7 @@ impl MonitoringImporter {
/// and returning measures in format of `Ok(Vec<(String, String)>)`
/// , where `(String, String)` is a tuple of measure `id` and `description`
/// (`name`)
#[tracing::instrument(name = "CM get metrics list mechanism", skip_all)]
pub async fn get_metrics_list(&self) -> anyhow::Result<Vec<(String, String)>> {
tracing::trace!("Trying ti get measures list from CM ...");
let client = Client::new();
let mut vec: Vec<(String, String)> = Vec::new();
let url = format!("http://{}/e-cmdb/api/query", self.ip);
@ -233,9 +212,7 @@ impl MonitoringImporter {
/// 3) spawns async tasks-grabbers to get measures info which
/// exprots all data by itselfs
///
#[tracing::instrument(name = "CM get measures info mechanism", skip_all)]
pub async fn get_measure_info(&self, measures: Arc<Vec<(String, String)>>) -> anyhow::Result<()> {
tracing::trace!("Trying ti get info about each measure ...");
let mut sys = sysinfo::System::new();
sys.refresh_cpu_all();
// adaptive permition on task spawm to prevent system overload
@ -290,7 +267,6 @@ impl MonitoringImporter {
/// This is a neccesary measure to handle two types of requests and URL restrictions
///
async fn process_endpoint(measure: Arc<String>, client: Arc<Client>, arc: Arc<Self>, hm: &HashMap<String, String>) -> anyhow::Result<PrometheusMetricsExtended> {
tracing::trace!("Processing CM endpoint with one or more measure names");
let resp = client
.get(format!("http://{}/e-nms/mirror/measure/{}", arc.ip, &measure))
.header("Content-Type", "application/json")
@ -345,7 +321,6 @@ impl MonitoringImporter {
/// Searches for certain fields and aggregates it in the `MetricOutputExtended`
/// object
async fn process_value(obj : &Map<String, Value>, hm: &HashMap<String, String>) -> anyhow::Result<MetricOutputExtended> {
tracing::trace!("Processing atomic Object value in CM JSON-response");
let id = obj.get("$id");
let val = obj.get("value");
let description = {

View File

@ -1,7 +1,7 @@
// module to handle unix-socket connection + pulling info from api
use anyhow::Result;
use tracing::{error, info};
use log::{error, info};
use rand::random;
use tokio::sync::mpsc::Receiver;
use tokio::time::{sleep, Duration};