BGP enhancements

+ add tcp idle callback function
+ extend bgp-sessions command
+ log update start/stop timestamp
+ decode raw update stream
+ ...

code/bngblaster/src/bbl_tcp.c

@@ -1,6 +1,6 @@
 /*
- * BNG Blaster (BBL) - LwIP
- *
+ * BNG Blaster (BBL) - TCP (lwIP)
+ * 
  * Christian Giese, February 2022
  *
  * Copyright (C) 2020-2021, RtBrick, Inc.
@@ -115,6 +115,9 @@ bbl_tcp_sent_cb(void *arg, struct tcp_pcb *tpcb, u16_t len) {
     } else if(tcpc->pcb->unacked == NULL && tcpc->pcb->unsent == NULL) {
         /* Idle means that it is save to replace buffer. */
         tcpc->state = BBL_TCP_STATE_IDLE;
+        if(tcpc->idle_cb) {
+            (tcpc->idle_cb)(tcpc->arg);
+        }
     }
     return ERR_OK;
 }

code/bngblaster/src/bbl_tcp.h

@@ -24,7 +24,7 @@ typedef enum bbl_tcp_state_ {
     BBL_TCP_STATE_SENDING,
 } bbl_tcp_state_t;
 
-typedef void (*bbl_tcp_connected_fn)(void *arg);
+typedef void (*bbl_tcp_callback_fn)(void *arg);
 typedef void (*bbl_tcp_receive_fn)(void *arg, uint8_t *buf, uint16_t len);
 typedef void (*bbl_tcp_error_fn)(void *arg, err_t err);
 typedef err_t (*bbl_tcp_poll_fn)(void *arg, struct tcp_pcb *tpcb);
@@ -45,7 +45,8 @@ typedef struct bbl_tcp_ctx_
     
     struct tcp_pcb *pcb;
 
-    bbl_tcp_connected_fn connected_cb; /* application connected callback */
+    bbl_tcp_callback_fn connected_cb; /* application connected callback */
+    bbl_tcp_callback_fn idle_cb; /* application idle callback */
 
     bbl_tcp_receive_fn receive_cb; /* application receive callback */
     bbl_tcp_error_fn error_cb; /* application error callback */

code/bngblaster/src/bgp/bgp.c

@@ -9,11 +9,16 @@
 #include "bgp.h"
 
 static bgp_raw_update_t *
