prometheus

Cargo.lock

@@ -1045,6 +1045,12 @@ dependencies = [
  "syn 0.15.44",
 ]
 
+[[package]]
+name = "doc-comment"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fea41bba32d969b513997752735605054bc0dfa92b4c56bf1189f2e174be7a10"
+
 [[package]]
 name = "eager"
 version = "0.1.0"
@@ -1240,6 +1246,12 @@ dependencies = [
  "percent-encoding",
 ]
 
+[[package]]
+name = "ftoa"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ca45aac12b6c561b6289bc68957cb1db3dccf870e1951d590202de5e24f1dd35"
+
 [[package]]
 name = "futures"
 version = "0.3.25"
@@ -1703,6 +1715,7 @@ checksum = "1885e79c1fc4b10f0e172c475f458b7f7b93061064d98c3293e98c5ba0c8b399"
 dependencies = [
  "autocfg",
  "hashbrown 0.12.3",
+ "serde",
 ]
 
 [[package]]
@@ -2100,6 +2113,7 @@ dependencies = [
  "procinfo",
  "serde",
  "serde_json",
+ "serde_prometheus",
  "solana-client",
  "solana-sdk",
  "solana-transaction-status",
@@ -3294,6 +3308,21 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_prometheus"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25fcd6131bac47a32328d1ba1ee15a27f8d91ab2e5920dba71dbe93d2648f6b1"
+dependencies = [
+ "ftoa",
+ "indexmap",
+ "itoa 0.4.8",
+ "lazy_static",
+ "regex",
+ "serde",
+ "snafu",
+]
+
 [[package]]
 name = "serde_urlencoded"
 version = "0.7.1"
@@ -3449,6 +3478,27 @@ version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a507befe795404456341dfab10cef66ead4c041f62b8b11bbb92bffe5d0953e0"
 
+[[package]]
+name = "snafu"
+version = "0.6.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eab12d3c261b2308b0d80c26fffb58d17eba81a4be97890101f416b478c79ca7"
+dependencies = [
+ "doc-comment",
+ "snafu-derive",
+]
+
+[[package]]
+name = "snafu-derive"
+version = "0.6.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1508efa03c362e23817f96cde18abed596a25219a8b2c66e8db33c03543d315b"
+dependencies = [
+ "proc-macro2 1.0.50",
+ "quote 1.0.23",
+ "syn 1.0.107",
+]
+
 [[package]]
 name = "socket2"
 version = "0.4.7"

Cargo.toml

@@ -37,3 +37,4 @@ procinfo = "0.4.2"
 tokio-postgres = "0.7.7"
 native-tls = "0.2.11"
 postgres-native-tls = "0.5.0"
+serde_prometheus = "0.1.6"

prometheus.yml

@@ -0,0 +1,13 @@
+global:
+  scrape_interval:     15s # By default, scrape targets every 15 seconds.
+
+scrape_configs:
+  # The job name is added as a label `job=<job_name>` to any timeseries scraped from this config.
+  - job_name: 'prometheus'
+    scrape_interval: 5s
+    static_configs:
+      - targets: ['localhost:9090']
+
+  - job_name: 'lite-rpc'
+    static_configs:
+      - targets: ['localhost:9500']

src/bridge.rs

@@ -4,8 +4,8 @@ use crate::{
     rpc::LiteRpcServer,
     tpu_manager::TpuManager,
     workers::{
-        BlockInformation, BlockListener, Cleaner, MetricsCapture, Postgres, TxSender,
-        WireTransaction,
+        BlockInformation, BlockListener, Cleaner, MetricsCapture, Postgres, PrometheusSync,
+        TxSender, WireTransaction,
     },
 };
 
@@ -100,7 +100,7 @@ impl LiteBridge {
         tx_send_interval: Duration,
         clean_interval: Duration,
         postgres_config: String,
-    ) -> anyhow::Result<[JoinHandle<anyhow::Result<()>>; 9]> {
+    ) -> anyhow::Result<[JoinHandle<anyhow::Result<()>>; 10]> {
         let (postgres_send, postgres_recv) = mpsc::unbounded_channel();
         let (postgres_connection, postgres) = Postgres::new(postgres_config).await?;
 
@@ -117,6 +117,7 @@ impl LiteBridge {
         );
 
         let metrics_capture = MetricsCapture::new(self.tx_sender.clone());
+        let prometheus_sync = PrometheusSync::new(metrics_capture.clone()).sync();
 
         let finalized_block_listenser = self
             .finalized_block_listenser
@@ -124,7 +125,6 @@ impl LiteBridge {
             .listen(Some(postgres_send.clone()));
 
         let confirmed_block_listenser = self.confirmed_block_listenser.clone().listen(None);
-
         let cleaner = Cleaner::new(
             self.tx_sender.clone(),
             [
@@ -173,6 +173,7 @@ impl LiteBridge {
             postgres_connection,
             postgres,
             metrics_capture.capture(Some(postgres_send)),
+            prometheus_sync,
             cleaner,
         ])
     }

src/workers/metrics_capture.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use log::{info, warn};
 use solana_transaction_status::TransactionConfirmationStatus;
 use tokio::{sync::RwLock, task::JoinHandle};
@@ -6,9 +8,10 @@ use super::{PostgresMpscSend, TxSender};
 use serde::{Deserialize, Serialize};
 
 /// Background worker which captures metrics
+#[derive(Clone)]
 pub struct MetricsCapture {
     tx_sender: TxSender,
-    metrics: RwLock<Metrics>,
+    metrics: Arc<RwLock<Metrics>>,
 }
 
 #[derive(Clone, Default, Debug, Serialize, Deserialize)]

src/workers/mod.rs

@@ -2,10 +2,12 @@ mod block_listenser;
 mod cleaner;
 mod metrics_capture;
 mod postgres;
+mod prometheus;
 mod tx_sender;
 
 pub use block_listenser::*;
 pub use cleaner::*;
 pub use metrics_capture::*;
 pub use postgres::*;
+pub use prometheus::*;
 pub use tx_sender::*;

src/workers/prometheus.rs

@@ -0,0 +1,59 @@
+use std::collections::HashMap;
+
+use tokio::{
+    io::AsyncWriteExt,
+    net::{TcpListener, TcpStream},
+    task::JoinHandle,
+};
+
+use super::MetricsCapture;
+
+#[derive(Clone)]
+pub struct PrometheusSync {
+    metrics_capture: MetricsCapture,
+}
+
+impl PrometheusSync {
+    pub fn new(metrics_capture: MetricsCapture) -> Self {
+        Self { metrics_capture }
+    }
+
+    fn create_response(payload: &str) -> String {
+        format!(
+            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
+            payload.len(),
+            payload
+        )
+    }
+
+    async fn handle_stream(&self, stream: &mut TcpStream) -> anyhow::Result<()> {
+        let metrics = self.metrics_capture.get_metrics().await;
+        let metrics = serde_prometheus::to_string(&metrics, None, HashMap::new())?;
+
+        let response = Self::create_response(&metrics);
+
+        stream.writable().await?;
+        stream.write_all(response.as_bytes()).await?;
+
+        stream.flush().await?;
+
+        Ok(())
+    }
+
+    pub fn sync(self) -> JoinHandle<anyhow::Result<()>> {
+        #[allow(unreachable_code)]
+        tokio::spawn(async move {
+            let listener = TcpListener::bind("[::]:9500").await?;
+
+            loop {
+                let Ok((mut stream, _addr)) =  listener.accept().await else {
+                    continue;
+                };
+
+                let _ = self.handle_stream(&mut stream).await;
+            }
+
+            Ok(())
+        })
+    }
+}