Compare commits

..

No commits in common. "master" and "feature/160" have entirely different histories.

9 changed files with 168 additions and 441 deletions

View File

@ -1,4 +0,0 @@
/target
Cargo.lock
*.sock
.env

View File

@ -1,21 +1,15 @@
# Template .env for API grabber
# 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 = "http(s)://ip.ip.ip.ip:port"
# VINTEO Jitter puller (needed to init Jitter native grab)
# -------------------------------
VINTEO_URL_BASE = "http(s)://ip.ip.ip.ip:port"
VINTEO_ENDPOINT_CONFERENCES = "/api/v1/to/something"
VINTEO_ENDPOINT_PARTICIPANTS = "/api/v1/to/something"
VINTEO_API_KEY = "6fe8b0db-62b4-4065-9c1e-441ec4228341.9acec20bd17d7178f332896f8c006452877a22b8627d089105ed39c5baef9711"
# Status Model API support
@ -23,25 +17,14 @@ VINTEO_API_KEY = "6fe8b0db-62b4-4065-9c1e-441ec4228341.9acec20bd17d7178f332896f8
STATUS_SYSTEM_URL = "http://192.168.2.39:9999/api/input"
# eNODE.Monitoring configuration
# -------------------------------
# eNODE.Monitoring server IP
ENODE_MONITORING_IP = "ip.ip.ip.ip"
# eNODE.Monitoring credentials
ENODE_MONITORING_LOGIN = "admin_user_enode_monitoring" # admin user is required
ENODE_MONITORING_PASSWORD = "admin_password_enode_monitoring" # # admin password is required
# List of target devices
ENODE_TARGET_DEVICES = "18, 19"
# to work with unlimit API-Token
ENODE_API_TOKEN = "sssswwwwaaaaffff"
ENODE_TARGET_DEVICES = "device$18,device$19"
# OPTIONAL SETTINGS
# -------------------------------
# IM configuration for max level of logging info
# for example DEBUG, INFO, WARN, ERROR, TRACE
IM_LOG_INFO = "INFO"
# IM configuration for setting up API connetion
# timeout (in secs). Default value - 10
IM_CONNECTION_TIMEOUT = "10"
# IM configuration for delay of requests
# delay (in secs). Default value - 5
IM_REQUEST_DELAY = "20"

View File

@ -5,11 +5,7 @@ RUN apt update && apt install -y musl-tools
RUN rustup target add x86_64-unknown-linux-musl
COPY . .
ENV CARGO_HTTP_DEBUG=true
ENV CARGO_HTTP_TIMEOUT=100
RUN cargo test --verbose
RUN cargo test
RUN cargo build --release --target=x86_64-unknown-linux-musl
FROM alpine:latest

158
README.md
View File

