Nest: Automatic channel reloads based on RPKI changes

If there are roa_check() calls in channel filters, then the channel
subscribes to ROA table notifications, which are sent when ROA tables
are updated (subject to settle time) and trigger channel reload or
refeed.

nest/proto.c

@@ -48,6 +48,7 @@ static char *e_states[] = { "DOWN", "FEEDING", "READY" };
 
 extern struct protocol proto_unix_iface;
 
+static void channel_request_reload(struct channel *c);
 static void proto_shutdown_loop(timer *);
 static void proto_rethink_goal(struct proto *p);
 static char *proto_state_name(struct proto *p);
@@ -180,6 +181,8 @@ proto_add_channel(struct proto *p, struct channel_config *cf)
   c->last_state_change = current_time();
   c->reloadable = 1;
 
+  init_list(&c->roa_subscriptions);
+
   CALL(c->channel->init, c, cf);
 
   add_tail(&p->channels, &c->n);
@@ -256,10 +259,15 @@ channel_feed_loop(void *ptr)
   if (c->export_state != ES_FEEDING)
     return;
 
+  /* Start feeding */
   if (!c->feed_active)
+  {
     if (c->proto->feed_begin)
       c->proto->feed_begin(c, !c->refeeding);
 
+    c->refeed_pending = 0;
+  }
+
   // DBG("Feeding protocol %s continued\n", p->name);
   if (!rt_feed_channel(c))
   {
@@ -289,9 +297,132 @@ channel_feed_loop(void *ptr)
 
   if (c->proto->feed_end)
     c->proto->feed_end(c);
+
+  /* Restart feeding */
+  if (c->refeed_pending)
+    channel_request_feeding(c);
 }
 
 
+static void
+channel_roa_in_changed(struct rt_subscription *s)
+{
+  struct channel *c = s->data;
+  int active = c->reload_event && ev_active(c->reload_event);
+
+  CD(c, "Reload triggered by RPKI change%s", active ? " - already active" : "");
+
+  if (!active)
+    channel_request_reload(c);
+  else
+    c->reload_pending = 1;
+}
+
+static void
+channel_roa_out_changed(struct rt_subscription *s)
+{
+  struct channel *c = s->data;
+  int active = (c->export_state == ES_FEEDING);
+
+  CD(c, "Feeding triggered by RPKI change%s", active ? " - already active" : "");
+
+  if (!active)
+    channel_request_feeding(c);
+  else
+    c->refeed_pending = 1;
+}
+
+/* Temporary code, subscriptions should be changed to resources */
+struct roa_subscription {
+  struct rt_subscription s;
+  node roa_node;
+};
+
+static int
+channel_roa_is_subscribed(struct channel *c, rtable *tab, int dir)
+{
+  void (*hook)(struct rt_subscription *) =
+    dir ? channel_roa_in_changed : channel_roa_out_changed;
+
+  struct roa_subscription *s;
+  node *n;
+
+  WALK_LIST2(s, n, c->roa_subscriptions, roa_node)
+    if ((s->s.tab == tab) && (s->s.hook == hook))
+      return 1;
+
+  return 0;
+}
+
+
+static void
+channel_roa_subscribe(struct channel *c, rtable *tab, int dir)
+{
+  if (channel_roa_is_subscribed(c, tab, dir))
+    return;
+
+  struct roa_subscription *s = mb_allocz(c->proto->pool, sizeof(struct roa_subscription));
+
+  s->s.hook = dir ? channel_roa_in_changed : channel_roa_out_changed;
+  s->s.data = c;
+  rt_subscribe(tab, &s->s);
+
+  add_tail(&c->roa_subscriptions, &s->roa_node);
+}
+
+static void
+channel_roa_unsubscribe(struct roa_subscription *s)
+{
+  rt_unsubscribe(&s->s);
+  rem_node(&s->roa_node);
+  mb_free(s);
+}
+
+static void
+channel_roa_subscribe_filter(struct channel *c, int dir)
+{
+  const struct filter *f = dir ? c->in_filter : c->out_filter;
+  struct rtable *tab;
+
+  if ((f == FILTER_ACCEPT) || (f == FILTER_REJECT))
+    return;
+
+  struct filter_iterator fit;
+  FILTER_ITERATE_INIT(&fit, f, c->proto->pool);
+
+  FILTER_ITERATE(&fit, fi)
+  {
+    switch (fi->fi_code)
+    {
+    case FI_ROA_CHECK_IMPLICIT:
+      tab = fi->i_FI_ROA_CHECK_IMPLICIT.rtc->table;
+      channel_roa_subscribe(c, tab, dir);
+      break;
+
+    case FI_ROA_CHECK_EXPLICIT:
+      tab = fi->i_FI_ROA_CHECK_EXPLICIT.rtc->table;
+      channel_roa_subscribe(c, tab, dir);
+      break;
+
+    default:
+      break;
+    }
+  }
+  FILTER_ITERATE_END;
+
+  FILTER_ITERATE_CLEANUP(&fit);
+}
+
+static void
+channel_roa_unsubscribe_all(struct channel *c)
+{
+  struct roa_subscription *s;
+  node *n, *x;
+
+  WALK_LIST2_DELSAFE(s, n, x, c->roa_subscriptions, roa_node)
+    channel_roa_unsubscribe(s);
+}
+
 static void
 channel_start_export(struct channel *c)
 {
@@ -329,11 +460,19 @@ channel_reload_loop(void *ptr)
 {
   struct channel *c = ptr;
 
+  /* Start reload */
+  if (!c->reload_active)
+    c->reload_pending = 0;
+
   if (!rt_reload_channel(c))
   {
     ev_schedule(c->reload_event);
     return;
   }
+
+  /* Restart reload */
+  if (c->reload_pending)
+    channel_request_reload(c);
 }
 
 static void
@@ -399,6 +538,14 @@ channel_do_start(struct channel *c)
   CALL(c->channel->start, c);
 }
 
