deployer3000
/
prometheus-exporter
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 6 Wiki Activity

Compare commits

base: deployer3000:master
Branches Tags
deployer3000:master
deployer3000:rc
deployer3000:feature/161
deployer3000:feature/1184
deployer3000:feature/readme
deployer3000:feature/1126
deployer3000:feature/#1108
deployer3000:feature/cm-metrics
deployer3000:0.2.10
deployer3000:0.2.9
deployer3000:0.2.8
deployer3000:0.2.7
deployer3000:0.2.6
deployer3000:0.2.5
deployer3000:0.2.4
deployer3000:0.2.3
deployer3000:0.2.0
deployer3000:0.1.0
..
compare: deployer3000:feature/1126
Branches Tags
deployer3000:master
deployer3000:rc
deployer3000:feature/161
deployer3000:feature/1184
deployer3000:feature/readme
deployer3000:feature/1126
deployer3000:feature/#1108
deployer3000:feature/cm-metrics
deployer3000:0.2.10
deployer3000:0.2.9
deployer3000:0.2.8
deployer3000:0.2.7
deployer3000:0.2.6
deployer3000:0.2.5
deployer3000:0.2.4
deployer3000:0.2.3
deployer3000:0.2.0
deployer3000:0.1.0

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

7 changed files with 216 additions and 456 deletions
Show Stats Expand all files Collapse all files

12
.env.example
View File

@ -1,13 +1 @@
# port binding for prometheus-exporter
# default value = 9100
PROMETHEUS_EXPORTER_PORT = 9100
# prefix for metric naming
# default value = "zvks"
PROMETHEUS_EXPORTER_PREFIX = "zvks"
# setting up max level of logging
# default - INFO
# values : WARN, ERROR, INFO, DEBUG, TRACE
PROMETHEUS_EXPORTER_LOG_LEVEL = "TRACE"

56
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 {
@ -34,22 +13,7 @@ pipeline {
}
steps {
script {
def hasTags = sh(script: "git tag -l | wc -l", returnStdout: true).trim().toInteger() > 0
def lastVersion = "0.0.0"
if (hasTags) {
lastVersion = sh(script: "git describe --tags --abbrev=0", returnStdout: true).trim()
}
echo "Last version: ${lastVersion}"
def (major, minor, patch) = lastVersion.tokenize('.')
def newVersion = "${major}.${minor}.${patch.toInteger() + 1}"
echo "New version: ${newVersion}"
env.IMAGE_TAG = newVersion
env.NEW_VERSION = newVersion
env.IMAGE_TAG = sh(script: "git describe --tags --abbrev=0", returnStdout: true).trim()
}
}
}
@ -94,30 +58,14 @@ pipeline {
echo "Attempting to merge PR ${env.CHANGE_ID} into master..."
withCredentials([usernamePassword(credentialsId: 'gitea_creds', usernameVariable: 'GITEA_USER', passwordVariable: 'GITEA_PASS')]) {
def prId = env.CHANGE_ID
sh """
curl -X POST \
-u "${GITEA_USER}:${GITEA_PASS}" \
-H "Content-Type: application/json" \
-d '{"do":"merge"}' \
http://git.entcor/api/v1/repos/deployer3000/${env.IMAGE_NAME}/pulls/${prId}/merge
http://git.entcor/api/v1/repos/deployer3000/prometheus-exporter/pulls/${prId}/merge
"""
def commitHash = sh(script: "git rev-parse HEAD~1", returnStdout: true).trim() // необходим для корректного отображения статусов
echo "PR ${prId} merged successfully into master!"
sleep(time: 15, unit: 'SECONDS')
sh "git checkout master && git pull origin master"
sh """
curl -v -X POST -u "${GITEA_USER}:${GITEA_PASS}" \
-H "Content-Type: application/json" \
-d '{"tag_name": "${env.NEW_VERSION}", "name": "Release ${env.NEW_VERSION}", "target_commitish": "master"}' \
"${env.GITEA_REPOSITORY_URL}deployer3000/${env.IMAGE_NAME}/releases"
"""
echo "New release succeeded!"
def context = "test-org/${env.IMAGE_NAME}/pipeline/pr-${env.CHANGE_TARGET}"
notify(context, GITEA_USER, GITEA_PASS, env.GITEA_REPOSITORY_URL, env.IMAGE_NAME, commitHash, "success")
}
}
}

