Merge "Decrease rpc timeout after agent receives SIGTERM"

etc/neutron/plugins/openvswitch/ovs_neutron_plugin.ini

@@ -99,6 +99,11 @@
 #
 # enable_distributed_routing = False
 
+# (IntOpt) Set new timeout in seconds for new rpc calls after agent receives
+# SIGTERM. If value is set to 0, rpc timeout won't be changed"
+#
+# quitting_rpc_timeout = 10
+
 [securitygroup]
 # Firewall driver for realizing neutron security group function.
 # firewall_driver = neutron.agent.firewall.NoopFirewallDriver

neutron/plugins/openvswitch/agent/ovs_neutron_agent.py

@@ -130,7 +130,8 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
                  ovsdb_monitor_respawn_interval=(
                      constants.DEFAULT_OVSDBMON_RESPAWN),
                  arp_responder=False,
-                 use_veth_interconnection=False):
+                 use_veth_interconnection=False,
+                 quitting_rpc_timeout=None):
         '''Constructor.
 
         :param integ_br: name of the integration bridge.
@@ -153,6 +154,8 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
                supported.
         :param use_veth_interconnection: use veths instead of patch ports to
                interconnect the integration bridge to physical bridges.
+        :param quitting_rpc_timeout: timeout in seconds for rpc calls after
+               SIGTERM is received
         '''
         super(OVSNeutronAgent, self).__init__()
         self.use_veth_interconnection = use_veth_interconnection
@@ -259,6 +262,8 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
         # The initialization is complete; we can start receiving messages
         self.connection.consume_in_threads()
 
+        self.quitting_rpc_timeout = quitting_rpc_timeout
+
     def _report_state(self):
         # How many devices are likely used by a VM
         self.agent_state.get('configurations')['devices'] = (
@@ -1516,6 +1521,13 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
     def _handle_sigterm(self, signum, frame):
         LOG.debug("Agent caught SIGTERM, quitting daemon loop.")
         self.run_daemon_loop = False
+        if self.quitting_rpc_timeout:
+            self.set_rpc_timeout(self.quitting_rpc_timeout)
+
+    def set_rpc_timeout(self, timeout):
+        for rpc_api in (self.plugin_rpc, self.sg_plugin_rpc,
+                        self.dvr_plugin_rpc, self.state_rpc):
+            rpc_api.client.timeout = timeout
 
 
 def _ofport_set_to_str(ofport_set):
@@ -1547,6 +1559,7 @@ def create_agent_config_map(config):
         l2_population=config.AGENT.l2_population,
         arp_responder=config.AGENT.arp_responder,
         use_veth_interconnection=config.OVS.use_veth_interconnection,
+        quitting_rpc_timeout=config.AGENT.quitting_rpc_timeout
     )
 
     # Verify the tunnel_types specified are valid

neutron/plugins/openvswitch/common/config.py

@@ -79,6 +79,10 @@ agent_opts = [
                        "outgoing IP packet carrying GRE/VXLAN tunnel.")),
     cfg.BoolOpt('enable_distributed_routing', default=False,
                 help=_("Make the l2 agent run in DVR mode.")),
+    cfg.IntOpt('quitting_rpc_timeout', default=10,
+               help=_("Set new timeout in seconds for new rpc calls after "
+                      "agent receives SIGTERM. If value is set to 0, rpc "
+                      "timeout won't be changed"))
 ]
 
 

neutron/tests/unit/openvswitch/test_ovs_neutron_agent.py

@@ -99,6 +99,7 @@ class TestOvsNeutronAgent(base.BaseTestCase):
         cfg.CONF.set_default('firewall_driver',
                              'neutron.agent.firewall.NoopFirewallDriver',
                              group='SECURITYGROUP')
+        cfg.CONF.set_default('quitting_rpc_timeout', 10, 'AGENT')
         kwargs = ovs_neutron_agent.create_agent_config_map(cfg.CONF)
 
         class MockFixedIntervalLoopingCall(object):
@@ -1023,6 +1024,20 @@ class TestOvsNeutronAgent(base.BaseTestCase):
         setup_int_br.assert_has_calls([mock.call()])
         setup_phys_br.assert_has_calls([mock.call({})])
 
+    def test_set_rpc_timeout(self):
+        self.agent._handle_sigterm(None, None)
+        for rpc_client in (self.agent.plugin_rpc.client,
+                           self.agent.sg_plugin_rpc.client,
+                           self.agent.dvr_plugin_rpc.client,
+                           self.agent.state_rpc.client):
+            self.assertEqual(10, rpc_client.timeout)
+
+    def test_set_rpc_timeout_no_value(self):
+        self.agent.quitting_rpc_timeout = None
+        with mock.patch.object(self.agent, 'set_rpc_timeout') as mock_set_rpc:
+            self.agent._handle_sigterm(None, None)
+        self.assertFalse(mock_set_rpc.called)
+
 
 class AncillaryBridgesTest(base.BaseTestCase):