@ -1,173 +1,39 @@
# Интеграционный модуль для проекта "Буревестник ВКС"
## Описание
`Интеграционный модуль (ИМ)` - Rust-пакет, предоставляющий функционал интеграционного модуля в проекте "Буревестник ВКС", состоящий из бинарных крейтов для:
`integr_mod` - Rust-пакет, предоставляющий функционал интеграционного модуля в проекте "Буревестник ВКС", состоящий из бинарных крейтов для:
- получение данных через API ВКС
- поддержку хранения, валидации и актуализации собственных конфигураций
- предобработку полученных данных и сохранение в БД
- интеграции с `еНОД.Мониторинг`
## Специфика работы
На даннный момент предусмотрено два режима работы:
1) **Нативный** - режим работы, производящий прямой опрос сервиса видео-конференц связи `Vinteo` и соотвествующий процесс `ETL`
2) **Статичный** - режим работы *"посредник"*, когда все метрические данные ВКС `Vinteo` получаются через `REST-Full API` средства `еНОД.Мониторинг`
3) **Системный** - аналогичный **статичному** режиму, но метрические данные (заведомо обогащенные нулевым статусом) отправялются не напрямую в модуль `Prometehus exporter`, а в `Статусную модель`
4) **Vinteo** - особый режим работы, предполагающий сбор определенного набора метрик напрямую с ВКС `Vinteo` механизмом многоэтапного `API-запроса`
> **Примечание**
По стандарту `ИМ` работает в **НАТИВНОМ** режиме и ожидает конфигурации в формате `.json`, однако приоритетным считается **СТАТИЧНЫЙ** режим. Подробная информация о настройке в пункте `Руководство`
- предобработку полученных данных и ~~отправку `Системе Мониторинга`~~ сохранение в БД
## Руководство
В данном разделе опсиан алгоритм настройки, сборки и запуска программного модуля `ИМ`
### Преднастройка
1. Выбор режима работы модуля, который скорректирует принцип настройки:
| Режим работы | .env | config-api.json | $STATUS_SYSTEM_URL | $EXPORTER_URL |
|---|---|---|---|---|
| Нативный | ❌ | ✅ | ❌ | ❌ |
| Статичный | ✅ | ❌ | ❌❌❌ | ✅ |
| Системный | ✅ | ❌ | ✅ | ❌ |
, где:
✅ -- следует настроить (предпринять)
❌ -- игнорируется системой, не стоит настраивать
❌❌❌ -- **НЕЛЬЗЯ** настраивать (предпринимать), возможны ошибки в работе
> Режим работы `Vinteo` *не описан* в таблице **намеренно**
### Настройка режима работы "Нативный"
Для настройки данного режима необходимо расположить в **активной** директории конфигурационный `config_api.json` файл:
``` json
{
"config": [
{
"id":"zvks",
"login" : "",
"pass" : "",
"api_key" : "6fe8b0db-62b4-4065-9c1e-441ec4228341.9acec20bd17d7178f332896f8c006452877a22b8627d089105ed39c5baef9711",
"period" : "",
"timeout" : "5",
"metrics" : [
{
"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",
"url": "https://demo.vcs.vinteo.dev/api/v1/accounts",
"measure": [
{ "id":"total", "type": "integer", "addr": "data.total" }
]
}
]
}
]
}
```
> **Примечание**
Название конфигурационного файла должно быть как в примере - `config_api.json`
### Настройка режима работы "Статичный"
Для настройки данного режима необходимо пополнить данными о сервере в `.env` файле по примеру:
1. Заполнить .env файл или установить переменные окружения в соотвествии с примером в `.env.example` файле
``` toml
...
EXPORTER_URL = "http(s)://ip.ip.ip.ip:port" # <--- экспорт данных (обязательно)
# Template .env for API grabber
# Prometheus-Exporter info
EXPORTER_URL = "http(s)://ip.ip.ip.ip:port"
# eNODE.Monitoring configuration
ENODE_MONITORING_IP = "ip.ip.ip.ip"
# admin user is required
ENODE_MONITORING_LOGIN = "admin_user_enode_monitoring"# ---> получение данных
ENODE_MONITORING_LOGIN = "admin_user_enode_monitoring"
# admin password is required
ENODE_MONITORING_PASSWORD = "admin_password_enode_monitoring"
...
```
### Настройка режима работы "Системный"
Для настройки данного режима необходимо пополнить данными о сервере в `.env` файле по примеру:
``` toml
...
STATUS_SYSTEM_URL = "http(s)://{ip}:{port}/api/input"# <--- экспорт данных
# eNODE.Monitoring configuration
ENODE_MONITORING_IP = "{ip}.{ip}.{ip}.{ip}"
# admin user is required
ENODE_MONITORING_LOGIN = "admin_user_enode_monitoring"# ---> получение данных
# admin password is required
ENODE_MONITORING_PASSWORD = "admin_password_enode_monitoring"
...
```
### Настройка режима работы "Vinteo"
Для работы в данном режиме необходимо установить переменные окружения в соотвествии со списком ниже
``` toml
...
VINTEO_URL_BASE = "https://demo.vcs.vinteo.dev"
VINTEO_ENDPOINT_CONFERENCES = "/api/v1/conferences"
VINTEO_ENDPOINT_PARTICIPANTS = "/api/v1/participants/"
VINTEO_API_KEY = "00000000000111111111.aaaaaaaaaaaaaaabbbbbbbbbbbbb"
...
```
### Настройка экспорта полученных и обработанных данных
Настройка *точки выхода* для полученных и обработанных метрик определеяется установленными в переменных окружения параметрами, варианта два:
1) **Экспорт в статусную модель** в рамках механизма сквозного прохода данных в проекте `Буревестник ВКС`
``` toml
...
STATUS_SYSTEM_URL = "{BASE_URL}/{ROUTE}"
...
```
2) **Экспорт в экспортер или иной потребитель данных**
``` toml
...
EXPORTER_URL = "{BASE_URL}/{ROUTE}"
...
```
> ### **ОЧЕНЬ ВАЖНОЕ ПРИМЕЧАНИЕ**
> ---
> Одновременное использование `$STATUS_SYSTEM_URL` и `$EXPORTER_URL` **НЕДОПУСТИМО** !! Вариант со ссылкой **на статусную модель** является _по стандарту_ **БОЛЕЕ ПРИОРИТЕТНЫМ**, второй затрется, использовать необходимо только один
2. Произвести сборку проекта командой :
``` bash
cargo build --release
```
3. Запустить
> **Debug** версия
> Debug версия
``` bash
cargo run --bin api-grub
```
или
> **Release** версия
> Release версия
``` bash
cargo run --release --bin api-grub
```
@ -176,6 +42,6 @@ cargo run --release --bin api-grub
| Крейт (подмодуль) | Прогресс |
|---|---|
|`api-grub` | ✅✅✅✅✅✅✅✅✅🛠️ |
|`config-delivery` [migrated] | ❌❌❌❌❌❌❌❌❌❌ |
|`config-delivery [migrated]` | ❌❌❌❌❌❌❌❌❌❌ |
|`integrs-structs` | ✅✅✅✅✅✅✅✅✅✅ |
|`preproc` [temp-deprecated] | ❌❌❌❌❌❌❌❌❌❌ | (разработка временно остановлена)

