Divide-and-conquer security group beasts

In one specific compute node, the security group rules
can be enormous quantity. This patch adds a step-by-step
processing method to deal with the large number of the
security group rules. And also changes or adds some LOG.

Related-Bug: #1813703
Related-Bug: #1813704
Related-Bug: #1813707

Change-Id: I57bf27ec75cf848271c5a28b22beee12b8bd5faa

neutron/agent/common/ovs_lib.py

@@ -447,10 +447,14 @@ class OVSBridge(BaseOVS):
         else:
             flow_strs = [_build_flow_expr_str(kw, action, strict)
                          for kw in kwargs_list]
+            LOG.debug("Processing %d OpenFlow rules.", len(flow_strs))
             if use_bundle:
                 extra_param.append('--bundle')
-            self.run_ofctl('%s-flows' % action, extra_param + ['-'],
-                           '\n'.join(flow_strs))
+
+            step = common_constants.AGENT_RES_PROCESSING_STEP
+            for i in range(0, len(flow_strs), step):
+                self.run_ofctl('%s-flows' % action, extra_param + ['-'],
+                               '\n'.join(flow_strs[i:i + step]))
 
     def add_flow(self, **kwargs):
         self.do_action_flows('add', [kwargs])

neutron/agent/securitygroups_rpc.py

@@ -22,6 +22,7 @@ from oslo_log import log as logging
 import oslo_messaging
 
 from neutron.agent import firewall
+from neutron.common import constants as common_constants
 from neutron.conf.agent import securitygroups_rpc as sc_cfg
 
 
@@ -132,12 +133,17 @@ class SecurityGroupAgentRpc(object):
         self._apply_port_filter(device_ids)
 
     def _apply_port_filter(self, device_ids, update_filter=False):
+        step = common_constants.AGENT_RES_PROCESSING_STEP
         if self.use_enhanced_rpc:
-            devices_info = self.plugin_rpc.security_group_info_for_devices(
-                self.context, list(device_ids))
-            devices = devices_info['devices']
-            security_groups = devices_info['security_groups']
-            security_group_member_ips = devices_info['sg_member_ips']
+            devices = {}
+            security_groups = {}
+            security_group_member_ips = {}
+            for i in range(0, len(device_ids), step):
+                devices_info = self.plugin_rpc.security_group_info_for_devices(
+                    self.context, list(device_ids)[i:i + step])
+                devices.update(devices_info['devices'])
+                security_groups.update(devices_info['security_groups'])
+                security_group_member_ips.update(devices_info['sg_member_ips'])
         else:
             devices = self.plugin_rpc.security_group_rules_for_devices(
                 self.context, list(device_ids))

neutron/common/constants.py

@@ -262,3 +262,15 @@ EXT_PARENT_PREFIX = 'ext_parent'
 
 RP_BANDWIDTHS = 'resource_provider_bandwidths'
 RP_INVENTORY_DEFAULTS = 'resource_provider_inventory_defaults'
+
+# Number of resources for neutron agent side functions to deal
+# with large sets.
+# Setting this value does not count on special conditions, it is just a human
+# countable or scalable number. [1] gives us the method to test the scale
+# issue. And we have tested the value of 1000, 500, 200, 100. But for 100,
+# ovs-agent will have a lower timeout probability. And according to the
+# testing result, step size 100 can indeed cost about 10% much more time
+# than 500/1000. But such extra time looks inevitably needed to be sacrificed
+# for the restart success rate.
+# [1] http://paste.openstack.org/show/745685/
+AGENT_RES_PROCESSING_STEP = 100

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

@@ -1776,8 +1776,8 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
         need_binding_devices = []
         skipped_devices = set()
         binding_no_activated_devices = set()
+        start = time.time()
         if devices_added_updated:
-            start = time.time()
             (skipped_devices, binding_no_activated_devices,
              need_binding_devices, failed_devices['added']) = (
                 self.treat_devices_added_or_updated(
@@ -1806,6 +1806,10 @@ class OVSNeutronAgent(l2population_rpc.L2populationRpcCallBackTunnelMixin,
         self._add_port_tag_info(need_binding_devices)
         self.sg_agent.setup_port_filters(added_ports,
                                          port_info.get('updated', set()))
+        LOG.info("process_network_ports - iteration:%(iter_num)d - "
+                 "agent port security group processed in %(elapsed).3f",
+                 {'iter_num': self.iter_num,
+                  'elapsed': time.time() - start})
         failed_devices['added'] |= self._bind_devices(need_binding_devices)
 
         if 'removed' in port_info and port_info['removed']:

neutron/tests/unit/agent/test_securitygroups_rpc.py

@@ -1001,7 +1001,7 @@ class SecurityGroupAgentEnhancedRpcTestCase(
                                         mock.call.defer_apply(),
                                         mock.call.remove_port_filter(
                                             self.fake_device),
-                                        ])
+                                        ], any_order=True)
 
     def test_security_groups_rule_updated_enhanced_rpc(self):
         sg_list = ['fake_sgid1', 'fake_sgid3']
@@ -1058,7 +1058,7 @@ class SecurityGroupAgentEnhancedRpcTestCase(
                  mock.call.update_port_filter(self.fake_device),
                  mock.call.process_trusted_ports([])]
 
-        self.firewall.assert_has_calls(calls)
+        self.firewall.assert_has_calls(calls, any_order=True)
 
     def test_refresh_firewall_devices_enhanced_rpc(self):
         self.agent.prepare_devices_filter(['fake_device'])
@@ -1081,7 +1081,7 @@ class SecurityGroupAgentEnhancedRpcTestCase(
                  mock.call.update_port_filter(self.fake_device),
                  mock.call.process_trusted_ports([]),
                  ]
-        self.firewall.assert_has_calls(calls)
+        self.firewall.assert_has_calls(calls, any_order=True)
 
     def test_refresh_firewall_none_enhanced_rpc(self):
         self.agent.refresh_firewall([])