BMP: Simplify route monitoring hooks

No need for *_begin(), *_commit(), and *_end() hooks. The hook *_notify()
is sufficient for everything.

proto/bgp/bgp.h

@@ -635,8 +635,7 @@ int bgp_get_attr(const struct eattr *e, byte *buf, int buflen);
 void bgp_get_route_info(struct rte *, byte *buf);
 int bgp_total_aigp_metric_(rte *e, u64 *metric, const struct adata **ad);
 
-struct bmp_proto;
+byte * bgp_bmp_encode_rte(struct bgp_channel *c, byte *buf, const net_addr *n, const struct rte *new, const struct rte_src *src);
-void bgp_bmp_encode_rte(struct bgp_channel *c, struct bmp_proto *bmp, const net_addr *n, const struct rte *new, const struct rte_src *src);
 
 #define BGP_AIGP_METRIC		1
 #define BGP_AIGP_MAX		U64(0xffffffffffffffff)

proto/bgp/packets.c

@@ -2472,13 +2472,11 @@ bgp_bmp_prepare_bgp_hdr(byte *buf, const u16 msg_size, const u8 msg_type)
   return buf + BGP_MSG_HDR_TYPE_POS + BGP_MSG_HDR_TYPE_SIZE;
 }
 
-void
+byte *
-bgp_bmp_encode_rte(struct bgp_channel *c, struct bmp_proto *bmp, const net_addr *n,
+bgp_bmp_encode_rte(struct bgp_channel *c, byte *buf, const net_addr *n,
 		   const struct rte *new, const struct rte_src *src)
 {
 //  struct bgp_proto *p = (void *) c->c.proto;
-
-  byte buf[BGP_MAX_EXT_MSG_LENGTH];
   byte *pkt = buf + BGP_HEADER_LENGTH;
 
   ea_list *attrs = new ? new->attrs->eattrs : NULL;
@@ -2502,11 +2500,11 @@ bgp_bmp_encode_rte(struct bgp_channel *c, struct bmp_proto *bmp, const net_addr
   add_tail(&b->prefixes, &px->buck_node);
 
   byte *end = bgp_create_update_bmp(c, pkt, b, !!new);
-  if (!end)
-    return;
 
-  bgp_bmp_prepare_bgp_hdr(buf, end - buf, PKT_UPDATE);
+  if (end)
-  bmp_route_monitor_put_update_in_pre_msg(bmp, buf, end - buf);
+    bgp_bmp_prepare_bgp_hdr(buf, end - buf, PKT_UPDATE);
+
+  return end;
 }
 
 #endif /* CONFIG_BMP */
@@ -2769,8 +2767,6 @@ bgp_rx_update(struct bgp_conn *conn, byte *pkt, uint len)
   s.ip_reach_len = len - pos;
   s.ip_reach_nlri = pkt + pos;
 
-  bmp_route_monitor_update_in_pre_begin();
-
   if (s.attr_len)
     ea = bgp_decode_attrs(&s, s.attrs, s.attr_len);
   else
@@ -2801,9 +2797,6 @@ bgp_rx_update(struct bgp_conn *conn, byte *pkt, uint len)
     bgp_decode_nlri(&s, s.mp_reach_af, s.mp_reach_nlri, s.mp_reach_len,
 		    ea, s.mp_next_hop_data, s.mp_next_hop_len);
 
-  bmp_route_monitor_update_in_pre_commit(p);
-  bmp_route_monitor_update_in_pre_end();
-
 done:
   rta_free(s.cached_rta);
   lp_restore(tmp_linpool, &tmpp);

proto/bmp/bmp.c

@@ -211,6 +211,12 @@ struct bmp_data_node {
   node n;
   byte *data;
   size_t data_size;
+
+  u32 remote_as;
+  u32 remote_id;
+  ip_addr remote_ip;
+  btime timestamp;
+  bool global_peer;
 };
 
 static void
@@ -401,9 +407,12 @@ static void
 bmp_route_monitor_msg_serialize(buffer *stream, const bool is_peer_global,
   const bool table_in_pre_policy, const u32 peer_as, const u32 peer_bgp_id,
   const bool as4_support, const ip_addr remote_addr, const byte *update_msg,
-  const size_t update_msg_size, u32 ts_sec, u32 ts_usec)
+  const size_t update_msg_size, btime timestamp)
 {
   const size_t data_size = BMP_PER_PEER_HDR_SIZE + update_msg_size;
+  u32 ts_sec = timestamp TO_S;
+  u32 ts_usec = timestamp - (ts_sec S);
+
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
   bmp_common_hdr_serialize(stream, BMP_ROUTE_MONITOR, data_size);
   bmp_per_peer_hdr_serialize(stream, is_peer_global, table_in_pre_policy,
@@ -581,56 +590,43 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
   bmp_buffer_free(&payload);
 }
 
-
 static void
-bmp_route_monitor_update_in_pre_begin_(struct bmp_proto *p)
+bmp_route_monitor_put_update(struct bmp_proto *p, const byte *data, size_t length, struct bgp_proto *bgp)
 {
-  if (!p->started)
-    return;
-
-  if (p->monitoring_rib.in_pre_policy == false)
-    return;
-
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    !EMPTY_LIST(p->rt_table_in_pre_policy.update_msg_queue),
-    "Previous BMP route monitoring update not finished yet"
-  );
-
-  gettimeofday(&p->rt_table_in_pre_policy.update_begin_time,NULL);
-  init_list(&p->rt_table_in_pre_policy.update_msg_queue);
-  p->rt_table_in_pre_policy.update_msg_size = 0;
-  p->rt_table_in_pre_policy.update_in_progress = true;
-}
-
-void
-bmp_route_monitor_update_in_pre_begin(void)
-{
-  struct bmp_proto *p; node *n;
-  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-    bmp_route_monitor_update_in_pre_begin_(p);
-}
-
-void
-bmp_route_monitor_put_update_in_pre_msg(struct bmp_proto *p, const byte *data, const size_t data_size)
-{
-  if (!p->started)
-    return;
-
-  if (p->monitoring_rib.in_pre_policy == false)
-    return;
-
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    !p->rt_table_in_pre_policy.update_in_progress,
-    "BMP route monitoring update not started yet"
-  );
-
   struct bmp_data_node *upd_msg = mb_alloc(p->update_msg_mem_pool,
                                sizeof (struct bmp_data_node));
-  upd_msg->data = mb_alloc(p->update_msg_mem_pool, data_size);
+  upd_msg->data = mb_alloc(p->update_msg_mem_pool, length);
-  memcpy(upd_msg->data, data, data_size);
+  memcpy(upd_msg->data, data, length);
-  upd_msg->data_size = data_size;
+  upd_msg->data_size = length;
-  p->rt_table_in_pre_policy.update_msg_size += data_size;
+  add_tail(&p->update_msg_queue, &upd_msg->n);
-  add_tail(&p->rt_table_in_pre_policy.update_msg_queue, &upd_msg->n);
+
+  /* Save some metadata */
+  upd_msg->remote_as = bgp->remote_as;
+  upd_msg->remote_id = bgp->remote_id;
+  upd_msg->remote_ip = bgp->remote_ip;
+  upd_msg->timestamp = current_time();
+  upd_msg->global_peer = bmp_is_peer_global_instance(bgp);
+
+  /* Kick the commit */
+  if (!ev_active(p->update_ev))
+    ev_schedule(p->update_ev);
+}
+
+static void
+bmp_route_monitor_update_in_notify_(struct bmp_proto *p, struct bgp_channel *c,
+					const net_addr *n, const struct rte *new, const struct rte_src *src)
+{
+  struct bgp_proto *bgp = (void *) c->c.proto;
+
+  if (!p->started)
+    return;
+
+  if (p->monitoring_rib.in_pre_policy == false)
+    return;
+
+  byte buf[BGP_MAX_EXT_MSG_LENGTH];
+  byte *end = bgp_bmp_encode_rte(c, buf, n, new, src);
+  bmp_route_monitor_put_update(p, buf, end - buf, bgp);
 }
 
 void
