Issue #1748: Implement inter-router connection group upgrade

This introduces the concept of a connection group ordinal. It
annotates inter-router connections. Its purpose is to assign a
precedence to connections within and inter-router connection
group. Those inter router connections with the highest ordinal become
the active routing path to a peer router. Connections within the group
that have a lower ordinal remain active but no new flows are sent over
them.

This is used to implement connection upgrade for certificate rotation.

Author: kgiusti

python/skupper_router/management/skrouter.json

@@ -1437,6 +1437,10 @@
                     "graph": true,
                     "description": "The total number of deliveries that have traversed this link."
                 },
+                "connectionId": {
+                    "type": "integer",
+                    "description": "The identity of the parent connection."
+                },
                 "presettledCount": {
                     "type": "integer",
                     "graph": true,

src/router_core/agent_connection.c

@@ -101,6 +101,7 @@ const char *qdr_connection_columns[] =
      "meshId",
      "tlsOrdinal",
      "groupCorrelationId",
+     "groupOrdinal",
      0};
 
 const char *CONNECTION_TYPE = "io.skupper.router.connection";

src/router_core/connections.c

@@ -39,10 +39,11 @@ DEQ_DECLARE(qdr_forward_deliver_info_t, qdr_forward_deliver_info_list_t);
 ALLOC_DEFINE(qdr_forward_deliver_info_t);
 
 
-// get the control link for a given inter-router connection
+// get the outgoing control link for a given inter-router connection
 static inline qdr_link_t *peer_router_control_link(qdr_core_t *core, int conn_mask)
 {
-    return (conn_mask >= 0) ? core->control_links_by_mask_bit[conn_mask] : 0;
+    qdr_connection_t *conn = (conn_mask >= 0) ? core->rnode_conns_by_mask_bit[conn_mask] : 0;
+    return (!!conn) ? conn->control_links[QD_OUTGOING] : 0;
 }
 
 
@@ -54,11 +55,15 @@ static inline qdr_link_t *peer_router_data_link(qdr_core_t *core,
     if (conn_mask < 0 || priority < 0)
         return 0;
 
+    qdr_connection_t *conn = core->rnode_conns_by_mask_bit[conn_mask];
+    if (!conn)
+        return 0;
+
     // Try to return the requested priority link, but if it does
     // not exist, return the closest one that is lower.
     qdr_link_t * link = 0;
     while (1) {
-        if ((link = core->data_links_by_mask_bit[conn_mask].links[priority]))
+        if ((link = conn->data_links.link[priority]))
             return link;
         if (-- priority < 0)
             return 0;

src/router_core/forwarder.c

@@ -240,23 +240,18 @@ void qdr_route_table_setup_CT(qdr_core_t *core)
 
         core->neighbor_free_mask = qd_bitmask(1);
 
-        core->routers_by_mask_bit       = NEW_PTR_ARRAY(qdr_node_t, qd_bitmask_width());
-        core->control_links_by_mask_bit = NEW_PTR_ARRAY(qdr_link_t, qd_bitmask_width());
-        core->rnode_conns_by_mask_bit   = NEW_PTR_ARRAY(qdr_connection_t, qd_bitmask_width());
-        core->data_links_by_mask_bit    = NEW_ARRAY(qdr_priority_sheaf_t, qd_bitmask_width());
+        core->routers_by_mask_bit             = NEW_PTR_ARRAY(qdr_node_t, qd_bitmask_width());
+        core->rnode_conns_by_mask_bit         = NEW_PTR_ARRAY(qdr_connection_t, qd_bitmask_width());
+        core->pending_rnode_conns_by_mask_bit = NEW_PTR_ARRAY(qdr_connection_t, qd_bitmask_width());
+        core->group_correlator_by_maskbit     = NEW_PTR_ARRAY(char, qd_bitmask_width());
         DEQ_INIT(core->unallocated_group_members);
-        core->group_correlator_by_maskbit = NEW_PTR_ARRAY(char, qd_bitmask_width());
 
         for (int idx = 0; idx < qd_bitmask_width(); idx++) {
-            core->routers_by_mask_bit[idx]          = 0;
-            core->control_links_by_mask_bit[idx]    = 0;
-            core->data_links_by_mask_bit[idx].count = 0;
-            core->rnode_conns_by_mask_bit[idx]      = 0;
-            for (int priority = 0; priority < QDR_N_PRIORITIES; ++ priority) {
-                core->data_links_by_mask_bit[idx].links[priority] = 0;
-            }
-            core->group_correlator_by_maskbit[idx] = (char*) malloc(QD_DISCRIMINATOR_SIZE);
-            core->group_correlator_by_maskbit[idx][0] = '\0';
+            core->routers_by_mask_bit[idx]             = 0;
+            core->rnode_conns_by_mask_bit[idx]         = 0;
+            core->pending_rnode_conns_by_mask_bit[idx] = 0;
+            core->group_correlator_by_maskbit[idx]     = (char*) qd_malloc(QD_DISCRIMINATOR_SIZE);
+            core->group_correlator_by_maskbit[idx][0]  = '\0';
         }
     }
 }