+static void
+channel_do_up(struct channel *c)
+{
+  /* Register RPKI/ROA subscriptions */
+  channel_roa_subscribe_filter(c, 1);
+  channel_roa_subscribe_filter(c, 0);
+}
+
 static void
 channel_do_flush(struct channel *c)
 {
@@ -415,6 +562,8 @@ channel_do_flush(struct channel *c)
   c->in_table = NULL;
   c->reload_event = NULL;
   c->out_table = NULL;
+
+  channel_roa_unsubscribe_all(c);
 }
 
 static void
@@ -484,6 +633,7 @@ channel_set_state(struct channel *c, uint state)
     if (!c->gr_wait && c->proto->rt_notify)
       channel_start_export(c);
 
+    channel_do_up(c);
     break;
 
   case CS_FLUSHING:
@@ -694,6 +844,14 @@ channel_reconfigure(struct channel *c, struct channel_config *cf)
   if (c->channel_state != CS_UP)
     goto done;
 
+  /* Update RPKI/ROA subscriptions */
+  if (import_changed || export_changed)
+  {
+    channel_roa_unsubscribe_all(c);
+    channel_roa_subscribe_filter(c, 1);
+    channel_roa_subscribe_filter(c, 0);
+  }
+
   if (reconfigure_type == RECONFIG_SOFT)
   {
     if (import_changed)

nest/protocol.h

@@ -549,7 +549,12 @@ struct channel {
   struct rte *reload_next_rte;		/* Route iterator in in_table used during reloading */
   u8 reload_active;			/* Iterator reload_fit is linked */
 
+  u8 reload_pending;			/* Reloading and another reload is scheduled */
+  u8 refeed_pending;			/* Refeeding and another refeed is scheduled */
+
   struct rtable *out_table;		/* Internal table for exported routes */
+
+  list roa_subscriptions;		/* List of active ROA table subscriptions based on filters roa_check() */
 };
 
 

nest/route.h

@@ -19,6 +19,7 @@ struct protocol;
 struct proto;
 struct rte_src;
 struct symbol;
+struct timer;
 struct filter;
 struct cli;
 
@@ -147,6 +148,8 @@ struct rtable_config {
   int gc_max_ops;			/* Maximum number of operations before GC is run */
   int gc_min_time;			/* Minimum time between two consecutive GC runs */
   byte sorted;				/* Routes of network are sorted according to rte_better() */
+  btime min_settle_time;		/* Minimum settle time for notifications */
+  btime max_settle_time;		/* Maximum settle time for notifications */
 };
 
 typedef struct rtable {
@@ -166,6 +169,8 @@ typedef struct rtable {
 					 * obstacle from this routing table.
 					 */
   struct event *rt_event;		/* Routing table event */
+  btime last_rt_change;			/* Last time when route changed */
+  btime base_settle_time;		/* Start time of rtable settling interval */
   btime gc_time;			/* Time of last GC */
   int gc_counter;			/* Number of operations since last GC */
   byte prune_state;			/* Table prune state, 1 -> scheduled, 2-> running */
@@ -173,8 +178,18 @@ typedef struct rtable {
   byte nhu_state;			/* Next Hop Update state */
   struct fib_iterator prune_fit;	/* Rtable prune FIB iterator */
   struct fib_iterator nhu_fit;		/* Next Hop Update FIB iterator */
+
+  list subscribers;			/* Subscribers for notifications */
+  struct timer *settle_timer;		/* Settle time for notifications */
 } rtable;
 
+struct rt_subscription {
+  node n;
+  rtable *tab;
+  void (*hook)(struct rt_subscription *b);
+  void *data;
+};
+
 #define NHU_CLEAN	0
 #define NHU_SCHEDULED	1
 #define NHU_RUNNING	2
@@ -294,6 +309,8 @@ void rt_preconfig(struct config *);
 void rt_commit(struct config *new, struct config *old);
 void rt_lock_table(rtable *);
 void rt_unlock_table(rtable *);
+void rt_subscribe(rtable *tab, struct rt_subscription *s);
+void rt_unsubscribe(struct rt_subscription *s);
 void rt_setup(pool *, rtable *, struct rtable_config *);
 static inline net *net_find(rtable *tab, const net_addr *addr) { return (net *) fib_find(&tab->fib, addr); }
 static inline net *net_find_valid(rtable *tab, const net_addr *addr)

nest/rt-table.c

@@ -36,6 +36,7 @@
 #include "nest/iface.h"
 #include "lib/resource.h"
 #include "lib/event.h"
+#include "lib/timer.h"
 #include "lib/string.h"
 #include "conf/conf.h"
 #include "filter/filter.h"
@@ -60,6 +61,7 @@ static void rt_notify_hostcache(rtable *tab, net *net);
 static void rt_update_hostcache(rtable *tab);
 static void rt_next_hop_update(rtable *tab);
 static inline void rt_prune_table(rtable *tab);
+static inline void rt_schedule_notify(rtable *tab);
 
 
 /* Like fib_route(), but skips empty net entries */
@@ -968,6 +970,8 @@ rte_announce(rtable *tab, uint type, net *net, rte *new, rte *old,
       rt_notify_hostcache(tab, net);
   }
 
+  rt_schedule_notify(tab);
+
   struct channel *c; node *n;
   WALK_LIST2(c, n, tab->channels, table_node)
   {
@@ -1211,6 +1215,9 @@ rte_recalculate(struct channel *c, net *net, rte *new, struct rte_src *src)
   else
     stats->imp_withdraws_ignored++;
 
+  if (old_ok || new_ok)
+    table->last_rt_change = current_time();
+
  skip_stats1:
 
   if (new)
@@ -1792,6 +1799,78 @@ rt_event(void *ptr)
   rt_unlock_table(tab);
 }
 
+
+static inline btime
+rt_settled_time(rtable *tab)
+{
+  ASSUME(tab->base_settle_time != 0);
+
+  return MIN(tab->last_rt_change + tab->config->min_settle_time,
+	     tab->base_settle_time + tab->config->max_settle_time);
+}
+
+static void
+rt_settle_timer(timer *t)
+{
+  rtable *tab = t->data;
+
+  if (!tab->base_settle_time)
+    return;
+
+  btime settled_time = rt_settled_time(tab);
+  if (current_time() < settled_time)
+  {
+    tm_set(tab->settle_timer, settled_time);
+    return;
+  }
+
+  /* Settled */
+  tab->base_settle_time = 0;
+
+  struct rt_subscription *s;
+  WALK_LIST(s, tab->subscribers)
+    s->hook(s);
+}
+
+static void
+rt_kick_settle_timer(rtable *tab)
+{
+  tab->base_settle_time = current_time();
+
+  if (!tab->settle_timer)
+    tab->settle_timer = tm_new_init(rt_table_pool, rt_settle_timer, tab, 0, 0);
+
+  if (!tm_active(tab->settle_timer))
+    tm_set(tab->settle_timer, rt_settled_time(tab));
+}
+
+static inline void
+rt_schedule_notify(rtable *tab)
+{
+  if (EMPTY_LIST(tab->subscribers))
+    return;
+
+  if (tab->base_settle_time)
+    return;
+
+  rt_kick_settle_timer(tab);
+}
+
+void
+rt_subscribe(rtable *tab, struct rt_subscription *s)
+{
+  s->tab = tab;
+  rt_lock_table(tab);
+  add_tail(&tab->subscribers, &s->n);
+}
+
+void
+rt_unsubscribe(struct rt_subscription *s)
+{
+  rem_node(&s->n);
+  rt_unlock_table(s->tab);
+}
+
 void
 rt_setup(pool *p, rtable *t, struct rtable_config *cf)
 {
@@ -1806,7 +1885,9 @@ rt_setup(pool *p, rtable *t, struct rtable_config *cf)
   hmap_set(&t->id_map, 0);
 
   t->rt_event = ev_new_init(p, rt_event, t);
-  t->gc_time = current_time();
+  t->last_rt_change = t->gc_time = current_time();
+
+  init_list(&t->subscribers);
 }
 
 /**
@@ -2204,6 +2285,8 @@ rt_new_table(struct symbol *s, uint addr_type)
   c->addr_type = addr_type;
   c->gc_max_ops = 1000;
   c->gc_min_time = 5;
+  c->min_settle_time = 1 S;
+  c->max_settle_time = 20 S;
 
   add_tail(&new_config->tables, &c->n);
 
@@ -2250,6 +2333,7 @@ rt_unlock_table(rtable *r)
       fib_free(&r->fib);
       hmap_free(&r->id_map);
       rfree(r->rt_event);
+      rfree(r->settle_timer);
       mb_free(r);
       config_del_obstacle(conf);
     }