Merge "openvswitch agent: add OVS_RESTARTED event"

neutron/callbacks/events.py

@@ -43,3 +43,5 @@ ABORT_DELETE = 'abort_delete'
 ABORT = 'abort_'
 BEFORE = 'before_'
 PRECOMMIT = 'precommit_'
+
+OVS_RESTARTED = 'ovs_restarted'

neutron/plugins/ml2/drivers/openvswitch/agent/ovs_neutron_agent.py

@@ -48,6 +48,7 @@ from neutron.api.rpc.handlers import dvr_rpc
 from neutron.api.rpc.handlers import securitygroups_rpc as sg_rpc
 from neutron.callbacks import events as callback_events
 from neutron.callbacks import registry
+from neutron.callbacks import resources as callback_resources
 from neutron.common import config
 from neutron.common import constants as c_const
 from neutron.common import topics
@@ -1968,6 +1969,11 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
                                                  self.patch_tun_ofport)
                     self.dvr_agent.reset_dvr_parameters()
                     self.dvr_agent.setup_dvr_flows()
+                # notify that OVS has restarted
+                registry.notify(
+                    callback_resources.AGENT,
+                    callback_events.OVS_RESTARTED,
+                    self)
                 # restart the polling manager so that it will signal as added
                 # all the current ports
                 # REVISIT (rossella_s) Define a method "reset" in

neutron/tests/functional/agent/test_l2_ovs_agent.py

@@ -14,8 +14,12 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
+import mock
 import time
 
+from neutron.callbacks import events
+from neutron.callbacks import registry
+from neutron.callbacks import resources
 from neutron.common import utils
 from neutron.plugins.ml2.drivers.openvswitch.agent.common import constants
 from neutron.tests.common import net_helpers
@@ -320,6 +324,24 @@ class TestOVSAgent(base.OVSAgentTestFramework):
             utils.WaitTimeout, self.wait_until_ports_state, [self.ports[1]],
             up=True, timeout=10)
 
+    def test_ovs_restarted_event(self):
+        callback = mock.Mock()
+
+        self.setup_agent_and_ports(
+            port_dicts=self.create_test_ports())
+
+        registry.subscribe(callback,
+                           resources.AGENT,
+                           events.OVS_RESTARTED)
+
+        self.agent.check_ovs_status.return_value = constants.OVS_RESTARTED
+
+        utils.wait_until_true(lambda: callback.call_count, timeout=10)
+
+        callback.assert_called_with(resources.AGENT,
+                                    events.OVS_RESTARTED,
+                                    mock.ANY)
+
 
 class TestOVSAgentExtensionConfig(base.OVSAgentTestFramework):
     def setUp(self):