Fixes skupperproject#1456 - Implemented the administrative ACCESS_POI…

…NT and LINK

This also separates the closure-of-all-connections from the freeing of the server module so that connections can be closed at an earlier point of the shutdown sequence.

This also adds accepted-connections to the listener reference counts for a more orderly freeing of resources.

Original work by Ted Ross. Ported to amqp adaptor and obfuscated by kgiusti

include/qpid/dispatch/vanflow.h

@@ -51,6 +51,7 @@ typedef enum vflow_record_type {
     VFLOW_RECORD_PROCESS_GROUP = 0x0c,  // A grouping of PROCESS
     VFLOW_RECORD_HOST          = 0x0d,  // Host (or Kubernetes Node) on which a process runs
     VFLOW_RECORD_LOG           = 0x0e,  // A notable router log event such as an error or warning
+    VFLOW_RECORD_ACCESS_POINT  = 0x0f,  // An access point for inter-router connections
 } vflow_record_type_t;
 
 // clang-format off
@@ -121,13 +122,17 @@ typedef enum vflow_attribute {
     VFLOW_ATTRIBUTE_LOG_TEXT         = 49,  // String
     VFLOW_ATTRIBUTE_SOURCE_FILE      = 50,  // String
     VFLOW_ATTRIBUTE_SOURCE_LINE      = 51,  // uint
+
+    VFLOW_ATTRIBUTE_LINK_COUNT       = 52,  // uint/counter
+    VFLOW_ATTRIBUTE_OPER_STATUS      = 53,  // String
+    VFLOW_ATTRIBUTE_ROLE             = 54,  // String
 } vflow_attribute_t;
 // clang-format on
 
 #define VALID_REF_ATTRS     0x00006000000000e6
