Fixes #1793: Move the LINK vflow record from the connector object to connector_config object. This ensures (#1794)

that only one LINK record exists per logical link

Signed-off-by: Ganesh Murthy       <[email protected]>

Author: ganeshmurthy

src/adaptors/amqp/amqp_adaptor.c

@@ -569,8 +569,8 @@ bool AMQP_rx_handler(qd_router_t *router, qd_link_t *link)
     //
     // Bump LINK metrics if appropriate
     //
-    if (!!conn->connector && !!conn->connector->vflow_record) {
-        vflow_inc_counter(conn->connector->vflow_record, VFLOW_ATTRIBUTE_OCTETS_REVERSE, (uint64_t) octets_received);
+    if (!!conn->connector && !!conn->connector->ctor_config && !!conn->connector->ctor_config->vflow_record) {
+        vflow_inc_counter(conn->connector->ctor_config->vflow_record, VFLOW_ATTRIBUTE_OCTETS_REVERSE, (uint64_t) octets_received);
     }
 
     // check if cut-through can be enabled or disabled
@@ -1535,8 +1535,8 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
                            && strncmp(key.start, QD_CONNECTION_PROPERTY_ACCESS_ID, key.size) == 0)) {
                     props_found += 1;
                     if (!pn_data_next(props)) break;
-                    if (!!connector && !!connector->vflow_record && pn_data_type(props) == PN_STRING) {
-                        vflow_set_ref_from_pn(connector->vflow_record, VFLOW_ATTRIBUTE_PEER, props);
+                    if (!!connector && !!connector->ctor_config && !!connector->ctor_config->vflow_record && pn_data_type(props) == PN_STRING) {
+                        vflow_set_ref_from_pn(connector->ctor_config->vflow_record, VFLOW_ATTRIBUTE_PEER, props);
                     }
 
                 } else {
@@ -2143,8 +2143,8 @@ static uint64_t CORE_link_deliver(void *context, qdr_link_t *link, qdr_delivery_
     //
     // Bump LINK metrics if appropriate
     //
-    if (!!qconn->connector && !!qconn->connector->vflow_record) {
-        vflow_inc_counter(qconn->connector->vflow_record, VFLOW_ATTRIBUTE_OCTETS, (uint64_t) octets_sent);
+    if (!!qconn->connector && !!qconn->connector->ctor_config && !!qconn->connector->ctor_config->vflow_record) {
+        vflow_inc_counter(qconn->connector->ctor_config->vflow_record, VFLOW_ATTRIBUTE_OCTETS, (uint64_t) octets_sent);
     }
 
     //

src/adaptors/amqp/qd_connector.c

@@ -108,7 +108,6 @@ static void qd_connector_config_cleanup_conns(void *context)
 static qd_error_t qd_connector_config_create_connectors(qd_connector_config_t *ctor_config)
 {
     ASSERT_MGMT_THREAD;  // only the mgmt thread can modify the connector list!
-
     qd_error_clear();
 
     // The connector configuration for inter-router connections may define a data connection count. This is the number
@@ -282,25 +281,6 @@ qd_connector_t *qd_connector_create(qd_connector_config_t *ctor_config, bool is_
     snprintf(item->host_port, hplen, "%s:%s", item->host , item->port);
     DEQ_INSERT_TAIL(connector->conn_info_list, item);
 
-    //
-    // Set up the vanflow record for this connector (LINK)
-    // Do this only for router-to-router connectors since the record represents an inter-router link
-    //
-    if ((strcmp(ctor_config->config.role, "inter-router") == 0 && !is_data_connector) ||
-        strcmp(ctor_config->config.role, "edge") == 0 ||
-        strcmp(ctor_config->config.role, "inter-edge") == 0) {
-        connector->vflow_record = vflow_start_record(VFLOW_RECORD_LINK, 0);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_NAME, ctor_config->config.name);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_ROLE, ctor_config->config.role);
-        vflow_set_uint64(connector->vflow_record, VFLOW_ATTRIBUTE_LINK_COST, ctor_config->config.inter_router_cost);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
-        vflow_set_uint64(connector->vflow_record, VFLOW_ATTRIBUTE_DOWN_COUNT, 0);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_PROTOCOL, item->scheme);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_DESTINATION_HOST, item->host);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_DESTINATION_PORT, item->port);
-        vflow_set_uint64(connector->vflow_record, VFLOW_ATTRIBUTE_OCTETS, 0);
-        vflow_set_uint64(connector->vflow_record, VFLOW_ATTRIBUTE_OCTETS_REVERSE, 0);
-    }
     return connector;
 }
 