@@ -641,97 +637,50 @@ bmp_route_monitor_update_in_notify(struct channel *C, const net_addr *n,
 
   struct bmp_proto *p; node *nx;
   WALK_LIST2(p, nx, bmp_proto_list, bmp_node)
-    bgp_bmp_encode_rte(c, p, n, new, src);
+    bmp_route_monitor_update_in_notify_(p, c, n, new, src);
 }
 
 static void
-bmp_route_monitor_update_in_pre_commit_(struct bmp_proto *p, const struct bgp_proto *bgp)
+bmp_route_monitor_commit(void *p_)
 {
+  struct bmp_proto *p = p_;
+
   if (!p->started)
     return;
 
   if (p->monitoring_rib.in_pre_policy == false)
     return;
 
-  const struct birdsock *sk = bmp_get_birdsock(bgp);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    sk,
-    "Failed to get bird socket from BGP proto"
-  );
-
-  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps(bgp);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    remote_caps,
-    "Failed to get remote capabilities from BGP proto"
-  );
-
-  bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
   buffer payload
-    = bmp_buffer_alloc(p->buffer_mpool,
+    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
-        p->rt_table_in_pre_policy.update_msg_size + DEFAULT_MEM_BLOCK_SIZE);
 
   buffer update_msgs
-    = bmp_buffer_alloc(p->buffer_mpool,
+    = bmp_buffer_alloc(p->buffer_mpool, BGP_MAX_EXT_MSG_LENGTH);
-        p->rt_table_in_pre_policy.update_msg_size);
 
