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

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

.env.example

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

src/endpoints.rs

@@ -82,13 +82,6 @@ pub async fn update_metrics(
 ) -> impl IntoResponse {
     trace!("post on /update");
     let service = &request.service_name;
-    // let endpoint = &request.endpoint_name;
-    // if request.with_device_status_and_module() {
-
-    // } else {
-        
-    // }
-
     let mut metrics = Vec::new();
 
     for i in request.metrics {
@@ -105,27 +98,10 @@ 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
-                //             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())
-                //             }
-                //         },
-                //     }
-                // }
             },
             MetricsValueType::ArrayOfStrings => {
                 trace!("processing an array of strings");
@@ -180,7 +156,6 @@ 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();
-            // dbg!(&metric_families);
             debug!("vec of metric families - {:?}", &metric_families);
             
             encoder.encode(&metric_families, &mut buffer).unwrap();
@@ -200,7 +175,6 @@ 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");
     let prod = CloneableCollector::from_boxed(metric);
@@ -253,44 +227,6 @@ pub fn update_or_insert_metric<'a>(
             },
         }
     }
-
-    // match registry.register(prod.get_collector()) {
-    //     Ok(_) => {
-    //         info!("Metric `{}` was registered!", metric_name);
-    //     },
-    //     Err(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 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(())
 }
 
@@ -309,13 +245,11 @@ impl CompareGaugeVec for GaugeVec {
             for metric in old_fam.get_metric() {
                 // labels for current old version of metric in vec
                 let lables = get_hashmap_lables_from_metric(metric);
-                dbg!(&lables);
                 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);
-                        dbg!(&new_lables);
                         if lables.len() != new_lables.len() {
                             error!("Trying to save invalid metric type. Reseting changes ...");
                             self = old.clone();
@@ -324,17 +258,14 @@ impl CompareGaugeVec for GaugeVec {
                             match (lables.get("status"), new_lables.get("status")) {
                                 (Some(&status), Some(_)) => {
                                     match (
-                                            (lables.get("device"), lables.get("module")), 
+                                            (lables.get("device"), lables.get("source_id")), 
-                                            (new_lables.get("device"), new_lables.get("module")),
+                                            (new_lables.get("device"), new_lables.get("source_id")),
                                         ) {
-                                            ((Some(&device), Some(&module)),
+                                            ((Some(&device), Some(&source_id)),
-                                            (Some(&new_device), Some(&new_module))) => {
+                                            (Some(&new_device), Some(&new_source_id))) => {
                                                 /* */
-                                                dbg!(1);
+                                                if device != new_device || source_id != new_source_id {
-                                                if device != new_device || module != new_module {
+                                                    self.with_label_values(&[device, source_id, status]).set(value);
-                                                    dbg!(2);
-                                                    self.with_label_values(&[device, module, status]).set(value);
-                                                    // continue 'outer;
                                                 }
                                             },
                                             ((Some(&device), None),
@@ -342,7 +273,6 @@ impl CompareGaugeVec for GaugeVec {
                                                 /* */
                                                 if device != new_device {
                                                     self.with_label_values(&[device, status]).set(value);
-                                                    // continue 'outer;
                                                 }
                                             },
                                             _ => { /* DEAD END */},

src/main.rs

@@ -11,7 +11,7 @@ use axum::{
     routing::{get, post}, 
     Router};
 use prometheus::Registry;
-use std::{str::FromStr, sync::{Arc, Mutex}};
+use std::{collections::HashMap, str::FromStr, sync::{Arc, Mutex}};
 use endpoints::*;
 use tokio::net::TcpListener;
 use tracing::info;
@@ -22,7 +22,8 @@ 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]
@@ -41,14 +42,19 @@ 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

@@ -45,26 +45,25 @@ impl MetricsProcesser {
         metric: &MetricOutput,
         metric_name: &str,
         metric_desc: &str,
-        // status: Option<isize>,
+        prefix: &str
-        // device: Option<isize>,
     ) -> 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(module) = metric.module {
+                if let Some(source_id) = &metric.source_id {
-                    let vec = GaugeVec::new(opts!(metric_name, metric_desc), &["status", "device", "module"]).unwrap();
+                    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(), &module.to_string()]).set(metric.value.as_f64().unwrap_or_else(|| 0.0));
+                    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!(metric_name, metric_desc), &["status", "device"]).unwrap();
+                    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!(metric_name, metric_desc), &["status"]).unwrap();
+                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));
@@ -72,7 +71,7 @@ impl MetricsProcesser {
         }
 
         let gauge = Gauge::new(
-            metric_name,
+            format!("{}_{}", prefix, metric.name),
             metric_desc
         );
 

src/structs.rs

@@ -11,6 +11,7 @@ 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,
@@ -19,13 +20,14 @@ pub mod v3 {
         pub desc : Option<Cow<'a, String>>,
         pub status: Option<isize>,
         pub device: Option<isize>,
-        pub module: 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>>,
     }
 }