Merge pull request #11138 from opensourcerouting/fix/rpki

bgpd: RPKI reverts, nits...

2 files changed, 90 insertions, 106 deletions

diff --git a/bgpd/bgp_rpki.c b/bgpd/bgp_rpki.c
index eb9d5f3f73..0612e078cf 100644
--- a/bgpd/bgp_rpki.c
+++ b/bgpd/bgp_rpki.c
@@ -63,9 +63,6 @@
 #include "bgpd/bgp_rpki_clippy.c"
 #endif
 
-static struct thread *t_rpki;
-static struct thread *t_rpki_start;
-
 DEFINE_MTYPE_STATIC(BGPD, BGP_RPKI_CACHE, "BGP RPKI Cache server");
 DEFINE_MTYPE_STATIC(BGPD, BGP_RPKI_CACHE_GROUP, "BGP RPKI Cache server group");
 DEFINE_MTYPE_STATIC(BGPD, BGP_RPKI_RTRLIB, "BGP RPKI RTRLib");
@@ -75,6 +72,8 @@ DEFINE_MTYPE_STATIC(BGPD, BGP_RPKI_RTRLIB, "BGP RPKI RTRLib");
 #define RETRY_INTERVAL_DEFAULT 600
 #define BGP_RPKI_CACHE_SERVER_SYNC_RETRY_TIMEOUT 3
 
+static struct thread *t_rpki_sync;
+
 #define RPKI_DEBUG(...)                                                        \
 	if (rpki_debug) {                                                      \
 		zlog_debug("RPKI: " __VA_ARGS__);                              \
@@ -123,9 +122,9 @@ static struct cache *find_cache(const uint8_t preference);
 static int add_tcp_cache(const char *host, const char *port,
 			 const uint8_t preference, const char *bindaddr);
 static void print_record(const struct pfx_record *record, struct vty *vty);
-static int is_synchronized(void);
-static int is_running(void);
-static int is_stopping(void);
+static bool is_synchronized(void);
+static bool is_running(void);
+static bool is_stopping(void);
 static void route_match_free(void *rule);
 static enum route_map_cmd_result_t route_match(void *rule,
 					       const struct prefix *prefix,
@@ -137,10 +136,11 @@ static void revalidate_all_routes(void);
 
 static struct rtr_mgr_config *rtr_config;
 static struct list *cache_list;
-static int rtr_is_running;
-static int rtr_is_stopping;
+static bool rtr_is_running;
+static bool rtr_is_stopping;
+static bool rtr_is_synced;
 static _Atomic int rtr_update_overflow;
-static int rpki_debug;
+static bool rpki_debug;
 static unsigned int polling_period;
 static unsigned int expire_interval;
 static unsigned int retry_interval;
@@ -333,17 +333,17 @@ static struct rtr_mgr_group *get_groups(void)
 	return rtr_mgr_groups;
 }
 
-inline int is_synchronized(void)
+inline bool is_synchronized(void)
 {
-	return is_running() && rtr_mgr_conf_in_sync(rtr_config);
+	return rtr_is_synced;
 }
 
-inline int is_running(void)
+inline bool is_running(void)
 {
 	return rtr_is_running;
 }
 
-inline int is_stopping(void)
+inline bool is_stopping(void)
 {
 	return rtr_is_stopping;
 }
@@ -372,13 +372,13 @@ static void bgpd_sync_callback(struct thread *thread)
 	struct listnode *node;
 	struct prefix *prefix;
 	struct pfx_record rec;
-	int retval;
-	int socket = THREAD_FD(thread);
 
-	thread_add_read(bm->master, bgpd_sync_callback, NULL, socket, &t_rpki);
+	thread_add_read(bm->master, bgpd_sync_callback, NULL,
+			rpki_sync_socket_bgpd, NULL);
 
 	if (atomic_load_explicit(&rtr_update_overflow, memory_order_seq_cst)) {
-		while (read(socket, &rec, sizeof(struct pfx_record)) != -1)
+		while (read(rpki_sync_socket_bgpd, &rec,
+			    sizeof(struct pfx_record)) != -1)
 			;
 
 		atomic_store_explicit(&rtr_update_overflow, 0,
@@ -387,20 +387,12 @@ static void bgpd_sync_callback(struct thread *thread)
 		return;
 	}
 
-	retval = read(socket, &rec, sizeof(struct pfx_record));
+	int retval =
+		read(rpki_sync_socket_bgpd, &rec, sizeof(struct pfx_record));
 	if (retval != sizeof(struct pfx_record)) {
-		RPKI_DEBUG("Could not read from socket");
-		return;
-	}
-
-	/* RTR-Server crashed/terminated, let's handle and switch
-	 * to the second available RTR-Server according to preference.
-	 */
-	if (rec.socket && rec.socket->state == RTR_ERROR_FATAL) {
-		reset(true);
+		RPKI_DEBUG("Could not read from rpki_sync_socket_bgpd");
 		return;
 	}
-
 	prefix = pfx_record_to_prefix(&rec);
 
 	afi_t afi = (rec.prefix.ver == LRTR_IPV4) ? AFI_IP : AFI_IP6;
@@ -458,53 +450,29 @@ static void revalidate_all_routes(void)
 {
 	struct bgp *bgp;
 	struct listnode *node;
-	afi_t afi;
-	safi_t safi;
 
 	for (ALL_LIST_ELEMENTS_RO(bm->bgp, node, bgp)) {
 		struct peer *peer;
 		struct listnode *peer_listnode;
 
 		for (ALL_LIST_ELEMENTS_RO(bgp->peer, peer_listnode, peer)) {
-			FOREACH_AFI_SAFI (afi, safi) {
-				if (!peer->afc_nego[afi][safi])
-					continue;
 
-				if (!peer->bgp->rib[afi][safi])
-					continue;
+			for (size_t i = 0; i < 2; i++) {
+				safi_t safi;
+				afi_t afi = (i == 0) ? AFI_IP : AFI_IP6;
+
+				for (safi = SAFI_UNICAST; safi < SAFI_MAX;
+				     safi++) {
+					if (!peer->bgp->rib[afi][safi])
+						continue;
 
-				bgp_soft_reconfig_in(peer, afi, safi);
+					bgp_soft_reconfig_in(peer, afi, safi);
+				}
 			}
 		}
 	}
 }
 
-static void rpki_connection_status_cb(const struct rtr_mgr_group *group
-				      __attribute__((unused)),
-				      enum rtr_mgr_status status,
-				      const struct rtr_socket *socket
-				      __attribute__((unused)),
-				      void *data __attribute__((unused)))
-{
-	struct pfx_record rec = {0};
-	int retval;
-
-	if (is_stopping() ||
-	    atomic_load_explicit(&rtr_update_overflow, memory_order_seq_cst))
-		return;
-
-	if (status == RTR_MGR_ERROR)
-		rec.socket = socket;
-
-	retval = write(rpki_sync_socket_rtr, &rec, sizeof(rec));
-	if (retval == -1 && (errno == EAGAIN || errno == EWOULDBLOCK))
-		atomic_store_explicit(&rtr_update_overflow, 1,
-				      memory_order_seq_cst);
-
-	else if (retval != sizeof(rec))
-		RPKI_DEBUG("Could not write to rpki_sync_socket_rtr");
-}
-
 static void rpki_update_cb_sync_rtr(struct pfx_table *p __attribute__((unused)),
 				    const struct pfx_record rec,
 				    const bool added __attribute__((unused)))
@@ -546,8 +514,9 @@ static void rpki_init_sync_socket(void)
 		goto err;
 	}
 
+
 	thread_add_read(bm->master, bgpd_sync_callback, NULL,
-			rpki_sync_socket_bgpd, &t_rpki);
+			rpki_sync_socket_bgpd, NULL);
 
 	return;
 
@@ -559,9 +528,10 @@ err:
 
 static int bgp_rpki_init(struct thread_master *master)
 {
-	rpki_debug = 0;
-	rtr_is_running = 0;
-	rtr_is_stopping = 0;
+	rpki_debug = false;
+	rtr_is_running = false;
+	rtr_is_stopping = false;
+	rtr_is_synced = false;
 
 	cache_list = list_new();
 	cache_list->del = (void (*)(void *)) & free_cache;
@@ -596,23 +566,27 @@ static int bgp_rpki_module_init(void)
 	return 0;
 }
 
-static void start_expired(struct thread *thread)
+static void sync_expired(struct thread *thread)
 {
 	if (!rtr_mgr_conf_in_sync(rtr_config)) {
-		thread_add_timer(bm->master, start_expired, NULL,
+		RPKI_DEBUG("rtr_mgr is not synced, retrying.");
+		thread_add_timer(bm->master, sync_expired, NULL,
 				 BGP_RPKI_CACHE_SERVER_SYNC_RETRY_TIMEOUT,
-				 &t_rpki_start);
+				 &t_rpki_sync);
 		return;
 	}
 
-	rtr_is_running = 1;
+	RPKI_DEBUG("rtr_mgr sync is done.");
+
+	rtr_is_synced = true;
 }
 
 static int start(void)
 {
 	int ret;
 
-	rtr_is_stopping = 0;
+	rtr_is_stopping = false;
+	rtr_is_synced = false;
 	rtr_update_overflow = 0;
 
 	if (list_isempty(cache_list)) {
@@ -627,8 +601,7 @@ static int start(void)
 	RPKI_DEBUG("Polling period: %d", polling_period);
 	ret = rtr_mgr_init(&rtr_config, groups, groups_len, polling_period,
 			   expire_interval, retry_interval,
-			   rpki_update_cb_sync_rtr, NULL,
-			   rpki_connection_status_cb, NULL);
+			   rpki_update_cb_sync_rtr, NULL, NULL, NULL);
 	if (ret == RTR_ERROR) {
 		RPKI_DEBUG("Init rtr_mgr failed.");
 		return ERROR;
@@ -642,21 +615,23 @@ static int start(void)
 		return ERROR;
 	}
 
-	thread_add_timer(bm->master, start_expired, NULL, 0, &t_rpki_start);
+	thread_add_timer(bm->master, sync_expired, NULL, 0, &t_rpki_sync);
 
 	XFREE(MTYPE_BGP_RPKI_CACHE_GROUP, groups);
 
+	rtr_is_running = true;
+
 	return SUCCESS;
 }
 
 static void stop(void)
 {
-	rtr_is_stopping = 1;
+	rtr_is_stopping = true;
 	if (is_running()) {
-		THREAD_OFF(t_rpki_start);
+		THREAD_OFF(t_rpki_sync);
 		rtr_mgr_stop(rtr_config);
 		rtr_mgr_free(rtr_config);
-		rtr_is_running = 0;
+		rtr_is_running = false;
 	}
 }
 
@@ -665,9 +640,6 @@ static int reset(bool force)
 	if (is_running() && !force)
 		return SUCCESS;
 
-	if (thread_is_scheduled(t_rpki_start))
-		return SUCCESS;
-
 	RPKI_DEBUG("Resetting RPKI Session");
 	stop();
 	return start();
@@ -1360,32 +1332,33 @@ DEFUN (show_rpki_cache_connection,
 	}
 	vty_out(vty, "Connected to group %d\n", group->preference);
 	for (ALL_LIST_ELEMENTS_RO(cache_list, cache_node, cache)) {
-		if (cache->preference == group->preference) {
-			struct tr_tcp_config *tcp_config;
+		struct tr_tcp_config *tcp_config;
 #if defined(FOUND_SSH)
-			struct tr_ssh_config *ssh_config;
+		struct tr_ssh_config *ssh_config;
 #endif
-
-			switch (cache->type) {
-			case TCP:
-				tcp_config = cache->tr_config.tcp_config;
-				vty_out(vty, "rpki tcp cache %s %s pref %hhu\n",
-					tcp_config->host, tcp_config->port,
-					cache->preference);
-				break;
-
+		switch (cache->type) {
+		case TCP:
+			tcp_config = cache->tr_config.tcp_config;
+			vty_out(vty, "rpki tcp cache %s %s pref %hhu%s\n",
+				tcp_config->host, tcp_config->port,
+				cache->preference,
+				cache->rtr_socket->state == RTR_ESTABLISHED
+					? " (connected)"
+					: "");
+			break;
 #if defined(FOUND_SSH)
-			case SSH:
-				ssh_config = cache->tr_config.ssh_config;
-				vty_out(vty, "rpki ssh cache %s %u pref %hhu\n",
-					ssh_config->host, ssh_config->port,
-					cache->preference);
-				break;
+		case SSH:
+			ssh_config = cache->tr_config.ssh_config;
+			vty_out(vty, "rpki ssh cache %s %u pref %hhu%s\n",
+				ssh_config->host, ssh_config->port,
+				cache->preference,
+				cache->rtr_socket->state == RTR_ESTABLISHED
+					? " (connected)"
+					: "");
+			break;
 #endif
-
-			default:
-				break;
-			}
+		default:
+			break;
 		}
 	}
 
@@ -1413,7 +1386,7 @@ DEFUN (debug_rpki,
        DEBUG_STR
        "Enable debugging for rpki\n")
 {
-	rpki_debug = 1;
+	rpki_debug = true;
 	return CMD_SUCCESS;
 }
 
@@ -1424,7 +1397,7 @@ DEFUN (no_debug_rpki,
        DEBUG_STR
        "Disable debugging for rpki\n")
 {
-	rpki_debug = 0;
+	rpki_debug = false;
 	return CMD_SUCCESS;
 }
 
@@ -1478,6 +1451,7 @@ static void install_cli_commands(void)
 	install_element(ENABLE_NODE, &bgp_rpki_stop_cmd);
 
 	/* Install rpki reset command */
+	install_element(ENABLE_NODE, &rpki_reset_cmd);
 	install_element(RPKI_NODE, &rpki_reset_cmd);
 
 	/* Install rpki polling period commands */
diff --git a/doc/user/rpki.rst b/doc/user/rpki.rst
index c1308913a4..c2f8ad1268 100644
--- a/doc/user/rpki.rst
+++ b/doc/user/rpki.rst
@@ -102,13 +102,23 @@ The following commands are independent of a specific cache server.
 
 .. clicmd:: rpki polling_period (1-3600)
 
-
    Set the number of seconds the router waits until the router asks the cache
    again for updated data.
 
    The default value is 300 seconds.
 
-   The following commands configure one or multiple cache servers.
+.. clicmd:: rpki expire_interval (600-172800)
+
+   Set the number of seconds the router waits until the router expires the cache.
+
+   The default value is 7200 seconds.
+
+.. clicmd:: rpki retry_interval (1-7200)
+
+   Set the number of seconds the router waits until retrying to connect to the
+   cache server.
+
+   The default value is 600 seconds.
 
 .. clicmd:: rpki cache (A.B.C.D|WORD) PORT [SSH_USERNAME] [SSH_PRIVKEY_PATH] [SSH_PUBKEY_PATH] [KNOWN_HOSTS_PATH] [source A.B.C.D] PREFERENCE