View File

@ -1,6 +1,6 @@
[package]
name = "api-grub"
version = "1.0.15"
version = "1.0.11"
edition = "2021"
authors = ["Vladislav Drozdov <maseeeeeeeed@gmail.com>"]
description = "API poller for ZVKS project"
@ -28,5 +28,3 @@ openssl = { version = "0.10", features = ["vendored"] }
tracing-subscriber = "0.3.19"
tracing = "0.1.41"
lazy_static = "1.5.0"
futures = "0.3.31"
async-stream = "0.3.6"

View File

@ -1,43 +1,22 @@
use std::collections::{HashMap, HashSet};
use integr_structs::api::v3::{PrometheusMetricsExtended, MetricOutputExtended};
use std::sync::Arc;
use integr_structs::api::v3::PrometheusMetricsExtended;
use reqwest::Client;
use lazy_static::lazy_static;
use tracing::{error, info, trace, warn};
use std::pin::Pin;
use tracing::{error, info, warn};
use serde_json::{Value, from_str};
use crate::export::Exporter;
use async_stream::stream;
use futures::{future, pin_mut, stream::Stream, StreamExt};
lazy_static! {
static ref VINTEO_BASE: String = std::env::var("VINTEO_URL_BASE").unwrap_or_else(|_| String::from("https://demo.vcs.vinteo.dev"));
static ref CONFERENCES_ENDPOINT: String = std::env::var("VINTEO_ENDPOINT_CONFERENCES").unwrap_or_else(|_| String::from("/api/v1/conferences"));
static ref USERS_ENDPOINT: String = std::env::var("VINTEO_ENDPOINT_PARTICIPANTS").unwrap_or_else(|_| String::from("/api/v1/participants/"));
static ref CONFERENCES_ENDPOINT: String = String::from("/api/v1/conferences");
static ref USERS_ENDPOINT: String = String::from("/api/v1/participants/");
}
// conferences ids
type Conferences = HashSet<(String, String)>;
type OutputConferences = HashMap<(String, String), Value>;
#[derive(Debug, Clone)]
struct ConferencesDigital(OutputConferences);
impl ConferencesDigital {
pub fn new(item: OutputConferences) -> Self {
Self(item)
}
pub fn go_across(self) -> impl Stream<Item = ((String, String), Arc<Value>)> {
stream! {
for (key, value) in self.0 {
info!("Conference {} ({}) is processing ...", key.0, key.1);
yield (key, Arc::new(value));
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
}
}
}
}
// hash map {CONFERENCE_ID - JITTER_VALUE}
type OutputConferences = HashMap<(String, String), u32>;
// hash map {PARTICIPANT_ID - {CONFERENCE_ID - JITTER_VALUE}}
type Jitter = HashMap<(String, String), OutputConferences>;
// to handle async http requests
#[derive(Debug)]
@ -50,13 +29,12 @@ struct Requester {
impl Requester {
pub fn new() -> anyhow::Result<Self> {
Ok(Self {
base: VINTEO_BASE.clone().to_owned(),
api_key: std::env::var("VINTEO_API_KEY")?,
base: String::from("https://demo.vcs.vinteo.dev"),
api_key: std::env::var("VINTEO_API_KEY").unwrap(),
client: Client::builder().user_agent("api-grub").build()?,
})
}
pub async fn get_conferences(&self) -> anyhow::Result<Value> {
trace!("getting conferences list from API");
let url = format!("{}{}", self.base, *CONFERENCES_ENDPOINT);
let req = self.client.get(url)
.timeout(tokio::time::Duration::from_secs(10))
@ -66,7 +44,6 @@ impl Requester {
Ok(from_str(&resp?.text().await?)?)
}
pub async fn get_conferences_from_value(conferences: Value) -> anyhow::Result<Conferences> {
trace!("extracting conferences list");
let mut hashset = Conferences::new();
match conferences.get("data") {
None => return Err(anyhow::Error::msg("Invalid JSON format: no `data` field")),
@ -102,9 +79,8 @@ impl Requester {
}
Ok(hashset)
}
pub async fn get_users_jitter_by_conference(&self, conferences : Conferences) -> anyhow::Result<OutputConferences> {
trace!("getting users list for each conference from API ...");
let mut output = OutputConferences::new();
pub async fn get_users_jitter_by_conference(&self, conferences : Conferences) -> anyhow::Result<Jitter> {
let mut output = Jitter::new();
for conf in conferences {
let url = format!("{}{}{}", self.base, *USERS_ENDPOINT, &conf.1);
let req = self.client.get(url)
@ -119,7 +95,70 @@ impl Requester {
Err(er) => error!("Cannot convert response into text: {}", er),
Ok(resp) => {
let resp: Value = from_str(&resp)?;
output.entry(conf).or_insert(resp);
if let Some(data) = resp.get("data") {
if let Some(parts) = data.get("participants") {
if let Some(parts) = parts.as_array() {
for participant in parts {
// save id
let id = {
if let Some(id) = participant.get("id") {
let id = serde_json::to_string(id)?.replace("\"", "");
if let Some(watermark) = participant.get("watermark") {
match watermark.as_str() {
Some(watermark) => (watermark.to_string(), id),
None => (String::new(), id),
}
} else {
error!("No `watermark` field was found in participant from {:?} conference. Skipping ...", &conf);
(String::new(), id)
}
} else {
error!("No `id` field was found in participant from {:?} conference. Skipping ...", &conf);
continue;
}
};
if let Some(params) = participant.get("params") {
let jitter = {
if let Some(jitter) = params.get("jBLen") {
// find substring
let temp = serde_json::to_string(jitter)?.replace(" ms", "").replace("\"", "");
// string to u32
if let Ok(jitter) = temp.parse::<u32>() {
jitter
} else {
error!("Invalid type of `jBLen` field for participant {:?} in conference {:?}. Skipping ...", &id, &conf);
continue;
}
} else {
error!("No `jBLen` field was found in participant {:?} from conference {:?}. Skipping ...", &id, &conf);
continue;
}
};
// hm push
output.entry(id)
.and_modify(|map| {
map.entry(conf.clone())
.and_modify(|val| *val = jitter)
.or_insert(jitter); } )
.or_insert( {
let mut hm = OutputConferences::new();
hm.insert(conf.clone(), jitter);
hm
} );
} else {
error!("No `params` field in participant {:?} from conference {:?}. Skipping ...", &id, &conf);
}
}
}
else {
return Err(anyhow::Error::msg("Invalid JSON format: `participants` is not an array"))
}
} else {
return Err(anyhow::Error::msg("Invalid JSON format: no `participants` field"))
}
} else {
return Err(anyhow::Error::msg("Invalid JSON format: no `data` field"))
}
}
}
},
@ -127,94 +166,28 @@ impl Requester {
}
Ok(output)
}
}
fn across_participants(conf: &Value) -> impl Stream<Item = Arc<Value>> + '_ {
trace!("fetching participants for conf and going across ...");
let participants_iter = conf
.get("data")
.and_then(|data| data.get("participants"))
.and_then(|participants| participants.as_array())
.into_iter()
.flatten()
.cloned();
stream! {
for participant in participants_iter {
yield Arc::new(participant);
}
}
}
async fn gather_futures(parts: Arc<Value>, conf_id: Arc<str>, conf_desc: Arc<str>, targets: Vec<(&str, &str)>) -> Vec<Option<MetricOutputExtended>> {
trace!("extracting {} measures for current participant in current conference {} ...", targets.len(), conf_id);
let mut futures: Vec<Pin<Box<dyn futures::Future<Output = Option<MetricOutputExtended>> + Send>>> = Vec::new();
for (target, description) in targets {
futures.push(Box::pin(extract_from_participant(
parts.clone(),
target,
conf_id.clone(),
conf_desc.clone(),
description
)));
}
future::join_all(futures).await
}
// GET participants/{conference_id} data.total
async fn extract_total_participants(conferences : Arc<Value>) -> Value {
if let Some(total) = conferences
.get("data")
.and_then(|data| data.get("total")) {
return total.clone()
}
async fn get_metrics_from_jitter(jitter: Jitter) -> integr_structs::api::v3::PrometheusMetricsExtended {
use integr_structs::api::v3::MetricOutputExtended;
let mut metrics = PrometheusMetricsExtended::new_empty_jitter();
// for ((name, id), confs) in jitter {
// confs.iter()
// .for_each(|((conf_name, conf_id), &value)| {todo!()});
// }
jitter.iter()
.map(|(participant, confs)| (confs.iter(), participant))
.for_each(|(conf_iter, (name, id))| {
conf_iter.for_each(|((conf_name, conf_id), &jitter)| {
let metric_id = format!("j{}{}", id, conf_id);
let desc = format!("Фазовое отклонение на цифровом канале (jBLen) у пользователя `{}` (id: {}) в конференции `{}` (id: {})", name, id, conf_name, conf_id);
let val = serde_json::to_value(jitter).unwrap_or_else(|er| {
error!("Error casting jitter value from participant {} (id: {}), conference - {} (id: {}). Error: {}", name, id, conf_name, conf_id, er);
Value::Null
});
metrics.add(MetricOutputExtended::new_with_slices(&metric_id, &metric_id,"int", "Vinteo native", &desc, None, None,val));
});
});
metrics
}
async fn extract_total_anon_participants(conferences : Arc<Value>) -> Value {
let sum = conferences
.get("data")
.and_then(|data| data.get("participants"))
.and_then(|parts| parts.as_array())
.iter()
.flat_map(|&val| val.iter())
.filter_map(|part| part.get("isAnonymous"))
.filter_map(|is_anon| is_anon.as_bool())
.filter(|is_anon| *is_anon == true)
.count();
Value::Number(sum.into())
}
// GET participants/{conference_id} data.participants[].isAnonymous
async fn extract_from_participant(participant : Arc<Value>, target: &str, conf_id: Arc<str>, conf_desc: Arc<str>, description: &str) -> Option<MetricOutputExtended> {
trace!("extracting `{}` measure for current participant ...", target);
if let Some(value) = participant.get("params")
.and_then(|params| params.get(target)) {
let name = participant.get("watermark").unwrap_or_else(|| &Value::Null).as_str().unwrap_or_else(|| "unknown");
let id = participant.get("id").unwrap_or_else(|| &Value::Null).as_str().unwrap_or_else(|| "unknown");
let metric_name = format!("{}_{}_{}", target, conf_id, id);
return Some(
MetricOutputExtended::new_with_slices(
&metric_name,
&metric_name,
"type",
"Vinteo native",
format!(
"{} для пользователя {} ({}) в конференции {} ({})",
description,
name,
id,
conf_desc,
conf_id
).as_ref(),
Some(18),
Some("module$12".to_string()),
value.clone()
)
)
}
None
}
pub async fn init_grubbing_jitter() -> anyhow::Result<()> {
@ -227,85 +200,12 @@ pub async fn init_grubbing_jitter() -> anyhow::Result<()> {
let conferences = Requester::get_conferences_from_value(conferences).await?;
info!("Getting jitter metrics by user in each conference ...");
let confs = requester.get_users_jitter_by_conference(conferences).await?;
let jitter = requester.get_users_jitter_by_conference(conferences).await?;
info!("Initializing extraction mechanism ...");
let conferences = ConferencesDigital::new(confs);
let extracted_conference = conferences.go_across();
let mut buffer: Vec<MetricOutputExtended> = Vec::new();
pin_mut!(extracted_conference);
info!("Casting jitter metrics into Metrics format for Prometheus ...");
let metrics = Requester::get_metrics_from_jitter(jitter).await;
while let Some(((desc, id), item)) = extracted_conference.next().await {
let parts_stream = across_participants(&item);
pin_mut!(parts_stream);
let total_participants = MetricOutputExtended::new_with_slices(
format!("TotalParticipants_{}", id).as_ref(),
format!("TotalParticipants_{}", id).as_ref(),
"type",
"Vinteo Native",
format!("Общее количество участников в конференции {}", &desc).as_ref(),
Some(18),
Some(String::from("module$12")),
extract_total_participants(item.clone()).await
);
// anon NON_STABLE
let total_anon_participants = MetricOutputExtended::new_with_slices(
format!("TotalAnonymousParticipants_{}", id).as_ref(),
format!("TotalAnonymousParticipants_{}", id).as_ref(),
"type",
"Vinteo Native",
format!("Общее количество анонимных участников в конференции {}", &desc).as_ref(),
Some(18),
Some(String::from("module$12")),
extract_total_anon_participants(item.clone()).await
);
// description
buffer.push(total_participants);
buffer.push(total_anon_participants);
while let Some(participant) = parts_stream.next().await {
buffer.append(
&mut gather_futures(
participant,
Arc::from(id.as_ref()),
Arc::from(desc.as_ref()),
vec![
("txBitrate", "Скорость отправки данных"),
("rxBitrate", "Скорость приёма данных"),
("txFPS", "Количество отправляемых кадров в секунду"),
("rxFPS", "Количество получаемых кадров в секунду"),
("rxLost", "Количество потерянных пакетов на приёме"),
("txLost", "Количество потерянных пакетов на отправке"),
("jBLen", "Фазовое отклонение на цифровом канале"),
("txLostPercent", "Процент потерянных пакетов на отправке"),
("rxLostPercent", "Процент потерянных пакетов на приёме"),
("txH239Bitrate", "Скорость передачи данных в дополнительном видео-потоке"),
("rxH239Bitrate", "Скорость приёма данных в дополнительном видео-потоке"),
("txVideoCodec", "Используемый видеокодек на отправке"),
("rxVideoCodec", "Используемый видеокодек на приёме"),
("txAudioCodec", "Используемый аудиокодек на отправке"),
("rxAudioCodec", "Используемый аудиокодек на приёме"),
("txResolution", "Разрешение передаваемого видео"),
("rxResolution", "Разрешение принимаемого видео"),
("txH239Resolution", "Разрешение дополнительного видео-потока"),
("rxH239Resolution", "Разрешение дополнительного видео-потока на принимающей стороне"),
("duration", "Длительсность сеанса (в минутах)"),
],
).await
.into_iter()
.filter(|metric| metric.is_some())
.map(|metric| metric.unwrap() )
.collect::<Vec<MetricOutputExtended>>()
);
}
}
// let metrics = Requester::get_metrics_from_jitter(jitter).await;
if !buffer.is_empty() {
let metrics = PrometheusMetricsExtended::new_zvks(buffer).await;
if !metrics.metrics.is_empty() {
match Exporter::export_extended_metrics(metrics).await {
Ok(bytes) => info!("Successfully transmitted metrics ({} bytes)", bytes),
Err(er) => error!("Cannot export data: {}", er),
@ -313,6 +213,8 @@ pub async fn init_grubbing_jitter() -> anyhow::Result<()> {
} else {
warn!("Metrics array is empty. Ignoring exporting ...");
}
// Exporter::export_extended_metrics(metrics)
tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
}
// get users' jitter info

View File

@ -1,4 +1,6 @@
use std::str::FromStr;
use chrono::Local;
use anyhow::Result;
use tracing::info;
@ -19,6 +21,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()

View File

@ -36,16 +36,10 @@ async fn main() -> Result<()>{
},
}
});
let request_delay = std::env::var("IM_REQUEST_DELAY")
.unwrap_or_else(|_| String::from("5"))
.parse::<u32>()
.unwrap_or_else(|_| {
warn!("No delay was set, setting up as 5 secs ..."); 5
});
let event_grub = tokio::spawn(async move {
// GRAB USING eNODE.MONITORING API GATEWAY
if std::env::var("ENODE_MONITORING_IP").is_ok() {
match get_metrics_from_monitoring(0, request_delay as usize).await {
match get_metrics_from_monitoring(0, 5).await {
Ok(_) => info!("Grabing (eNODE.Monitoring) task de-initialized"),
Err(er) => error!("Grabing task returned an error : {}", er),
}

View File

@ -79,18 +79,7 @@ pub trait LazyUnzipInstance {
impl LazyUnzipInstance for &[MetricInstance] {
fn lazy_unzip(&self) -> HashMap<String, MetricMeta> {
self.iter().map(
|obj|
(
obj.dola_id.to_string(),
MetricMeta::new(
&obj.name,
&obj.desc,
&obj.device,
&obj.source
)
)
).collect()
self.iter().map(|obj| (obj.dola_id.to_string(), MetricMeta::new(&obj.name, &obj.desc, &obj.device, &obj.source))).collect()
}
}
@ -194,7 +183,6 @@ pub struct MonitoringImporter {
login : String,
password : String,
access_token : String,
api_token : String,
ts : String,
timeout : usize,
}
@ -217,7 +205,6 @@ impl MonitoringImporter {
login : env::var("ENODE_MONITORING_LOGIN").unwrap_or_else(|_| String::new()),
password : env::var("ENODE_MONITORING_PASSWORD").unwrap_or_else(|_| String::new()),
access_token : String::new(),
api_token : env::var("ENODE_API_TOKEN").unwrap_or_else(|_| String::new()),
ts : String::new(),
timeout : std::env::var("IM_CONNECTION_TIMEOUT").unwrap_or_else(|_| "10".to_string()).parse().unwrap_or_else(|_| 10)
}
@ -226,13 +213,7 @@ impl MonitoringImporter {
/// and can be used to pull and push info to and from CM
///
async fn is_valid(&self) -> bool {
!self.ip.is_empty() && ((!self.login.is_empty() && !self.password.is_empty() ) || !self.api_token.is_empty())
}
/// Function that checks is current `MonitoringImporter` valid
/// and can be used to pull and push info to and from CM
///
async fn is_minimal(&self) -> bool {
(self.login.is_empty() || self.password.is_empty()) && !self.api_token.is_empty()
!self.ip.is_empty() && !self.login.is_empty() && !self.password.is_empty()
}
/// A setter of `timestamp`
///
@ -254,15 +235,8 @@ impl MonitoringImporter {
#[tracing::instrument(name = "cm_fn_session_start", skip_all)]
pub async fn start_session(&mut self) -> anyhow::Result<()> {
if !self.is_valid().await {
if self.is_minimal().await {
return Err(Error::msg(format!("Given API-Token is no more actual now ({})", &self.access_token)));
}
return Err(Error::msg("Invalid eNODE-Monitoring configuration"));
}
if !self.api_token.is_empty() {
std::mem::swap(&mut self.access_token, &mut self.api_token);
info!("API-Token that was in the ENODE configuration was set as access-token");
} else {
let client = Client::new();
let url = format!("http://{}/e-data-front/auth/login", self.ip);
let fortoken = ForTokenCredentials::new(&self.login, &self.password);
@ -274,7 +248,10 @@ impl MonitoringImporter {
.timeout(tokio::time::Duration::from_secs(self.timeout as u64))
.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?;
match resp.json::<AuthResponse>().await {
Ok(auth) => {
self.set_ts(&fortoken.ts).await;
@ -290,7 +267,6 @@ impl MonitoringImporter {
delay = delay * 2;
}
info!("Started a new CM session");
}
Ok(())
}