MDM Agent: Handle NetlinkDumpInterrupted except in ip.get_links()

modules/sc-mesh-secure-deployment/src/nats/src/cbma_adaptation.py

@@ -15,6 +15,7 @@
 import ipaddress
 import errno
 from pyroute2 import IPRoute, NetlinkError, arp  # type: ignore[import-not-found, import-untyped]
+from pyroute2.netlink.exceptions import NetlinkDumpInterrupted
 
 from src import cbma_paths
 from src.comms_controller import CommsController
@@ -280,7 +281,16 @@ def __delete_vlan_interfaces(self) -> bool:
     def __get_interfaces(self) -> None:
         interfaces = []
         ip = IPRoute()
-        for link in ip.get_links():
+
+        ip_links = []
+        while True:
+            try:
+                ip_links = ip.get_links()
+                break
+            except NetlinkDumpInterrupted:
+                time.sleep(1)
+
+        for link in ip_links:
             ifname = link.get_attr("IFLA_IFNAME")
             ifstate = link.get_attr("IFLA_OPERSTATE")
             mac_address = link.get_attr("IFLA_ADDRESS")

modules/sc-mesh-secure-deployment/src/nats/src/comms_if_monitor.py

@@ -6,8 +6,10 @@
 # pylint: disable=too-few-public-methods, too-many-nested-blocks
 from typing import Callable, List, Dict
 import subprocess
+import time
 from copy import deepcopy
 from pyroute2 import IPRoute
+from pyroute2.netlink.exceptions import NetlinkDumpInterrupted
 
 DUMMY_INTERFACE_NAME = 'ifdummy0'
 
@@ -24,7 +26,15 @@ def __init__(self, callback: Callable[[List[Dict]], None]) -> None:
         self.__ipr = IPRoute()
 
     def __get_initial_interfaces(self):
-        for link in self.__ipr.get_links():
+        ip_links = []
+        while True:
+            try:
+                ip_links = self.__ipr.get_links()
+                break
+            except NetlinkDumpInterrupted:
+                time.sleep(1)
+
+        for link in ip_links:
             interface_info = self.__get_interface_info(link)
             if interface_info:
                 self.__interfaces.append(interface_info)