[testbed] ptf data plane connection for multi-servers testbed (#15881)

What is the motivation for this PR?
We need to make sure ptf data plane connectivity during tests running.

How did you do it?
Start ptf_nn_agent on every ptf container and connect them with ptf dataplane.

How did you verify/test it?
Run test case with ptf dataplane on single server testbed and multi-servers testbed.

Author: w1nda

tests/common/multi_servers_utils.py

@@ -0,0 +1 @@
+../../ansible/module_utils/multi_servers_utils.py

tests/common/plugins/ptfadapter/__init__.py

@@ -3,7 +3,7 @@
 import pytest
 import time
 
-from .ptfadapter import PtfTestAdapter
+from .ptfadapter import PtfTestAdapter, PtfAgent
 import ptf.testutils
 
 from tests.common import constants
@@ -112,7 +112,7 @@ def get_ifaces_map(ifaces, ptf_port_mapping_mode, need_backplane=False):
 
 
 @pytest.fixture(scope='module')
-def ptfadapter(ptfhost, tbinfo, request, duthost):
+def ptfadapter(ptfhosts, tbinfo, request, duthost):
     """return ptf test adapter object.
     The fixture is module scope, because usually there is not need to
     restart PTF nn agent and reinitialize data plane thread on every
@@ -127,22 +127,13 @@ def ptfadapter(ptfhost, tbinfo, request, duthost):
     else:
         ptf_port_mapping_mode = 'use_orig_interface'
 
-    need_backplane = False
-    if 'ciscovs-7nodes' in tbinfo['topo']['name']:
-        need_backplane = True
-
-    # get the eth interfaces from PTF and initialize ifaces_map
-    res = ptfhost.command('cat /proc/net/dev')
-    ifaces = get_ifaces(res['stdout'])
-    ifaces_map = get_ifaces_map(ifaces, ptf_port_mapping_mode, need_backplane)
-
-    def start_ptf_nn_agent():
+    def start_ptf_nn_agent(device_num):
         for i in range(MAX_RETRY_TIME):
             ptf_nn_port = random.randint(*DEFAULT_PTF_NN_PORT_RANGE)
 
             # generate supervisor configuration for ptf_nn_agent
             ptfhost.host.options['variable_manager'].extra_vars.update({
-                'device_num': DEFAULT_DEVICE_NUM,
+                'device_num': device_num,
                 'ptf_nn_port': ptf_nn_port,
                 'ifaces_map': ifaces_map,
             })
@@ -163,8 +154,17 @@ def start_ptf_nn_agent():
                 return ptf_nn_port
         return None
 
-    ptf_nn_agent_port = start_ptf_nn_agent()
-    assert ptf_nn_agent_port is not None
+    need_backplane = False
+    if 'ciscovs-7nodes' in tbinfo['topo']['name']:
+        need_backplane = True
+    ptfagents = []
+    for seq, ptfhost in enumerate(ptfhosts):
+        res = ptfhost.command('cat /proc/net/dev')
+        ifaces = get_ifaces(res['stdout'])
+        ifaces_map = get_ifaces_map(ifaces, ptf_port_mapping_mode, need_backplane)
+        ptf_nn_agent_port = start_ptf_nn_agent(seq)
+        ptfagents.append(PtfAgent(ptfhost.mgmt_ip, ptfhost.mgmt_ipv6, ptf_nn_agent_port, seq, ifaces_map))
+        assert ptf_nn_agent_port is not None
 
     def check_if_use_minigraph_from_tbinfo(tbinfo):
         if 'properties' in tbinfo['topo'] and "init_cfg_profile" in tbinfo['topo']['properties']:
@@ -174,8 +174,7 @@ def check_if_use_minigraph_from_tbinfo(tbinfo):
             return False
         return True
 
-    with PtfTestAdapter(tbinfo['ptf_ip'], tbinfo['ptf_ipv6'],
-                        ptf_nn_agent_port, 0, list(ifaces_map.keys()), ptfhost) as adapter:
+    with PtfTestAdapter(ptfagents, ptfhosts) as adapter:
         if not request.config.option.keep_payload:
             override_ptf_functions()
             node_id = request.module.__name__
@@ -189,12 +188,12 @@ def check_if_use_minigraph_from_tbinfo(tbinfo):
 
 
 @pytest.fixture(scope='module')
-def nbr_device_numbers(nbrhosts):
+def nbr_device_numbers(nbrhosts, ptfhosts):
     """return the mapping of neighbor devices name to ptf device number.
     """
     numbers = sorted(nbrhosts.keys())
     device_numbers = {
-        nbr_name: numbers.index(nbr_name) + DEFAULT_DEVICE_NUM + 1
+        nbr_name: numbers.index(nbr_name) + len(ptfhosts)
         for nbr_name in list(nbrhosts.keys())}
     return device_numbers
 

tests/common/plugins/ptfadapter/ptfadapter.py

@@ -11,6 +11,15 @@
 import logging
 
 
+class PtfAgent:
+    def __init__(self, ptf_ip, ptf_ipv6, ptf_nn_port, device_num, ptf_port_set):
+        self.ptf_ip = ptf_ip
+        self.ptf_ipv6 = ptf_ipv6
+        self.ptf_nn_port = ptf_nn_port
+        self.device_num = device_num
+        self.ptf_port_set = ptf_port_set
+
+
 class PtfAdapterNNConnectionError(Exception):
 
     def __init__(self, remote_sock_addr):
@@ -30,21 +39,17 @@ class PtfTestAdapter(BaseTest):
     # the number of currently established connections
     NN_STAT_CURRENT_CONNECTIONS = 201
 
-    def __init__(self, ptf_ip, ptf_ipv6, ptf_nn_port, device_num, ptf_port_set, ptfhost):
+    def __init__(self, ptfagents, ptfhosts):
         """ initialize PtfTestAdapter
-        :param ptf_ip: PTF host IP
-        :param ptf_ipv6: PTF host IPv6 address
-        :param ptf_nn_port: PTF nanomessage agent port
-        :param device_num: device number
-        :param ptf_port_set: PTF ports
-        :return:
         """
         self.runTest = lambda: None    # set a no op runTest attribute to satisfy BaseTest interface
         super(PtfTestAdapter, self).__init__()
         self.payload_pattern = ""
         self.connected = False
-        self.ptfhost = ptfhost
-        self._init_ptf_dataplane(ptf_ip, ptf_ipv6, ptf_nn_port, device_num, ptf_port_set)
+        self.ptfhosts = ptfhosts
+        self.ptfagents = ptfagents
+        self.ptf_port_set = [k for a in ptfagents for k in a.ptf_port_set.keys()]
+        self._init_ptf_dataplane()
 
     def __enter__(self):
         """ enter in 'with' block """
@@ -65,42 +70,33 @@ def _check_ptf_nn_agent_availability(self, socket_addr):
         finally:
             sock.close()
 
-    def _init_ptf_dataplane(self, ptf_ip, ptf_ipv6, ptf_nn_port, device_num, ptf_port_set, ptf_config=None):
+    def _init_ptf_dataplane(self, ptf_config=None):
         """
         initialize ptf framework and establish connection to ptf_nn_agent
         running on PTF host
-        :param ptf_ip: PTF host IP
-        :param ptf_ipv6: PTF host IPv6 address
-        :param ptf_nn_port: PTF nanomessage agent port
-        :param device_num: device number
-        :param ptf_port_set: PTF ports
         :return:
         """
-        self.ptf_ip = ptf_ip
-        self.ptf_ipv6 = ptf_ipv6
-        self.ptf_nn_port = ptf_nn_port
-        self.device_num = device_num
-        self.ptf_port_set = ptf_port_set
         self.connected = False
 
         ptfutils.default_timeout = self.DEFAULT_PTF_TIMEOUT
         ptfutils.default_negative_timeout = self.DEFAULT_PTF_NEG_TIMEOUT
 
-        ptf_nn_sock_addr = 'tcp://{}:{}'.format(ptf_ip, ptf_nn_port)
-
         ptf.config.update({
             'platform': 'nn',
-            'device_sockets': [
-                (device_num, ptf_port_set, ptf_nn_sock_addr)
-            ],
+            'device_sockets': [],
             'qlen': self.DEFAULT_PTF_QUEUE_LEN,
             'relax': True,
         })
+
         if ptf_config is not None:
             ptf.config.update(ptf_config)
 
-        if not self._check_ptf_nn_agent_availability(ptf_nn_sock_addr):
-            raise PtfAdapterNNConnectionError(ptf_nn_sock_addr)
+        for ptfagent in self.ptfagents:
+            ptf_nn_sock_addr = 'tcp://{}:{}'.format(ptfagent.ptf_ip, ptfagent.ptf_nn_port)
+            ptf.config['device_sockets'].append((ptfagent.device_num, ptfagent.ptf_port_set, ptf_nn_sock_addr))
+
+            if not self._check_ptf_nn_agent_availability(ptf_nn_sock_addr):
+                raise PtfAdapterNNConnectionError(ptf_nn_sock_addr)
 
         # update ptf.config based on NN platform and create dataplane instance
         nn.platform_config_update(ptf.config)
@@ -112,6 +108,26 @@ def _init_ptf_dataplane(self, ptf_ip, ptf_ipv6, ptf_nn_port, device_num, ptf_por
             device_id, port_id = id
             ptf.dataplane_instance.port_add(ifname, device_id, port_id)
         self.connected = True
+        ptf.dataplane_instance.port_device_map = {p: d for d, p in ptf.dataplane_instance.ports.keys()}
+        ptf.dataplane_instance.port_to_device = lambda port: ptf.dataplane_instance.port_device_map[port]
+        ptf.dataplane_instance.port_to_tuple = lambda port: (ptf.dataplane_instance.port_device_map[port], port)
+        ptf.dataplane_instance._poll = ptf.dataplane_instance.poll
+        ptf.dataplane_instance.poll = lambda device_number, port_number=None, timeout=None, exp_pkt=None, filters=[]: \
+            ptf.dataplane_instance._poll(
+                ptf.dataplane_instance.port_to_device(port_number)
+                if port_number is not None else device_number if device_number is not None else None,
+                port_number,
+                timeout,
+                exp_pkt,
+                filters
+            )
+        ptf.dataplane_instance._send = ptf.dataplane_instance.send
+        ptf.dataplane_instance.send = lambda device_number, port_number, packet: \
+            ptf.dataplane_instance._send(
+                ptf.dataplane_instance.port_to_device(port_number),
+                port_number,
+                packet
+            )
         self.dataplane = ptf.dataplane_instance
 
     def kill(self):
@@ -136,12 +152,12 @@ def reinit(self, ptf_config=None):
 
         # Restart ptf_nn_agent to close any TCP connection from the server side
         logging.info("Restarting ptf_nn_agent")
-        self.ptfhost.command('supervisorctl reread')
-        self.ptfhost.command('supervisorctl update')
-        self.ptfhost.command('supervisorctl restart ptf_nn_agent')
+        for ptfhost in self.ptfhosts:
+            ptfhost.command('supervisorctl reread')
+            ptfhost.command('supervisorctl update')
+            ptfhost.command('supervisorctl restart ptf_nn_agent')
 
-        self._init_ptf_dataplane(self.ptf_ip, self.ptf_ipv6, self.ptf_nn_port,
-                                 self.device_num, self.ptf_port_set, ptf_config)
+        self._init_ptf_dataplane(ptf_config)
 
     def update_payload(self, pkt):
         """Update the payload of packet to the default pattern when certain conditions are met.

tests/common/ptf_agent_updater.py

@@ -5,61 +5,66 @@ class PtfAgentUpdater(object):
     """
     PtfAgentUpdater class for updating ptf_nn_agent on PTF host
     """
-    def __init__(self, ptfhost, ptfadapter, ptf_nn_agent_template):
+    def __init__(self, ptfhosts, ptfadapter, ptf_nn_agent_template):
         """
         Initialize an object for updating ptf_nn_agent
 
         Args:
-            ptfhost: PTF host object
+            ptfhosts: PTF hosts object
             ptfadapter: PTF adapter
             ptf_nn_agent_template: ptf_nn_agent template
         """
-        self.ptfhost = ptfhost
+        self.ptfhosts = ptfhosts
         self.ptfadapter = ptfadapter
         self.ptf_nn_agent_template = ptf_nn_agent_template
 
-    def configure_ptf_nn_agent(self, ifaces):
+    def configure_ptf_nn_agent(self, ifaces, ptf_index=0):
         """
         Add new interfaces to interfaces map of ptfadapter
 
         Args:
             ifaces: List of interface names
+            ptf_index: The index of ptfhost in ptfhosts to configure ptf_nn_agent
         """
+        ptfhost = self.ptfhosts[ptf_index]
         ifaces = [ifaces] if not isinstance(ifaces, list) else ifaces
-        last_iface_id = sorted(self.ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'].keys())[-1]
+        last_iface_id = sorted(ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'].keys())[-1]
 
         for iface_id, iface in enumerate(ifaces, last_iface_id+1):
-            self.ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'][iface_id] = iface
+            ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'][iface_id] = iface
             self.ptfadapter.ptf_port_set.append(iface_id)
 
-        self._restart_ptf_nn_agent()
+        self._restart_ptf_nn_agent(ptf_index)
 
         self.ptfadapter.reinit()
 
-    def cleanup_ptf_nn_agent(self, ifaces):
+    def cleanup_ptf_nn_agent(self, ifaces, ptf_index=0):
         """
         Remove interfaces from interfaces map of ptfadapter
 
         Args:
             ifaces: List of interface names
+            ptf_index: The index of ptfhost in ptfhosts to configure ptf_nn_agent
         """
+        ptfhost = self.ptfhosts[ptf_index]
         ifaces = [ifaces] if not isinstance(ifaces, list) else ifaces
-        ifaces_map = self.ptfhost.host.options['variable_manager'].extra_vars['ifaces_map']
+        ifaces_map = ptfhost.host.options['variable_manager'].extra_vars['ifaces_map']
         config_port_indices = {v: k for k, v in list(ifaces_map.items())}
 
         for iface in ifaces:
-            self.ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'].pop(config_port_indices[iface])
+            ptfhost.host.options['variable_manager'].extra_vars['ifaces_map'].pop(config_port_indices[iface])
             self.ptfadapter.ptf_port_set.remove(config_port_indices[iface])
 
-        self._restart_ptf_nn_agent()
+        self._restart_ptf_nn_agent(ptf_index)
 
         self.ptfadapter.reinit()
 
-    def _restart_ptf_nn_agent(self):
+    def _restart_ptf_nn_agent(self, ptf_index=0):
         """
         Restart ptf_nn_agent
         """
-        self.ptfhost.template(src=self.ptf_nn_agent_template, dest=PTF_NN_AGENT_CONF)
-        self.ptfhost.command('supervisorctl reread')
-        self.ptfhost.command('supervisorctl update')
-        self.ptfhost.command('supervisorctl restart ptf_nn_agent')
+        ptfhost = self.ptfhosts[ptf_index]
+        ptfhost.template(src=self.ptf_nn_agent_template, dest=PTF_NN_AGENT_CONF)
+        ptfhost.command('supervisorctl reread')
+        ptfhost.command('supervisorctl update')
+        ptfhost.command('supervisorctl restart ptf_nn_agent')

tests/common/testbed.py

@@ -107,10 +107,10 @@ def _read_testbed_topo_from_yaml(self):
         with open(self.testbed_filename) as f:
             tb_info = yaml.safe_load(f)
             for tb in tb_info:
-                if tb["ptf_ip"]:
+                if "ptf_ip" in tb and tb["ptf_ip"]:
                     tb["ptf_ip"], tb["ptf_netmask"] = \
                         self._cidr_to_ip_mask(tb["ptf_ip"])
-                if tb["ptf_ipv6"]:
+                if "ptf_ipv6" in tb and tb["ptf_ipv6"]:
                     tb["ptf_ipv6"], tb["ptf_netmask_v6"] = \
                         self._cidr_to_ip_mask(tb["ptf_ipv6"])
                 tb["duts"] = tb.pop("dut")