US changes
parent
ac218ea845
commit
ab1ed5a57c
|
|
@ -3,14 +3,28 @@ use axum::{
|
||||||
response::IntoResponse,
|
response::IntoResponse,
|
||||||
http
|
http
|
||||||
};
|
};
|
||||||
use tracing_subscriber::field::debug;
|
|
||||||
use crate::structs::v3::PrometheusMetrics;
|
use crate::structs::v3::PrometheusMetrics;
|
||||||
use prometheus::{ Encoder, Gauge, Registry, TextEncoder};
|
use prometheus::{ core::Collector, Encoder, Gauge, Registry, TextEncoder};
|
||||||
use std::sync::{ Arc, MutexGuard };
|
use std::{cell::RefCell, rc::Rc, sync::{ Arc, MutexGuard }};
|
||||||
use crate::AppState;
|
use crate::AppState;
|
||||||
use tracing::{ debug, error, info, warn, trace };
|
use tracing::{ debug, error, info, warn, trace };
|
||||||
use crate::metrics::{MetricsProcesser, MetricsValueType};
|
use crate::metrics::{MetricsProcesser, MetricsValueType};
|
||||||
|
|
||||||
|
struct BoxCollectorProducer {
|
||||||
|
inner : *mut dyn Collector,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl BoxCollectorProducer {
|
||||||
|
pub fn new(target: Box<dyn Collector>) -> Self {
|
||||||
|
Self {
|
||||||
|
inner : Box::into_raw(target)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
pub fn new_box(&self) -> Box<dyn Collector> {
|
||||||
|
unsafe { Box::from_raw(self.inner) }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// An `Update` endpoint
|
/// An `Update` endpoint
|
||||||
///
|
///
|
||||||
/// Used to registrate new metrics and to update already
|
/// Used to registrate new metrics and to update already
|
||||||
|
|
@ -115,14 +129,17 @@ pub async fn metrics_handler(State(state): State<Arc<AppState>>) -> String {
|
||||||
/// `Registry`
|
/// `Registry`
|
||||||
///
|
///
|
||||||
pub fn update_or_insert_metric<'a>(
|
pub fn update_or_insert_metric<'a>(
|
||||||
metric: Gauge,
|
metric: Box<dyn Collector>,
|
||||||
registry: MutexGuard<'a, Registry>,
|
registry: MutexGuard<'a, Registry>,
|
||||||
metric_name: &str
|
metric_name: &str
|
||||||
) -> anyhow::Result<()> {
|
) -> anyhow::Result<()> {
|
||||||
trace!("fn update_or_insert_metric is running");
|
trace!("fn update_or_insert_metric is running");
|
||||||
use prometheus::Error;
|
use prometheus::Error;
|
||||||
// let mut counter = 0;
|
// let mut counter = 0;
|
||||||
match registry.register(Box::new(metric.clone())) {
|
// let ptr = Box::into_raw(metric);
|
||||||
|
let prod = BoxCollectorProducer::new(metric);
|
||||||
|
|
||||||
|
match registry.register(prod.new_box()) {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
info!("Metric `{}` was registered!", metric_name);
|
info!("Metric `{}` was registered!", metric_name);
|
||||||
},
|
},
|
||||||
|
|
@ -131,9 +148,9 @@ pub fn update_or_insert_metric<'a>(
|
||||||
match er {
|
match er {
|
||||||
Error::AlreadyReg => {
|
Error::AlreadyReg => {
|
||||||
trace!("processing already regged metric");
|
trace!("processing already regged metric");
|
||||||
match registry.unregister(Box::new(metric.clone())) {
|
match registry.unregister(prod.new_box()) {
|
||||||
Ok(_) => {
|
Ok(_) => {
|
||||||
if let Err(er) = registry.register(Box::new(metric)) {
|
if let Err(er) = registry.register(prod.new_box()) {
|
||||||
warn!("Cannot update metric `{}`", metric_name);
|
warn!("Cannot update metric `{}`", metric_name);
|
||||||
return Err(anyhow::Error::msg(
|
return Err(anyhow::Error::msg(
|
||||||
format!("Cannot update metric `{}` due to {}", metric_name, er)
|
format!("Cannot update metric `{}` due to {}", metric_name, er)
|
||||||
|
|
|
||||||
|
|
@ -27,6 +27,8 @@ struct AppState {
|
||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main() -> anyhow::Result<()> {
|
async fn main() -> anyhow::Result<()> {
|
||||||
|
dotenv().ok();
|
||||||
|
|
||||||
let log_level = std::env::var("PROMETHEUS_EXPORTER_LOG_LEVEL")
|
let log_level = std::env::var("PROMETHEUS_EXPORTER_LOG_LEVEL")
|
||||||
.unwrap_or_else(|_| "INFO".to_owned());
|
.unwrap_or_else(|_| "INFO".to_owned());
|
||||||
|
|
||||||
|
|
@ -41,7 +43,6 @@ async fn main() -> anyhow::Result<()> {
|
||||||
.init();
|
.init();
|
||||||
|
|
||||||
info!("Loading env vars from .env if exists ...");
|
info!("Loading env vars from .env if exists ...");
|
||||||
dotenv().ok();
|
|
||||||
|
|
||||||
info!("Initializing local Prometehus metrics registry ...");
|
info!("Initializing local Prometehus metrics registry ...");
|
||||||
let registry = Registry::new();
|
let registry = Registry::new();
|
||||||
|
|
|
||||||
|
|
@ -1,10 +1,11 @@
|
||||||
|
use std::sync::Arc;
|
||||||
|
|
||||||
use crate::structs::v3::MetricOutput;
|
use crate::structs::v3::MetricOutput;
|
||||||
use serde::de;
|
|
||||||
use serde_json::{Map, Value};
|
use serde_json::{Map, Value};
|
||||||
use prometheus::Gauge;
|
use prometheus::Gauge;
|
||||||
use tracing::error;
|
use tracing::error;
|
||||||
use prometheus::Opts;
|
use prometheus::Opts;
|
||||||
use prometheus::GaugeVec;
|
use prometheus::{opts, GaugeVec, core::Collector};
|
||||||
use tracing::{debug, trace};
|
use tracing::{debug, trace};
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
|
|
@ -20,7 +21,6 @@ pub struct MetricsProcesser;
|
||||||
|
|
||||||
|
|
||||||
impl MetricsProcesser {
|
impl MetricsProcesser {
|
||||||
|
|
||||||
pub fn get_type_of_value(metrics: &MetricOutput) -> MetricsValueType {
|
pub fn get_type_of_value(metrics: &MetricOutput) -> MetricsValueType {
|
||||||
trace!("defining metric type");
|
trace!("defining metric type");
|
||||||
if Self::is_number(metrics) {
|
if Self::is_number(metrics) {
|
||||||
|
|
@ -48,8 +48,14 @@ impl MetricsProcesser {
|
||||||
metric: &MetricOutput,
|
metric: &MetricOutput,
|
||||||
metric_name: &str,
|
metric_name: &str,
|
||||||
metric_desc: &str
|
metric_desc: &str
|
||||||
) -> Option<Gauge> {
|
) -> Option<Box<dyn Collector>> {
|
||||||
trace!("fn gauge_from_number is running");
|
trace!("fn gauge_from_number is running");
|
||||||
|
if let Some(status) = metric.status {
|
||||||
|
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));
|
||||||
|
return Some(Box::new(vec));
|
||||||
|
}
|
||||||
|
|
||||||
let gauge = Gauge::new(
|
let gauge = Gauge::new(
|
||||||
metric_name,
|
metric_name,
|
||||||
metric_desc
|
metric_desc
|
||||||
|
|
@ -57,10 +63,6 @@ impl MetricsProcesser {
|
||||||
|
|
||||||
match gauge {
|
match gauge {
|
||||||
Ok(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() {
|
let val = match metric.value.as_number() {
|
||||||
Some(val) => {
|
Some(val) => {
|
||||||
val.as_f64().unwrap_or_else(||
|
val.as_f64().unwrap_or_else(||
|
||||||
|
|
@ -74,7 +76,7 @@ impl MetricsProcesser {
|
||||||
};
|
};
|
||||||
gauge.set(val);
|
gauge.set(val);
|
||||||
debug!("processed metric: {:?}", &gauge);
|
debug!("processed metric: {:?}", &gauge);
|
||||||
return Some(gauge);
|
return Some(Box::new(gauge));
|
||||||
},
|
},
|
||||||
Err(er) => error!("Cannot create Gauge metric {} due to {}", &metric_name, er),
|
Err(er) => error!("Cannot create Gauge metric {} due to {}", &metric_name, er),
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue