Merge pull request 'rc' (#22) from rc into master

Reviewed-on: http://git.enode/deployer3000/prometheus-exporter/pulls/21

.env.example

@@ -0,0 +1,13 @@
+# 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"

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "exporter"
-version = "0.1.0"
+version = "0.2.3"
 edition = "2021"
 
 [dependencies]
@@ -13,4 +13,5 @@ tracing = "0.1.41"
 tracing-subscriber = "0.3.19"
 serde = { version = "1.0.217", features = ["derive"] }
 serde_json = "1.0.138"
+dotenv = "0.15.0"
 

Jenkinsfile

@@ -1,3 +1,24 @@
+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 {
@@ -13,7 +34,22 @@ pipeline {
             }
             steps {
                 script {
-                    env.IMAGE_TAG = sh(script: "git describe --tags --abbrev=0", returnStdout: true).trim()
+                    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
                 }
             }
         }
@@ -58,14 +94,30 @@ 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/prometheus-exporter/pulls/${prId}/merge
+                             http://git.entcor/api/v1/repos/deployer3000/${env.IMAGE_NAME}/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")
                     }
                 }
             }

README.md

@@ -1,3 +1,115 @@
 # Prometheus exporter 
 
-to handle updates of ZVKS metrics in local Registry and to share with Prometheus later
+Программный компонент проекта "Буревестник ВКС", предоставляющий функционал
+"кастомного" `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
+```

src/endpoints.rs

@@ -4,37 +4,116 @@ use axum::{
     http
 };
 use crate::structs::v3::PrometheusMetrics;
-use prometheus::{ Encoder, Gauge, Registry, TextEncoder};
+use prometheus::{ core::Collector, Encoder, GaugeVec, Registry, TextEncoder, proto::Metric};
-use std::sync::{ Arc, MutexGuard };
+use std::{collections::HashMap, fmt::Display, sync::{ Arc, MutexGuard }};
 use crate::AppState;
-use tracing::{ error, debug, info, warn };
+use tracing::{ debug, error, info, warn, trace };
 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
+/// 
+/// Used to registrate new metrics and to update already 
+/// existing in local metrics `Registry`
+/// 
+/// # Usage
+/// 
+/// ``` bash
+/// curl -X POST -d '...' 'http://127.0.0.1:9100/update' -d ...
+/// ```
+/// 
 pub async fn update_metrics(
     State(state): State<Arc<AppState>>,
-    Json(request) : Json<PrometheusMetrics>
+    Json(request) : Json<PrometheusMetrics<'_>>
 ) -> impl IntoResponse {
-    info!("post on /update");
+    trace!("post on /update");
-//     let resp = Response::new("body");
-    // debug!("{:?}", request);
-    // debug!("{:?}", MetricsProcesser::get_type_of_value(&request));
     let service = &request.service_name;
-    let endpoint = &request.endpoint_name;
+    let mut metrics = Vec::new();
 
     for i in request.metrics {
-        // debug!("{:?}", &i);
+        debug!("Processing metric: {:?}", &i);
-        // debug!("{:?}", MetricsProcesser::get_type_of_value(&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
+                    &metric_name,
+                    &i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id)),
+                    &state.prefix
                 );
+                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 {
                     match state.registry.lock() {
                         Err(er) => {
@@ -42,36 +121,43 @@ pub async fn update_metrics(
                         },
                         Ok(registry) => {
                             //  todo: error handler
-                            let _ = update_or_insert_metric(
+                            if let Err(er) = update_or_insert_metric(
                                 gauge, 
                                 registry,
-                                &metric_name
+                            ) {
-                            );
+                                error!("Update or insert metric crushed: {}", er);
+                                return (http::StatusCode::INTERNAL_SERVER_ERROR, er.to_string())
+                            }
                         },
                     }
                 }
-                // dbg!(gauge);
-            },
-            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
-}
+/// 
-
+/// Needed in sharing current local metrics `Registry` state
+/// and all stored metrics  
+/// 
+/// # Usage
+/// 
+/// ``` bash
+/// curl -X GET 'http://127.0.0.1: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()
         },
