Replace keepalived notifier bash script with Python ip monitor

Previously L3 HA generated a bash script and copied it to a per-router
configuration directory that was visible to that router's keepalived
instance. This patch changes the in-line generated Bash script to a
Python script that can be maintained in the repository.
The bash script was used as a keepalived notifier script, that was invoked
by keepalived whenever a state transition occured. These notifier scripts
may be invoked by keepalived out of order in case it transitions quickly
twice. For example, if the master failed and two slaves fight for the new
master role. One will transition to master, and the other will often
transition to master and then immidiately back to standby. In this case,
the transition scripts were often fired out of order, resulting in the
wrong state being reported.

The proposed approach is to get rid of the keepalived notifier scripts
entirely. Instead, monitor IP changes on the HA device. If the omnipresent
IP address was configured on the HA device, it means that we're looking
at a master instance. If it was deleted, the router transition to standby
or fault.

In order to keep the L3 agent CPU usage down, it will spawn a process
per HA router. That process will start the ip address monitor.
Whenever it gets an IP address change event, it will notify the L3 agent
via a unix domain socket.

Partially-Implements: blueprint report-ha-router-master
Change-Id: I2022bced330d5f108fbedd40548a901225d7ea1c
Closes-Bug: #1402010
Closes-Bug: #1367705

etc/neutron/rootwrap.d/l3.filters

@@ -49,3 +49,6 @@ kill_keepalived: KillFilter, root, /usr/sbin/keepalived, -HUP, -15, -9
 
 # l3 agent to delete floatingip's conntrack state
 conntrack: CommandFilter, conntrack, root
+
+# keepalived state change monitor
+keepalived_state_change: CommandFilter, neutron-keepalived-state-change, root

neutron/agent/l3/ha.py

@@ -15,17 +15,20 @@
 
 import os
 
+import eventlet
 from oslo_config import cfg
 from oslo_log import log as logging
+import webob
 
 from neutron.agent.linux import keepalived
-from neutron.agent.linux import utils
+from neutron.agent.linux import utils as agent_utils
 from neutron.common import constants as l3_constants
-from neutron.i18n import _LE
+from neutron.i18n import _LE, _LI
 
 LOG = logging.getLogger(__name__)
 
 HA_DEV_PREFIX = 'ha-'
