BGP Dynamic Routing: introduce BgpDrAgent

This patch implements a new agent named "BgpDrAgent". The new agent
will host different BGP speaking drivers and makes the required BGP
peering session/s for neutron. The agent takes the needed "peer/s and
route/s" information from the BGP speaker entity and synchronize the
same to the registerd driver.

For realizing HA, two BgpDrAgents should host the same BGP speaker.

Partially-Implements: blueprint bgp-dynamic-routing
Co-Authored-By: Ryan Tidwell <ryan.tidwell@hpe.com>
Co-Authored-By: Jaume Devesa <devvesa@gmail.com>
Co-Authored-By: Numan Siddique <nusiddiq@redhat.com>
Change-Id: I3217795bdd0fa2d9d4b39274f4f95fc013c8d29d

devstack/lib/bgp

@@ -5,3 +5,21 @@ function configure_bgp_service_plugin {
 function configure_bgp {
     configure_bgp_service_plugin
 }
+
+function configure_bgp_dragent {
+    cp $NEUTRON_DIR/etc/bgp_dragent.ini.sample $Q_BGP_DRAGENT_CONF_FILE
+
+    iniset $Q_BGP_DRAGENT_CONF_FILE DEFAULT verbose True
+    iniset $Q_BGP_DRAGENT_CONF_FILE DEFAULT debug $ENABLE_DEBUG_LOG_LEVEL
+    if [ -n "$BGP_ROUTER_ID" ]; then
+        iniset $Q_BGP_DRAGENT_CONF_FILE BGP bgp_router_id $BGP_ROUTER_ID
+    fi
+}
+
+function start_bgp_dragent {
+    run_process q-bgp-agt "$AGENT_BGP_BINARY --config-file $NEUTRON_CONF --config-file /$Q_BGP_DRAGENT_CONF_FILE"
+}
+
+function stop_bgp_dragent {
+    stop_process q-bgp-agt
+}

devstack/plugin.sh

@@ -24,6 +24,9 @@ if [[ "$1" == "stack" ]]; then
             if is_service_enabled q-agt; then
                 configure_l2_agent
             fi
+            if is_service_enabled q-bgp && is_service_enabled q-bgp-agt; then
+                configure_bgp_dragent
+            fi
             #Note: sriov agent should run with OVS or linux bridge agent
             #because they are the mechanisms that bind the DHCP and router ports.
             #Currently devstack lacks the option to run two agents on the same node.
@@ -39,10 +42,16 @@ if [[ "$1" == "stack" ]]; then
             if is_service_enabled q-sriov-agt; then
                 start_l2_agent_sriov
             fi
+            if is_service_enabled q-bgp && is_service_enabled q-bgp-agt; then
+                start_bgp_dragent
+            fi
             ;;
     esac
 elif [[ "$1" == "unstack" ]]; then
     if is_service_enabled q-sriov-agt; then
         stop_l2_agent_sriov
     fi
+    if is_service_enabled q-bgp && is_service_enabled q-bgp-agt; then
+        stop_bgp_dragent
+    fi
 fi

devstack/settings

@@ -1 +1,6 @@
 L2_AGENT_EXTENSIONS=${L2_AGENT_EXTENSIONS:-}
+
+#BGP binary and config information
+AGENT_BGP_BINARY=${AGENT_BGP_BINARY:-"$NEUTRON_BIN_DIR/neutron-bgp-dragent"}
+Q_BGP_DRAGENT_CONF_FILE=${Q_BGP_DRAGENT_CONF_FILE:-"$NEUTRON_CONF_DIR/bgp_dragent.ini"}
+BGP_ROUTER_ID=${BGP_ROUTER_ID:-}

etc/oslo-config-generator/bgp_dragent.ini

@@ -0,0 +1,7 @@
+[DEFAULT]
+output_file = etc/bgp_dragent.ini.sample
+wrap_width = 79
+
+namespace = neutron.base.agent
+namespace = neutron.bgp.agent
+namespace = oslo.log

neutron/api/rpc/agentnotifiers/bgp_dr_rpc_agent_api.py

@@ -0,0 +1,105 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 oslo_messaging
+
+from neutron.common import rpc as n_rpc
+from neutron.services.bgp.common import constants as bgp_consts
+
+
+class BgpDrAgentNotifyApi(object):
+    """API for plugin to notify BGP DrAgent.
+
+    This class implements the client side of an rpc interface.  The server side
+    is neutron.services.bgp_speaker.agent.bgp_dragent.BgpDrAgent. For more
+    information about rpc interfaces, please see doc/source/devref/rpc_api.rst.
+    """
+
+    def __init__(self, topic=bgp_consts.BGP_DRAGENT):
+        target = oslo_messaging.Target(topic=topic, version='1.0')
+        self.client = n_rpc.get_client(target)
+        self.topic = topic
+
+    def bgp_routes_advertisement(self, context, bgp_speaker_id,
+                                 routes, host):
+        """Tell BgpDrAgent to begin advertising the given route.
+
+        Invoked on FIP association, adding router port to a tenant network,
+        and new DVR port-host bindings, and subnet creation(?).
+        """
+        self._notification_host_cast(context, 'bgp_routes_advertisement_end',
+                {'advertise_routes': {'speaker_id': bgp_speaker_id,
+                                      'routes': routes}}, host)
+
+    def bgp_routes_withdrawal(self, context, bgp_speaker_id,
+                              routes, host):
+        """Tell BgpDrAgent to stop advertising the given route.
+
+        Invoked on FIP disassociation, removal of a router port on a
+        network, and removal of DVR port-host binding, and subnet delete(?).
+        """
+        self._notification_host_cast(context, 'bgp_routes_withdrawal_end',
+                {'withdraw_routes': {'speaker_id': bgp_speaker_id,
+                                     'routes': routes}}, host)
+
+    def bgp_peer_disassociated(self, context, bgp_speaker_id,
+                               bgp_peer_ip, host):
+        """Tell BgpDrAgent about a new BGP Peer association.
+
+        This effectively tells the BgpDrAgent to stop a peering session.
+        """
+        self._notification_host_cast(context, 'bgp_peer_disassociation_end',
+                {'bgp_peer': {'speaker_id': bgp_speaker_id,
+                              'peer_ip': bgp_peer_ip}}, host)
+
+    def bgp_peer_associated(self, context, bgp_speaker_id,
+                            bgp_peer_id, host):
+        """Tell BgpDrAgent about a BGP Peer disassociation.
+
+        This effectively tells the bgp_dragent to open a peering session.
+        """
+        self._notification_host_cast(context, 'bgp_peer_association_end',
+                {'bgp_peer': {'speaker_id': bgp_speaker_id,
+                              'peer_id': bgp_peer_id}}, host)
+
+    def bgp_speaker_created(self, context, bgp_speaker_id, host):
+        """Tell BgpDrAgent about the creation of a BGP Speaker.
+
+        Because a BGP Speaker can be created with BgpPeer binding in place,
+        we need to inform the BgpDrAgent of a new BGP Speaker in case a
+        peering session needs to opened immediately.
+        """
+        self._notification_host_cast(context, 'bgp_speaker_create_end',
+                {'bgp_speaker': {'id': bgp_speaker_id}}, host)
+
+    def bgp_speaker_removed(self, context, bgp_speaker_id, host):
+        """Tell BgpDrAgent about the removal of a BGP Speaker.
+
+        Because a BGP Speaker can be removed with BGP Peer binding in
+        place, we need to inform the BgpDrAgent of the removal of a
+        BGP Speaker in case peering sessions need to be stopped.
+        """
+        self._notification_host_cast(context, 'bgp_speaker_remove_end',
+                {'bgp_speaker': {'id': bgp_speaker_id}}, host)
+
+    def _notification_host_cast(self, context, method, payload, host):
+        """Send payload to BgpDrAgent in the cast mode"""
+        cctxt = self.client.prepare(topic=self.topic, server=host)
+        cctxt.cast(context, method, payload=payload)
+
+    def _notification_host_call(self, context, method, payload, host):
+        """Send payload to BgpDrAgent in the call mode"""
+        cctxt = self.client.prepare(topic=self.topic, server=host)
+        cctxt.call(context, method, payload=payload)

neutron/api/rpc/handlers/bgp_speaker_rpc.py

@@ -0,0 +1,65 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 oslo_messaging
+
+from neutron.extensions import bgp as bgp_ext
+from neutron import manager
+
+
+class BgpSpeakerRpcCallback(object):
+    """BgpDrAgent RPC callback in plugin implementations.
+
+    This class implements the server side of an RPC interface.
+    The client side of this interface can be found in
+    neutron.services.bgp_speaker.agent.bgp_dragent.BgpDrPluginApi.
+    For more information about changing RPC interfaces,
+    see doc/source/devref/rpc_api.rst.
+    """
+
+    # API version history:
+    # 1.0 BGPDRPluginApi BASE_RPC_API_VERSION
+    target = oslo_messaging.Target(version='1.0')
+
+    @property
+    def plugin(self):
+        if not hasattr(self, '_plugin'):
+            self._plugin = manager.NeutronManager.get_service_plugins().get(
+                bgp_ext.BGP_EXT_ALIAS)
+        return self._plugin
+
+    def get_bgp_speaker_info(self, context, bgp_speaker_id):
+        """Return BGP Speaker details such as peer list and local_as.
+
+        Invoked by the BgpDrAgent to lookup the details of a BGP Speaker.
+        """
+        return self.plugin.get_bgp_speaker_with_advertised_routes(
+                                context, bgp_speaker_id)
+
+    def get_bgp_peer_info(self, context, bgp_peer_id):
+        """Return BgpPeer details such as IP, remote_as, and credentials.
+
+        Invoked by the BgpDrAgent to lookup the details of a BGP peer.
+        """
+        return self.plugin.get_bgp_peer(context, bgp_peer_id,
+                                        ['peer_ip', 'remote_as',
+                                         'auth_type', 'password'])
+
+    def get_bgp_speakers(self, context, host=None, **kwargs):
+        """Returns the list of all BgpSpeakers.
+
+        Typically invoked by the BgpDrAgent as part of its bootstrap process.
+        """
+        return self.plugin.get_bgp_speakers_for_agent_host(context, host)

neutron/cmd/eventlet/agents/bgp_dragent.py

@@ -0,0 +1,20 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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.services.bgp.agent import entry as bgp_dragent
+
+
+def main():
+    bgp_dragent.main()

neutron/db/bgp_db.py

@@ -129,6 +129,20 @@ class BgpDbMixin(common_db.CommonDbMixin):
             bgp_speaker = self._get_bgp_speaker(context, bgp_speaker_id)
             return self._make_bgp_speaker_dict(bgp_speaker, fields)
 
+    def get_bgp_speaker_with_advertised_routes(self, context,
+                                               bgp_speaker_id):
+        bgp_speaker_attrs = ['id', 'local_as', 'tenant_id']
+        bgp_peer_attrs = ['peer_ip', 'remote_as', 'password']
+        with context.session.begin(subtransactions=True):
+            bgp_speaker = self.get_bgp_speaker(context, bgp_speaker_id,
+                                               fields=bgp_speaker_attrs)
+            res = dict((k, bgp_speaker[k]) for k in bgp_speaker_attrs)
+            res['peers'] = self.get_bgp_peers_by_bgp_speaker(context,
+                                                         bgp_speaker['id'],
+                                                         fields=bgp_peer_attrs)
+            res['advertised_routes'] = []
+            return res
+
     def update_bgp_speaker(self, context, bgp_speaker_id, bgp_speaker):
         bp = bgp_speaker[bgp_ext.BGP_SPEAKER_BODY_KEY_NAME]
         with context.session.begin(subtransactions=True):
@@ -211,6 +225,15 @@ class BgpDbMixin(common_db.CommonDbMixin):
                                     sorts=sorts, limit=limit,
                                     page_reverse=page_reverse)
 
