Issue #1748: Implement inter-router connection group upgrade

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

@@ -23,9 +23,10 @@
 
 import time
 from system_test import TestCase, main_module, Qdrouterd, unittest, retry
-from system_test import CA_CERT, SSL_PROFILE_TYPE, CONNECTION_TYPE
+from system_test import CA_CERT, SSL_PROFILE_TYPE, CONNECTION_TYPE, ROUTER_LINK_TYPE
 from system_test import CLIENT_CERTIFICATE, CLIENT_PRIVATE_KEY, CLIENT_PRIVATE_KEY_PASSWORD
 from system_test import SERVER_CERTIFICATE, SERVER_PRIVATE_KEY, SERVER_PRIVATE_KEY_PASSWORD
+from tcp_streamer import TcpStreamerThread
 
 
 class InterRouterCertRotationTest(TestCase):
@@ -56,6 +57,23 @@ def wait_inter_router_conns(self, router, count):
                    len(self.get_inter_router_conns(rtr)) == ct)
         self.assertTrue(ok, f"Failed to get {count} i.r. conns: {self.get_inter_router_conns(router)}")
 
+    def get_inter_router_data_conns(self, router):
+        dconns = self.get_inter_router_conns(router)
+        return [c for c in dconns if c['role'] == 'inter-router-data']
+
+    def get_links_by_conn_id(self, router, conn_id):
+        mgmt = router.management
+        links = mgmt.query(type=ROUTER_LINK_TYPE).get_dicts()
+        return [link for link in links if link['connectionId'] == conn_id]
+
+    def get_streaming_data_links(self, router):
+        ir_conns = self.get_inter_router_data_conns(router)
+        links = []
+        for conn in ir_conns:
+            links.extend([link for link in self.get_links_by_conn_id(router, conn['identity'])
+                          if link['linkType'] == 'endpoint'])
+        return links
+
     def test_01_ordinal_updates(self):
         """
         Verify that ordinal updates create new inter-router connections. Verify
@@ -194,6 +212,95 @@ def test_02_drop_old(self):
         router_L.teardown()
         router_C.teardown()
 
+    def test_03_tcp_streams(self):
+        """
+        Verify that existing TCP streams are not interrupted when new
+        inter-router connections are established.
+        """
+        data_conn_count = 4
+        inter_router_port = self.tester.get_port()
+        tcp_listener_port = self.tester.get_port()
+        tcp_connector_port = self.tester.get_port()
+
+        router_L = self.router("RouterL",
+                               [('sslProfile', {'name': 'ListenerSslProfile',
+                                                'caCertFile': CA_CERT,
+                                                'certFile': SERVER_CERTIFICATE,
+                                                'privateKeyFile': SERVER_PRIVATE_KEY,
+                                                'password': SERVER_PRIVATE_KEY_PASSWORD}),
+                                ('listener', {'name': 'Listener01',
+                                              'role': 'inter-router',
+                                              'host': '0.0.0.0',
+                                              'port': inter_router_port,
+                                              'requireSsl': 'yes',
+                                              'sslProfile': 'ListenerSslProfile'}),
+                                ('tcpListener', {'name': 'tcpListener01',
+                                                 'address': 'tcp/streaming',
+                                                 'port': tcp_listener_port})],
+                               data_conn_count, wait=False)
+        router_C = self.router("RouterC",
+                               [('sslProfile', {'name': "ConnectorSslProfile",
+                                                'ordinal': 0,
+                                                'oldestValidOrdinal': 0,
+                                                'caCertFile': CA_CERT,
+                                                'certFile': CLIENT_CERTIFICATE,
+                                                'privateKeyFile': CLIENT_PRIVATE_KEY,
+                                                'password': CLIENT_PRIVATE_KEY_PASSWORD}),
+                                ('connector', {'role': 'inter-router',
+                                               'host': 'localhost',
+                                               'port': inter_router_port,
+                                               'verifyHostname': 'yes',
+                                               'sslProfile': 'ConnectorSslProfile'}),
+                                ('tcpConnector', {'name': 'tcpConnector01',
+                                                  'address': 'tcp/streaming',
+                                                  'host': 'localhost',
+                                                  'port': tcp_connector_port})],
+                               data_conn_count, wait=True)
+        router_C.wait_router_connected("RouterL")
+
+        # wait for the inter-router connections to come up
+        self.wait_inter_router_conns(router_C, data_conn_count + 1)
+
+        # start TCP streaming connections across the routers
+        tcp_streamer = TcpStreamerThread(client_addr=('localhost', tcp_listener_port),
+                                         server_addr=('0.0.0.0', tcp_connector_port),
+                                         client_count=10, poll_timeout=0.2)
+
+        # Now wait until the streams have established (2 links per client) and
+        # traffic is passing
+        ok = retry(lambda rtr=router_C:
+                   len(self.get_streaming_data_links(rtr)) == 20)
+        self.assertTrue(ok, f"Failed to get 20 links: {self.get_streaming_data_links(router_C)}")
+        begin_recv = tcp_streamer.bytes_received
+        ok = retry(lambda: tcp_streamer.bytes_received > begin_recv)
+        self.assertTrue(ok, f"Failed to stream data {tcp_streamer.bytes_received}")
+
+        # Now rotate the certs
+        # update tlsOrdinal to 3 and wait for new conns to appear
+        router_C.management.update(type=SSL_PROFILE_TYPE,
+                                   attributes={'ordinal': 3},
+                                   name='ConnectorSslProfile')
+        self.wait_inter_router_conns(router_C, 2 * (data_conn_count + 1))
+
+        # verify that the streamer is still running and the streams are still passing traffic
+        begin_recv = tcp_streamer.bytes_received
+        ok = retry(lambda: tcp_streamer.bytes_received > begin_recv)
+        self.assertTrue(ok, f"Failed to stream data {tcp_streamer.bytes_received}")
+        self.assertTrue(tcp_streamer.is_alive, "Streamer has failed!")
+
+        # Update oldestValidOrdinal to 3. Expect the connections that carry the
+        # streaming data to close
+        router_C.management.update(type=SSL_PROFILE_TYPE,
+                                   attributes={'oldestValidOrdinal': 3},
+                                   name='ConnectorSslProfile')
+        self.wait_inter_router_conns(router_C, data_conn_count + 1)
+        ok = retry(lambda: tcp_streamer.is_alive is False)
+        self.assertTrue(ok, "Failed to terminate the streamer")
+
+        router_L.teardown()
+        router_C.teardown()
+        tcp_streamer.join()
+
 
 if __name__ == '__main__':
     unittest.main(main_module())