+KEEPALIVED_STATE_CHANGE_SERVER_BACKLOG = 4096
 
 OPTS = [
     cfg.StrOpt('ha_confs_path',
@@ -45,14 +48,83 @@ OPTS = [
 ]
 
 
+class KeepalivedStateChangeHandler(object):
+    def __init__(self, agent):
+        self.agent = agent
+
+    @webob.dec.wsgify(RequestClass=webob.Request)
+    def __call__(self, req):
+        router_id = req.headers['X-Neutron-Router-Id']
+        state = req.headers['X-Neutron-State']
+        self.enqueue(router_id, state)
+
+    def enqueue(self, router_id, state):
+        LOG.debug('Handling notification for router '
+                  '%(router_id)s, state %(state)s', {'router_id': router_id,
+                                                     'state': state})
+        self.agent.enqueue_state_change(router_id, state)
+
+
+class L3AgentKeepalivedStateChangeServer(object):
+    def __init__(self, agent, conf):
+        self.agent = agent
+        self.conf = conf
+
+        agent_utils.ensure_directory_exists_without_file(
+            self.get_keepalived_state_change_socket_path(self.conf))
+
+    @classmethod
+    def get_keepalived_state_change_socket_path(cls, conf):
+        return os.path.join(conf.state_path, 'keepalived-state-change')
+
+    def run(self):
+        server = agent_utils.UnixDomainWSGIServer(
+            'neutron-keepalived-state-change')
+        server.start(KeepalivedStateChangeHandler(self.agent),
+                     self.get_keepalived_state_change_socket_path(self.conf),
+                     workers=0,
+                     backlog=KEEPALIVED_STATE_CHANGE_SERVER_BACKLOG)
+        server.wait()
+
+
 class AgentMixin(object):
     def __init__(self, host):
         self._init_ha_conf_path()
         super(AgentMixin, self).__init__(host)
+        eventlet.spawn(self._start_keepalived_notifications_server)
+
+    def _start_keepalived_notifications_server(self):
+        state_change_server = (
+            L3AgentKeepalivedStateChangeServer(self, self.conf))
+        state_change_server.run()
+
+    def enqueue_state_change(self, router_id, state):
+        LOG.info(_LI('Router %(router_id)s transitioned to %(state)s'),
+                 {'router_id': router_id,
+                  'state': state})
+        self._update_metadata_proxy(router_id, state)
+
+    def _update_metadata_proxy(self, router_id, state):
+        try:
+            ri = self.router_info[router_id]
+        except AttributeError:
+            LOG.info(_LI('Router %s is not managed by this agent. It was '
+                         'possibly deleted concurrently.'), router_id)
+            return
+
+        if state == 'master':
+            LOG.debug('Spawning metadata proxy for router %s', router_id)
+            self.metadata_driver.spawn_monitored_metadata_proxy(
+                self.process_monitor, ri.ns_name, self.conf.metadata_port,
+                self.conf, router_id=ri.router_id)
+        else:
+            LOG.debug('Closing metadata proxy for router %s', router_id)
+            self.metadata_driver.destroy_monitored_metadata_proxy(
+                self.process_monitor, ri.router_id, ri.ns_name, self.conf)
 
     def _init_ha_conf_path(self):
         ha_full_path = os.path.dirname("/%s/" % self.conf.ha_confs_path)
-        utils.ensure_dir(ha_full_path)
+        agent_utils.ensure_dir(ha_full_path)
 
     def process_ha_router_added(self, ri):
         ha_port = ri.router.get(l3_constants.HA_INTERFACE_KEY)
@@ -69,7 +141,9 @@ class AgentMixin(object):
                             ha_port['ip_cidr'],
                             ha_port['mac_address'])
 
-        ri._add_keepalived_notifiers()
+        ri.update_initial_state(self.enqueue_state_change)
+        ri.spawn_state_change_monitor(self.process_monitor)
 
     def process_ha_router_removed(self, ri):
+        ri.destroy_state_change_monitor(self.process_monitor)
         ri.ha_network_removed()

neutron/agent/l3/ha_router.py

@@ -12,21 +12,23 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import os
 import shutil
-import signal
 
 import netaddr
 from oslo_log import log as logging
 
 from neutron.agent.l3 import router_info as router
+from neutron.agent.linux import external_process
 from neutron.agent.linux import ip_lib
 from neutron.agent.linux import keepalived
-from neutron.agent.metadata import driver as metadata_driver
 from neutron.common import constants as n_consts
 from neutron.common import utils as common_utils
+from neutron.i18n import _LE
 
 LOG = logging.getLogger(__name__)
 HA_DEV_PREFIX = 'ha-'
+IP_MONITOR_PROCESS_SERVICE = 'ip_monitor'
 
 
 class HaRouter(router.RouterInfo):
@@ -69,6 +71,18 @@ class HaRouter(router.RouterInfo):
             LOG.debug('Error while reading HA state for %s', self.router_id)
             return None
 
+    @ha_state.setter
+    def ha_state(self, new_state):
+        self._verify_ha()
+        ha_state_path = self.keepalived_manager._get_full_config_file_path(
+            'state')
+        try:
+            with open(ha_state_path, 'w') as f:
+                f.write(new_state)
+        except (OSError, IOError):
+            LOG.error(_LE('Error while writing HA state for %s'),
+                      self.router_id)
+
     def _init_keepalived_manager(self, process_monitor):
         self.keepalived_manager = keepalived.KeepalivedManager(
             self.router['id'],
@@ -107,27 +121,6 @@ class HaRouter(router.RouterInfo):
         conf_dir = self.keepalived_manager.get_conf_dir()
         shutil.rmtree(conf_dir)
 
-    def _add_keepalived_notifiers(self):
-        callback = (
-            metadata_driver.MetadataDriver._get_metadata_proxy_callback(
-                self.agent_conf.metadata_port, self.agent_conf,
-                router_id=self.router_id))
-        # TODO(mangelajo): use the process monitor in keepalived when
-        #                  keepalived stops killing/starting metadata
-        #                  proxy on its own
-        pm = (
-            metadata_driver.MetadataDriver.
-            _get_metadata_proxy_process_manager(self.router_id,
-                                                self.ns_name,
-                                                self.agent_conf))
-        pid = pm.get_pid_file_name()
-        self.keepalived_manager.add_notifier(
-            callback(pid), 'master', self.ha_vr_id)
-        for state in ('backup', 'fault'):
-            self.keepalived_manager.add_notifier(
-                ['kill', '-%s' % signal.SIGKILL,
-                 '$(cat ' + pid + ')'], state, self.ha_vr_id)
-
     def _get_keepalived_instance(self):
         return self.keepalived_manager.config.get_instance(self.ha_vr_id)
 
@@ -276,3 +269,53 @@ class HaRouter(router.RouterInfo):
 
         interface_name = self.get_internal_device_name(port['id'])
         self._clear_vips(interface_name)
+
+    def _get_state_change_monitor_process_manager(self):
+        return external_process.ProcessManager(
+            self.agent_conf,
+            '%s.monitor' % self.router_id,
+            self.ns_name,
+            default_cmd_callback=self._get_state_change_monitor_callback())
+
+    def _get_state_change_monitor_callback(self):
+        ha_device = self.get_ha_device_name(self.ha_port['id'])
+        ha_cidr = self._get_primary_vip()
+
+        def callback(pid_file):
+            cmd = [
+                'neutron-keepalived-state-change',
+                '--router_id=%s' % self.router_id,
+                '--namespace=%s' % self.ns_name,
+                '--conf_dir=%s' % self.keepalived_manager.get_conf_dir(),
+                '--monitor_interface=%s' % ha_device,
+                '--monitor_cidr=%s' % ha_cidr,
+                '--pid_file=%s' % pid_file,
+                '--state_path=%s' % self.agent_conf.state_path,
+                '--user=%s' % os.geteuid(),
+                '--group=%s' % os.getegid()]
+            return cmd
+
+        return callback
+
+    def spawn_state_change_monitor(self, process_monitor):
+        pm = self._get_state_change_monitor_process_manager()
+        pm.enable()
+        process_monitor.register(
+            self.router_id, IP_MONITOR_PROCESS_SERVICE, pm)
+
+    def destroy_state_change_monitor(self, process_monitor):
+        pm = self._get_state_change_monitor_process_manager()
+        process_monitor.unregister(
+            self.router_id, IP_MONITOR_PROCESS_SERVICE)
+        pm.disable()
+
+    def update_initial_state(self, callback):
+        ha_device = ip_lib.IPDevice(
+            self.get_ha_device_name(self.ha_port['id']),
+            self.ns_name)
+        addresses = ha_device.addr.list()
+        cidrs = (address['cidr'] for address in addresses)
+        ha_cidr = self._get_primary_vip()
+        state = 'master' if ha_cidr in cidrs else 'backup'
+        self.ha_state = state
+        callback(self.router_id, state)

neutron/agent/l3/keepalived_state_change.py

@@ -0,0 +1,144 @@
+# Copyright (c) 2015 Red Hat Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+import sys
+
+import httplib2
+from oslo_config import cfg
+from oslo_log import log as logging
+import requests
+
+from neutron.agent.l3 import ha
+from neutron.agent.linux import daemon
+from neutron.agent.linux import ip_monitor
+from neutron.agent.linux import utils as agent_utils
+from neutron.common import config
+from neutron.i18n import _LE
+
+
+LOG = logging.getLogger(__name__)
+
+
+class KeepalivedUnixDomainConnection(agent_utils.UnixDomainHTTPConnection):
+    def __init__(self, *args, **kwargs):
+        # Old style super initialization is required!
+        agent_utils.UnixDomainHTTPConnection.__init__(
+            self, *args, **kwargs)
+        self.socket_path = (
+            ha.L3AgentKeepalivedStateChangeServer.
+            get_keepalived_state_change_socket_path(cfg.CONF))
+
+
+class MonitorDaemon(daemon.Daemon):
+    def __init__(self, pidfile, router_id, user, group, namespace, conf_dir,
+                 interface, cidr):
+        self.router_id = router_id
+        self.namespace = namespace
+        self.conf_dir = conf_dir
+        self.interface = interface
+        self.cidr = cidr
+        super(MonitorDaemon, self).__init__(pidfile, uuid=router_id,
+                                            user=user, group=group)
+
+    def run(self, run_as_root=False):
+        monitor = ip_monitor.IPMonitor(namespace=self.namespace,
+                                       run_as_root=run_as_root)
+        monitor.start()
+        # Only drop privileges if the process is currently running as root
+        # (The run_as_root variable name here is unfortunate - It means to
+        # use a root helper when the running process is NOT already running
+        # as root
+        if not run_as_root:
+            super(MonitorDaemon, self).run()
+        for iterable in monitor:
+            self.parse_and_handle_event(iterable)
+
+    def parse_and_handle_event(self, iterable):
+        try:
+            event = ip_monitor.IPMonitorEvent.from_text(iterable)
+            if event.interface == self.interface and event.cidr == self.cidr:
+                new_state = 'master' if event.added else 'backup'
+                self.write_state_change(new_state)
+                self.notify_agent(new_state)
+        except Exception:
+            LOG.exception(_LE(
+                'Failed to process or handle event for line %s'), iterable)
+
+    def write_state_change(self, state):
+        with open(os.path.join(
+                self.conf_dir, 'state'), 'w') as state_file:
+            state_file.write(state)
+        LOG.debug('Wrote router %s state %s', self.router_id, state)
+
+    def notify_agent(self, state):
+        resp, content = httplib2.Http().request(
+            # Note that the message is sent via a Unix domain socket so that
+            # the URL doesn't matter.
+            'http://127.0.0.1/',
+            headers={'X-Neutron-Router-Id': self.router_id,
+                     'X-Neutron-State': state},
+            connection_type=KeepalivedUnixDomainConnection)
+
+        if resp.status != requests.codes.ok:
+            raise Exception(_('Unexpected response: %s') % resp)
+
+        LOG.debug('Notified agent router %s, state %s', self.router_id, state)
+
+
+def register_opts(conf):
+    conf.register_cli_opt(
+        cfg.StrOpt('router_id', help=_('ID of the router')))
+    conf.register_cli_opt(
+        cfg.StrOpt('namespace', help=_('Namespace of the router')))
+    conf.register_cli_opt(
+        cfg.StrOpt('conf_dir', help=_('Path to the router directory')))
+    conf.register_cli_opt(
+        cfg.StrOpt('monitor_interface', help=_('Interface to monitor')))
+    conf.register_cli_opt(
+        cfg.StrOpt('monitor_cidr', help=_('CIDR to monitor')))
+    conf.register_cli_opt(
+        cfg.StrOpt('pid_file', help=_('Path to PID file for this process')))
+    conf.register_cli_opt(
+        cfg.StrOpt('user', help=_('User (uid or name) running this process '
+                                  'after its initialization')))
+    conf.register_cli_opt(
+        cfg.StrOpt('group', help=_('Group (gid or name) running this process '
+                                   'after its initialization')))
+    conf.register_opt(
+        cfg.StrOpt('metadata_proxy_socket',
+                   default='$state_path/metadata_proxy',
+                   help=_('Location of Metadata Proxy UNIX domain '
+                          'socket')))
+
+
+def configure(conf):
+    config.init(sys.argv[1:])
+    conf.set_override('log_dir', cfg.CONF.conf_dir)
+    conf.set_override('debug', True)
+    conf.set_override('verbose', True)
+    config.setup_logging()
+
+
+def main():
+    register_opts(cfg.CONF)
+    configure(cfg.CONF)
+    MonitorDaemon(cfg.CONF.pid_file,
+                  cfg.CONF.router_id,
+                  cfg.CONF.user,
+                  cfg.CONF.group,
+                  cfg.CONF.namespace,
+                  cfg.CONF.conf_dir,
+                  cfg.CONF.monitor_interface,
+                  cfg.CONF.monitor_cidr).start()

neutron/agent/linux/ip_monitor.py

@@ -60,7 +60,7 @@ class IPMonitor(async_process.AsyncProcess):
     """Wrapper over `ip monitor address`.
 
     To monitor and react indefinitely:
-        m = IPMonitor(namespace='tmp')
+        m = IPMonitor(namespace='tmp', root_as_root=True)
         m.start()
         for iterable in m:
             event = IPMonitorEvent.from_text(iterable)
@@ -69,9 +69,10 @@ class IPMonitor(async_process.AsyncProcess):
 
     def __init__(self,
                  namespace=None,
+                 run_as_root=True,
                  respawn_interval=None):
         super(IPMonitor, self).__init__(['ip', '-o', 'monitor', 'address'],
-                                        run_as_root=True,
+                                        run_as_root=run_as_root,
                                         respawn_interval=respawn_interval,
                                         namespace=namespace)
 

neutron/agent/linux/keepalived.py

@@ -15,7 +15,6 @@
 import errno
 import itertools
 import os
-import stat
 
 import netaddr
 from oslo_config import cfg
@@ -26,7 +25,6 @@ from neutron.agent.linux import utils
 from neutron.common import exceptions
 
 VALID_STATES = ['MASTER', 'BACKUP']
-VALID_NOTIFY_STATES = ['master', 'backup', 'fault']
 VALID_AUTH_TYPES = ['AH', 'PASS']
 HA_DEFAULT_PRIORITY = 50
 PRIMARY_VIP_RANGE_SIZE = 24
@@ -69,16 +67,6 @@ class InvalidInstanceStateException(exceptions.NeutronException):
         super(InvalidInstanceStateException, self).__init__(**kwargs)
 
 
-class InvalidNotifyStateException(exceptions.NeutronException):
-    message = _('Invalid notify state: %(state)s, valid states are: '
-                '%(valid_notify_states)s')
-
-    def __init__(self, **kwargs):
-        if 'valid_notify_states' not in kwargs:
-            kwargs['valid_notify_states'] = ', '.join(VALID_NOTIFY_STATES)
-        super(InvalidNotifyStateException, self).__init__(**kwargs)
-
-
 class InvalidAuthenticationTypeException(exceptions.NeutronException):
     message = _('Invalid authentication type: %(auth_type)s, '
                 'valid types are: %(valid_auth_types)s')
@@ -141,7 +129,6 @@ class KeepalivedInstance(object):
         self.vips = []
         self.virtual_routes = []
         self.authentication = None
-        self.notifiers = []
         metadata_cidr = '169.254.169.254/32'
         self.primary_vip_range = get_free_range(
             parent_range='169.254.0.0/16',
@@ -174,11 +161,6 @@ class KeepalivedInstance(object):
         return [vip.ip_address for vip in self.vips
                 if vip.interface_name == interface_name]
 
-    def set_notify(self, state, path):
-        if state not in VALID_NOTIFY_STATES:
-            raise InvalidNotifyStateException(state=state)
-        self.notifiers.append((state, path))
-
     def _build_track_interface_config(self):
         return itertools.chain(
             ['    track_interface {'],
@@ -234,10 +216,6 @@ class KeepalivedInstance(object):
                                 for route in self.virtual_routes),
                                ['    }'])
 
-    def _build_notify_scripts(self):
-        return itertools.chain(('    notify_%s "%s"' % (state, path)
-                                for state, path in self.notifiers))
-
     def build_config(self):
         config = ['vrrp_instance %s {' % self.name,
                   '    state %s' % self.state,
@@ -270,9 +248,6 @@ class KeepalivedInstance(object):
         if self.virtual_routes:
             config.extend(self._build_virtual_routes_config())
 
-        if self.notifiers:
-            config.extend(self._build_notify_scripts())
-
         config.append('}')
 
         return config
@@ -309,53 +284,7 @@ class KeepalivedConf(object):
         return '\n'.join(self.build_config())
 
 
-class KeepalivedNotifierMixin(object):
-    def _get_notifier_path(self, state):
-        return self._get_full_config_file_path('notify_%s.sh' % state)
-
-    def _write_notify_script(self, state, script):
-        name = self._get_notifier_path(state)
-        utils.replace_file(name, script)
-        st = os.stat(name)
-        os.chmod(name, st.st_mode | stat.S_IEXEC)
-
-        return name
-
-    def _prepend_shebang(self, script):
-        return '#!/bin/sh\n%s' % script
-
-    def _append_state(self, script, state):
-        state_path = self._get_full_config_file_path('state')
-        return '%s\necho -n %s > %s' % (script, state, state_path)
-
-    def add_notifier(self, script, state, vrouter_id):
-        """Add a master, backup or fault notifier.
-
-        These notifiers are executed when keepalived invokes a state
-        transition. Write a notifier to disk and add it to the
-        configuration.
-        """
-
-        script_with_prefix = self._prepend_shebang(' '.join(script))
-        full_script = self._append_state(script_with_prefix, state)
-        self._write_notify_script(state, full_script)
-
-        vr_instance = self.config.get_instance(vrouter_id)
-        vr_instance.set_notify(state, self._get_notifier_path(state))
-
-    def get_conf_dir(self):
-        confs_dir = os.path.abspath(os.path.normpath(self.conf_path))
-        conf_dir = os.path.join(confs_dir, self.resource_id)
-        return conf_dir
-
-    def _get_full_config_file_path(self, filename, ensure_conf_dir=True):
-        conf_dir = self.get_conf_dir()
-        if ensure_conf_dir:
-            utils.ensure_dir(conf_dir)
-        return os.path.join(conf_dir, filename)
-
-
-class KeepalivedManager(KeepalivedNotifierMixin):
+class KeepalivedManager(object):
     """Wrapper for keepalived.
 
     This wrapper permits to write keepalived config files, to start/restart
@@ -372,6 +301,17 @@ class KeepalivedManager(KeepalivedNotifierMixin):
         self.conf_path = conf_path
         self.process = None
 
+    def get_conf_dir(self):
+        confs_dir = os.path.abspath(os.path.normpath(self.conf_path))
+        conf_dir = os.path.join(confs_dir, self.resource_id)
+        return conf_dir
+
+    def _get_full_config_file_path(self, filename, ensure_conf_dir=True):
+        conf_dir = self.get_conf_dir()
+        if ensure_conf_dir:
+            utils.ensure_dir(conf_dir)
+        return os.path.join(conf_dir, filename)
+
     def _output_config_file(self):
         config_str = self.config.get_config_str()
         config_path = self._get_full_config_file_path('keepalived.conf')

neutron/cmd/keepalived_state_change.py

@@ -0,0 +1,19 @@
+# Copyright (c) 2015 Red Hat Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+from neutron.agent.l3 import keepalived_state_change
+
+
+def main():
+    keepalived_state_change.main()

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

@@ -0,0 +1,73 @@
+# Copyright (c) 2015 Red Hat Inc.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+
+import mock
+from oslo_config import cfg
+
+from neutron.agent.l3 import keepalived_state_change
+from neutron.openstack.common import uuidutils
+from neutron.tests.functional import base
+
+
+class TestKeepalivedStateChange(base.BaseSudoTestCase):
+    def setUp(self):
+        super(TestKeepalivedStateChange, self).setUp()
+        cfg.CONF.register_opt(
+            cfg.StrOpt('metadata_proxy_socket',
+                       default='$state_path/metadata_proxy',
+                       help=_('Location of Metadata Proxy UNIX domain '
+                              'socket')))
+
+        self.router_id = uuidutils.generate_uuid()
+        self.conf_dir = self.get_default_temp_dir().path
+        self.cidr = '169.254.128.1/24'
+        self.interface_name = 'interface'
+        self.monitor = keepalived_state_change.MonitorDaemon(
+            self.get_temp_file_path('monitor.pid'),
+            self.router_id,
+            1,
+            2,
+            'namespace',
+            self.conf_dir,
+            self.interface_name,
+            self.cidr)
+        mock.patch.object(self.monitor, 'notify_agent').start()
+        self.line = '1: %s    inet %s' % (self.interface_name, self.cidr)
+
+    def test_parse_and_handle_event_wrong_device_completes_without_error(self):
+        self.monitor.parse_and_handle_event(
+            '1: wrong_device    inet wrong_cidr')
+
+    def _get_state(self):
+        with open(os.path.join(self.monitor.conf_dir, 'state')) as state_file:
+            return state_file.read()
+
+    def test_parse_and_handle_event_writes_to_file(self):
+        self.monitor.parse_and_handle_event('Deleted %s' % self.line)
+        self.assertEqual('backup', self._get_state())
+
+        self.monitor.parse_and_handle_event(self.line)
+        self.assertEqual('master', self._get_state())
+
+    def test_parse_and_handle_event_fails_writing_state(self):
+        with mock.patch.object(
+                self.monitor, 'write_state_change', side_effect=OSError):
+            self.monitor.parse_and_handle_event(self.line)
+
+    def test_parse_and_handle_event_fails_notifying_agent(self):
+        with mock.patch.object(
+                self.monitor, 'notify_agent', side_effect=Exception):
+            self.monitor.parse_and_handle_event(self.line)

neutron/tests/functional/agent/test_l3_agent.py

@@ -147,7 +147,6 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
             expected_device['mac_address'], namespace)
 
     def get_expected_keepalive_configuration(self, router):
-        ha_confs_path = self.agent.conf.ha_confs_path
         router_id = router.router_id
         ha_device_name = router.get_ha_device_name(router.ha_port['id'])
         ha_device_cidr = router.ha_port['ip_cidr']
@@ -191,11 +190,7 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
         0.0.0.0/0 via %(default_gateway_ip)s dev %(external_device_name)s
         8.8.8.0/24 via 19.4.4.4
     }