@@ -360,8 +340,6 @@ void qd_connector_decref(qd_connector_t* connector)
         assert(connector->qd_conn == 0);
 
         qd_connector_config_decref(connector->ctor_config);
-        vflow_end_record(connector->vflow_record);
-        connector->vflow_record = 0;
         qd_timer_free(connector->reconnect_timer);
         sys_mutex_free(&connector->lock);
         sys_atomic_destroy(&connector->ref_count);
@@ -473,25 +451,34 @@ void qd_connector_remote_opened(qd_connector_t *connector)
 /**
  * Set the child connection of the connector
  */
-void qd_connector_add_connection(qd_connector_t *connector, qd_connection_t *ctx)
+void qd_connector_add_connection(qd_connector_t *connector, qd_connection_t *qd_conn)
 {
-    assert(ctx->connector == 0);
-
+    assert(qd_conn->connector == 0);
     sys_atomic_inc(&connector->ref_count);
-    ctx->connector = connector;
-    connector->qd_conn = ctx;
+    qd_conn->connector = connector;
+    connector->qd_conn = qd_conn;
+    if (!connector->is_data_connector) {
+        qd_connector_config_t *ctor_config = connector->ctor_config;
+        sys_atomic_inc(&ctor_config->active_control_conn_count);
+    }    
 }
 
 
 void qd_connector_add_link(qd_connector_t *connector)
 {
     if (!connector->is_data_connector) {
-        if (connector->vflow_record && connector->ctor_config->tls_config) {
-            // connector->ctor_config->tls_ordinal is set in the handle_connector_ssl_profile_mgmt_update() callback
-            vflow_set_uint64(connector->vflow_record, VFLOW_ATTRIBUTE_ACTIVE_TLS_ORDINAL, connector->ctor_config->tls_ordinal);
+        qd_connector_config_t *ctor_config = connector->ctor_config;
+        if (ctor_config && ctor_config->vflow_record) {
+            if (ctor_config->tls_config) {
+                // connector->ctor_config->tls_ordinal is set in the handle_connector_ssl_profile_mgmt_update() callback
+                vflow_set_uint64(ctor_config->vflow_record, VFLOW_ATTRIBUTE_ACTIVE_TLS_ORDINAL, connector->ctor_config->tls_ordinal);
+            }        
+            if (sys_atomic_get(&ctor_config->active_control_conn_count) == 1) {
+                vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "up");
+                vflow_set_timestamp_now(ctor_config->vflow_record, VFLOW_ATTRIBUTE_UP_TIMESTAMP);
+            }
+
         }
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "up");
-        vflow_set_timestamp_now(connector->vflow_record, VFLOW_ATTRIBUTE_UP_TIMESTAMP);
         connector->oper_status_down = false;
     }
 }
