Apic drivers enhancements (second approach): Sync

- Model synchronization

Implements blueprint: apic-driver-enhancements

Change-Id: I4264afe5c140c3576951c1a5a28a8d7666481147

etc/neutron/plugins/ml2/ml2_conf_cisco.ini

@@ -50,47 +50,49 @@
 [ml2_cisco_apic]
 
 # Hostname:port list of APIC controllers
-# apic_hosts=1.1.1.1:80, 1.1.1.2:8080, 1.1.1.3:80
+# apic_hosts = 1.1.1.1:80, 1.1.1.2:8080, 1.1.1.3:80
 
 # Username for the APIC controller
-# apic_username=user
+# apic_username = user
 
 # Password for the APIC controller
-# apic_password=password
+# apic_password = password
 
 # Whether use SSl for connecting to the APIC controller or not
-# apic_use_ssl=True
+# apic_use_ssl = True
 
 # How to map names to APIC: use_uuid or use_name
-# apic_name_mapping=use_name
+# apic_name_mapping = use_name
 
 # Names for APIC objects used by Neutron
 # Note: When deploying multiple clouds against one APIC,
 #       these names must be unique between the clouds.
-# apic_vmm_domain=openstack
+# apic_vmm_domain = openstack
-# apic_vlan_ns_name=openstack_ns
+# apic_vlan_ns_name = openstack_ns
-# apic_node_profile=openstack_profile
+# apic_node_profile = openstack_profile
-# apic_entity_profile=openstack_entity
+# apic_entity_profile = openstack_entity
-# apic_function_profile=openstack_function
+# apic_function_profile = openstack_function
-# apic_app_profile_name=openstack_app
+# apic_app_profile_name = openstack_app
+# Agent timers for State reporting and topology discovery
+# apic_sync_interval = 30
 
 # Specify your network topology.
 # This section indicates how your compute nodes are connected to the fabric's
 # switches and ports. The format is as follows:
 #
 # [apic_switch:<swich_id_from_the_apic>]
-# <compute_host>,<compute_host>=<switchport_the_host(s)_are_connected_to>
+# <compute_host>,<compute_host> = <switchport_the_host(s)_are_connected_to>
 #
 # You can have multiple sections, one for each switch in your fabric that is
 # participating in Openstack. e.g.
 #
 # [apic_switch:17]
-# ubuntu,ubuntu1=1/10
+# ubuntu,ubuntu1 = 1/10
-# ubuntu2,ubuntu3=1/11
+# ubuntu2,ubuntu3 = 1/11
 #
 # [apic_switch:18]
-# ubuntu5,ubuntu6=1/1
+# ubuntu5,ubuntu6 = 1/1
-# ubuntu7,ubuntu8=1/2
+# ubuntu7,ubuntu8 = 1/2
 
 # Describe external connectivity.
 # In this section you can specify the external network configuration in order
@@ -99,11 +101,11 @@
 # format is as follows:
 #
 # [apic_external_network:<externalNetworkName>]
-# switch=<switch_id_from_the_apic>
+# switch = <switch_id_from_the_apic>
-# port=<switchport_the_external_router_is_connected_to>
+# port = <switchport_the_external_router_is_connected_to>
-# encap=<encapsulation>
+# encap = <encapsulation>
-# cidr_exposed=<cidr_exposed_to_the_external_router>
+# cidr_exposed = <cidr_exposed_to_the_external_router>
-# gateway_ip=<ip_of_the_external_gateway>
+# gateway_ip = <ip_of_the_external_gateway>
 #
 # An example follows:
 # [apic_external_network:network_ext]

neutron/plugins/ml2/drivers/cisco/apic/apic_sync.py

