[nic_simulator] Add timeout and common options (#6005)

Approach
What is the motivation for this PR?
mgmt client could not talks to the gRPC server.
Signed-off-by: Longxiang Lyu <[email protected]>

How did you do it?
Bring up loopback device to enable mgmt service talks to each gRPC server interacting with SONiC.
Add common gRPC options to the nic_simulator.
Add timeout for the gRPC calls from mgmt service.

How did you verify/test it?
Verify mgmt client could talks to the mgmt service of `nic_simulator.

Any platform specific information?

Author: lolyu

ansible/dualtor/nic_simulator/nic_simulator.py

@@ -44,6 +44,17 @@
 ACTIVE_ACTIVE_INTERFACE_PATTERN = "iaa-[\w-]+-\d+"
 SERVER_NIC_INTERFACE_TEMPLATE = "nic-%s-%d"
 SERVER_NIC_INTERFACE_PATTERN = "nic-[\w-]+-\d+"
+GRPC_TIMEOUT = 0.5
+GRPC_SERVER_OPTIONS = [
+    ('grpc.http2.min_ping_interval_without_data_ms', 1000),
+    ('grpc.http2.max_ping_strikes',  0)
+]
+GRPC_CLIENT_OPTIONS = [
+    ('grpc.keepalive_timeout_ms', 8000),
+    ('grpc.keepalive_time_ms', 4000),
+    ('grpc.keepalive_permit_without_calls', True),
+    ('grpc.http2.max_pings_without_data', 0)
+]
 
 
 def get_ip_address(ifname):
@@ -530,9 +541,10 @@ def QueryServerVersion(self, request, context):
 
     def _run_server(self, binding_port):
         """Run the gRPC server."""
-        self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=THREAD_CONCURRENCY_PER_SERVER),
-                                                   options=[('grpc.http2.min_ping_interval_without_data_ms', 1000),
-                                                            ('grpc.http2.max_ping_strikes',  0)])
+        self.server = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=THREAD_CONCURRENCY_PER_SERVER),
+            options=GRPC_SERVER_OPTIONS
+        )
         nic_simulator_grpc_service_pb2_grpc.add_DualToRActiveServicer_to_server(
             self,
             self.server
@@ -569,7 +581,10 @@ def _get_client_stub(self, nic_address):
             client_stub = self.client_stubs[nic_address]
         else:
             client_stub = nic_simulator_grpc_service_pb2_grpc.DualToRActiveStub(
-                grpc.insecure_channel("%s:%s" % (nic_address, self.binding_port))
+                grpc.insecure_channel(
+                    "%s:%s" % (nic_address, self.binding_port),
+                    options=GRPC_CLIENT_OPTIONS
+                )
             )
             self.client_stubs[nic_address] = client_stub
         return client_stub
@@ -585,7 +600,8 @@ def QueryAdminForwardingPortState(self, request, context):
                     nic_simulator_grpc_service_pb2.AdminRequest(
                         portid=[0, 1],
                         state=[True, True]
-                    )
+                    ),
+                    timeout=GRPC_TIMEOUT
                 )
                 query_responses.append(state)
             except Exception as e:
@@ -608,7 +624,8 @@ def SetAdminForwardingPortState(self, request, context):
             client_stub = self._get_client_stub(nic_address)
             try:
                 state = client_stub.SetAdminForwardingPortState(
-                    admin_request
+                    admin_request,
+                    timeout=GRPC_TIMEOUT
                 )
                 set_responses.append(state)
             except Exception as e:
@@ -626,9 +643,10 @@ def QueryOperationPortState(self, request, context):
         return nic_simulator_grpc_mgmt_service_pb2.ListOfOperationReply()
 
     def start(self):
-        self.server = grpc.server(futures.ThreadPoolExecutor(max_workers=THREAD_CONCURRENCY_PER_SERVER),
-                                                   options=[('grpc.http2.min_ping_interval_without_data_ms', 1000),
-                                                            ('grpc.http2.max_ping_strikes',  0)])
+        self.server = grpc.server(
+            futures.ThreadPoolExecutor(max_workers=THREAD_CONCURRENCY_PER_SERVER),
+            options=GRPC_SERVER_OPTIONS
+        )
         nic_simulator_grpc_mgmt_service_pb2_grpc.add_DualTorMgmtServiceServicer_to_server(self, self.server)
         self.server.add_insecure_port("%s:%s" % (self.binding_address, self.binding_port))
         self.server.start()

ansible/roles/vm_set/library/vm_topology.py

@@ -1033,6 +1033,10 @@ def add_host_ports(self):
                     vlan_id = self.vlan_ids[str(intf)]
                     self.add_dut_vlan_subif_to_docker(ptf_if, vlan_separator, vlan_id)
 
+    def enable_netns_loopback(self):
+        """Enable loopback device in the netns."""
+        VMTopology.cmd("ip netns exec %s ifconfig lo up" % self.netns)        
+
     def setup_netns_source_routing(self):
         """Setup policy-based routing to forward packet to its igress ports."""
 
@@ -1593,6 +1597,7 @@ def main():
             if net.netns:
                 net.add_network_namespace()
                 net.add_mgmt_port_to_netns(mgmt_bridge, netns_mgmt_ip_addr, ptf_mgmt_ip_gw)
+                net.enable_netns_loopback()
 
             if hostif_exists:
                 net.add_host_ports()
@@ -1722,6 +1727,7 @@ def main():
             if net.netns:
                 net.add_network_namespace()
                 net.add_mgmt_port_to_netns(mgmt_bridge, netns_mgmt_ip_addr, ptf_mgmt_ip_gw)
+                net.enable_netns_loopback()
 
             if hostif_exists:
                 net.add_host_ports()