+    def get_bgp_peers_by_bgp_speaker(self, context,
+                                     bgp_speaker_id, fields=None):
+        filters = [BgpSpeakerPeerBinding.bgp_speaker_id == bgp_speaker_id,
+                   BgpSpeakerPeerBinding.bgp_peer_id == BgpPeer.id]
+        with context.session.begin(subtransactions=True):
+            query = context.session.query(BgpPeer)
+            query = query.filter(*filters)
+            return [self._make_bgp_peer_dict(x) for x in query.all()]
+
     def get_bgp_peer(self, context, bgp_peer_id, fields=None):
         bgp_peer_db = self._get_bgp_peer(context, bgp_peer_id)
         return self._make_bgp_peer_dict(bgp_peer_db, fields=fields)

neutron/db/bgp_dragentscheduler_db.py

@@ -18,6 +18,7 @@ from oslo_db import exception as db_exc
 from oslo_log import log as logging
 import sqlalchemy as sa
 from sqlalchemy import orm
+from sqlalchemy.orm import exc
 
 from neutron._i18n import _
 from neutron._i18n import _LW
@@ -36,7 +37,7 @@ BGP_DRAGENT_SCHEDULER_OPTS = [
         'bgp_drscheduler_driver',
         default='neutron.services.bgp.scheduler'
                 '.bgp_dragent_scheduler.ChanceScheduler',
-        help=_('Driver used for scheduling BGP speakers to BGP DrAgent')),
+        help=_('Driver used for scheduling BGP speakers to BGP DrAgent'))
 ]
 
 cfg.CONF.register_opts(BGP_DRAGENT_SCHEDULER_OPTS)
