chore: migrate to gasket prometheus exporter (#780)

Author: scarmuega

Cargo.lock

@@ -26,7 +26,8 @@ u5c = ["tonic"]
 # pallas = { path = "../pallas/pallas" }
 pallas = { git = "https://github.com/txpipe/pallas" }
 
-gasket = { version = "^0.6", features = ["derive"] }
+gasket = { version = "^0.7", features = ["derive"] }
+gasket-prometheus = { version = "^0.7" }
 # gasket = { path = "../../construkts/gasket-rs/gasket", features = ["derive"] }
 # gasket = { git = "https://github.com/construkts/gasket-rs.git", features = ["derive"] }
 
@@ -46,11 +47,9 @@ serde = { version = "1.0.152", features = ["derive"] }
 serde_json = { version = "1.0.104", features = ["arbitrary_precision"] }
 strum = "0.24"
 strum_macros = "0.25"
-prometheus_exporter_base = { version = "1.4.0", features = ["hyper", "hyper_server"] }
 unicode-truncate = "0.2.0"
 thiserror = "1.0.39"
 indicatif = "0.17.3"
-lazy_static = "1.4.0"
 tracing = "0.1.37"
 tracing-subscriber = "0.3.17"
 anyhow = "1.0.77"

Cargo.toml

@@ -1,20 +1,5 @@
-use std::{
-    sync::Mutex,
-    time::{Duration, Instant},
-};
-
-use gasket::{metrics::Reading, runtime::Tether};
-use lazy_static::lazy_static;
-use tracing::{debug, error, warn};
-
-#[derive(clap::ValueEnum, Clone, Default)]
-pub enum Mode {
-    /// shows progress as a plain sequence of logs
-    #[default]
-    Plain,
-    /// shows aggregated progress and metrics
-    Tui,
-}
+use gasket::{daemon::Daemon, metrics::Reading, runtime::Tether};
+use std::{sync::Arc, time::Duration};
 
 struct TuiConsole {
     chainsync_progress: indicatif::ProgressBar,
@@ -116,77 +101,15 @@ impl TuiConsole {
     }
 }
 
-struct PlainConsole {
-    last_report: Mutex<Instant>,
-}
-
-impl PlainConsole {
-    fn new() -> Self {
-        Self {
-            last_report: Mutex::new(Instant::now()),
-        }
+pub async fn render(daemon: Arc<Daemon>, tui_enabled: bool) {
+    if !tui_enabled {
+        return;
     }
 
-    fn refresh<'a>(&self, tethers: impl Iterator<Item = &'a Tether>) {
-        let mut last_report = self.last_report.lock().unwrap();
-
-        if last_report.elapsed() <= Duration::from_secs(10) {
-            return;
-        }
-
-        for tether in tethers {
-            match tether.check_state() {
-                gasket::runtime::TetherState::Dropped => {
-                    error!("[{}] stage tether has been dropped", tether.name());
-                }
-                gasket::runtime::TetherState::Blocked(_) => {
-                    warn!(
-                        "[{}] stage tehter is blocked or not reporting state",
-                        tether.name()
-                    );
-                }
-                gasket::runtime::TetherState::Alive(state) => {
-                    debug!("[{}] stage is alive with state: {:?}", tether.name(), state);
-                    match tether.read_metrics() {
-                        Ok(readings) => {
-                            for (key, value) in readings {
-                                debug!("[{}] metric `{}` = {:?}", tether.name(), key, value);
-                            }
-                        }
-                        Err(err) => {
-                            error!("[{}] error reading metrics: {}", tether.name(), err)
-                        }
-                    }
-                }
-            }
-        }
-
-        *last_report = Instant::now();
-    }
-}
-
-lazy_static! {
-    static ref TUI_CONSOLE: TuiConsole = TuiConsole::new();
-}
-
-lazy_static! {
-    static ref PLAIN_CONSOLE: PlainConsole = PlainConsole::new();
-}
-
-pub fn initialize(mode: &Option<Mode>) {
-    if !matches!(mode, Some(Mode::Tui)) {
-        tracing::subscriber::set_global_default(
-            tracing_subscriber::FmtSubscriber::builder()
-                .with_max_level(tracing::Level::DEBUG)
-                .finish(),
-        )
-        .unwrap();
-    }
-}
+    let tui = TuiConsole::new();
 
-pub fn refresh<'a>(mode: &Option<Mode>, tethers: impl Iterator<Item = &'a Tether>) {
-    match mode {
-        Some(Mode::Tui) => TUI_CONSOLE.refresh(tethers),
-        _ => PLAIN_CONSOLE.refresh(tethers),
+    loop {
+        tui.refresh(daemon.tethers());
+        tokio::time::sleep(Duration::from_secs(1)).await;
     }
 }