@@ -441,7 +436,13 @@ static void qdr_set_link_CT(qdr_core_t *core, qdr_action_t *action, bool discard
         return;
     }
 
-    if (core->control_links_by_mask_bit[conn_maskbit] == 0) {
+    qdr_connection_t *conn = core->rnode_conns_by_mask_bit[conn_maskbit];
+    if (conn == 0) {
+        qd_log(LOG_ROUTER_CORE, QD_LOG_CRITICAL, "set_link: Invalid conn reference: %d", conn_maskbit);
+        return;
+    }
+
+    if (conn->control_links[QD_OUTGOING] == 0) {
         qd_log(LOG_ROUTER_CORE, QD_LOG_CRITICAL, "set_link: Invalid link reference: %d", conn_maskbit);
         return;
     }

src/router_core/route_tables.c

@@ -351,11 +351,10 @@ void qdr_core_free(qdr_core_t *core)
     assert(DEQ_IS_EMPTY(core->action_list_background));
     assert(DEQ_IS_EMPTY(core->streaming_connections));
 
-    if (core->routers_by_mask_bit)         free(core->routers_by_mask_bit);
-    if (core->control_links_by_mask_bit)   free(core->control_links_by_mask_bit);
-    if (core->data_links_by_mask_bit)      free(core->data_links_by_mask_bit);
-    if (core->neighbor_free_mask)          qd_bitmask_free(core->neighbor_free_mask);
-    if (core->rnode_conns_by_mask_bit)     free(core->rnode_conns_by_mask_bit);
+    if (core->routers_by_mask_bit)             free(core->routers_by_mask_bit);
+    if (core->neighbor_free_mask)              qd_bitmask_free(core->neighbor_free_mask);
+    if (core->rnode_conns_by_mask_bit)         free(core->rnode_conns_by_mask_bit);
+    if (core->pending_rnode_conns_by_mask_bit) free(core->pending_rnode_conns_by_mask_bit);
     if (core->group_correlator_by_maskbit) {
         for (int idx = 0; idx < qd_bitmask_width(); idx++) {
             free(core->group_correlator_by_maskbit[idx]);
@@ -1063,11 +1062,10 @@ uint64_t qdr_identifier(qdr_core_t* core)
     return id;
 }
 
-void qdr_reset_sheaf(qdr_core_t *core, uint8_t n)
+
+void qdr_reset_sheaf(qdr_connection_t *conn)
 {
-  qdr_priority_sheaf_t *sheaf = core->data_links_by_mask_bit + n;
-  sheaf->count = 0;
-  memset(sheaf->links, 0, QDR_N_PRIORITIES * sizeof(void *));
+    conn->data_links = (qdr_priority_sheaf_t) {0};
 }
 
 

src/router_core/router_core.c

@@ -635,6 +635,14 @@ void qdr_core_remove_address_config(qdr_core_t *core, qdr_address_config_t *addr
 bool qdr_is_addr_treatment_multicast(qdr_address_t *addr);
 const char *get_address_treatment_string(qd_address_treatment_t  treatment);
 
+// non-streaming inter-router links sorted by priority
+//
+typedef struct qdr_priority_sheaf_t {
+    qdr_link_t *link[QDR_N_PRIORITIES];
+    int count;
+} qdr_priority_sheaf_t;
+
+
 //
 // Connection Information
 //
@@ -682,26 +690,29 @@ struct qdr_connection_t {
     bool                        closed; // This bit is used in the case where a client is trying to force close this connection.
     uint8_t                     next_pri;  // for incoming inter-router data links
     qdr_connection_role_t       role;
-    int                         inter_router_cost;
     qdr_conn_identifier_t      *conn_id;
     qdr_conn_identifier_t      *alt_conn_id;
     bool                        strip_annotations_in;
     bool                        strip_annotations_out;
+    bool                        enable_protocol_trace; // Has trace level logging been turned on for this connection.
+    bool                        has_streaming_links;   ///< one or more of this connection's links are for streaming messages
+    int                         inter_router_cost;
     int                         link_capacity;
-    int                         mask_bit;  ///< set only if inter-router connection
+    int                         mask_bit;  ///< set only if inter-router control connection
+    int                         group_parent_mask_bit;  ///< if inter-router data connection maskbit of group parent inter-router control conn
     qdr_connection_work_list_t  work_list;
     sys_mutex_t                 work_lock;
     qdr_link_ref_list_t         links;
     qdr_link_ref_list_t         links_with_work[QDR_N_PRIORITIES];
     qdr_connection_info_t      *connection_info;
     void                       *user_context; /* Updated from IO thread, use work_lock */
+    qdr_link_t                 *control_links[2];  // QD_LINK_CONTROL links [QD_INCOMING/QD_OUTGOING] (inter-router conn only)
+    qdr_priority_sheaf_t        data_links;  // links for non-streaming messages (by priority)  (inter-router conn only)
     qd_conn_oper_status_t       oper_status;
     qd_conn_admin_status_t      admin_status;
     qdr_error_t                *error;
     uint32_t                    conn_uptime; // Timestamp which can be used to calculate the number of seconds this connection has been up and running.
     uint32_t                    last_delivery_time; // Timestamp which can be used to calculate the number of seconds since the last delivery arrived on this connection.
-    bool                        enable_protocol_trace; // Has trace level logging been turned on for this connection.
-    bool                        has_streaming_links;   ///< one or more of this connection's links are for streaming messages
     qdr_link_list_t             streaming_link_pool;   ///< pool of links available for streaming messages
     const qd_policy_spec_t     *policy_spec;
     qdr_connection_list_t       connection_group;      ///< List of associated connection group members
@@ -767,11 +778,6 @@ struct qdr_conn_identifier_t {
     qdr_auto_link_list_t       auto_link_refs;
 };
 
-typedef struct qdr_priority_sheaf_t {
-    qdr_link_t *links[QDR_N_PRIORITIES];
-    int count;
-} qdr_priority_sheaf_t;
-
 
 struct qdr_protocol_adaptor_t {
     DEQ_LINKS(qdr_protocol_adaptor_t);
@@ -882,14 +888,13 @@ struct qdr_core_t {
     qdr_address_t             *router_addr_T;
     qdr_address_t             *routerma_addr_T;
 
-    qdr_node_list_t         routers;                      ///< List of routers, in order of cost, from lowest to highest
-    qd_bitmask_t           *neighbor_free_mask;           ///< bits available for new conns (qd_connection_t->mask_bit values)
-    qdr_node_t            **routers_by_mask_bit;          ///< indexed by qdr_node_t->mask_bit
-    qdr_connection_t      **rnode_conns_by_mask_bit;      ///< inter-router conns indexed by conn->mask_bit
-    qdr_link_t            **control_links_by_mask_bit;    ///< indexed by qdr_node_t->link_mask_bit, qdr_connection_t->mask_bit
-    qdr_priority_sheaf_t   *data_links_by_mask_bit;       ///< indexed by qdr_node_t->link_mask_bit, qdr_connection_t->mask_bit
-    qdr_connection_list_t   unallocated_group_members;    ///< List of unallocated group members (i.e. before the group is given a maskbit)
-    char                  **group_correlator_by_maskbit;  ///< Group correlator number indexed by conn->maskbit
+    qdr_node_list_t         routers;                          ///< List of routers, in order of cost, from lowest to highest
+    qd_bitmask_t           *neighbor_free_mask;               ///< bits available for new conns (qd_connection_t->mask_bit values)
+    qdr_node_t            **routers_by_mask_bit;              ///< indexed by qdr_node_t->mask_bit
+    qdr_connection_t      **rnode_conns_by_mask_bit;          ///< inter-router conns indexed by conn->mask_bit
+    qdr_connection_t      **pending_rnode_conns_by_mask_bit;  ///< higher precedence inter-router conns pending upgrade [conn->mask_bit]
+    qdr_connection_list_t   unallocated_group_members;        ///< List of unallocated group members (i.e. before the group is given a maskbit)
+    char                  **group_correlator_by_maskbit;      ///< Group correlator number indexed by conn->maskbit
     uint64_t                cost_epoch;
 
     uint64_t              next_tag;
@@ -1061,10 +1066,9 @@ void qdr_core_timer_free_CT(qdr_core_t *core, qdr_core_timer_t *timer);
  * Clears the sheaf of priority links in a connection.
  * Call this when a connection is being closed, when the mask-bit
  * for that sheaf is being returned to the core for re-use.
- * @param core Pointer to the core object returned by qd_core()
- * @param n uint8_t index for the sheaf to be reset prior to re-use.
+ * @param conn Pointer to the connection owning the sheaf
  */
-void qdr_reset_sheaf(qdr_core_t *core, uint8_t n);
+void qdr_reset_sheaf(qdr_connection_t *conn);
 
 /**
  * Run in an IO thread.

src/router_core/router_core_private.h

@@ -1204,6 +1204,66 @@ def get_inter_router_conns(self):
         conns = self.management.query(type=CONNECTION_TYPE).get_dicts()
         return [c for c in conns if 'inter-router' in c['role']]
 
+    def get_inter_router_data_conns(self):
+        """
+        Return a list of all inter-router-data connections present
+        """
+        dconns = self.get_inter_router_conns()
+        return [c for c in dconns if c['role'] == 'inter-router-data']
+
+    def get_inter_router_control_conns(self):
+        """
+        Return a list of all inter-router control connections present
+        """
+        dconns = self.get_inter_router_conns()
+        return [c for c in dconns if c['role'] == 'inter-router']
+
+    def get_links_by_conn_id(self, connection_id):
+        """
+        Return a list of all active AMQP links for the given connection
+        """
+        links = self.management.query(type=ROUTER_LINK_TYPE).get_dicts()
+        return [link for link in links if link['connectionId'] == connection_id]
+
+    def get_active_inter_router_data_links(self):
+        # Get the list of active inter-router-data links for the router. These
+        # are the dynamically provisioned links for passing streaming data
+        # between routers. For example there will be at least one streaming
+        # data link for every TCP session passing through the router.
+        ir_conns = self.get_inter_router_data_conns()
+        links = []
+        for conn in ir_conns:
+            links.extend([link for link in self.get_links_by_conn_id(conn['identity'])
+                          if link['linkType'] == 'endpoint' and
+                          link['operStatus'] == 'up'])
+        return links
+
+    def get_active_inter_router_control_links(self):
+        # Get the list of active inter-router control links for the
+        # router. These are the links that carry the inter-router Hello message
+        # traffic as well as priority-based non-streaming messages.
+        # There will be two for each inter-router path (1 incoming link, 1
+        # outgoing link)
+        ir_conns = self.get_inter_router_control_conns()
+        links = []
+        for conn in ir_conns:
+            links.extend([link for link in self.get_links_by_conn_id(conn['identity'])
+                          if link['linkType'] == 'router-control' and
+                          link['operStatus'] == 'up'])
+        return links
+
+    def get_last_topology_change(self):
+        """
+        Get the timestamp when this router last re-computed topology. Returns
+        None if the router is not interior.
+        """
+        try:
+            node_state = self.management.read(type=ROUTER_NODE_TYPE,
+                                              identity=f"router.node/{self.config.router_id}")
+            return node_state["lastTopoChange"]
+        except:
+            return None
+
 
 class NcatException(Exception):
     def __init__(self, error=None):