-  struct bmp_data_node *data;
+  struct bmp_data_node *data, *data_next;
-  WALK_LIST(data, p->rt_table_in_pre_policy.update_msg_queue)
+  WALK_LIST_DELSAFE(data, data_next, p->update_msg_queue)
   {
     bmp_put_data(&update_msgs, data->data, data->data_size);
     bmp_route_monitor_msg_serialize(&payload,
-      is_global_instance_peer, true /* TODO: Hardcoded pre-policy Adj-Rib-In */,
+      data->global_peer, true /* TODO: Hardcoded pre-policy Adj-Rib-In */,
-      bgp->conn->received_as, bgp->remote_id, remote_caps->as4_support,
+      data->remote_as, data->remote_id, true,
-      sk->daddr, bmp_buffer_data(&update_msgs), bmp_buffer_pos(&update_msgs),
+      data->remote_ip, bmp_buffer_data(&update_msgs), bmp_buffer_pos(&update_msgs),
-      p->rt_table_in_pre_policy.update_begin_time.tv_sec,
+      data->timestamp);
-      p->rt_table_in_pre_policy.update_begin_time.tv_usec);
 
     bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
 
     bmp_buffer_flush(&payload);
     bmp_buffer_flush(&update_msgs);
+
+    mb_free(data->data);
+    rem_node(&data->n);
+    mb_free(data);
   }
 
   bmp_buffer_free(&payload);
   bmp_buffer_free(&update_msgs);
 }
 
