implement basic kafka pipeline

Cargo.toml

@@ -12,6 +12,7 @@ bytes = "1.6.0"
 chrono = "0.4.38"
 hex = "0.4.3"
 ipnet = "2.9.0"
+kafka = "0.10.0"
 prefix-trie = "0.3.0"
 serde = { version = "1.0.213", features = ["derive"] }
 tokio = { version = "1.37.0", features = ["full"] }

src/api.rs

@@ -1,9 +1,9 @@
 use axum::{extract::State, routing::get, Json, Router};
-use bgpkit_parser::models::{AsPath, AsPathSegment, MetaCommunity};
 use core::net::{IpAddr, Ipv4Addr};
 use serde::{Deserialize, Serialize};
 
 use crate::db::DB;
+use crate::update::{construct_as_path, construct_communities};
 
 #[derive(Debug, Serialize, Deserialize)]
 struct APIRouter {
@@ -33,43 +33,6 @@ pub fn app(db: DB) -> Router {
     Router::new().route("/", get(root).with_state(db))
 }
 
-fn construct_as_path(path: Option<AsPath>) -> Vec<u32> {
-    match path {
-        Some(mut path) => {
-            let mut contructed_path: Vec<u32> = Vec::new();
-            path.dedup_coalesce();
-            for segment in path.into_segments_iter() {
-                match segment {
-                    AsPathSegment::AsSequence(dedup_asns) => {
-                        for asn in dedup_asns {
-                            contructed_path.push(asn.to_u32());
-                        }
-                    }
-                    _ => (),
-                }
-            }
-            contructed_path
-        }
-        None => Vec::new(),
-    }
-}
-
-fn construct_communities(communities: Vec<MetaCommunity>) -> Vec<(u32, u16)> {
-    let mut constructed_communities = Vec::new();
-    for community in communities {
-        match community {
-            MetaCommunity::Plain(community) => match community {
-                bgpkit_parser::models::Community::Custom(asn, value) => {
-                    constructed_communities.push((asn.to_u32(), value));
-                }
-                _ => (), // TODO
-            },
-            _ => (), // TODO
-        }
-    }
-    constructed_communities
-}
-
 fn format(db: DB) -> Vec<APIRouter> {
     let routers = db.routers.lock().unwrap();
     let mut api_routers = Vec::new();

src/bmp.rs

@@ -8,8 +8,9 @@ use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
 
 use crate::db::DB;
+use crate::pipeline::send_to_kafka;
 use crate::router::new_router;
-use crate::update::decode_updates;
+use crate::update::{decode_updates, format_update};
 
 pub async fn unmarshal_bmp_packet(socket: &mut TcpStream) -> io::Result<Option<BmpMessage>> {
     // Get minimal packet length to get how many bytes to remove from the socket
@@ -81,9 +82,11 @@ pub async fn handle(socket: &mut TcpStream, db: DB) {
                 }
             }
 
-            // TODO: Handle streaming pipeline configuration (stdout, CSV file, Kafka, ...)
+            // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in legitimate_updates {
+                let update = format_update(&router, &peer, &update);
                 println!("{:?}", update);
+                send_to_kafka("broker.nxthdr.dev:9092", "bgp-updates", update.as_bytes());
             }
         }
         BmpMessageBody::PeerDownNotification(_) => {
@@ -102,9 +105,11 @@ pub async fn handle(socket: &mut TcpStream, db: DB) {
             // And we then update the internal state
             router.remove_peer(&peer);
 
-            // TODO: Handle streaming pipeline configuration (stdout, CSV file, Kafka, ...)
+            // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in synthetic_updates {
+                let update = format_update(&router, &peer, &update);
                 println!("{:?}", update);
+                send_to_kafka("broker.nxthdr.dev:9092", "bgp-updates", update.as_bytes());
             }
         }
         _ => (),

src/main.rs

@@ -1,6 +1,7 @@
 mod api;
 mod bmp;
 mod db;
+mod pipeline;
 mod router;
 mod update;
 

src/pipeline.rs

@@ -0,0 +1,24 @@
+use std::time::Duration;
+
+use kafka::error::Error as KafkaError;
+use kafka::producer::{Producer, Record, RequiredAcks};
+
+pub fn send_to_kafka(broker: &str, topic: &str, data: &[u8]) {
+    if let Err(e) = produce_message(data, topic, vec![broker.to_owned()]) {
+        println!("Failed producing messages: {}", e);
+    }
+}
+
+fn produce_message<'a, 'b>(
+    data: &'a [u8],
+    topic: &'b str,
+    brokers: Vec<String>,
+) -> Result<(), KafkaError> {
+    let mut producer = Producer::from_hosts(brokers)
+        .with_ack_timeout(Duration::from_secs(1))
+        .with_required_acks(RequiredAcks::One)
+        .create()?;
+
+    producer.send(&Record::from_value(topic, data))?;
+    Ok(())
+}