@@ -81,76 +167,132 @@ pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
     }
 }
 
+/// A function-registrator 
+/// 
+/// Registrates or updates metrics state in local 
+/// `Registry`
+/// 
 pub fn update_or_insert_metric<'a>(
-    metric: Gauge, 
+    metric: Box<dyn Collector>, 
     registry: MutexGuard<'a, Registry>,
-    metric_name: &str
 ) -> anyhow::Result<()> {
-    use prometheus::Error;
+    trace!("fn update_or_insert_metric is running");
-    // let mut counter = 0;
+    let prod = CloneableCollector::from_boxed(metric);
-    match registry.register(Box::new(metric.clone())) {
-        Ok(_) => {
-            info!("Metric `{}` was registered!", metric_name);
-        },
-        Err(er) => {
-            // update or throw away
-            match er {
-                Error::AlreadyReg => {
 
-                    match registry.unregister(Box::new(metric.clone())) {
+    let family = registry.gather();
-                        Ok(_) => {
+
-                            if let Err(er) = registry.register(Box::new(metric)) {
+    let new_metric = prod.get_collector().collect();
-                                warn!("Cannot update metric `{}`", metric_name);
+    let new_metric_family = &new_metric[0];
-                                return Err(anyhow::Error::msg(
+    let new_metric_name = new_metric_family.get_name();
-                                    format!("Cannot update metric `{}` due to {}", metric_name, er)
+
-                                ))
+    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 {
-                                info!("OK on metric `{}` update", metric_name);
+        match registry.register(prod.get_collector()) {
-                            }
+            Ok(_) => info!("Metric `{}` was registered!", new_metric_name),
-                        },
             Err(er) => {
-                            error!("Cannot unregister metric `{}` due to {}", metric_name, er);
+                error!("Cannot register new metric `{}` due to {}", new_metric_name, er);
                 return Err(anyhow::Error::msg(
-                                format!("Cannot unregister metric `{}` due to {}", metric_name, er)
+                    format!("Cannot register new metric `{}` due to {}", new_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)
-                    ))
-                }
-            }
-        },
     }
     Ok(())
-    // registry.gather()
+}
-    //     .iter()
+
-    //     .filter(|fam| fam.get_name().)
+
-         
+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 */ },
+                            }
+                        }
+                    }
+                }
+            }
+        }
+
+        self
+    }
+}
+
+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>>()
 }

src/main.rs

@@ -1,83 +1,75 @@
+/// Prometheus-exporter - a Rust binary crate to provide functional 
+/// of the metrics Exporter.
+/// 
+/// It handles metrics list in it's own local `Registry`
+
 mod endpoints;
-// mod logger;
 mod metrics;
 mod structs;
 
-// use logger::setup_logger;
 use axum::{
     routing::{get, post}, 
     Router};
 use prometheus::Registry;
-use std::sync::{Arc, Mutex};
+use std::{collections::HashMap, str::FromStr, sync::{Arc, Mutex}};
 use endpoints::*;
 use tokio::net::TcpListener;
-// use log::{warn, info, error};
 use tracing::info;
+use dotenv::dotenv;
 
+/// Shared state of the Web-server 
+/// 
+/// Used to store and share state of the metrics `Registry`
+/// 
 struct AppState {
     registry : Mutex<Registry>,
-    // counter: Mutex<Counter>,
+    prefix : String,
-    // sum : Mutex<Summary>,
 }
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    // let _ = setup_logger().await;
+    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::DEBUG)
+    .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();
 
+    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!("Initializing local Prometehus metrics registry ...");
-
     let registry = Registry::new();
