Merge "Read IP monitor changes in a parallel thread"

neutron/agent/linux/ip_lib.py

@@ -1474,43 +1474,64 @@ def ip_monitor(namespace, queue, event_stop, event_started):
     cannot use privsep because is a blocking function and can exhaust the
     number of working threads.
     """
-    def get_device_name(ip, index):
+    def get_device_name(index):
         try:
-            device = ip.link('get', index=index)
-            if device:
-                attrs = device[0].get('attrs', [])
-                for attr in (attr for attr in attrs
-                             if attr[0] == 'IFLA_IFNAME'):
-                    return attr[1]
+            with privileged.get_iproute(namespace) as ip:
+                device = ip.link('get', index=index)
+                if device:
+                    attrs = device[0].get('attrs', [])
+                    for attr in (attr for attr in attrs
+                                 if attr[0] == 'IFLA_IFNAME'):
+                        return attr[1]
         except netlink_exceptions.NetlinkError as e:
             if e.code == errno.ENODEV:
                 return
             raise
 
-    try:
+    def read_ip_updates(_queue):
+        """Read Pyroute2.IPRoute input socket
+
+        The aim of this function is to open and bind an IPRoute socket only for
+        reading the netlink changes; no other operations are done with this
+        opened socket. This function is executed in a separate thread,
+        dedicated only to this task.
+        """
         with privileged.get_iproute(namespace) as ip:
             ip.bind()
-            cache_devices = {}
+            while True:
+                eventlet.sleep(0)
+                ip_addresses = ip.get()
+                for ip_address in ip_addresses:
+                    _queue.put(ip_address)
+
+    _queue = eventlet.Queue()
+    try:
+        cache_devices = {}
+        with privileged.get_iproute(namespace) as ip:
             for device in ip.get_links():
                 cache_devices[device['index']] = get_attr(device,
                                                           'IFLA_IFNAME')
-            event_started.send()
-            while not event_stop.ready():
-                eventlet.sleep(0)
-                ip_address = []
-                with common_utils.Timer(timeout=2, raise_exception=False):
-                    ip_address = ip.get()
-                if not ip_address:
+        pool = eventlet.GreenPool(1)
+        ip_updates_thread = pool.spawn(read_ip_updates, _queue)
+        event_started.send()
+        while not event_stop.ready():
+            eventlet.sleep(0)
+            try:
+                ip_address = _queue.get(timeout=2)
+            except eventlet.queue.Empty:
+                continue
+            if 'index' in ip_address and 'prefixlen' in ip_address:
+                index = ip_address['index']
+                name = (get_device_name(index) or
+                        cache_devices.get(index))
+                if not name:
                     continue
-                if 'index' in ip_address[0] and 'prefixlen' in ip_address[0]:
-                    index = ip_address[0]['index']
-                    name = (get_device_name(ip, index) or
-                            cache_devices.get(index))
-                    if not name:
-                        continue
 
-                    cache_devices[index] = name
-                    queue.put(_parse_ip_address(ip_address[0], name))
+                cache_devices[index] = name
+                queue.put(_parse_ip_address(ip_address, name))
+
+        ip_updates_thread.kill()
+
     except OSError as e:
         if e.errno == errno.ENOENT:
             raise privileged.NetworkNamespaceNotFound(netns_name=namespace)

neutron/tests/functional/agent/linux/test_ip_lib.py

@@ -788,17 +788,39 @@ class IpMonitorTestCase(testscenarios.WithScenarios,
             self.ip_wrapper.add_dummy(device)
         utils.wait_until_true(lambda: self._read_file({}), timeout=30)
         ip_addresses = [
-            {'cidr': '192.168.250.21/24', 'event': 'added',
+            {'cidr': '192.168.251.21/24', 'event': 'added',
              'name': self.devices[0]},
-            {'cidr': '192.168.250.22/24', 'event': 'added',
+            {'cidr': '192.168.251.22/24', 'event': 'added',
              'name': self.devices[1]}]
 
         self._handle_ip_addresses('added', ip_addresses)
         self._check_read_file(ip_addresses)
 
         self.ip_wrapper.add_dummy(self.devices[-1])
-        ip_addresses.append({'cidr': '192.168.250.23/24', 'event': 'added',
+        ip_addresses.append({'cidr': '192.168.251.23/24', 'event': 'added',
                              'name': self.devices[-1]})
 
         self._handle_ip_addresses('added', [ip_addresses[-1]])
         self._check_read_file(ip_addresses)
+
+    def test_add_and_remove_multiple_ips(self):
+        # NOTE(ralonsoh): testing [1], adding multiple IPs.
+        # [1] https://bugs.launchpad.net/neutron/+bug/1832307
+        utils.wait_until_true(lambda: self._read_file({}), timeout=30)
+        self.ip_wrapper.add_dummy(self.devices[0])
+        ip_addresses = []
+        for i in range(250):
+            _cidr = str(netaddr.IPNetwork('192.168.252.1/32').ip + i) + '/32'
+            ip_addresses.append({'cidr': _cidr, 'event': 'added',
+                                 'name': self.devices[0]})
+
+        self._handle_ip_addresses('added', ip_addresses)
+        self._check_read_file(ip_addresses)
+
+        for i in range(250):
+            _cidr = str(netaddr.IPNetwork('192.168.252.1/32').ip + i) + '/32'
+            ip_addresses.append({'cidr': _cidr, 'event': 'removed',
+                                 'name': self.devices[0]})
+
+        self._handle_ip_addresses('removed', ip_addresses)
+        self._check_read_file(ip_addresses)