northd: Process ACL changes incrementally.

Any change to ACL would result in full lflow recompute. Instead of
that reprocess only the relevant logical switch. In order to get
track the relation between ACL and LS from the ACL side we need to
store all related ACL uuids in the ls_stateful_record.

There isn't any significant memory impact that would happen as
result of tracking the uuids for each switch. Tested on database
with 76k ACLs and 25k PGs.

Reported-at: https://issues.redhat.com/browse/FDP-755
Acked-by: Mark Michelson <mmichels@redhat.com>
Acked-by: Dumitru Ceara <dceara@redhat.com>
Signed-off-by: Ales Musil <amusil@redhat.com>

Author: almusil

northd/en-ls-stateful.c

@@ -65,20 +65,13 @@ static void ls_stateful_record_destroy(struct ls_stateful_record *);
 static void ls_stateful_record_init(
     struct ls_stateful_record *,
     const struct ovn_datapath *,
-    const struct ls_port_group *,
-    const struct ls_port_group_table *);
-static void ls_stateful_record_reinit(
-    struct ls_stateful_record *,
-    const struct ovn_datapath *,
-    const struct ls_port_group *,
     const struct ls_port_group_table *);
 static bool ls_has_lb_vip(const struct ovn_datapath *);
-static void ls_stateful_record_set_acl_flags(
+static void ls_stateful_record_set_acls(
     struct ls_stateful_record *, const struct ovn_datapath *,
-    const struct ls_port_group *, const struct ls_port_group_table *);
-static bool ls_stateful_record_set_acl_flags_(struct ls_stateful_record *,
-                                              struct nbrec_acl **,
-                                              size_t n_acls);
+    const struct ls_port_group_table *);
+static void ls_stateful_record_set_acls_(struct ls_stateful_record *,
+                                         struct nbrec_acl **, size_t n_acls);
 static struct ls_stateful_input ls_stateful_get_input_data(
     struct engine_node *);
 
@@ -148,30 +141,31 @@ ls_stateful_northd_handler(struct engine_node *node, void *data_)
     struct ed_type_ls_stateful *data = data_;
     struct hmapx_node *hmapx_node;
 
-    struct hmapx changed_stateful_od = HMAPX_INITIALIZER(&changed_stateful_od);
     HMAPX_FOR_EACH (hmapx_node, &nd_changes->ls_with_changed_lbs) {
-        hmapx_add(&changed_stateful_od, hmapx_node->data);
-    }
+        const struct ovn_datapath *od = hmapx_node->data;
 
-    HMAPX_FOR_EACH (hmapx_node, &nd_changes->ls_with_changed_acls) {
-        hmapx_add(&changed_stateful_od, hmapx_node->data);
+        struct ls_stateful_record *ls_stateful_rec =
+            ls_stateful_table_find_(&data->table, od->nbs);
+        ovs_assert(ls_stateful_rec);
+        ls_stateful_rec->has_lb_vip = ls_has_lb_vip(od);
+
+        /* Add the ls_stateful_rec to the tracking data. */
+        hmapx_add(&data->trk_data.crupdated, ls_stateful_rec);
     }
 
-    HMAPX_FOR_EACH (hmapx_node, &changed_stateful_od) {
+    HMAPX_FOR_EACH (hmapx_node, &nd_changes->ls_with_changed_acls) {
         const struct ovn_datapath *od = hmapx_node->data;
 
-        struct ls_stateful_record *ls_stateful_rec = ls_stateful_table_find_(
-            &data->table, od->nbs);
+        struct ls_stateful_record *ls_stateful_rec =
+            ls_stateful_table_find_(&data->table, od->nbs);
         ovs_assert(ls_stateful_rec);
-        ls_stateful_record_reinit(ls_stateful_rec, od, NULL,
-                                  input_data.ls_port_groups);
+        ls_stateful_record_set_acls(ls_stateful_rec, od,
+                                    input_data.ls_port_groups);
 
         /* Add the ls_stateful_rec to the tracking data. */
         hmapx_add(&data->trk_data.crupdated, ls_stateful_rec);
     }
 
-    hmapx_destroy(&changed_stateful_od);
-
     if (ls_stateful_has_tracked_data(&data->trk_data)) {
         return EN_HANDLED_UPDATED;
     }
@@ -189,43 +183,34 @@ ls_stateful_port_group_handler(struct engine_node *node, void *data_)
         return EN_UNHANDLED;
     }
 