@@ -0,0 +1,111 @@
+# Copyright (c) 2014 Cisco Systems Inc.
+# All Rights Reserved.
+#
+#    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.
+#
+# @author: Ivar Lazzaro (ivar-lazzaro), Cisco Systems Inc.
+
+from neutron.common import constants as n_constants
+from neutron import context
+from neutron import manager
+from neutron.openstack.common.gettextutils import _LW
+from neutron.openstack.common import log
+from neutron.openstack.common import loopingcall
+from neutron.plugins.ml2 import db as l2_db
+from neutron.plugins.ml2 import driver_context
+
+LOG = log.getLogger(__name__)
+
+
+class SynchronizerBase(object):
+
+    def __init__(self, driver, interval=None):
+        self.core_plugin = manager.NeutronManager.get_plugin()
+        self.driver = driver
+        self.interval = interval
+
+    def sync(self, f, *args, **kwargs):
+        """Fire synchronization based on interval.
+
+        Interval can be 0 for 'sync once' >0 for 'sync periodically' and
+        <0 for 'no sync'
+        """
+        if self.interval:
+            if self.interval > 0:
+                loop_call = loopingcall.FixedIntervalLoopingCall(f, *args,
+                                                                 **kwargs)
+                loop_call.start(interval=self.interval)
+                return loop_call
+        else:
+            # Fire once
+            f(*args, **kwargs)
+
+
+class ApicBaseSynchronizer(SynchronizerBase):
+
+    def sync_base(self):
+        self.sync(self._sync_base)
+
+    def _sync_base(self):
+        ctx = context.get_admin_context()
+        # Sync Networks
+        for network in self.core_plugin.get_networks(ctx):
+            mech_context = driver_context.NetworkContext(self.core_plugin, ctx,
+                                                         network)
+            try:
+                self.driver.create_network_postcommit(mech_context)
+            except Exception:
+                LOG.warn(_LW("Create network postcommit failed for "
+                             "network %s"), network['id'])
+
+        # Sync Subnets
+        for subnet in self.core_plugin.get_subnets(ctx):
+            mech_context = driver_context.SubnetContext(self.core_plugin, ctx,
+                                                        subnet)
+            try:
+                self.driver.create_subnet_postcommit(mech_context)
+            except Exception:
+                LOG.warn(_LW("Create subnet postcommit failed for"
+                             " subnet %s"), subnet['id'])
+
+        # Sync Ports (compute/gateway/dhcp)
+        for port in self.core_plugin.get_ports(ctx):
+            _, binding = l2_db.get_locked_port_and_binding(ctx.session,
+                                                           port['id'])
+            network = self.core_plugin.get_network(ctx, port['network_id'])
+            mech_context = driver_context.PortContext(self.core_plugin, ctx,
+                                                      port, network, binding)
+            try:
+                self.driver.create_port_postcommit(mech_context)
+            except Exception:
+                LOG.warn(_LW("Create port postcommit failed for"
+                             " port %s"), port['id'])
+
+
+class ApicRouterSynchronizer(SynchronizerBase):
+
+    def sync_router(self):
+        self.sync(self._sync_router)
+
+    def _sync_router(self):
+        ctx = context.get_admin_context()
+        # Sync Router Interfaces
+        filters = {'device_owner': [n_constants.DEVICE_OWNER_ROUTER_INTF]}
+        for interface in self.core_plugin.get_ports(ctx, filters=filters):
+            try:
+                self.driver.add_router_interface_postcommit(
+                    ctx, interface['device_id'],
+                    {'port_id': interface['id']})
+            except Exception:
+                LOG.warn(_LW("Add interface postcommit failed for "
+                             "port %s"), interface['id'])

neutron/plugins/ml2/drivers/cisco/apic/config.py

@@ -81,6 +81,9 @@ apic_opts = [
     cfg.StrOpt('root_helper',
                default=DEFAULT_ROOT_HELPER,
                help=_("Setup root helper as rootwrap or sudo")),
+    cfg.IntOpt('apic_sync_interval',
+               default=0,
+               help=_("Synchronization interval in seconds")),
 ]
 
 

neutron/plugins/ml2/drivers/cisco/apic/mechanism_apic.py

@@ -27,6 +27,7 @@ from neutron.openstack.common import log
 from neutron.plugins.common import constants
 from neutron.plugins.ml2 import driver_api as api
 from neutron.plugins.ml2.drivers.cisco.apic import apic_model
+from neutron.plugins.ml2.drivers.cisco.apic import apic_sync
 from neutron.plugins.ml2.drivers.cisco.apic import config
 from neutron.plugins.ml2 import models
 
@@ -54,13 +55,35 @@ class APICMechanismDriver(api.MechanismDriver):
                                         keyclient_param, keystone_authtoken,
                                         apic_system_id)
 
+    @staticmethod
+    def get_base_synchronizer(inst):
+        apic_config = cfg.CONF.ml2_cisco_apic
+        return apic_sync.ApicBaseSynchronizer(inst,
+                                              apic_config.apic_sync_interval)
+
+    @staticmethod
+    def get_router_synchronizer(inst):
+        apic_config = cfg.CONF.ml2_cisco_apic
+        return apic_sync.ApicRouterSynchronizer(inst,
+                                                apic_config.apic_sync_interval)
+
     def initialize(self):
         # initialize apic
         self.apic_manager = APICMechanismDriver.get_apic_manager()
         self.name_mapper = self.apic_manager.apic_mapper
