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:0.2.8
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 "0.2.8" have entirely different histories.
master ... 0.2.8

5 changed files with 92 additions and 34 deletions
Show Stats Expand all files Collapse all files

5
.env.example
View File

@ -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

88
src/endpoints.rs
View File

@ -82,6 +82,13 @@ 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 {
@ -98,10 +105,27 @@ 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)),
&state.prefix
&i.desc.clone().unwrap_or_else(|| std::borrow::Cow::Borrowed(&i.id))
);
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");
@ -156,6 +180,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();
// dbg!(&metric_families);
debug!("vec of metric families - {:?}", &metric_families);
encoder.encode(&metric_families, &mut buffer).unwrap();
@ -175,6 +200,7 @@ 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);
@ -198,7 +224,7 @@ pub fn update_or_insert_metric<'a>(
.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() {
@ -227,6 +253,44 @@ 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(())
}
@ -245,11 +309,13 @@ 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();
@ -258,14 +324,17 @@ impl CompareGaugeVec for GaugeVec {
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")),
(lables.get("device"), lables.get("module")),
(new_lables.get("device"), new_lables.get("module")),
) {
((Some(&device), Some(&source_id)),
(Some(&new_device), Some(&new_source_id))) => {
((Some(&device), Some(&module)),
(Some(&new_device), Some(&new_module))) => {
/* */
if device != new_device || source_id != new_source_id {
self.with_label_values(&[device, source_id, status]).set(value);
dbg!(1);
if device != new_device || module != new_module {
dbg!(2);
self.with_label_values(&[device, module, status]).set(value);
// continue 'outer;
}
},
((Some(&device), None),
@ -273,6 +342,7 @@ impl CompareGaugeVec for GaugeVec {
/* */
if device != new_device {
self.with_label_values(&[device, status]).set(value);
// continue 'outer;
}
},
_ => { /* DEAD END */},

12
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::{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>,
prefix : String,
registry: Mutex<Registry>,
}
#[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),
prefix : prefix,
registry: Mutex::new(registry),
});

15
src/metrics.rs
View File

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

6
src/structs.rs
View File

@ -11,7 +11,6 @@ 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,
@ -20,14 +19,13 @@ pub mod v3 {
pub desc : Option<Cow<'a, String>>,
pub status: Option<isize>,
pub device: Option<isize>,
#[serde(rename = "source")]
pub source_id: Option<String>,
pub module: Option<isize>,
}
#[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>>,
}
}