-    /* port_group engine node doesn't provide the tracking data yet.
-     * Loop through all the ls port groups and update the ls_stateful_rec.
-     * This is still better than returning false. */
-    struct ls_stateful_input input_data = ls_stateful_get_input_data(node);
     struct ed_type_ls_stateful *data = data_;
-    const struct ls_port_group *ls_pg;
+    if (ls_stateful_has_tracked_data(&data->trk_data)) {
+        return EN_HANDLED_UPDATED;
+    }
+    return EN_HANDLED_UNCHANGED;
+}
 
-    LS_PORT_GROUP_TABLE_FOR_EACH (ls_pg, input_data.ls_port_groups) {
-        struct ls_stateful_record *ls_stateful_rec =
-            ls_stateful_table_find_(&data->table, ls_pg->nbs);
-        ovs_assert(ls_stateful_rec);
-        const struct ovn_datapath *od =
-            ovn_datapaths_find_by_index(input_data.ls_datapaths,
-                                        ls_stateful_rec->ls_index);
-        bool had_stateful_acl = ls_stateful_rec->has_stateful_acl;
-        struct acl_tier old_max = ls_stateful_rec->max_acl_tier;
-        bool had_acls = ls_stateful_rec->has_acls;
-        bool modified = false;
-
-        ls_stateful_record_reinit(ls_stateful_rec, od, ls_pg,
-                                  input_data.ls_port_groups);
-
-        struct acl_tier new_max = ls_stateful_rec->max_acl_tier;
-
-        /* Using memcmp for struct acl_tier is fine since there is no padding
-         * in the struct. However, if the structure is changed, the memcmp
-         * may need to be updated to compare individual struct fields.
-         */
-        if ((had_stateful_acl != ls_stateful_rec->has_stateful_acl)
-            || (had_acls != ls_stateful_rec->has_acls)
-            || memcmp(&old_max, &new_max, sizeof(old_max))) {
-            modified = true;
+enum engine_input_handler_result
+ls_stateful_acl_handler(struct engine_node *node, void *data_)
+{
+    struct ed_type_ls_stateful *data = data_;
+    const struct nbrec_acl_table *nbrec_acl_table =
+        EN_OVSDB_GET(engine_get_input("NB_acl", node));
+
+    const struct nbrec_acl *acl;
+    NBREC_ACL_TABLE_FOR_EACH_TRACKED (acl, nbrec_acl_table) {
+        /* The creation and deletion is handled in relation to LS/PG rather
+         * than the ACL itself. */
+        if (nbrec_acl_is_new(acl) || nbrec_acl_is_deleted(acl)) {
+            continue;
         }
 
-        if (modified) {
-            /* Add the ls_stateful_rec to the tracking data. */
-            hmapx_add(&data->trk_data.crupdated, ls_stateful_rec);
+        struct ls_stateful_record *ls_stateful_rec;
+        LS_STATEFUL_TABLE_FOR_EACH (ls_stateful_rec, &data->table) {
+            if (uuidset_contains(&ls_stateful_rec->related_acls,
+                                 &acl->header_.uuid)) {
+                hmapx_add(&data->trk_data.crupdated, ls_stateful_rec);
+            }
         }
     }
 
@@ -295,7 +280,8 @@ ls_stateful_record_create(struct ls_stateful_table *table,
         xzalloc(sizeof *ls_stateful_rec);
     ls_stateful_rec->ls_index = od->index;
     ls_stateful_rec->nbs_uuid = od->nbs->header_.uuid;
-    ls_stateful_record_init(ls_stateful_rec, od, NULL, ls_pgs);
+    uuidset_init(&ls_stateful_rec->related_acls);
+    ls_stateful_record_init(ls_stateful_rec, od, ls_pgs);
     ls_stateful_rec->lflow_ref = lflow_ref_create();
 
     hmap_insert(&table->entries, &ls_stateful_rec->key_node,
@@ -307,27 +293,18 @@ ls_stateful_record_create(struct ls_stateful_table *table,
 static void
 ls_stateful_record_destroy(struct ls_stateful_record *ls_stateful_rec)
 {
+    uuidset_destroy(&ls_stateful_rec->related_acls);
     lflow_ref_destroy(ls_stateful_rec->lflow_ref);
     free(ls_stateful_rec);
 }
 
 static void
 ls_stateful_record_init(struct ls_stateful_record *ls_stateful_rec,
                       const struct ovn_datapath *od,
-                      const struct ls_port_group *ls_pg,
                       const struct ls_port_group_table *ls_pgs)
 {
     ls_stateful_rec->has_lb_vip = ls_has_lb_vip(od);
-    ls_stateful_record_set_acl_flags(ls_stateful_rec, od, ls_pg, ls_pgs);
-}
-
-static void
-ls_stateful_record_reinit(struct ls_stateful_record *ls_stateful_rec,
-                          const struct ovn_datapath *od,
-                          const struct ls_port_group *ls_pg,
-                          const struct ls_port_group_table *ls_pgs)
-{
-    ls_stateful_record_init(ls_stateful_rec, od, ls_pg, ls_pgs);
+    ls_stateful_record_set_acls(ls_stateful_rec, od, ls_pgs);
 }
 
 static bool
@@ -365,36 +342,28 @@ ls_has_lb_vip(const struct ovn_datapath *od)
 }
 
 static void
-ls_stateful_record_set_acl_flags(struct ls_stateful_record *ls_stateful_rec,
-                                 const struct ovn_datapath *od,
-                                 const struct ls_port_group *ls_pg,
-                                 const struct ls_port_group_table *ls_pgs)
+ls_stateful_record_set_acls(struct ls_stateful_record *ls_stateful_rec,
+                            const struct ovn_datapath *od,
+                            const struct ls_port_group_table *ls_pgs)
 {
     ls_stateful_rec->has_stateful_acl = false;
     memset(&ls_stateful_rec->max_acl_tier, 0,
            sizeof ls_stateful_rec->max_acl_tier);
     ls_stateful_rec->has_acls = false;
+    uuidset_clear(&ls_stateful_rec->related_acls);
 
-    if (ls_stateful_record_set_acl_flags_(ls_stateful_rec, od->nbs->acls,
-                                          od->nbs->n_acls)) {
-        return;
-    }
-
-    if (!ls_pg) {
-        ls_pg = ls_port_group_table_find(ls_pgs, od->nbs);
-    }
+    ls_stateful_record_set_acls_(ls_stateful_rec, od->nbs->acls,
+                                 od->nbs->n_acls);
 
+    struct ls_port_group *ls_pg = ls_port_group_table_find(ls_pgs, od->nbs);
     if (!ls_pg) {
         return;
     }
 
     const struct ls_port_group_record *ls_pg_rec;
     HMAP_FOR_EACH (ls_pg_rec, key_node, &ls_pg->nb_pgs) {
-        if (ls_stateful_record_set_acl_flags_(ls_stateful_rec,
-                                              ls_pg_rec->nb_pg->acls,
-                                              ls_pg_rec->nb_pg->n_acls)) {
-            return;
-        }
+        ls_stateful_record_set_acls_(ls_stateful_rec, ls_pg_rec->nb_pg->acls,
+                                     ls_pg_rec->nb_pg->n_acls);
     }
 }
 
@@ -421,46 +390,24 @@ update_ls_max_acl_tier(struct ls_stateful_record *ls_stateful_rec,
     *tier = MAX(*tier, acl->tier);
 }
 
-static bool
-ls_acl_tiers_are_maxed_out(struct acl_tier *acl_tier,
-                           uint64_t max_allowed_acl_tier)
-{
-    return acl_tier->ingress_post_lb == max_allowed_acl_tier &&
-           acl_tier->ingress_pre_lb == max_allowed_acl_tier &&
-           acl_tier->egress == max_allowed_acl_tier;
-}
-
-static bool
-ls_stateful_record_set_acl_flags_(struct ls_stateful_record *ls_stateful_rec,
-                                  struct nbrec_acl **acls,
-                                  size_t n_acls)
+static void
+ls_stateful_record_set_acls_(struct ls_stateful_record *ls_stateful_rec,
+                             struct nbrec_acl **acls, size_t n_acls)
 {
-    /* A true return indicates that there are no possible ACL flags
-     * left to set on ls_stateful record. A false return indicates that
-     * further ACLs should be explored in case more flags need to be
-     * set on ls_stateful record.
-     */
     if (!n_acls) {
-        return false;
+        return;
     }
 
     ls_stateful_rec->has_acls = true;
     for (size_t i = 0; i < n_acls; i++) {
         const struct nbrec_acl *acl = acls[i];
         update_ls_max_acl_tier(ls_stateful_rec, acl);
+        uuidset_insert(&ls_stateful_rec->related_acls, &acl->header_.uuid);
         if (!ls_stateful_rec->has_stateful_acl
                 && !strcmp(acl->action, "allow-related")) {
             ls_stateful_rec->has_stateful_acl = true;
         }
-        if (ls_stateful_rec->has_stateful_acl &&
-            ls_acl_tiers_are_maxed_out(
-                &ls_stateful_rec->max_acl_tier,
-                nbrec_acl_col_tier.type.key.integer.max)) {
-            return true;
-        }
     }
-
-    return false;
 }
 
 static struct ls_stateful_input

northd/en-ls-stateful.h

@@ -20,6 +20,7 @@
 
 /* OVS includes. */
 #include "lib/hmapx.h"
+#include "lib/uuidset.h"
 #include "openvswitch/hmap.h"
 #include "sset.h"
 
@@ -54,6 +55,9 @@ struct ls_stateful_record {
     bool has_acls;
     struct acl_tier max_acl_tier;
 
+    /* Set of ACLs that are related to this LS. */
+    struct uuidset related_acls;
+
     /* 'lflow_ref' is used to reference logical flows generated for
      * this ls_stateful record.
      *
@@ -109,6 +113,8 @@ enum engine_input_handler_result
 ls_stateful_northd_handler(struct engine_node *, void *data);
 enum engine_input_handler_result
 ls_stateful_port_group_handler(struct engine_node *, void *data);
+enum engine_input_handler_result
+ls_stateful_acl_handler(struct engine_node *node, void *data);
 
 const struct ls_stateful_record *ls_stateful_table_find(
     const struct ls_stateful_table *, const struct nbrec_logical_switch *);

northd/inc-proc-northd.c

@@ -259,6 +259,7 @@ void inc_proc_northd_init(struct ovsdb_idl_loop *nb,
     engine_add_input(&en_ls_stateful, &en_northd, ls_stateful_northd_handler);
     engine_add_input(&en_ls_stateful, &en_port_group,
                      ls_stateful_port_group_handler);
+    engine_add_input(&en_ls_stateful, &en_nb_acl, ls_stateful_acl_handler);
 
     engine_add_input(&en_mac_binding_aging, &en_sb_mac_binding, NULL);
     engine_add_input(&en_mac_binding_aging, &en_northd, NULL);
@@ -330,7 +331,6 @@ void inc_proc_northd_init(struct ovsdb_idl_loop *nb,
     engine_add_input(&en_multicast_igmp, &en_sb_multicast_group, NULL);
     engine_add_input(&en_multicast_igmp, &en_sb_igmp_group, NULL);
 
-    engine_add_input(&en_lflow, &en_nb_acl, NULL);
     engine_add_input(&en_lflow, &en_sync_meters, NULL);
     engine_add_input(&en_lflow, &en_sb_logical_flow, NULL);
     engine_add_input(&en_lflow, &en_sb_multicast_group, NULL);