[dualtor_io] Allow duplications for link down downstream I/O (#17909)

What is the motivation for this PR?
The following two link failure cases are failing on Cisco/MLNX:

test_active_link_down_downstream_active
test_active_link_down_downstream_active_soc
The reason is that, after link down, between the fdb flush and tunnel route add (due to mux toggle-to-standby), the ASIC has no l2 information for server/soc neighbors, downstream traffic will flood to all vlan member ports on Cisco/MLNX platform.
Those two testcase has no tolerance for packet duplications due to that, on Broadcom platform, traffic to neighbors with no l2 information will be simply dropped.

Let's adapt to Cisco/MLNX platform, by allowing packet duplications for those two testcases.

How did you verify/test it?
dualtor_io/test_link_failure.py::test_active_link_down_downstream_active[active-active] PASSED                                                                                                                                                                         [100%]
dualtor_io/test_link_failure.py::test_active_link_down_downstream_active_soc[active-active] PASSED                                                                                                                                                                     [100%]

Signed-off-by: Longxiang Lyu <lolv@microsoft.com>

Author: lolyu

tests/common/dualtor/data_plane_utils.py

@@ -41,7 +41,8 @@ def arp_setup(ptfhost):
 
 
 def validate_traffic_results(tor_IO, allowed_disruption, delay,
-                             allow_disruption_before_traffic=False):
+                             allow_disruption_before_traffic=False,
+                             allowed_duplication=None):
     """
     Generates a report (dictionary) of I/O metrics that were calculated as part
     of the dataplane test. This report is to be used by testcases to verify the
@@ -107,7 +108,12 @@ def validate_traffic_results(tor_IO, allowed_disruption, delay,
                             "Maximum allowed disruption: {}s"
                             .format(server_ip, longest_disruption, delay))
 
-        if total_duplications > allowed_disruption:
+        # NOTE: Not all testcases set the allowed duplication threshold and the duplication check
+        # uses the allowed disruption threshold here.q So let's set the allowed duplication to
+        # allowed disruption if the allowed duplication is provided here.
+        if allowed_duplication is None:
+            allowed_duplication = allowed_disruption
+        if total_duplications > allowed_duplication:
             failures.append("Traffic to server {} was duplicated {} times. "
                             "Allowed number of duplications: {}"
                             .format(server_ip, total_duplications, allowed_disruption))
@@ -150,11 +156,12 @@ def _validate_long_disruption(disruptions, allowed_disruption, delay):
 
 
 def verify_and_report(tor_IO, verify, delay, allowed_disruption,
-                      allow_disruption_before_traffic=False):
+                      allow_disruption_before_traffic=False, allowed_duplication=None):
     # Wait for the IO to complete before doing checks
     if verify:
         validate_traffic_results(tor_IO, allowed_disruption=allowed_disruption, delay=delay,
-                                 allow_disruption_before_traffic=allow_disruption_before_traffic)
+                                 allow_disruption_before_traffic=allow_disruption_before_traffic,
+                                 allowed_duplication=allowed_duplication)
     return tor_IO.get_test_results()
 
 
@@ -267,7 +274,8 @@ def send_t1_to_server_with_action(duthosts, ptfhost, ptfadapter, tbinfo,
 
     def t1_to_server_io_test(activehost, tor_vlan_port=None,
                              delay=0, allowed_disruption=0, action=None, verify=False, send_interval=0.1,
-                             stop_after=None, allow_disruption_before_traffic=False):
+                             stop_after=None, allow_disruption_before_traffic=False,
+                             allowed_duplication=None):
         """
         Helper method for `send_t1_to_server_with_action`.
         Starts sender and sniffer before performing the action on the tor host.
