Merge pull request #29 from nxthdr/add-tests

adding test of processors function

Author: matthieugouel

README.md

@@ -4,9 +4,9 @@
 
 # Risotto
 
-Risotto 😋 is a BGP updates collector that gathers BGP updates from routers via the BMP protocol. This repository contains the Risotto collector application and the Risotto library.
+Risotto 😋 is a BGP collector that processes BMP protocol messages from routers and publishes updates to Kafka/Redpanda. This repository includes both the Risotto collector application and the Risotto library.
 
-The application streams the BGP updates to a Kafka topic, which can be consumed by other components downstream. The library provides all the necessary components to decode BMP messages and produce BGP updates.
+The collector application streams BGP updates to a Kafka topic, enabling downstream components to consume them. The library offers essential components for decoding BMP messages and generating BGP updates.
 
 ## State Management
 
@@ -17,7 +17,7 @@ This state addresses two challenges when handling BMP data:
 
 Duplicate announcements could, in theory, be handled by the database, but less data manipulation is better. Instead, Risotto checks each incoming update against its state. If the prefix is already present, the update is discarded.
 
-For Peer Down notifications, Risotto leverages its state to generate synthetic withdraws for the prefixes announced by the downed peer.
+Missing withdraws are generated synthetically when receiving Peer Down notifications for the prefixes already announced by the downed peer.
 
 For persistance, Risotto dumps its state at specified interval, and fetches it at startup. Risotto is able to infer any missing withdraws that would have occured during downtime, from the initial peer up flow. This ensures the database remains accurate, even if the collector is restarted. On the other hand, a restart may result in duplicate announcements.
 In other words, Risotto guaranties that the database is always in a consistent state, but may contain some duplicate announcements.

risotto-lib/Cargo.toml

@@ -18,5 +18,5 @@ bgpkit-parser = { version = "0.11.0", features = ["serde"] }
 bytes = "1.9.0"
 rand = "0.9.0"
 serde = { version = "1.0.217", features = ["derive"] }
-tokio = "1.42.0"
+tokio = { version = "1.42.0", features = ["macros"] }
 tracing = "0.1.41"

risotto-lib/src/lib.rs

@@ -1,3 +1,90 @@
-pub mod handler;
+pub mod processor;
 pub mod state;
 pub mod update;