+        self.synchronizer = None
         self.apic_manager.ensure_infra_created_on_apic()
         self.apic_manager.ensure_bgp_pod_policy_created_on_apic()
 
+    def sync_init(f):
+        def inner(inst, *args, **kwargs):
+            if not inst.synchronizer:
+                inst.synchronizer = (
+                    APICMechanismDriver.get_base_synchronizer(inst))
+                inst.synchronizer.sync_base()
+            return f(inst, *args, **kwargs)
+        return inner
+
     @lockutils.synchronized('apic-portlock')
     def _perform_path_port_operations(self, context, port):
         # Get network
@@ -172,6 +195,7 @@ class APICMechanismDriver(api.MechanismDriver):
             network_id = self.name_mapper.network(context, network_id)
             return tenant_id, network_id, gateway_ip
 
+    @sync_init
     def create_port_postcommit(self, context):
         self._perform_port_operations(context)
 
@@ -184,6 +208,7 @@ class APICMechanismDriver(api.MechanismDriver):
                 resource='Port', msg='Port device owner and id cannot be '
                                      'changed.')
 
+    @sync_init
     def update_port_postcommit(self, context):
         self._perform_port_operations(context)
 
@@ -197,6 +222,7 @@ class APICMechanismDriver(api.MechanismDriver):
         elif port.get('device_owner') == n_constants.DEVICE_OWNER_DHCP:
             self._delete_path_if_last(context)
 
+    @sync_init
     def create_network_postcommit(self, context):
         if not context.current.get('router:external'):
             tenant_id = context.current['tenant_id']
@@ -214,6 +240,10 @@ class APICMechanismDriver(api.MechanismDriver):
                 self.apic_manager.ensure_epg_created(
                     tenant_id, network_id, transaction=trs)
 
+    @sync_init
+    def update_network_postcommit(self, context):
+        super(APICMechanismDriver, self).update_network_postcommit(context)
+
     def delete_network_postcommit(self, context):
         if not context.current.get('router:external'):
             tenant_id = context.current['tenant_id']
@@ -236,6 +266,7 @@ class APICMechanismDriver(api.MechanismDriver):
                                                       context.current['id'])
                 self.apic_manager.delete_external_routed_network(network_id)
 
+    @sync_init
     def create_subnet_postcommit(self, context):
         info = self._get_subnet_info(context, context.current)
         if info:
@@ -244,6 +275,7 @@ class APICMechanismDriver(api.MechanismDriver):
             self.apic_manager.ensure_subnet_created_on_apic(
                 tenant_id, network_id, gateway_ip)
 
+    @sync_init
     def update_subnet_postcommit(self, context):
         if context.current['gateway_ip'] != context.original['gateway_ip']:
             with self.apic_manager.apic.transaction() as trs:

neutron/services/l3_router/l3_apic.py

@@ -36,6 +36,7 @@ class ApicL3ServicePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         super(ApicL3ServicePlugin, self).__init__()
         self.manager = mechanism_apic.APICMechanismDriver.get_apic_manager()
         self.name_mapper = self.manager.apic_mapper
+        self.synchronizer = None
         self.manager.ensure_infra_created_on_apic()
         self.manager.ensure_bgp_pod_policy_created_on_apic()
 
