feat: Test metric for the price - draft

Cargo.toml

@@ -16,7 +16,6 @@ serde_json = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 once_cell = "1.10"
 prometheus = "0.13"
-lazy_static = "1.4"
 warp = "0.3.7"
 
 

docker-compose.yml

@@ -20,8 +20,6 @@ services:
       - ./grafana/prometheus.yml:/etc/prometheus/prometheus.yml
     ports:
       - "9090:9090"
-    networks:
-      - oracle
   grafana:
     container_name: grafana
     image: grafana/grafana
@@ -30,8 +28,7 @@ services:
       - "3000:3000"
     volumes:
       - ./grafana/provisioning/datasources:/etc/grafana/provisioning/datasources
-    networks:
-      - oracle
-networks:
-  oracle:
-    driver: bridge
+#networks:
+#  oracle:
+#    driver: bridge
+#    external: false

grafana/prometheus.yml

@@ -1,8 +1,8 @@
 scrape_configs:
   - job_name: 'RingBenchMonitor'
-    metrics_path: '/actuator/prometheus'
-    scrape_interval: 3s
+    metrics_path: '/metrics'
+    scrape_interval: 1s
     static_configs:
-      - targets: ['host.docker.internal:8080']
+      - targets: ['host.docker.internal:3030']
         labels:
           application: 'RingBenchMonitor Prometheus data'

src/domain/services/trade_history_service.rs

@@ -5,6 +5,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use once_cell::sync::Lazy;
 use tokio::sync::Mutex;
 use crate::domain::entities::trade::{TradeData, TradeSD};
+use crate::ports::stats::stats_recorder_service::StatsRecorderService;
 
 // Static singleton for global trade history storage
 pub static TRADE_HISTORY: Lazy<Arc<Mutex<VecDeque<TradeData>>>> = Lazy::new(|| {
@@ -23,9 +24,11 @@ impl TradeHistoryService {
         let current_time = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as u64;
 
         log::info!("New Trade: {}",&trade_sd.data);
+
+        StatsRecorderService::track_price(trade_sd.data.price.parse::<f64>().unwrap_or(0.0));
         // Keep trades within a larger window (e.g., 70 seconds)
         while let Some(oldest_trade) = trades.front() {
-            if current_time - oldest_trade.trade_time > 70_000 {
+            if current_time >= oldest_trade.trade_time && current_time - oldest_trade.trade_time > 70_000 {
                 trades.pop_front();
             } else {
                 break;

src/main.rs

@@ -1,3 +1,6 @@
+use std::sync::Arc;
+use warp::Filter;
+use crate::ports::stats::stats_recorder_service::StatsRecorderService;
 use crate::ports::ws_client_trade;
 
 mod ports;
@@ -11,12 +14,24 @@ async fn main() {
     // Log the start of the application
     log::info!("Starting application...");
 
-    //let trade_history_service = Arc::new(TradeHistoryService);
-
     let websocket_trade_handle = tokio::spawn( async{
         log::info!("Starting Trade Stream WebSocket client...");
         ws_client_trade::start_websocket().await;
     });
 
-    tokio::try_join!(websocket_trade_handle).unwrap();
+    // Serve metrics on port 3030
+    let metrics_server = tokio::spawn(async {
+
+        // Metrics route for Prometheus
+        let metrics_route = warp::path("metrics")
+            .and(warp::get())
+            .map(|| {
+                let metrics = StatsRecorderService::serve_metrics();
+                warp::reply::with_header(metrics, "Content-Type", "text/plain")
+            });
+
+        warp::serve(metrics_route).run(([0, 0, 0, 0], 3030)).await;
+    });
+
+    tokio::try_join!(websocket_trade_handle,metrics_server).unwrap();
 }

src/ports/mod.rs

@@ -1 +1,2 @@
-pub mod ws_client_trade;
+pub mod ws_client_trade;
+pub mod stats;

src/ports/stats/mod.rs

@@ -1 +1 @@
-mod stats_recorder;
+pub mod stats_recorder_service;

src/ports/stats/stats_recorder_service.rs

@@ -0,0 +1,43 @@
+use std::collections::VecDeque;
+use prometheus::{Gauge, Encoder, TextEncoder, register_gauge, default_registry, Registry};
+use std::sync::{Arc, Mutex};
+use once_cell::sync::Lazy;
+use crate::config::CONFIG;
+use crate::domain::entities::trade::TradeData;
+
+// Static singleton for global price gauge storage
+pub static PRICE_GAUGE: Lazy<Arc<Mutex<Gauge>>> = Lazy::new(|| {
+    let gauge = register_gauge!(
+        "crypto_price",
+        "Current price of the trading pair"
+    ).unwrap();
+
+    let registry = Registry::new();
+
+    registry.register(Box::new(gauge.clone())).expect("Failed to register gauge");
+    // Register gauge with the default registry manually
+    //default_registry().register(Box::new(gauge.clone())).expect("Failed to register gauge");
+    log::info!("Prometheus Gauge  was registered");
+    Arc::new(Mutex::new(gauge))
+});
+
+#[derive(Debug, Clone, Default)]
+pub struct StatsRecorderService;
+
+impl StatsRecorderService{
+    pub fn track_price(price: f64) {
+        // Update the shared gauge
+        let gauge = PRICE_GAUGE.lock().unwrap();
+        //log::info!("PRICE:: {}",price);
+        gauge.set(price);
+    }
+
+    pub fn serve_metrics() -> String {
+        // Serve the Prometheus metrics via HTTP
+        let encoder = TextEncoder::new();
+        let mut buffer = Vec::new();
+        let metrics = prometheus::gather();
+        encoder.encode(&metrics, &mut buffer).unwrap();
+        String::from_utf8(buffer).unwrap()
+    }
+}