+
+use bgpkit_parser::parser::bmp::messages::BmpMessageBody;
+use bytes::Bytes;
+use core::net::IpAddr;
+use std::sync::mpsc::Sender;
+use tracing::{error, info, trace};
+
+use crate::processor::{
+    decode_bmp_message, peer_down_notification, peer_up_notification, route_monitoring,
+};
+use crate::state::AsyncState;
+use crate::update::{new_metadata, Update};
+
+pub async fn process_bmp_message(
+    state: Option<AsyncState>,
+    tx: Sender<Update>,
+    router_addr: IpAddr,
+    router_port: u16,
+    bytes: &mut Bytes,
+) {
+    // Parse the BMP message
+    let message = match decode_bmp_message(bytes) {
+        Ok(message) => message,
+        Err(e) => {
+            error!("failed to decode BMP message: {}", e);
+            return;
+        }
+    };
+
+    // Extract header and peer information
+    let metadata = match new_metadata(router_addr, router_port, message.clone()) {
+        Some(m) => m,
+        None => return,
+    };
+
+    match message.message_body {
+        BmpMessageBody::InitiationMessage(_) => {
+            info!(
+                "InitiationMessage: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            )
+            // No-Op
+        }
+        BmpMessageBody::PeerUpNotification(body) => {
+            trace!("{:?}", body);
+            info!(
+                "PeerUpNotification: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            );
+            peer_up_notification(state, tx, metadata, body).await;
+        }
+        BmpMessageBody::RouteMonitoring(body) => {
+            trace!("{:?}", body);
+            route_monitoring(state, tx, metadata, body).await;
+        }
+        BmpMessageBody::RouteMirroring(_) => {
+            info!(
+                "RouteMirroring: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            )
+            // No-Op
+        }
+        BmpMessageBody::PeerDownNotification(body) => {
+            trace!("{:?}", body);
+            info!(
+                "PeerDownNotification: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            );
+            peer_down_notification(state, tx, metadata, body).await;
+        }
+
+        BmpMessageBody::TerminationMessage(_) => {
+            info!(
+                "TerminationMessage: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            )
+            // No-Op
+        }
+        BmpMessageBody::StatsReport(_) => {
+            info!(
+                "StatsReport: {} - {}",
+                metadata.router_addr, metadata.peer_addr
+            )
+            // No-Op
+        }
+    }
+}

risotto-lib/src/handler.rs risotto-lib/src/processor.rsrisotto-lib/src/handler.rs renamed to risotto-lib/src/processor.rs

@@ -1,15 +1,14 @@
 use anyhow::Result;
 use bgpkit_parser::bmp::messages::{PeerDownNotification, PeerUpNotification, RouteMonitoring};
 use bgpkit_parser::parse_bmp_msg;
-use bgpkit_parser::parser::bmp::messages::{BmpMessage, BmpMessageBody};
+use bgpkit_parser::parser::bmp::messages::BmpMessage;
 use bytes::Bytes;
-use core::net::IpAddr;
 use std::sync::mpsc::Sender;
-use tracing::{error, info, trace};
+use tracing::trace;
 
 use crate::state::AsyncState;
 use crate::state::{peer_up_withdraws_handler, synthesize_withdraw_update};
-use crate::update::{decode_updates, new_metadata, new_peer_from_metadata, Update, UpdateMetadata};
+use crate::update::{decode_updates, new_peer_from_metadata, Update, UpdateMetadata};
 
 pub fn decode_bmp_message(bytes: &mut Bytes) -> Result<BmpMessage> {
     let message = match parse_bmp_msg(bytes) {
@@ -20,7 +19,7 @@ pub fn decode_bmp_message(bytes: &mut Bytes) -> Result<BmpMessage> {
     Ok(message)
 }
 
-pub async fn process_peer_up_notification(
+pub async fn peer_up_notification(
     state: Option<AsyncState>,
     tx: Sender<Update>,
     metadata: UpdateMetadata,
@@ -34,7 +33,7 @@ pub async fn process_peer_up_notification(
     }
 }
 
-pub async fn process_route_monitoring(
+pub async fn route_monitoring(
     state: Option<AsyncState>,
     tx: Sender<Update>,
     metadata: UpdateMetadata,
@@ -66,7 +65,7 @@ pub async fn process_route_monitoring(
     }
 }
 
-pub async fn process_peer_down_notification(
+pub async fn peer_down_notification(
     state: Option<AsyncState>,
     tx: Sender<Update>,
     metadata: UpdateMetadata,
@@ -100,78 +99,3 @@ pub async fn process_peer_down_notification(
         }
     }
 }
-
-pub async fn process_bmp_message(
-    state: Option<AsyncState>,
-    tx: Sender<Update>,
-    router_addr: IpAddr,
-    router_port: u16,
-    bytes: &mut Bytes,
-) {
-    // Parse the BMP message
-    let message = match decode_bmp_message(bytes) {
-        Ok(message) => message,
-        Err(e) => {
-            error!("failed to decode BMP message: {}", e);
-            return;
-        }
-    };
-
-    // Extract header and peer information
-    let metadata = match new_metadata(router_addr, router_port, message.clone()) {
-        Some(m) => m,
-        None => return,
-    };
-
-    match message.message_body {
-        BmpMessageBody::InitiationMessage(_) => {
-            info!(
-                "InitiationMessage: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            )
-            // No-Op
-        }
-        BmpMessageBody::PeerUpNotification(body) => {
-            trace!("{:?}", body);
-            info!(
-                "PeerUpNotification: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            );
-            process_peer_up_notification(state, tx, metadata, body).await;
-        }
-        BmpMessageBody::RouteMonitoring(body) => {
-            trace!("{:?}", body);
-            process_route_monitoring(state, tx, metadata, body).await;
-        }
-        BmpMessageBody::RouteMirroring(_) => {
-            info!(
-                "RouteMirroring: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            )
-            // No-Op
-        }
-        BmpMessageBody::PeerDownNotification(body) => {
-            trace!("{:?}", body);
-            info!(
-                "PeerDownNotification: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            );
-            process_peer_down_notification(state, tx, metadata, body).await;
-        }
-
-        BmpMessageBody::TerminationMessage(_) => {
-            info!(
-                "TerminationMessage: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            )
-            // No-Op
-        }
-        BmpMessageBody::StatsReport(_) => {
-            info!(
-                "StatsReport: {} - {}",
-                metadata.router_addr, metadata.peer_addr
-            )
-            // No-Op
-        }
-    }
-}

risotto-lib/src/state.rs

@@ -1,4 +1,4 @@
-use bgpkit_parser::models::{Origin, Peer as BGPkitPeer};
+use bgpkit_parser::models::Peer as BGPkitPeer;
 use chrono::Utc;
 use core::net::IpAddr;
 use rand::Rng;
@@ -223,7 +223,7 @@ pub fn synthesize_withdraw_update(prefix: TimedPrefix, metadata: UpdateMetadata)
         prefix_addr: prefix.prefix_addr,
         prefix_len: prefix.prefix_len,
         announced: false,
-        origin: Origin::INCOMPLETE,
+        origin: "INCOMPLETE".to_string(),
         path: vec![],
         communities: vec![],
         is_post_policy: prefix.is_post_policy,

risotto-lib/src/update.rs

@@ -30,7 +30,7 @@ pub struct Update {
     pub announced: bool,
     pub is_post_policy: bool,
     pub is_adj_rib_out: bool,
-    pub origin: Origin,
+    pub origin: String,
     pub path: Vec<u32>,
     pub communities: Vec<(u32, u16)>,
     pub synthetic: bool,
@@ -95,7 +95,7 @@ pub fn decode_updates(message: RouteMonitoring, metadata: UpdateMetadata) -> Opt
                     announced,
                     is_post_policy: metadata.is_post_policy,
                     is_adj_rib_out: metadata.is_adj_rib_out,
-                    origin,
+                    origin: origin.to_string(),
                     path: new_path(path.clone()),
                     communities: new_communities(&communities.clone()),
                     synthetic: false,