implement basic logging

Cargo.toml

@@ -12,9 +12,11 @@ bytes = "1.6.0"
 chrono = "0.4.38"
 clap = { version = "4.5.20", features = ["derive"] }
 config = "0.14.1"
+env_logger = "0.11.5"
 hex = "0.4.3"
 ipnet = "2.9.0"
 kafka = "0.10.0"
+log = "0.4.22"
 prefix-trie = "0.3.0"
 serde = { version = "1.0.213", features = ["derive"] }
 tokio = { version = "1.37.0", features = ["full"] }

Dockerfile

@@ -9,7 +9,6 @@ RUN cargo build --release
 FROM rust:latest
 
 COPY --from=builder /app/target/release/risotto /app/risotto
-COPY example/risotto.yml /config/risotto.yml
 
 EXPOSE 3000
 EXPOSE 4000

src/bmp.rs

@@ -4,6 +4,7 @@ use bgpkit_parser::parser::bmp::messages::{BmpMessage, BmpMessageBody};
 use bytes::Bytes;
 use chrono::Utc;
 use config::Config;
+use log::debug;
 use std::io;
 use tokio::io::AsyncReadExt;
 use tokio::net::TcpStream;
@@ -75,11 +76,13 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
         .or_insert_with(|| new_router(router_ip, router_port));
 
     match message.message_body {
-        BmpMessageBody::PeerUpNotification(_) => {
+        BmpMessageBody::PeerUpNotification(body) => {
+            debug!("PEER_UP_NOTIFICATION - {:?}", body);
             // Simply add the peer if we did not see it before
             router.add_peer(&peer);
         }
         BmpMessageBody::RouteMonitoring(body) => {
+            debug!("ROUTE_MONITORING - {:?}", body);
             let potential_updates = decode_updates(body).unwrap_or(Vec::new());
 
             let mut legitimate_updates = Vec::new();
@@ -93,11 +96,12 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
             // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in legitimate_updates {
                 let update = format_update(&router, &peer, &update);
-                println!("{:?}", update);
+                debug!("UPDATE - {:?}", update);
                 send_to_kafka(&kafka_host, &kafka_topic, update.as_bytes());
             }
         }
-        BmpMessageBody::PeerDownNotification(_) => {
+        BmpMessageBody::PeerDownNotification(body) => {
+            debug!("PEER_DOWN_NOTIFICATION - {:?}", body);
             // Remove the peer and the associated prefixes
             // To do so, we start by emiting synthetic withdraw updates
             let mut synthetic_updates = Vec::new();
@@ -116,7 +120,7 @@ pub async fn handle(socket: &mut TcpStream, db: DB, settings: Config) {
             // TODO: Handle multiple event pipelines (stdout, CSV file, Kafka, ...)
             for update in synthetic_updates {
                 let update = format_update(&router, &peer, &update);
-                println!("{:?}", update);
+                debug!("UPDATE - {:?}", update);
                 send_to_kafka(&kafka_host, &kafka_topic, update.as_bytes());
             }
         }

src/main.rs

@@ -6,18 +6,55 @@ mod router;
 mod settings;
 mod update;
 
+use chrono::Local;
 use clap::Parser;
 use config::Config;
+use env_logger::Builder;
+use log::debug;
+use log::LevelFilter;
+use std::io::Write;
 use tokio::net::TcpListener;
 
 use crate::db::DB;
 
+#[derive(Parser, Debug)]
+#[command(version, about, long_about = None)]
+struct CLI {
+    #[arg(short, long)]
+    config: String,
+
+    #[arg(short, long)]
+    debug: bool,
+}
+
+fn set_logging(cli: &CLI) {
+    Builder::new()
+        .format(|buf, record| {
+            writeln!(
+                buf,
+                "{} [{}] - {}",
+                Local::now().format("%Y-%m-%dT%H:%M:%S"),
+                record.level(),
+                record.args()
+            )
+        })
+        .filter(
+            None,
+            if cli.debug {
+                LevelFilter::Debug
+            } else {
+                LevelFilter::Info
+            },
+        )
+        .init();
+}
+
 async fn api_handler(settings: Config, db: DB) {
     let address = settings.get_string("api.address").unwrap();
     let port = settings.get_int("api.port").unwrap();
     let host = settings::host(address, port, false);
 
-    println!("Binding API listener to {}", host);
+    debug!("Binding API listener to {}", host);
 
     let api_listener = TcpListener::bind(host).await.unwrap();
     let app = api::app(db.clone());
@@ -30,7 +67,7 @@ async fn bmp_handler(settings: Config, db: DB) {
     let port = settings.get_int("bmp.port").unwrap();
     let host = settings::host(address, port, false);
 
-    println!("Binding BMP listener to {}", host);
+    debug!("Binding BMP listener to {}", host);
 
     let bmp_listener = TcpListener::bind(host).await.unwrap();
     loop {
@@ -45,13 +82,6 @@ async fn bmp_handler(settings: Config, db: DB) {
     }
 }
 
-#[derive(Parser, Debug)]
-#[command(version, about, long_about = None)]
-struct CLI {
-    #[arg(short, long)]
-    config: String,
-}
-
 #[tokio::main]
 async fn main() {
     let cli = CLI::parse();
@@ -64,6 +94,8 @@ async fn main() {
 
     let db = db::new_db().await;
 
+    set_logging(&cli);
+
     let api_handler = tokio::spawn(api_handler(settings.clone(), db.clone()));
     let bmp_handler = tokio::spawn(bmp_handler(settings.clone(), db.clone()));
 

src/pipeline.rs

@@ -2,10 +2,11 @@ use std::time::Duration;
 
 use kafka::error::Error as KafkaError;
 use kafka::producer::{Producer, Record, RequiredAcks};
+use log::error;
 
 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);
+        error!("Failed producing messages: {}", e);
     }
 }
 

testbed/compose.yml

@@ -7,7 +7,7 @@ services:
       - NET_BROADCAST
       - NET_RAW
     volumes:
-      - ./config/bird_10.conf:/etc/bird/bird.conf:ro
+      - ./config/bird/bird_10.conf:/etc/bird/bird.conf:ro
     networks:
       testbed:
         ipv4_address: 10.0.0.10
@@ -19,12 +19,15 @@ services:
       - NET_BROADCAST
       - NET_RAW
     volumes:
-      - ./config/bird_20.conf:/etc/bird/bird.conf:ro
+      - ./config/bird/bird_20.conf:/etc/bird/bird.conf:ro
     networks:
       testbed:
         ipv4_address: 10.0.0.20
   risotto:
     build: ..
+    command: --debug --config /config/risotto
+    volumes:
+      - ./config/risotto/risotto.yml:/config/risotto.yml
     networks:
       testbed:
         ipv4_address: 10.0.0.100

example/risotto.yml → testbed/config/risotto/risotto.yml

@@ -1,12 +1,12 @@
 api:
-  address: '::'
+  address: 0.0.0.0
   port: 3000
 
 bmp:
-  address: '::'
+  address: 0.0.0.0
   port: 4000
 
 kafka:
   host: kafka.example.com
-  port: 9000
+  port: 9092
   topic: bgp-updates