@@ -73,7 +74,12 @@ class BgpDrAgentSchedulerDbMixin(bgp_dras_ext.BgpDrSchedulerPluginBase,
 
     def schedule_bgp_speaker(self, context, created_bgp_speaker):
         if self.bgp_drscheduler:
-            self.bgp_drscheduler.schedule(context, created_bgp_speaker)
+            agents = self.bgp_drscheduler.schedule(context,
+                                                   created_bgp_speaker)
+            for agent in agents:
+                self._bgp_rpc.bgp_speaker_created(context,
+                                                  created_bgp_speaker['id'],
+                                                  agent.host)
         else:
             LOG.warning(_LW("Cannot schedule BgpSpeaker to DrAgent. "
                             "Reason: No scheduler registered."))
@@ -107,6 +113,8 @@ class BgpDrAgentSchedulerDbMixin(bgp_dras_ext.BgpDrSchedulerPluginBase,
             binding.agent_id = agent_id
             context.session.add(binding)
 
+        self._bgp_rpc.bgp_speaker_created(context, speaker_id, agent_db.host)
+
     def remove_bgp_speaker_from_dragent(self, context, agent_id, speaker_id):
         with context.session.begin(subtransactions=True):
             agent_db = self._get_agent(context, agent_id)
@@ -129,6 +137,8 @@ class BgpDrAgentSchedulerDbMixin(bgp_dras_ext.BgpDrSchedulerPluginBase,
                       {'bgp_speaker_id': speaker_id,
                        'agent_id': agent_id})
 
+        self._bgp_rpc.bgp_speaker_removed(context, speaker_id, agent_db.host)
+
     def get_dragents_hosting_bgp_speakers(self, context, bgp_speaker_ids,
                                           active=None, admin_state_up=None):
         query = context.session.query(BgpSpeakerDrAgentBinding)
@@ -175,3 +185,31 @@ class BgpDrAgentSchedulerDbMixin(bgp_dras_ext.BgpDrSchedulerPluginBase,
         return {'bgp_speakers':
                 self.get_bgp_speakers(context,
                                       filters={'id': bgp_speaker_ids})}
+
+    def get_bgp_speakers_for_agent_host(self, context, host):
+        agent = self._get_agent_by_type_and_host(
+            context, bgp_consts.AGENT_TYPE_BGP_ROUTING, host)
+        if not agent.admin_state_up:
+            return {}
+
+        query = context.session.query(BgpSpeakerDrAgentBinding)
+        query = query.filter(BgpSpeakerDrAgentBinding.agent_id == agent.id)
+        try:
+            binding = query.one()
+        except exc.NoResultFound:
+            return []
+        bgp_speaker = self.get_bgp_speaker_with_advertised_routes(
+                                context, binding['bgp_speaker_id'])
+        return [bgp_speaker]
+
+    def get_bgp_speaker_by_speaker_id(self, context, bgp_speaker_id):
+        try:
+            return self.get_bgp_speaker(context, bgp_speaker_id)
+        except exc.NoResultFound:
+            return {}
+
+    def get_bgp_peer_by_peer_id(self, context, bgp_peer_id):
+        try:
+            return self.get_bgp_peer(context, bgp_peer_id)
+        except exc.NoResultFound:
+            return {}

neutron/extensions/bgp_dragentscheduler.py

@@ -169,3 +169,15 @@ class BgpDrSchedulerPluginBase(object):
     @abc.abstractmethod
     def list_bgp_speaker_on_dragent(self, context, agent_id):
         pass
+
+    @abc.abstractmethod
+    def get_bgp_speakers_for_agent_host(self, context, host):
+        pass
+
+    @abc.abstractmethod
+    def get_bgp_speaker_by_speaker_id(self, context, speaker_id):
+        pass
+
+    @abc.abstractmethod
+    def get_bgp_peer_by_peer_id(self, context, bgp_peer_id):
+        pass

neutron/services/bgp/agent/bgp_dragent.py

@@ -0,0 +1,631 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 collections
+
+from oslo_config import cfg
+from oslo_log import log as logging
+import oslo_messaging
+from oslo_service import loopingcall
+from oslo_service import periodic_task
+
+from neutron.agent import rpc as agent_rpc
+from neutron.common import constants
+from neutron.common import rpc as n_rpc
+from neutron.common import topics
+from neutron.common import utils
+from neutron import context
+from neutron.extensions import bgp as bgp_ext
+from neutron._i18n import _, _LE, _LI, _LW
+from neutron import manager
+from neutron.services.bgp.common import constants as bgp_consts
+
+LOG = logging.getLogger(__name__)
+
+
+class BgpDrAgent(manager.Manager):
+    """BGP Dynamic Routing agent service manager.
+
+    Note that the public methods of this class are exposed as the server side
+    of an rpc interface.  The neutron server uses
+    neutron.api.rpc.agentnotifiers.bgp_dr_rpc_agent_api.
+    BgpDrAgentNotifyApi as the client side to execute the methods
+    here. For more information about changing rpc interfaces, see
+    doc/source/devref/rpc_api.rst.
+
+    API version history:
+        1.0 initial Version
+    """
+    target = oslo_messaging.Target(version='1.0')
+
+    def __init__(self, host, conf=None):
+        super(BgpDrAgent, self).__init__()
+        self.conf = conf
+        self.needs_resync_reasons = collections.defaultdict(list)
+        self.needs_full_sync_reason = None
+
+        self.cache = BgpSpeakerCache()
+        self.context = context.get_admin_context_without_session()
+        self.plugin_rpc = BgpDrPluginApi(bgp_consts.BGP_PLUGIN,
+                                         self.context, host)
+
+    def after_start(self):
+        self.run()
+        LOG.info(_LI("BGP Dynamic Routing agent started"))
+
+    def run(self):
+        """Activate BGP Dynamic Routing agent."""
+        self.sync_state(self.context)
+        self.periodic_resync(self.context)
+
+    @utils.synchronized('bgp-dragent')
+    def sync_state(self, context, full_sync=None, bgp_speakers=None):
+        try:
+            hosted_bgp_speakers = self.plugin_rpc.get_bgp_speakers(context)
+            hosted_bgp_speaker_ids = [bgp_speaker['id']
+                                      for bgp_speaker in hosted_bgp_speakers]
+            cached_bgp_speakers = self.cache.get_bgp_speaker_ids()
+            for bgp_speaker_id in cached_bgp_speakers:
+                if bgp_speaker_id not in hosted_bgp_speaker_ids:
+                    self.remove_bgp_speaker_from_dragent(bgp_speaker_id)
+
+            resync_all = not bgp_speakers or full_sync
+            only_bs = set() if resync_all else set(bgp_speakers)
+            for hosted_bgp_speaker in hosted_bgp_speakers:
+                hosted_bs_id = hosted_bgp_speaker['id']
+                if resync_all or hosted_bs_id in only_bs:
+                    if not self.cache.is_bgp_speaker_added(hosted_bs_id):
+                        self.safe_configure_dragent_for_bgp_speaker(
+                            hosted_bgp_speaker)
+                        continue
+                    self.sync_bgp_speaker(hosted_bgp_speaker)
+                    resync_reason = "Periodic route cache refresh"
+                    self.schedule_resync(speaker_id=hosted_bs_id,
+                                         reason=resync_reason)
+        except Exception as e:
+            self.schedule_full_resync(reason=e)
+            LOG.error(_LE('Unable to sync BGP speaker state.'))
+
+    def sync_bgp_speaker(self, bgp_speaker):
+        # sync BGP Speakers
+        bgp_peer_ips = set(
+            [bgp_peer['peer_ip'] for bgp_peer in bgp_speaker['peers']])
+        cached_bgp_peer_ips = set(
+            self.cache.get_bgp_peer_ips(bgp_speaker['id']))
+        removed_bgp_peer_ips = cached_bgp_peer_ips - bgp_peer_ips
+
+        for bgp_peer_ip in removed_bgp_peer_ips:
+            self.remove_bgp_peer_from_bgp_speaker(bgp_speaker['id'],
+                                                  bgp_peer_ip)
+        if bgp_peer_ips:
+            self.add_bgp_peers_to_bgp_speaker(bgp_speaker)
+
+        # sync advertise routes
+        cached_adv_routes = self.cache.get_adv_routes(bgp_speaker['id'])
+        adv_routes = bgp_speaker['advertised_routes']
+        if cached_adv_routes == adv_routes:
+            return
+
+        for cached_route in cached_adv_routes:
+            if cached_route not in adv_routes:
+                self.withdraw_route_via_bgp_speaker(bgp_speaker['id'],
+                                                    bgp_speaker['local_as'],
+                                                    cached_route)
+
+        self.advertise_routes_via_bgp_speaker(bgp_speaker)
+
+    @utils.exception_logger()
+    def _periodic_resync_helper(self, context):
+        """Resync the BgpDrAgent state at the configured interval."""
+        if self.needs_resync_reasons or self.needs_full_sync_reason:
+            full_sync = self.needs_full_sync_reason
+            reasons = self.needs_resync_reasons
+            # Reset old reasons
+            self.needs_full_sync_reason = None
+            self.needs_resync_reasons = collections.defaultdict(list)
+            if full_sync:
+                LOG.debug("resync all: %(reason)s", {"reason": full_sync})
+            for bgp_speaker, reason in reasons.items():
+                LOG.debug("resync (%(bgp_speaker)s): %(reason)s",
+                          {"reason": reason, "bgp_speaker": bgp_speaker})
+            self.sync_state(
+                context, full_sync=full_sync, bgp_speakers=reasons.keys())
+
+    # NOTE: spacing is set 1 sec. The actual interval is controlled
+    # by neutron/service.py which defaults to CONF.periodic_interval
+    @periodic_task.periodic_task(spacing=1)
+    def periodic_resync(self, context):
+        LOG.debug("Started periodic resync.")
+        self._periodic_resync_helper(context)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_speaker_create_end(self, context, payload):
+        """Handle bgp_speaker_create_end notification event."""
+        bgp_speaker_id = payload['bgp_speaker']['id']
+        LOG.debug('Received BGP speaker create notification for '
+                  'speaker_id=%(speaker_id)s from the neutron server.',
+                  {'speaker_id': bgp_speaker_id})
+        self.add_bgp_speaker_helper(bgp_speaker_id)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_speaker_remove_end(self, context, payload):
+        """Handle bgp_speaker_create_end notification event."""
+
+        bgp_speaker_id = payload['bgp_speaker']['id']
+        LOG.debug('Received BGP speaker remove notification for '
+                  'speaker_id=%(speaker_id)s from the neutron server.',
+                  {'speaker_id': bgp_speaker_id})
+        self.remove_bgp_speaker_from_dragent(bgp_speaker_id)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_peer_association_end(self, context, payload):
+        """Handle bgp_peer_association_end notification event."""
+
+        bgp_peer_id = payload['bgp_peer']['peer_id']
+        bgp_speaker_id = payload['bgp_peer']['speaker_id']
+        LOG.debug('Received BGP peer associate notification for '
+                  'speaker_id=%(speaker_id)s peer_id=%(peer_id)s '
+                  'from the neutron server.',
+                  {'speaker_id': bgp_speaker_id,
+                   'peer_id': bgp_peer_id})
+        self.add_bgp_peer_helper(bgp_speaker_id, bgp_peer_id)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_peer_disassociation_end(self, context, payload):
+        """Handle bgp_peer_disassociation_end notification event."""
+
+        bgp_peer_ip = payload['bgp_peer']['peer_ip']
+        bgp_speaker_id = payload['bgp_peer']['speaker_id']
+        LOG.debug('Received BGP peer disassociate notification for '
+                  'speaker_id=%(speaker_id)s peer_ip=%(peer_ip)s '
+                  'from the neutron server.',
+                  {'speaker_id': bgp_speaker_id,
+                   'peer_ip': bgp_peer_ip})
+        self.remove_bgp_peer_from_bgp_speaker(bgp_speaker_id, bgp_peer_ip)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_routes_advertisement_end(self, context, payload):
+        """Handle bgp_routes_advertisement_end notification event."""
+
+        bgp_speaker_id = payload['advertise_routes']['speaker_id']
+        LOG.debug('Received routes advertisement end notification '
+                  'for speaker_id=%(speaker_id)s from the neutron server.',
+                  {'speaker_id': bgp_speaker_id})
+        routes = payload['advertise_routes']['routes']
+        self.add_routes_helper(bgp_speaker_id, routes)
+
+    @utils.synchronized('bgp-dr-agent')
+    def bgp_routes_withdrawal_end(self, context, payload):
+        """Handle bgp_routes_withdrawal_end notification event."""
+
+        bgp_speaker_id = payload['withdraw_routes']['speaker_id']
+        LOG.debug('Received route withdrawal notification for '
+                  'speaker_id=%(speaker_id)s from the neutron server.',
+                  {'speaker_id': bgp_speaker_id})
+        routes = payload['withdraw_routes']['routes']
+        self.withdraw_routes_helper(bgp_speaker_id, routes)
+
+    def add_bgp_speaker_helper(self, bgp_speaker_id):
+        """Add BGP speaker."""
+        bgp_speaker = self.safe_get_bgp_speaker_info(bgp_speaker_id)
+        if bgp_speaker:
+            self.add_bgp_speaker_on_dragent(bgp_speaker)
+
+    def add_bgp_peer_helper(self, bgp_speaker_id, bgp_peer_id):
+        """Add BGP peer."""
+        # Check if the BGP Speaker is already added or not
+        if not self.cache.is_bgp_speaker_added(bgp_speaker_id):
+            # Something went wrong. Let's re-sync
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason="BGP Speaker Out-of-sync")
+            return
+
+        bgp_peer = self.safe_get_bgp_peer_info(bgp_speaker_id,
+                                               bgp_peer_id)
+        if bgp_peer:
+            bgp_speaker_as = self.cache.get_bgp_speaker_local_as(
+                                                            bgp_speaker_id)
+            self.add_bgp_peer_to_bgp_speaker(bgp_speaker_id,
+                                             bgp_speaker_as,
+                                             bgp_peer)
+
+    def add_routes_helper(self, bgp_speaker_id, routes):
+        """Advertise routes to BGP speaker."""
+        # Check if the BGP Speaker is already added or not
+        if not self.cache.is_bgp_speaker_added(bgp_speaker_id):
+            # Something went wrong. Let's re-sync
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason="BGP Speaker Out-of-sync")
+            return
+
+        bgp_speaker_as = self.cache.get_bgp_speaker_local_as(bgp_speaker_id)
+        for route in routes:
+            self.advertise_route_via_bgp_speaker(bgp_speaker_id,
+                                                 bgp_speaker_as,
+                                                 route)
+            if self.is_resync_scheduled(bgp_speaker_id):
+                break
+
+    def withdraw_routes_helper(self, bgp_speaker_id, routes):
+        """Withdraw routes advertised by BGP speaker."""
+        if not self.cache.is_bgp_speaker_added(bgp_speaker_id):
+            # Something went wrong. Let's re-sync
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason="BGP Speaker Out-of-sync")
+            return
+
+        bgp_speaker_as = self.cache.get_bgp_speaker_local_as(bgp_speaker_id)
+        for route in routes:
+            self.withdraw_route_via_bgp_speaker(bgp_speaker_id,
+                                                bgp_speaker_as,
+                                                route)
+            if self.is_resync_scheduled(bgp_speaker_id):
+                break
+
+    def safe_get_bgp_speaker_info(self, bgp_speaker_id):
+        try:
+            bgp_speaker = self.plugin_rpc.get_bgp_speaker_info(self.context,
+                                                               bgp_speaker_id)
+            if not bgp_speaker:
+                LOG.warning(_LW('BGP Speaker %s has been deleted.'),
+                            bgp_speaker_id)
+            return bgp_speaker
+        except Exception as e:
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason=e)
+            LOG.error(_LE('BGP Speaker %(bgp_speaker)s info call '
+                          'failed with reason=%(e)s.'),
+                      {'bgp_speaker': bgp_speaker_id, 'e': e})
+
+    def safe_get_bgp_peer_info(self, bgp_speaker_id, bgp_peer_id):
+        try:
+            bgp_peer = self.plugin_rpc.get_bgp_peer_info(self.context,
+                                                         bgp_peer_id)
+            if not bgp_peer:
+                LOG.warning(_LW('BGP Peer %s has been deleted.'), bgp_peer)
+            return bgp_peer
+        except Exception as e:
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason=e)
+            LOG.error(_LE('BGP peer %(bgp_peer)s info call '
+                          'failed with reason=%(e)s.'),
+                      {'bgp_peer': bgp_peer_id, 'e': e})
+
+    @utils.exception_logger()
+    def safe_configure_dragent_for_bgp_speaker(self, bgp_speaker):
+        try:
+            self.add_bgp_speaker_on_dragent(bgp_speaker)
+        except (bgp_ext.BgpSpeakerNotFound, RuntimeError):
+            LOG.warning(_LW('BGP speaker %s may have been deleted and its '
+                            'resources may have already been disposed.'),
+                     bgp_speaker['id'])
+
+    def add_bgp_speaker_on_dragent(self, bgp_speaker):
+        # Caching BGP speaker details in BGPSpeakerCache. Will be used
+        # during smooth.
+        self.cache.put_bgp_speaker(bgp_speaker)
+
+        LOG.debug('Calling driver for adding BGP speaker %(speaker_id)s,'
+                  ' speaking for local_as %(local_as)s',
+                  {'speaker_id': bgp_speaker['id'],
+                   'local_as': bgp_speaker['local_as']})
+
+        # Add peer and route information to the driver.
+        self.add_bgp_peers_to_bgp_speaker(bgp_speaker)
+        self.advertise_routes_via_bgp_speaker(bgp_speaker)
+        self.schedule_resync(speaker_id=bgp_speaker['id'],
+                             reason="Periodic route cache refresh")
+
+    def remove_bgp_speaker_from_dragent(self, bgp_speaker_id):
+        if self.cache.is_bgp_speaker_added(bgp_speaker_id):
+            bgp_speaker_as = self.cache.get_bgp_speaker_local_as(
+                                                        bgp_speaker_id)
+            self.cache.remove_bgp_speaker_by_id(bgp_speaker_id)
+
+            LOG.debug('Calling driver for removing BGP speaker %(speaker_as)s',
+                      {'speaker_as': bgp_speaker_as})
+            return
+
+        # Something went wrong. Let's re-sync
+        self.schedule_resync(speaker_id=bgp_speaker_id,
+                             reason="BGP Speaker Out-of-sync")
+
+    def add_bgp_peers_to_bgp_speaker(self, bgp_speaker):
+        for bgp_peer in bgp_speaker['peers']:
+            self.add_bgp_peer_to_bgp_speaker(bgp_speaker['id'],
+                                             bgp_speaker['local_as'],
+                                             bgp_peer)
+            if self.is_resync_scheduled(bgp_speaker['id']):
+                break
+
+    def add_bgp_peer_to_bgp_speaker(self, bgp_speaker_id,
+                                    bgp_speaker_as, bgp_peer):
+        if self.cache.get_bgp_peer_by_ip(bgp_speaker_id, bgp_peer['peer_ip']):
+            return
+
+        self.cache.put_bgp_peer(bgp_speaker_id, bgp_peer)
+
+        LOG.debug('Calling driver interface for adding BGP peer %(peer_ip)s '
+                  'remote_as=%(remote_as)s to BGP Speaker running for '
+                  'local_as=%(local_as)d',
+                  {'peer_ip': bgp_peer['peer_ip'],
+                   'remote_as': bgp_peer['remote_as'],
+                   'local_as': bgp_speaker_as})
+
+    def remove_bgp_peer_from_bgp_speaker(self, bgp_speaker_id, bgp_peer_ip):
+        if not self.cache.is_bgp_speaker_added(bgp_speaker_id):
+            # Something went wrong. Let's re-sync
+            self.schedule_resync(speaker_id=bgp_speaker_id,
+                                 reason="BGP Speaker Out-of-sync")
+            return
+
+        if self.cache.is_bgp_peer_added(bgp_speaker_id, bgp_peer_ip):
+            self.cache.remove_bgp_peer_by_ip(bgp_speaker_id, bgp_peer_ip)
+
+            bgp_speaker_as = self.cache.get_bgp_speaker_local_as(
+                                                        bgp_speaker_id)
+
+            LOG.debug('Calling driver interface to remove BGP peer '
+                      '%(peer_ip)s from BGP Speaker running for '
+                      'local_as=%(local_as)d',
+                      {'peer_ip': bgp_peer_ip, 'local_as': bgp_speaker_as})
+            return
+
+        # Peer should have been found, Some problem, Let's re-sync
+        self.schedule_resync(speaker_id=bgp_speaker_id,
+                             reason="BGP Peer Out-of-sync")
+
+    def advertise_routes_via_bgp_speaker(self, bgp_speaker):
+        for route in bgp_speaker['advertised_routes']:
+            self.advertise_route_via_bgp_speaker(bgp_speaker['id'],
+                                                 bgp_speaker['local_as'],
+                                                 route)
+            if self.is_resync_scheduled(bgp_speaker['id']):
+                break
+
+    def advertise_route_via_bgp_speaker(self, bgp_speaker_id,
+                                        bgp_speaker_as, route):
+        if self.cache.is_route_advertised(bgp_speaker_id, route):
+            # Requested route already advertised. Hence, Nothing to be done.
+            return
+        self.cache.put_adv_route(bgp_speaker_id, route)
+
+        LOG.debug('Calling driver for advertising prefix: %(cidr)s, '
+                  'next_hop: %(nexthop)s',
+                  {'cidr': route['destination'],
+                   'nexthop': route['next_hop']})
+
+    def withdraw_route_via_bgp_speaker(self, bgp_speaker_id,
+                                       bgp_speaker_as, route):
+        if self.cache.is_route_advertised(bgp_speaker_id, route):
+            self.cache.remove_adv_route(bgp_speaker_id, route)
+            LOG.debug('Calling driver for withdrawing prefix: %(cidr)s, '
+                  'next_hop: %(nexthop)s',
+                  {'cidr': route['destination'],
+                   'nexthop': route['next_hop']})
+            return
+        # Something went wrong. Let's re-sync
+        self.schedule_resync(speaker_id=bgp_speaker_id,
+                             reason="Advertised routes Out-of-sync")
+
+    def schedule_full_resync(self, reason):
+        LOG.debug('Recording full resync request for all BGP Speakers '
+                  'with reason=%s', reason)
+        self.needs_full_sync_reason = reason
+
+    def schedule_resync(self, reason, speaker_id):
+        """Schedule a full resync for a given BGP Speaker.
+        If no BGP Speaker is specified, resync all BGP Speakers.
+        """
+        LOG.debug('Recording resync request for BGP Speaker %s '
+                  'with reason=%s', speaker_id, reason)
+        self.needs_resync_reasons[speaker_id].append(reason)
+
+    def is_resync_scheduled(self, bgp_speaker_id):
+        if bgp_speaker_id not in self.needs_resync_reasons:
+            return False
+
+        reason = self.needs_resync_reasons[bgp_speaker_id]
+        # Re-sync scheduled for the queried BGP speaker. No point
+        # continuing further. Let's stop processing and wait for
+        # re-sync to happen.
+        LOG.debug('Re-sync already scheduled for BGP Speaker %s '
+                  'with reason=%s', bgp_speaker_id, reason)
+        return True
+
+
+class BgpDrPluginApi(object):
+    """Agent side of BgpDrAgent RPC API.
+
+    This class implements the client side of an rpc interface.
+    The server side of this interface can be found in
+    neutron.api.rpc.handlers.bgp_speaker_rpc.BgpSpeakerRpcCallback.
+    For more information about changing rpc interfaces, see
+    doc/source/devref/rpc_api.rst.
+
+    API version history:
+        1.0 - Initial version.
+    """
+    def __init__(self, topic, context, host):
+        self.context = context
+        self.host = host
+        target = oslo_messaging.Target(topic=topic, version='1.0')
+        self.client = n_rpc.get_client(target)
+
+    def get_bgp_speakers(self, context):
+        """Make a remote process call to retrieve all BGP speakers info."""
+        cctxt = self.client.prepare()
+        return cctxt.call(context, 'get_bgp_speakers', host=self.host)
+
+    def get_bgp_speaker_info(self, context, bgp_speaker_id):
+        """Make a remote process call to retrieve a BGP speaker info."""
+        cctxt = self.client.prepare()
+        return cctxt.call(context, 'get_bgp_speaker_info',
+                          bgp_speaker_id=bgp_speaker_id)
+
+    def get_bgp_peer_info(self, context, bgp_peer_id):
+        """Make a remote process call to retrieve a BGP peer info."""
+        cctxt = self.client.prepare()
+        return cctxt.call(context, 'get_bgp_peer_info',
+                          bgp_peer_id=bgp_peer_id)
+
+
+class BgpSpeakerCache(object):
+    """Agent cache of the current BGP speaker state.
+
+    This class is designed to support the advertisement for
+    multiple BGP speaker via a single driver interface.
+
+    Version history:
+        1.0 - Initial version for caching the state of BGP speaker.
+    """
+    def __init__(self):
+        self.cache = {}
+
+    def get_bgp_speaker_ids(self):
+        return self.cache.keys()
+
+    def put_bgp_speaker(self, bgp_speaker):
+        if bgp_speaker['id'] in self.cache:
+            self.remove_bgp_speaker_by_id(self.cache[bgp_speaker['id']])
+        self.cache[bgp_speaker['id']] = {'bgp_speaker': bgp_speaker,
+                                         'peers': {},
+                                         'advertised_routes': []}
+
+    def get_bgp_speaker_by_id(self, bgp_speaker_id):
+        if bgp_speaker_id in self.cache:
+            return self.cache[bgp_speaker_id]['bgp_speaker']
+
+    def get_bgp_speaker_local_as(self, bgp_speaker_id):
+        bgp_speaker = self.get_bgp_speaker_by_id(bgp_speaker_id)
+        if bgp_speaker:
+            return bgp_speaker['local_as']
+
+    def is_bgp_speaker_added(self, bgp_speaker_id):
+        return self.get_bgp_speaker_by_id(bgp_speaker_id)
+
+    def remove_bgp_speaker_by_id(self, bgp_speaker_id):
+        if bgp_speaker_id in self.cache:
+            del self.cache[bgp_speaker_id]
+
+    def put_bgp_peer(self, bgp_speaker_id, bgp_peer):
+        if bgp_peer['peer_ip'] in self.get_bgp_peer_ips(bgp_speaker_id):
+            del self.cache[bgp_speaker_id]['peers'][bgp_peer['peer_ip']]
+
+        self.cache[bgp_speaker_id]['peers'][bgp_peer['peer_ip']] = bgp_peer
+
+    def is_bgp_peer_added(self, bgp_speaker_id, bgp_peer_ip):
+        return self.get_bgp_peer_by_ip(bgp_speaker_id, bgp_peer_ip)
+
+    def get_bgp_peer_ips(self, bgp_speaker_id):
+        bgp_speaker = self.get_bgp_speaker_by_id(bgp_speaker_id)
+        if bgp_speaker:
+            return self.cache[bgp_speaker_id]['peers'].keys()
+
+    def get_bgp_peer_by_ip(self, bgp_speaker_id, bgp_peer_ip):
+        bgp_speaker = self.get_bgp_speaker_by_id(bgp_speaker_id)
+        if bgp_speaker:
+            return self.cache[bgp_speaker_id]['peers'].get(bgp_peer_ip)
+
+    def remove_bgp_peer_by_ip(self, bgp_speaker_id, bgp_peer_ip):
+        if bgp_peer_ip in self.get_bgp_peer_ips(bgp_speaker_id):
+            del self.cache[bgp_speaker_id]['peers'][bgp_peer_ip]
+
+    def put_adv_route(self, bgp_speaker_id, route):
+        self.cache[bgp_speaker_id]['advertised_routes'].append(route)
+
+    def is_route_advertised(self, bgp_speaker_id, route):
+        routes = self.cache[bgp_speaker_id]['advertised_routes']
+        for r in routes:
+            if r['destination'] == route['destination'] and (
+                    r['next_hop'] == route['next_hop']):
+                return True
+        return False
+
+    def remove_adv_route(self, bgp_speaker_id, route):
+        routes = self.cache[bgp_speaker_id]['advertised_routes']
+        updated_routes = [r for r in routes if (
+            r['destination'] != route['destination'])]
+        self.cache[bgp_speaker_id]['advertised_routes'] = updated_routes
+
+    def get_adv_routes(self, bgp_speaker_id):
+        return self.cache[bgp_speaker_id]['advertised_routes']
+
+    def get_state(self):
+        bgp_speaker_ids = self.get_bgp_speaker_ids()
+        num_bgp_speakers = len(bgp_speaker_ids)
+        num_bgp_peers = 0
+        num_advertised_routes = 0
+        for bgp_speaker_id in bgp_speaker_ids:
+            bgp_speaker = self.get_bgp_speaker_by_id(bgp_speaker_id)
+            num_bgp_peers += len(bgp_speaker['peers'])
+            num_advertised_routes += len(bgp_speaker['advertised_routes'])
+        return {'bgp_speakers': num_bgp_speakers,
+                'bgp_peers': num_bgp_peers,
+                'advertise_routes': num_advertised_routes}
+
+
+class BgpDrAgentWithStateReport(BgpDrAgent):
+    def __init__(self, host, conf=None):
+        super(BgpDrAgentWithStateReport,
+              self).__init__(host, conf)
+        self.state_rpc = agent_rpc.PluginReportStateAPI(topics.PLUGIN)
+        self.agent_state = {
+            'agent_type': bgp_consts.AGENT_TYPE_BGP_ROUTING,
+            'binary': 'neutron-bgp-dragent',
+            'configurations': {},
+            'host': host,
+            'topic': bgp_consts.BGP_DRAGENT,
+            'start_flag': True}
+        report_interval = cfg.CONF.AGENT.report_interval
+        if report_interval:
+            self.heartbeat = loopingcall.FixedIntervalLoopingCall(
+                self._report_state)
+            self.heartbeat.start(interval=report_interval)
+
+    def _report_state(self):
+        LOG.debug("Report state task started")
+        try:
+            self.agent_state.get('configurations').update(
+                self.cache.get_state())
+            ctx = context.get_admin_context_without_session()
+            agent_status = self.state_rpc.report_state(ctx, self.agent_state,
+                                                       True)
+            if agent_status == constants.AGENT_REVIVED:
+                LOG.info(_LI("Agent has just been revived. "
+                             "Scheduling full sync"))
+                self.schedule_full_resync(
+                        reason=_("Agent has just been revived"))
+        except AttributeError:
+            # This means the server does not support report_state
+            LOG.warning(_LW("Neutron server does not support state report. "
+                            "State report for this agent will be disabled."))
+            self.heartbeat.stop()
+            self.run()
+            return
+        except Exception:
+            LOG.exception(_LE("Failed reporting state!"))
+            return
+        if self.agent_state.pop('start_flag', None):
+            self.run()
+
+    def agent_updated(self, context, payload):
+        """Handle the agent_updated notification event."""
+        self.schedule_full_resync(
+                reason=_("BgpDrAgent updated: %s") % payload)
+        LOG.info(_LI("agent_updated by server side %s!"), payload)
+
+    def after_start(self):
+        LOG.info(_LI("BGP dynamic routing agent started"))

