src/endpoints.rs

@@ -4,12 +4,10 @@ use axum::{
     http
 };
 use crate::structs::v3::PrometheusMetrics;
-// use integr_structs::api::v3::PrometheusMetrics;
+use prometheus::{ Encoder, Gauge, Registry, TextEncoder};
-use prometheus::{Encoder, Registry, TextEncoder, Gauge};
+use std::sync::{ Arc, MutexGuard };
-use std::sync::{Arc, MutexGuard};
 use crate::AppState;
-// use log::{warn, info, error};
+use tracing::{ error, debug, info, warn };
-use tracing::{info, error, warn};
 use crate::metrics::{MetricsProcesser, MetricsValueType};
 
 pub async fn update_metrics(
@@ -18,7 +16,7 @@ pub async fn update_metrics(
 ) -> impl IntoResponse {
     info!("post on /update");
 //     let resp = Response::new("body");
-    // println!("{:?}", request);
+    // debug!("{:?}", request);
     // debug!("{:?}", MetricsProcesser::get_type_of_value(&request));
     let service = &request.service_name;
     let endpoint = &request.endpoint_name;
@@ -43,7 +41,8 @@ pub async fn update_metrics(
                             error!("Cannot lock Metric Registry due to {} ", er)
                         },
                         Ok(registry) => {
-                            update_or_insert_metric(
+                            //  todo: error handler
+                            let _ = update_or_insert_metric(
                                 gauge, 
                                 registry,
                                 &metric_name
@@ -86,16 +85,70 @@ pub fn update_or_insert_metric<'a>(
     metric: Gauge, 
     registry: MutexGuard<'a, Registry>,
     metric_name: &str
-) {
+) -> anyhow::Result<()> {
+    use prometheus::Error;
     // let mut counter = 0;
-    match registry.register(Box::new(metric)) {
+    match registry.register(Box::new(metric.clone())) {
         Ok(_) => {
-            info!("Metric `{}` was added!", metric_name);
+            info!("Metric `{}` was registered!", metric_name);
         },
-        Err(_er) => {
+        Err(er) => {
-            // update
+            // update or throw away
+            match er {
+                Error::AlreadyReg => {
+
+                    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)
+                    ))
+                }
+            }
         },
     }
+    Ok(())
     // registry.gather()
     //     .iter()
     //     .filter(|fam| fam.get_name().)

src/main.rs

@@ -21,8 +21,15 @@ struct AppState {
 }
 
 #[tokio::main]
-async fn main() {
+async fn main() -> anyhow::Result<()> {
     // let _ = setup_logger().await;
+
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::DEBUG)
+        .init();
+
+    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");
@@ -52,6 +59,8 @@ async fn main() {
     // 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),
@@ -60,20 +69,15 @@ async fn main() {
 
     // info!("Configurating Web-Server...");
 
-    tracing_subscriber::fmt()
+    info!("Configurating internals of Prometheus Exporter web-server...");
-        .with_max_level(tracing::Level::DEBUG)
-        .init();
-
-    info!("Configurating Web-Server...");
 
     let app = Router::new()
         .route("/metrics", get(metrics_handler))
-        // .route("/increment", get(increment_handler))
-        // .route("/sum", get(summary_handler))
         .route("/update", post(update_metrics))
         .with_state(state.clone());
 
     let listener = TcpListener::bind("0.0.0.0:9100").await.unwrap();
     info!("Serving on ...:9100");
-    axum::serve(listener, app).await.unwrap();
+    axum::serve(listener, app).await?;
+    Ok(())
 }

src/metrics.rs

@@ -2,6 +2,8 @@ use crate::structs::v3::MetricOutput;
 use serde_json::{Map, Value};
 use prometheus::Gauge;
 use tracing::error;
+use prometheus::Opts;
+use prometheus::GaugeVec;
 
 #[derive(Debug)]
 pub enum MetricsValueType { 
@@ -32,6 +34,8 @@ impl MetricsProcesser {
         }
         MetricsValueType::None
     }
+
+    // TODO: i64 and f63 support
     pub fn gauge_from_number(
         metric: &MetricOutput,
         metric_name: &str,
@@ -102,9 +106,6 @@ impl MetricsProcesser {
                             metric_value = value.as_f64().unwrap_or(0.0)
                         }
                     });
-                use prometheus::Opts;
-                use prometheus::GaugeVec;
-
                 let opts = Opts::new(&name, &help);
                 let gauge_vec = GaugeVec::new(opts, &[&label_name]);
                 match gauge_vec {
@@ -178,3 +179,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
+    }
+}