"keepalived_state_change" needs to use threading to send arping

"keepalived_state_change" monitor does not use eventlet but normal
Python threads. When "send_ip_addr_adv_notif" is called from inside
the monitor, the arping command is never sent because the eventlet
thread does not start. In order to be able to be called from this
process, this method should also have an alternative implementation
using "threading".

"TestMonitorDaemon.test_new_fip_sends_garp" is also modified to
actually test the GARP sent. The test was originally implemented with
only one interface in the monitored namespace.
"keepalived_state_change" sends a GARP when a new IP address is added
in a interface other than the monitored one. That's why this patch
creates a new interface and sets it as the monitor interface. When
a new IP address is added to the other interface, the monitor populates
it by sending a GARP through the modified interface [1].

[1] 8ee34655b8/neutron/agent/l3/keepalived_state_change.py (L90)

Change-Id: Ib69e21b4645cef71db07595019fac9af77fefaa1
Closes-Bug: #1870313
(cherry picked from commit 21935365f2)

neutron/agent/l3/keepalived_state_change.py

@@ -143,7 +143,8 @@ class MonitorDaemon(daemon.Daemon):
             self.namespace,
             event['name'],
             ip_address,
-            log_exception=False
+            log_exception=False,
+            use_eventlet=False
         )
         LOG.debug('Sent GARP to %(ip_address)s from %(device_name)s',
                   {'ip_address': ip_address, 'device_name': event['name']})

neutron/agent/linux/ip_lib.py

@@ -1033,7 +1033,8 @@ def _arping(ns_name, iface_name, address, count, log_exception):
 
 
 def send_ip_addr_adv_notif(
-        ns_name, iface_name, address, count=3, log_exception=True):
+        ns_name, iface_name, address, count=3, log_exception=True,
+        use_eventlet=True):
     """Send advance notification of an IP address assignment.
 
     If the address is in the IPv4 family, send gratuitous ARP.
@@ -1051,12 +1052,18 @@ def send_ip_addr_adv_notif(
     :param log_exception: (Optional) True if possible failures should be logged
                           on exception level. Otherwise they are logged on
                           WARNING level. Default is True.
+    :param use_eventlet: (Optional) True if the arping command will be spawned
+                         using eventlet, False to use Python threads
+                         (threading).
     """
     def arping():
         _arping(ns_name, iface_name, address, count, log_exception)
 
     if count > 0 and netaddr.IPAddress(address).version == 4:
-        eventlet.spawn_n(arping)
+        if use_eventlet:
+            eventlet.spawn_n(arping)
+        else:
+            threading.Thread(target=arping).start()
 
 
 def sysctl(cmd, namespace=None, log_fail_as_error=True):

neutron/tests/functional/agent/l3/test_keepalived_state_change.py

@@ -52,19 +52,7 @@ class TestMonitorDaemon(base.BaseLoggingTestCase):
         self.router, self.peer = self.machines.machines[:2]
         self.router_id = uuidutils.generate_uuid()
 
-        self.cmd_opts = [
-            ha_router.STATE_CHANGE_PROC_NAME,
-            '--router_id=%s' % self.router_id,
-            '--namespace=%s' % self.router.namespace,
-            '--conf_dir=%s' % self.conf_dir,
-            '--log-file=%s' % self.log_file,
-            '--monitor_interface=%s' % self.router.port.name,
-            '--monitor_cidr=%s' % self.cidr,
-            '--pid_file=%s' % self.pid_file,
-            '--state_path=%s' % self.conf_dir,
-            '--user=%s' % os.geteuid(),
-            '--group=%s' % os.getegid()
-        ]
+        self._generate_cmd_opts()
         self.ext_process = external_process.ProcessManager(
             None, self.router_id, namespace=self.router.namespace,
             service='test_ip_mon', pids_path=self.conf_dir,
@@ -86,6 +74,22 @@ class TestMonitorDaemon(base.BaseLoggingTestCase):
     def _callback(self, *args):
         return self.cmd_opts
 
+    def _generate_cmd_opts(self, monitor_interface=None, cidr=None):
+        monitor_interface = monitor_interface or self.router.port.name
+        cidr = cidr or self.cidr
+        self.cmd_opts = [
+            ha_router.STATE_CHANGE_PROC_NAME,
+            '--router_id=%s' % self.router_id,
+            '--namespace=%s' % self.router.namespace,
+            '--conf_dir=%s' % self.conf_dir,
+            '--log-file=%s' % self.log_file,
+            '--monitor_interface=%s' % monitor_interface,
+            '--monitor_cidr=%s' % cidr,
+            '--pid_file=%s' % self.pid_file,
+            '--state_path=%s' % self.conf_dir,
+            '--user=%s' % os.geteuid(),
+            '--group=%s' % os.getegid()]
+
     def _search_in_file(self, file_name, text):
         def text_in_file():
             try:
@@ -103,11 +107,22 @@ class TestMonitorDaemon(base.BaseLoggingTestCase):
                  'file_content': open(file_name).read()})
 
     def test_new_fip_sends_garp(self):
+        ns_ip_wrapper = ip_lib.IPWrapper(self.router.namespace)
+        new_interface = ns_ip_wrapper.add_dummy('new_interface')
+        new_interface_cidr = '169.254.152.1/24'
+        new_interface.link.set_up()
+        new_interface.addr.add(new_interface_cidr)
+        self._generate_cmd_opts(monitor_interface='new_interface',
+                                cidr=new_interface_cidr)
+
         self._run_monitor()
+
         next_ip_cidr = net_helpers.increment_ip_cidr(self.machines.ip_cidr, 2)
         expected_ip = str(netaddr.IPNetwork(next_ip_cidr).ip)
         # Create incomplete ARP entry
         self.peer.assert_no_ping(expected_ip)
+        # Wait for ping expiration
+        eventlet.sleep(1)
         has_entry = has_expected_arp_entry(
             self.peer.port.name,
             self.peer.namespace,
@@ -130,6 +145,9 @@ class TestMonitorDaemon(base.BaseLoggingTestCase):
                 self.router.port.link.address))
         utils.wait_until_true(has_arp_entry_predicate, timeout=15,
                               exception=exc)
+        msg = ('Sent GARP to %(cidr)s from %(device)s' %
+               {'cidr': expected_ip, 'device': self.router.port.name})
+        self._search_in_file(self.log_file, msg)
 
     def test_read_queue_change_state(self):
         self._run_monitor()