-#define VALID_UINT_ATTRS    0x00099ffa07800119
-#define VALID_COUNTER_ATTRS 0x0000035000800000
-#define VALID_STRING_ATTRS  0x00060005787ffe00
+#define VALID_UINT_ATTRS    0x00199ffa07800119
+#define VALID_COUNTER_ATTRS 0x0010035000800000
+#define VALID_STRING_ATTRS  0x00660005787ffe00
 #define VALID_TRACE_ATTRS   0x0000000080000000
 
 typedef enum vflow_log_severity {

src/adaptors/amqp/amqp_adaptor.c

@@ -20,6 +20,7 @@
 #include "private.h"
 #include "qd_connector.h"
 #include "qd_connection.h"
+#include "qd_listener.h"
 #include "container.h"
 #include "node_type.h"
 
@@ -2366,8 +2367,12 @@ static void qd_amqp_adaptor_final(void *adaptor_context)
         if (ctx->policy_settings)
             qd_policy_settings_free(ctx->policy_settings);
         if (ctx->connector) {
-            ctx->connector->qd_conn = 0;
-            qd_connector_decref(ctx->connector);
+            qd_connector_remove_connection(ctx->connector);
+            ctx->connector = 0;
+        }
+        if (ctx->listener) {
+            qd_listener_remove_connection(ctx->listener, ctx);
+            ctx->listener = 0;
         }
         sys_atomic_destroy(&ctx->wake_core);
         sys_atomic_destroy(&ctx->wake_cutthrough_inbound);

src/adaptors/amqp/connection_manager.c

@@ -30,6 +30,7 @@
 #include "qpid/dispatch/ctools.h"
 #include "qpid/dispatch/failoverlist.h"
 #include "qpid/dispatch/threading.h"
+#include "qpid/dispatch/vanflow.h"
 
 #include <proton/listener.h>
 
@@ -147,7 +148,7 @@ static void log_config(qd_server_config_t *c, const char *what, bool create)
 QD_EXPORT qd_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_entity_t *entity)
 {
     qd_connection_manager_t *cm = qd->connection_manager;
-    qd_listener_t *li = qd_server_listener(qd->server);
+    qd_listener_t *li = qd_listener(qd->server);
     if (!li || qd_server_config_load(qd, &li->config, entity, true, 0) != QD_ERROR_NONE) {
         qd_log(LOG_CONN_MGR, QD_LOG_ERROR, "Unable to create listener: %s", qd_error_message());
         qd_listener_decref(li);
@@ -166,6 +167,19 @@ QD_EXPORT qd_listener_t *qd_dispatch_configure_listener(qd_dispatch_t *qd, qd_en
     } else {
         li->config.failover_list = 0;
     }
+
+    //
+    // Set up the vanflow record for this listener (ACCESS_POINT).
+    // Do this only for router-to-routers links: not mgmt/metrics/healthz/websockets listeners
+    //
+    if (strcmp(li->config.role, "inter-router") == 0 ||
+        strcmp(li->config.role, "edge") == 0 ||
+        strcmp(li->config.role, "inter-edge") == 0) {
+        li->vflow_record = vflow_start_record(VFLOW_RECORD_ACCESS_POINT, 0);
+        vflow_set_string(li->vflow_record, VFLOW_ATTRIBUTE_ROLE, li->config.role);
+        vflow_set_uint64(li->vflow_record, VFLOW_ATTRIBUTE_LINK_COUNT, 0);
+    }
+
     DEQ_ITEM_INIT(li);
     DEQ_INSERT_TAIL(cm->listeners, li);
     log_config(&li->config, "Listener", true);
@@ -414,8 +428,17 @@ QD_EXPORT qd_connector_t *qd_dispatch_configure_connector(qd_dispatch_t *qd, qd_
         item->host_port = malloc(hplen);
         snprintf(item->host_port, hplen, "%s:%s", item->host , item->port);
 
-        DEQ_INSERT_TAIL(ct->conn_info_list, item);
+        //
+        // Set up the vanflow record for this connector (LINK)
+        //
+        ct->vflow_record = vflow_start_record(VFLOW_RECORD_LINK, 0);
+        vflow_set_string(ct->vflow_record, VFLOW_ATTRIBUTE_ROLE, ct->config.role);
+        vflow_set_string(ct->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
+        vflow_set_string(ct->vflow_record, VFLOW_ATTRIBUTE_PROTOCOL, item->scheme);
+        vflow_set_string(ct->vflow_record, VFLOW_ATTRIBUTE_DESTINATION_HOST, item->host);
+        vflow_set_string(ct->vflow_record, VFLOW_ATTRIBUTE_DESTINATION_PORT, item->port);
 
+        DEQ_INSERT_TAIL(ct->conn_info_list, item);
         return ct;
     }
 

src/adaptors/amqp/qd_connection.c

@@ -27,6 +27,7 @@
 #include "qpid/dispatch/proton_utils.h"
 #include "qpid/dispatch/protocol_adaptor.h"
 #include "qpid/dispatch/timer.h"
+#include "qpid/dispatch/vanflow.h"
 
 #include <proton/proactor.h>
 #include <proton/sasl.h>
@@ -445,10 +446,8 @@ static void decorate_connection(qd_connection_t *ctx, const qd_server_config_t *
  * Does not allocate any managed objects and therefore
  * does not take ENTITY_CACHE lock.
  */
-qd_connection_t *qd_server_connection_impl(qd_server_t *server, qd_server_config_t *config, qd_connection_t *ctx, qd_connector_t *connector)
+void qd_connection_init(qd_connection_t *ctx, qd_server_t *server, qd_server_config_t *config, qd_connector_t *connector, qd_listener_t *listener)
 {
-    assert(ctx);
-    ZERO(ctx);
     ctx->pn_conn = pn_connection();
     assert(ctx->pn_conn);
     sys_mutex_init(&ctx->deferred_call_lock);
@@ -466,29 +465,23 @@ qd_connection_t *qd_server_connection_impl(qd_server_t *server, qd_server_config
     DEQ_INIT(ctx->deferred_calls);
     DEQ_INIT(ctx->free_link_session_list);
 
+    // note: setup connector or listener before decorating the connection since
+    // decoration involves accessing the connection's parent.
+
     if (!!connector) {
-        ctx->connector = connector;
-        strncpy(ctx->group_correlator, connector->group_correlator, QD_DISCRIMINATOR_SIZE);
+        assert(!listener);
+        qd_connector_add_connection(connector, ctx);
+    } else if (!!listener) {
+        qd_listener_add_connection(listener, ctx);
     }
+
     decorate_connection(ctx, config);
 
     sys_mutex_lock(&amqp_adaptor.lock);
     DEQ_INSERT_TAIL(amqp_adaptor.conn_list, ctx);
     sys_mutex_unlock(&amqp_adaptor.lock);
-
-    return ctx;
 }
 
-/* Construct a new qd_connection. Thread safe.
- * Allocates a qd_connection_t object and therefore
- * takes ENTITY_CACHE lock.
- */
-qd_connection_t *qd_server_connection(qd_server_t *server, qd_server_config_t *config)
-{
-    qd_connection_t *ctx = new_qd_connection_t();
-    if (!ctx) return NULL;
-    return qd_server_connection_impl(server, config, ctx, 0);
-}
 
 qd_connector_t* qd_connection_connector(const qd_connection_t *c) {
     return c->connector;
@@ -553,18 +546,21 @@ void qd_connection_set_user(qd_connection_t *conn)
 
 static void qd_connection_free(qd_connection_t *qd_conn)
 {
-    // If this is a dispatch connector uncouple it from the
-    // connection and restart the re-connect timer if necessary
+    sys_mutex_lock(&amqp_adaptor.lock);
+    DEQ_REMOVE(amqp_adaptor.conn_list, qd_conn);
+    sys_mutex_unlock(&amqp_adaptor.lock);
+
+    // If this connection is from a connector uncouple it and restart the re-connect timer if necessary
 
     if (qd_conn->connector) {
-        qd_connector_release_connection(qd_conn->connector, qd_conn);
-        qd_connector_decref(qd_conn->connector);  // drop connection's reference
+        qd_connector_remove_connection(qd_conn->connector);
         qd_conn->connector = 0;
     }
 
-    sys_mutex_lock(&amqp_adaptor.lock);
-    DEQ_REMOVE(amqp_adaptor.conn_list, qd_conn);
-    sys_mutex_unlock(&amqp_adaptor.lock);
+    if (qd_conn->listener) {
+        qd_listener_remove_connection(qd_conn->listener, qd_conn);
+        qd_conn->listener = 0;
+    }
 
     // If counted for policy enforcement, notify it has closed
     if (qd_conn->policy_counted) {

src/adaptors/amqp/qd_connection.h

@@ -115,8 +115,9 @@ struct qd_connection_t {
 
 ALLOC_DECLARE_SAFE(qd_connection_t);
 
-qd_connection_t *qd_server_connection(qd_server_t *server, qd_server_config_t* config);
-qd_connection_t *qd_server_connection_impl(qd_server_t *server, qd_server_config_t *config, qd_connection_t *ctx, qd_connector_t *connector);
+// initialize a newly allocated qd_connection_t
+void qd_connection_init(qd_connection_t *qd_conn, qd_server_t *server, qd_server_config_t *config,
+                        qd_connector_t *connector, qd_listener_t *listener);
 
 qd_connector_t* qd_connection_connector(const qd_connection_t *c);
 

src/adaptors/amqp/qd_connector.c

@@ -23,6 +23,7 @@
 
 #include "qpid/dispatch/alloc_pool.h"
 #include "qpid/dispatch/timer.h"
+#include "qpid/dispatch/vanflow.h"
 
 #include <proton/proactor.h>
 
@@ -46,22 +47,10 @@ static qd_failover_item_t *qd_connector_get_conn_info_lh(qd_connector_t *ct) TA_
 
 
 /* Timer callback to try/retry connection open, connector->lock held */
-static void try_open_lh(qd_connector_t *connector, qd_connection_t *connection) TA_REQ(connector->lock)
+static void try_open_lh(qd_connector_t *connector, qd_connection_t *qd_conn) TA_REQ(connector->lock)
 {
-    // connection until pn_proactor_connect is called below
-    qd_connection_t *qd_conn = qd_server_connection_impl(connector->server, &connector->config, connection, connector);
-    if (!qd_conn) {                 /* Try again later */
-        qd_log(LOG_SERVER, QD_LOG_CRITICAL, "Allocation failure connecting to %s",
-               connector->config.host_port);
-        connector->delay = 10000;
-        connector->state = CXTR_STATE_CONNECTING;
-        qd_timer_schedule(connector->timer, connector->delay);
-        return;
-    }
+    qd_connection_init(qd_conn, connector->server, &connector->config, connector, 0);
 
-    sys_atomic_inc(&connector->ref_count);
-
-    connector->qd_conn = qd_conn;
     connector->state   = CXTR_STATE_OPEN;
     connector->delay   = 5000;
 
@@ -101,13 +90,7 @@ static void try_open_cb(void *context)
     // Allocate connection before taking connector lock to avoid
     // CONNECTOR - ENTITY_CACHE lock inversion deadlock window.
     qd_connection_t *ctx = new_qd_connection_t();
-    if (!ctx) {
-        qd_log(LOG_SERVER, QD_LOG_CRITICAL, "Allocation failure connecting to %s", ct->config.host_port);
-        ct->delay = 10000;
-        ct->state = CXTR_STATE_CONNECTING;
-        qd_timer_schedule(ct->timer, ct->delay);
-        return;
-    }
+    ZERO(ctx);
 
     sys_mutex_lock(&ct->lock);
 
@@ -185,6 +168,8 @@ void qd_connector_decref(qd_connector_t* connector)
         assert(connector->state == CXTR_STATE_DELETED);
         assert(connector->qd_conn == 0);
 
+        vflow_end_record(connector->vflow_record);
+        connector->vflow_record = 0;
         qd_server_config_free(&connector->config);
         qd_timer_free(connector->timer);
         sys_mutex_free(&connector->lock);
@@ -291,16 +276,35 @@ void qd_connector_remote_opened(qd_connector_t *connector)
     sys_mutex_unlock(&connector->lock);
 }
 
+/**
+ * Set the child connection of the connector
+ */
+void qd_connector_add_connection(qd_connector_t *connector, qd_connection_t *ctx)
+{
+    assert(ctx->connector == 0);
+
+    sys_atomic_inc(&connector->ref_count);
+    ctx->connector = connector;
+    connector->qd_conn = ctx;
+
+    strncpy(ctx->group_correlator, connector->group_correlator, QD_DISCRIMINATOR_SIZE);
+    vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "up");
+}
+
 
 /**
  * The connection associated with this connector is about to be freed,
  * clean up all related state
  */
-void qd_connector_release_connection(qd_connector_t *connector, qd_connection_t *qd_conn)
+void qd_connector_remove_connection(qd_connector_t *connector)
 {
     sys_mutex_lock(&connector->lock);
-    assert(connector->qd_conn == qd_conn);
-    connector->qd_conn = 0;  // this connection to be freed
+
+    qd_connection_t *ctx = connector->qd_conn;
+    vflow_set_string(connector->vflow_record, VFLOW_ATTRIBUTE_OPER_STATUS, "down");
+    connector->qd_conn = 0;
+    ctx->connector = 0;
+
     if (connector->state != CXTR_STATE_DELETED) {
         // Increment the connection index by so that we can try connecting to the failover url (if any).
         bool has_failover = qd_connector_has_failover_info(connector);
@@ -316,4 +320,7 @@ void qd_connector_release_connection(qd_connector_t *connector, qd_connection_t
         qd_timer_schedule(connector->timer, delay);
     }
     sys_mutex_unlock(&connector->lock);
+
+    // Drop reference held by connection.
+    qd_connector_decref(connector);
 }

src/adaptors/amqp/qd_connector.h

@@ -32,6 +32,7 @@
 typedef struct qd_timer_t  qd_timer_t;
 typedef struct qd_server_t qd_server_t;
 typedef struct qd_connection_t qd_connection_t;
+typedef struct vflow_record_t  vflow_record_t;
 
 typedef enum {
     CXTR_STATE_INIT = 0,
@@ -58,6 +59,7 @@ typedef struct qd_connector_t {
     sys_mutex_t               lock;
     cxtr_state_t              state;
     qd_connection_t          *qd_conn;
+    vflow_record_t           *vflow_record;
 
     /* This conn_list contains all the connection information needed to make a connection. It also includes failover connection information */
     qd_failover_item_list_t   conn_info_list;
@@ -94,5 +96,12 @@ bool qd_connector_has_failover_info(const qd_connector_t* ct);
 const char *qd_connector_policy_vhost(const qd_connector_t* ct);
 void qd_connector_handle_transport_error(qd_connector_t *connector, uint64_t connection_id, pn_condition_t *condition);
 void qd_connector_remote_opened(qd_connector_t *connector);
-void qd_connector_release_connection(qd_connector_t *connector, qd_connection_t *qd_conn);
+
+// add a new connection to the parent connector
+void qd_connector_add_connection(qd_connector_t *connector, qd_connection_t *ctx);
+
+// remove the child connection
+// NOTE WELL: this may free the connector if the connection is holding the last
+// reference to it
+void qd_connector_remove_connection(qd_connector_t *connector);
 #endif