@@ -302,7 +310,8 @@ def t1_to_server_io_test(activehost, tor_vlan_port=None,
         if delay and not allowed_disruption:
             allowed_disruption = 1
 
-        return verify_and_report(tor_IO, verify, delay, allowed_disruption, allow_disruption_before_traffic)
+        return verify_and_report(tor_IO, verify, delay, allowed_disruption, allow_disruption_before_traffic,
+                                 allowed_duplication=allowed_duplication)
 
     yield t1_to_server_io_test
 
@@ -416,7 +425,7 @@ def send_t1_to_soc_with_action(duthosts, ptfhost, ptfadapter, tbinfo,
 
     def t1_to_soc_io_test(activehost, tor_vlan_port=None,
                           delay=0, allowed_disruption=0, action=None, verify=False, send_interval=0.01,
-                          stop_after=None):
+                          stop_after=None, allowed_duplication=None):
 
         tor_IO = run_test(duthosts, activehost, ptfhost, ptfadapter, vmhost,
                           action, tbinfo, tor_vlan_port, send_interval,
@@ -432,7 +441,8 @@ def t1_to_soc_io_test(activehost, tor_vlan_port=None,
         if asic_type == "vs":
             logging.info("Skipping verify on VS platform")
             return
-        return verify_and_report(tor_IO, verify, delay, allowed_disruption)
+        return verify_and_report(tor_IO, verify, delay, allowed_disruption,
+                                 allowed_duplication=allowed_duplication)
 
     yield t1_to_soc_io_test
 

tests/common/dualtor/dual_tor_io.py

@@ -11,7 +11,6 @@
 import six
 import scapy.all as scapyall
 import ptf.testutils as testutils
-from operator import itemgetter
 from itertools import groupby
 
 from tests.common.dualtor.dual_tor_common import CableType
@@ -793,16 +792,37 @@ def examine_each_packet(self, server_ip, packets):
             logger.error("Sniffer failed to filter any traffic from DUT")
         else:
             # Find ranges of consecutive packets that have been duplicated
-            # All packets within the same consecutive range will have the same
-            # difference between the packet index and the sequence number
-            for _, grouper in groupby(enumerate(duplicate_packet_list), lambda t: t[0] - t[1][0]):
-                group = list(map(itemgetter(1), grouper))
-                duplicate_start, duplicate_end = group[0], group[-1]
+            # All consecutive packets with the same payload will be grouped as one
+            # duplication group.
+            # For example, for the duplication list as the following:
+            # [(70, 1744253633.499116), (70, 1744253633.499151), (70, 1744253633.499186),
+            #  (81, 1744253635.49922), (81, 1744253635.499255)]
+            # two duplications will be reported:
+            # "duplications": [
+            #     {
+            #         "start_time": 1744253633.499116,
+            #         "end_time": 1744253633.499186,
+            #         "start_id": 70,
+            #         "end_id": 70,
+            #         "duplication_count": 3
+            #     },
+            #     {
+            #         "start_time": 1744253635.49922,
+            #         "end_time": 1744253635.499255,
+            #         "start_id": 81,
+            #         "end_id": 81,
+            #         "duplication_count": 2
+            #     }
+            # ]
+            for _, grouper in groupby(duplicate_packet_list, lambda d: d[0]):
+                duplicates = list(grouper)
+                duplicate_start, duplicate_end = duplicates[0], duplicates[-1]
                 duplicate_dict = {
                     'start_time': duplicate_start[1],
                     'end_time': duplicate_end[1],
                     'start_id': duplicate_start[0],
-                    'end_id': duplicate_end[0]
+                    'end_id': duplicate_end[0],
+                    'duplication_count': len(duplicates)
                 }
                 duplicate_ranges.append(duplicate_dict)
 

tests/dualtor_io/test_link_failure.py

@@ -84,7 +84,8 @@ def test_active_link_down_downstream_active(
     if cable_type == CableType.active_active:
         send_t1_to_server_with_action(
             upper_tor_host, verify=True, delay=MUX_SIM_ALLOWED_DISRUPTION_SEC,
-            allowed_disruption=1, action=shutdown_fanout_upper_tor_intfs
+            allowed_disruption=1, allowed_duplication=1,
+            action=shutdown_fanout_upper_tor_intfs
         )
         verify_tor_states(
             expected_active_host=lower_tor_host,
@@ -332,7 +333,8 @@ def test_active_link_down_downstream_active_soc(
     if cable_type == CableType.active_active:
         send_t1_to_soc_with_action(
             upper_tor_host, verify=True, delay=MUX_SIM_ALLOWED_DISRUPTION_SEC,
-            allowed_disruption=1, action=shutdown_fanout_upper_tor_intfs
+            allowed_disruption=1, allowed_duplication=1,
+            action=shutdown_fanout_upper_tor_intfs
         )
         verify_tor_states(
             expected_active_host=lower_tor_host,