.env.example

@@ -2,11 +2,6 @@
 # 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

Jenkinsfile

@@ -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 {
@@ -35,6 +14,7 @@ pipeline {
             steps {
                 script {
                     def hasTags = sh(script: "git tag -l | wc -l", returnStdout: true).trim().toInteger() > 0
+                    echo "${hasTags}"
                         
                     def lastVersion = "0.0.0"
                         
@@ -113,7 +93,6 @@ pipeline {
                             -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}"

src/endpoints.rs

@@ -4,12 +4,11 @@ use axum::{
     http
 };
 use crate::structs::v3::PrometheusMetrics;
-use prometheus::{ core::Collector, Encoder, GaugeVec, Registry, TextEncoder, proto::Metric};
+use prometheus::{ core::Collector, Encoder, Registry, TextEncoder};
-use std::{collections::HashMap, fmt::Display, sync::{ Arc, MutexGuard }};
+use std::sync::{ Arc, MutexGuard };
 use crate::AppState;
 use tracing::{ debug, error, info, warn, trace };
 use crate::metrics::{MetricsProcesser, MetricsValueType};
-use prometheus::opts;
 
 #[derive(Clone)]
 struct CloneableCollector(Arc<dyn Collector>);
@@ -20,41 +19,7 @@ impl CloneableCollector {
     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()
@@ -82,11 +47,11 @@ pub async fn update_metrics(
 ) -> impl IntoResponse {
     trace!("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 => {
@@ -98,10 +63,23 @@ pub async fn update_metrics(
                 let gauge = MetricsProcesser::gauge_from_number(
                     &i, 
                     &metric_name,
-                    &i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id)),
+                    &i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id))
-                    &state.prefix
                 );
-                metrics.push(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
+                            let _ = update_or_insert_metric(
+                                gauge, 
+                                registry,
+                                &metric_name
+                            );
+                        },
+                    }
+                }
             },
             MetricsValueType::ArrayOfStrings => {
                 trace!("processing an array of strings");
@@ -113,26 +91,8 @@ pub async fn update_metrics(
             }
         }
     }
-    for gauge in metrics {
+
-        if let Some(gauge) = gauge {
+    (http::StatusCode::ACCEPTED, "Ok")
-                    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(
-                                gauge, 
-                                registry,
-                            ) {
-                                error!("Update or insert metric crushed: {}", er);
-                                return (http::StatusCode::INTERNAL_SERVER_ERROR, er.to_string())
-                            }
-                        },
-                    }
-                }
-    }
-    (http::StatusCode::ACCEPTED, String::from("Ok"))
 }
 
 /// An `Metrics` endpoint
@@ -156,6 +116,7 @@ pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
             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();
@@ -175,124 +136,48 @@ pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
 pub fn update_or_insert_metric<'a>(
     metric: Box<dyn Collector>, 
     registry: MutexGuard<'a, Registry>,
+    metric_name: &str
 ) -> anyhow::Result<()> {
     trace!("fn update_or_insert_metric is running");
+    use prometheus::Error;
     let prod = CloneableCollector::from_boxed(metric);
 
-    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),
+        Ok(_) => {
+            info!("Metric `{}` was registered!", metric_name);
+        },
         Err(er) => {
-                error!("Cannot register new metric `{}` due to {}", new_metric_name, er);
+            // update or throw away
+            match er {
+                Error::AlreadyReg => {
+                    trace!("processing already regged metric");
+                    match registry.unregister(prod.get_collector()) {
+                        Ok(_) => {
+                            if let Err(er) = registry.register(prod.get_collector()) {
+                                warn!("Cannot update metric `{}`", metric_name);
                                 return Err(anyhow::Error::msg(
-                    format!("Cannot register new metric `{}` due to {}", new_metric_name, er)
+                                    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)
                             ))
                         },
                     }
+                },
+                _ => {
+                    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(())
 }
-
-
-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

@@ -11,7 +11,7 @@ use axum::{
     routing::{get, post}, 
     Router};
 use prometheus::Registry;
-use std::{collections::HashMap, str::FromStr, sync::{Arc, Mutex}};
+use std::{str::FromStr, sync::{Arc, Mutex}};
 use endpoints::*;
 use tokio::net::TcpListener;
 use tracing::info;
@@ -22,8 +22,7 @@ use dotenv::dotenv;
 /// Used to store and share state of the metrics `Registry`
 /// 
 struct AppState {
-    registry : Mutex<Registry>,
+    registry: Mutex<Registry>,
-    prefix : String,
 }
 
 #[tokio::main]
@@ -42,19 +41,14 @@ async fn main() -> anyhow::Result<()> {
     .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();
 
     info!("Initializing shared state for Prometheus Exporter web-server ...");
     let state = Arc::new(AppState {
-        registry : Mutex::new(registry),
+        registry: Mutex::new(registry),
-        prefix : prefix,
     });
 
 

src/metrics.rs

@@ -1,7 +1,10 @@
+use std::sync::Arc;
+
 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 tracing::{debug, trace};
 
@@ -44,34 +47,18 @@ impl MetricsProcesser {
     pub fn gauge_from_number(
         metric: &MetricOutput,
         metric_name: &str,
-        metric_desc: &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 {
+            let vec = GaugeVec::new(opts!(metric_name, metric_desc), &["status"]).unwrap();
-                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(
-            format!("{}_{}", prefix, metric.name),
+            metric_name,
             metric_desc
         );
 
@@ -96,6 +83,89 @@ impl MetricsProcesser {
         }
         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) => {
+                        match vec.get_metric_with_label_values(&[&label_value]) {
+                            Ok(metric) => {
+                                metric.set(metric_value);
+                                return Some(metric.clone()); 
+                            },
+                            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 +175,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 +191,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
+    }
+}

src/structs.rs

@@ -11,23 +11,19 @@ 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 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>>,
     }
 }