114
README.md
View File

@ -1,115 +1,3 @@
# Prometheus exporter
Программный компонент проекта "Буревестник ВКС", предоставляющий функционал
"кастомного" `Prometheus` экспортера для обработки метрических данных
### Функциональные особенности
1) приемка метрических данных с "Интеграционного модуля" в особом формате
2) приведение принятых метрических данных к виду `Prometheus`-метрик
3) хранение и актуализация метрических данных формата `Prometheus`-метрик в локальном `Registry`
4) предоставления веб-интерфейса (аналогичного интерфейсу стандартного `Prometheus` экспортера) для получения
### Прикладные ограничения локальной сборки пакета
Для локальной сборки Rust-пакета `prometheus-exporter` необходимы:
1) `cargo` (>=1.85.0)
2) `rustc` (>=1.85.0)
3) `rustup` (>=1.28.1) для выбора `target` ОС и архитектуры сборки
### Руководство по сборке пакета
1) На первом этапе следует удостовериться в отсутствии багов и неточностей в работе модуля предвариетльно запустив юнит-тесты командой:
```
cargo test
```
> Пакетный менеджер `cargo` автоматически соберет проект в необходимой для тестирования конфигурации и запустит тесты, в терминале будут отображены результаты
>> ВАЖНО! При возникновении какой-либо ошибки в процессе тестирования убедительная просьба - оставить соответствующий `Issue` в данном репозитории и не использовать пакет до устранения проблемы
2) Сама сборка Rust-пакета `prometheus-exporter` осуществляется введением команды
* для сборки `Debug`-версии проекта
``` bash
cargo build
```
* для сборки `Release`-версии проекта
``` bash
cargo build --release
```
### Запуск проекта
Для запуска проекта необходимо воспользоваться одной из команд:
* для запуска `Debug`-версии проекта
``` bash
cargo run
```
* для запуска `Release`-версии проекта
``` bash
cargo run --release
```
> Если необходимо работать с бинарными файлами, их можно найти по пути `<путь_до_пакета/target/<debug или release>/таргет_сборки/>`
### Преднастройка запуска
Преднастройка пакета `prometheus-exporter` осуществляется внутри `.env` файла по примеру, описанному в файле `.env.example`
``` toml
# port binding for prometheus-exporter
# default value = 9100
PROMETHEUS_EXPORTER_PORT = 9100
```
### Проверка работоспособности
С точки зрения специфики программного модуля `Prometehus exporter` существует **2 способа** проверки работоспособности
#### Прикладные требования для проверки работоспособности `prometheus-exporter`
* наличие утилиты `curl` или ближайшего аналога
#### Способы
1) Попытка получить данные
``` bash
curl -X GET http://localhost:9100/metrics
```
2) Попытка загрузить тестовую метрику
Вид тестовой метрики по формату который принимает пакет `prometheus-exporter`:
``` json
{
"service_name" : "test",
"endpoint_name" : "test",
"metrics" : [{
"id" : "test",
"type" : "",
"addr" : "test",
"value" : 12,
"description" : "test help",
}]
}
```
Команда загрузки тестовой метрики в `prometheus-exporter`:
``` bash
curl -X POST -H 'Content-Type: application/json' 'http://localhost:9100/update' -d '{ "service_name" : "test", "endpoint_name" : "test", "metrics" : [{ "id" : "test", "type" : "", "addr" : "test", "value" : 12, "description" : "test help" }]}'
```
Ответом от сервера если метрика загрузится будет:
```
Ok
```
to handle updates of ZVKS metrics in local Registry and to share with Prometheus later

291
src/endpoints.rs
View File