@@ -505,14 +492,25 @@ void qd_connector_remove_connection(qd_connector_t *connector, bool final, const
 {
     sys_mutex_lock(&connector->lock);
 
+    if (!connector->is_data_connector) {
+        qd_connector_config_t *ctor_config = connector->ctor_config;
+        sys_atomic_dec(&ctor_config->active_control_conn_count);
+    }     
+    
+
     qd_connection_t *ctx = connector->qd_conn;
     if (!connector->is_data_connector && !connector->oper_status_down  && !final) {
         connector->oper_status_down = true;
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
-        vflow_inc_counter(connector->vflow_record, VFLOW_ATTRIBUTE_DOWN_COUNT, 1);
-        vflow_set_timestamp_now(connector->vflow_record, VFLOW_ATTRIBUTE_DOWN_TIMESTAMP);
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_RESULT, condition_name        ? condition_name : "unknown");
-        vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_REASON, condition_description ? condition_description : "");
+        qd_connector_config_t *ctor_config = connector->ctor_config;
+        // If there are no active control connections, we can safely assume that
+        // the operation status of the LINK record is "down"
+        if (ctor_config && ctor_config->vflow_record && sys_atomic_get(&ctor_config->active_control_conn_count) == 0) {
+            vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
+            vflow_inc_counter(ctor_config->vflow_record, VFLOW_ATTRIBUTE_DOWN_COUNT, 1);
+            vflow_set_timestamp_now(ctor_config->vflow_record, VFLOW_ATTRIBUTE_DOWN_TIMESTAMP);
+            vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_RESULT, condition_name        ? condition_name : "unknown");
+            vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_REASON, condition_description ? condition_description : "");
+        }
     }
     connector->qd_conn = 0;
     ctx->connector = 0;
@@ -644,6 +642,7 @@ qd_connector_config_t *qd_connector_config_create(qd_dispatch_t *qd, qd_entity_t
     ZERO(ctor_config);
     DEQ_ITEM_INIT(ctor_config);
     sys_atomic_init(&ctor_config->ref_count, 1);  // for caller
+    sys_atomic_init(&ctor_config->active_control_conn_count, 0);
     ctor_config->server = qd_dispatch_get_server(qd);
     DEQ_INIT(ctor_config->connectors);
     ctor_config->cleanup_timer = qd_timer(amqp_adaptor.dispatch, qd_connector_config_cleanup_conns, ctor_config);
@@ -663,6 +662,7 @@ qd_connector_config_t *qd_connector_config_create(qd_dispatch_t *qd, qd_entity_t
 
     const bool is_inter_router = strcmp(ctor_config->config.role, "inter-router") == 0;
     const bool is_edge = strcmp(ctor_config->config.role, "edge") == 0;
+    const bool is_inter_edge = strcmp(ctor_config->config.role, "inter-edge") == 0;
 
     //
     // If an sslProfile is configured allocate a TLS config to be used by all child connector's connections
@@ -684,9 +684,25 @@ qd_connector_config_t *qd_connector_config_create(qd_dispatch_t *qd, qd_entity_t
                                                    handle_connector_ssl_profile_mgmt_update);
         }
     }
-
+    if (is_inter_router || is_edge || is_inter_edge) {
+        ctor_config->vflow_record = vflow_start_record(VFLOW_RECORD_LINK, 0);
+        vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_NAME, ctor_config->config.name);
+        vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_ROLE, ctor_config->config.role);
+        vflow_set_uint64(ctor_config->vflow_record, VFLOW_ATTRIBUTE_LINK_COST, ctor_config->config.inter_router_cost);
+        vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
+        vflow_set_uint64(ctor_config->vflow_record, VFLOW_ATTRIBUTE_DOWN_COUNT, 0);
+        if (ctor_config->config.ssl_required) {
+            vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_PROTOCOL, "amqps");
+        } else {
+            vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_PROTOCOL, "amqp");
+        }
+                vflow_set_string(ctor_config->vflow_record, VFLOW_ATTRIBUTE_DESTINATION_PORT, ctor_config->config.port);
+        vflow_set_uint64(ctor_config->vflow_record, VFLOW_ATTRIBUTE_OCTETS, 0);
+        vflow_set_uint64(ctor_config->vflow_record, VFLOW_ATTRIBUTE_OCTETS_REVERSE, 0);
+    }    
+    //
     // For inter-router connectors generate a group correlator and configure the data connection count