-void
-bmp_route_monitor_update_in_pre_commit(const struct bgp_proto *bgp)
-{
-  struct bmp_proto *p; node *n;
-  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-    bmp_route_monitor_update_in_pre_commit_(p, bgp);
-}
-
-static void
-bmp_route_monitor_update_in_pre_end_(struct bmp_proto *p)
-{
-  if (!p->started)
-    return;
-
-  if (p->monitoring_rib.in_pre_policy == false)
-    return;
-
-  struct bmp_data_node *upd_msg;
-  struct bmp_data_node *upd_msg_next;
-  WALK_LIST_DELSAFE(upd_msg, upd_msg_next, p->rt_table_in_pre_policy.update_msg_queue)
-  {
-    mb_free(upd_msg->data);
-    rem_node((node *) upd_msg);
-    mb_free(upd_msg);
-  }
-
-  p->rt_table_in_pre_policy.update_in_progress = false;
-}
-
-void
-bmp_route_monitor_update_in_pre_end(void)
-{
-  struct bmp_proto *p; node *n;
-  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
-    bmp_route_monitor_update_in_pre_end_(p);
-}
-
 static void
 bmp_route_monitor_end_of_rib_msg(struct bmp_proto *p, struct bgp_channel *c)
 {
@@ -746,10 +695,7 @@ bmp_route_monitor_end_of_rib_msg(struct bmp_proto *p, struct bgp_channel *c)
   put_u16(rx_end_payload + BGP_MSG_HDR_LENGTH_POS, pos - rx_end_payload);
   put_u8(rx_end_payload + BGP_MSG_HDR_TYPE_POS, PKT_UPDATE);
 
-  bmp_route_monitor_update_in_pre_begin_(p);
+  bmp_route_monitor_put_update(p, rx_end_payload, pos - rx_end_payload, bgp);
-  bmp_route_monitor_put_update_in_pre_msg(p, rx_end_payload, pos - rx_end_payload);
-  bmp_route_monitor_update_in_pre_commit_(p, bgp);
-  bmp_route_monitor_update_in_pre_end_(p);
 }
 
 static void
@@ -773,14 +719,10 @@ bmp_route_monitor_pre_policy_table_in_snapshot(struct bmp_proto *p, struct bgp_c
     if (P->proto->class != PROTOCOL_BGP)
       continue;
 
-    bmp_route_monitor_update_in_pre_begin_(p);
-
     rte *e;
     for (e = n->routes; e; e = e->next)
-      bgp_bmp_encode_rte(c, p, n->n.addr, e, e->src);
+      bmp_route_monitor_update_in_notify_(p, c, n->n.addr, e, e->src);
 
-    bmp_route_monitor_update_in_pre_commit_(p, (struct bgp_proto *) P);
-    bmp_route_monitor_update_in_pre_end_(p);
     ++cnt;
   }
   FIB_ITERATE_END;
@@ -1071,14 +1013,15 @@ bmp_start(struct proto *P)
   p->map_mem_pool = rp_new(P->pool, "BMP Map");
   p->tx_mem_pool = rp_new(P->pool, "BMP Tx");
   p->update_msg_mem_pool = rp_new(P->pool, "BMP Update");
-  p->tx_ev = ev_new_init(p->tx_mem_pool, bmp_fire_tx, p);
+  p->tx_ev = ev_new_init(p->p.pool, bmp_fire_tx, p);
+  p->update_ev = ev_new_init(p->p.pool, bmp_route_monitor_commit, p);
   p->connect_retry_timer = tm_new_init(p->p.pool, bmp_connection_retry, p, 0, 0);
   p->sk = NULL;
 
   // bmp_peer_map_init(&p->bgp_peers, p->map_mem_pool);
 
   init_list(&p->tx_queue);
-  init_list(&p->rt_table_in_pre_policy.update_msg_queue);
+  init_list(&p->update_msg_queue);
   p->started = false;
   p->sock_err = 0;
   add_tail(&bmp_proto_list, &p->bmp_node);

proto/bmp/bmp.h