-    notify_master "%(ha_confs_path)s/%(router_id)s/notify_master.sh"
-    notify_backup "%(ha_confs_path)s/%(router_id)s/notify_backup.sh"
-    notify_fault "%(ha_confs_path)s/%(router_id)s/notify_fault.sh"
 }""" % {
-            'ha_confs_path': ha_confs_path,
             'router_id': router_id,
             'ha_device_name': ha_device_name,
             'ha_device_cidr': ha_device_cidr,
@@ -219,7 +214,8 @@ class L3AgentTestFramework(base.BaseOVSLinuxTestCase):
         # then the devices and iptable rules have also been deleted,
         # so there's no need to check that explicitly.
         self.assertFalse(self._namespace_exists(router.ns_name))
-        self.assertFalse(self._metadata_proxy_exists(self.agent.conf, router))
+        utils.wait_until_true(
+            lambda: not self._metadata_proxy_exists(self.agent.conf, router))
 
     def _assert_snat_chains(self, router):
         self.assertFalse(router.iptables_manager.is_chain_empty(
@@ -283,6 +279,25 @@ class L3AgentTestCase(L3AgentTestFramework):
     def test_observer_notifications_ha_router(self):
         self._test_observer_notifications(enable_ha=True)
 
+    def test_keepalived_state_change_notification(self):
+        enqueue_mock = mock.patch.object(
+            self.agent, 'enqueue_state_change').start()
+        router_info = self.generate_router_info(enable_ha=True)
+        router = self.manage_router(self.agent, router_info)
+        utils.wait_until_true(lambda: router.ha_state == 'master')
+
+        device_name = router.get_ha_device_name(
+            router.router[l3_constants.HA_INTERFACE_KEY]['id'])
+        ha_device = ip_lib.IPDevice(device_name, router.ns_name)
+        ha_device.link.set_down()
+        utils.wait_until_true(lambda: router.ha_state == 'backup')
+
+        utils.wait_until_true(lambda: enqueue_mock.call_count == 3)
+        calls = [args[0] for args in enqueue_mock.call_args_list]
+        self.assertEqual((router.router_id, 'backup'), calls[0])
+        self.assertEqual((router.router_id, 'master'), calls[1])
+        self.assertEqual((router.router_id, 'backup'), calls[2])
+
     def _test_observer_notifications(self, enable_ha):
         """Test create, update, delete of router and notifications."""
         with mock.patch.object(
@@ -419,7 +434,8 @@ class L3AgentTestCase(L3AgentTestFramework):
             utils.wait_until_true(device_exists)
 
         self.assertTrue(self._namespace_exists(router.ns_name))
-        self.assertTrue(self._metadata_proxy_exists(self.agent.conf, router))
+        utils.wait_until_true(
+            lambda: self._metadata_proxy_exists(self.agent.conf, router))
         self._assert_internal_devices(router)
         self._assert_external_device(router)
         if ip_version == 4:
@@ -538,7 +554,7 @@ class L3HATestFramework(L3AgentTestFramework):
         ha_device.link.set_down()
 
         utils.wait_until_true(lambda: router2.ha_state == 'master')
-        utils.wait_until_true(lambda: router1.ha_state == 'fault')
+        utils.wait_until_true(lambda: router1.ha_state == 'backup')
 
 
 class MetadataFakeProxyHandler(object):

neutron/tests/unit/agent/linux/test_keepalived.py

@@ -66,7 +66,6 @@ class KeepalivedConfBaseMixin(object):
                                                   advert_int=5)
         instance1.set_authentication('AH', 'pass123')
         instance1.track_interfaces.append("eth0")
-        instance1.set_notify('master', '/tmp/script.sh')
 
         vip_address1 = keepalived.KeepalivedVipAddress('192.168.1.0/24',
                                                        'eth1')
@@ -136,7 +135,6 @@ class KeepalivedConfTestCase(base.BaseTestCase,
     virtual_routes {
         0.0.0.0/0 via 192.168.1.1 dev eth1
     }
-    notify_master "/tmp/script.sh"
 }
 vrrp_instance VR_2 {
     state MASTER
@@ -177,20 +175,12 @@ vrrp_instance VR_2 {
 
 class KeepalivedStateExceptionTestCase(base.BaseTestCase):
     def test_state_exception(self):
-        instance = keepalived.KeepalivedInstance('MASTER', 'eth0', 1,
-                                                 '169.254.192.0/18')
-
-        invalid_notify_state = 'a seal walks'
-        self.assertRaises(keepalived.InvalidNotifyStateException,
-                          instance.set_notify,
-                          invalid_notify_state, '/tmp/script.sh')
-
-        invalid_vrrp_state = 'into a club'
+        invalid_vrrp_state = 'a seal walks'
         self.assertRaises(keepalived.InvalidInstanceStateException,
                           keepalived.KeepalivedInstance,
                           invalid_vrrp_state, 'eth0', 33, '169.254.192.0/18')
 
-        invalid_auth_type = '[hip, hip]'
+        invalid_auth_type = 'into a club'
         instance = keepalived.KeepalivedInstance('MASTER', 'eth0', 1,
                                                  '169.254.192.0/18')
         self.assertRaises(keepalived.InvalidAuthenticationTypeException,
@@ -233,7 +223,6 @@ class KeepalivedInstanceTestCase(base.BaseTestCase,
     virtual_routes {
         0.0.0.0/0 via 192.168.1.1 dev eth1
     }
-    notify_master "/tmp/script.sh"
 }
 vrrp_instance VR_2 {
     state MASTER

neutron/tests/unit/agent/metadata/test_driver.py

@@ -63,6 +63,7 @@ class TestMetadataDriverProcess(base.BaseTestCase):
 
     def setUp(self):
         super(TestMetadataDriverProcess, self).setUp()
+        mock.patch('eventlet.spawn').start()
         agent_config.register_interface_driver_opts_helper(cfg.CONF)
         cfg.CONF.set_override('interface_driver',
                               'neutron.agent.linux.interface.NullDriver')

neutron/tests/unit/test_l3_agent.py

@@ -177,6 +177,7 @@ class BasicRouterOperationsFramework(base.BaseTestCase):
 
     def setUp(self):
         super(BasicRouterOperationsFramework, self).setUp()
+        mock.patch('eventlet.spawn').start()
         self.conf = agent_config.setup_conf()
         self.conf.register_opts(base_config.core_opts)
         log.register_options(self.conf)
@@ -200,7 +201,7 @@ class BasicRouterOperationsFramework(base.BaseTestCase):
         self.ensure_dir = mock.patch('neutron.agent.linux.utils'
                                      '.ensure_dir').start()
 
-        mock.patch('neutron.agent.linux.keepalived.KeepalivedNotifierMixin'
+        mock.patch('neutron.agent.linux.keepalived.KeepalivedManager'
                    '._get_full_config_file_path').start()
 
         self.utils_exec_p = mock.patch(

setup.cfg

@@ -92,6 +92,7 @@ console_scripts =
     neutron-debug = neutron.debug.shell:main
     neutron-dhcp-agent = neutron.cmd.eventlet.agents.dhcp:main
     neutron-hyperv-agent = neutron.plugins.hyperv.agent.hyperv_neutron_agent:main
+    neutron-keepalived-state-change = neutron.cmd.keepalived_state_change:main
     neutron-ibm-agent = neutron.plugins.ibm.agent.sdnve_neutron_agent:main
     neutron-l3-agent = neutron.cmd.eventlet.agents.l3:main
     neutron-linuxbridge-agent = neutron.plugins.linuxbridge.agent.linuxbridge_neutron_agent:main