-bgp_raw_update_load(char *file) {
-    bgp_raw_update_t *raw_update;
+bgp_raw_update_load(char *file, bool decode_file) {
+    bgp_raw_update_t *raw_update = NULL;
     long fsize;
     double fsize_kb;
 
+    uint8_t *buf = NULL;
+    uint32_t len = 0;
+    uint16_t msg_len;
+    uint8_t  msg_type;
+
     /* Open file */
     FILE *f = fopen(file, "rb");
     if (f == NULL) {
@@ -28,19 +33,64 @@ bgp_raw_update_load(char *file) {
     fseek(f, 0, SEEK_SET);
 
     /* Load file into memory */
-    LOG(INFO, "Load BGP RAW update file %s (%.2f KB)\n", file, fsize_kb);
     raw_update = malloc(sizeof(bgp_raw_update_t));
     raw_update->file = file;
     raw_update->buf = malloc(fsize);
     raw_update->len = fsize;
     if(fread(raw_update->buf, fsize, 1, f) != 1) {
+        fclose(f);
         LOG(ERROR, "Failed to read BGP RAW update file %s\n", file);
-        free(raw_update->buf);
-        free(raw_update);
-        return NULL;
+        goto ERROR;
     }
     fclose(f);
+
+    if(decode_file) {
+        /* Decode update stream */
+        buf = raw_update->buf;
+        len = raw_update->len;
+        while(len) {
+            if(len < BGP_MIN_MESSAGE_SIZE) {
+                goto DECODE_ERROR;
+            }
+            if(*(uint64_t*)buf != UINT64_MAX) {
+                goto DECODE_ERROR;
+            }
+            BUMP_BUFFER(buf, len, sizeof(uint64_t));
+            if(*(uint64_t*)buf != UINT64_MAX) {
+                goto DECODE_ERROR;
+            }
+            BUMP_BUFFER(buf, len, sizeof(uint64_t));
+            msg_len = be16toh(*(uint16_t*)buf);
+            BUMP_BUFFER(buf, len, sizeof(uint16_t));
+            msg_type = *buf;
+            BUMP_BUFFER(buf, len, sizeof(uint8_t));
+            if(msg_len < BGP_MIN_MESSAGE_SIZE ||
+               msg_len > BGP_MAX_MESSAGE_SIZE) {
+                goto DECODE_ERROR;
+            }
+            if((msg_len - BGP_MIN_MESSAGE_SIZE) > len) {
+                goto DECODE_ERROR;
+            }
+            if(msg_type == BGP_MSG_UPDATE) {
+                raw_update->updates++;
+            }
+            BUMP_BUFFER(buf, len, (msg_len - BGP_MIN_MESSAGE_SIZE));
+        }
+    }
+    LOG(INFO, "Loaded BGP RAW update file %s (%.2f KB, %u updates)\n", 
+        file, fsize_kb, raw_update->updates);
     return raw_update;
+
+DECODE_ERROR:
+    LOG(ERROR, "Failed to decode BGP RAW update file %s\n", file);
+ERROR:
+    if(raw_update) {
+        if(raw_update->buf) {
+            free(raw_update->buf);
+        }
+        free(raw_update);
+    }
+    return NULL;
 }
 
 /**
@@ -106,9 +156,13 @@ bgp_init(bbl_ctx_s *ctx) {
                 raw_update = raw_update->next;
             }
             if(!session->raw_update) {
-                session->raw_update = bgp_raw_update_load(config->raw_update_file);
-                session->raw_update->next = raw_update_root;
-                raw_update_root = session->raw_update;
+                session->raw_update = bgp_raw_update_load(config->raw_update_file, true);
+                if(session->raw_update) {
+                    session->raw_update->next = raw_update_root;
+                    raw_update_root = session->raw_update;
+                } else {
+                    return false;
+                }
             }
         }
 

code/bngblaster/src/bgp/bgp_ctrl.c

@@ -11,12 +11,25 @@
 json_t *
 bgp_ctrl_session(bgp_session_t *session) {
     json_t *root = NULL;
+    json_t *stats = NULL;
 
     if(!session) {
         return NULL;
     }
 
-    root = json_pack("{ss ss ss si si ss ss si si ss }",
+    stats = json_pack("{si si si si si si}",
+                      "messages-rx", session->stats.message_rx,
+                      "messages-tx", session->stats.message_tx,
+                      "keepalive-rx", session->stats.keepalive_rx,
+                      "keepalive-tx", session->stats.keepalive_tx,
+                      "update-rx", session->stats.update_rx,
+                      "update-tx", session->stats.update_tx);
+
+    if(!stats) {
+        return NULL;
+    }
+
+    root = json_pack("{ss ss ss si si ss ss si si ss so}",
                      "interface", session->interface->name,
                      "local-address", format_ipv4_address(&session->ipv4_local_address),
                      "local-id", format_ipv4_address(&session->config->id),
@@ -26,7 +39,11 @@ bgp_ctrl_session(bgp_session_t *session) {
                      "peer-id", format_ipv4_address(&session->peer.id),
                      "peer-as", session->peer.as,
                      "peer-holdtime", session->peer.holdtime,
-                     "state", bgp_session_state_string(session->state));
+                     "state", bgp_session_state_string(session->state),
+                     "stats", stats);
 
+    if(!root) {
+        if(stats) json_decref(stats);
+    }
     return root;
 }

code/bngblaster/src/bgp/bgp_def.h

@@ -12,8 +12,8 @@
 /* DEFINITIONS ... */
 
 #define BGP_PORT                    179
-#define BGP_MIN_MESSAGE_SIZE        19
-#define BGP_MAX_MESSAGE_SIZE        4096
+#define BGP_MIN_MESSAGE_SIZE        19U
+#define BGP_MAX_MESSAGE_SIZE        4096U
 #define BGP_BUF_SIZE                256*1024
 #define BGP_DEFAULT_AS              65000
 #define BGP_DEFAULT_HOLDTIME        90
@@ -43,6 +43,7 @@ typedef struct bgp_raw_update_ {
 
     uint8_t *buf;
     uint32_t len;
+    uint32_t updates;
 
     /* Pointer to next instance */
     struct bgp_raw_update_ *next;
@@ -97,7 +98,6 @@ typedef struct bgp_session_ {
     io_buffer_t write_buf;
 
     bgp_state_t state;
-    uint16_t keepalive_countdown;
 
     struct {
         uint32_t as;
@@ -106,13 +106,21 @@ typedef struct bgp_session_ {
     } peer;
 
     struct {
+        uint32_t message_rx;
+        uint32_t message_tx;
         uint32_t keepalive_rx;
+        uint32_t keepalive_tx;
         uint32_t update_rx;
+        uint32_t update_tx;
     } stats;
 
     bgp_raw_update_t *raw_update;
     bool raw_update_send;
 
+    struct timespec established_timestamp;
+    struct timespec update_start_timestamp;
+    struct timespec update_stop_timestamp;
+
     bool teardown;
     uint8_t error_code;
     uint8_t error_subcode;

code/bngblaster/src/bgp/bgp_receive.c

@@ -69,10 +69,25 @@ struct keyval_ bgp_notification_cease_error_values[] = {
     { 0, NULL}
 };
 
+static void 
+bgp_decode_error(bgp_session_t *session) {
+    LOG(BGP, "BGP (%s %s - %s) invalid message received\n",
+        session->interface->name,
+        format_ipv4_address(&session->ipv4_local_address),
+        format_ipv4_address(&session->ipv4_peer_address));
+
+    if(!session->error_code) {
+        session->error_code = 1; /* Message header error */
+        session->error_subcode = 2; /* Bad message length */
+    }
+    bgp_session_close(session);
+}
+
 static void
 bgp_open(bgp_session_t *session, uint8_t *start, uint16_t length) {
     if(length < 28) {
-        goto ERROR;
+        bgp_decode_error(session);
+        return;
     }
     session->peer.as = read_be_uint(start+20, 2);
     session->peer.holdtime = read_be_uint(start+22, 2);
@@ -86,18 +101,6 @@ bgp_open(bgp_session_t *session, uint8_t *start, uint16_t length) {
 
     bgp_session_state_change(session, BGP_OPENCONFIRM);
     return;
-
-ERROR:
-    LOG(BGP, "BGP (%s %s - %s) invalid open message received\n",
-        session->interface->name,
-        format_ipv4_address(&session->ipv4_local_address),
-        format_ipv4_address(&session->ipv4_peer_address));
-
-    if(!session->error_code) {
-        session->error_code = 1; /* Message header error */
-        session->error_subcode = 2; /* Bad message length */
-    }
-    bgp_session_close(session);
 }
 
 static void
@@ -105,7 +108,8 @@ bgp_notification(bgp_session_t *session, uint8_t *start, uint16_t length) {
     uint8_t error_code, error_subcode;
 
     if(length < 21) {
-        goto ERROR;
+        bgp_decode_error(session);
+        return;
     }
 
     error_code = *(start+19);
@@ -154,18 +158,6 @@ bgp_notification(bgp_session_t *session, uint8_t *start, uint16_t length) {
     session->error_code = 0;
     bgp_session_close(session);
     return;
-
-ERROR:
-    LOG(BGP, "BGP (%s %s - %s) invalid notification message received\n",
-        session->interface->name,
-        format_ipv4_address(&session->ipv4_local_address),
-        format_ipv4_address(&session->ipv4_peer_address));
-
-    if(!session->error_code) {
-        session->error_code = 1; /* Message header error */
-        session->error_subcode = 2; /* Bad message length */
-    }
-    bgp_session_close(session);
 }
 
 /*
@@ -202,17 +194,25 @@ bpg_read(bgp_session_t *session) {
         size = buffer->idx - buffer->start_idx;
 
         /* Minimum message size */
-        if (size < 19) {
+        if (size < BGP_MIN_MESSAGE_SIZE) {
+            break;
+        }
+
+        length = read_be_uint(start+16, 2);
+        if (length < BGP_MIN_MESSAGE_SIZE ||
+            length > BGP_MAX_MESSAGE_SIZE) {
+            bgp_decode_error(session);
             break;
         }
 
         /* Full message on the wire to consume? */
-        length = read_be_uint(start+16, 2);
-        type = *(start+18);
         if (length > size) {
             break;
         }
 
+        type = *(start+18);
+
+        session->stats.message_rx++;
         LOG(DEBUG, "BGP (%s %s - %s) read %s message\n",
             session->interface->name,
             format_ipv4_address(&session->ipv4_local_address),

code/bngblaster/src/bgp/bgp_session.c

@@ -75,11 +75,36 @@ bgp_session_keepalive_job(timer_s *timer) {
         if(session->tcpc && session->tcpc->state == BBL_TCP_STATE_IDLE) {
             bgp_session_reset_write_buffer(session);
             bgp_push_keepalive_message(session);
-            bgp_session_send(session);
+            if(bgp_session_send(session)) {
+                session->stats.message_tx++;
+                session->stats.keepalive_tx++;
+            }
         }
     }
 }
 
+void 
+bgp_raw_update_stop_cb(void *arg) {
+    bgp_session_t *session = (bgp_session_t*)arg;
+    struct timespec time_diff;
+
+    session->tcpc->idle_cb = NULL;
+
+    clock_gettime(CLOCK_MONOTONIC, &session->update_stop_timestamp);
+    timespec_sub(&time_diff, 
+                 &session->update_stop_timestamp, 
+                 &session->update_start_timestamp);
+    
+    session->stats.message_tx += session->raw_update->updates;
+    session->stats.update_tx += session->raw_update->updates;
+
+    LOG(BGP, "BGP (%s %s - %s) raw update stop after %lds\n",
+        session->interface->name,
+        format_ipv4_address(&session->ipv4_local_address),
+        format_ipv4_address(&session->ipv4_peer_address),
+        time_diff.tv_sec);
+}
+
 void
 bgp_session_update_job(timer_s *timer) {
     bgp_session_t *session = timer->data;
@@ -89,6 +114,15 @@ bgp_session_update_job(timer_s *timer) {
         if(session->raw_update && !session->raw_update_send) {
             if(bbl_tcp_send(session->tcpc, session->raw_update->buf, session->raw_update->len)) {
                 session->raw_update_send = true;
+
+                LOG(BGP, "BGP (%s %s - %s) raw update start\n",
+                    session->interface->name,
+                    format_ipv4_address(&session->ipv4_local_address),
+                    format_ipv4_address(&session->ipv4_peer_address));
+
+                clock_gettime(CLOCK_MONOTONIC, &session->update_start_timestamp);
+                session->tcpc->idle_cb = bgp_raw_update_stop_cb;
+
             } else {
                 goto RETRY;
             }
@@ -110,6 +144,7 @@ bgp_session_state_opensent(bgp_session_t *session) {
     bgp_session_reset_write_buffer(session);
     bgp_push_open_message(session);
     bgp_session_send(session);
+    session->stats.message_tx++;
 }
 
 static void
@@ -117,6 +152,8 @@ bgp_session_state_openconfirm(bgp_session_t *session) {
     bgp_session_reset_write_buffer(session);
     bgp_push_keepalive_message(session);
     bgp_session_send(session);
+    session->stats.message_tx++;
+    session->stats.keepalive_tx++;
 }
 
 static void
@@ -124,6 +161,8 @@ bgp_session_state_estbalished(bgp_session_t *session) {
     bbl_ctx_s *ctx = session->interface->ctx;
     time_t keepalive_interval;
 
+    clock_gettime(CLOCK_MONOTONIC, &session->established_timestamp);
+
     /* Start BGP keepalive */
     keepalive_interval = session->peer.holdtime/2U;
     if(!keepalive_interval) {
@@ -256,16 +295,33 @@ bgp_session_connect(bgp_session_t *session, time_t delay) {
     if(session->state == BGP_CLOSED) {
         bbl_tcp_ctx_free(session->tcpc);
         session->tcpc = NULL;
+
         bgp_session_reset_read_buffer(session);
         bgp_session_reset_write_buffer(session);
+
         session->peer.as = 0;
         session->peer.id = 0;
         session->peer.holdtime = 0;
+
+        session->stats.message_rx = 0;
+        session->stats.message_tx = 0;
         session->stats.keepalive_rx = 0;
+        session->stats.keepalive_tx = 0;
         session->stats.update_rx = 0;
+        session->stats.update_tx = 0;
+        
         session->raw_update_send = false;
+
+        session->established_timestamp.tv_sec = 0;
+        session->established_timestamp.tv_nsec = 0;
+        session->update_start_timestamp.tv_sec = 0;
+        session->update_start_timestamp.tv_nsec = 0;
+        session->update_stop_timestamp.tv_sec = 0;
+        session->update_stop_timestamp.tv_nsec = 0;
+
         session->error_code = 0;
         session->error_subcode = 0;
+
         bgp_session_state_change(session, BGP_IDLE);
 
         timer_add(&ctx->timer_root, &session->connect_timer, 
@@ -317,6 +373,7 @@ bgp_session_close(bgp_session_t *session) {
         bgp_session_reset_write_buffer(session);
         bgp_push_notification_message(session);
         bgp_session_send(session);
+        session->stats.message_tx++;
         bgp_session_state_change(session, BGP_CLOSING);
         delay = 3;
     }