Add load tests and prevent partial message read (#20)

* Add streaming API for addRoutes

* Add list routes and list peers streaming APIs

* Implement basic load test

* Implement logger

* Fix load test

* Prevent partial message read

* Make load test realistic

* Support RibType for ListRoutes API

Author: sungwoncho

Cargo.lock

@@ -3,6 +3,7 @@ members = [
     "core",
     "daemon",
     "cli",
+    "loadtests",
 ]
 resolver = "2"
 

Cargo.toml

@@ -1,4 +1,4 @@
-.PHONY: all build clean run test fmt release setup
+.PHONY: all build clean run test fmt release setup loadtest
 
 all: build
 
@@ -19,7 +19,12 @@ run: setup
 
 test: setup
 	cargo clippy --all-targets --all-features -- -D warnings
-	cargo test
+	cargo test --workspace --exclude loadtests
 
 fmt:
 	cargo fmt
+
+loadtest: setup
+	@echo "Building bgpggd and running load tests..."
+	cargo build --bin bgpggd
+	cargo test -p loadtests --release -- --nocapture --test-threads=1

Makefile

@@ -102,9 +102,10 @@ pub async fn handle(addr: String, cmd: GlobalCommands) -> Result<(), Box<dyn std
         },
 
         GlobalCommands::Info => {
-            let (listen_addr, listen_port) = client.get_server_info().await?;
+            let (listen_addr, listen_port, num_routes) = client.get_server_info().await?;
             println!("BGP Server Information:");
             println!("  Listen Address: {}:{}", listen_addr, listen_port);
+            println!("  Routes in RIB: {}", num_routes);
         }
 
         GlobalCommands::Summary => {

cli/src/commands/global.rs

@@ -2,3 +2,4 @@ asn: 65000
 listen_addr: "127.0.0.1:1790"
 router_id: "1.1.1.1"
 grpc_listen_addr: "[::1]:50051"
+log_level: "info"

config.yaml

@@ -134,6 +134,7 @@ pub async fn read_bgp_message<R: AsyncReadExt + Unpin>(
     mut stream: R,
 ) -> Result<BgpMessage, ParserError> {
     let mut header_buffer = [0u8; BGP_HEADER_SIZE_BYTES];
+
     stream
         .read_exact(&mut header_buffer)
         .await

core/src/bgp/msg.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::warn;
-
 use super::msg::{Message, MessageType};
 use super::utils::ParserError;
 
@@ -171,28 +169,40 @@ impl BgpError {
             ErrorCode::MessageHeaderError => {
                 let err = MessageHeaderError::from(err_sub_code);
                 if matches!(err, MessageHeaderError::Unknown(_)) {
-                    warn!("received NOTIFICATION with unknown MessageHeaderError subcode", "subcode" => err_sub_code);
+                    eprintln!(
+                        "Warning: received NOTIFICATION with unknown MessageHeaderError subcode, subcode={:?}",
+                        err_sub_code
+                    );
                 }
                 BgpError::MessageHeaderError(err)
             }
             ErrorCode::OpenMessageError => {
                 let err = OpenMessageError::from(err_sub_code);
                 if matches!(err, OpenMessageError::Unknown(_)) {
-                    warn!("received NOTIFICATION with unknown OpenMessageError subcode", "subcode" => err_sub_code);
+                    eprintln!(
+                        "Warning: received NOTIFICATION with unknown OpenMessageError subcode, subcode={:?}",
+                        err_sub_code
+                    );
                 }
                 BgpError::OpenMessageError(err)
             }
             ErrorCode::UpdateMessageError => {
                 let err = UpdateMessageError::from(err_sub_code);
                 if matches!(err, UpdateMessageError::Unknown(_)) {
-                    warn!("received NOTIFICATION with unknown UpdateMessageError subcode", "subcode" => err_sub_code);
+                    eprintln!(
+                        "Warning: received NOTIFICATION with unknown UpdateMessageError subcode, subcode={:?}",
+                        err_sub_code
+                    );
                 }
                 BgpError::UpdateMessageError(err)
             }
             ErrorCode::HoldTimerExpired => BgpError::HoldTimerExpired,
             ErrorCode::FiniteStateMachineError => BgpError::FiniteStateMachineError,
             ErrorCode::Unknown => {
-                warn!("received NOTIFICATION with unknown error code", "code" => err_code);
+                eprintln!(
+                    "Warning: received NOTIFICATION with unknown error code, code={:?}",
+                    err_code
+                );
                 BgpError::Unknown
             }
         }
@@ -280,7 +290,10 @@ impl NotificationMessage {
         // RFC 4271: errors in NOTIFICATION messages cannot be reported back via NOTIFICATION,
         // so we log locally and return Unknown for malformed messages.
         if bytes.len() < 2 {
-            warn!("received malformed NOTIFICATION message", "len" => bytes.len());
+            eprintln!(
+                "Warning: received malformed NOTIFICATION message, len={:?}",
+                bytes.len()
+            );
             return NotificationMessage {
                 error: BgpError::Unknown,
                 data: bytes,

core/src/bgp/msg_notification.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::warn;
 use std::error::Error;
 use std::fmt::{Display, Formatter};
 use std::net::{Ipv4Addr, Ipv6Addr};
@@ -99,7 +98,10 @@ pub fn parse_nlri_list(bytes: &[u8]) -> Result<Vec<IpNetwork>, ParserError> {
 
         // Semantic check: skip multicast prefixes (224.0.0.0/4)
         if net.is_multicast() {
-            warn!("ignoring multicast NLRI prefix", "prefix" => format!("{:?}", net));
+            eprintln!(
+                "Warning: ignoring multicast NLRI prefix, prefix={:?}",
+                format!("{:?}", net)
+            );
             cursor += byte_len;
             continue;
         }

core/src/bgp/utils.rs

@@ -13,8 +13,10 @@
 // limitations under the License.
 
 use super::msg::BmpMessage;
+use crate::log::Logger;
 use crate::{error, info};
 use std::net::SocketAddr;
+use std::sync::Arc;
 use std::time::Duration;
 use tokio::io::AsyncWriteExt;
 use tokio::net::TcpStream;
@@ -24,14 +26,16 @@ pub struct BmpTcpClient {
     addr: SocketAddr,
     conn: Option<TcpStream>,
     reconnect_delay: Duration,
+    logger: Arc<Logger>,
 }
 
 impl BmpTcpClient {
-    pub fn new(addr: SocketAddr) -> Self {
+    pub fn new(addr: SocketAddr, logger: Arc<Logger>) -> Self {
         Self {
             addr,
             conn: None,
             reconnect_delay: Duration::from_secs(1),
+            logger,
         }
     }
 
@@ -46,13 +50,13 @@ impl BmpTcpClient {
 
         match TcpStream::connect(self.addr).await {
             Ok(stream) => {
-                info!("BMP: Connected to collector", "addr" => &self.addr);
+                info!(&self.logger, "BMP: Connected to collector", "addr" => &self.addr);
                 self.conn = Some(stream);
                 self.reconnect_delay = Duration::from_secs(1); // Reset delay
                 true
             }
             Err(e) => {
-                error!("BMP: Failed to connect to collector", "addr" => &self.addr, "error" => &e.to_string());
+                error!(&self.logger, "BMP: Failed to connect to collector", "addr" => &self.addr, "error" => &e.to_string());
                 // Exponential backoff: 1s, 2s, 4s, 8s, 16s, 30s (max)
                 self.reconnect_delay =
                     std::cmp::min(self.reconnect_delay * 2, Duration::from_secs(30));
@@ -70,7 +74,7 @@ impl BmpTcpClient {
             match conn.write_all(data).await {
                 Ok(_) => true,
                 Err(e) => {
-                    error!("BMP: Write failed", "addr" => &self.addr, "error" => &e.to_string());
+                    error!(&self.logger, "BMP: Write failed", "addr" => &self.addr, "error" => &e.to_string());
                     self.conn = None; // Trigger reconnect
                     false
                 }