fixing prometheus server

bin/autobahn-router/src/edge_updater.rs

@@ -403,7 +403,7 @@ impl EdgeUpdater {
         state.latest_slot_processed = state.latest_slot_pending;
 
         if started_at.elapsed() > Duration::from_millis(100) {
-            info!(
+            debug!(
                 "{} - refresh {} - took - {:?}",
                 self.dex.name,
                 refreshed_edges.len(),

bin/autobahn-router/src/main.rs

@@ -99,6 +99,14 @@ async fn main() -> anyhow::Result<()> {
     let config = Config::load(&args[1])?;
     let router_version = RouterVersion::OverestimateAmount;
 
+    if config.metrics.output_http {
+        let prom_bind_addr = config
+            .metrics
+            .prometheus_address
+            .clone()
+            .expect("prometheus_address must be set");
+        PrometheusSync::sync(prom_bind_addr);
+    }
     let hot_mints = Arc::new(RwLock::new(HotMintsCache::new(&config.hot_mints)));
 
     let mango_data = match mango::mango_fetcher::fetch_mango_data().await {
@@ -200,14 +208,6 @@ async fn main() -> anyhow::Result<()> {
         exit(-1);
     };
 
-    if config.metrics.output_http {
-        let prom_bind_addr = config
-            .metrics
-            .prometheus_address
-            .clone()
-            .expect("prometheus_address must be set");
-        let _prometheus = PrometheusSync::sync(prom_bind_addr);
-    }
     if config.metrics.output_stdout {
         warn!("metrics output to stdout is not supported yet");
     }

bin/autobahn-router/src/prometheus_sync.rs

@@ -1,25 +1,25 @@
-use std::time::Duration;
-
+use axum::{routing, Router};
 use prometheus::{Encoder, TextEncoder};
+use tokio::net::{TcpListener, ToSocketAddrs};
 use tokio::task::JoinHandle;
-use tokio::{
-    io::AsyncWriteExt,
-    net::{TcpListener, TcpStream, ToSocketAddrs},
-};
-use tracing::error;
+use tower_http::cors::{AllowHeaders, AllowMethods, Any, CorsLayer};
+use tracing::{error, info};
+
+use crate::server::errors::AppError;
 
 pub struct PrometheusSync;
 
 impl PrometheusSync {
-    fn create_response(payload: &str) -> String {
+    fn create_response(payload: String) -> String {
         format!(
             "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
             payload.len(),
             payload
         )
     }
 
-    async fn handle_stream(stream: &mut TcpStream) -> anyhow::Result<()> {
+    async fn get_prometheus_stream() -> Result<String, AppError> {
+        error!("got message for prometheus");
         let mut metrics_buffer = Vec::new();
         let encoder = TextEncoder::new();
 
@@ -29,29 +29,29 @@ impl PrometheusSync {
             .unwrap();
 
         let metrics_buffer = String::from_utf8(metrics_buffer).unwrap();
-        let response = Self::create_response(&metrics_buffer);
-
-        stream.writable().await?;
-        stream.write_all(response.as_bytes()).await?;
-
-        stream.flush().await?;
-
-        Ok(())
+        Ok(Self::create_response(metrics_buffer))
     }
 
     pub fn sync(addr: impl ToSocketAddrs + Send + 'static) -> JoinHandle<anyhow::Result<()>> {
         tokio::spawn(async move {
             let listener = TcpListener::bind(addr).await?;
 
-            loop {
-                let Ok((mut stream, _addr)) = listener.accept().await else {
-                    error!("Error accepting prometheus stream");
-                    tokio::time::sleep(Duration::from_millis(1)).await;
-                    continue;
-                };
+            let mut router: Router<()> = Router::new();
+            router = router.route("/metrics", routing::get(Self::get_prometheus_stream));
+
+            let cors = CorsLayer::new()
+                .allow_methods(AllowMethods::any())
+                .allow_headers(AllowHeaders::any())
+                .allow_origin(Any);
+
+            router = router.layer(cors);
+
+            let handle = axum::serve(listener, router);
+
+            info!("Prometheus Server started");
 
-                let _ = Self::handle_stream(&mut stream).await;
-            }
+            handle.await.expect("Prometheus Server failed");
+            Ok(())
         })
     }
 }

bin/autobahn-router/src/server/mod.rs

@@ -1,5 +1,5 @@
 pub mod alt_provider;
-mod errors;
+pub mod errors;
 pub mod hash_provider;
 pub mod http_server;
 pub mod live_account_provider;