Temporarily stop creation of vflow observability records if too many at (#1792)

* Temporarily stop creation of vflow observability records if too many at
once. To prevent large memory growth in some test scenarios.

* Respond to feedback

* Make lower threshold close to upper.

---------

Co-authored-by: mick <[email protected]>

Author: mgoulish

include/qpid/dispatch/vanflow.h

@@ -57,6 +57,34 @@ typedef enum vflow_record_type {
     VFLOW_RECORD_BIFLOW_APP    = 0x11,  // Bidirectional Application (L7) flow
 } vflow_record_type_t;
 
+#define VFLOW_RECORD_MAX VFLOW_RECORD_BIFLOW_APP // Update this if you add new record types to vflow_record_type
+
+// Most record-types support critical functionality and must be emitted every time the corresponding
+// event occurs. But a few record-types support discretionary functionality that can be interrupted,
+// if necessary, to prevent excessive memory growth.
+// This array identifies which is which.
+static const bool discretionary_records[VFLOW_RECORD_MAX + 1] = {
+    false,     // VFLOW_RECORD_SITE          (0x00)
+    false,     // VFLOW_RECORD_ROUTER        (0x01)
+    false,     // VFLOW_RECORD_LINK          (0x02)
+    false,     // VFLOW_RECORD_CONTROLLER    (0x03)
+    false,     // VFLOW_RECORD_LISTENER      (0x04)
+    false,     // VFLOW_RECORD_CONNECTOR     (0x05)
+    true,      // VFLOW_RECORD_FLOW          (0x06)
+    false,     // VFLOW_RECORD_PROCESS       (0x07)
+    false,     // VFLOW_RECORD_IMAGE         (0x08)
+    false,     // VFLOW_RECORD_INGRESS       (0x09)
+    false,     // VFLOW_RECORD_EGRESS        (0x0a)
+    false,     // VFLOW_RECORD_COLLECTOR     (0x0b)
+    false,     // VFLOW_RECORD_PROCESS_GROUP (0x0c)
+    false,     // VFLOW_RECORD_HOST          (0x0d)
+    false,     // VFLOW_RECORD_LOG           (0x0e)
+    false,     // VFLOW_RECORD_ROUTER_ACCESS (0x0f)
+    true,      // VFLOW_RECORD_BIFLOW_TPORT  (0x10)
+    true,      // VFLOW_RECORD_BIFLOW_APP    (0x11)
+
+};
+
 // clang-format off
 typedef enum vflow_attribute {
     // Note: these values are shared with the Skupper control plane - do not re-use or change them without updating the

src/vanflow.c

@@ -46,6 +46,15 @@
 #define DEFERRED_DELETION_TICKS 25  // Five seconds
 #define VFLOW_ID_CUSTOM 0xffffffffffffffff
 
+// If the number of discretionary records rises above
+// this threshold, stop production of them to avoid
+// excessive memory growth.
+#define DISCRETIONARY_RECORDS_STOP_THRESHOLD    5000
+
+// If the number of discretionary records falls below
+// this threshold, allow their production to start again.
+#define DISCRETIONARY_RECORDS_START_THRESHOLD   4990
+
 //
 // If the record_id value is VFLOW_ID_CUSTOM, use the full_id for arbitrary strings, otherwise use ${s.source_id}:${record_id}
 //
@@ -148,6 +157,10 @@ static const int   rate_span                   = 10;    // Ten-second rolling av
 static sys_atomic_t site_configured;
 
 typedef struct {
+    // How many records of types that support discretionary currently exist?
+    // Their creation can be interrupted if necessary to avoid excessive memory growth.
+    sys_atomic_t         discretionary_record_count;
+    sys_atomic_t         emit_discretionary_records;
     qdr_core_t          *router_core;
     sys_mutex_t          lock;
     sys_mutex_t          id_lock;
@@ -864,6 +877,15 @@ static void _vflow_create_router_record(void)
  */
 static void _vflow_free_record_TH(vflow_record_t *record, bool recursive)
 {
+    // If this is one of the record types that supports discretionary functionality,
+    // count it. If the number currently existing is now below the
+    // resumption threshold, indicate that we should resume producing them.
+    if (discretionary_records[record->record_type]) {
+        if (sys_atomic_dec(&state->discretionary_record_count) <= DISCRETIONARY_RECORDS_START_THRESHOLD) {
+            sys_atomic_set(&state->emit_discretionary_records, 1);
+        }
+    }
+
     //
     // If this record is a child of a parent, remove it from the parent's child list
     //
@@ -1830,6 +1852,19 @@ vflow_record_t *vflow_start_record_custom_id(vflow_record_type_t record_type, vf
 //=====================================================================================
 vflow_record_t *vflow_start_record(vflow_record_type_t record_type, vflow_record_t *parent)
 {
+    // If this is one of the record types that supports discretionary functionality,
+    // and if we already have the maximum allowed number of those, just don't produce
+    // another one.
+    // Otherwise, produce and count it, and check if this one put us over the limit.
+    if (discretionary_records[record_type]) {
+        if (0 == sys_atomic_get(&state->emit_discretionary_records)) {
+            return 0;
+        }
+        if (sys_atomic_inc(&state->discretionary_record_count) >= DISCRETIONARY_RECORDS_STOP_THRESHOLD) {
+            sys_atomic_set(&state->emit_discretionary_records, 0);
+        }
+    }
+
     vflow_record_t *record = new_vflow_record_t();
     vflow_work_t   *work   = _vflow_work(_vflow_start_record_TH);
     ZERO(record);
@@ -1862,6 +1897,12 @@ vflow_record_t *vflow_start_co_record_iter(vflow_record_type_t record_type, qd_i
     // search/replacement algorithm in _vflow_process_co_record_TH will need to be re-written in a more general way.
     //
     assert(record_type == VFLOW_RECORD_BIFLOW_TPORT);
+    // The above record type is 'discretionary', which
+    // means we may have been told to temporarily halt
+    // production of them.
+    if (sys_atomic_get(&state->emit_discretionary_records) == 0) {
+        return 0;
+    }
     vflow_record_t *record = new_vflow_record_t();
     ZERO(record);
     record->record_type        = record_type;
@@ -1902,14 +1943,15 @@ void vflow_end_record(vflow_record_t *record)
 void vflow_serialize_identity(const vflow_record_t *record, qd_composed_field_t *field)
 {
     char buffer[IDENTITY_MAX + 1];
-    assert(!!record);
     if (!!record) {
         if (record->identity.record_id == VFLOW_ID_CUSTOM) {
             qd_compose_insert_string(field, record->identity.s.full_id);
         } else {
             snprintf(buffer, IDENTITY_MAX, "%s:%"PRIu64, record->identity.s.source_id, record->identity.record_id);
             qd_compose_insert_string(field, buffer);
         }
+    } else {
+        qd_compose_insert_null(field);
     }
 }
 
@@ -2440,6 +2482,8 @@ static void _vflow_init(qdr_core_t *core, void **adaptor_context)
     state = NEW(vflow_state_t);
     ZERO(state);
 
+    sys_atomic_init(&state->discretionary_record_count, 0);
+    sys_atomic_init(&state->emit_discretionary_records, 1);
     state->router_core = core;
     state->hostname = getenv("HOSTNAME");
     size_t hostLength = !!state->hostname ? strlen(state->hostname) : 0;