add prometheus bmp metrics

Author: matthieugouel

README.md

@@ -26,7 +26,7 @@ Conversely, Risotto can be configured to stream updates as is to the event pipel
 
 ## Quick Start
 
-The easiest way to use risotto is using Docker.
+The easiest way to use Risotto with Docker.
 
 * Create a `risotto.yml` configuration file
 

integration/README.md

@@ -16,7 +16,7 @@ The `bmp.from_kafka` table is using the ClickHouse [Kafka engine](https://clickh
 * Start the environment
 
 ```sh
-docker compose up --build --force-recreate --renew-anon-volumes
+docker compose up -d --build --force-recreate --renew-anon-volumes
 ```
 
 * Check Risotto state:

integration/config/risotto/risotto.yml

@@ -13,6 +13,6 @@ kafka:
   topic: risotto-updates
 
 state:
-  enable: false
+  enable: true
   path: /app/state.json
   save_interval: 10

risotto-lib/src/lib.rs

@@ -1,11 +1,13 @@
 pub mod processor;
 pub mod state;
 pub mod state_store;
+pub mod statistics;
 pub mod update;
 
 use bgpkit_parser::parser::bmp::messages::BmpMessageBody;
 use bytes::Bytes;
 use core::net::IpAddr;
+use statistics::AsyncStatistics;
 use std::sync::mpsc::Sender;
 use tracing::{debug, error, info, trace};
 
@@ -18,6 +20,7 @@ use crate::update::{new_metadata, Update};
 
 pub async fn process_bmp_message<T: StateStore>(
     state: Option<AsyncState<T>>,
+    statistics: AsyncStatistics,
     tx: Sender<Update>,
     router_addr: IpAddr,
     router_port: u16,
@@ -32,22 +35,27 @@ pub async fn process_bmp_message<T: StateStore>(
         }
     };
 
+    let mut statistics_lock = statistics.lock().await;
+
     trace!("[{}]:{} - {:?}", router_addr, router_port, message);
+    statistics_lock.rx_bmp_messages += 1;
 
     // Extract header and peer information
     let metadata = new_metadata(router_addr, router_port, &message);
 
     match message.message_body {
         BmpMessageBody::InitiationMessage(body) => {
+            trace!("{:?}", body);
             let tlvs_info = body
                 .tlvs
                 .iter()
                 .map(|tlv| tlv.info.clone())
                 .collect::<Vec<_>>();
-            info!(
+            debug!(
                 "[{}]:{} - InitiationMessage: {:?}",
                 router_addr, router_port, tlvs_info
             );
+            statistics_lock.rx_bmp_initiation += 1;
             // No-Op
         }
         BmpMessageBody::PeerUpNotification(body) => {
@@ -60,10 +68,11 @@ pub async fn process_bmp_message<T: StateStore>(
                 return;
             }
             let metadata = metadata.unwrap();
-            info!(
+            debug!(
                 "[{}]:{} - PeerUpNotification - {}",
                 metadata.router_addr, metadata.router_port, metadata.peer_addr
             );
+            statistics_lock.rx_bmp_peer_up += 1;
             peer_up_notification(state, tx, metadata, body).await;
         }
         BmpMessageBody::RouteMonitoring(body) => {
@@ -76,10 +85,13 @@ pub async fn process_bmp_message<T: StateStore>(
                 return;
             }
             let metadata = metadata.unwrap();
+            statistics_lock.rx_bmp_route_monitoring += 1;
             route_monitoring(state, tx, metadata, body).await;
         }
-        BmpMessageBody::RouteMirroring(_) => {
-            info!("[{}]:{} - RouteMirroring", router_addr, router_port)
+        BmpMessageBody::RouteMirroring(body) => {
+            trace!("{:?}", body);
+            debug!("[{}]:{} - RouteMirroring", router_addr, router_port);
+            statistics_lock.rx_bmp_route_mirroring += 1;
             // No-Op
         }
         BmpMessageBody::PeerDownNotification(body) => {
@@ -92,20 +104,25 @@ pub async fn process_bmp_message<T: StateStore>(
                 return;
             }
             let metadata = metadata.unwrap();
-            info!(
+            debug!(
                 "[{}]:{} - PeerDownNotification: - {}",
                 metadata.router_addr, metadata.router_port, metadata.peer_addr
             );
+            statistics_lock.rx_bmp_peer_down += 1;
             peer_down_notification(state, tx, metadata, body).await;
         }
 
-        BmpMessageBody::TerminationMessage(_) => {
-            info!("[{}]:{} - TerminationMessage", router_addr, router_port)
+        BmpMessageBody::TerminationMessage(body) => {
+            trace!("{:?}", body);
+            info!("[{}]:{} - TerminationMessage", router_addr, router_port);
+            statistics_lock.rx_bmp_termination += 1;
             // No-Op
         }
-        BmpMessageBody::StatsReport(_) => {
-            info!("[{}]:{} - StatsReport", router_addr, router_port)
+        BmpMessageBody::StatsReport(body) => {
+            trace!("{:?}", body);
+            info!("[{}]:{} - StatsReport", router_addr, router_port);
+            statistics_lock.rx_bmp_stats_report += 1;
             // No-Op
         }
-    }
+    };
 }

risotto-lib/src/processor.rs

@@ -50,7 +50,7 @@ pub async fn route_monitoring<T: StateStore>(
 
     let mut legitimate_updates = Vec::new();
     if let Some(state) = &state {
-        let mut state_lock = state.lock().unwrap();
+        let mut state_lock = state.lock().await;
         for update in potential_updates {
             let is_updated = state_lock
                 .update(&metadata.router_addr.clone(), &metadata.peer_addr, &update)
@@ -80,7 +80,7 @@ pub async fn peer_down_notification<T: StateStore>(
     if let Some(state) = state {
         // Remove the peer and the associated updates from the state
         // We start by emiting synthetic withdraw updates
-        let mut state_lock = state.lock().unwrap();
+        let mut state_lock = state.lock().await;
 
         let mut synthetic_updates = Vec::new();
         let updates = state_lock

risotto-lib/src/state.rs

@@ -5,8 +5,9 @@ use serde::{Deserialize, Serialize};
 use std::error::Error;
 use std::hash::{Hash, Hasher};
 use std::sync::mpsc::Sender;
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 use std::time::Duration;
+use tokio::sync::Mutex;
 use tokio::time::sleep;
 use tracing::{info, trace};
 
@@ -125,7 +126,7 @@ pub async fn peer_up_withdraws_handler<T: StateStore>(
         metadata.router_addr, metadata.router_port, metadata.peer_addr, startup, sleep_time
     );
 
-    let state_lock = state.lock().unwrap();
+    let state_lock = state.lock().await;
     let timed_prefixes = state_lock
         .store
         .get_updates_by_peer(&metadata.router_addr, &metadata.peer_addr);
@@ -149,7 +150,7 @@ pub async fn peer_up_withdraws_handler<T: StateStore>(
         synthetic_updates.len()
     );
 
-    let mut state_lock = state.lock().unwrap();
+    let mut state_lock = state.lock().await;
     for update in &mut synthetic_updates {
         trace!("{:?}", update);
 

risotto-lib/src/statistics.rs

@@ -0,0 +1,34 @@
+use std::sync::Arc;
+use tokio::sync::Mutex;
+
+pub type AsyncStatistics = Arc<Mutex<ProcessorStatistics>>;
+
+pub fn new_statistics() -> AsyncStatistics {
+    Arc::new(Mutex::new(ProcessorStatistics::default()))
+}
+
+pub struct ProcessorStatistics {
+    pub rx_bmp_messages: usize,
+    pub rx_bmp_initiation: usize,
+    pub rx_bmp_peer_up: usize,
+    pub rx_bmp_route_monitoring: usize,
+    pub rx_bmp_route_mirroring: usize,
+    pub rx_bmp_peer_down: usize,
+    pub rx_bmp_termination: usize,
+    pub rx_bmp_stats_report: usize,
+}
+
+impl Default for ProcessorStatistics {
+    fn default() -> Self {
+        Self {
+            rx_bmp_messages: 0,
+            rx_bmp_initiation: 0,
+            rx_bmp_peer_up: 0,
+            rx_bmp_route_monitoring: 0,
+            rx_bmp_route_mirroring: 0,
+            rx_bmp_peer_down: 0,
+            rx_bmp_termination: 0,
+            rx_bmp_stats_report: 0,
+        }
+    }
+}

risotto/src/api.rs

@@ -1,13 +1,10 @@
 use axum::{extract::State as AxumState, routing::get, Json, Router};
 use core::net::IpAddr;
-use metrics::{Key, Label, Recorder};
-use metrics_exporter_prometheus::PrometheusBuilder;
+use metrics::{counter, gauge, Label};
+use metrics_exporter_prometheus::PrometheusHandle;
 use serde::{Deserialize, Serialize};
 
-use risotto_lib::{state::AsyncState, state_store::store::StateStore};
-
-static METADATA: metrics::Metadata =
-    metrics::Metadata::new(module_path!(), metrics::Level::INFO, Some(module_path!()));
+use risotto_lib::{state::AsyncState, state_store::store::StateStore, statistics::AsyncStatistics};
 
 #[derive(Debug, Serialize, Deserialize)]
 struct APIRouter {
@@ -24,28 +21,38 @@ struct APIPeer {
 
 struct AppState<T: StateStore> {
     state: Option<AsyncState<T>>,
+    statistics: AsyncStatistics,
+    prom_handle: PrometheusHandle,
 }
 
 impl<T: StateStore> Clone for AppState<T> {
     fn clone(&self) -> Self {
         Self {
             state: self.state.clone(),
+            statistics: self.statistics.clone(),
+            prom_handle: self.prom_handle.clone(),
         }
     }
 }
 
-pub fn app<T: StateStore>(state: Option<AsyncState<T>>) -> Router {
-    let app_state = AppState { state };
-
+pub fn app<T: StateStore>(
+    state: Option<AsyncState<T>>,
+    statistics: AsyncStatistics,
+    prom_handle: PrometheusHandle,
+) -> Router {
+    let app_state = AppState {
+        state,
+        statistics,
+        prom_handle,
+    };
     Router::new()
         .route("/", get(root).with_state(app_state.clone()))
         .route("/metrics", get(metrics).with_state(app_state.clone()))
 }
 
 async fn format<T: StateStore>(state: AsyncState<T>) -> Vec<APIRouter> {
     let mut api_routers: Vec<APIRouter> = Vec::new();
-    let state = state.lock().unwrap();
-
+    let state = state.lock().await;
     for (router_addr, peer_addr, update_prefix) in state.get_all().unwrap() {
         // Find the router in the list of routers
         let mut router = None;
@@ -102,7 +109,7 @@ async fn format<T: StateStore>(state: AsyncState<T>) -> Vec<APIRouter> {
 }
 
 async fn root<T: StateStore>(
-    AxumState(AppState { state }): AxumState<AppState<T>>,
+    AxumState(AppState { state, .. }): AxumState<AppState<T>>,
 ) -> Json<Vec<APIRouter>> {
     match state.as_ref() {
         Some(state) => {
@@ -113,45 +120,52 @@ async fn root<T: StateStore>(
     }
 }
 
-async fn metrics<T: StateStore>(AxumState(AppState { state }): AxumState<AppState<T>>) -> String {
-    if state.is_none() {
-        return String::new();
-    }
-
-    let state = state.unwrap();
-
-    let recorder = PrometheusBuilder::new().build_recorder();
-    let api_routers = format(state).await;
-
-    recorder.describe_gauge(
-        "risotto_bgp_peers".into(),
-        None,
-        "Number of BGP peers per router".into(),
-    );
-    for api_router in &api_routers {
-        let labels = vec![Label::new("router", api_router.router_addr.to_string())];
-        let key = Key::from_parts("risotto_bgp_peers", labels);
-        recorder
-            .register_gauge(&key, &METADATA)
-            .set(api_router.peers.len() as f64);
-    }
+async fn metrics<T: StateStore>(
+    AxumState(AppState {
+        state,
+        statistics,
+        prom_handle,
+    }): AxumState<AppState<T>>,
+) -> String {
+    // Set the state metrics if enabled
+    if !state.is_none() {
+        let state = state.unwrap();
+        let api_routers = format(state).await;
+
+        for api_router in &api_routers {
+            let labels = vec![Label::new("router", api_router.router_addr.to_string())];
+            gauge!("risotto_state_bgp_peers", labels).set(api_router.peers.len() as f64);
+        }
 
-    recorder.describe_gauge(
-        "risotto_bgp_updates".into(),
-        None,
-        "Number of BGP updates per (router, peer)".into(),
-    );
-    for api_router in &api_routers {
-        for api_peer in &api_router.peers {
-            let total = api_peer.ipv4 + api_peer.ipv6;
-            let labels = vec![
-                Label::new("router", api_router.router_addr.to_string()),
-                Label::new("peer", api_peer.peer_addr.to_string()),
-            ];
-            let key = Key::from_parts("risotto_bgp_updates", labels);
-            recorder.register_gauge(&key, &METADATA).set(total as f64);
+        for api_router in &api_routers {
+            for api_peer in &api_router.peers {
+                let total = api_peer.ipv4 + api_peer.ipv6;
+                let labels = vec![
+                    Label::new("router", api_router.router_addr.to_string()),
+                    Label::new("peer", api_peer.peer_addr.to_string()),
+                ];
+                gauge!("risotto_state_bgp_updates", labels).set(total as f64);
+            }
         }
     }
 
-    recorder.handle().render()
+    // Set the statistics metrics
+    let statistics = statistics.lock().await;
+    let metric_name = "risotto_bmp_messages_total";
+    counter!(metric_name, vec![Label::new("type", "initiation")])
+        .absolute(statistics.rx_bmp_initiation as u64);
+    counter!(metric_name, vec![Label::new("type", "peer_up")])
+        .absolute(statistics.rx_bmp_peer_up as u64);
+    counter!(metric_name, vec![Label::new("type", "route_monitoring")])
+        .absolute(statistics.rx_bmp_route_monitoring as u64);
+    counter!(metric_name, vec![Label::new("type", "route_mirroring")])
+        .absolute(statistics.rx_bmp_route_mirroring as u64);
+    counter!(metric_name, vec![Label::new("type", "peer_down")])
+        .absolute(statistics.rx_bmp_peer_down as u64);
+    counter!(metric_name, vec![Label::new("type", "termination")])
+        .absolute(statistics.rx_bmp_termination as u64);
+    counter!(metric_name, vec![Label::new("type", "stats_report")])
+        .absolute(statistics.rx_bmp_stats_report as u64);
+
+    prom_handle.render()
 }