-    // let counter_opts = Opts::new("example_counter", "Пример счётчика");
-    // let histogram_opts = Opts::new("example_histogram", "Пример histogram");
-
-    // use prometheus::proto::{Summary, Quantile};
-    // use prometheus::proto::
-
-    // let guage = prometheus::ProtobufEncoder::new();
-
-
-
-    // let mut sunops = Summary::new();
-    // let mut q1 = Quantile::new();
-    // let mut q2 = Quantile::new();
-
-    // q1.set_quantile(25.0);
-    // q2.set_quantile(75.0);
-    // // prometheus::proto::Metric::
-
-    // let vq = vec![q1, q2];
-    // sunops.set_quantile(vq.into());
-
-    // let counter = Counter::with_opts(counter_opts).unwrap();
-    // // counter.desc()
-    // registry.register(Box::new(counter.clone())).unwrap();
-    // registry.register(Box::new(prometheus::proto::MetricFamily::));
-    // registry.register(Box::new(sunops.clone())).unwrap();
-    
 
     info!("Initializing shared state for Prometheus Exporter web-server ...");
-
     let state = Arc::new(AppState {
         registry : Mutex::new(registry),
-        // counter: Mutex::new(counter),
+        prefix : prefix,
-        // sum : Mutex::new(sunops)
     });
 
-    // info!("Configurating Web-Server...");
 
     info!("Configurating internals of Prometheus Exporter web-server...");
-
     let app = Router::new()
         .route("/metrics", get(metrics_handler))
         .route("/update", post(update_metrics))
         .with_state(state.clone());
 
-    let listener = TcpListener::bind("0.0.0.0:9100").await.unwrap();
+    let port = std::env::var("PROMETHEUS_EXPORTER_PORT")
-    info!("Serving on ...:9100");
+    .unwrap_or_else(|_| "9100".to_owned());
+    let bind_address = format!("0.0.0.0:{}", &port);
+    let listener = TcpListener::bind(bind_address).await?;
+    
+    info!("Serving on ...:{}", &port);
     axum::serve(listener, app).await?;
     Ok(())
 }

src/metrics.rs

@@ -2,8 +2,8 @@ use crate::structs::v3::MetricOutput;
 use serde_json::{Map, Value};
 use prometheus::Gauge;
 use tracing::error;
-use prometheus::Opts;
+use prometheus::{opts, GaugeVec, core::Collector};
-use prometheus::GaugeVec;
+use tracing::{debug, trace};
 
 #[derive(Debug)]
 pub enum MetricsValueType { 
@@ -18,20 +18,25 @@ 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
     }
 
@@ -39,18 +44,39 @@ impl MetricsProcesser {
     pub fn gauge_from_number(
         metric: &MetricOutput,
         metric_name: &str,
-    ) -> Option<Gauge> {        
+        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));
+            }
+        }
+
         let gauge = Gauge::new(
-            metric_name,
+            format!("{}_{}", prefix, metric.name),
-            &metric.id
+            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(||
@@ -63,97 +89,13 @@ impl MetricsProcesser {
                     },
                 };
                 gauge.set(val);
-                return Some(gauge);
+                debug!("processed metric: {:?}", &gauge);
+                return Some(Box::new(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();
@@ -163,9 +105,6 @@ 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()
     }
@@ -179,18 +118,3 @@ 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
-    }
-}

src/structs.rs

@@ -1,25 +1,33 @@
 // use std::collections::HashMap;
 use serde::{Serialize, Deserialize};
 use serde_json::Value;
+use std::borrow::Cow;
 // use anyhow::Result;
 // use std::sync::Arc;
 
 pub mod v3 {
     pub use super::*;
     // to prometheus and nmns
-    #[derive(Serialize, Deserialize, Debug)]
+    #[derive(Serialize, Deserialize, Debug, Clone)]
-    pub struct MetricOutput {
+    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 {
+    pub struct PrometheusMetrics<'a> {
         pub service_name: String,
-        pub endpoint_name: String,
+        // pub endpoint_name: String,
-        pub metrics: Vec<MetricOutput>,
+        pub metrics: Vec<MetricOutput<'a>>,
     }
 }