northd: Add lr_stateful handler for lflow engine node.

Acked-by: Dumitru Ceara <dceara@redhat.com>
Acked-by: Han Zhou <hzhou@ovn.org>
Signed-off-by: Numan Siddique <numans@ovn.org>
(cherry picked from commit e5ae7a9)

northd/en-lflow.c

@@ -163,6 +163,33 @@ lflow_port_group_handler(struct engine_node *node, void *data OVS_UNUSED)
     return true;
 }
 
+bool
+lflow_lr_stateful_handler(struct engine_node *node, void *data)
+{
+    struct ed_type_lr_stateful *lr_sful_data =
+        engine_get_input_data("lr_stateful", node);
+
+    if (!lr_stateful_has_tracked_data(&lr_sful_data->trk_data)
+        || lr_sful_data->trk_data.vip_nats_changed) {
+        return false;
+    }
+
+    const struct engine_context *eng_ctx = engine_get_context();
+    struct lflow_data *lflow_data = data;
+    struct lflow_input lflow_input;
+
+    lflow_get_input_data(node, &lflow_input);
+    if (!lflow_handle_lr_stateful_changes(eng_ctx->ovnsb_idl_txn,
+                                          &lr_sful_data->trk_data,
+                                          &lflow_input,
+                                          lflow_data->lflow_table)) {
+        return false;
+    }
+
+    engine_set_node_state(node, EN_UPDATED);
+    return true;
+}
+
 void *en_lflow_init(struct engine_node *node OVS_UNUSED,
                      struct engine_arg *arg OVS_UNUSED)
 {

northd/en-lflow.h

@@ -20,5 +20,6 @@ void *en_lflow_init(struct engine_node *node, struct engine_arg *arg);
 void en_lflow_cleanup(void *data);
 bool lflow_northd_handler(struct engine_node *, void *data);
 bool lflow_port_group_handler(struct engine_node *, void *data);
+bool lflow_lr_stateful_handler(struct engine_node *, void *data);
 
 #endif /* EN_LFLOW_H */

northd/en-lr-stateful.c

@@ -40,6 +40,7 @@
 #include "lib/ovn-sb-idl.h"
 #include "lib/ovn-util.h"
 #include "lib/stopwatch-names.h"
+#include "lflow-mgr.h"
 #include "northd.h"
 
 VLOG_DEFINE_THIS_MODULE(en_lr_stateful);
@@ -81,7 +82,7 @@ static void remove_lrouter_lb_reachable_ips(struct lr_stateful_record *,
                                             enum lb_neighbor_responder_mode,
                                             const struct sset *lb_ips_v4,
                                             const struct sset *lb_ips_v6);
-static void lr_stateful_rebuild_vip_nats(struct lr_stateful_record *);
+static bool lr_stateful_rebuild_vip_nats(struct lr_stateful_record *);
 
 /* 'lr_stateful' engine node manages the NB logical router LB data.
  */
@@ -109,6 +110,7 @@ en_lr_stateful_clear_tracked_data(void *data_)
     struct ed_type_lr_stateful *data = data_;
 
     hmapx_clear(&data->trk_data.crupdated);
+    data->trk_data.vip_nats_changed = false;
 }
 
 void
@@ -196,6 +198,10 @@ lr_stateful_lb_data_handler(struct engine_node *node, void *data_)
 
             /* Add the lr_stateful_rec rec to the tracking data. */
             hmapx_add(&data->trk_data.crupdated, lr_stateful_rec);
+
+            if (!sset_is_empty(&lr_stateful_rec->vip_nats)) {
+                data->trk_data.vip_nats_changed = true;
+            }
             continue;
         }
 
@@ -313,7 +319,9 @@ lr_stateful_lb_data_handler(struct engine_node *node, void *data_)
 
         HMAPX_FOR_EACH (hmapx_node, &data->trk_data.crupdated) {
             struct lr_stateful_record *lr_stateful_rec = hmapx_node->data;
-            lr_stateful_rebuild_vip_nats(lr_stateful_rec);
+            if (lr_stateful_rebuild_vip_nats(lr_stateful_rec)) {
+                data->trk_data.vip_nats_changed = true;
+            }
             const struct ovn_datapath *od =
                 ovn_datapaths_find_by_index(input_data.lr_datapaths,
                                             lr_stateful_rec->lr_index);
@@ -352,8 +360,13 @@ lr_stateful_lr_nat_handler(struct engine_node *node, void *data_)
                 lr_stateful_record_create(&data->table, lrnat_rec, od,
                                           input_data.lb_datapaths_map,
                                           input_data.lbgrp_datapaths_map);
+            if (!sset_is_empty(&lr_stateful_rec->vip_nats)) {
+                data->trk_data.vip_nats_changed = true;
+            }
         } else {
-            lr_stateful_rebuild_vip_nats(lr_stateful_rec);
+            if (lr_stateful_rebuild_vip_nats(lr_stateful_rec)) {
+                data->trk_data.vip_nats_changed = true;
+            }
         }
 
         /* Add the lr_stateful_rec rec to the tracking data. */