@@ -58,6 +59,16 @@ class ApicL3ServicePlugin(db_base_plugin_v2.NeutronDbPluginV2,
         """Returns string description of the plugin."""
         return _("L3 Router Service Plugin for basic L3 using the APIC")
 
+    def sync_init(f):
+        def inner(inst, *args, **kwargs):
+            if not inst.synchronizer:
+                inst.synchronizer = (
+                    mechanism_apic.APICMechanismDriver.
+                    get_router_synchronizer(inst))
+                inst.synchronizer.sync_router()
+            return f(inst, *args, **kwargs)
+        return inner
+
     def add_router_interface_postcommit(self, context, router_id,
                                         interface_info):
         # Update router's state first
@@ -121,12 +132,30 @@ class ApicL3ServicePlugin(db_base_plugin_v2.NeutronDbPluginV2,
 
     # Router API
 
+    @sync_init
+    def create_router(self, *args, **kwargs):
+        return super(ApicL3ServicePlugin, self).create_router(*args, **kwargs)
+
+    @sync_init
     def update_router(self, context, id, router):
         result = super(ApicL3ServicePlugin, self).update_router(context,
                                                                 id, router)
         self.update_router_postcommit(context, result)
         return result
 
+    @sync_init
+    def get_router(self, *args, **kwargs):
+        return super(ApicL3ServicePlugin, self).get_router(*args, **kwargs)
+
+    @sync_init
+    def get_routers(self, *args, **kwargs):
+        return super(ApicL3ServicePlugin, self).get_routers(*args, **kwargs)
+
+    @sync_init
+    def get_routers_count(self, *args, **kwargs):
+        return super(ApicL3ServicePlugin, self).get_routers_count(*args,
+                                                                  **kwargs)
+
     def delete_router(self, context, router_id):
         self.delete_router_precommit(context, router_id)
         result = super(ApicL3ServicePlugin, self).delete_router(context,
@@ -135,6 +164,7 @@ class ApicL3ServicePlugin(db_base_plugin_v2.NeutronDbPluginV2,
 
     # Router Interface API
 
+    @sync_init
     def add_router_interface(self, context, router_id, interface_info):
         # Create interface in parent
         result = super(ApicL3ServicePlugin, self).add_router_interface(

neutron/tests/unit/ml2/drivers/cisco/apic/test_cisco_apic_sync.py

@@ -0,0 +1,79 @@
+# Copyright (c) 2014 Cisco Systems
+# All Rights Reserved.
+#
+#    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.
+#
+# @author: Ivar Lazzaro (ivarlazzaro@gmail.com), Cisco Systems, Inc.
+
+import sys
+
+import mock
+
+sys.modules["apicapi"] = mock.Mock()
+
+from neutron.plugins.ml2.drivers.cisco.apic import apic_sync
+from neutron.tests import base
+
+LOOPING_CALL = 'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall'
+GET_PLUGIN = 'neutron.manager.NeutronManager.get_plugin'
+GET_ADMIN_CONTEXT = 'neutron.context.get_admin_context'
+L2_DB = 'neutron.plugins.ml2.db.get_locked_port_and_binding'
+NETWORK_CONTEXT = 'neutron.plugins.ml2.driver_context.NetworkContext'
+SUBNET_CONTEXT = 'neutron.plugins.ml2.driver_context.SubnetContext'
+PORT_CONTEXT = 'neutron.plugins.ml2.driver_context.PortContext'
+
+
+class TestCiscoApicSync(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestCiscoApicSync, self).setUp()
+        self.driver = mock.Mock()
+        # Patch looping call
+        loopingcall_c = mock.patch(LOOPING_CALL).start()
+        self.loopingcall = mock.Mock()
+        loopingcall_c.return_value = self.loopingcall
+        # Patch get plugin
+        self.get_plugin = mock.patch(GET_PLUGIN).start()
+        self.get_plugin.return_value = mock.Mock()
+        # Patch get admin context
+        self.get_admin_context = mock.patch(GET_ADMIN_CONTEXT).start()
+        self.get_admin_context.return_value = mock.Mock()
+        # Patch get locked port and binding
+        self.get_locked_port_and_binding = mock.patch(L2_DB).start()
+        self.get_locked_port_and_binding.return_value = [mock.Mock()] * 2
+        # Patch driver context
+        mock.patch(NETWORK_CONTEXT).start()
+        mock.patch(SUBNET_CONTEXT).start()
+        mock.patch(PORT_CONTEXT).start()
+
+    def test_sync_base(self):
+        sync = apic_sync.ApicBaseSynchronizer(self.driver)
+        sync.core_plugin = mock.Mock()
+        sync.core_plugin.get_networks.return_value = [{'id': 'net'}]
+        sync.core_plugin.get_subnets.return_value = [{'id': 'sub'}]
+        sync.core_plugin.get_ports.return_value = [{'id': 'port',
+                                                    'network_id': 'net'}]
+        sync.sync_base()
+        self.driver.create_network_postcommit.assert_called_once()
+        self.driver.create_subnet_postcommit.assert_called_once()
+        self.get_locked_port_and_binding.assert_called_once()
+        self.driver.create_port_postcommit.assert_called_once()
+
+    def test_sync_router(self):
+        sync = apic_sync.ApicRouterSynchronizer(self.driver)
+        sync.core_plugin = mock.Mock()
+        sync.core_plugin.get_ports.return_value = [{'id': 'port',
+                                                    'network_id': 'net',
+                                                    'device_id': 'dev'}]
+        sync.sync_router()
+        self.driver.add_router_interface_postcommit.assert_called_once()