-
+    //
     if (is_inter_router) {
         qd_generate_discriminator(ctor_config->group_correlator);
         ctor_config->data_connection_count = qd_dispatch_get_data_connection_count(qd);
@@ -742,13 +758,19 @@ void qd_connector_config_decref(qd_connector_config_t *ctor_config)
     assert(rc > 0);  // else underflow
 
     if (rc == 1) {
+        if (ctor_config->vflow_record) {
+           vflow_end_record(ctor_config->vflow_record);
+           ctor_config->vflow_record = 0;
+        }
+        
         // Expect: all connectors hold the ref_count so this must be empty
         assert(DEQ_IS_EMPTY(ctor_config->connectors));
 
         // free the timer first otherwise the callback can run and attempt to access ctor_config while it is being torn
         // down:
         qd_timer_free(ctor_config->cleanup_timer);
         sys_atomic_destroy(&ctor_config->ref_count);
+        sys_atomic_destroy(&ctor_config->active_control_conn_count);
         free(ctor_config->policy_vhost);
         qd_tls_config_decref(ctor_config->tls_config);
         qd_server_config_free(&ctor_config->config);

src/adaptors/amqp/qd_connector.h

@@ -64,7 +64,6 @@ typedef struct qd_connector_t {
     sys_mutex_t               lock;
     connector_state_t         state;
     qd_connection_t          *qd_conn;
-    vflow_record_t           *vflow_record;
     bool                      oper_status_down;  // set when oper-status transitions to 'down' to avoid repeated error indications.
     bool                      is_data_connector; // inter-router conn for streaming messages
 
@@ -96,13 +95,13 @@ struct qd_connector_config_t {
     qd_server_t              *server;
     char                     *policy_vhost;  /* Optional policy vhost name */
     qd_timer_t               *cleanup_timer; /* remove quiesced connectors */
-
+    vflow_record_t           *vflow_record; /* vflow record for VFLOW_RECORD_LINK */
     // TLS Configuration. Keep a local copy of the TLS ordinals to monitor changes by management
     qd_tls_config_t          *tls_config;
     uint64_t                  tls_ordinal;
     uint64_t                  tls_oldest_valid_ordinal;
     uint32_t                  data_connection_count;  // # of child inter-router data connections
-
+    sys_atomic_t              active_control_conn_count;
     // The group correlation id for all child connectors/connections
     char                      group_correlator[QD_DISCRIMINATOR_SIZE];
 

tests/system_tests_cert_rotation.py

@@ -31,6 +31,7 @@
 from system_test import CLIENT2_CERTIFICATE, CLIENT2_PRIVATE_KEY, CLIENT2_PRIVATE_KEY_PASSWORD
 from system_test import SERVER2_CERTIFICATE, SERVER2_PRIVATE_KEY, SERVER2_PRIVATE_KEY_PASSWORD
 from tcp_streamer import TcpStreamerThread
+from vanflow_snooper import VFlowSnooperThread, ANY_VALUE
 
 
 class InterRouterCertRotationTest(TestCase):
@@ -122,13 +123,55 @@ def test_01_ordinal_updates(self):
         zero_ordinals = [c for c in irc if c['tlsOrdinal'] == 0]
         self.assertEqual(data_conn_count + 1, len(zero_ordinals), f"Missing conns: {zero_ordinals}")
 
+        snooper_thread = VFlowSnooperThread(router_C.addresses[0])
+
+        expected = {
+            'RouterL': [('ROUTER_ACCESS', {'LINK_COUNT': 1,
+                                           'ROLE': 'inter-router',
+                                           'IDENTITY': ANY_VALUE})],
+            'RouterC': [('LINK', {'PEER': ANY_VALUE,
+                                  'OPER_STATUS': 'up',
+                                  'ACTIVE_TLS_ORDINAL': 0,
+                                  'ROLE': 'inter-router'})]
+        }
+        success = retry(lambda: snooper_thread.match_records(expected), delay=2)
+        self.assertTrue(success, f"Failed to match records {snooper_thread.get_results()}")
+
+        # Before incrementing the ordinal on the sslProfile, get the vflow LINK record
+        # and capture its identity and start time. After we update the ordinal, we will
+        # make sure that the LINK record we obtain then has the same ordinal and start time
+        # as the current LINK record. This will prove that the LINK records were not recreated
+        # across ordinal updates.
+        link_vflow_recs = snooper_thread.get_router_records("RouterC", record_type='LINK')
+        link_vflow_dict = link_vflow_recs[0]
+        link_identity = link_vflow_dict['IDENTITY']
+        link_start_time = link_vflow_dict['START_TIME']
+
         # 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))
         self.wait_inter_router_conns(router_L, 2 * (data_conn_count + 1))
 
+        # The ordinal has been updated to 3, check to see if the LINK has the correct
+        # ordinal value of 3
+        expected = {
+            'RouterC': [('LINK', {'PEER': ANY_VALUE,
+                                  'OPER_STATUS': 'up',
+                                  'ACTIVE_TLS_ORDINAL': 3,
+                                  'ROLE': 'inter-router'})]
+        }
+        success = retry(lambda: snooper_thread.match_records(expected), delay=2)
+        self.assertTrue(success, f"Failed to match records {snooper_thread.get_results()}")
+
+        # Check to see if there is still the same link
+        # record as from before the ordinal was updated.
+        link_vflow_recs = snooper_thread.get_router_records("RouterC", record_type='LINK')
+        link_vflow_dict = link_vflow_recs[0]
+        self.assertEqual(link_identity, link_vflow_dict['IDENTITY'])
+        self.assertEqual(link_start_time, link_vflow_dict['START_TIME'])
+
         # Update oldestValidOrdinal to 3. Expect the older connections with an
         # ordinal value of 0 to be deleted
         router_C.management.update(type=SSL_PROFILE_TYPE,
@@ -137,13 +180,30 @@ def test_01_ordinal_updates(self):
         self.wait_inter_router_conns(router_L, data_conn_count + 1)
         self.wait_inter_router_conns(router_C, data_conn_count + 1)
 
+        # The oldestValidOrdinal has been updated to 3, we will check to see if there is
+        # still the same link record as from before the ordinal was updated.
+        link_vflow_recs = snooper_thread.get_router_records("RouterC", record_type='LINK')
+        link_vflow_dict = link_vflow_recs[0]
+        self.assertEqual(link_identity, link_vflow_dict['IDENTITY'])
+        self.assertEqual(link_start_time, link_vflow_dict['START_TIME'])
+
         # Verify all group Ordinals are 3 (same as connector tlsOrdinal)
         irc = router_C.get_inter_router_conns()
         irc.extend(router_L.get_inter_router_conns())
         self.assertEqual(2 * (data_conn_count + 1),
                          len([c for c in irc if c['groupOrdinal'] == 3]),
                          f"Unexpected conns: {irc}")
         router_L.teardown()
+
+        # Router L has now been torn down, check to see if the RouterC's OPER_STATUS on the LINK record is "down"
+        expected = {
+            'RouterC': [('LINK', {'PEER': ANY_VALUE,
+                                  'OPER_STATUS': 'down',
+                                  "PROTOCOL": "amqp",
+                                  'ROLE': 'inter-router'})]
+        }
+
+        success = retry(lambda: snooper_thread.match_records(expected), delay=2)
         router_C.teardown()
 
     def test_02_drop_old(self):

tests/vanflow_snooper.py

@@ -127,7 +127,8 @@
     62: "PROXY_HOST",
     63: "PROXY_PORT",
     64: "ERROR_LISTENER_SIDE",
-    65: "ERROR_CONNECTOR_SIDE"
+    65: "ERROR_CONNECTOR_SIDE",
+    66: "ACTIVE_TLS_ORDINAL"
 }