@ -4,66 +4,11 @@ use axum::{
http
};
use crate::structs::v3::PrometheusMetrics;
use prometheus::{ core::Collector, Encoder, GaugeVec, Registry, TextEncoder, proto::Metric};
use std::{collections::HashMap, fmt::Display, sync::{ Arc, MutexGuard }};
use prometheus::{ Encoder, Gauge, Registry, TextEncoder};
use std::sync::{ Arc, MutexGuard };
use crate::AppState;
use tracing::{ debug, error, info, warn, trace };
use tracing::{ debug, error, info, warn };
use crate::metrics::{MetricsProcesser, MetricsValueType};
use prometheus::opts;
#[derive(Clone)]
struct CloneableCollector(Arc<dyn Collector>);
impl CloneableCollector {
fn from_boxed(collector: Box<dyn Collector>) -> Self {
CloneableCollector(Arc::from(collector))
}
fn get_collector(&self) -> Box<dyn Collector> {
Box::new(self.clone())
}
fn to_gauge_vec(&self) -> GaugeVec {
let mut fam = self.0.collect()[0].clone();
let metric_name = fam.take_name();
let metric_help = fam.take_help();
let lables = fam.get_metric().iter()
.enumerate()
.filter(|(id, _)| id == &0)
.map(|(_, metric)| metric.get_label())
.flat_map(|a| a.iter())
.map(|a| a.get_name())
.collect::<Vec<&str>>();
let gv = GaugeVec::new(opts!(&metric_name, &metric_help), &lables).unwrap();
for metric in fam.get_metric() {
// gather values
let vals = metric.get_label()
.iter()
.map(|pair| pair.get_value())
.collect::<Vec<&str>>();
gv.with_label_values(&vals).set(metric.get_gauge().get_value());
}
gv
}
}
impl Display for CloneableCollector {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
self.0.collect();
write!(f, "{:#?}", self.0.collect())
}
}
impl Collector for CloneableCollector {
fn desc(&self) -> Vec<&prometheus::core::Desc> {
self.0.desc()
}
fn collect(&self) -> Vec<prometheus::proto::MetricFamily> {
self.0.collect()
}
}
/// An `Update` endpoint
///
@ -73,66 +18,57 @@ impl Collector for CloneableCollector {
/// # Usage
///
/// ``` bash
/// curl -X POST -d '...' 'http://127.0.0.1:9100/update' -d ...
/// curl -X POST -d '"id" : ...' 'http::/localhost:9100/update'
/// ```
///
pub async fn update_metrics(
State(state): State<Arc<AppState>>,
Json(request) : Json<PrometheusMetrics<'_>>
) -> impl IntoResponse {
trace!("post on /update");
info!("post on /update");
let service = &request.service_name;
let mut metrics = Vec::new();
let endpoint = &request.endpoint_name;
for i in request.metrics {
debug!("Processing metric: {:?}", &i);
let metric_name = format!("{}_{}", service, &i.id);
let metric_name = format!("{}_{}_{}", service, endpoint, &i.id);
match MetricsProcesser::get_type_of_value(&i) {
MetricsValueType::Array |
MetricsValueType::TaggedArray => {
trace!("processing an array of metrics");
// ...
},
MetricsValueType::Number => {
trace!("processing a number type of metric");
let gauge = MetricsProcesser::gauge_from_number(
&i,
&metric_name,
&i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id)),
&state.prefix
&i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id))
);
metrics.push(gauge);
},
MetricsValueType::ArrayOfStrings => {
trace!("processing an array of strings");
warn!("String arrays are unsupported, ignoring ...");
},
_ => {
trace!("processing unrecognized type of metric");
warn!("Unrecognized metric type was supplied, ignoring ...");
}
}
}
for gauge in metrics {
if let Some(gauge) = gauge {
if let Some(gauge) = gauge {
match state.registry.lock() {
Err(er) => {
error!("Cannot lock Metric Registry due to {} ", er)
},
Ok(registry) => {
// todo: error handler
if let Err(er) = update_or_insert_metric(
let _ = update_or_insert_metric(
gauge,
registry,
) {
error!("Update or insert metric crushed: {}", er);
return (http::StatusCode::INTERNAL_SERVER_ERROR, er.to_string())
}
&metric_name
);
},
}
}
},
MetricsValueType::ArrayOfStrings => {
warn!("String arrays are unsupported, ignoring ...");
},
_ => {
warn!("Unrecognized metric type was supplied, ignoring ...");
}
}
}
(http::StatusCode::ACCEPTED, String::from("Ok"))
(http::StatusCode::ACCEPTED, "Ok")
}
/// An `Metrics` endpoint
@ -143,21 +79,17 @@ pub async fn update_metrics(
/// # Usage
///
/// ``` bash
/// curl -X GET 'http://127.0.0.1:9100/metrics'
/// curl -X GET 'http::/localhost:9100/metrics'
/// ```
///
pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
trace!("get on /metrics");
let registry = state.registry.lock();
debug!("registry mutex lock is {}", registry.is_ok());
return match registry {
Ok(registry) => {
let encoder = TextEncoder::new();
let mut buffer = Vec::new();
let metric_families = registry.gather();
debug!("vec of metric families - {:?}", &metric_families);
encoder.encode(&metric_families, &mut buffer).unwrap();
String::from_utf8(buffer).unwrap()
},
@ -173,126 +105,75 @@ pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
/// `Registry`
///
pub fn update_or_insert_metric<'a>(
metric: Box<dyn Collector>,
metric: Gauge,
registry: MutexGuard<'a, Registry>,
metric_name: &str
) -> anyhow::Result<()> {
trace!("fn update_or_insert_metric is running");
let prod = CloneableCollector::from_boxed(metric);
use prometheus::Error;
// let mut counter = 0;
match registry.register(Box::new(metric.clone())) {
Ok(_) => {
info!("Metric `{}` was registered!", metric_name);
},
Err(er) => {
// update or throw away
match er {
Error::AlreadyReg => {
let family = registry.gather();
let new_metric = prod.get_collector().collect();
let new_metric_family = &new_metric[0];
let new_metric_name = new_metric_family.get_name();
if let Some(mut fam) = family.into_iter()
.find(|fam| fam.get_name() == new_metric_name) {
trace!("found exising metric. processing update ...");
debug!("found metric : {:?}", &fam);
let metric_name = fam.take_name();
let metric_help = fam.take_help();
let lables = fam.get_metric().iter()
.enumerate()
.filter(|(id, _)| id == &0)
.map(|(_, metric)| metric.get_label())
.flat_map(|a| a.iter())
.map(|a| a.get_name())
.collect::<Vec<&str>>();
let gv = GaugeVec::new(opts!(&metric_name, &metric_help), &lables)?;
for metric in fam.get_metric() {
let vals = metric.get_label()
.iter()
.map(|pair| pair.get_value())
.collect::<Vec<&str>>();
gv.with_label_values(&vals).set(metric.get_gauge().get_value());
}
trace!("recreated metric as gauge vec {:?}", gv.collect());
let new_vec = prod.to_gauge_vec().compare_with_old(&gv);
registry.unregister(Box::new(gv))?;
info!("Unregistered old metric!");
registry.register(Box::new(new_vec))?;
info!("Metric `{}` was re-registered!", new_metric_name);
} else {
match registry.register(prod.get_collector()) {
Ok(_) => info!("Metric `{}` was registered!", new_metric_name),
Err(er) => {
error!("Cannot register new metric `{}` due to {}", new_metric_name, er);
return Err(anyhow::Error::msg(
format!("Cannot register new metric `{}` due to {}", new_metric_name, er)
))
},
}
}
Ok(())
}
trait CompareGaugeVec {
fn compare_with_old(self, old: &GaugeVec) -> Self;
}
impl CompareGaugeVec for GaugeVec {
fn compare_with_old(mut self, old: &GaugeVec) -> Self {
// old.
let old_fam = old.collect();
let new = self.collect();
'outer: for old_fam in old_fam {
for metric in old_fam.get_metric() {
// labels for current old version of metric in vec
let lables = get_hashmap_lables_from_metric(metric);
let value = metric.get_gauge().get_value();
for new in &new {
for new_metric in new.get_metric() {
let new_lables = get_hashmap_lables_from_metric(new_metric);
if lables.len() != new_lables.len() {
error!("Trying to save invalid metric type. Reseting changes ...");
self = old.clone();
break 'outer;
} else {
match (lables.get("status"), new_lables.get("status")) {
(Some(&status), Some(_)) => {
match (
(lables.get("device"), lables.get("source_id")),
(new_lables.get("device"), new_lables.get("source_id")),
) {
((Some(&device), Some(&source_id)),
(Some(&new_device), Some(&new_source_id))) => {
/* */
if device != new_device || source_id != new_source_id {
self.with_label_values(&[device, source_id, status]).set(value);
}
},
((Some(&device), None),
(Some(&new_device), None)) => {
/* */
if device != new_device {
self.with_label_values(&[device, status]).set(value);
}
},
_ => { /* DEAD END */},
}
},
_ => { /* DEAD END */ },
match registry.unregister(Box::new(metric.clone())) {
Ok(_) => {
if let Err(er) = registry.register(Box::new(metric)) {
warn!("Cannot update metric `{}`", metric_name);
return Err(anyhow::Error::msg(
format!("Cannot update metric `{}` due to {}", metric_name, er)
))
} else {
info!("OK on metric `{}` update", metric_name);
}
}
},
Err(er) => {
error!("Cannot unregister metric `{}` due to {}", metric_name, er);
return Err(anyhow::Error::msg(
format!("Cannot unregister metric `{}` due to {}", metric_name, er)
))
},
}
// use prometheus::opts;
// use prometheus::GaugeVec;
// let vec = GaugeVec::new(opts!("test", "test_help"), &["label"]).unwrap();
// // vec.with_label_values(&["default"]).set(42.0);
// if registry.unregister(Box::new(vec)).is_err() {
// debug!("unregister failed");
// };
// let vec = GaugeVec::new(opts!("test1", "test_help1"), &["label"]).unwrap();
// vec.with_label_values(&["goood!"]).set(412.0);
// let _ = registry.register(Box::new(vec));
// registry
// .gather()
// .iter_mut()
// .filter(|target| target.get_name() == metric_name.trim())
// .for_each(|family| {
// // let prev: &mut GaugeVec = family.mut_metric()[0].mut_gauge();
// // GaugeVec::
// // info!("Metric `{}` was updated, new value - {}", metric_name, new);
// });
},
_ => {
error!("Cannot register new metric `{}` due to {}", metric_name, er);
return Err(anyhow::Error::msg(
format!("Cannot register new metric `{}` due to {}", metric_name, er)
))
}
}
}
self
},
}
}
Ok(())
// registry.gather()
// .iter()
// .filter(|fam| fam.get_name().)
fn get_hashmap_lables_from_metric(metric: &Metric) -> HashMap<&str, &str> {
metric.get_label()
.iter()
.map(|pair| (pair.get_name(), pair.get_value()))
.collect::<HashMap<&str, &str>>()
}

37
src/main.rs
View File

@ -11,7 +11,7 @@ use axum::{
routing::{get, post},
Router};
use prometheus::Registry;
use std::{collections::HashMap, str::FromStr, sync::{Arc, Mutex}};
use std::sync::{Arc, Mutex};
use endpoints::*;
use tokio::net::TcpListener;
use tracing::info;
@ -22,39 +22,24 @@ use dotenv::dotenv;
/// Used to store and share state of the metrics `Registry`
///
struct AppState {
registry : Mutex<Registry>,
prefix : String,
registry: Mutex<Registry>,
}
#[tokio::main]
async fn main() -> anyhow::Result<()> {
dotenv().ok();
let log_level = std::env::var("PROMETHEUS_EXPORTER_LOG_LEVEL")
.unwrap_or_else(|_| "INFO".to_owned());
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_line_number(false)
.with_target(false)
.with_file(false)
.compact()
.init();
.with_max_level(tracing::Level::DEBUG)
.init();
info!("Logger was created and configurated, dotenv vars were loaded (if exist)");
let prefix = std::env::var("PROMETHEUS_EXPORTER_PREFIX")
.unwrap_or_else(|_| "zvks".to_owned());
info!("Loading env vars from .env if exists ...");
dotenv().ok();
info!("Initializing local Prometehus metrics registry ...");
let registry = Registry::new();
info!("Initializing shared state for Prometheus Exporter web-server ...");
let state = Arc::new(AppState {
registry : Mutex::new(registry),
prefix : prefix,
registry: Mutex::new(registry),
});
@ -64,12 +49,10 @@ async fn main() -> anyhow::Result<()> {
.route("/update", post(update_metrics))
.with_state(state.clone());
let port = std::env::var("PROMETHEUS_EXPORTER_PORT")
.unwrap_or_else(|_| "9100".to_owned());
let bind_address = format!("0.0.0.0:{}", &port);
let listener = TcpListener::bind(bind_address).await?;
let bind_address = format!("0.0.0.0:{}", std::env::var("PROMETHEUS_EXPORTER_PORT").unwrap_or_else(|_| "9100".to_owned()));
let listener = TcpListener::bind(bind_address).await.unwrap();
info!("Serving on ...:{}", &port);
info!("Serving on ...:9100");
axum::serve(listener, app).await?;
Ok(())
}

151
src/metrics.rs
View File

@ -2,8 +2,8 @@ use crate::structs::v3::MetricOutput;
use serde_json::{Map, Value};
use prometheus::Gauge;
use tracing::error;
use prometheus::{opts, GaugeVec, core::Collector};
use tracing::{debug, trace};
use prometheus::Opts;
use prometheus::GaugeVec;
#[derive(Debug)]
pub enum MetricsValueType {
@ -18,25 +18,20 @@ pub struct MetricsProcesser;
impl MetricsProcesser {
pub fn get_type_of_value(metrics: &MetricOutput) -> MetricsValueType {
trace!("defining metric type");
if Self::is_number(metrics) {
debug!("processing Number");
return MetricsValueType::Number;
}
else if Self::is_array(metrics) {
if Self::is_tagged_array(metrics) {
debug!("processing TaggedArray");
return MetricsValueType::TaggedArray;
}
if Self::is_array_of_string_values(metrics) {
debug!("processing ArrayOfStrings");
return MetricsValueType::ArrayOfStrings;
}
debug!("processing Array");
return MetricsValueType::Array;
}
debug!("processing undefined type");
MetricsValueType::None
}
@ -44,39 +39,19 @@ impl MetricsProcesser {
pub fn gauge_from_number(
metric: &MetricOutput,
metric_name: &str,
metric_desc: &str,
prefix: &str
) -> Option<Box<dyn Collector>> {
trace!("fn gauge_from_number is running");
if let Some(status) = metric.status {
if let Some(device) = metric.device {
if let Some(source_id) = &metric.source_id {
let vec = GaugeVec::new(opts!(format!("{}_{}", prefix, metric.name), metric_desc), &["status", "device", "source_id"]).unwrap();
vec.with_label_values(&[&status.to_string(), &device.to_string(), &source_id.to_string()]).set(metric.value.as_f64().unwrap_or_else(|| 0.0));
debug!("processed metric: {:?}", &vec);
return Some(Box::new(vec));
}
else {
let vec = GaugeVec::new(opts!(format!("{}_{}", prefix, metric.name), metric_desc), &["status", "device"]).unwrap();
vec.with_label_values(&[&status.to_string(), &device.to_string()]).set(metric.value.as_f64().unwrap_or_else(|| 0.0));
debug!("processed metric: {:?}", &vec);
return Some(Box::new(vec));
}
} else {
let vec = GaugeVec::new(opts!(format!("{}_{}", prefix, metric.name), metric_desc), &["status"]).unwrap();
vec.with_label_values(&[&status.to_string()]).set(metric.value.as_f64().unwrap_or_else(|| 0.0));
debug!("processed metric: {:?}", &vec);
return Some(Box::new(vec));
}
}
metric_desc: &str
) -> Option<Gauge> {
let gauge = Gauge::new(
format!("{}_{}", prefix, metric.name),
metric_name,
metric_desc
);
match gauge {
Ok(gauge) => {
// let value = metric.value.as_number().unwrap_or({
// error!("Cannot convert {} metric value to f64 type. Value was set to 0.0", &metric_name);
// });
// let value = value.as_f64()
let val = match metric.value.as_number() {
Some(val) => {
val.as_f64().unwrap_or_else(||
@ -89,13 +64,97 @@ impl MetricsProcesser {
},
};
gauge.set(val);
debug!("processed metric: {:?}", &gauge);
return Some(Box::new(gauge));
return Some(gauge);
},
Err(er) => error!("Cannot create Gauge metric {} due to {}", &metric_name, er),
}
None
}
pub fn gauge_from_map_metrics(
map: &Map<String, Value>,
service: &str,
endpoint: &str
) -> Option<Gauge> {
let map = map.clone();
let help: String = map.keys()
.enumerate()
.map(|(idx, key)| {
if idx == 1 {
return key.to_owned();
}
"".to_owned()
})
.collect();
let name = format!("{}_{}_{}", service, endpoint, &help);
if map.len() > 1 {
// tagged
if map.len() > 2 {
error!("Cannot create Gauge {}. It can be only 1 tag", &name);
} else {
let mut label_name = String::new();
let mut label_value = String::new();
let mut metric_value = 0.0;
map.iter()
.enumerate()
.for_each(|(idx, (key, value))|{
if idx == 0 {
label_name = key.to_owned();
label_value = value.as_str()
.unwrap_or("")
.to_owned();
} else {
metric_value = value.as_f64().unwrap_or(0.0)
}
});
let opts = Opts::new(&name, &help);
let gauge_vec = GaugeVec::new(opts, &[&label_name]);
match gauge_vec {
Ok(vec) => {
// vec.get_metric_with_label_values(vals)
match vec.get_metric_with_label_values(&[&label_value]) {
Ok(metric) => {
metric.set(metric_value); // Устанавливаем значение метрики
return Some(metric.clone()); // Возвращаем `Gauge`
},
Err(er) => {
error!("Cannot create Gauge {} due to {}", &name, er);
},
}
},
Err(er) => error!("Cannot create Gauge {} due to {}", &name, er),
}
}
} else {
// not-tagged
let metric = Gauge::new(&name, &help);
match metric {
Ok(gauge) => {
let mut value = 0.0;
map.values()
.map(|val| val.clone().as_f64())
.for_each(|val| {
value = val.unwrap_or(0.0)
});
gauge.set(value);
return Some(gauge);
},
Err(er) => {
error!("Cannot create Gauge {} due to {}", &name, er);
}
}
}
None
}
pub fn get_value_as_vec_map(metrics: &MetricOutput) -> Vec<Map<String, Value>>{
let mut vec: Vec<Map<String, Value>> = Vec::new();
let arr = metrics.value.as_array().unwrap();
arr.iter()
.for_each(|a| {
vec.push(serde_json::from_value(a.clone()).unwrap());
});
vec
}
pub fn is_array_of_string_values(metrics: &MetricOutput) -> bool {
let arr = metrics.value.clone();
let arr = arr.as_array().unwrap();
@ -105,6 +164,9 @@ impl MetricsProcesser {
map.values()
.all(|val| val.is_string())
}
// fn is_valid(metrics: &PrometheusMetrics) -> bool {
// false
// }
fn is_array(metrics: &MetricOutput) -> bool {
metrics.value.is_array()
}
@ -118,3 +180,18 @@ impl MetricsProcesser {
.value.is_number()
}
}
trait IsTaggedArray {
fn is_tagged_array(&self) -> bool;
}
impl IsTaggedArray for Value {
fn is_tagged_array(&self) -> bool {
if let Some(arr) = self.as_array() {
return arr[0].get("tag_name").is_some();
}
false
}
}

7
src/structs.rs
View File

@ -11,23 +11,18 @@ pub mod v3 {
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct MetricOutput<'a> {
pub id : String,
pub name : String,
#[serde(rename = "type")]
json_type : String,
addr : String,
pub value : Value,
#[serde(rename = "description")]
pub desc : Option<Cow<'a, String>>,
pub status: Option<isize>,
pub device: Option<isize>,
#[serde(rename = "source")]
pub source_id: Option<String>,
}
#[derive(Serialize, Deserialize, Debug)]
pub struct PrometheusMetrics<'a> {
pub service_name: String,
// pub endpoint_name: String,
pub endpoint_name: String,
pub metrics: Vec<MetricOutput<'a>>,
}
}