@@ -455,6 +468,7 @@ lr_stateful_record_create(struct lr_stateful_table *table,
     lr_stateful_rec->nbr_uuid = od->nbr->header_.uuid;
     lr_stateful_rec->lrnat_rec = lrnat_rec;
     lr_stateful_rec->lr_index = od->index;
+    lr_stateful_rec->lflow_ref = lflow_ref_create();
 
 
     /* Initialize the record.
@@ -543,6 +557,7 @@ lr_stateful_record_destroy(struct lr_stateful_record *lr_stateful_rec)
 {
     ovn_lb_ip_set_destroy(lr_stateful_rec->lb_ips);
     sset_destroy(&lr_stateful_rec->vip_nats);
+    lflow_ref_destroy(lr_stateful_rec->lflow_ref);
     free(lr_stateful_rec);
 }
 
@@ -657,10 +672,12 @@ remove_lrouter_lb_reachable_ips(struct lr_stateful_record *lr_stateful_rec,
     }
 }
 
-static void
+static bool
 lr_stateful_rebuild_vip_nats(struct lr_stateful_record *lr_stateful_rec)
 {
-    sset_clear(&lr_stateful_rec->vip_nats);
+    struct sset old_vip_nats = SSET_INITIALIZER(&old_vip_nats);
+    sset_swap(&lr_stateful_rec->vip_nats, &old_vip_nats);
+
     const char *external_ip;
     SSET_FOR_EACH (external_ip, &lr_stateful_rec->lrnat_rec->external_ips) {
         bool is_vip_nat = false;
@@ -676,4 +693,10 @@ lr_stateful_rebuild_vip_nats(struct lr_stateful_record *lr_stateful_rec)
             sset_add(&lr_stateful_rec->vip_nats, external_ip);
         }
     }
+
+    bool vip_nats_changed = !sset_equals(&lr_stateful_rec->vip_nats,
+                                         &old_vip_nats);
+    sset_destroy(&old_vip_nats);
+
+    return vip_nats_changed;
 }

northd/en-lr-stateful.h

@@ -32,6 +32,7 @@
 
 struct ovn_datapath;
 struct lr_nat_record;
+struct lflow_ref;
 
 /* lr_stateful_table:  This represents a table of logical routers with
  *                     stateful related data.
@@ -63,6 +64,30 @@ struct lr_stateful_record {
 
     /* sset of vips which are also part of lr nats. */
     struct sset vip_nats;
+
+    /* 'lflow_ref' is used to reference logical flows generated for
+     * this lr_stateful_record.
+     *
+     * This data is initialized and destroyed by the en_lr_stateful node,
+     * but populated and used only by the en_lflow node. Ideally this data
+     * should be maintained as part of en_lflow's data.  However, it would
+     * be less efficient and more complex:
+     *
+     * 1. It would require an extra search (using the index) to find the
+     * lflows.
+     *
+     * 2. Building the index needs to be thread-safe, using either a global
+     * lock which is obviously less efficient, or hash-based lock array which
+     * is more complex.
+     *
+     * Adding the lflow_ref here is more straightforward. The drawback is that
+     * we need to keep in mind that this data belongs to en_lflow node, so
+     * never access it from any other nodes.
+     *
+     * Note: lflow_ref is not thread safe.  Only one thread should
+     * access lr_stateful_record->lflow_ref at any given time.
+     */
+    struct lflow_ref *lflow_ref;
 };
 
 struct lr_stateful_table {
@@ -82,6 +107,10 @@ struct lr_stateful_table {
 struct lr_stateful_tracked_data {
     /* Created or updated logical router with LB and/or NAT data. */
     struct hmapx crupdated; /* Stores 'struct lr_stateful_record'. */
+
+    /* Indicates if any router's NATs changed which were also LB vips
+     * or vice versa. */
+    bool vip_nats_changed;
 };
 
 struct ed_type_lr_stateful {
@@ -113,7 +142,7 @@ const struct lr_stateful_record *lr_stateful_table_find_by_index(
 static inline bool
 lr_stateful_has_tracked_data(struct lr_stateful_tracked_data *trk_data)
 {
-    return !hmapx_is_empty(&trk_data->crupdated);
+    return !hmapx_is_empty(&trk_data->crupdated) || trk_data->vip_nats_changed;
 }
 
 static inline bool

northd/inc-proc-northd.c

@@ -228,11 +228,11 @@ void inc_proc_northd_init(struct ovsdb_idl_loop *nb,
     engine_add_input(&en_lflow, &en_sb_logical_flow, NULL);
     engine_add_input(&en_lflow, &en_sb_multicast_group, NULL);
     engine_add_input(&en_lflow, &en_sb_igmp_group, NULL);
-    engine_add_input(&en_lflow, &en_lr_stateful, NULL);
     engine_add_input(&en_lflow, &en_ls_stateful, NULL);
     engine_add_input(&en_lflow, &en_sb_logical_dp_group, NULL);
     engine_add_input(&en_lflow, &en_northd, lflow_northd_handler);
     engine_add_input(&en_lflow, &en_port_group, lflow_port_group_handler);
+    engine_add_input(&en_lflow, &en_lr_stateful, lflow_lr_stateful_handler);
 
     engine_add_input(&en_sync_to_sb_addr_set, &en_nb_address_set,
                      sync_to_sb_addr_set_nb_address_set_handler);