@@ -45,21 +45,13 @@ struct bmp_config {
 struct bgp_proto;
 struct bmp_proto;
 
-// Keeps necessary information during composing BGP UPDATE MSG which is going
-// to be sent to the BMP collector
-struct rt_table_info {
-  list update_msg_queue;         // Stores all composed BGP UPDATE MSGs
-  size_t update_msg_size;        // Size of all BGP UPDATE MSGs
-  struct timeval update_begin_time; // Keeps timestamp of starting BGP UPDATE MSGs composing
-  bool update_in_progress;       // Holds information whether composing process is still in progress
-};
-
 struct bmp_proto {
   struct proto p;                  // Parent proto
   const struct bmp_config *cf;     // Shortcut to BMP configuration
   node bmp_node;                   // Node in bmp_proto_list
   sock *sk;                        // TCP connection
   event *tx_ev;                    // TX event
+  event *update_ev;                // Update event
   char sys_descr[MIB_II_STR_LEN];  // sysDescr MIB-II [RFC1213] object
   char sys_name[MIB_II_STR_LEN];   // sysName MIB-II [RFC1213] object
   ip_addr local_addr;              // Source local IP address
@@ -74,7 +66,7 @@ struct bmp_proto {
   pool *update_msg_mem_pool;       // Memory pool used for BPG UPDATE MSG allocations
   list tx_queue;                   // Stores queued packets going to be sent
   timer *connect_retry_timer;      // Timer for retrying connection to the BMP collector
-  struct rt_table_info rt_table_in_pre_policy; // Pre-policy route import table
+  list update_msg_queue;           // Stores all composed BGP UPDATE MSGs
   bool started;                    // Flag that stores running status of BMP instance
   int sock_err;                    // Last socket error code
 };
@@ -91,32 +83,12 @@ bmp_peer_up(const struct bgp_proto *bgp,
 	    const byte *rx_open_msg, uint rx_open_length);
 
 /**
- * The following 5 functions create BMP Route Monitoring message based on
+ * bmp_route_monitor_update_in_notify - send notification that rte vas received
- * pre-policy Adj-RIB-In. Composing Route Monitoring message consist of few
- * stages. First of all call bmp_route_monitor_update_in_pre_begin() in order
- * to start composing message. As a second step, call
- * bmp_route_monitor_update_in_notify() to announce each rte, which internally
- * calls bmp_route_monitor_put_update_in_pre_msg() in order to save BGP UPDATE
- * msg. As a third step call bmp_route_monitor_update_in_pre_commit() in order
- * to send BMP Route Monitoring message to the BMP collector. As a last step,
- * call bmp_route_monitor_update_in_pre_end() in order to release resources.
  */
-void
-bmp_route_monitor_update_in_pre_begin(void);
-
 void
 bmp_route_monitor_update_in_notify(struct channel *C, const net_addr *n,
 				   const struct rte *new, const struct rte_src *src);
 
-void
-bmp_route_monitor_update_in_pre_commit(const struct bgp_proto *bgp);
-
-void
-bmp_route_monitor_update_in_pre_end(void);
-
-void
-bmp_route_monitor_put_update_in_pre_msg(struct bmp_proto *p, const byte *data, const size_t data_size);
-
 /**
  * bmp_peer_down - send notification that BGP peer connection is not in
  * established state
@@ -129,9 +101,6 @@ bmp_peer_down(const struct bgp_proto *bgp, const int err_class, const byte *pkt,
 #else /* BMP build disabled */
 
 static inline void bmp_peer_up(const struct bgp_proto *bgp UNUSED, const byte *tx_open_msg UNUSED, uint tx_open_length UNUSED, const byte *rx_open_msg UNUSED, uint rx_open_length UNUSED) { }
-static inline void bmp_route_monitor_update_in_pre_begin(void) { }
-static inline void bmp_route_monitor_update_in_pre_commit(const struct bgp_proto *bgp UNUSED) { }
-static inline void bmp_route_monitor_update_in_pre_end(void) { }
 static inline void bmp_peer_down(const struct bgp_proto *bgp UNUSED, const int err_class UNUSED, const byte *pkt UNUSED, size_t pkt_size UNUSED) { }
 
 #endif /* CONFIG_BMP */