flowprobe: fix corrupted packets sent after feature disabling

When IPFIX flow record generation is enabled on an interface and the
active timer is set, flows will be saved and then exported according to
the active and passive timers. If then disable the feature on the
interface, the flow entries currently saved will remain in the state
tables. They will gradually expire and be exported. The problem is that
the template for them has already been removed. And they will be sent
with zero template ID which will make them unreadable.

A similar problem will occur if feature settings are "changed" on the
interface - i.e. disable the feature and re-enable it with different
settings (e.g. set a different datapath). The remaining flows that
correspond to the previous feature settings will be eventually sent
either with zero template ID or with template ID that corresponds to the
current feature settings on the interface (and look like garbage data).

With this fix, flush the current buffers before template removal and
clear the remaining flows of the interface during feature disabling.

Type: fix
Change-Id: I1e57db06adfdd3a02fed1a6a89b5418f85a35e16
Signed-off-by: Alexander Chernavin <achernavin@netgate.com>

4 files changed, 92 insertions, 2 deletions

diff --git a/src/plugins/flowprobe/flowprobe.c b/src/plugins/flowprobe/flowprobe.c
index 3b060bec529..fb05158576e 100644
--- a/src/plugins/flowprobe/flowprobe.c
+++ b/src/plugins/flowprobe/flowprobe.c
@@ -503,6 +503,40 @@ flowprobe_create_state_tables (u32 active_timer)
   return error;
 }
 
+static clib_error_t *
+flowprobe_clear_state_if_index (u32 sw_if_index)
+{
+  flowprobe_main_t *fm = &flowprobe_main;
+  clib_error_t *error = 0;
+  u32 worker_i;
+  u32 entry_i;
+
+  if (fm->active_timer > 0)
+    {
+      vec_foreach_index (worker_i, fm->pool_per_worker)
+	{
+	  pool_foreach_index (entry_i, fm->pool_per_worker[worker_i])
+	    {
+	      flowprobe_entry_t *e =
+		pool_elt_at_index (fm->pool_per_worker[worker_i], entry_i);
+	      if (e->key.rx_sw_if_index == sw_if_index ||
+		  e->key.tx_sw_if_index == sw_if_index)
+		{
+		  if (fm->passive_timer > 0)
+		    {
+		      tw_timer_stop_2t_1w_2048sl (
+			fm->timers_per_worker[worker_i],
+			e->passive_timer_handle);
+		    }
+		  flowprobe_delete_by_index (worker_i, entry_i);
+		}
+	    }
+	}
+    }
+
+  return error;
+}
+
 static int
 validate_feature_on_interface (flowprobe_main_t * fm, u32 sw_if_index,
 			       u8 which)