neutron/services/bgp/agent/config.py

@@ -0,0 +1,18 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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.
+
+BGP_DRIVER_OPTS = []
+
+BGP_PROTO_CONFIG_OPTS = []

neutron/services/bgp/agent/entry.py

@@ -0,0 +1,47 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 sys
+
+from oslo_config import cfg
+from oslo_service import service
+
+from neutron.agent.common import config
+from neutron.agent.linux import external_process
+from neutron.common import config as common_config
+from neutron import service as neutron_service
+from neutron.services.bgp.agent import config as bgp_dragent_config
+from neutron.services.bgp.common import constants as bgp_consts
+
+
+def register_options():
+    config.register_agent_state_opts_helper(cfg.CONF)
+    config.register_root_helper(cfg.CONF)
+    cfg.CONF.register_opts(bgp_dragent_config.BGP_DRIVER_OPTS, 'BGP')
+    cfg.CONF.register_opts(bgp_dragent_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
+    cfg.CONF.register_opts(external_process.OPTS)
+
+
+def main():
+    register_options()
+    common_config.init(sys.argv[1:])
+    config.setup_logging()
+    server = neutron_service.Service.create(
+        binary='neutron-bgp-dragent',
+        topic=bgp_consts.BGP_DRAGENT,
+        report_interval=cfg.CONF.AGENT.report_interval,
+        manager='neutron.services.bgp.agent.bgp_dragent.'
+                'BgpDrAgentWithStateReport')
+    service.launch(cfg.CONF, server).wait()

neutron/services/bgp/bgp_plugin.py

@@ -13,15 +13,21 @@
 # under the License.
 
 from oslo_config import cfg
+from oslo_log import log as logging
 from oslo_utils import importutils
 
+from neutron.api.rpc.agentnotifiers import bgp_dr_rpc_agent_api
+from neutron.api.rpc.handlers import bgp_speaker_rpc as bs_rpc
+from neutron.common import rpc as n_rpc
 from neutron.db import bgp_db
 from neutron.db import bgp_dragentscheduler_db
 from neutron.extensions import bgp as bgp_ext
 from neutron.extensions import bgp_dragentscheduler as dras_ext
+from neutron.services.bgp.common import constants as bgp_consts
 from neutron.services import service_base
 
 PLUGIN_NAME = bgp_ext.BGP_EXT_ALIAS + '_svc_plugin'
+LOG = logging.getLogger(__name__)
 
 
 class BgpPlugin(service_base.ServicePluginBase,
@@ -35,6 +41,7 @@ class BgpPlugin(service_base.ServicePluginBase,
         super(BgpPlugin, self).__init__()
         self.bgp_drscheduler = importutils.import_object(
             cfg.CONF.bgp_drscheduler_driver)
+        self._setup_rpc()
 
     def get_plugin_name(self):
         return PLUGIN_NAME
@@ -46,3 +53,42 @@ class BgpPlugin(service_base.ServicePluginBase,
         """returns string description of the plugin."""
         return ("BGP dynamic routing service for announcement of next-hops "
                 "for tenant networks, floating IP's, and DVR host routes.")
+
+    def _setup_rpc(self):
+        self.topic = bgp_consts.BGP_PLUGIN
+        self.conn = n_rpc.create_connection()
+        self.agent_notifiers[bgp_consts.AGENT_TYPE_BGP_ROUTING] = (
+            bgp_dr_rpc_agent_api.BgpDrAgentNotifyApi()
+        )
+        self._bgp_rpc = self.agent_notifiers[bgp_consts.AGENT_TYPE_BGP_ROUTING]
+        self.endpoints = [bs_rpc.BgpSpeakerRpcCallback()]
+        self.conn.create_consumer(self.topic, self.endpoints,
+                                  fanout=False)
+        self.conn.consume_in_threads()
+
+    def add_bgp_peer(self, context, bgp_speaker_id, bgp_peer_info):
+        ret_value = super(BgpPlugin, self).add_bgp_peer(context,
+                                                        bgp_speaker_id,
+                                                        bgp_peer_info)
+        hosted_bgp_dragents = self.get_dragents_hosting_bgp_speakers(
+                                                             context,
+                                                             [bgp_speaker_id])
+        for agent in hosted_bgp_dragents:
+            self._bgp_rpc.bgp_peer_associated(context, bgp_speaker_id,
+                                              ret_value['bgp_peer_id'],
+                                              agent.host)
+        return ret_value
+
+    def remove_bgp_peer(self, context, bgp_speaker_id, bgp_peer_info):
+        hosted_bgp_dragents = self.get_dragents_hosting_bgp_speakers(
+            context, [bgp_speaker_id])
+
+        ret_value = super(BgpPlugin, self).remove_bgp_peer(context,
+                                                           bgp_speaker_id,
+                                                           bgp_peer_info)
+
+        for agent in hosted_bgp_dragents:
+            self._bgp_rpc.bgp_peer_disassociated(context,
+                                                 bgp_speaker_id,
+                                                 ret_value['bgp_peer_id'],
+                                                 agent.host)

neutron/services/bgp/common/constants.py

@@ -14,3 +14,7 @@
 # under the License.
 
 AGENT_TYPE_BGP_ROUTING = 'BGP dynamic routing agent'
+
+BGP_DRAGENT = 'bgp_dragent'
+
+BGP_PLUGIN = 'q-bgp-plugin'

neutron/services/bgp/common/opts.py

@@ -0,0 +1,28 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+# 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.
+
+import itertools
+
+import neutron.services.bgp.agent.config
+
+
+def list_bgp_agent_opts():
+    return [
+        ('BGP',
+         itertools.chain(
+             neutron.services.bgp.agent.config.BGP_DRIVER_OPTS,
+             neutron.services.bgp.agent.config.BGP_PROTO_CONFIG_OPTS)
+         )
+    ]

neutron/tests/unit/api/rpc/agentnotifiers/test_bgp_dr_rpc_agent_api.py

@@ -0,0 +1,83 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 mock
+
+from neutron.api.rpc.agentnotifiers import bgp_dr_rpc_agent_api
+from neutron import context
+from neutron.tests import base
+
+
+class TestBgpDrAgentNotifyApi(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestBgpDrAgentNotifyApi, self).setUp()
+        self.notifier = (
+            bgp_dr_rpc_agent_api.BgpDrAgentNotifyApi())
+
+        mock_cast_p = mock.patch.object(self.notifier,
+                                        '_notification_host_cast')
+        self.mock_cast = mock_cast_p.start()
+        mock_call_p = mock.patch.object(self.notifier,
+                                        '_notification_host_call')
+        self.mock_call = mock_call_p.start()
+        self.context = context.get_admin_context()
+        self.host = 'host-1'
+
+    def test_notify_dragent_bgp_routes_advertisement(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        routes = [{'destination': '1.1.1.1', 'next_hop': '2.2.2.2'}]
+        self.notifier.bgp_routes_advertisement(self.context, bgp_speaker_id,
+                                               routes, self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)
+
+    def test_notify_dragent_bgp_routes_withdrawal(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        routes = [{'destination': '1.1.1.1'}]
+        self.notifier.bgp_routes_withdrawal(self.context, bgp_speaker_id,
+                                            routes, self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)
+
+    def test_notify_bgp_peer_disassociated(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        bgp_peer_ip = '1.1.1.1'
+        self.notifier.bgp_peer_disassociated(self.context, bgp_speaker_id,
+                                             bgp_peer_ip, self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)
+
+    def test_notify_bgp_peer_associated(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        bgp_peer_id = 'bgp-peer-1'
+        self.notifier.bgp_peer_associated(self.context, bgp_speaker_id,
+                                          bgp_peer_id, self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)
+
+    def test_notify_bgp_speaker_created(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        self.notifier.bgp_speaker_created(self.context, bgp_speaker_id,
+                                          self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)
+
+    def test_notify_bgp_speaker_removed(self):
+        bgp_speaker_id = 'bgp-speaker-1'
+        self.notifier.bgp_speaker_removed(self.context, bgp_speaker_id,
+                                          self.host)
+        self.assertEqual(1, self.mock_cast.call_count)
+        self.assertEqual(0, self.mock_call.call_count)

neutron/tests/unit/api/rpc/handlers/test_bgp_speaker_rpc.py

@@ -0,0 +1,44 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 mock
+
+from neutron.api.rpc.handlers import bgp_speaker_rpc
+from neutron.tests import base
+
+
+class TestBgpSpeakerRpcCallback(base.BaseTestCase):
+
+    def setUp(self):
+        self.plugin_p = mock.patch('neutron.manager.NeutronManager.'
+                                   'get_service_plugins')
+        self.plugin = self.plugin_p.start()
+        self.callback = bgp_speaker_rpc.BgpSpeakerRpcCallback()
+        super(TestBgpSpeakerRpcCallback, self).setUp()
+
+    def test_get_bgp_speaker_info(self):
+        self.callback.get_bgp_speaker_info(mock.Mock(),
+                                           bgp_speaker_id='id1')
+        self.assertIsNotNone(len(self.plugin.mock_calls))
+
+    def test_get_bgp_peer_info(self):
+        self.callback.get_bgp_peer_info(mock.Mock(),
+                                        bgp_peer_id='id1')
+        self.assertIsNotNone(len(self.plugin.mock_calls))
+
+    def test_get_bgp_speakers(self):
+        self.callback.get_bgp_speakers(mock.Mock(),
+                                       host='host')
+        self.assertIsNotNone(len(self.plugin.mock_calls))

neutron/tests/unit/services/bgp/agent/test_bgp_dragent.py

@@ -0,0 +1,727 @@
+# Copyright 2016 Huawei Technologies India Pvt. Ltd.
+#
+# 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 copy
+import sys
+import uuid
+
+import eventlet
+import mock
+from oslo_config import cfg
+import testtools
+
+from neutron.common import config as common_config
+from neutron import context
+from neutron.services.bgp.agent import bgp_dragent
+from neutron.services.bgp.agent import config as bgp_config
+from neutron.services.bgp.agent import entry
+from neutron.tests import base
+
+HOSTNAME = 'hostname'
+rpc_api = bgp_dragent.BgpDrPluginApi
+BGP_PLUGIN = '%s.%s' % (rpc_api.__module__, rpc_api.__name__)
+
+FAKE_BGPSPEAKER_UUID = str(uuid.uuid4())
+FAKE_BGPPEER_UUID = str(uuid.uuid4())
+
+FAKE_BGP_SPEAKER = {'id': FAKE_BGPSPEAKER_UUID,
+                    'local_as': 12345,
+                    'peers': [{'remote_as': '2345',
+                               'peer_ip': '1.1.1.1',
+                               'password': ''}],
+                    'advertised_routes': []}
+
+FAKE_BGP_PEER = {'id': FAKE_BGPPEER_UUID,
+                 'remote_as': '2345',
+                 'peer_ip': '1.1.1.1',
+                 'password': ''}
+
+FAKE_ROUTE = {'id': FAKE_BGPSPEAKER_UUID,
+              'destination': '2.2.2.2/32',
+              'next_hop': '3.3.3.3'}
+
+FAKE_ROUTES = {'routes': {'id': FAKE_BGPSPEAKER_UUID,
+                          'destination': '2.2.2.2/32',
+                          'next_hop': '3.3.3.3'}
+               }
+
+
+class TestBgpDrAgent(base.BaseTestCase):
+    def setUp(self):
+        super(TestBgpDrAgent, self).setUp()
+        cfg.CONF.register_opts(bgp_config.BGP_DRIVER_OPTS, 'BGP')
+        cfg.CONF.register_opts(bgp_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
+        mock_log_p = mock.patch.object(bgp_dragent, 'LOG')
+        self.mock_log = mock_log_p.start()
+        self.context = context.get_admin_context()
+
+    def test_bgp_dragent_manager(self):
+        state_rpc_str = 'neutron.agent.rpc.PluginReportStateAPI'
+        # sync_state is needed for this test
+        with mock.patch.object(bgp_dragent.BgpDrAgentWithStateReport,
+                               'sync_state',
+                               autospec=True) as mock_sync_state:
+            with mock.patch(state_rpc_str) as state_rpc:
+                with mock.patch.object(sys, 'argv') as sys_argv:
+                    sys_argv.return_value = [
+                        'bgp_dragent', '--config-file',
+                        base.etcdir('neutron.conf')]
+                    common_config.init(sys.argv[1:])
+                    agent_mgr = bgp_dragent.BgpDrAgentWithStateReport(
+                        'testhost')
+                    eventlet.greenthread.sleep(1)
+                    agent_mgr.after_start()
+                    self.assertIsNotNone(len(mock_sync_state.mock_calls))
+                    state_rpc.assert_has_calls(
+                        [mock.call(mock.ANY),
+                         mock.call().report_state(mock.ANY, mock.ANY,
+                                                  mock.ANY)])
+
+    def test_bgp_dragent_main_agent_manager(self):
+        logging_str = 'neutron.agent.common.config.setup_logging'
+        launcher_str = 'oslo_service.service.ServiceLauncher'
+        with mock.patch(logging_str):
+            with mock.patch.object(sys, 'argv') as sys_argv:
+                with mock.patch(launcher_str) as launcher:
+                    sys_argv.return_value = ['bgp_dragent', '--config-file',
+                                             base.etcdir('neutron.conf')]
+                    entry.main()
+                    launcher.assert_has_calls(
+                        [mock.call(cfg.CONF),
+                         mock.call().launch_service(mock.ANY),
+                         mock.call().wait()])
+
+    def test_run_completes_single_pass(self):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+        with mock.patch.object(bgp_dr, 'sync_state') as sync_state:
+            bgp_dr.run()
+            self.assertIsNotNone(len(sync_state.mock_calls))
+
+    def test_after_start(self):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+        with mock.patch.object(bgp_dr, 'sync_state') as sync_state:
+            bgp_dr.after_start()
+            self.assertIsNotNone(len(sync_state.mock_calls))
+
+    def _test_sync_state_helper(self, bgp_speaker_list=None,
+                                cached_info=None,
+                                safe_configure_call_count=0,
+                                sync_bgp_speaker_call_count=0,
+                                remove_bgp_speaker_call_count=0,
+                                remove_bgp_speaker_ids=None,
+                                added_bgp_speakers=None,
+                                synced_bgp_speakers=None):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+
+        attrs_to_mock = dict(
+            [(a, mock.MagicMock())
+             for a in ['plugin_rpc', 'sync_bgp_speaker',
+                       'safe_configure_dragent_for_bgp_speaker',
+                       'remove_bgp_speaker_from_dragent']])
+
+        with mock.patch.multiple(bgp_dr, **attrs_to_mock):
+            if not cached_info:
+                cached_info = {}
+            if not added_bgp_speakers:
+                added_bgp_speakers = []
+            if not remove_bgp_speaker_ids:
+                remove_bgp_speaker_ids = []
+            if not synced_bgp_speakers:
+                synced_bgp_speakers = []
+
+            bgp_dr.plugin_rpc.get_bgp_speakers.return_value = bgp_speaker_list
+            bgp_dr.cache.cache = cached_info
+            bgp_dr.cache.clear_cache = mock.Mock()
+            bgp_dr.sync_state(mock.ANY)
+
+            self.assertEqual(
+                remove_bgp_speaker_call_count,
+                bgp_dr.remove_bgp_speaker_from_dragent.call_count)
+
+            if remove_bgp_speaker_call_count:
+                expected_calls = [mock.call(bgp_speaker_id)
+                                  for bgp_speaker_id in remove_bgp_speaker_ids]
+                bgp_dr.remove_bgp_speaker_from_dragent.assert_has_calls(
+                    expected_calls)
+
+            self.assertEqual(
+                safe_configure_call_count,
+                bgp_dr.safe_configure_dragent_for_bgp_speaker.call_count)
+
+            if safe_configure_call_count:
+                expected_calls = [mock.call(bgp_speaker)
+                                  for bgp_speaker in added_bgp_speakers]
+                bgp_dr.safe_configure_dragent_for_bgp_speaker.assert_has_calls(
+                    expected_calls)
+
+            self.assertEqual(sync_bgp_speaker_call_count,
+                             bgp_dr.sync_bgp_speaker.call_count)
+
+            if sync_bgp_speaker_call_count:
+                expected_calls = [mock.call(bgp_speaker)
+                                  for bgp_speaker in synced_bgp_speakers]
+                bgp_dr.sync_bgp_speaker.assert_has_calls(expected_calls)
+
+    def test_sync_state_bgp_speaker_added(self):
+        bgp_speaker_list = [{'id': 'foo-id',
+                             'local_as': 12345,
+                             'peers': [],
+                             'advertised_routes': []}]
+        self._test_sync_state_helper(bgp_speaker_list=bgp_speaker_list,
+                                     safe_configure_call_count=1,
+                                     added_bgp_speakers=bgp_speaker_list)
+
+    def test_sync_state_bgp_speaker_deleted(self):
+        bgp_speaker_list = []
+        cached_bgp_speaker = {'id': 'foo-id',
+                              'local_as': 12345,
+                              'peers': ['peer-1'],
+                              'advertised_routes': []}
+        cached_info = {'foo-id': cached_bgp_speaker}
+        self._test_sync_state_helper(bgp_speaker_list=bgp_speaker_list,
+                                     cached_info=cached_info,
+                                     remove_bgp_speaker_call_count=1,
+                                     remove_bgp_speaker_ids=['foo-id'])
+
+    def test_sync_state_added_and_deleted(self):
+        bgp_speaker_list = [{'id': 'foo-id',
+                             'local_as': 12345,
+                             'peers': [],
+                             'advertised_routes': []}]
+        cached_bgp_speaker = {'bgp_speaker': {'local_as': 12345},
+                              'peers': ['peer-1'],
+                              'advertised_routes': []}
+        cached_info = {'bar-id': cached_bgp_speaker}
+
+        self._test_sync_state_helper(bgp_speaker_list=bgp_speaker_list,
+                                     cached_info=cached_info,
+                                     remove_bgp_speaker_call_count=1,
+                                     remove_bgp_speaker_ids=['bar-id'],
+                                     safe_configure_call_count=1,
+                                     added_bgp_speakers=bgp_speaker_list)
+
+    def test_sync_state_added_and_synced(self):
+        bgp_speaker_list = [{'id': 'foo-id',
+                             'local_as': 12345,
+                             'peers': [],
+                             'advertised_routes': []},
+                            {'id': 'bar-id', 'peers': ['peer-2'],
+                             'advertised_routes': []},
+                            {'id': 'temp-id', 'peers': ['temp-1'],
+                                'advertised_routes': []}]
+
+        cached_bgp_speaker = {'id': 'bar-id', 'bgp_speaker': {'id': 'bar-id'},
+                              'peers': ['peer-1'],
+                              'advertised_routes': []}
+        cached_bgp_speaker_2 = {'id': 'temp-id',
+                                'bgp_speaker': {'id': 'temp-id'},
+                                'peers': ['temp-1'],
+                                'advertised_routes': []}
+        cached_info = {'bar-id': cached_bgp_speaker,
+                       'temp-id': cached_bgp_speaker_2}
+
+        self._test_sync_state_helper(bgp_speaker_list=bgp_speaker_list,
+                                     cached_info=cached_info,
+                                     safe_configure_call_count=1,
+                                     added_bgp_speakers=[bgp_speaker_list[0]],
+                                     sync_bgp_speaker_call_count=2,
+                                     synced_bgp_speakers=[bgp_speaker_list[1],
+                                                          bgp_speaker_list[2]]
+                                     )
+
+    def test_sync_state_added_synced_and_removed(self):
+        bgp_speaker_list = [{'id': 'foo-id',
+                             'local_as': 12345,
+                             'peers': [],
+                             'advertised_routes': []},
+                            {'id': 'bar-id', 'peers': ['peer-2'],
+                             'advertised_routes': []}]
+        cached_bgp_speaker = {'id': 'bar-id',
+                              'bgp_speaker': {'id': 'bar-id'},
+                              'peers': ['peer-1'],
+                              'advertised_routes': []}
+        cached_bgp_speaker_2 = {'id': 'temp-id',
+                                'bgp_speaker': {'id': 'temp-id'},
+                                'peers': ['temp-1'],
+                                'advertised_routes': []}
+        cached_info = {'bar-id': cached_bgp_speaker,
+                       'temp-id': cached_bgp_speaker_2}
+
+        self._test_sync_state_helper(bgp_speaker_list=bgp_speaker_list,
+                                     cached_info=cached_info,
+                                     remove_bgp_speaker_call_count=1,
+                                     remove_bgp_speaker_ids=['temp-id'],
+                                     safe_configure_call_count=1,
+                                     added_bgp_speakers=[bgp_speaker_list[0]],
+                                     sync_bgp_speaker_call_count=1,
+                                     synced_bgp_speakers=[bgp_speaker_list[1]])
+
+    def _test_sync_bgp_speaker_helper(self, bgp_speaker, cached_info=None,
+                                      remove_bgp_peer_call_count=0,
+                                      removed_bgp_peer_ip_list=None,
+                                      withdraw_route_call_count=0,
+                                      withdraw_routes_list=None,
+                                      add_bgp_peers_called=False,
+                                      advertise_routes_called=False):
+        if not cached_info:
+            cached_info = {}
+        if not removed_bgp_peer_ip_list:
+            removed_bgp_peer_ip_list = []
+        if not withdraw_routes_list:
+            withdraw_routes_list = []
+
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+
+        attrs_to_mock = dict(
+            [(a, mock.MagicMock())
+             for a in ['remove_bgp_peer_from_bgp_speaker',
+                       'add_bgp_peers_to_bgp_speaker',
+                       'advertise_routes_via_bgp_speaker',
+                       'withdraw_route_via_bgp_speaker']])
+
+        with mock.patch.multiple(bgp_dr, **attrs_to_mock):
+            bgp_dr.cache.cache = cached_info
+            bgp_dr.sync_bgp_speaker(bgp_speaker)
+
+            self.assertEqual(
+                remove_bgp_peer_call_count,
+                bgp_dr.remove_bgp_peer_from_bgp_speaker.call_count)
+
+            if remove_bgp_peer_call_count:
+                expected_calls = [mock.call(bgp_speaker['id'], peer_ip)
+                                  for peer_ip in removed_bgp_peer_ip_list]
+                bgp_dr.remove_bgp_peer_from_bgp_speaker.assert_has_calls(
+                    expected_calls)
+
+            self.assertEqual(add_bgp_peers_called,
+                             bgp_dr.add_bgp_peers_to_bgp_speaker.called)
+
+            if add_bgp_peers_called:
+                bgp_dr.add_bgp_peers_to_bgp_speaker.assert_called_with(
+                    bgp_speaker)
+
+            self.assertEqual(
+                withdraw_route_call_count,
+                bgp_dr.withdraw_route_via_bgp_speaker.call_count)
+
+            if withdraw_route_call_count:
+                expected_calls = [mock.call(bgp_speaker['id'], 12345, route)
+                                  for route in withdraw_routes_list]
+                bgp_dr.withdraw_route_via_bgp_speaker.assert_has_calls(
+                    expected_calls)
+
+            self.assertEqual(advertise_routes_called,
+                             bgp_dr.advertise_routes_via_bgp_speaker.called)
+
+            if advertise_routes_called:
+                bgp_dr.advertise_routes_via_bgp_speaker.assert_called_with(
+                    bgp_speaker)
+
+    def test_sync_bgp_speaker_bgp_peers_updated(self):
+        peers = [{'id': 'peer-1', 'peer_ip': '1.1.1.1'},
+                 {'id': 'peer-2', 'peer_ip': '2.2.2.2'}]
+        bgp_speaker = {'id': 'foo-id',
+                       'local_as': 12345,
+                       'peers': peers,
+                       'advertised_routes': []}
+
+        cached_peers = {'1.1.1.1': {'id': 'peer-2', 'peer_ip': '1.1.1.1'},
+                        '3.3.3.3': {'id': 'peer-3', 'peer_ip': '3.3.3.3'}}
+
+        cached_bgp_speaker = {'foo-id': {'bgp_speaker': {'local_as': 12345},
+                                         'peers': cached_peers,
+                                         'advertised_routes': []}}
+        self._test_sync_bgp_speaker_helper(
+            bgp_speaker, cached_info=cached_bgp_speaker,
+            remove_bgp_peer_call_count=1,
+            removed_bgp_peer_ip_list=['3.3.3.3'],
+            add_bgp_peers_called=True,
+            advertise_routes_called=False)
+
+    def test_sync_bgp_speaker_routes_updated(self):
+        adv_routes = [{'destination': '10.0.0.0/24', 'next_hop': '1.1.1.1'},
+                      {'destination': '20.0.0.0/24', 'next_hop': '2.2.2.2'}]
+        bgp_speaker = {'id': 'foo-id',
+                       'local_as': 12345,
+                       'peers': {},
+                       'advertised_routes': adv_routes}
+
+        cached_adv_routes = [{'destination': '20.0.0.0/24',
+                              'next_hop': '2.2.2.2'},
+                             {'destination': '30.0.0.0/24',
+                              'next_hop': '3.3.3.3'}]
+
+        cached_bgp_speaker = {
+            'foo-id': {'bgp_speaker': {'local_as': 12345},
+                       'peers': {},
+                       'advertised_routes': cached_adv_routes}}
+
+        self._test_sync_bgp_speaker_helper(
+            bgp_speaker, cached_info=cached_bgp_speaker,
+            withdraw_route_call_count=1,
+            withdraw_routes_list=[cached_adv_routes[1]],
+            add_bgp_peers_called=False,
+            advertise_routes_called=True)
+
+    def test_sync_bgp_speaker_peers_routes_added(self):
+        peers = [{'id': 'peer-1', 'peer_ip': '1.1.1.1'},
+                 {'id': 'peer-2', 'peer_ip': '2.2.2.2'}]
+        adv_routes = [{'destination': '10.0.0.0/24',
+                       'next_hop': '1.1.1.1'},
+                      {'destination': '20.0.0.0/24',
+                       'next_hop': '2.2.2.2'}]
+        bgp_speaker = {'id': 'foo-id',
+                       'local_as': 12345,
+                       'peers': peers,
+                       'advertised_routes': adv_routes}
+
+        cached_bgp_speaker = {
+            'foo-id': {'bgp_speaker': {'local_as': 12345},
+                       'peers': {},
+                       'advertised_routes': []}}
+
+        self._test_sync_bgp_speaker_helper(
+            bgp_speaker, cached_info=cached_bgp_speaker,
+            add_bgp_peers_called=True,
+            advertise_routes_called=True)
+
+    def test_sync_state_plugin_error(self):
+        with mock.patch(BGP_PLUGIN) as plug:
+            mock_plugin = mock.Mock()
+            mock_plugin.get_bgp_speakers.side_effect = Exception
+            plug.return_value = mock_plugin
+
+            with mock.patch.object(bgp_dragent.LOG, 'error') as log:
+                bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+                with mock.patch.object(bgp_dr,
+                        'schedule_full_resync') as schedule_full_resync:
+                    bgp_dr.sync_state(mock.ANY)
+
+                    self.assertTrue(log.called)
+                    self.assertTrue(schedule_full_resync.called)
+
+    def test_periodic_resync(self):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+        with mock.patch.object(bgp_dr,
+                               '_periodic_resync_helper') as resync_helper:
+            bgp_dr.periodic_resync(self.context)
+            self.assertTrue(resync_helper.called)
+
+    def test_periodic_resync_helper(self):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+        bgp_dr.schedule_resync('foo reason', 'foo-id')
+        with mock.patch.object(bgp_dr, 'sync_state') as sync_state:
+            sync_state.side_effect = RuntimeError
+            with testtools.ExpectedException(RuntimeError):
+                bgp_dr._periodic_resync_helper(self.context)
+            self.assertTrue(sync_state.called)
+            self.assertEqual(len(bgp_dr.needs_resync_reasons), 0)
+
+    def _test_add_bgp_peer_helper(self, bgp_speaker_id,
+                                  bgp_peer, cached_bgp_speaker,
+                                  put_bgp_peer_called=True):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+
+        bgp_dr.cache.cache = cached_bgp_speaker
+        with mock.patch.object(
+                bgp_dr.cache, 'put_bgp_peer') as mock_put_bgp_peer:
+            bgp_dr.add_bgp_peer_to_bgp_speaker('foo-id', 12345, bgp_peer)
+            if put_bgp_peer_called:
+                mock_put_bgp_peer.assert_called_once_with(
+                    bgp_speaker_id, bgp_peer)
+            else:
+                self.assertFalse(mock_put_bgp_peer.called)
+
+    def test_add_bgp_peer_not_cached(self):
+        bgp_peer = {'peer_ip': '1.1.1.1', 'remote_as': 34567,
+                    'password': 'abc'}
+        cached_bgp_speaker = {'foo-id': {'bgp_speaker': {'local_as': 12345},
+                                         'peers': {},
+                                         'advertised_routes': []}}
+
+        self._test_add_bgp_peer_helper('foo-id', bgp_peer, cached_bgp_speaker)
+
+    def test_add_bgp_peer_already_cached(self):
+        bgp_peer = {'peer_ip': '1.1.1.1', 'remote_as': 34567,
+                    'password': 'abc'}
+        cached_peers = {'1.1.1.1': {'peer_ip': '1.1.1.1', 'remote_as': 34567}}
+        cached_bgp_speaker = {'foo-id': {'bgp_speaker': {'local_as': 12345},
+                                         'peers': cached_peers,
+                                         'advertised_routes': []}}
+
+        self._test_add_bgp_peer_helper('foo-id', bgp_peer, cached_bgp_speaker,
+                                       put_bgp_peer_called=False)
+
+    def _test_advertise_route_helper(self, bgp_speaker_id,
+                                     route, cached_bgp_speaker,
+                                     put_adv_route_called=True):
+        bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+
+        bgp_dr.cache.cache = cached_bgp_speaker
+        with mock.patch.object(
+                bgp_dr.cache, 'put_adv_route') as mock_put_adv_route:
+            bgp_dr.advertise_route_via_bgp_speaker(bgp_speaker_id, 12345,
+                                                   route)
+            if put_adv_route_called:
+                mock_put_adv_route.assert_called_once_with(
+                    bgp_speaker_id, route)
+            else:
+                self.assertFalse(mock_put_adv_route.called)
+
+    def test_advertise_route_helper_not_cached(self):
+        route = {'destination': '10.0.0.0/24', 'next_hop': '1.1.1.1'}
+        cached_bgp_speaker = {'foo-id': {'bgp_speaker': {'local_as': 12345},
+                                         'peers': {},
+                                         'advertised_routes': []}}
+
+        self._test_advertise_route_helper('foo-id', route, cached_bgp_speaker,
+                                          put_adv_route_called=True)
+
+    def test_advertise_route_helper_already_cached(self):
+        route = {'destination': '10.0.0.0/24', 'next_hop': '1.1.1.1'}
+        cached_bgp_speaker = {'foo-id': {'bgp_speaker': {'local_as': 12345},
+                                         'peers': {},
+                                         'advertised_routes': [route]}}
+
+        self._test_advertise_route_helper('foo-id', route, cached_bgp_speaker,
+                                          put_adv_route_called=False)
+
+
+class TestBgpDrAgentEventHandler(base.BaseTestCase):
+
+    cache_cls = 'neutron.services.bgp.agent.bgp_dragent.BgpSpeakerCache'
+
+    def setUp(self):
+        super(TestBgpDrAgentEventHandler, self).setUp()
+        cfg.CONF.register_opts(bgp_config.BGP_DRIVER_OPTS, 'BGP')
+        cfg.CONF.register_opts(bgp_config.BGP_PROTO_CONFIG_OPTS, 'BGP')
+
+        mock_log_p = mock.patch.object(bgp_dragent, 'LOG')
+        self.mock_log = mock_log_p.start()
+
+        self.plugin_p = mock.patch(BGP_PLUGIN)
+        plugin_cls = self.plugin_p.start()
+        self.plugin = mock.Mock()
+        plugin_cls.return_value = self.plugin
+
+        self.cache_p = mock.patch(self.cache_cls)
+        cache_cls = self.cache_p.start()
+        self.cache = mock.Mock()
+        cache_cls.return_value = self.cache
+
+        self.bgp_dr = bgp_dragent.BgpDrAgent(HOSTNAME)
+        self.schedule_full_resync_p = mock.patch.object(
+                                        self.bgp_dr, 'schedule_full_resync')
+        self.schedule_full_resync = self.schedule_full_resync_p.start()
+        self.context = mock.Mock()
+
+    def test_bgp_speaker_create_end(self):
+        payload = {'bgp_speaker': {'id': FAKE_BGPSPEAKER_UUID}}
+
+        with mock.patch.object(self.bgp_dr,
+                               'add_bgp_speaker_helper') as enable:
+            self.bgp_dr.bgp_speaker_create_end(None, payload)
+            enable.assert_called_once_with(FAKE_BGP_SPEAKER['id'])
+
+    def test_bgp_peer_association_end(self):
+        payload = {'bgp_peer': {'speaker_id': FAKE_BGPSPEAKER_UUID,
+                                'peer_id': FAKE_BGPPEER_UUID}}
+
+        with mock.patch.object(self.bgp_dr,
+                               'add_bgp_peer_helper') as enable:
+            self.bgp_dr.bgp_peer_association_end(None, payload)
+            enable.assert_called_once_with(FAKE_BGP_SPEAKER['id'],
+                                           FAKE_BGP_PEER['id'])
+
+    def test_route_advertisement_end(self):
+        routes = [{'destination': '2.2.2.2/32', 'next_hop': '3.3.3.3'},
+                  {'destination': '4.4.4.4/32', 'next_hop': '5.5.5.5'}]
+        payload = {'advertise_routes': {'speaker_id': FAKE_BGPSPEAKER_UUID,
+                                        'routes': routes}}
+
+        expected_calls = [mock.call(FAKE_BGP_SPEAKER['id'], routes)]
+
+        with mock.patch.object(self.bgp_dr,
+                               'add_routes_helper') as enable:
+            self.bgp_dr.bgp_routes_advertisement_end(None, payload)
+            enable.assert_has_calls(expected_calls)
+
+    def test_add_bgp_speaker_helper(self):
+        self.plugin.get_bgp_speaker_info.return_value = FAKE_BGP_SPEAKER
+        add_bs_p = mock.patch.object(self.bgp_dr,
+                                   'add_bgp_speaker_on_dragent')
+        add_bs = add_bs_p.start()
+        self.bgp_dr.add_bgp_speaker_helper(FAKE_BGP_SPEAKER['id'])
+        self.plugin.assert_has_calls([
+            mock.call.get_bgp_speaker_info(mock.ANY,
+                                           FAKE_BGP_SPEAKER['id'])])
+        add_bs.assert_called_once_with(FAKE_BGP_SPEAKER)
+
+    def test_add_bgp_peer_helper(self):
+        self.plugin.get_bgp_peer_info.return_value = FAKE_BGP_PEER
+        add_bp_p = mock.patch.object(self.bgp_dr,
+                                     'add_bgp_peer_to_bgp_speaker')
+        add_bp = add_bp_p.start()
+        self.bgp_dr.add_bgp_peer_helper(FAKE_BGP_SPEAKER['id'],
+                                        FAKE_BGP_PEER['id'])
+        self.plugin.assert_has_calls([
+            mock.call.get_bgp_peer_info(mock.ANY,
+                                        FAKE_BGP_PEER['id'])])
+        self.assertEqual(1, add_bp.call_count)
+
+    def test_add_routes_helper(self):
+        add_rt_p = mock.patch.object(self.bgp_dr,
+                                     'advertise_route_via_bgp_speaker')
+        add_bp = add_rt_p.start()
+        self.bgp_dr.add_routes_helper(FAKE_BGP_SPEAKER['id'], FAKE_ROUTES)
+        self.assertEqual(1, add_bp.call_count)
+
+    def test_bgp_speaker_remove_end(self):
+        payload = {'bgp_speaker': {'id': FAKE_BGPSPEAKER_UUID}}
+
+        with mock.patch.object(self.bgp_dr,
+                               'remove_bgp_speaker_from_dragent') as disable:
+            self.bgp_dr.bgp_speaker_remove_end(None, payload)
+            disable.assert_called_once_with(FAKE_BGP_SPEAKER['id'])
+
+    def test_bgp_peer_disassociation_end(self):
+        payload = {'bgp_peer': {'speaker_id': FAKE_BGPSPEAKER_UUID,
+                                'peer_ip': '1.1.1.1'}}
+
+        with mock.patch.object(self.bgp_dr,
+                               'remove_bgp_peer_from_bgp_speaker') as disable:
+            self.bgp_dr.bgp_peer_disassociation_end(None, payload)
+            disable.assert_called_once_with(FAKE_BGPSPEAKER_UUID,
+                                            FAKE_BGP_PEER['peer_ip'])
+
+    def test_bgp_routes_withdrawal_end(self):
+        withdraw_routes = [{'destination': '2.2.2.2/32'},
+                           {'destination': '3.3.3.3/32'}]
+        payload = {'withdraw_routes': {'speaker_id': FAKE_BGPSPEAKER_UUID,
+                                       'routes': withdraw_routes}}
+
+        expected_calls = [mock.call(FAKE_BGP_SPEAKER['id'], withdraw_routes)]
+
+        with mock.patch.object(self.bgp_dr,
+                               'withdraw_routes_helper') as disable:
+            self.bgp_dr.bgp_routes_withdrawal_end(None, payload)
+            disable.assert_has_calls(expected_calls)
+
+
+class TestBGPSpeakerCache(base.BaseTestCase):
+
+    def setUp(self):
+        super(TestBGPSpeakerCache, self).setUp()
+        self.expected_cache = {FAKE_BGP_SPEAKER['id']:
+                               {'bgp_speaker': FAKE_BGP_SPEAKER,
+                                'peers': {},
+                                'advertised_routes': []}}
+        self.bs_cache = bgp_dragent.BgpSpeakerCache()
+
+    def test_put_bgp_speaker(self):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+        self.assertEqual(self.expected_cache, self.bs_cache.cache)
+
+    def test_put_bgp_speaker_existing(self):
+        prev_bs_info = {'id': 'foo-id'}
+        with mock.patch.object(self.bs_cache,
+                               'remove_bgp_speaker_by_id') as remove:
+            self.bs_cache.cache[FAKE_BGP_SPEAKER['id']] = prev_bs_info
+            self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+            remove.assert_called_once_with(prev_bs_info)
+        self.assertEqual(self.expected_cache, self.bs_cache.cache)
+
+    def remove_bgp_speaker_by_id(self):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+        self.assertEqual(1, len(self.bs_cache.cache))
+        self.bs_cache.remove_bgp_speaker_by_id(FAKE_BGP_SPEAKER['id'])
+        self.assertEqual(0, len(self.bs_cache.cache))
+
+    def test_get_bgp_speaker_by_id(self):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+
+        self.assertEqual(
+            FAKE_BGP_SPEAKER,
+            self.bs_cache.get_bgp_speaker_by_id(FAKE_BGP_SPEAKER['id']))
+
+    def test_get_bgp_speaker_ids(self):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+
+        self.assertEqual([FAKE_BGP_SPEAKER['id']],
+                         list(self.bs_cache.get_bgp_speaker_ids()))
+
+    def _test_bgp_peer_helper(self, remove=False):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+        self.bs_cache.put_bgp_peer(FAKE_BGP_SPEAKER['id'], FAKE_BGP_PEER)
+        expected_cache = copy.deepcopy(self.expected_cache)
+        expected_cache[FAKE_BGP_SPEAKER['id']]['peers'] = {
+            FAKE_BGP_PEER['peer_ip']: FAKE_BGP_PEER}
+        self.assertEqual(expected_cache, self.bs_cache.cache)
+
+        if remove:
+            self.bs_cache.remove_bgp_peer_by_ip(FAKE_BGP_SPEAKER['id'],
+                                                'foo-ip')
+            self.assertEqual(expected_cache, self.bs_cache.cache)
+
+            self.bs_cache.remove_bgp_peer_by_ip(FAKE_BGP_SPEAKER['id'],
+                                                FAKE_BGP_PEER['peer_ip'])
+            self.assertEqual(self.expected_cache, self.bs_cache.cache)
+
+    def test_put_bgp_peer(self):
+        self._test_bgp_peer_helper()
+
+    def test_remove_bgp_peer(self):
+        self._test_bgp_peer_helper(remove=True)
+
+    def _test_bgp_speaker_adv_route_helper(self, remove=False):
+        self.bs_cache.put_bgp_speaker(FAKE_BGP_SPEAKER)
+        self.bs_cache.put_adv_route(FAKE_BGP_SPEAKER['id'], FAKE_ROUTE)
+        expected_cache = copy.deepcopy(self.expected_cache)
+        expected_cache[FAKE_BGP_SPEAKER['id']]['advertised_routes'].append(
+            FAKE_ROUTE)
+        self.assertEqual(expected_cache, self.bs_cache.cache)
+
+        fake_route_2 = copy.deepcopy(FAKE_ROUTE)
+        fake_route_2['destination'] = '4.4.4.4/32'
+        self.bs_cache.put_adv_route(FAKE_BGP_SPEAKER['id'], fake_route_2)
+
+        expected_cache[FAKE_BGP_SPEAKER['id']]['advertised_routes'].append(
+            fake_route_2)
+        self.assertEqual(expected_cache, self.bs_cache.cache)
+
+        if remove:
+            self.bs_cache.remove_adv_route(FAKE_BGP_SPEAKER['id'],
+                                           fake_route_2)
+            expected_cache[FAKE_BGP_SPEAKER['id']]['advertised_routes'] = (
+                [FAKE_ROUTE])
+            self.assertEqual(expected_cache, self.bs_cache.cache)
+
+            self.bs_cache.remove_adv_route(FAKE_BGP_SPEAKER['id'],
+                                           FAKE_ROUTE)
+            self.assertEqual(self.expected_cache, self.bs_cache.cache)
+
+    def test_put_bgp_speaker_adv_route(self):
+        self._test_bgp_speaker_adv_route_helper()
+
+    def test_remove_bgp_speaker_adv_route(self):
+        self._test_bgp_speaker_adv_route_helper(remove=True)
+
+    def test_is_bgp_speaker_adv_route_present(self):
+        self._test_bgp_speaker_adv_route_helper()
+        self.assertTrue(self.bs_cache.is_route_advertised(
+            FAKE_BGP_SPEAKER['id'], FAKE_ROUTE))
+        self.assertFalse(self.bs_cache.is_route_advertised(
+            FAKE_BGP_SPEAKER['id'], {'destination': 'foo-destination',
+                                     'next_hop': 'foo-next-hop'}))

setup.cfg

@@ -40,6 +40,7 @@ scripts =
 
 [entry_points]
 console_scripts =
+    neutron-bgp-dragent = neutron.cmd.eventlet.agents.bgp_dragent:main
     neutron-db-manage = neutron.db.migration.cli:main
     neutron-debug = neutron.debug.shell:main
     neutron-dhcp-agent = neutron.cmd.eventlet.agents.dhcp:main
@@ -127,6 +128,7 @@ oslo.config.opts =
     neutron = neutron.opts:list_opts
     neutron.agent = neutron.opts:list_agent_opts
     neutron.base.agent = neutron.opts:list_base_agent_opts
+    neutron.bgp.agent = neutron.services.bgp.common.opts:list_bgp_agent_opts
     neutron.db = neutron.opts:list_db_opts
     neutron.dhcp.agent = neutron.opts:list_dhcp_agent_opts
     neutron.extensions = neutron.opts:list_extension_opts