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.

4 files changed, 265 insertions, 1 deletions

diff --git a/nest/proto.c b/nest/proto.c
index 7b359152..6bd2427b 100644
--- a/nest/proto.c
+++ b/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,10 +297,133 @@ 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)
 {
   ASSERT(c->channel_state == CS_UP);
@@ -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
@@ -400,6 +539,14 @@ channel_do_start(struct channel *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)
 {
   rt_schedule_prune(c->table);
@@ -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)
diff --git a/nest/protocol.h b/nest/protocol.h
index d82e3983..17d10fcb 100644
--- a/nest/protocol.h
+++ b/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() */
 };
 
 
diff --git a/nest/route.h b/nest/route.h
index 1b4f2866..53cdcee8 100644
--- a/nest/route.h
+++ b/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)
diff --git a/nest/rt-table.c b/nest/rt-table.c
index 298320d9..626c2fb8 100644
--- a/nest/rt-table.c
+++ b/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);
     }