@@ -548,6 +582,10 @@ flowprobe_interface_add_del_feature (flowprobe_main_t *fm, u32 sw_if_index,
     {
       if (which == FLOW_VARIANT_L2)
 	{
+	  if (!is_add)
+	    {
+	      flowprobe_flush_callback_l2 ();
+	    }
 	  if (fm->record & FLOW_RECORD_L2)
 	    {
 	      rv = flowprobe_template_add_del (1, UDP_DST_PORT_ipfix, flags,
@@ -581,6 +619,10 @@ flowprobe_interface_add_del_feature (flowprobe_main_t *fm, u32 sw_if_index,
 	}
       else if (which == FLOW_VARIANT_IP4)
 	{
+	  if (!is_add)
+	    {
+	      flowprobe_flush_callback_ip4 ();
+	    }
 	  rv = flowprobe_template_add_del (
 	    1, UDP_DST_PORT_ipfix, flags, flowprobe_data_callback_ip4,
 	    flowprobe_template_rewrite_ip4, is_add, &template_id);
@@ -588,6 +630,10 @@ flowprobe_interface_add_del_feature (flowprobe_main_t *fm, u32 sw_if_index,
 	}
       else if (which == FLOW_VARIANT_IP6)
 	{
+	  if (!is_add)
+	    {
+	      flowprobe_flush_callback_ip6 ();
+	    }
 	  rv = flowprobe_template_add_del (
 	    1, UDP_DST_PORT_ipfix, flags, flowprobe_data_callback_ip6,
 	    flowprobe_template_rewrite_ip6, is_add, &template_id);
@@ -647,6 +693,11 @@ flowprobe_interface_add_del_feature (flowprobe_main_t *fm, u32 sw_if_index,
 	vlib_process_signal_event (vm, flowprobe_timer_node.index, 1, 0);
     }
 
+  if (!is_add && fm->initialized)
+    {
+      flowprobe_clear_state_if_index (sw_if_index);
+    }
+
   return 0;
 }
 
diff --git a/src/plugins/flowprobe/flowprobe.h b/src/plugins/flowprobe/flowprobe.h
index 3174a844c2a..0a70303b470 100644
--- a/src/plugins/flowprobe/flowprobe.h
+++ b/src/plugins/flowprobe/flowprobe.h
@@ -168,6 +168,8 @@ typedef struct
 extern flowprobe_main_t flowprobe_main;
 extern vlib_node_registration_t flowprobe_walker_node;
 
+void flowprobe_delete_by_index (u32 my_cpu_number, u32 poolindex);
+
 void flowprobe_flush_callback_ip4 (void);
 void flowprobe_flush_callback_ip6 (void);
 void flowprobe_flush_callback_l2 (void);
diff --git a/src/plugins/flowprobe/node.c b/src/plugins/flowprobe/node.c
index e9fa4b89665..44096d6245f 100644
--- a/src/plugins/flowprobe/node.c
+++ b/src/plugins/flowprobe/node.c
@@ -955,8 +955,7 @@ flowprobe_flush_callback_l2 (void)
   flush_record (FLOW_VARIANT_L2_IP6);
 }
 
-
-static void
+void
 flowprobe_delete_by_index (u32 my_cpu_number, u32 poolindex)
 {
   flowprobe_main_t *fm = &flowprobe_main;
diff --git a/test/test_flowprobe.py b/test/test_flowprobe.py
index 19571f71235..0bdbd3d6789 100644
--- a/test/test_flowprobe.py
+++ b/test/test_flowprobe.py
@@ -1291,6 +1291,44 @@ class DisableFP(MethodHolder):
         ipfix.remove_vpp_config()
         self.logger.info("FFP_TEST_FINISH_0001")
 
+    def test_no_leftover_flows_after_disabling(self):
+        """disable flowprobe feature and expect no leftover flows"""
+        self.pg_enable_capture(self.pg_interfaces)
+        self.pkts = []
+
+        # enable ip4 datapath for an interface
+        # set active and passive timers
+        ipfix = VppCFLOW(
+            test=self,
+            active=3,
+            passive=4,
+            intf="pg3",
+            layer="l3",
+            datapath="ip4",
+            direction="rx",
+            mtu=100,
+        )
+        ipfix.add_vpp_config()
+
+        # template packet should arrive immediately
+        ipfix.verify_templates(count=1)
+
+        # send some ip4 packets
+        self.create_stream(src_if=self.pg3, dst_if=self.pg4, packets=5)
+        self.send_packets(src_if=self.pg3, dst_if=self.pg4)
+
+        # disable feature for the interface
+        # currently stored ip4 flows should be removed
+        ipfix.disable_flowprobe_feature()
+
+        # no leftover ip4 flows are expected
+        self.pg_enable_capture([self.collector])
+        self.sleep(12, "wait for leftover ip4 flows during three passive intervals")
+        self.collector.assert_nothing_captured()
+
+        # cleanup
+        ipfix.disable_exporter()
+
 
 @unittest.skipUnless(config.extended, "part of extended tests")
 class ReenableFP(MethodHolder):